"""v0.20.0 — on-site framework-specs surface source. Sibling of `docs_sessions.py` (v0.19.0 / roadmap item #30): the framework mediates a gitea fetch on behalf of the browser so the rendered `/docs/specs/*` surface inherits the same chrome as `/docs/user-guide` and `/docs/sessions/*` and stays free of any cross-origin gestures from the frontend. Two read endpoints, both anonymous-reachable: GET /api/docs/specs/manifest — the configured spec list GET /api/docs/specs/ — a single spec body (markdown) The framework-default manifest is OHM-flavored (rfc-app's own SPEC.md + flotilla's SPEC.md on `git.wiggleverse.org`) for the same reason `docs_sessions.py`'s defaults are: OHM is the only deployment to date. A deployment running its own spec set overrides the manifest via the `OHM_DOCS_SPECS` env var (set through flotilla's overlay). History is intentionally not surfaced here — the operator-stated intent is "current version only; git is the history surface". Per-spec entries carry three fields: name — URL-safe slug (`[a-z0-9-]+`) — the path segment title — human-readable label shown in the nav and the page header url — the upstream raw URL the framework fetches Validation: - The configured list must be a JSON array of `{name, title, url}` objects. A malformed `OHM_DOCS_SPECS` value (bad JSON, wrong shape, invalid slug) logs a warning and falls back to the default so a typo in the overlay doesn't crash startup. - Each `name` is checked against `^[a-z0-9-]+$` before the manifest is accepted. The route layer also validates the path-bound `name` parameter before any network call, so a malformed URL never reaches the cache or the upstream. Cache shape mirrors `docs_sessions.py`: in-process `dict` + monotonic TTL check, negative results (404) cached, no external dep. The manifest is cheap (parsed from an env var, no network), so it has no TTL — every request re-derives it. Per-spec content has a 5-minute default TTL (env-tunable via `OHM_DOCS_SPECS_CONTENT_TTL_SEC`). §3 invariant 1 is preserved: the framework holds no secret bytes; the upstream specs are public-repo raw URLs, the fetch carries no auth header. """ from __future__ import annotations import json import logging import os import re import threading import time from typing import Any import httpx log = logging.getLogger(__name__) # The framework-default spec set. OHM-flavored per the same precedent # `docs_sessions.py` set: the only live deployment is OHM, so the # default points there. A deployment running its own specs overrides # `OHM_DOCS_SPECS` via the overlay. _DEFAULT_SPECS: list[dict[str, str]] = [ { "name": "rfc-app", "title": "rfc-app SPEC", "url": ( "https://git.wiggleverse.org/ben.stull/rfc-app/" "raw/branch/main/SPEC.md" ), }, { "name": "flotilla", "title": "flotilla SPEC", "url": ( "https://git.wiggleverse.org/wiggleverse/ohm-rfc-app-flotilla/" "raw/branch/main/SPEC.md" ), }, ] _DEFAULT_CONTENT_TTL_SEC = 300.0 # URL-safe slug. Matches `docs_sessions.py`'s `_SESSION_DIR_RE` spirit # (rejecting anything that could resolve outside the intended layout) # but with the lowercase-alphanumeric-plus-dash shape the manifest # enforces. Path traversal (`..`), separators (`/`), tilde, uppercase, # and whitespace all fail this regex; the route layer rejects 400 # before any cache or network call. _NAME_RE = re.compile(r"^[a-z0-9-]+$") _HTTP_TIMEOUT_SEC = 5.0 def _env_float(name: str, default: float) -> float: raw = os.environ.get(name, "").strip() if not raw: return default try: return float(raw) except ValueError: log.warning("invalid %s=%r — falling back to %s", name, raw, default) return default def _content_ttl() -> float: return _env_float("OHM_DOCS_SPECS_CONTENT_TTL_SEC", _DEFAULT_CONTENT_TTL_SEC) def _is_valid_name(name: str) -> bool: """Slug guard for path-bound `name` parameters. Mirrors `docs_sessions._is_valid_session_dir`'s contract: the route layer calls this before any network or cache work, so a malformed name never escapes the FastAPI surface. """ return bool(isinstance(name, str) and _NAME_RE.match(name)) def _parse_spec_entry(entry: Any) -> dict[str, str] | None: """Validate a single manifest entry; return None if invalid. Required fields: `name`, `title`, `url`. All three must be non-empty strings; `name` must match `_NAME_RE`. The validator is strict: an entry that fails any check is dropped from the manifest (and the caller logs at warning level). """ if not isinstance(entry, dict): return None name = entry.get("name") title = entry.get("title") url = entry.get("url") if not isinstance(name, str) or not _is_valid_name(name): return None if not isinstance(title, str) or not title.strip(): return None if not isinstance(url, str) or not url.strip(): return None return {"name": name, "title": title.strip(), "url": url.strip()} def _load_configured_specs() -> list[dict[str, str]]: """Parse `OHM_DOCS_SPECS` (if set) or return the default list. Malformed JSON or wrong-shape values log a warning and fall back to the default — the deployment continues to render the spec surface rather than crashing startup. The strict validation (each entry's name slug, presence of all three fields) drops bad entries one-by-one; if every entry is dropped, the default applies. """ raw = os.environ.get("OHM_DOCS_SPECS", "").strip() if not raw: return list(_DEFAULT_SPECS) try: parsed = json.loads(raw) except (json.JSONDecodeError, ValueError) as e: log.warning( "OHM_DOCS_SPECS is not valid JSON (%s) — falling back to default", e ) return list(_DEFAULT_SPECS) if not isinstance(parsed, list): log.warning( "OHM_DOCS_SPECS must be a JSON array — falling back to default" ) return list(_DEFAULT_SPECS) out: list[dict[str, str]] = [] seen: set[str] = set() for entry in parsed: validated = _parse_spec_entry(entry) if validated is None: log.warning( "OHM_DOCS_SPECS entry %r failed validation — dropped", entry ) continue if validated["name"] in seen: log.warning( "OHM_DOCS_SPECS has duplicate name %r — dropped", validated["name"] ) continue seen.add(validated["name"]) out.append(validated) if not out: log.warning( "OHM_DOCS_SPECS yielded no valid entries — falling back to default" ) return list(_DEFAULT_SPECS) return out # --------------------------------------------------------------------------- # In-process TTL cache # --------------------------------------------------------------------------- # # Same shape as `docs_sessions.py`: plain dict + `time.monotonic()` check, # no external dep. The cache value is a `(stored_at, payload)` tuple; # `payload` may carry an error-shape sentinel for negative caching (404s). # Lock guards read-modify-write across worker tasks; entries are immutable # once stored so reads under the lock are fast. _lock = threading.Lock() _cache: dict[str, tuple[float, dict[str, Any]]] = {} def _cache_get(key: str, ttl_sec: float) -> dict[str, Any] | None: with _lock: entry = _cache.get(key) if entry is None: return None stored_at, payload = entry if time.monotonic() - stored_at > ttl_sec: return None return payload def _cache_put(key: str, payload: dict[str, Any]) -> None: with _lock: _cache[key] = (time.monotonic(), payload) def reset_cache() -> None: """Drop every cached entry. Test seam — not called in production.""" with _lock: _cache.clear() # --------------------------------------------------------------------------- # Public fetch surface # --------------------------------------------------------------------------- # # Each fetcher returns a `{status, ...}` dict, same convention as # `docs_sessions.py`: # "ok" — payload field carries the body / manifest # "404" — gitea returned 404 (or the configured name doesn't exist) # "error" — gitea returned 5xx, timed out, or returned malformed data # # The route layer maps these onto HTTP responses; keeping the mapping # out of this module makes the cache transparent to the test harness. async def _http_get(url: str) -> tuple[int, str]: """Perform a single GET against `url`; return (status_code, body). On timeout or network error, returns (599, error_message). The 599 pseudo-status maps to a 502 at the route layer the same way an upstream 5xx does — the caller doesn't care which leg of the network broke. """ try: async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT_SEC) as client: r = await client.get(url) return r.status_code, r.text except httpx.HTTPError as e: log.warning("specs fetch failed for %s: %s", url, e) return 599, f"fetch error: {e}" def fetch_specs_manifest() -> dict[str, Any]: """Return the configured spec manifest. The manifest is derived from the `OHM_DOCS_SPECS` env var (or the framework default if unset / malformed) and carries no network work — it's safe to call on every request. The return shape mirrors the docs_sessions manifest endpoint for frontend consistency: {"status": "ok", "specs": [{"name", "title", "url"}, ...]} The "url" field is exposed in the manifest so the frontend can offer a "view source on gitea" affordance alongside each rendered spec (operator-stated intent: "include the history so you can see it in git" — that gesture lives in the source link, not on the rendered page). """ specs = _load_configured_specs() return {"status": "ok", "specs": specs} async def fetch_spec(name: str) -> dict[str, Any]: """Fetch a single spec body by its manifest `name`. The caller is expected to have validated `name` against `_is_valid_name` before calling this — an invalid name shouldn't reach the network. We re-check inside as defense-in-depth: a bogus name here returns the same `{status: "404"}` shape so the route layer's `404 → HTTP 404` mapping handles it uniformly. Returns one of: {"status": "ok", "body": "..."} {"status": "404"} — no such spec OR upstream 404 {"status": "error", "detail": "..."} — upstream 5xx / timeout """ if not _is_valid_name(name): return {"status": "404"} cache_key = f"spec:{name}" cached = _cache_get(cache_key, _content_ttl()) if cached is not None: return cached specs = _load_configured_specs() match = next((s for s in specs if s["name"] == name), None) if match is None: # Cache the negative — a deployment with an unstable manifest # would still benefit from the TTL window, and the cached 404 # is automatically displaced when the next request happens # after TTL expiry. payload: dict[str, Any] = {"status": "404"} _cache_put(cache_key, payload) return payload url = match["url"] status, body = await _http_get(url) if status == 200: payload = {"status": "ok", "body": body} _cache_put(cache_key, payload) return payload if status == 404: payload = {"status": "404"} _cache_put(cache_key, payload) return payload return {"status": "error", "detail": f"upstream returned {status}"}