"""Expose the IAM-private Cloud Run producer to OBS on loopback only. OBS Browser sources cannot mint Google identity tokens. This relay does that on their behalf, sends the token only over HTTPS to Cloud Run, and accepts browser requests only from this machine. It is a trusted render-host component, not a public or LAN-facing proxy. """ from __future__ import annotations import argparse import base64 import http.client import json import shutil import ssl import subprocess import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import SplitResult, urlsplit, urlunsplit LOOPBACK = "127.0.0.1" DEFAULT_PORT = 8766 _HOP_BY_HOP = { "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade", } _FORWARDED_REQUEST_HEADERS = { "accept", "accept-encoding", "accept-language", "if-modified-since", "if-none-match", "range", "user-agent", } class RelayError(RuntimeError): """A safe-to-display relay startup or upstream error.""" def _jwt_exp(token: str) -> int | None: """Read an ID token's untrusted expiry for caching; Cloud Run still verifies the signature.""" try: encoded = token.split(".", 2)[1] payload = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) return int(json.loads(payload)["exp"]) except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError): return None class GcloudIdentityToken: """Mint and briefly cache an ID token from the signed-in operator's gcloud identity.""" def __init__(self, gcloud: str = "gcloud", *, runner=subprocess.run, clock=time.time): self._gcloud = gcloud self._runner = runner self._clock = clock self._token = "" self._refresh_at = 0.0 self._lock = threading.Lock() def get(self, *, force: bool = False) -> str: with self._lock: now = self._clock() if not force and self._token and now < self._refresh_at: return self._token result = self._runner( [self._gcloud, "auth", "print-identity-token"], capture_output=True, text=True, timeout=30, check=False, ) token = result.stdout.strip() if result.returncode or not token: detail = (result.stderr or "gcloud returned no identity token").strip().splitlines()[-1] raise RelayError(f"Cannot authenticate the private producer: {detail}") expires = _jwt_exp(token) self._token = token self._refresh_at = max(now + 30, (expires - 300) if expires else now + 300) return token def discover_upstream( *, service: str, project: str, region: str, gcloud: str = "gcloud", runner=subprocess.run ) -> str: result = runner( [gcloud, "run", "services", "describe", service, "--project", project, "--region", region, "--format=value(status.url)"], capture_output=True, text=True, timeout=30, check=False, ) upstream = result.stdout.strip().rstrip("/") if result.returncode or not upstream: detail = (result.stderr or "service URL was empty").strip().splitlines()[-1] raise RelayError(f"Cannot locate the private producer: {detail}") parsed = urlsplit(upstream) if parsed.scheme != "https" or not parsed.hostname or parsed.path not in ("", "/"): raise RelayError("The private producer must resolve to a bare HTTPS origin") return upstream class PrivateFeedRelay(ThreadingHTTPServer): """Threaded reverse proxy deliberately fixed to IPv4 loopback.""" allow_reuse_address = True daemon_threads = True def __init__(self, port: int, upstream: str, tokens: GcloudIdentityToken): parsed = urlsplit(upstream.rstrip("/")) if parsed.scheme not in ("http", "https") or not parsed.hostname: raise ValueError("upstream must be an HTTP(S) origin") self.upstream = parsed self.tokens = tokens super().__init__((LOOPBACK, port), RelayHandler) class RelayHandler(BaseHTTPRequestHandler): """Minimal render proxy with no CORS grant and no credential logging.""" protocol_version = "HTTP/1.1" server: PrivateFeedRelay def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API self._proxy(method="GET") def do_HEAD(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API self._proxy(method="HEAD") def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API parsed = urlsplit(self.path) if parsed.path != "/client-stats" or parsed.query: self._error(405, "Only the renderer health report may be written") return if (self.headers.get("Content-Type") or "").split(";", 1)[0].strip() != "application/json": self._error(415, "Renderer health must be JSON") return try: length = int(self.headers.get("Content-Length") or "") except ValueError: length = -1 if not 0 <= length <= 16384: self._error(413, "Renderer health body is missing or too large") return self._proxy(method="POST", body=self.rfile.read(length)) def do_PUT(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API self._error(405, "The private feed relay is read-only") def do_DELETE(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API self._error(405, "The private feed relay is read-only") def _proxy(self, *, method: str, body: bytes | None = None) -> None: if not self._local_request_allowed(): self._error(403, "Loopback origin required") return request_target = urlsplit(self.path) if request_target.scheme or request_target.netloc or not request_target.path.startswith("/"): self._error(400, "Origin-form request target required") return target = urlunsplit(SplitResult("", "", request_target.path, request_target.query, "")) try: connection, response = self._open_upstream(target, method=method, body=body) if response.status in (401, 403): response.close() connection.close() connection, response = self._open_upstream( target, method=method, body=body, refresh=True ) except (OSError, RelayError, http.client.HTTPException) as exc: self._error(502, f"Private producer unavailable: {exc}") return try: self.send_response(response.status, response.reason) for name, value in response.getheaders(): lower = name.lower() if lower in _HOP_BY_HOP or lower in {"access-control-allow-origin", "set-cookie"}: continue if lower == "cache-control": continue self.send_header(name, value) self.send_header("Cache-Control", "no-store") self.send_header("Referrer-Policy", "no-referrer") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Connection", "close") self.end_headers() self.close_connection = True if method != "HEAD": content_type = response.getheader("Content-Type", "").lower() if content_type.startswith("text/event-stream"): self._copy_sse(response) else: self._copy_body(response) except (BrokenPipeError, ConnectionResetError): pass finally: response.close() connection.close() def _local_request_allowed(self) -> bool: expected_port = self.server.server_address[1] host = (self.headers.get("Host") or "").lower() if host not in {f"127.0.0.1:{expected_port}", f"localhost:{expected_port}"}: return False return (self.headers.get("Sec-Fetch-Site") or "").lower() != "cross-site" def _open_upstream( self, target: str, *, method: str, body: bytes | None = None, refresh: bool = False ): upstream = self.server.upstream connection_type = http.client.HTTPSConnection if upstream.scheme == "https" else http.client.HTTPConnection kwargs = {"timeout": 35} if upstream.scheme == "https": kwargs["context"] = ssl.create_default_context() connection = connection_type(upstream.hostname, upstream.port, **kwargs) headers = {name: value for name, value in self.headers.items() if name.lower() in _FORWARDED_REQUEST_HEADERS} headers["Authorization"] = f"Bearer {self.server.tokens.get(force=refresh)}" if body is not None: headers["Content-Type"] = "application/json" headers["Content-Length"] = str(len(body)) connection.request(method, target, body=body, headers=headers) return connection, connection.getresponse() def _copy_sse(self, response: http.client.HTTPResponse) -> None: while line := response.readline(65536): self.wfile.write(line) self.wfile.flush() def _copy_body(self, response: http.client.HTTPResponse) -> None: while chunk := response.read(65536): self.wfile.write(chunk) def _error(self, status: int, message: str) -> None: body = json.dumps({"error": message}).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Connection", "close") self.end_headers() self.close_connection = True if self.command != "HEAD": self.wfile.write(body) def log_message(self, fmt: str, *args) -> None: # Never log a query string: production does not use URL tokens, but a mistaken local URL must # not turn one into a durable credential leak. Identity headers never enter this method. path = urlsplit(self.path).path status = args[1] if len(args) > 1 else "-" size = args[2] if len(args) > 2 else "-" print(f'relay {self.address_string()} "{self.command} {path}" {status} {size}') def _preflight(upstream: str, tokens: GcloudIdentityToken) -> None: parsed = urlsplit(upstream) connection = http.client.HTTPSConnection(parsed.hostname, parsed.port, timeout=30, context=ssl.create_default_context()) try: connection.request("GET", "/livez", headers={"Authorization": f"Bearer {tokens.get()}"}) response = connection.getresponse() response.read() if response.status != 200: raise RelayError(f"Private producer preflight returned HTTP {response.status}") finally: connection.close() def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--port", type=int, default=DEFAULT_PORT) parser.add_argument("--service", default="trap-v3-producer") parser.add_argument("--project", default="apps-viewregatta") parser.add_argument("--region", default="europe-north1") parser.add_argument("--gcloud", default="gcloud") args = parser.parse_args(argv) gcloud = shutil.which(args.gcloud) if not gcloud: parser.error(f"{args.gcloud!r} was not found on PATH") upstream = discover_upstream(service=args.service, project=args.project, region=args.region, gcloud=gcloud) tokens = GcloudIdentityToken(gcloud) _preflight(upstream, tokens) relay = PrivateFeedRelay(args.port, upstream, tokens) port = relay.server_address[1] print(f"Private producer: {upstream}") print(f"OBS programme: http://127.0.0.1:{port}/core/web/index.html?feed=live") print("LAN access: disabled (IPv4 loopback only)") try: relay.serve_forever() except KeyboardInterrupt: pass finally: relay.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())