#!/usr/bin/env python3 """Range-capable loopback fixtures for the real native network playback check.""" from __future__ import annotations import argparse import json import re import subprocess import sys import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, unquote, urlsplit class _Gate: def __init__(self, total: int, prefix_ends: dict[str, float]): self.total = total self.prefix_ends = prefix_ends self.entered = threading.Event() self.release = threading.Event() self.claimed = False self.waiting = 0 self.requests = 0 self.ranges = 0 self.bytes = 0 self.connections = set() class StreamingServer: """Context-managed server; safe to reuse for actual resolver/playback smoke checks.""" headers = { "av.mp4": "combined", "streaming.mp4": "combined", "cache.mp4": "combined", "quality-1080.mp4": "combined", "quality-720.mp4": "combined", "frame-step.mp4": "combined", "video.mp4": "video", "frame-step-video.mp4": "video", "audio.wav": "audio", "frame-step-audio.m4a": "audio", "network-vod.m3u8": "combined", **{f"network-vod-{index}.ts": "combined" for index in range(4)}, } resolved_token = "token=part%2Fone+two%26tail" resolved_secret = "resolved-header-secret" def __init__(self, fixtures: str | Path): self.fixtures = Path(fixtures).resolve() self.lock = threading.Lock() self.gates: dict[str, _Gate] = {} self.stats: dict[str, dict[str, int]] = {} self.prefixes: dict[tuple[str, int], dict[str, float]] = {} self.resolved_stats: dict[str, dict[str, dict[str, int]]] = {} self.sink_stats = {"requests": 0, "credentials": 0} self.stopping = threading.Event() owner = self class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *_args): pass # Media URLs/headers never appear in logs. def do_GET(self): owner._get(self) def handle(self): try: super().handle() except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError): # Cancellation can close a body read or an idle keep-alive connection. self.close_connection = True class SinkHandler(Handler): def do_GET(self): with owner.lock: owner.sink_stats["requests"] += 1 owner.sink_stats["credentials"] += int( self.headers.get("Authorization") is not None or self.headers.get("X-Terebi-Fixture") is not None) owner._reply(self, 403, b"Unrelated origin") self.sink = ThreadingHTTPServer(("127.0.0.1", 0), SinkHandler) self.sink.daemon_threads = True self.sink_thread = threading.Thread(target=self.sink.serve_forever, daemon=True) self.sink_url = f"http://127.0.0.1:{self.sink.server_port}" self.http = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.http.daemon_threads = True self.thread = threading.Thread(target=self.http.serve_forever, daemon=True) self.base_url = f"http://127.0.0.1:{self.http.server_port}" def __enter__(self): self.thread.start() self.sink_thread.start() return self def __exit__(self, *_args): self.stopping.set() with self.lock: for gate in self.gates.values(): gate.release.set() self.http.shutdown() self.http.server_close() self.thread.join() self.sink.shutdown() self.sink.server_close() self.sink_thread.join() @staticmethod def _reply(request, status: int, body: bytes, content_type="application/json"): request.send_response(status) request.send_header("Content-Type", content_type) request.send_header("Content-Length", str(len(body))) request.send_header("Connection", "close") request.end_headers() request.wfile.write(body) request.close_connection = True @staticmethod def _range(value: str | None, size: int): if value is None: return 0, size - 1 match = re.fullmatch(r"bytes=(\d*)-(\d*)", value) if not match or not any(match.groups()): return None first, last = match.groups() if not first: suffix = int(last) return (max(0, size - suffix), size - 1) if suffix else None start = int(first) end = min(int(last), size - 1) if last else size - 1 return (start, end) if 0 <= start <= end < size else None def _prefix_ends(self, name: str, offset: int) -> dict[str, float]: """Timestamp upper bounds, never a byte fraction interpreted as playback time.""" key = (name, offset) if key not in self.prefixes: probe = subprocess.run( ["ffprobe", "-v", "error", "-show_packets", "-show_streams", "-show_format", "-show_entries", "format=start_time:stream=index,codec_type:packet=stream_index,pts_time,duration_time,pos,size", "-of", "json", str(self.fixtures / name)], capture_output=True, text=True, check=True, timeout=30, ) metadata = json.loads(probe.stdout) tracks = {stream["index"]: stream["codec_type"] for stream in metadata["streams"]} origin = float(metadata.get("format", {}).get("start_time", 0)) ends = {"audio": 0.0, "video": 0.0} first_missing = {"audio": float("inf"), "video": float("inf")} for packet in metadata["packets"]: kind = tracks.get(packet["stream_index"]) if kind not in ends or "pts_time" not in packet or "pos" not in packet: continue # A packet crossing the gate is not playable, even if its header arrived. if int(packet["pos"]) + int(packet["size"]) <= offset: end = float(packet["pts_time"]) - origin + float(packet.get("duration_time", 0)) ends[kind] = max(ends[kind], end) else: first_missing[kind] = min(first_missing[kind], max(0, float(packet["pts_time"]) - origin)) # A reordered reference packet's later PTS cannot bridge an undelivered B-frame. self.prefixes[key] = {kind: min(end, first_missing[kind]) for kind, end in ends.items()} return self.prefixes[key] def _timestamp_offset(self, name: str, seconds: float) -> int: probe = subprocess.run( ["ffprobe", "-v", "error", "-show_packets", "-show_format", "-show_entries", "format=start_time:packet=pts_time,pos", "-of", "json", str(self.fixtures / name)], capture_output=True, text=True, check=True, timeout=30, ) metadata = json.loads(probe.stdout) origin = float(metadata.get("format", {}).get("start_time", 0)) return min(int(packet["pos"]) for packet in metadata["packets"] if "pos" in packet and "pts_time" in packet and float(packet["pts_time"]) - origin >= seconds) def _resolved(self, request, parsed, parts): scenario = parts[1] relative = "/".join(parts[2:]) kind = ("master" if relative == "master.m3u8" else "variant" if relative == "nested/media.m3u8" else "key" if relative == "keys/key.bin" else "init" if relative in ("init.mp4", "init.jpeg") else "segment") with self.lock: stats = self.resolved_stats.setdefault(scenario, {}).setdefault( kind, {"requests": 0, "denied": 0, "redirects": 0}) stats["requests"] += 1 denied = (parsed.query != self.resolved_token or request.headers.get("X-Terebi-Fixture") != "combined" or request.headers.get("Authorization") != "Bearer " + self.resolved_secret) stats["denied"] += int(denied) if denied: self._reply(request, 403, b"Required source credentials or complete token missing") return if scenario == "redirect-" + kind or scenario == "redirect-same-master" and kind == "master": with self.lock: stats["redirects"] += 1 request.send_response(302) request.send_header("Location", self.base_url + "/media/network-vod.m3u8" if scenario == "redirect-same-master" else self.sink_url + "/leak") request.send_header("Content-Length", "0") request.send_header("Connection", "close") request.end_headers() request.close_connection = True return fragmented = scenario in ("fmp4", "image-fmp4", "redirect-init", "external-init") image_named = scenario.startswith("image-") directory = self.fixtures / "resolved-hls" / ("fmp4" if fragmented else "encrypted") def uri(path, resource_kind): if image_named and resource_kind in ("segment", "init"): path = path.rsplit(".", 1)[0] + (".jpeg" if fragmented else ".jpg") address = path + "?" + self.resolved_token if scenario == "external-" + resource_kind: return self.sink_url + "/leak/" + address return address if kind == "master": body = ("#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=500000\n" + uri("nested/media.m3u8", "variant") + "\n").encode() elif kind == "variant": playlist = (directory / "media.m3u8").read_text() playlist = re.sub(r'URI="[^"]+"', lambda match: 'URI="' + uri( "../init.mp4" if fragmented else "../keys/key.bin", "init" if fragmented else "key") + '"', playlist) playlist = re.sub(r"(?m)^(seg-[^\n]+)$", lambda match: uri("../segments/" + match[1], "segment"), playlist) body = playlist.encode() else: name = relative.removeprefix("segments/").removeprefix("keys/") # Resource routing is deliberately finite; never serve arbitrary paths. if not re.fullmatch(r"key\.bin|init\.(?:mp4|jpeg)|seg-\d+\.(?:ts|m4s|jpg|jpeg)", name): self._reply(request, 404, b"Unknown HLS resource") return if image_named: name = (name.rsplit(".", 1)[0] + ".mp4" if kind == "init" else name.rsplit(".", 1)[0] + (".m4s" if fragmented else ".ts") if kind == "segment" else name) body = (directory / name).read_bytes() self._reply(request, 200, body, "application/vnd.apple.mpegurl" if kind in ("master", "variant") else "image/jpeg" if image_named and kind in ("segment", "init") else "application/octet-stream") def _get(self, request): parsed = urlsplit(request.path) parts = unquote(parsed.path).strip("/").split("/") if parts == ["control", "resolved-stats"]: with self.lock: body = json.dumps({"sources": self.resolved_stats, "sink": self.sink_stats}).encode() self._reply(request, 200, body) return if len(parts) >= 3 and parts[0] == "resolved": self._resolved(request, parsed, parts) return if parts == ["youtube-redirect", "av.mp4"]: request.send_response(302) request.send_header("Location", self.base_url + "/media/av.mp4") request.send_header("Content-Length", "0") request.send_header("Connection", "close") request.end_headers() request.close_connection = True return if parts == ["control", "stats"]: with self.lock: body = json.dumps(self.stats).encode() self._reply(request, 200, body) return if len(parts) == 3 and parts[0] == "control" and parts[1] in ("status", "release"): with self.lock: gate = self.gates.get(parts[2]) if gate is None: body = {"entered": False, "released": False, "waiting": 0, "requests": 0, "ranges": 0, "bytes": 0, "total": 0, "connections": 0, "prefix_ends": {}} else: if parts[1] == "release": gate.release.set() body = {"entered": gate.entered.is_set(), "released": gate.release.is_set(), "waiting": gate.waiting, "requests": gate.requests, "ranges": gate.ranges, "bytes": gate.bytes, "total": gate.total, "connections": len(gate.connections), "prefix_ends": gate.prefix_ends} self._reply(request, 200, json.dumps(body).encode()) return mode = parts[0] token = None if len(parts) == 3 and mode in ("stall", "stall-open", "chunked-media", "disconnect"): token, name = parts[1:] if not re.fullmatch(r"[A-Za-z0-9_-]+", token): self._reply(request, 400, b"Invalid gate token", "text/plain") return elif len(parts) == 2 and mode in ("media", "no-range", "failure"): name = parts[1] else: self._reply(request, 404, b"Unknown fixture endpoint", "text/plain") return if name not in self.headers or not (self.fixtures / name).is_file(): self._reply(request, 404, b"Fixture not found", "text/plain") return with self.lock: stats = self.stats.setdefault(name, {"requests": 0, "ranges": 0, "bytes": 0, "denied": 0}) stats["requests"] += 1 if request.headers.get("X-Terebi-Fixture") != self.headers[name]: stats["denied"] += 1 denied = True else: denied = False if denied: self._reply(request, 403, b"Required fixture header missing or incorrect", "text/plain") return if mode == "failure": self._reply(request, 503, b"Controlled fixture extraction failure", "text/plain") return path = self.fixtures / name size = path.stat().st_size range_header = request.headers.get("Range") if mode != "no-range" else None byte_range = self._range(range_header, size) if byte_range is None: request.send_response(416) request.send_header("Content-Range", f"bytes */{size}") request.send_header("Content-Length", "0") request.send_header("Connection", "close") request.end_headers() request.close_connection = True return start, end = byte_range gate = None gate_at = 0 should_gate = False if token is not None: query = parse_qs(parsed.query) try: gate_at = (self._timestamp_offset(name, float(query["after_seconds"][0])) if mode in ("stall", "chunked-media", "disconnect") and "after_seconds" in query else int(query.get("after", ["131072"])[0]) if mode in ("stall", "chunked-media", "disconnect") else start) except (ValueError, OverflowError): self._reply(request, 400, b"Invalid gate offset", "text/plain") return with self.lock: if token not in self.gates: prefix = self._prefix_ends(name, gate_at) if mode in ("stall", "chunked-media", "disconnect") else {} self.gates[token] = _Gate(size, prefix) gate = self.gates[token] gate.requests += 1 gate.connections.add(request.client_address) gate.ranges += int(range_header is not None) # Only the original prefix connection is gated: a seek can create a new # range request while the old read is still blocked and never released. # Simulate a CDN that throttles oversized/open-ended ranges, while # finite requests of at most 10 MiB run at full available speed. oversized = mode == "chunked-media" and ( range_header is None or not range_header.split("-", 1)[1] or int(range_header.split("-", 1)[1]) - start + 1 > 10 * 1024 * 1024) should_gate = (mode != "chunked-media" or oversized) and not gate.claimed and start <= gate_at <= end if should_gate: gate.claimed = True with self.lock: stats["ranges"] += int(range_header is not None) request.send_response(206 if range_header is not None else 200) request.send_header("Accept-Ranges", "none" if mode == "no-range" else "bytes") content_types = {".wav": "audio/wav", ".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t"} request.send_header("Content-Type", content_types.get(path.suffix, "video/mp4")) request.send_header("Content-Length", str(end - start + 1)) if range_header is not None: request.send_header("Content-Range", f"bytes {start}-{end}/{size}") persistent = mode == "chunked-media" and parse_qs(parsed.query).get("keep_alive") == ["1"] request.send_header("Connection", "keep-alive" if persistent else "close") request.end_headers() request.wfile.flush() request.close_connection = not persistent if (mode == "disconnect" and not should_gate and start <= gate_at and query.get("fail_reconnect") == ["1"]): return # Every resumed response at the broken offset is also truncated. cursor = start with path.open("rb") as media: media.seek(start) while cursor <= end and not self.stopping.is_set(): if should_gate and cursor == gate_at: with self.lock: gate.waiting += 1 gate.entered.set() try: gate.release.wait() # No timer silently unblocks a cancellation check. finally: with self.lock: gate.waiting -= 1 should_gate = False if mode == "disconnect": return # Advertise the complete range, then cut its body short. amount = min(16384, end - cursor + 1) if should_gate: amount = min(amount, gate_at - cursor) chunk = media.read(amount) if not chunk: break request.wfile.write(chunk) request.wfile.flush() cursor += len(chunk) with self.lock: stats["bytes"] += len(chunk) if gate is not None: gate.bytes += len(chunk) def resolved_fixtures(fixtures: Path): """Generate deterministic AES-128 TS and fMP4 fixtures without external services.""" for fragmented in (False, True): directory = fixtures.resolve() / "resolved-hls" / ("fmp4" if fragmented else "encrypted") directory.mkdir(parents=True, exist_ok=True) command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(fixtures.resolve() / "av.mp4"), "-c:v", "libx264", "-g", "24", "-keyint_min", "24", "-sc_threshold", "0", "-c:a", "aac", "-hls_time", "1", "-hls_playlist_type", "vod"] if fragmented: command += ["-hls_segment_type", "fmp4", "-hls_fmp4_init_filename", "init.mp4"] else: (directory / "key.bin").write_bytes(bytes(range(16))) (directory / "keyinfo").write_text("key.bin\n" + str(directory / "key.bin") + "\n000102030405060708090a0b0c0d0e0f\n") command += ["-hls_key_info_file", str(directory / "keyinfo")] command += ["-hls_segment_filename", str(directory / ("seg-%d.m4s" if fragmented else "seg-%d.ts")), str(directory / "media.m3u8")] subprocess.run(command, check=True, timeout=30) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("binary", type=Path) parser.add_argument("fixtures", type=Path) args = parser.parse_args() required = ("av.mp4", "video.mp4", "audio.wav", "frame-step-video.mp4", "frame-step-audio.m4a", "streaming.mp4", "cache.mp4", "network-vod.m3u8") missing = [name for name in required if not (args.fixtures / name).is_file()] if missing: parser.error("Missing streaming fixtures: " + ", ".join(missing)) resolved_fixtures(args.fixtures) with StreamingServer(args.fixtures) as server: try: result = subprocess.run([str(args.binary.resolve()), server.base_url, str(args.fixtures.resolve())], timeout=180, check=False, capture_output=True) except subprocess.TimeoutExpired: raise SystemExit("FAIL: native network check exceeded its infrastructure deadline") private = (StreamingServer.resolved_token.encode(), StreamingServer.resolved_secret.encode(), b"Bearer ", server.base_url.encode(), server.sink_url.encode()) if any(secret in result.stdout or secret in result.stderr for secret in private): raise SystemExit("FAIL: native diagnostics exposed source URL or credentials") print(result.stdout.decode(), end="") print(result.stderr.decode(), end="", file=sys.stderr) if server.sink_stats["requests"] or server.sink_stats["credentials"]: raise SystemExit("FAIL: resolved playback contacted an unrelated origin") raise SystemExit(result.returncode) if __name__ == "__main__": main()