fix(api): RBAC default-deny + constant-time API key (#1133 #1134)

api/rbac.py

@@ -3,11 +3,17 @@
 
 Opt-in via ``api.rbac.enabled``; requires Pro+ tier (``dashboard_rbac`` in tier_features).
 Roles: ``admin`` (all), ``dashboard``, ``scanner``, ``reports_reader``, ``config_admin``.
+
+When RBAC is active, policy is **default-deny**: only routes on the public allowlist or the
+explicit role map are reachable; unclassified routes return 403.
 """
 
 from __future__ import annotations
 
+import hmac
 import json
+import re
+from enum import Enum
 from typing import Any
 
 from starlette.requests import Request
@@ -23,6 +29,19 @@
     {"admin", "dashboard", "scanner", "reports_reader", "config_admin"}
 )
 
+SCAN = frozenset({"scanner", "admin"})
+REP = frozenset({"reports_reader", "admin"})
+DASH = frozenset({"dashboard", "admin"})
+CFG = frozenset({"config_admin", "admin"})
+
+
+class RouteRbacClass(str, Enum):
+    """How RBAC classifies a route when enforcement is active."""
+
+    PUBLIC = "public"
+    PROTECTED = "protected"
+    UNCLASSIFIED = "unclassified"
+
 
 def _normalize_role_list(raw: list[str] | None, fallback: list[str]) -> list[str]:
     source = raw if raw else fallback
@@ -52,6 +71,127 @@ def _parse_roles_json(s: str | None) -> list[str] | None:
     return None
 
 
+def _concrete_path_for_policy(path: str) -> str:
+    """Map FastAPI route templates to a concrete path for policy classification."""
+    out = path.replace("{locale_slug}", "en")
+    return re.sub(r"\{[^}]+\}", "sample-id", out)
+
+
+def is_rbac_public_route(method: str, path: str) -> bool:
+    """Explicit allowlist: no RBAC check when enforcement is active."""
+    m = method.upper()
+    concrete = _concrete_path_for_policy(path)
+    if concrete == "/health" or concrete.startswith("/static/"):
+        return True
+    if concrete.startswith("/auth/webauthn"):
+        return True
+    if concrete == "/about/json":
+        return True
+
+    slug, rest = locale_path_segments(concrete)
+    if (
+        slug
+        and m in ("GET", "HEAD")
+        and len(rest) == 1
+        and rest[0] in ("help", "about", "login")
+    ):
+        return True
+    return False
+
+
+def required_roles_for_route(method: str, path: str) -> frozenset[str] | None:
+    """
+    Return roles that may access this protected route (any one is enough unless admin).
+
+    None means the route is public (see ``is_rbac_public_route``) or unclassified when
+    used alone — prefer ``classify_route_rbac`` for middleware.
+    """
+    classification, roles = classify_route_rbac(method, path)
+    if classification == RouteRbacClass.PROTECTED:
+        assert roles is not None
+        return roles
+    return None
+
+
+def classify_route_rbac(
+    method: str, path: str
+) -> tuple[RouteRbacClass, frozenset[str] | None]:
+    """
+    Classify a route for RBAC when enforcement is active.
+
+    PUBLIC: allow without role check.
+    PROTECTED: requires principal with one of the returned roles (or admin).
+    UNCLASSIFIED: deny (403) — route must be added to the policy map.
+    """
+    if is_rbac_public_route(method, path):
+        return RouteRbacClass.PUBLIC, None
+
+    m = method.upper()
+    concrete = _concrete_path_for_policy(path)
+
+    if concrete in ("/scan", "/start") and m == "POST":
+        return RouteRbacClass.PROTECTED, SCAN
+    if concrete == "/scan_database" and m == "POST":
+        return RouteRbacClass.PROTECTED, SCAN
+    if concrete == "/scan_pdf" and m == "POST":
+        return RouteRbacClass.PROTECTED, REP
+    if concrete in ("/report", "/heatmap") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, REP
+    if concrete.startswith("/reports/") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, REP
+    if concrete.startswith("/heatmap/") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, REP
+    if concrete in ("/findings", "/findings/csv") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, REP
+    if concrete.startswith("/findings/") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, REP
+    if concrete in ("/status", "/list") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, DASH
+    if concrete in ("/openapi.json", "/docs", "/redoc") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, DASH
+    if concrete.startswith("/docs/") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, DASH
+    if concrete.startswith("/logs") and m in ("GET", "HEAD"):
+        return RouteRbacClass.PROTECTED, DASH
+    if concrete.startswith("/sessions/") and m == "PATCH":
+        return RouteRbacClass.PROTECTED, DASH
+
+    slug, rest = locale_path_segments(concrete)
+    if slug:
+        if rest == ["reports"] and m in ("GET", "HEAD"):
+            return RouteRbacClass.PROTECTED, REP
+        if rest == ["config"] and m in ("GET", "HEAD", "POST"):
+            return RouteRbacClass.PROTECTED, CFG
+        if len(rest) == 0 and m in ("GET", "HEAD"):
+            return RouteRbacClass.PROTECTED, DASH
+        if rest == ["assessment"] or rest == ["assessment", "export"]:
+            if m in ("GET", "HEAD"):
+                return RouteRbacClass.PROTECTED, DASH
+        if rest == ["assessment"] and m == "POST":
+            return RouteRbacClass.PROTECTED, DASH
+
+    return RouteRbacClass.UNCLASSIFIED, None
+
+
+def iter_fastapi_route_specs(app: Any) -> list[tuple[str, str]]:
+    """(method, path_template) for every HTTP route on the app (for invariant tests)."""
+    from starlette.routing import BaseRoute
+
+    specs: list[tuple[str, str]] = []
+    for route in app.routes:
+        if not isinstance(route, BaseRoute):
+            continue
+        methods = getattr(route, "methods", None) or set()
+        path = getattr(route, "path", None)
+        if not path:
+            continue
+        for method in sorted(methods):
+            if method in ("OPTIONS",):
+                continue
+            specs.append((method, path))
+    return specs
+
+
 def resolve_effective_roles_for_request(
     request: Request, cfg: dict[str, Any], db_manager: Any
 ) -> list[str] | None:
@@ -72,7 +212,9 @@ def resolve_effective_roles_for_request(
             auth = request.headers.get("authorization", "").strip()
             if auth.lower().startswith("bearer "):
                 provided = auth[7:].strip()
-        if provided and provided == expected:
+        if provided and hmac.compare_digest(
+            provided.encode("utf-8"), expected.encode("utf-8")
+        ):
             return _normalize_role_list(api_key_roles, default_roles)
 
     wa = webauthn_block(cfg)
@@ -100,69 +242,6 @@ def principal_allows(required: frozenset[str], roles: list[str]) -> bool:
     return bool(required.intersection(roles))
 
 
-def required_roles_for_route(method: str, path: str) -> frozenset[str] | None:
-    """
-    Return roles that may access this route (any one is enough unless using admin).
-
-    None means this path is not covered by RBAC (public or unknown to policy).
-    """
-    m = method.upper()
-    if path == "/health" or path.startswith("/static/"):
-        return None
-    if path.startswith("/auth/webauthn"):
-        return None
-    if path == "/about/json":
-        return None
-
-    slug, rest = locale_path_segments(path)
-    if (
-        slug
-        and m in ("GET", "HEAD")
-        and len(rest) == 1
-        and rest[0] in ("help", "about", "login")
-    ):
-        return None
-
-    SCAN = frozenset({"scanner", "admin"})
-    REP = frozenset({"reports_reader", "admin"})
-    DASH = frozenset({"dashboard", "admin"})
-    CFG = frozenset({"config_admin", "admin"})
-
-    if path in ("/scan", "/start") and m == "POST":
-        return SCAN
-    if path == "/scan_database" and m == "POST":
-        return SCAN
-    if path == "/scan_pdf" and m == "POST":
-        return REP
-    if path in ("/report", "/heatmap") and m in ("GET", "HEAD"):
-        return REP
-    if path.startswith("/reports/") and m in ("GET", "HEAD"):
-        return REP
-    if path.startswith("/heatmap/") and m in ("GET", "HEAD"):
-        return REP
-    if path in ("/status", "/list") and m in ("GET", "HEAD"):
-        return DASH
-    if path.startswith("/logs") and m in ("GET", "HEAD"):
-        return DASH
-    if path.startswith("/sessions/") and m == "PATCH":
-        return DASH
-
-    if slug:
-        if rest == ["reports"] and m in ("GET", "HEAD"):
-            return REP
-        if rest == ["config"] and m in ("GET", "HEAD", "POST"):
-            return CFG
-        if len(rest) == 0 and m in ("GET", "HEAD"):
-            return DASH
-        if rest == ["assessment"] or rest == ["assessment", "export"]:
-            if m in ("GET", "HEAD"):
-                return DASH
-        if rest == ["assessment"] and m == "POST":
-            return DASH
-
-    return None
-
-
 async def rbac_middleware_handler(
     request: Request,
     call_next: Any,
@@ -172,9 +251,17 @@ async def rbac_middleware_handler(
     cfg = get_config()
     if not rbac_enforcement_active(cfg):
         return await call_next(request)
-    req_roles = required_roles_for_route(request.method, request.url.path)
-    if req_roles is None:
+
+    classification, req_roles = classify_route_rbac(request.method, request.url.path)
+    if classification == RouteRbacClass.PUBLIC:
         return await call_next(request)
+    if classification == RouteRbacClass.UNCLASSIFIED:
+        return JSONResponse(
+            status_code=403,
+            content={"detail": "Route not classified in RBAC policy."},
+        )
+    assert req_roles is not None
+
     engine = get_engine()
     roles = resolve_effective_roles_for_request(request, cfg, engine.db_manager)
     if roles is None:

tests/test_rbac.py

@@ -3,11 +3,18 @@
 from __future__ import annotations
 
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 import yaml
 from fastapi.testclient import TestClient
 
+from api.rbac import (
+    RouteRbacClass,
+    classify_route_rbac,
+    iter_fastapi_route_specs,
+    resolve_effective_roles_for_request,
+)
 from core.rbac_settings import rbac_enforcement_active
 from core.webauthn_rp.session_cookie import sign_session
 from core.webauthn_rp.settings import user_id_bytes, webauthn_block
@@ -164,3 +171,57 @@ def test_rbac_webauthn_session_roles_json_reports_only(rbac_pro_client):
 
     assert client.get("/en/reports").status_code == 200
     assert client.get("/en/").status_code == 403
+
+
+def test_all_registered_routes_classified_for_rbac():
+    """#1133 invariant: every app route is public or explicitly role-mapped."""
+    import api.routes as routes
+
+    unclassified: list[str] = []
+    for method, path in iter_fastapi_route_specs(routes.app):
+        kind, _roles = classify_route_rbac(method, path)
+        if kind == RouteRbacClass.UNCLASSIFIED:
+            unclassified.append(f"{method} {path}")
+    assert not unclassified, (
+        "Unclassified routes (default-deny when RBAC active):\n"
+        + "\n".join(unclassified)
+    )
+
+
+def test_rbac_findings_requires_role_when_enforced(rbac_pro_client):
+    client, _ = rbac_pro_client
+    assert client.get("/findings").status_code == 401
+    h = {"X-API-Key": "test-secret-key"}
+    assert client.get("/findings", headers=h).status_code == 403
+
+
+def test_rbac_api_key_compare_uses_hmac_compare_digest(rbac_pro_client):
+    """#1134: API key path in rbac must use hmac.compare_digest, not ==."""
+    client, routes_mod = rbac_pro_client
+    cfg = routes_mod._get_config()
+    engine = routes_mod._get_engine()
+    request = type(
+        "Req",
+        (),
+        {
+            "headers": {"x-api-key": "test-secret-key"},
+            "cookies": {},
+        },
+    )()
+
+    with patch("api.rbac.hmac.compare_digest", return_value=True) as mock_digest:
+        roles = resolve_effective_roles_for_request(request, cfg, engine.db_manager)
+        mock_digest.assert_called_once_with(b"test-secret-key", b"test-secret-key")
+    assert roles is not None
+    assert "dashboard" in roles
+
+
+def test_rbac_non_ascii_api_key_returns_401_not_500(rbac_pro_client):
+    """#1150: non-ASCII X-API-Key must not crash compare_digest (401, not 500)."""
+    client, _ = rbac_pro_client
+    # httpx rejects non-ASCII str headers; send UTF-8 bytes like a real client may.
+    r = client.get(
+        "/status",
+        headers=[(b"x-api-key", "café-🔑".encode("utf-8"))],
+    )
+    assert r.status_code == 401