460 lines
22 KiB
Python
Executable File
460 lines
22 KiB
Python
Executable File
#!/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()
|