#!/usr/bin/env python3 """Serve pinned public snapshots only while an authenticated native state is fresh.""" from __future__ import annotations import argparse import json import os import pathlib import time from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any STATE_SCHEMA = "guanghu.native-public-projection-state/v1" def load_snapshot(path: pathlib.Path, schema: str, source_commit: str) -> bytes: payload = path.read_bytes() value = json.loads(payload) if not isinstance(value, dict) or value.get("schema") != schema: raise ValueError(f"snapshot schema mismatch: {path}") if schema == "guanghu.public-navigation-anchor/v1": observed_commit = value.get("navigation_source", {}).get("source_commit") else: observed_commit = value.get("commit") if value.get("full_forgejo_equivalence") is not False: raise ValueError("code snapshot cannot claim full Forgejo equivalence") if observed_commit != source_commit: raise ValueError(f"snapshot source commit mismatch: {path}") return payload class Projection: def __init__(self, args: argparse.Namespace) -> None: self.state_file = pathlib.Path(args.state_file) self.expected = { "schema": STATE_SCHEMA, "node_id": args.expected_node, "relay_node": args.expected_relay, "candidate_sha256": args.expected_candidate_sha256, "source_commit": args.expected_source_commit, "transport_authentication": "SSH_FORCED_COMMAND", } self.max_age_seconds = args.max_age_seconds self.anchor = load_snapshot( pathlib.Path(args.anchor_file), "guanghu.public-navigation-anchor/v1", args.expected_source_commit, ) self.code = load_snapshot( pathlib.Path(args.code_file), "guanghu.native-code-channel-read-only/v1", args.expected_source_commit, ) def state(self) -> tuple[bool, str, dict[str, Any] | None]: try: if self.state_file.is_symlink(): return False, "state_file_symlink", None value = json.loads(self.state_file.read_text(encoding="utf-8")) except FileNotFoundError: return False, "state_missing", None except (OSError, json.JSONDecodeError): return False, "state_invalid", None if not isinstance(value, dict): return False, "state_invalid", None if any(value.get(key) != expected for key, expected in self.expected.items()): return False, "state_binding_mismatch", value if value.get("status") != "READY_NATIVE_RESIDENT": return False, "native_not_ready", value accepted_at = value.get("accepted_at_epoch") if not isinstance(accepted_at, (int, float)): return False, "state_time_invalid", value age = time.time() - float(accepted_at) if age < -5: return False, "state_time_in_future", value if age > self.max_age_seconds: return False, "projection_expired", value return True, "ready", value def handler_for(projection: Projection) -> type[BaseHTTPRequestHandler]: class Handler(BaseHTTPRequestHandler): server_version = "GuanghuNativeProjection/1" def log_message(self, _format: str, *_args: object) -> None: return def send_bytes(self, status: int, payload: bytes, content_type: str) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() if self.command != "HEAD": self.wfile.write(payload) def send_json(self, status: int, value: dict[str, object]) -> None: payload = (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() self.send_bytes(status, payload, "application/json; charset=utf-8") def do_HEAD(self) -> None: # noqa: N802 self.do_GET() def do_GET(self) -> None: # noqa: N802 fresh, reason, state = projection.state() if self.path == "/healthz": self.send_json( HTTPStatus.OK, { "schema": "guanghu.native-public-projection-health/v1", "native_projection_fresh": fresh, "reason": reason, "node_id": projection.expected["node_id"], }, ) return if self.path not in { "/api/ai/v1/anchor", "/api/ai/v1/code-channel", }: self.send_json(HTTPStatus.NOT_FOUND, {"error": "not_found"}) return if not fresh: self.send_json( HTTPStatus.SERVICE_UNAVAILABLE, { "schema": "guanghu.native-public-projection-unavailable/v1", "native_projection_fresh": False, "reason": reason, }, ) return assert state is not None payload = ( projection.anchor if self.path == "/api/ai/v1/anchor" else projection.code ) self.send_bytes(HTTPStatus.OK, payload, "application/json; charset=utf-8") def do_POST(self) -> None: # noqa: N802 self.send_json(HTTPStatus.METHOD_NOT_ALLOWED, {"error": "method_not_allowed"}) return Handler def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--listen-host", default="127.0.0.1") parser.add_argument("--listen-port", required=True, type=int) parser.add_argument("--state-file", required=True) parser.add_argument("--anchor-file", required=True) parser.add_argument("--code-file", required=True) parser.add_argument("--expected-node", required=True) parser.add_argument("--expected-relay", required=True) parser.add_argument("--expected-candidate-sha256", required=True) parser.add_argument("--expected-source-commit", required=True) parser.add_argument("--max-age-seconds", type=float, default=15.0) args = parser.parse_args() if args.listen_host not in {"127.0.0.1", "::1"}: raise SystemExit("projection gateway must remain loopback-only") if not 1 <= args.listen_port <= 65535: raise SystemExit("invalid listen port") if args.max_age_seconds <= 0: raise SystemExit("max age must be greater than zero") projection = Projection(args) server = ThreadingHTTPServer((args.listen_host, args.listen_port), handler_for(projection)) server.serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())