patristic-be / src /api /config.py
Mario33333's picture
deploy: reader-access security hardening (feature/audiobook@06a5ed8)
7c2d250 verified
Raw
History Blame Contribute Delete
5.13 kB
"""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))