"""API-process config: port, git sha resolution, and config-masking helpers. Stage 1 deliberately keeps this thin — no CORS allowlist (CONTRACT.md §2: v1 runs same-origin through the Express proxy), no session secret yet (Stage 2 adds auth). What lives here: - The uvicorn port (default 8000; override via ``API_PORT``). - :func:`current_git_sha` — read at process start so ``/system/version`` can echo it without shelling out per request. - :func:`masked_config` — walks ``config.yaml`` and redacts anything that looks like a secret. Used by ``/system/version``. Anything that grows into "session config" or "CORS origins" lands here in later stages; Stage 1 deliberately doesn't pre-build those slots. """ from __future__ import annotations import os import re import subprocess from copy import deepcopy from functools import lru_cache from typing import Any from src.config import REPO_ROOT, load_config # ---------- Port ---------------------------------------------------------- def api_port() -> int: """Port the API listens on. Default 8000; override via ``API_PORT``. The number itself doesn't matter to the frontend — the Express proxy in ``Patristic-AI-Fe/server.ts`` maps ``/api/*`` to this port. Keep it in env for ops flexibility (e.g., running two instances on the same machine during a cutover). """ raw = os.environ.get("API_PORT", "8000").strip() try: return int(raw) except ValueError: return 8000 # ---------- Git SHA ------------------------------------------------------- @lru_cache(maxsize=1) def current_git_sha() -> str: """Return the current git short-SHA, or ``"unknown"`` if not available. Cached for the process lifetime: the value can't change without a restart (we don't run inside a checked-out worktree that's being modified by anything other than us). Failures (no git on PATH, not a repo) degrade to ``"unknown"`` rather than raising — ``/system/version`` should always answer. """ try: out = subprocess.run( ["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True, timeout=2, cwd=str(REPO_ROOT), ) except (subprocess.TimeoutExpired, OSError, FileNotFoundError): return "unknown" if out.returncode != 0: return "unknown" return out.stdout.strip() or "unknown" # ---------- Config masking ------------------------------------------------ # Keys whose values are redacted in /system/version output. Conservative: # anything that smells like a credential gets masked regardless of where it # sits in the tree. The matcher is case-insensitive. _SECRET_KEY_PATTERNS = ( re.compile(r"(api[_-]?key|token|secret|password|auth)", re.IGNORECASE), ) # A URL with embedded credentials (e.g. libsql://user:pass@host) gets the # userinfo segment stripped before being emitted. Plain http URLs pass # through unchanged. _URL_WITH_CREDS = re.compile(r"^([a-z][a-z0-9+\-.]*://)[^/@\s]+@", re.IGNORECASE) def mask_url_credentials(value: str) -> str: """Strip a userinfo segment from a URL. ``scheme://user:pass@host`` → ``scheme://***@host``; credential-free or non-URL strings pass through unchanged. Single source of truth for this rule — both :func:`_walk_and_mask` here and the ``/config`` router import it so the redaction can't drift between the two endpoints. """ if not value: return value return _URL_WITH_CREDS.sub(r"\1***@", value) def _looks_like_secret_key(key: str) -> bool: return any(p.search(key) for p in _SECRET_KEY_PATTERNS) def _redact(value: Any) -> Any: if isinstance(value, str): if not value: return value # Show first 2 / last 2 chars for ops triage but keep the bulk hidden. if len(value) <= 6: return "***" return f"{value[:2]}***{value[-2:]}" # Non-string secrets are unusual; just blank them. return "***" def _walk_and_mask(node: Any) -> Any: """Recursively mask anything that looks secret in a dict/list tree.""" if isinstance(node, dict): out: dict[str, Any] = {} for k, v in node.items(): if isinstance(k, str) and _looks_like_secret_key(k): out[k] = _redact(v) else: out[k] = _walk_and_mask(v) return out if isinstance(node, list): return [_walk_and_mask(item) for item in node] if isinstance(node, str) and _URL_WITH_CREDS.match(node): return mask_url_credentials(node) return node def masked_config() -> dict[str, Any]: """Return ``config.yaml`` with secret-shaped values redacted. Reads from the cached :func:`src.config.load_config` so the underlying config doesn't get re-parsed per request. Deep-copied before masking so we never mutate the shared cache. """ cfg = load_config() return _walk_and_mask(deepcopy(cfg.raw))