scripts/nginx-test/anilist_stub.py
Rama main · réplica de lectura del repositorio original.
"""Stand-in for AniList's image CDN (s4.anilist.co) in the nginx harness.
:443 speaks TLS 1.3 with a certificate for s4.anilist.co that the harness's
test root CA issued through two intermediate CAs; the harness mounts that
root into nginx as its only trusted CA. Session tickets are off, so no
connection can resume an earlier session: every connection shows nginx a
certificate to check, including right after the certificate is switched.
GET /file/anilistcdn/<path> answers 200 with "original <path> v<version>"
and a Last-Modified that moves with the version; an If-Modified-Since at or
after it gets a 304. Both carry every header that would stop a naive cache
from storing them (Cache-Control: no-store, an Expires in the past, Vary: *,
Set-Cookie, X-Accel-Expires: 0, X-Accel-Redirect) and headers that must never
reach our visitors (Set-Cookie, Strict-Transport-Security, Report-To, NEL).
A 200 also carries the CDN's notes about itself, as the real one does: how
old its own copy is (Age), its request and cache ids, and its storage
backend's file metadata. Error answers carry headers that invite a cache to
keep them.
:9000 is plain HTTP, for the harness. <path> is the part after
/file/anilistcdn/.
/set?path=<path>[&status=N][&retry_after=S][&version=N][&delay=S]
/all?status=N answer every path with N (0: back to per-path rules)
/listen?on=0|1 stop, or resume, accepting connections on :443
/cert?mode=good|untrusted|wrong_name
/hits[?path=<path>] requests answered, in all or for one path
/last?field=F from the last request: target, sni, status,
header_names, or a request header (lower case)
/targets every request target seen, one per line
"""
import email.utils
import socket
import ssl
import threading
import time
from collections.abc import Callable
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
PREFIX = "/file/anilistcdn/"
CERT_DIR = "/certs"
# Last-Modified of version n is n days after this (2026-01-01T00:00:00Z).
FIRST_MODIFIED = 1767225600
DAY = 86400
STORE_REPELLENT = (
("Cache-Control", "private, no-store"),
("Expires", "Thu, 01 Jan 1970 00:00:00 GMT"),
("Vary", "*"),
("Set-Cookie", "anilist_test=1; Path=/"),
("X-Accel-Expires", "0"),
("X-Accel-Redirect", "/maintenance.html"),
("Strict-Transport-Security", "max-age=1"),
("Report-To", '{"group":"cf-nel","max_age":1,"endpoints":[{"url":"https://reports.invalid/"}]}'),
("NEL", '{"report_to":"cf-nel","max_age":1}'),
)
CDN_BOOKKEEPING = (
("Age", "1171085"),
("CF-Ray", "0123456789abcdef-SYD"),
("CF-Cache-Status", "HIT"),
("X-Bz-Upload-Timestamp", "1743617779515"),
("x-bz-content-sha1", "1e43bfb5e49b2c7a40fc0270be87e7a10b853164"),
("x-bz-file-id", "4_ztest_ftest"),
("x-bz-file-name", "media/anime/cover/large/test.png"),
("x-bz-info-src_last_modified_millis", "1743617779240"),
)
ERROR_CACHEABLE = (
("Cache-Control", "public, max-age=3600"),
("X-Accel-Expires", "3600"),
)
Headers = tuple[tuple[str, str], ...]
lock = threading.Lock()
rules: dict[str, dict[str, float]] = {}
hits_by_path: dict[str, int] = {}
targets: list[str] = []
last: dict[str, str] = {}
state = {"all": 0, "cert": "good", "sni": "", "hits": 0}
def tls_context(name: str) -> ssl.SSLContext:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
ctx.num_tickets = 0
ctx.options |= ssl.OP_NO_TICKET
ctx.load_cert_chain(f"{CERT_DIR}/{name}-chain.pem", f"{CERT_DIR}/{name}.key")
return ctx
CONTEXTS = {mode: tls_context(mode) for mode in ("good", "untrusted", "wrong_name")}
LISTENING = tls_context("good")
def choose_certificate(conn: ssl.SSLSocket, server_name: str | None, _: ssl.SSLContext) -> None:
with lock:
state["sni"] = server_name or ""
mode = state["cert"]
conn.context = CONTEXTS[mode]
LISTENING.sni_callback = choose_certificate
class TLSServer(ThreadingHTTPServer):
daemon_threads = True
request_queue_size = 256
def finish_request(self, request: socket.socket, client_address: tuple) -> None:
# The handshake runs here, in the connection's own thread, so a slow
# or refused one never holds up the accept loop.
with lock:
state["sni"] = ""
try:
conn = LISTENING.wrap_socket(request, server_side=True)
except OSError:
return # nginx refused the certificate, or went away
try:
self.RequestHandlerClass(conn, client_address, self)
finally:
conn.close()
class CDN(BaseHTTPRequestHandler):
def log_message(self, *args: object) -> None:
pass
def do_GET(self) -> None:
target = self.path
path = urlsplit(target).path
name = path[len(PREFIX):] if path.startswith(PREFIX) else path
with lock:
state["hits"] += 1
hits_by_path[name] = hits_by_path.get(name, 0) + 1
targets.append(target)
last.clear()
last.update({k.lower(): v for k, v in self.headers.items()})
last["target"] = target
last["sni"] = str(state["sni"])
last["header_names"] = " ".join(sorted({k.lower() for k in self.headers.keys()}))
rule = dict(rules.get(name, {}))
status = int(state["all"]) or int(rule.get("status", 200))
time.sleep(rule.get("delay", 0))
if status == 200:
self.answer_original(name, int(rule.get("version", 1)))
else:
self.answer_error(status, int(rule.get("retry_after", 0)))
def answer_original(self, name: str, version: int) -> None:
modified_at = FIRST_MODIFIED + version * DAY
modified = email.utils.formatdate(modified_at, usegmt=True)
since = self.headers.get("If-Modified-Since")
if since and email.utils.parsedate_to_datetime(since).timestamp() >= modified_at:
self.answer(304, b"", (("Last-Modified", modified),) + STORE_REPELLENT)
return
body = f"original {name} v{version}\n".encode()
headers = (("Content-Type", "image/png"), ("Last-Modified", modified))
self.answer(200, body, headers + STORE_REPELLENT + CDN_BOOKKEEPING)
def answer_error(self, status: int, retry_after: int) -> None:
headers: Headers = (("Content-Type", "text/plain"),) + ERROR_CACHEABLE
if retry_after:
headers += (("Retry-After", str(retry_after)),)
self.answer(status, f"stub {status}\n".encode(), headers)
def answer(self, status: int, body: bytes, headers: Headers) -> None:
with lock:
last["status"] = str(status)
self.send_response(status)
for key, value in headers:
self.send_header(key, value)
if status != 304:
self.send_header("Content-Length", str(len(body)))
self.end_headers()
if status != 304:
self.wfile.write(body)
tls_server: TLSServer | None = None
listen_lock = threading.Lock()
def listen(on: bool) -> None:
global tls_server
with listen_lock:
if on and tls_server is None:
tls_server = TLSServer(("0.0.0.0", 443), CDN)
threading.Thread(
target=tls_server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
).start()
elif not on and tls_server is not None:
tls_server.shutdown()
tls_server.server_close()
tls_server = None
def set_rule(q: dict[str, str]) -> str:
update = {k: int(q[k]) for k in ("status", "retry_after", "version") if k in q}
if "delay" in q:
update["delay"] = float(q["delay"])
with lock:
rules[q["path"]] = {**rules.get(q["path"], {}), **update}
return "ok"
def set_all(q: dict[str, str]) -> str:
with lock:
state["all"] = int(q["status"])
return "ok"
def set_listen(q: dict[str, str]) -> str:
listen(q["on"] == "1")
return "ok"
def set_cert(q: dict[str, str]) -> str:
if q["mode"] not in CONTEXTS:
raise ValueError(f"unknown certificate mode {q['mode']}")
with lock:
state["cert"] = q["mode"]
return "ok"
def get_hits(q: dict[str, str]) -> str:
with lock:
return str(hits_by_path.get(q["path"], 0) if "path" in q else state["hits"])
def get_last(q: dict[str, str]) -> str:
with lock:
return last.get(q["field"].lower(), "")
def get_targets(_: dict[str, str]) -> str:
with lock:
return "\n".join(targets)
COMMANDS: dict[str, Callable[[dict[str, str]], str]] = {
"set": set_rule,
"all": set_all,
"listen": set_listen,
"cert": set_cert,
"hits": get_hits,
"last": get_last,
"targets": get_targets,
}
class Control(BaseHTTPRequestHandler):
def log_message(self, *args: object) -> None:
pass
def do_GET(self) -> None:
url = urlsplit(self.path)
query = {k: v[0] for k, v in parse_qs(url.query).items()}
command = COMMANDS.get(url.path.strip("/"))
if command is None:
self.reply(404, "unknown command\n")
return
try:
self.reply(200, command(query) + "\n")
except (KeyError, ValueError) as error:
self.reply(400, f"bad request: {error}\n")
def reply(self, status: int, text: str) -> None:
body = text.encode()
self.send_response(status)
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
listen(True)
ThreadingHTTPServer(("0.0.0.0", 9000), Control).serve_forever()