#!/usr/bin/env python3 """Local-only checks for aiohttp session and SOCKS5 domain-handoff boundaries. Requires: Python 3.9+, aiohttp 3.13.5, aiohttp-socks 0.12.0 for the tested run. It binds only 127.0.0.1 on ephemeral ports. It never contacts a real proxy, resolves lab.invalid, forwards a request, changes TLS verification, or logs secrets. """ import asyncio import json import socket import sys from typing import Dict, Tuple import aiohttp import aiohttp_socks import python_socks from aiohttp_socks import ProxyConnector, ProxyType HOST = "127.0.0.1" LAB_HOST = "lab.invalid" TIMEOUT = aiohttp.ClientTimeout(total=2) class LoopbackOnlyResolver(aiohttp.abc.AbstractResolver): """Resolve exactly localhost without consulting a system or external resolver.""" async def resolve(self, host: str, port: int = 0, family: int = socket.AF_INET): if host != "localhost": raise OSError("fixture resolver permits only localhost") return [{ "hostname": host, "host": HOST, "port": port, "family": socket.AF_INET, "proto": 0, "flags": socket.AI_NUMERICHOST, }] async def close(self) -> None: return None async def read_headers(reader: asyncio.StreamReader) -> bytes: return await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=2) async def cookie_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: """A tiny loopback HTTP server that has only /set and /echo.""" try: for _ in range(3): # The fixture issues exactly three local HTTP requests. request = await read_headers(reader) first_line = request.split(b"\r\n", 1)[0] if first_line.startswith(b"GET /set "): headers = b"Set-Cookie: fixture=green; Path=/\r\n" body = b'{"cookie":"set"}' elif first_line.startswith(b"GET /echo "): saw_cookie = b"Cookie: fixture=green" in request headers = b"" body = json.dumps({"fixture_cookie_seen": saw_cookie}).encode() else: headers = b"" body = b'{"error":"only /set and /echo are allowed"}' writer.write( b"HTTP/1.1 200 OK\r\n" + headers + b"Content-Type: application/json\r\nContent-Length: " + str(len(body)).encode() + b"\r\n\r\n" + body ) await writer.drain() except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, asyncio.TimeoutError): pass finally: writer.close() await writer.wait_closed() async def socks_handler( reader: asyncio.StreamReader, writer: asyncio.StreamWriter, observed: Dict[str, object] ) -> None: """Accept one SOCKS5 domain-form CONNECT for a fixed lab host, then stop.""" try: greeting = await asyncio.wait_for(reader.readexactly(3), timeout=2) if greeting != b"\x05\x01\x00": raise ValueError("only SOCKS5/no-auth greeting is allowed") writer.write(b"\x05\x00") await writer.drain() ver, command, reserved, atyp = await asyncio.wait_for(reader.readexactly(4), timeout=2) if (ver, command, reserved, atyp) != (5, 1, 0, 3): raise ValueError("only SOCKS5 domain-form CONNECT is allowed") length = (await asyncio.wait_for(reader.readexactly(1), timeout=2))[0] hostname = (await asyncio.wait_for(reader.readexactly(length), timeout=2)).decode("ascii") port = int.from_bytes(await asyncio.wait_for(reader.readexactly(2), timeout=2), "big") if hostname != LAB_HOST or port != 80: raise ValueError("fixture permits only lab.invalid:80") observed.update({"atyp": atyp, "hostname": hostname, "port": port}) # Success says only that this loopback stub accepted the request shape. writer.write(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00") await writer.drain() # Consume but do not forward the HTTP request. This also prevents the # synthetic response from racing the client's post-handshake request. await read_headers(reader) # These bytes are a local synthetic HTTP reply, not a forwarded response. writer.write(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n") await writer.drain() except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, asyncio.TimeoutError, ValueError) as error: observed["error"] = str(error) finally: writer.close() await writer.wait_closed() def port_of(server: asyncio.AbstractServer) -> int: socket = server.sockets[0] return int(socket.getsockname()[1]) async def get_json(session: aiohttp.ClientSession, url: str) -> Dict[str, object]: async with session.get(url) as response: response.raise_for_status() return await response.json() async def run_fixture() -> Dict[str, object]: cookie_server = await asyncio.start_server(cookie_handler, HOST, 0) cookies_url = f"http://localhost:{port_of(cookie_server)}" observed: Dict[str, object] = {} socks_server = await asyncio.start_server( lambda reader, writer: socks_handler(reader, writer, observed), HOST, 0 ) resolver = LoopbackOnlyResolver() connector = aiohttp.TCPConnector(limit=2, resolver=resolver) try: async with aiohttp.ClientSession( connector=connector, connector_owner=False, timeout=TIMEOUT ) as first_session: await get_json(first_session, cookies_url + "/set") same_session = await get_json(first_session, cookies_url + "/echo") async with aiohttp.ClientSession( connector=connector, connector_owner=False, timeout=TIMEOUT ) as second_session: separate_session = await get_json(second_session, cookies_url + "/echo") socks_connector = ProxyConnector( proxy_type=ProxyType.SOCKS5, host=HOST, port=port_of(socks_server), rdns=True ) try: async with aiohttp.ClientSession(connector=socks_connector, timeout=TIMEOUT) as session: async with session.get(f"http://{LAB_HOST}/") as response: socks_status = response.status finally: await socks_connector.close() result = { "same_session_cookie_seen": same_session["fixture_cookie_seen"], "separate_session_cookie_seen": separate_session["fixture_cookie_seen"], "socks_status": socks_status, "socks_request": observed, } assert result == { "same_session_cookie_seen": True, "separate_session_cookie_seen": False, "socks_status": 204, "socks_request": {"atyp": 3, "hostname": LAB_HOST, "port": 80}, }, result return result finally: await connector.close() await resolver.close() socks_server.close() cookie_server.close() await socks_server.wait_closed() await cookie_server.wait_closed() def main() -> int: if len(sys.argv) > 2 or (len(sys.argv) == 2 and sys.argv[1] != "--self-test"): print("usage: aiohttp-proxy-integration-checks.py [--self-test]", file=sys.stderr) return 2 result = asyncio.run(run_fixture()) print(json.dumps(result, sort_keys=True)) print( "ENV: python={} aiohttp={} aiohttp-socks={} python-socks={}".format( sys.version.split()[0], aiohttp.__version__, aiohttp_socks.__version__, python_socks.__version__, ) ) print( "PASS: local cookie isolation and SOCKS5 domain-form handoff only; " "no external DNS, upstream routing, TLS, provider exit, stickiness, or cancellation was tested." ) return 0 if __name__ == "__main__": sys.exit(main())