diff --git a/api_access.py b/api_access.py index 2f4442c..789232b 100644 --- a/api_access.py +++ b/api_access.py @@ -1,70 +1,15 @@ from __future__ import annotations -import re -from typing import Literal +from backend.http import AccessRole, ApiRouteRegistry -AccessRole = Literal["authenticated", "member", "admin"] - -MEMBER_GET_PATHS = frozenset( - { - "/api/screener/setup", - "/api/screener/tracking", - "/api/mentors/setup", - "/api/mentors/messages", - "/api/heaven/setup", - "/api/heaven/readings", - "/api/assistant/messages", - } -) - -MEMBER_POST_PATHS = frozenset( - { - "/api/screener/sync", - "/api/screener/compile", - "/api/screener/strategies", - "/api/screener/run", - "/api/screener/tracking", - "/api/screener/tracking/refresh", - "/api/mentors/chat", - "/api/mentors/preferences", - "/api/heaven/hexagram", - "/api/heaven/personal", - "/api/heaven/interpret", - "/api/assistant/chat", - } -) - -ADMIN_POST_PATHS = frozenset( - { - "/api/backfill", - "/api/reasons", - "/api/seat-aliases", - "/api/heaven/sector-phases", - } -) +ROUTES = ApiRouteRegistry.load() def required_role(method: str, path: str) -> AccessRole: - method = method.upper() - if path.startswith("/api/admin/"): - return "admin" - if method == "GET" and path in MEMBER_GET_PATHS: - return "member" - if method == "POST": - if path in ADMIN_POST_PATHS: - return "admin" - if path in MEMBER_POST_PATHS: - return "member" - if method == "DELETE": - if re.fullmatch(r"/api/heaven/sector-phases/.+", path): - return "admin" - if path in {"/api/mentors/messages", "/api/assistant/messages"} or re.fullmatch( - r"/api/screener/strategies/\d+", path - ): - return "member" - if re.fullmatch(r"/api/heaven/readings/\d+", path): - return "member" - if re.fullmatch(r"/api/screener/tracking/\d+", path): - return "member" - return "authenticated" + """Compatibility access lookup backed by the authoritative route registry.""" + route = ROUTES.resolve(method, path) + return route.access if route else "authenticated" + + +__all__ = ["ROUTES", "AccessRole", "required_role"] diff --git a/backend/http/__init__.py b/backend/http/__init__.py new file mode 100644 index 0000000..61b4b88 --- /dev/null +++ b/backend/http/__init__.py @@ -0,0 +1,8 @@ +from .context import correlation_id +from .errors import normalize_error_payload +from .router import AccessRole, ApiRoute, ApiRouteRegistry, RouteRegistryError + +__all__ = [ + "AccessRole", "ApiRoute", "ApiRouteRegistry", "RouteRegistryError", + "correlation_id", "normalize_error_payload", +] diff --git a/backend/http/context.py b/backend/http/context.py new file mode 100644 index 0000000..fcf36ab --- /dev/null +++ b/backend/http/context.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +import re +import uuid + + +REQUEST_ID_PATTERN = re.compile(r"[A-Za-z0-9._-]{8,80}") + + +def correlation_id(supplied: str = "") -> str: + value = str(supplied or "").strip() + return value if REQUEST_ID_PATTERN.fullmatch(value) else uuid.uuid4().hex diff --git a/backend/http/errors.py b/backend/http/errors.py new file mode 100644 index 0000000..1535d82 --- /dev/null +++ b/backend/http/errors.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from http import HTTPStatus +from typing import Any + + +STATUS_CODES = { + HTTPStatus.BAD_REQUEST: "bad_request", + HTTPStatus.UNAUTHORIZED: "authentication_required", + HTTPStatus.FORBIDDEN: "access_denied", + HTTPStatus.NOT_FOUND: "not_found", + HTTPStatus.CONFLICT: "conflict", + HTTPStatus.INTERNAL_SERVER_ERROR: "internal_error", + HTTPStatus.SERVICE_UNAVAILABLE: "service_unavailable", +} + + +def normalize_error_payload( + payload: dict[str, Any], status: int | HTTPStatus, request_id: str, +) -> dict[str, Any]: + if "error" not in payload: + return payload + status_value = HTTPStatus(int(status)) + message = str(payload.get("message") or payload.get("error") or status_value.phrase) + return { + **payload, + "error": message, + "code": str(payload.get("code") or STATUS_CODES.get(status_value) or "request_failed"), + "message": message, + "request_id": request_id, + } diff --git a/backend/http/router.py b/backend/http/router.py new file mode 100644 index 0000000..337edf2 --- /dev/null +++ b/backend/http/router.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Literal, cast + +from app_config import APP_DIR + + +AccessRole = Literal["public", "authenticated", "member", "admin"] +MatchType = Literal["exact", "regex"] + + +class RouteRegistryError(RuntimeError): + pass + + +@dataclass(frozen=True) +class ApiRoute: + method: str + path: str + match: MatchType + feature: str + access: AccessRole + + def matches(self, method: str, path: str) -> bool: + if self.method != method.upper(): + return False + return self.path == path if self.match == "exact" else re.fullmatch(self.path, path) is not None + + +class ApiRouteRegistry: + def __init__(self, routes: tuple[ApiRoute, ...]) -> None: + self.routes = routes + self._exact: dict[tuple[str, str], ApiRoute] = {} + regex_routes: list[ApiRoute] = [] + seen: set[tuple[str, str]] = set() + for route in routes: + key = (route.method, route.path) + if key in seen: + raise RouteRegistryError(f"Duplicate API route: {route.method} {route.path}") + seen.add(key) + if route.match == "exact": + self._exact[key] = route + else: + try: + re.compile(route.path) + except re.error as exc: + raise RouteRegistryError(f"Invalid API route regex: {route.path}") from exc + regex_routes.append(route) + self._regex = tuple(regex_routes) + + @classmethod + def load(cls, path: Path | None = None) -> "ApiRouteRegistry": + config_path = path or APP_DIR / "config" / "api.config.json" + payload = json.loads(config_path.read_text(encoding="utf-8")) + routes = tuple( + ApiRoute( + method=str(item["method"]).upper(), + path=str(item["path"]), + match=cast(MatchType, str(item["match"])), + feature=str(item["feature"]), + access=cast(AccessRole, str(item["access"])), + ) + for item in payload.get("routes") or [] + ) + if not routes: + raise RouteRegistryError("API route registry is empty") + return cls(routes) + + def resolve(self, method: str, path: str) -> ApiRoute | None: + normalized = method.upper() + exact = self._exact.get((normalized, path)) + if exact: + return exact + return next((route for route in self._regex if route.matches(normalized, path)), None) diff --git a/docs/governance/architecture-inventory.json b/docs/governance/architecture-inventory.json index 8595f96..baef7f9 100644 --- a/docs/governance/architecture-inventory.json +++ b/docs/governance/architecture-inventory.json @@ -265,8 +265,8 @@ }, { "path": "server.py", - "bytes": 267824, - "lines": 5947 + "bytes": 268422, + "lines": 5960 }, { "path": "static/redesign-v2.css", diff --git a/docs/governance/stage-11-http-governance.md b/docs/governance/stage-11-http-governance.md new file mode 100644 index 0000000..509a517 --- /dev/null +++ b/docs/governance/stage-11-http-governance.md @@ -0,0 +1,28 @@ +# Stage 11: HTTP Route, Access, and Error Governance + +Date: 2026-07-29 + +## Result + +- Promoted `config/api.config.json` from a transitional inventory to the runtime route and + access registry. +- Added deterministic exact and regex route resolution with duplicate and regex validation. +- Replaced hand-maintained member/admin path sets with the registered access contract. +- Rejected unregistered API routes before business dispatch. +- Added safe request correlation IDs to JSON responses and `X-Request-ID` headers. +- Extended every legacy JSON error with stable `code`, `message`, and `request_id` fields while + preserving the existing `error` field used by the browser. +- Kept the current request handler and all route response bodies compatible while feature + route modules are migrated incrementally. + +## Runtime Authority + +Changing or adding an API now requires one coherent change to the handler and API registry. +The generated source inventory test prevents either side from drifting. Backend access remains +authoritative; frontend visibility cannot grant a route. + +## Residual Migration + +Individual dispatch branches still live in the compatibility request handler. Feature-owned +controllers will move behind the same registry in later stages without changing route identity, +authorization, or error serialization. diff --git a/server.py b/server.py index 0ac19f6..f854460 100644 --- a/server.py +++ b/server.py @@ -16,8 +16,9 @@ from typing import Any from urllib.parse import parse_qs, unquote, urlparse from assistant_agent import ReviewAssistantError, stream_review_assistant -from api_access import required_role +from api_access import ROUTES from backend.bootstrap import build_application_container, load_runtime_settings +from backend.http import correlation_id, normalize_error_payload from chart_data_provider import ChartDataError from app_config import ( DATA_DIR, @@ -5581,7 +5582,13 @@ class RequestHandler(BaseHTTPRequestHandler): return False def require_access(self, method: str, path: str) -> bool: - role = required_role(method, path) + route = ROUTES.resolve(method, path) + if route is None: + self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND) + return False + role = route.access + if role == "public": + return True if role == "admin": return self.require_admin() if role == "member": @@ -5912,11 +5919,17 @@ class RequestHandler(BaseHTTPRequestHandler): status: HTTPStatus = HTTPStatus.OK, headers: dict[str, str] | None = None, ) -> None: + request_id = getattr(self, "_correlation_id", "") + if not request_id: + request_id = correlation_id(self.headers.get("X-Request-ID", "")) + self._correlation_id = request_id + payload = normalize_error_payload(payload, status, request_id) content = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(content))) self.send_header("Cache-Control", "no-store") + self.send_header("X-Request-ID", request_id) for name, value in (headers or {}).items(): self.send_header(name, value) self.end_headers() diff --git a/tests/test_http_governance.py b/tests/test_http_governance.py new file mode 100644 index 0000000..d951eea --- /dev/null +++ b/tests/test_http_governance.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import unittest +from http import HTTPStatus + +from api_access import ROUTES, required_role +from backend.http import correlation_id, normalize_error_payload +from tools.build_api_registry import build as build_api_registry + + +class HttpGovernanceTests(unittest.TestCase): + def test_runtime_registry_resolves_every_declared_route(self) -> None: + for item in build_api_registry()["routes"]: + path = ( + item["path"] + .replace("(\\d{6})", "000001") + .replace("(\\d+)", "1") + .replace("(.+)", "sample") + ) + resolved = ROUTES.resolve(item["method"], path) + self.assertIsNotNone(resolved, f"{item['method']} {path}") + self.assertEqual(resolved.feature, item["feature"]) + self.assertEqual(resolved.access, item["access"]) + + def test_unknown_route_has_no_runtime_match(self) -> None: + self.assertIsNone(ROUTES.resolve("GET", "/api/not-registered")) + + def test_compatibility_access_function_uses_runtime_registry(self) -> None: + self.assertEqual(required_role("GET", "/api/screener/setup"), "member") + self.assertEqual(required_role("POST", "/api/admin/settings"), "admin") + self.assertEqual(required_role("GET", "/api/dashboard"), "authenticated") + + def test_errors_keep_legacy_field_and_add_stable_contract(self) -> None: + result = normalize_error_payload( + {"error": "invalid input"}, HTTPStatus.BAD_REQUEST, "request-123" + ) + self.assertEqual(result["error"], "invalid input") + self.assertEqual(result["message"], "invalid input") + self.assertEqual(result["code"], "bad_request") + self.assertEqual(result["request_id"], "request-123") + + def test_correlation_id_rejects_header_injection(self) -> None: + self.assertEqual(correlation_id("client-request-123"), "client-request-123") + self.assertRegex(correlation_id("bad\r\nheader"), r"^[0-9a-f]{32}$") + + +if __name__ == "__main__": + unittest.main()