Selenium proxy setup and authentication

import asyncio
import base64
import threading

HOP_BY_HOP = ("proxy-authorization:", "proxy-connection:", "connection:", "keep-alive:")


def start_forwarder(username: str, password: str, gateway: str = "scale-us.netdash.io:2000") -> str:
    """Start a local proxy on 127.0.0.1 that forwards to the netdash gateway with credentials.

    Returns its URL (for example http://127.0.0.1:53211) to pass to the browser. It adds the
    Proxy-Authorization header and passes everything else through: HTTPS stays a CONNECT tunnel,
    so the forwarder never sees page content. It runs until your script exits.
    """
    host, port = gateway.rsplit(":", 1)
    token = base64.b64encode(f"{username}:{password}".encode()).decode()

    async def pipe(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        try:
            while True:
                data = await reader.read(65536)
                if not data:
                    break
                writer.write(data)
                await writer.drain()
        except (ConnectionError, asyncio.CancelledError):
            pass
        finally:
            writer.close()

    async def handle(client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter) -> None:
        try:
            head = await client_reader.readuntil(b"\r\n\r\n")
            upstream = asyncio.open_connection(host, int(port))
            upstream_reader, upstream_writer = await asyncio.wait_for(upstream, timeout=30)
        except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, asyncio.TimeoutError, OSError):
            client_writer.close()
            return
        lines = head.decode("latin-1").split("\r\n")
        request_line = lines[0]
        headers = [h for h in lines[1:] if h and not h.lower().startswith(HOP_BY_HOP)]
        headers.insert(0, f"Proxy-Authorization: Basic {token}")
        if not request_line.startswith("CONNECT "):
            headers.append("Connection: close")  # one request per connection, each with credentials
        upstream_writer.write(("\r\n".join([request_line, *headers]) + "\r\n\r\n").encode("latin-1"))
        await asyncio.gather(pipe(client_reader, upstream_writer), pipe(upstream_reader, client_writer))

    loop = asyncio.new_event_loop()
    server = loop.run_until_complete(asyncio.start_server(handle, "127.0.0.1", 0))
    threading.Thread(target=loop.run_forever, daemon=True).start()
    return f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}"

import os

from selenium import webdriver

from netdash_forwarder import start_forwarder

username = f"{os.environ['NETDASH_USERNAME']}-cc-US"
local_proxy = start_forwarder(username, os.environ["NETDASH_PASSWORD"], "scale-us.netdash.io:2000")

options = webdriver.ChromeOptions()
options.add_argument(f"--proxy-server={local_proxy}")

driver = webdriver.Chrome(options=options)
try:
    driver.set_page_load_timeout(30)
    driver.get("https://api.ipify.org?format=json")
    print(driver.find_element("tag name", "body").text)
finally:
    driver.quit()

import os
import secrets
from typing import Optional

from selenium import webdriver

from netdash_forwarder import start_forwarder


def netdash_driver(country: str = "US", session_id: Optional[str] = None, ttl_minutes: int = 30,
                   gateway: str = "scale-us.netdash.io:2000") -> webdriver.Chrome:
    """Chrome through netdash: rotating without session_id, sticky with one."""
    cc = country.upper()
    cc = "GB" if cc == "UK" else cc
    username = f"{os.environ['NETDASH_USERNAME']}-cc-{cc}"
    if session_id is not None:
        username += f"-session-{session_id}-ttl-{ttl_minutes}"
    local_proxy = start_forwarder(username, os.environ["NETDASH_PASSWORD"], gateway)
    options = webdriver.ChromeOptions()
    options.add_argument(f"--proxy-server={local_proxy}")
    driver = webdriver.Chrome(options=options)
    driver.set_page_load_timeout(30)
    return driver


# A sticky German session through the Europe gateway: one exit IP for the whole test
driver = netdash_driver("DE", session_id=secrets.token_hex(4), gateway="scale-eu.netdash.io:2000")
try:
    for _ in range(2):
        driver.get("https://api.ipify.org?format=json")
        print(driver.find_element("tag name", "body").text)  # the same exit IP both times
finally:
    driver.quit()