"""Cases for the aiohttp proxy lab. Each case is a complete script body appended to PRELUDE. A case is (name, environment variables, python -W value or None, code). The runner starts every case in a new interpreter with a clean environment, so a case sees only the variables listed for it. """ PRELUDE = '''\ import asyncio, os, ssl import aiohttp HTTP_URL = "http://127.0.0.1:8080/" HTTPS_URL = "https://127.0.0.1:8443/" ECHO_URL = "https://127.0.0.1:8444/" # answers with the names of the headers it received PROXY = "http://127.0.0.1:8888" TLS = ssl.create_default_context(cafile=os.environ["LAB_CA"]) # trusts the lab's self-signed target def run(main): try: asyncio.run(main()) except Exception as exc: print(f"EXCEPTION {type(exc).__module__}.{type(exc).__qualname__}: {exc}") async def show(response): body = (await response.text()).strip() print(f"status={response.status} body={body!r} proxy_authenticate={response.headers.get('Proxy-Authenticate')!r}") ''' def get(url: str, request_args: str = "", session_args: str = "") -> str: """One GET in one session; request_args and session_args are spliced in verbatim.""" arguments = ", ".join(part for part in (url, "ssl=TLS" if url in ("HTTPS_URL", "ECHO_URL") else "", request_args) if part) return f''' async def main(): async with aiohttp.ClientSession({session_args}) as session: async with session.get({arguments}) as response: await show(response) run(main) ''' def socks(connector_args: str, url: str) -> str: arguments = url + (", ssl=TLS" if url == "HTTPS_URL" else "") return f''' from aiohttp_socks import ProxyConnector async def main(): connector = ProxyConnector.from_url({connector_args}) async with aiohttp.ClientSession(connector=connector) as session: async with session.get({arguments}) as response: await show(response) run(main) ''' BASIC_AUTH = 'proxy=PROXY, proxy_auth=aiohttp.BasicAuth("user", "pass")' GOOD_HEADERS = 'proxy_headers={"Proxy-Authorization": aiohttp.encode_basic_auth("user", "pass")}' WRONG_HEADERS = 'proxy_headers={"Proxy-Authorization": aiohttp.encode_basic_auth("user", "wrong")}' REQUEST_HEADER = 'headers={"Proxy-Authorization": aiohttp.encode_basic_auth("user", "pass")}' AUTHORIZATION_KEY = 'proxy_headers={"Authorization": aiohttp.encode_basic_auth("user", "pass")}' URL_GOOD = 'proxy="http://user:pass@127.0.0.1:8888"' URL_WRONG = 'proxy="http://user:wrong@127.0.0.1:8888"' ENV_PROXY = "http://user:pass@127.0.0.1:8888" # The deprecated call lives in an imported module here, not in the script Python was started with. IN_MODULE = ''' import pathlib pathlib.Path("legacy_client.py").write_text("\\n".join([ "import aiohttp", "", "async def fetch(url, proxy):", " async with aiohttp.ClientSession() as session:", " async with session.get(url, proxy=proxy, proxy_auth=aiohttp.BasicAuth('user', 'pass')) as response:", " return response.status", "", ])) import legacy_client async def main(): print("status", await legacy_client.fetch(HTTP_URL, PROXY)) run(main) ''' CASES = [ # --- Authentication: three ways to send the same credentials -------------------------------- ("auth-basicauth-http", {}, "always", get("HTTP_URL", BASIC_AUTH)), ("auth-basicauth-https", {}, "always", get("HTTPS_URL", BASIC_AUTH)), ("auth-headers-http", {}, "always", get("HTTP_URL", f"proxy=PROXY, {GOOD_HEADERS}")), ("auth-headers-https", {}, "always", get("HTTPS_URL", f"proxy=PROXY, {GOOD_HEADERS}")), ("auth-url-http", {}, "always", get("HTTP_URL", URL_GOOD)), ("auth-url-https", {}, "always", get("HTTPS_URL", URL_GOOD)), # --- Other places the header could go -------------------------------------------------------- ("auth-request-header-http", {}, "always", get("HTTP_URL", f"proxy=PROXY, {REQUEST_HEADER}")), ("auth-request-header-https", {}, "always", get("HTTPS_URL", f"proxy=PROXY, {REQUEST_HEADER}")), ("auth-both-headers-http", {}, "always", get("HTTP_URL", f"proxy=PROXY, {GOOD_HEADERS}, {REQUEST_HEADER}")), ("auth-both-headers-https", {}, "always", get("HTTPS_URL", f"proxy=PROXY, {GOOD_HEADERS}, {REQUEST_HEADER}")), ("auth-proxy-headers-authorization-http", {}, "always", get("HTTP_URL", f"proxy=PROXY, {AUTHORIZATION_KEY}")), ("auth-proxy-headers-authorization-https", {}, "always", get("HTTPS_URL", f"proxy=PROXY, {AUTHORIZATION_KEY}")), # --- Which hop receives which header ------------------------------------------------------ ("proxy-headers-custom-http", {}, "always", get("HTTP_URL", URL_GOOD + ', proxy_headers={"X-Lab": "1"}')), ("proxy-headers-custom-https", {}, "always", get("HTTPS_URL", URL_GOOD + ', proxy_headers={"X-Lab": "1"}')), ("proxy-headers-https-at-target", {}, "always", get("ECHO_URL", f"proxy=PROXY, {GOOD_HEADERS}")), ("request-header-https-at-target", {}, "always", get("ECHO_URL", f"proxy=PROXY, {GOOD_HEADERS}, {REQUEST_HEADER}")), # --- Wrong or missing credentials ----------------------------------------------------------- ("wrong-url-https", {}, "always", get("HTTPS_URL", URL_WRONG)), ("wrong-url-https-status-and-message", {}, "always", ''' async def main(): async with aiohttp.ClientSession() as session: try: async with session.get(HTTPS_URL, ssl=TLS, proxy="http://user:wrong@127.0.0.1:8888") as response: await show(response) except aiohttp.ClientHttpProxyError as exc: print(f"status={exc.status} message={exc.message!r}") print("subclass of ClientResponseError:", isinstance(exc, aiohttp.ClientResponseError)) run(main) '''), ("wrong-headers-https", {}, "always", get("HTTPS_URL", f"proxy=PROXY, {WRONG_HEADERS}")), ("wrong-basicauth-https", {}, "always", get("HTTPS_URL", 'proxy=PROXY, proxy_auth=aiohttp.BasicAuth("user", "wrong")')), ("wrong-url-http", {}, "always", get("HTTP_URL", URL_WRONG)), ("wrong-url-http-raise", {}, "always", get("HTTP_URL", URL_WRONG, "raise_for_status=True")), ("noauth-http", {}, "always", get("HTTP_URL", "proxy=PROXY")), ("noauth-https", {}, "always", get("HTTPS_URL", "proxy=PROXY")), ("noauth-http-session-raise", {}, "always", get("HTTP_URL", "proxy=PROXY", "raise_for_status=True")), ("noauth-http-request-raise", {}, "always", get("HTTP_URL", "proxy=PROXY, raise_for_status=True")), ("noauth-http-method-raise", {}, "always", ''' async def main(): async with aiohttp.ClientSession() as session: async with session.get(HTTP_URL, proxy=PROXY) as response: print(f"status={response.status} ok={response.ok}") response.raise_for_status() run(main) '''), # --- Session-level proxy settings ----------------------------------------------------------- ("session-proxy-request-headers-http", {}, "always", get("HTTP_URL", GOOD_HEADERS, "proxy=PROXY")), ("session-proxy-request-headers-https", {}, "always", get("HTTPS_URL", GOOD_HEADERS, "proxy=PROXY")), ("session-proxy-url-credentials-http", {}, "always", get("HTTP_URL", "", URL_GOOD)), ("session-proxy-url-credentials-https", {}, "always", get("HTTPS_URL", "", URL_GOOD)), ("session-proxy-auth", {}, "always", get("HTTP_URL", "", BASIC_AUTH)), ("session-proxy-headers", {}, "always", get("HTTP_URL", "", f"proxy=PROXY, {GOOD_HEADERS}")), # --- The BasicAuth deprecation on its own --------------------------------------------------- ("basicauth-construct", {}, "always", 'print(aiohttp.BasicAuth("user", "pass"))\n'), ("encode-basic-auth", {}, "always", 'print(aiohttp.encode_basic_auth("user", "pass"))\n'), ("server-auth-basicauth", {}, "always", get("HTTP_URL", 'auth=aiohttp.BasicAuth("user", "pass")')), ("server-auth-header", {}, "always", get("HTTP_URL", 'headers={"Authorization": aiohttp.encode_basic_auth("user", "pass")}')), ("encoding-defaults", {}, "always", 'print(aiohttp.encode_basic_auth("user", "p\u00e4ss"))\nprint(aiohttp.BasicAuth("user", "p\u00e4ss").encode())\nprint(aiohttp.encode_basic_auth("user", "p\u00e4ss", encoding="latin1"))\n'), ("basicauth-default-warnings", {}, None, get("HTTP_URL", BASIC_AUTH)), ("basicauth-in-module-default-warnings", {}, None, IN_MODULE), ("basicauth-in-module-warnings-always", {}, "always", IN_MODULE), ("basicauth-warnings-as-errors", {}, "error", get("HTTP_URL", BASIC_AUTH)), # --- Environment variables ------------------------------------------------------------------ ("env-default-session", {"HTTP_PROXY": ENV_PROXY}, "always", get("HTTP_URL")), ("env-trust-env", {"HTTP_PROXY": ENV_PROXY}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-trust-env-lowercase", {"http_proxy": ENV_PROXY}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-trust-env-https", {"HTTPS_PROXY": ENV_PROXY}, "always", get("HTTPS_URL", "", "trust_env=True")), ("env-trust-env-https-wrong-password", {"HTTPS_PROXY": "http://user:wrong@127.0.0.1:8888"}, "always", get("HTTPS_URL", "", "trust_env=True")), ("env-http-var-https-url", {"HTTP_PROXY": ENV_PROXY}, "always", get("HTTPS_URL", "", "trust_env=True")), ("env-all-proxy", {"ALL_PROXY": ENV_PROXY}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-no-credentials", {"HTTP_PROXY": "http://127.0.0.1:8888"}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-no-proxy", {"HTTP_PROXY": ENV_PROXY, "NO_PROXY": "127.0.0.1"}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-no-proxy-other-host", {"HTTP_PROXY": ENV_PROXY, "NO_PROXY": "example.com"}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-no-proxy-explicit-proxy", {"NO_PROXY": "127.0.0.1"}, "always", get("HTTP_URL", URL_GOOD, "trust_env=True")), ("env-explicit-proxy-wins", {"HTTP_PROXY": "http://127.0.0.1:9"}, "always", get("HTTP_URL", URL_GOOD, "trust_env=True")), ("env-dead-proxy", {"HTTP_PROXY": "http://127.0.0.1:9"}, "always", get("HTTP_URL", "", "trust_env=True")), ("env-https-scheme-proxy", {"HTTP_PROXY": "https://user:pass@127.0.0.1:8888"}, "always", get("HTTP_URL", "", "trust_env=True")), # --- SOCKS ---------------------------------------------------------------------------------- ("socks-scheme-http", {}, "always", get("HTTP_URL", 'proxy="socks5://127.0.0.1:1080"')), ("socks-scheme-https", {}, "always", get("HTTPS_URL", 'proxy="socks5://127.0.0.1:1080"')), ("socks5h-scheme-http", {}, "always", get("HTTP_URL", 'proxy="socks5h://127.0.0.1:1080"')), ("aiohttp-socks-http", {}, "always", socks('"socks5://127.0.0.1:1080"', "HTTP_URL")), ("aiohttp-socks-https", {}, "always", socks('"socks5://127.0.0.1:1080"', "HTTPS_URL")), ("aiohttp-socks-socks5h", {}, "always", socks('"socks5h://127.0.0.1:1080"', "HTTP_URL")), ("aiohttp-socks-hostname", {}, "always", socks('"socks5://127.0.0.1:1080"', '"http://localhost:8080/"')), ("aiohttp-socks-hostname-rdns", {}, "always", socks('"socks5://127.0.0.1:1080", rdns=True', '"http://localhost:8080/"')), ("aiohttp-socks-hostname-rdns-false", {}, "always", socks('"socks5://127.0.0.1:1080", rdns=False', '"http://localhost:8080/"')), # --- An https:// proxy URL pointed at a plain-HTTP proxy port ------------------------------- ("https-proxy-url-http-target", {}, "always", get("HTTP_URL", 'proxy="https://user:pass@127.0.0.1:8888"')), ("https-proxy-url-https-target", {}, "always", get("HTTPS_URL", 'proxy="https://user:pass@127.0.0.1:8888"')), ]