Spaces:
Running
Running
| """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 ------------------------------------------------------- | |
| 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)) | |