from __future__ import annotations import hashlib import ipaddress import json import os import re import shutil import stat import tempfile import time import urllib.error import urllib.parse import urllib.request import uuid from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from account_registry import Account, AuthorityBinding, AuthorityMode from health_client import AccountSnapshot, HealthStatus _ENDPOINTS_FILE = "authority_endpoints.json" _FORBIDDEN_AUTH_FILES = { "auth.json", ".credentials.json", "claude.json", } _PROVIDER_ENV_PREFIXES = ( "ANTHROPIC_", "AWS_", "AZURE_", "CHATGPT_", "CLAUDE_", "CODEX_", "GOOGLE_", "OPENAI_", "SUBROUTER_", ) _PROVIDER_ENV_NAMES = { "ALL_PROXY", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "all_proxy", "http_proxy", "https_proxy", "no_proxy", } class AuthorityConfigurationError(RuntimeError): pass @dataclass(frozen=True) class AuthorityEndpoint: name: str origin: str @dataclass(frozen=True) class GatewayStatus: state: str detail: str checked_at: float grant_expires_at_ms: int | None = None @dataclass(frozen=True) class AuthorityDataPlane: responses_url: str proxy_key: str grant_expires_at_ms: int sanitized_status: GatewayStatus @dataclass(frozen=True) class AuthorityLaunch: executable: str argv: tuple[str, ...] environment: dict[str, str] gateway_home: Path sanitized_status: GatewayStatus def _fingerprint(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest()[:12] def _atomic_write(path: Path, body: str, mode: int) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temporary_path = Path(temporary) try: os.fchmod(fd, mode) with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(body) handle.flush() os.fsync(handle.fileno()) os.replace(temporary_path, path) directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(directory_fd) finally: os.close(directory_fd) finally: temporary_path.unlink(missing_ok=True) def _sanitized_environment(source: dict[str, str]) -> dict[str, str]: return { name: value for name, value in source.items() if name not in _PROVIDER_ENV_NAMES and not name.startswith(_PROVIDER_ENV_PREFIXES) } def build_credentialless_info_environment( base_dir: Path, tool: str, *, inherited_environment: dict[str, str] | None = None, ) -> dict[str, str]: if tool not in {"claude", "codex"}: raise AuthorityConfigurationError("Gateway: migration required") root = base_dir / "credentialless-info-homes" / tool _materialize_immutable_home(root, {}) environment = _sanitized_environment( dict(os.environ if inherited_environment is None else inherited_environment) ) environment["HOME"] = str(root) if tool == "codex": environment["CODEX_HOME"] = str(root) environment["CODEX_SQLITE_HOME"] = str(root / "sqlite") else: environment["CLAUDE_CONFIG_DIR"] = str(root) return environment def _open_protected_parent(path: Path) -> int: descriptor = os.open( "/", os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW, ) try: for component in path.parent.parts[1:]: next_descriptor = os.open( component, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW, dir_fd=descriptor, ) os.close(descriptor) descriptor = next_descriptor info = os.fstat(descriptor) if stat.S_IMODE(info.st_mode) & 0o022: raise AuthorityConfigurationError("Gateway: migration required") if os.fstat(descriptor).st_uid not in {0, os.geteuid()}: raise AuthorityConfigurationError("Gateway: migration required") return descriptor except Exception: os.close(descriptor) raise def _read_protected_text(path: Path, *, max_bytes: int, private: bool) -> str: if not path.is_absolute() or path.name in {"", ".", ".."}: raise AuthorityConfigurationError("Gateway: migration required") parent_fd = -1 file_fd = -1 try: parent_fd = _open_protected_parent(path) file_fd = os.open( path.name, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, dir_fd=parent_fd, ) info = os.fstat(file_fd) if ( not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.geteuid() ): raise AuthorityConfigurationError("Gateway: migration required") forbidden_mode = 0o077 if private else 0o022 if stat.S_IMODE(info.st_mode) & forbidden_mode: raise AuthorityConfigurationError("Gateway: migration required") if info.st_size <= 0 or info.st_size > max_bytes: raise AuthorityConfigurationError("Gateway: migration required") body = bytearray() while len(body) <= max_bytes: chunk = os.read(file_fd, min(4096, max_bytes + 1 - len(body))) if not chunk: break body.extend(chunk) if not body or len(body) > max_bytes: raise AuthorityConfigurationError("Gateway: migration required") return bytes(body).decode("utf-8") except AuthorityConfigurationError: raise except (OSError, UnicodeDecodeError) as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc finally: if file_fd >= 0: os.close(file_fd) if parent_fd >= 0: os.close(parent_fd) LOCAL_AUTHORITY_NAME = "subrouter-primary" LOCAL_AUTHORITY_ORIGIN = "http://127.0.0.1:31415" def ensure_local_authority_endpoint( base_dir: Path, *, authority_name: str = LOCAL_AUTHORITY_NAME, origin: str = LOCAL_AUTHORITY_ORIGIN, ) -> AuthorityEndpoint: """Ensure the workstation-local Subrouter endpoint is registered for Systray. Systray is the account registry. This file contains only non-secret endpoint metadata and never provider credentials or proxy grants. """ if authority_name != LOCAL_AUTHORITY_NAME or origin != LOCAL_AUTHORITY_ORIGIN: raise AuthorityConfigurationError("Gateway: migration required") path = base_dir / _ENDPOINTS_FILE authorities: dict[str, object] = {} if path.exists(): try: payload = json.loads(_read_protected_text(path, max_bytes=65_536, private=False)) except (AuthorityConfigurationError, json.JSONDecodeError) as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc if not isinstance(payload, dict) or set(payload) != {"authorities"}: raise AuthorityConfigurationError("Gateway: migration required") raw = payload.get("authorities") if not isinstance(raw, dict): raise AuthorityConfigurationError("Gateway: migration required") authorities = dict(raw) existing = authorities.get(authority_name) if existing is not None and existing != {"origin": origin}: raise AuthorityConfigurationError("Gateway: migration required") authorities[authority_name] = {"origin": origin} _atomic_write( path, json.dumps({"authorities": authorities}, indent=2, sort_keys=True) + "\n", 0o600, ) return AuthorityEndpoint(authority_name, origin) def resolve_authority_endpoint(base_dir: Path, binding: AuthorityBinding) -> AuthorityEndpoint: path = base_dir / _ENDPOINTS_FILE try: payload = json.loads(_read_protected_text(path, max_bytes=65_536, private=False)) except AuthorityConfigurationError: raise except json.JSONDecodeError as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc if not isinstance(payload, dict) or set(payload) != {"authorities"}: raise AuthorityConfigurationError("Gateway: migration required") authorities = payload["authorities"] if not isinstance(authorities, dict): raise AuthorityConfigurationError("Gateway: migration required") record = authorities.get(binding.authority_name) if not isinstance(record, dict) or set(record) != {"origin"}: raise AuthorityConfigurationError("Gateway: migration required") origin = record["origin"] if not isinstance(origin, str): raise AuthorityConfigurationError("Gateway: migration required") parsed = urllib.parse.urlsplit(origin) try: _port = parsed.port except ValueError as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc if ( parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username is not None or parsed.password is not None ): raise AuthorityConfigurationError("Gateway: migration required") if parsed.scheme == "http": hostname = parsed.hostname.rstrip(".").lower() try: loopback = ipaddress.ip_address(hostname).is_loopback except ValueError: loopback = hostname == "localhost" if not loopback: raise AuthorityConfigurationError("Gateway: migration required") if parsed.query or parsed.fragment or parsed.path not in {"", "/"}: raise AuthorityConfigurationError("Gateway: migration required") return AuthorityEndpoint(binding.authority_name, origin.rstrip("/")) def _read_proxy_key(path: Path) -> str: value = _read_protected_text(path, max_bytes=4096, private=True).strip() if not value or any(character.isspace() for character in value): raise AuthorityConfigurationError("Gateway: migration required") return value def _route_base(endpoint: AuthorityEndpoint, binding: AuthorityBinding, tool: str) -> str: route = urllib.parse.quote(binding.route_id, safe="") suffix = "/v1" if tool == "codex" else "" return f"{endpoint.origin}/r/{route}{suffix}" def _gateway_root(base_dir: Path, account: Account, binding: AuthorityBinding) -> Path: identity = _fingerprint(f"{account.tray_key}\0{binding.authority_name}\0{binding.route_id}") return base_dir / "gateway-homes" / account.tool / identity def _validate_gateway_root(root: Path, expected: dict[str, str]) -> None: root_fd = -1 try: root_fd = os.open( root, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW, ) except FileNotFoundError: return except OSError as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc try: root_info = os.fstat(root_fd) if ( not stat.S_ISDIR(root_info.st_mode) or root_info.st_uid != os.geteuid() or stat.S_IMODE(root_info.st_mode) != 0o700 ): raise AuthorityConfigurationError("Gateway: migration required") try: names = os.listdir(root_fd) except OSError as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc if set(names) != set(expected): raise AuthorityConfigurationError("Gateway: migration required") for name in names: if name in _FORBIDDEN_AUTH_FILES: raise AuthorityConfigurationError("Gateway: migration required") file_fd = -1 try: file_fd = os.open( name, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW, dir_fd=root_fd, ) info = os.fstat(file_fd) if ( not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o600 or info.st_size > 65_536 ): raise AuthorityConfigurationError("Gateway: migration required") body_bytes = bytearray() while len(body_bytes) <= 65_536: chunk = os.read(file_fd, min(4096, 65_537 - len(body_bytes))) if not chunk: break body_bytes.extend(chunk) if len(body_bytes) > 65_536: raise AuthorityConfigurationError("Gateway: migration required") body = bytes(body_bytes).decode("utf-8") except AuthorityConfigurationError: raise except (OSError, UnicodeDecodeError) as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc finally: if file_fd >= 0: os.close(file_fd) if body != expected[name]: raise AuthorityConfigurationError("Gateway: migration required") try: value = json.loads(body) except json.JSONDecodeError: continue if _contains_token_material(value): raise AuthorityConfigurationError("Gateway: migration required") finally: os.close(root_fd) def _contains_token_material(value: object) -> bool: if isinstance(value, dict): return any( (isinstance(key, str) and any(term in key.lower() for term in ("token", "secret", "password", "refresh"))) or _contains_token_material(nested) for key, nested in value.items() ) if isinstance(value, list): return any(_contains_token_material(item) for item in value) return False def _materialize_immutable_home(root: Path, files: dict[str, str]) -> Path: parent = root.parent try: parent.mkdir(parents=True, exist_ok=True, mode=0o700) parent_info = parent.lstat() except OSError as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc if ( stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode) or parent_info.st_uid != os.geteuid() or stat.S_IMODE(parent_info.st_mode) & 0o022 ): raise AuthorityConfigurationError("Gateway: migration required") if root.exists() or root.is_symlink(): _validate_gateway_root(root, files) return root staging = Path(tempfile.mkdtemp(prefix=f".{root.name}.", dir=parent)) try: os.chmod(staging, 0o700) for name, body in files.items(): _atomic_write(staging / name, body, 0o600) _validate_gateway_root(staging, files) try: os.rename(staging, root) except FileExistsError: _validate_gateway_root(root, files) directory_fd = os.open(parent, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC) try: os.fsync(directory_fd) finally: os.close(directory_fd) except AuthorityConfigurationError: raise except OSError as exc: raise AuthorityConfigurationError("Gateway: migration required") from exc finally: if staging.exists(): shutil.rmtree(staging) _validate_gateway_root(root, files) return root def materialize_gateway_home( base_dir: Path, account: Account, endpoint: AuthorityEndpoint, binding: AuthorityBinding, ) -> Path: root = _gateway_root(base_dir, account, binding) allowed = {"gateway-manifest.json"} if account.tool == "codex": allowed.add("config.toml") manifest = { "schema": 1, "authority": binding.authority_name, "route_fingerprint": _fingerprint(binding.route_id), "provider": binding.provider, "allowed_files": sorted(allowed), "provider_auth_files_present": False, } files = { "gateway-manifest.json": json.dumps(manifest, indent=2, sort_keys=True) + "\n" } if account.tool == "codex": base_url = _route_base(endpoint, binding, account.tool) files["config.toml"] = ( 'model_provider = "subrouter"\n\n' '[model_providers.subrouter]\n' 'name = "Subrouter"\n' f'base_url = {json.dumps(base_url)}\n' 'env_key = "SUBROUTER_PROXY_KEY"\n' 'wire_api = "responses"\n' ) return _materialize_immutable_home(root, files) class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *_args: object, **_kwargs: object) -> None: return None def _fetch_gateway_status( endpoint: AuthorityEndpoint, binding: AuthorityBinding, key: str, *, timeout_secs: float, ) -> GatewayStatus: route = urllib.parse.quote(binding.route_id, safe="") url = f"{endpoint.origin}/r/{route}/_subrouter/status" request = urllib.request.Request( url, headers={"Authorization": f"Bearer {key}", "Accept": "application/json"}, method="GET", ) opener = urllib.request.build_opener( urllib.request.ProxyHandler({}), _NoRedirect(), ) try: with opener.open(request, timeout=timeout_secs) as response: if response.status != 200 or response.geturl() != url: raise AuthorityConfigurationError("Gateway: unavailable") body = response.read(16_385) if len(body) > 16_384: raise AuthorityConfigurationError("Gateway: unavailable") except AuthorityConfigurationError: raise except (OSError, urllib.error.URLError, urllib.error.HTTPError) as exc: raise AuthorityConfigurationError("Gateway: unavailable") from exc try: payload = json.loads(body) except json.JSONDecodeError as exc: raise AuthorityConfigurationError("Gateway: unavailable") from exc expected_fields = { "state", "provider", "route_fingerprint", "grant_fingerprint", "grant_expires_at", "grant_revoked", "account_availability", } if not isinstance(payload, dict) or set(payload) != expected_fields: raise AuthorityConfigurationError("Gateway: unavailable") state = payload["state"] availability = payload["account_availability"] expected_availability = { "ready": "available", "migration-required": "unavailable", "blocked": "ambiguous", "exhausted": "exhausted", } if ( not isinstance(state, str) or state not in expected_availability or payload["provider"] != binding.provider or payload["route_fingerprint"] != _fingerprint(binding.route_id) or not isinstance(payload["grant_fingerprint"], str) or re.fullmatch(r"[0-9a-f]{12}", payload["grant_fingerprint"]) is None or payload["grant_revoked"] is not False or availability != expected_availability[state] ): raise AuthorityConfigurationError("Gateway: unavailable") expiry = payload["grant_expires_at"] if not isinstance(expiry, str): raise AuthorityConfigurationError("Gateway: unavailable") try: expires_at = datetime.fromisoformat(expiry.replace("Z", "+00:00")) except ValueError as exc: raise AuthorityConfigurationError("Gateway: unavailable") from exc if expires_at.tzinfo is None or expires_at.astimezone(UTC).timestamp() <= time.time(): raise AuthorityConfigurationError("Gateway: unavailable") detail = { "ready": "Gateway: ready", "migration-required": "Gateway: migration required", "blocked": "Gateway: repair required", "exhausted": "Gateway: quota exhausted", }[state] expiry_ms = int(expires_at.astimezone(UTC).timestamp() * 1000) return GatewayStatus( state=state, detail=detail, checked_at=time.time(), grant_expires_at_ms=expiry_ms ) def fetch_gateway_status( base_dir: Path, binding: AuthorityBinding, *, timeout_secs: float = 10.0, ) -> GatewayStatus: endpoint = resolve_authority_endpoint(base_dir, binding) key = _read_proxy_key(binding.proxy_grant_ref) return _fetch_gateway_status( endpoint, binding, key, timeout_secs=timeout_secs, ) def gateway_health_snapshot( base_dir: Path, account: Account, *, timeout_secs: float = 10.0, ) -> AccountSnapshot: binding = account.authority_binding if binding is None: raise AuthorityConfigurationError("Gateway: migration required") try: status = fetch_gateway_status(base_dir, binding, timeout_secs=timeout_secs) except AuthorityConfigurationError as exc: return AccountSnapshot( status=HealthStatus.UNKNOWN, primary_used_pct=None, secondary_used_pct=None, checked_at=time.time(), detail=str(exc), ) health = HealthStatus.OK if status.state == "ready" else HealthStatus.BROKEN return AccountSnapshot( status=health, primary_used_pct=None, secondary_used_pct=None, checked_at=status.checked_at, detail=status.detail, ) def resolve_authority_data_plane( base_dir: Path, binding: AuthorityBinding, *, timeout_secs: float = 10.0, ) -> AuthorityDataPlane: if ( binding.mode != AuthorityMode.SUBROUTER or binding.quiesced or binding.provider != "codex" ): raise AuthorityConfigurationError("Gateway: migration required") endpoint = resolve_authority_endpoint(base_dir, binding) key = _read_proxy_key(binding.proxy_grant_ref) status = _fetch_gateway_status(endpoint, binding, key, timeout_secs=timeout_secs) if status.grant_expires_at_ms is None: raise AuthorityConfigurationError("Gateway: unavailable") return AuthorityDataPlane( responses_url=f"{_route_base(endpoint, binding, 'codex')}/responses", proxy_key=key, grant_expires_at_ms=status.grant_expires_at_ms, sanitized_status=status, ) def build_authority_launch( base_dir: Path, account: Account, tool: str, argv: list[str], *, inherited_environment: dict[str, str] | None = None, ) -> AuthorityLaunch: binding = account.authority_binding if binding is None or binding.mode == AuthorityMode.NATIVE: raise AuthorityConfigurationError("Gateway: migration required") if tool != account.tool or binding.provider != tool: raise AuthorityConfigurationError("Gateway: migration required") endpoint = resolve_authority_endpoint(base_dir, binding) key = _read_proxy_key(binding.proxy_grant_ref) status = _fetch_gateway_status(endpoint, binding, key, timeout_secs=10.0) gateway_home = materialize_gateway_home(base_dir, account, endpoint, binding) environment = _sanitized_environment( dict(os.environ if inherited_environment is None else inherited_environment) ) session_id = str(uuid.uuid4()) environment["SUBROUTER_SESSION_ID"] = session_id environment["SUBROUTER_PROVIDER"] = tool environment["HOME"] = str(gateway_home) base_url = _route_base(endpoint, binding, tool) forwarded = list(argv) if tool == "codex": environment["CODEX_HOME"] = str(gateway_home) environment["CODEX_SQLITE_HOME"] = str(gateway_home / "sqlite") environment["SUBROUTER_PROXY_KEY"] = key elif tool == "claude": environment["CLAUDE_CONFIG_DIR"] = str(gateway_home) environment["ANTHROPIC_BASE_URL"] = base_url environment["ANTHROPIC_AUTH_TOKEN"] = key else: raise AuthorityConfigurationError("Gateway: migration required") return AuthorityLaunch( executable=tool, argv=tuple(forwarded), environment=environment, gateway_home=gateway_home, sanitized_status=status, )