Selenium 为 Chrome 配置静态代理

"""Selenium + Chrome through a static proxy authorised by IP allowlist.

Install: python -m pip install selenium
Before running: add this machine's public IP to the allowlist of your account.
"""
import os

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

HOST = os.environ.get("PROXY_HOST", "gw.steadyproxy.com")
PORT = os.environ.get("PROXY_PORT", "7777")

options = webdriver.ChromeOptions()
# Host and port only: Chrome's --proxy-server switch has no field for a username or password
options.add_argument("--proxy-server=http://" + HOST + ":" + PORT)

driver = webdriver.Chrome(options=options)
try:
    driver.get("https://api.ipify.org?format=json")
    print("Exit IP:", driver.find_element(By.TAG_NAME, "body").text)
finally:
    driver.quit()

"""Local forwarder: Chrome -> 127.0.0.1:8899 (no login) -> static proxy (with login).

It adds the Proxy-Authorization header for you. Keep it bound to 127.0.0.1.
Run: python local_forwarder.py   (standard library only)
"""
import asyncio
import base64
import os

LISTEN_PORT = int(os.environ.get("LOCAL_PORT", "8899"))
UPSTREAM_HOST = os.environ.get("PROXY_HOST", "gw.steadyproxy.com")
UPSTREAM_PORT = int(os.environ.get("PROXY_PORT", "7777"))
CREDENTIALS = os.environ.get("PROXY_USER", "USERNAME") + ":" + os.environ.get("PROXY_PASS", "PASSWORD")
AUTH = "Proxy-Authorization: Basic " + base64.b64encode(CREDENTIALS.encode()).decode()
DROP = ("proxy-authorization:", "proxy-connection:", "connection:")


async def pipe(reader, writer):
    try:
        while data := await reader.read(65536):
            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")
    except (asyncio.IncompleteReadError, asyncio.LimitOverrunError):
        client_writer.close()
        return
    lines = head.decode("latin-1").split("\r\n")
    headers = [h for h in lines[1:] if h and not h.lower().startswith(DROP)]
    headers.append(AUTH)
    if not lines[0].upper().startswith("CONNECT "):
        headers.append("Connection: close")  # plain http:// - one request per connection
    try:
        up_reader, up_writer = await asyncio.open_connection(UPSTREAM_HOST, UPSTREAM_PORT)
    except OSError:
        client_writer.write(b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\n\r\n")
        client_writer.close()
        return
    up_writer.write(("\r\n".join([lines[0]] + headers) + "\r\n\r\n").encode("latin-1"))
    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", LISTEN_PORT)
    print("Forwarding 127.0.0.1:%d -> %s:%d" % (LISTEN_PORT, UPSTREAM_HOST, UPSTREAM_PORT))
    async with server:
        await server.serve_forever()


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

"""Selenium + Chrome through the local forwarder (username/password kept out of Chrome).

Start local_forwarder.py first, then run this file.
"""
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")

driver = webdriver.Chrome(options=options)
try:
    driver.get("https://api.ipify.org?format=json")
    print("Exit IP:", driver.find_element(By.TAG_NAME, "body").text)
finally:
    driver.quit()