"""Local-only regression checks. No third-party source or credentials are used. Run: python tests/gateway_integration.py Requires PHP CLI with ext-curl. The temporary test copy ONLY permits loopback; the real gateway.php is never changed. All subprocesses and fixtures are closed. """ import gzip import http.client import os from pathlib import Path import shutil import socket import subprocess import tempfile import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlencode TOKEN = "local-test-token-" + "a" * 48 PAYLOAD = bytes(range(256)) * 256 observed = [] class Origin(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *_): pass def do_HEAD(self): self.do_GET() def do_GET(self): observed.append((self.path, dict(self.headers))) if self.path in ("/redirect", "/cross", "/forbidden", "/loop"): location = { "/redirect": "/binary", "/cross": f"http://127.0.0.1:{second.server_port}/binary", "/forbidden": "http://localhost:1/private", "/loop": "/loop", }[self.path] self.send_response(302) self.send_header("Location", location) self.send_header("Content-Length", "8") self.end_headers() if self.command != "HEAD": self.wfile.write(b"redirect") return if self.path == "/stream": self.send_response(200) self.send_header("Content-Type", "video/mp2t") self.send_header("Transfer-Encoding", "chunked") self.end_headers() self.wfile.write(b"5\r\nfirst\r\n") self.wfile.flush() time.sleep(1.2) self.wfile.write(b"4\r\nlast\r\n0\r\n\r\n") return if self.path == "/huge": self.send_response(200) for _ in range(20): self.send_header("X-Large", "a" * 4000) self.end_headers() self.close_connection = True return status, data = 200, PAYLOAD if self.path == "/playlist.m3u8": data = b"#EXTM3U\n#EXTINF:5,\nsegment.ts\n" if self.path == "/missing": status, data = 404, b"missing" if self.headers.get("If-None-Match") == '"fixture"': status, data = 304, b"" if self.headers.get("Range") == "bytes=10-19": status, data = 206, PAYLOAD[10:20] if self.path == "/gzip": data = gzip.compress(PAYLOAD) self.send_response(status) self.send_header("Content-Type", "application/vnd.apple.mpegurl" if self.path.endswith("m3u8") else "video/mp2t") self.send_header("Content-Length", str(len(data))) self.send_header("ETag", '"fixture"') self.send_header("Set-Cookie", "must-not-leak=1") self.send_header("Accept-Ranges", "bytes") if self.path == "/gzip": self.send_header("Content-Encoding", "gzip") if status == 206: self.send_header("Content-Range", f"bytes 10-19/{len(PAYLOAD)}") self.end_headers() if self.command != "HEAD": self.wfile.write(data) def unused_port(): with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] first = ThreadingHTTPServer(("127.0.0.1", 0), Origin) second = ThreadingHTTPServer(("127.0.0.1", 0), Origin) for server in (first, second): threading.Thread(target=server.serve_forever, daemon=True).start() php = shutil.which("php") assert php, "PHP CLI must be on PATH" port = unused_port() source = (Path(__file__).resolve().parents[1] / "gateway.php").read_text(encoding="utf-8") source = source.replace("'https://media.example.com:443',", f"'http://127.0.0.1:{first.server_port}', 'http://127.0.0.1:{second.server_port}',") source = source.replace("const FORWARD_UPSTREAM_CREDENTIALS = false;", "const FORWARD_UPSTREAM_CREDENTIALS = true;") count = 0 def check(condition, message): global count assert condition, message count += 1 def request(path="/binary", *, method="GET", headers=None, action="play", token=TOKEN, script="gateway.php", raw_query=None): query = raw_query if raw_query is not None else urlencode({ "url": f"http://127.0.0.1:{first.server_port}{path}", "action": action}) conn = http.client.HTTPConnection("127.0.0.1", port, timeout=8) conn.request(method, f"/{script}?{query}", headers={"X-Gateway-Token": token, **(headers or {})}) response = conn.getresponse() body = response.read() result = response.status, dict((k.lower(), v) for k, v in response.getheaders()), body conn.close() return result try: with tempfile.TemporaryDirectory(prefix="gateway-tests-") as directory: root = Path(directory) (root / "secure.php").write_text(source, encoding="utf-8") test_source = source.replace("function publicIpv4(string $ip): bool\n{", "function publicIpv4(string $ip): bool\n{\n if ($ip === '127.0.0.1') { return true; } // LOCAL TEST ONLY") assert source != test_source (root / "gateway.php").write_text(test_source, encoding="utf-8") env = {**os.environ, "MEDIA_GATEWAY_TOKEN": TOKEN} with (root / "php.log").open("wb") as log: process = subprocess.Popen([php, "-S", f"127.0.0.1:{port}", "-t", directory], env=env, stdout=log, stderr=log, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) try: for attempt in range(50): try: with socket.create_connection(("127.0.0.1", port), timeout=.2): break except OSError: time.sleep(.1) check(request(token="wrong")[0] == 403, "authentication") check(request(script="secure.php")[0] == 403, "production blocks allowlisted private IP") check(request(method="POST")[0] == 405, "method restriction") check(request(raw_query="url[]=bad")[0] == 400, "array input") check(request(raw_query=urlencode({"url": "file:///etc/passwd"}))[0] == 400, "scheme restriction") check(request("/forbidden")[0] == 403, "redirect allowlist") check(request("/loop")[0] == 502, "redirect limit") check(request("/huge")[0] == 502, "header size bound") status, hdr, body = request(headers={"User-Agent": "GatewayFixture/1.0", "Authorization": "Bearer upstream-test", "Cookie": "fixture=1"}) check(status == 200 and body == PAYLOAD, "binary byte integrity") upstream = {k.lower(): v for k, v in observed[-1][1].items()} check(upstream.get("user-agent") == "GatewayFixture/1.0", "user-agent preservation") check(upstream.get("authorization") == "Bearer upstream-test", "opt-in upstream credentials") check("x-gateway-token" not in upstream and "set-cookie" not in hdr, "secret and cookie isolation") status, hdr, body = request(headers={"Range": "bytes=10-19"}) check(status == 206 and body == PAYLOAD[10:20] and "content-range" in hdr, "range preservation") check(request(headers={"If-None-Match": '"fixture"'})[0] == 304, "conditional response") status, hdr, body = request(action="download") check(body == PAYLOAD and hdr["content-disposition"].startswith("attachment;"), "binary download") status, hdr, body = request("/playlist.m3u8", action="download") check(body.endswith(b"segment.ts\n") and 'filename="playlist.m3u8"' in hdr["content-disposition"], "playlist raw download") status, hdr, body = request(method="HEAD", action="download") check(status == 200 and body == b"" and int(hdr["content-length"]) == len(PAYLOAD), "HEAD semantics") check(request("/missing")[0] == 404, "upstream error status") check(request("/redirect")[2] == PAYLOAD, "relative redirect") check(request("/cross", headers={"Authorization": "Bearer upstream-test", "Cookie": "fixture=1"})[2] == PAYLOAD, "cross-origin redirect") upstream = {k.lower(): v for k, v in observed[-1][1].items()} check("authorization" not in upstream and "cookie" not in upstream, "cross-origin credentials stripped") status, hdr, body = request("/gzip") check(hdr["content-encoding"] == "gzip" and gzip.decompress(body) == PAYLOAD and len(body) == int(hdr["content-length"]), "encoded byte integrity") conn = http.client.HTTPConnection("127.0.0.1", port, timeout=8) query = urlencode({"url": f"http://127.0.0.1:{first.server_port}/stream"}) start = time.monotonic() conn.request("GET", f"/gateway.php?{query}", headers={"X-Gateway-Token": TOKEN}) response = conn.getresponse() prefix = response.read(5) elapsed = time.monotonic() - start check(prefix == b"first" and elapsed < 1, "first bytes arrive before upstream completes") check(response.read() == b"last", "chunked stream integrity") conn.close() finally: process.terminate() process.wait(timeout=10) finally: for server in (first, second): server.shutdown() server.server_close() print(f"PASS: {count} checks; local fixtures only; production gateway unchanged.")