#!/usr/bin/env python3 """Run the Scrapy proxy lab on loopback and write results.json. Starts seven fixtures on 127.0.0.1 (python -m http.server on 8080, the same with TLS on 8443, the jobs target on 8081, three Basic-auth proxies on 8888, 8889 and 8890, and a logging SOCKS5 server on 1080), then runs every case from cases.py with `python -m scrapy runspider` in a new interpreter with a clean environment. For each case it records what the spider printed, the Scrapy log, the crawl stats, and what the proxies, the SOCKS server and the jobs target logged while the case ran. python3 -m venv .venv && . .venv/bin/activate python -m pip install -r requirements.txt python run_lab.py # writes results.json; pass a file name to keep another run Needs Python 3.14 (http.server --tls-cert) and the openssl command for the lab certificate. """ import json import pathlib import platform import re import socket import subprocess import sys import tempfile import time from importlib import metadata from cases import CASES HERE = pathlib.Path(__file__).resolve().parent PORTS = {"http_target": 8080, "https_target": 8443, "jobs_target": 8081, "proxy": 8888, "proxy_b": 8889, "proxy_c": 8890, "socks": 1080} SETTINGS = {"LOG_FORMAT": "%(levelname)s [%(name)s] %(message)s", "TELNETCONSOLE_ENABLED": "False", "REMOTE_CONTROL_ENABLED": "False", "DOWNLOAD_TIMEOUT": "15"} RECORD = re.compile(r"^(DEBUG|INFO|WARNING|ERROR|CRITICAL) \[([^\]]+)\] ") # Left out of the recorded log: start-up records that are the same in every case (settings, enabled # components, versions) and the per-byte DEBUG trace of the httpcore2 library. SKIPPED_LOGGERS = {"scrapy.utils.log", "scrapy.addons", "scrapy.middleware", "scrapy.crawler", "scrapy.extensions.logstats", "scrapy.extensions.telnet", "scrapy.statscollectors", "asyncio"} STATS = re.compile(r"^[ {]*'([^']+)': (.+?),?}?$") STATS_KEPT = ("downloader/request_count", "downloader/request_method_count/", "downloader/response_status_count/", "downloader/exception_type_count/", "retry/", "httperror/", "finish_reason", "log_count/", "response_received_count") def wait_for(port: int) -> None: for _ in range(100): try: socket.create_connection(("127.0.0.1", port), timeout=0.2).close() return except OSError: time.sleep(0.1) raise SystemExit(f"nothing is listening on 127.0.0.1:{port}") def new_lines(path: pathlib.Path, offset: int) -> tuple[list[str], int]: text = path.read_text(encoding="utf-8") if path.exists() else "" return text[offset:].splitlines(), len(text) def split_log(stderr: str) -> tuple[list[str], dict[str, str]]: """Return the log without the start-up records, and the kept keys of the final stats dump.""" kept: list[str] = [] stats: dict[str, str] = {} logger = "" for line in stderr.splitlines(): if match := RECORD.match(line): logger = match.group(2) if logger == "scrapy.statscollectors": if (match := STATS.match(line)) and match.group(1).startswith(STATS_KEPT): stats[match.group(1)] = match.group(2) elif logger not in SKIPPED_LOGGERS and not logger.startswith("httpcore2"): kept.append(line) return kept, stats def main() -> None: for name, port in PORTS.items(): try: socket.create_connection(("127.0.0.1", port), timeout=0.2).close() except OSError: continue raise SystemExit(f"port {port} ({name}) is already in use") work = pathlib.Path(tempfile.mkdtemp(prefix="scrapy-proxy-lab-")) docroot, home, cwd = work / "docroot", work / "home", work / "cwd" for directory in (docroot, docroot / "dir", home, cwd): directory.mkdir() (docroot / "index.html").write_text("lab target ok\n", encoding="utf-8") (docroot / "dir" / "index.html").write_text("redirected ok\n", encoding="utf-8") cert, key = work / "cert.pem", work / "key.pem" subprocess.run(["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "2", "-keyout", str(key), "-out", str(cert), "-subj", "/CN=localhost", "-addext", "subjectAltName=IP:127.0.0.1,DNS:localhost"], check=True, capture_output=True) proxy_log, socks_log, target_log = work / "proxy.jsonl", work / "socks.log", work / "target.jsonl" python = sys.executable quiet = {"stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL} def proxy(port: int, password: str) -> subprocess.Popen: return subprocess.Popen([python, str(HERE / "auth_proxy.py"), str(port), "--user", "user", "--password", password, "--log", str(proxy_log)], stdout=subprocess.DEVNULL) fixtures = [ subprocess.Popen([python, "-m", "http.server", "--bind", "127.0.0.1", "--directory", str(docroot), str(PORTS["http_target"])], **quiet), subprocess.Popen([python, "-m", "http.server", "--bind", "127.0.0.1", "--directory", str(docroot), "--tls-cert", str(cert), "--tls-key", str(key), str(PORTS["https_target"])], **quiet), subprocess.Popen([python, str(HERE / "jobs_target.py"), str(PORTS["jobs_target"]), "--log", str(target_log)], stdout=subprocess.DEVNULL), proxy(PORTS["proxy"], "pass"), proxy(PORTS["proxy_b"], "pass"), proxy(PORTS["proxy_c"], "p@ss:w/rd"), subprocess.Popen([python, "-u", str(HERE / "socks5_server.py"), str(PORTS["socks"])], stdout=socks_log.open("w", encoding="utf-8")), ] settings = [argument for name, value in SETTINGS.items() for argument in ("-s", f"{name}={value}")] venv = str(pathlib.Path(python).parent.parent) results = [] try: for port in PORTS.values(): wait_for(port) # The readiness probes above reach the fixtures; start the per-case logs after them. time.sleep(0.3) _, proxy_offset = new_lines(proxy_log, 0) _, socks_offset = new_lines(socks_log, 0) _, target_offset = new_lines(target_log, 0) for name, env, code in CASES: script = cwd / f"{name}.py" script.write_text(code, encoding="utf-8") command = [python, "-m", "scrapy", "runspider", script.name, *settings] clean_env = {"PATH": "/usr/bin:/bin", "HOME": str(home), "PYTHONPATH": str(HERE), **env} done = subprocess.run(command, cwd=cwd, env=clean_env, capture_output=True, text=True, timeout=120) time.sleep(0.2) proxy_lines, proxy_offset = new_lines(proxy_log, proxy_offset) socks_lines, socks_offset = new_lines(socks_log, socks_offset) target_lines, target_offset = new_lines(target_log, target_offset) stderr = done.stderr.replace(str(cwd) + "/", "").replace(venv, "").replace(str(HERE), "") log, stats = split_log(stderr) results.append({ "case": name, "env": env, "command": "python -m scrapy runspider " + " ".join(command[4:]), "code": code.strip(), "exit_code": done.returncode, "stdout": done.stdout.strip().splitlines(), "log": log, "stats": stats, "proxy_log": [json.loads(line) for line in proxy_lines], "socks_log": socks_lines, "target_log": [json.loads(line) for line in target_lines], }) print(f"{name}: exit {done.returncode}; proxies saw {len(proxy_lines)}, socks saw {len(socks_lines)}, jobs target saw {len(target_lines)}; " f"{' | '.join(done.stdout.strip().splitlines())[:220]}") finally: for fixture in fixtures: fixture.terminate() def version(package: str) -> str: try: return metadata.version(package) except metadata.PackageNotFoundError: return "not installed" versions = {package: version(package) for package in ("Scrapy", "Twisted", "pyOpenSSL", "cryptography", "w3lib", "httpx2", "httpcore2", "socksio")} report = {"recorded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "python": platform.python_version(), "platform": platform.platform(), "versions": versions, "ports": PORTS, "settings": SETTINGS, "lab_common": (HERE / "lab_common.py").read_text(encoding="utf-8"), "cases": results} output = HERE / (sys.argv[1] if len(sys.argv) > 1 else "results.json") output.write_text(json.dumps(report, indent=1, ensure_ascii=False) + "\n", encoding="utf-8") print(f"wrote {output.name}: {len(results)} cases, Scrapy {versions['Scrapy']}, Twisted {versions['Twisted']}, Python {report['python']}") if __name__ == "__main__": main()