diff --git a/.cursor/rules/check-all-gate.mdc b/.cursor/rules/check-all-gate.mdc index 58588ea5..1042c1bf 100644 --- a/.cursor/rules/check-all-gate.mdc +++ b/.cursor/rules/check-all-gate.mdc @@ -6,12 +6,14 @@ alwaysApply: true # Automation scripts (token-aware) +**Two tiers (do not conflate):** **Local `git commit`** — pre-commit hooks (`uv run pre-commit install` once per clone) plus fast slice checks as needed (`lint-only`, `quick-test`, etc.). **`git push` / PR / merge publish to GitHub** — full **`./scripts/check-all.sh`** or **`.\scripts\check-all.ps1`** (exit 0, same session); prefer **`commit-or-pr.ps1`** and **`pr-merge-when-green.ps1`**. **Do not** use ad-hoc pytest, `lint-only`, or **`-RunTests`** alone as pre-push proof. + **Prefer existing scripts over ad-hoc commands** when it saves tokens or time. | When | Script | Example | | ------ | -------- | -------- | -| Full gate (lint + tests) | `.\scripts\check-all.ps1` (Windows) or `./scripts/check-all.sh` (Linux / macOS) | Before commit/PR | -| Quick iterations (tests only) | `.\scripts\check-all.ps1 -SkipPreCommit` or `./scripts/check-all.sh --skip-pre-commit` | After code change | +| Full gate before push/PR to GitHub | `.\scripts\check-all.ps1` (Windows) or `./scripts/check-all.sh` (Linux / macOS) | Before **`git push`**, **`gh pr create`**, or merge publish — not required before every local commit | +| Quick iterations (tests only) | `.\scripts\check-all.ps1 -SkipPreCommit` or `./scripts/check-all.sh --skip-pre-commit` | After code change — re-run full **`check-all`** before push | | Lint/format only (no pytest) | `.\scripts\lint-only.ps1` or `./scripts/lint-only.sh` | After docs/templates/style change — runs **`pre-commit run --all-files`** (Ruff, plans-stats, plans-hub, markdown, pt-BR, commercial guard, external plan links) | | Subset of tests | `.\scripts\quick-test.ps1 -Path tests/test_foo.py` or `-Keyword "content_type"` · `./scripts/quick-test.sh --path …` / `--keyword …` | When iterating on one area | | **`.github/workflows/` YAML edits** | `.\scripts\quick-test.ps1 -Path tests/test_github_workflows.py` · `./scripts/quick-test.sh --path tests/test_github_workflows.py` | Same module runs inside full **`check-all`** / **`ci.yml`** (`uv run pytest`); guards Slack webhook step pattern, Action SHA pins, `ci.yml` lint = **`pre-commit run --all-files`**. See **`docs/TESTING.md`**, **`docs/ops/OPERATOR_NOTIFICATION_CHANNELS.md`** §4.1.1, **ADR 0005**. | @@ -22,15 +24,16 @@ alwaysApply: true | Doc-bundle recovery (post-`cat` / copy mess) | `.\scripts\recovery-doc-bundle-sanity.ps1` (optional `-BundlePath`, `-Headerless`); **loose files in `final_round_bucket`:** `uv run python scripts/build_final_round_bucket_concat.py` (`--map-names`, `--sweep`, `--quiet-gaps`) | Heuristic forensics only — **not** a substitute for `check-all`. Narrative: **`docs/ops/DOC_BUNDLE_RECOVERY_PLAYBOOK.md`**. | | Day boundary (**carryover-sweep** / **eod-sync**) | `.\scripts\operator-day-ritual.ps1 -Mode Morning` / `-Mode Eod` | Session tokens in **`.cursor/rules/session-mode-keywords.mdc`**; git/gh path hints, not a substitute for `check-all`. | | PII fresh-clone self-audit (Windows) | `.\scripts\pii-fresh-clone-audit.ps1` (optional `-IncludeTalentGuards`, `-KeepClone`) | Not `check-all`; validates **public** clone + full-history guard; **temp-only** (primary-dev-workstation-safe). See **`docs/ops/PII_FRESH_CLONE_AUDIT.md`**, **`PRIMARY_WINDOWS_WORKSTATION_PROTECTION.md`**. Session **`pii-fresh-audit`**. | -| **Commit / PR** | `.\scripts\commit-or-pr.ps1` | When user asks to commit, push, or create a PR: use Preview first to see files and propose title/body, then `-Action Commit` or `-Action PR -RunTests`. Prefer `.\scripts\preview-commit.ps1` for preview-only; for PR with multi-line body use a one-off .ps1 that sets `$Title`/`$Body` and calls commit-or-pr.ps1, or `.\scripts\create-pr.ps1 -Title "..." -BodyFilePath path`. | +| **Commit / PR** | `.\scripts\commit-or-pr.ps1` | Local commit: Preview → `-Action Commit`. **Push/PR:** run full **`check-all`** first, then `-Action PR` (do not treat **`-RunTests`** as **`check-all`**). | -**Full workflow (check → pre-commit → commit → describe → safe synced PR)** — use in this order to save tokens and keep one source of truth: +**Full workflow (iterate → commit → check-all → push/PR)** — use in this order to save tokens: -1. **Check + pre-commit:** `.\scripts\check-all.ps1` or `./scripts/check-all.sh` (Ruff, format, markdown, pytest in one run). +1. **While editing:** `lint-only`, `quick-test`, or pre-commit hooks on commit — slice-appropriate; not pre-push proof. +1. **Before `git push` / PR:** `.\scripts\check-all.ps1` or `./scripts/check-all.sh` (exit 0). 1. **Preview:** `.\scripts\preview-commit.ps1` to see files and diff; propose a short title and bullet body. -1. **Commit + describe + PR:** `.\scripts\commit-or-pr.ps1 -Action PR -Title "..." -Body "..." -RunTests` (or `.\scripts\create-pr.ps1 -Title "..." -BodyFilePath path` / one-off .ps1 for long body). This commits, runs tests again, fetch+rebase if behind, push, open PR. +1. **Commit + describe + PR:** `.\scripts\commit-or-pr.ps1 -Action PR -Title "..." -Body "..."` after step 2 passed in the same session. -**Merge-ready slice (operator expectation):** When work is clearly ready to land (tests/docs committed, no open blocker), **run `check-all`**, then merge to `main` and **`git push origin main`** (or open/update PR) **without** asking permission to run the standard gate — the gate is the default safety step, not a decision point. +**Merge-ready slice (operator expectation):** When work is ready to land on GitHub, **run full `check-all`**, then push/open PR or merge — **without** asking permission to run the standard pre-push gate. Local commits between iterations do **not** require full `check-all`. Avoid raw `pytest`/`ruff` invocations when a script covers the need; scripts keep behaviour consistent and reduce back-and-forth. For **commit, description, push, and PR**, prefer the commit-or-pr workflow (and helpers) over ad-hoc `git add`/`git commit`/`git push` so behaviour is consistent and tokens are saved. See `.cursor/skills/token-aware-automation/SKILL.md` for when to use which. diff --git a/api/rbac.py b/api/rbac.py index 9b9561d0..297caad8 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,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: diff --git a/tests/test_rbac.py b/tests/test_rbac.py index 334ab56a..a0473307 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,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