how to add proxy to non seleniumbase tools

[neochine]

import os
from pathlib import Path
from typing import Optional


class ChromeProxyExtension:
    EXT_DIR = Path("extensions/proxy")

    @staticmethod
    def parse_proxy(proxy: str):
        proxy = proxy.strip()

        username = password = None

        if "@" in proxy:
            creds, hostport = proxy.rsplit("@", 1)
            if ":" in creds:
                username, password = creds.split(":", 1)
            else:
                username = creds
                password = ""
        else:
            hostport = proxy

        if ":" not in hostport:
            raise ValueError("Proxy must be host:port or user:pass@host:port")

        host, port = hostport.rsplit(":", 1)
        return host, port, username, password  # keep port as string

    @classmethod
    def build(cls, proxy: str, bypass_list: Optional[list[str]] = None) -> str:
        host, port, user, pwd = cls.parse_proxy(proxy)

        cls.EXT_DIR.mkdir(parents=True, exist_ok=True)

        bypass = bypass_list or ["localhost", "127.0.0.1"]
        bypass_js = ", ".join(f'"{x}"' for x in bypass)

        manifest = """{
  "version": "1.0.0",
  "manifest_version": 3,
  "name": "Proxy Extension",
  "permissions": [
    "proxy",
    "storage",
    "webRequest",
    "webRequestAuthProvider"
  ],
  "host_permissions": ["<all_urls>"],
  "background": {
    "service_worker": "background.js"
  }
}
"""

        if user is not None:
            auth_js = f"""
function callbackFn(details) {{
    return {{
        authCredentials: {{
            username: "{user}",
            password: "{pwd}"
        }}
    }};
}}

chrome.webRequest.onAuthRequired.addListener(
    callbackFn,
    {{urls: ["<all_urls>"]}},
    ["blocking"]
);
"""
        else:
            auth_js = "// No auth required\n"

        # IMPORTANT CHANGE: port is now parseInt("{port}")
        background = f"""
var config = {{
    mode: "fixed_servers",
    rules: {{
        singleProxy: {{
            scheme: "http",
            host: "{host}",
            port: parseInt("{port}")
        }},
        bypassList: [{bypass_js}]
    }}
}};

chrome.proxy.settings.set(
    {{value: config, scope: "regular"}},
    function() {{}}
);

{auth_js}
"""

        (cls.EXT_DIR / "manifest.json").write_text(manifest, encoding="utf-8")
        (cls.EXT_DIR / "background.js").write_text(background, encoding="utf-8")

        return os.path.abspath(str(cls.EXT_DIR))

import zendriver as zd
import asyncio
import time
from seleniumbase import SB

from core.proxyhelper import ChromeProxyExtension

proxy ="user:pass@host:port"

async def main():
    # 1. build extension
    ext_path = ChromeProxyExtension.build(proxy)

    config = zd.Config()
    config.add_extension("extensions/nopecha")
    config.add_extension(ext_path)


    driver = await zd.start(config=config)

    await driver.get("https://httpbin.org/ip")
    await asyncio.sleep(10)
    await driver.get("https://icanhazip.com")
    await asyncio.sleep(10)
    await driver.get("https://httpbin.org/ip")

    time.sleep(40)


def main2():
    with SB(browser="chrome", uc=True, extension_dir="extensions/nopecha", headed=True, proxy=proxy,) as sb:
        sb.open("https://httpbin.org/ip")
        time.sleep(40)


main2()
#asyncio.run(main())

Thing is seleniumbase doesnt show original ip in httpbin but zendriver does. How do i make sure its not detected

[neochine]

i was trying to do seleniumwire like inspection in seleniumbase but it doesnt support that well whereas zendriver supports it well. so i want to add proxy like seleniumbase

[mdmintz]

SeleniumBase supports all packet inspection, including XHR:
Eg. SeleniumBase/examples/cdp_mode/raw_xhr_sb.py
(Lots of other examples around that too.)

For setting a proxy with SeleniumBase, use the proxy arg.
Formats: "server:port" or "user:pass@server:port".

[mdmintz]

This requires login. Show me an example with a public-facing website.

[neochine]

create test account please i have hard time finding another reproducible website with similar scenario
You can use this test account though. Replace (dot) with actual .
MTUxNTA2MTEwNjI4MzExODY5Mw(dot)Gko8u8(dot)d3oiGfRzpcto0I87hBR7MUX2OzG6tiFwtBMdII
Run the minimal reproducible code I shared above, once you hit ptpython press F12, use this token and in console write this

function login(token) {
setInterval(() => {
document.body.appendChild(document.createElement `iframe`).contentWindow.localStorage.token = `"${token}"`
}, 50);
setTimeout(() => {
location.reload();
}, 250);
}
login("MTUxNTA2MTEwNjI4MzExODY5Mw(dot)Gko8u8(dot)d3oiGfRzpcto0I87hBR7MUX2OzG6tiFwtBMdII")

replace (dot) with .
Press Ctrl + D to exit ptpython. Then let the script run. You wont even get single XHR request @mdmintz

[neochine]

However if you reload manually you can see few XHR requests to /science with Authorization header

[mdmintz]

I not going to log in to Discord with automation. But I will see if I can find another way to see why you're not seeing all the XHR that you should. Perhaps there's a public-facing site that reproduces the issue. Maybe there was a CDP update and I just need to refresh some methods to see everything.

[neochine]

Alright I am converting entire codebase to async then

[neochine]

@mdmintz

187     tab = await sb.get("about:blank")
188     listenXHR(tab)
189
190     # Change url to something that makes ajax requests
191     tab = await sb.get('https://discord.com/channels/@me') #
192
193     xhr_responses = await receiveXHR(tab, xhr_requests)
194     #xhr_responses = loop.run_until_complete(receiveXHR(tab, xhr_requests))
195     for response in xhr_responses:
196         print(c1 + "*** ==> XHR Request URL <== ***" + cr)
197         print(f'{response["url"]}')
198         is_base64 = response["is_base64"]
199         b64_data = "Base64 encoded data"
200         try:
201             headers = ast.literal_eval(response["body"])["headers"]
202             print(c2 + "*** ==> XHR Response Headers <== ***" + cr)
203             print(headers if not is_base64 else b64_data)
204         except Exception:
205             response_body = response["body"]
206             print(c2 + "*** ==> XHR Response Body <== ***" + cr)
207             print(response_body if not is_base64 else b64_data)
208
209     #sb.cdp.add_handler(mycdp.network.RequestWillBeSent, on_request)
210     #sb.cdp.add_handler(mycdp.network.ResponseReceived, on_response)
211
212     url = await sb.get('https://discord.com/channels/@me')
213     await sb.sleep(1)
214     sb.find_element("//span[text()='Create Account']")
215
216     xhr_responses = await receiveXHR(url, xhr_requests)
217
218     n = 0
219     while n < 15:
220         n += 1
221         await sb.sleep(1)
222     assert bool(xhr_responses ), "Failed"
223
224     from ptpython.repl import embed
225     embed(globals(), locals())

This doesnt work as well

[neochine]

    await sb.sleep(1)
       ^^^^^^^^^^^^^^
AssertionError: Failed

[neochine]

listenXHR and recieveXHR is same copy paste

 34 import ast
 35 import mycdp
 36 import sys
 37 import time
 38 from seleniumbase import SB #note that i imported cdp_driver above and didnt use SB() which is sync
 39
 40 xhr_requests = []
 41 last_xhr_request = None
 42
 43 def listenXHR(page):
 44     async def handler(evt):
 45         # Get AJAX requests
 46         if evt.type_ is mycdp.network.ResourceType.XHR:
 47             xhr_requests.append([evt.response.url, evt.request_id])
 48             global last_xhr_request
 49             last_xhr_request = time.time()
 50     page.add_handler(mycdp.network.ResponseReceived, handler)
 51
 52
 53 async def receiveXHR(page, requests):
 54     responses = []
 55     retries = 0
 56     max_retries = 5
 57     # Wait at least 2 seconds after last XHR request for more
 58     while True:
 59         if last_xhr_request is None or retries > max_retries:
 60             break
 61         if time.time() - last_xhr_request <= 2:
 62             retries = retries + 1
 63             time.sleep(2)
 64             continue
 65         else:
 66             break
 67     await page
 68     # Loop through gathered requests and get response body
 69     for request in requests:
 70         try:
 71             res = await page.send(mycdp.network.get_response_body(request[1]))
 72             if res is None:
 73                 continue
 74             responses.append({
 75                 "url": request[0],
 76                 "body": res[0],
 77                 "is_base64": res[1],
 78             })
 79         except Exception as e:
 80             print("Error getting response:", e)
 81     return responses
 82

[mdmintz]

There was a MyCDP update that may have helped.
Upgrade to SeleniumBase 4.49.11 - https://github.com/seleniumbase/SeleniumBase/releases/tag/v4.49.11

[neochine]

ASYNC works now!! I updated to 49.11 modified code a bit and it could detect now

    tab = await sb.get("about:blank")
    listenXHR(tab)

    # Change url to something that makes ajax requests
    tab = await sb.get('https://discord.com/channels/@me') #

    xhr_responses = await receiveXHR(tab, xhr_requests)
    if not xhr_responses:
        n = 0
        while n < 15:
            n += 1
            await sb.sleep(1)
        print(xhr_responses)
        xhr_responses = await receiveXHR(tab, xhr_requests)

    n = 0
    while n < 15:
        n += 1
        await sb.sleep(1)

    #xhr_responses = loop.run_until_complete(receiveXHR(tab, xhr_requests))
    for response in xhr_responses:
        print("*** ==> XHR Request URL <== ***")
        print(f'{response["url"]}')
        is_base64 = response["is_base64"]
        b64_data = "Base64 encoded data"
        try:
            headers = ast.literal_eval(response["body"])["headers"]
            print("*** ==> XHR Response Headers <== ***")
            print(headers if not is_base64 else b64_data)
        except Exception:
            response_body = response["body"]
            print("*** ==> XHR Response Body <== ***")
            print(response_body if not is_base64 else b64_data)

    #sb.cdp.add_handler(mycdp.network.RequestWillBeSent, on_request)
    #sb.cdp.add_handler(mycdp.network.ResponseReceived, on_response)

    url = await sb.get('https://discord.com/channels/@me') #
    if not xhr_responses:
        xhr_responses = await receiveXHR(url, xhr_requests)

        n = 0
        while n < 15:
            n += 1
            await sb.sleep(1)
        assert bool(xhr_responses ), "Failed"

    from ptpython.repl import embed
    embed(globals(), locals())

Sync code still doesnt work but at least async works

[neochine]

Btw discord.com/login still sends /science request. You could write unittests to detect /science XHR request on visiting https://discord.com/login

[neochine]

for both sync and async. if 'science' in event.request.url, test pass