Selenium和认证代理：实用的本地转发器

import asyncio
import base64
import os

UPSTREAM = ("gw.crawlrelay.com", 7777)
USER = os.environ.get("CRAWLRELAY_USER", "USERNAME-country-us")
PASSWORD = os.environ.get("CRAWLRELAY_PASSWORD", "PASSWORD")
AUTH = b"Proxy-Authorization: Basic " + base64.b64encode(("%s:%s" % (USER, PASSWORD)).encode())


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


async def handle(client_reader, client_writer):
    try:
        head = await client_reader.readuntil(b"\r\n\r\n")  # request line and headers from the browser
    except asyncio.IncompleteReadError:
        client_writer.close()
        return
    lines = head.split(b"\r\n")
    lines.insert(1, AUTH)  # add the credentials right after the request line
    up_reader, up_writer = await asyncio.open_connection(*UPSTREAM)
    up_writer.write(b"\r\n".join(lines))
    await up_writer.drain()
    await asyncio.gather(pipe(client_reader, up_writer), pipe(up_reader, client_writer))


async def main():
    server = await asyncio.start_server(handle, "127.0.0.1", 8899)
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())

from selenium import webdriver
from selenium.webdriver.common.by import By

options = webdriver.ChromeOptions()
options.add_argument("--proxy-server=http://127.0.0.1:8899")  # the local forwarder
options.add_argument("--headless=new")

driver = webdriver.Chrome(options=options)
try:
    driver.set_page_load_timeout(60)
    driver.get("https://api.ipify.org")
    print("exit IP:", driver.find_element(By.TAG_NAME, "body").text)
finally:
    driver.quit()