From 5d98dbd5ed4925ada8be28714d1a8df403221dc1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?FabioLeit=C3=A3o?= Date: Thu, 2 Jul 2026 21:39:58 -0300 Subject: [PATCH 1/3] fix(api): RBAC default-deny with full route map (#1133) When api.rbac is active, unclassified routes return 403; public allowlist covers health/static/webauthn/help/about/login. Add findings and OpenAPI doc routes to the role map. --- api/rbac.py | 217 ++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 151 insertions(+), 66 deletions(-) diff --git a/api/rbac.py b/api/rbac.py index 9b9561d06..aa5ee609c 100644 --- a/api/rbac.py +++ b/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,7 @@ 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, expected): return _normalize_role_list(api_key_roles, default_roles) wa = webauthn_block(cfg) @@ -100,69 +240,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 +249,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: From 8dab6d3f669211c61e65be3399fa1e85b44e007e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?FabioLeit=C3=A3o?= Date: Thu, 2 Jul 2026 21:40:09 -0300 Subject: [PATCH 2/3] test(api): RBAC route invariant and constant-time API key (#1133 #1134) Fail CI when a new app route lacks RBAC classification; assert rbac resolve path uses hmac.compare_digest for API keys. --- tests/test_rbac.py | 50 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/tests/test_rbac.py b/tests/test_rbac.py index 334ab56a8..7927c41f1 100644 --- a/tests/test_rbac.py +++ b/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,46 @@ 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() + assert roles is not None + assert "dashboard" in roles From e19700541ca721c4461ef71d32c60457bb380715 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?FabioLeit=C3=A3o?= Date: Fri, 3 Jul 2026 10:12:36 -0300 Subject: [PATCH 3/3] fix(security): UTF-8 bytes for rbac API key compare_digest (#1150) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit hmac.compare_digest on str rejects non-ASCII and raised TypeError → 500. Encode provided/expected as UTF-8 bytes (constant-time, no-match → 401). Closes #1150 --- api/rbac.py | 4 +++- tests/test_rbac.py | 13 ++++++++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/api/rbac.py b/api/rbac.py index aa5ee609c..297caad81 100644 --- a/api/rbac.py +++ b/api/rbac.py @@ -212,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 hmac.compare_digest(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) diff --git a/tests/test_rbac.py b/tests/test_rbac.py index 7927c41f1..a0473307e 100644 --- a/tests/test_rbac.py +++ b/tests/test_rbac.py @@ -211,6 +211,17 @@ def test_rbac_api_key_compare_uses_hmac_compare_digest(rbac_pro_client): 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() + 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