From 98b783b6b50301aa1975e8f6a7ebba5f65ad2132 Mon Sep 17 00:00:00 2001 From: Chris Tonjes Date: Wed, 30 Sep 2026 20:54:12 -0400 Subject: [PATCH] Phase 0 milestone 0.6: 501-stub routers for remaining entities 26 stub entities (every matrix entity except Registrant, User, Role, AuditLog) get a router from framework/stub_router.py: one route per matrix action (List, Search, Create, View, Edit, Delete, Restore, VersionHistory). Each route runs the real RBAC check, writes an audit row, then returns 501. Source RFP C.5.7.27. - Registry in src/ypms/stubs.py; test fails if a matrix entity is neither real nor stubbed. - tests/test_stubs.py sweeps entity x role x route against the matrix JSON: 501 + exactly one audit row when granted, 403 + none when not. FERPA/CFSA stubs pinned closed to non-examiner roles. - Registrant restore now gated on the matrix's Restore action (was Edit, the vendored-factory convention); only the super-role holds it. - Test fixtures (roles, make_user, api) lifted to conftest; one app per session (route construction dominated runtime) and a shared password hash for minted-token users. 138 tests passing, ruff + mypy clean. Co-Authored-By: Claude Opus 5.5 --- docs/coverage/registrant.md | 4 +- src/ypms/framework/stub_router.py | 105 +++++++++++++++++++++++++++++ src/ypms/main.py | 3 + src/ypms/stubs.py | 61 +++++++++++++++++ src/ypms/youth/router.py | 5 +- tests/conftest.py | 81 +++++++++++++++++++++- tests/test_stubs.py | 104 ++++++++++++++++++++++++++++ tests/youth/test_registrant_api.py | 76 +++++++-------------- 8 files changed, 381 insertions(+), 58 deletions(-) create mode 100644 src/ypms/framework/stub_router.py create mode 100644 src/ypms/stubs.py create mode 100644 tests/test_stubs.py diff --git a/docs/coverage/registrant.md b/docs/coverage/registrant.md index f3b7038..036288d 100644 --- a/docs/coverage/registrant.md +++ b/docs/coverage/registrant.md @@ -17,7 +17,7 @@ evidence it. Auditors: file paths are relative to `ypms/`. | `POST` | `/registrants` | `Create` | 403 | — (create audit row) | | `PATCH` | `/registrants/{id}` | `Edit` | own row only (YOUTH_USER); else 404 | — (update audit row) | | `DELETE` | `/registrants/{id}` | `Delete` | 403 | — (delete audit row) | -| `POST` | `/registrants/{id}/restore` | `Edit` | 403 | — (restore audit row) | +| `POST` | `/registrants/{id}/restore` | `Restore` | 403 | — (restore audit row) | | `GET` | `/registrants/{id}/history` | `VersionHistory` | 403 | `pii_read` row, `entity_id` = id | "Portal-only" = every role the caller holds is in `{YOUTH_USER, PARENT_USER, HOST_USER}`. A caller with any staff role gets staff behaviour. PARENT_USER sees no rows until the Guardian link lands (Phase 1). No bulk endpoints for Registrant in Phase 0. @@ -34,7 +34,7 @@ evidence it. Auditors: file paths are relative to `ypms/`. | NIST AU-2, AU-3, AU-12 — every state change audited | Create / update / delete / restore emit audit rows with actor, before, after | `src/ypms/framework/crud.py` | `test_create_emits_redacted_audit_and_version_snapshot`, `test_update_bumps_version_and_audits`, `test_delete_requires_delete_permission_then_restore` | | NIST AU-14; source RFP `C.5.6.9` #5 — record access to sensitive data | GET, list, and history emit `pii_read` rows with actor and client IP | `src/ypms/framework/pii_audit.py`, `src/ypms/youth/router.py` | `test_get_emits_pii_read`, `test_list_emits_one_pii_read_with_ids`, `test_history_requires_version_history_permission` | | NIST AU-9 (protection of audit info) — no secrets in the audit trail | `__audit_redact__` replaces `ssn_encrypted` with `[REDACTED]` in audit before/after and version-history snapshots | `src/ypms/framework/crud.py` (`_to_dict`), `src/ypms/youth/models.py` | `test_create_emits_redacted_audit_and_version_snapshot`, `test_update_bumps_version_and_audits` | -| NIST AC-3, AC-6 — least privilege | Data-driven RBAC per route; history gated on `VersionHistory` (not `View`); delete requires `Delete` (OYP_STAFF lacks it) | `src/ypms/youth/router.py`, `src/ypms/seed/permissions_matrix.json` | `test_history_requires_version_history_permission`, `test_delete_requires_delete_permission_then_restore`, `test_host_user_denied` | +| NIST AC-3, AC-6 — least privilege | Data-driven RBAC per route; history gated on `VersionHistory` (not `View`); delete requires `Delete` and restore requires `Restore` (only the super-role holds either today) | `src/ypms/youth/router.py`, `src/ypms/seed/permissions_matrix.json` | `test_history_requires_version_history_permission`, `test_delete_requires_delete_permission_then_restore`, `test_host_user_denied` | | NIST AC-3 — row-level restriction for external users (Phase 0 minimum) | Portal-only callers limited to own row by `user_id`; out-of-scope → 404 (no id probing) | `src/ypms/youth/router.py` (`_get_scoped`, `_require_staff`) | `test_youth_can_read_and_edit_only_own_row`, `test_youth_blocked_from_staff_routes`, `test_parent_sees_nothing_until_guardian_link` | | NIST SI-10 — input validation | SSN format + SSA never-issued ranges; DOB not in future; ZIP, email, length limits; NOT NULL fields can't be PATCHed to null; `user_id` not writable | `src/ypms/youth/schemas.py` | `test_create_validation`, `test_update_rejects_null_on_required_field`, `test_update_cannot_set_user_id` | | ARCH §3 "minors first" — minor status | `minor_flag` derived from DOB at read time, never stored | `src/ypms/youth/models.py` (`minor_flag`, `age_on`) | `test_minor_flag_derived_from_dob`, `test_age_on_boundaries` | diff --git a/src/ypms/framework/stub_router.py b/src/ypms/framework/stub_router.py new file mode 100644 index 0000000..1f74d62 --- /dev/null +++ b/src/ypms/framework/stub_router.py @@ -0,0 +1,105 @@ +"""501-stub router factory — YPMS addition (milestone 0.6). + +Source RFP `C.5.7.27` requires access enforcement even on routes whose +business logic isn't built yet. Each stub router exposes one route per matrix +action; every route runs the real RBAC check, writes an audit row, and only +then returns 501. Denied calls get 403 exactly as they will once the route is +real, so permission-matrix gaps surface now rather than at go-live. + +Routes (entity ids are ints; `/search` is registered before `/{id}`): + + GET "" List + GET "/search" Search + POST "" Create + GET "/{id}" View + PATCH "/{id}" Edit + DELETE "/{id}" Delete + POST "/{id}/restore" Restore + GET "/{id}/history" VersionHistory + +Audit rows follow the vendored `router._attach_stub` convention: `action` is +the matrix action name, `after.status` is `stub_invoked`. + +No request bodies are read — stubs never see PII. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from fastapi import APIRouter, Depends, status +from sqlalchemy.orm import Session + +from ypms.audit import service as audit +from ypms.db import get_db +from ypms.security.rbac import require_permission +from ypms.users.models import User + +STUB_STATUS = "stub_invoked" + +# (method, path, matrix action) — order matters: /search before /{id}. +STUB_ROUTES: tuple[tuple[str, str, str], ...] = ( + ("GET", "", "List"), + ("GET", "/search", "Search"), + ("POST", "", "Create"), + ("GET", "/{id}", "View"), + ("PATCH", "/{id}", "Edit"), + ("DELETE", "/{id}", "Delete"), + ("POST", "/{id}/restore", "Restore"), + ("GET", "/{id}/history", "VersionHistory"), +) + + +def _body(entity: str, action: str, id: int | None) -> dict[str, Any]: + return { + "status": "not_implemented", + "action": action, + "entity": entity, + "id": id, + "phase": "Phase 0 stub — RBAC enforced, audit recorded, no business logic yet", + } + + +def _handler(entity: str, action: str, with_id: bool) -> Callable[..., dict[str, Any]]: + def _record(db: Session, actor: User, id: int | None) -> dict[str, Any]: + audit.record( + db, actor=actor, action=action, entity=entity, entity_id=id, + after={"status": STUB_STATUS}, + ) + db.commit() + return _body(entity, action, id) + + if with_id: + + def with_id_handler( + id: int, + db: Session = Depends(get_db), + actor: User = Depends(require_permission(action, entity)), + ) -> dict[str, Any]: + return _record(db, actor, id) + + return with_id_handler + + def no_id_handler( + db: Session = Depends(get_db), + actor: User = Depends(require_permission(action, entity)), + ) -> dict[str, Any]: + return _record(db, actor, None) + + return no_id_handler + + +def make_stub_router(*, entity: str, prefix: str) -> APIRouter: + router = APIRouter(prefix=prefix, tags=[f"{entity} (stub)"]) + for method, path, action in STUB_ROUTES: + handler = _handler(entity, action, with_id="{id}" in path) + handler.__name__ = f"stub_{action.lower()}_{entity}" + router.add_api_route( + path, + handler, + methods=[method], + status_code=status.HTTP_501_NOT_IMPLEMENTED, + name=handler.__name__, + ) + return router diff --git a/src/ypms/main.py b/src/ypms/main.py index 31a54e0..bf94c0e 100644 --- a/src/ypms/main.py +++ b/src/ypms/main.py @@ -17,6 +17,7 @@ from ypms.db import engine from ypms.logging import configure_logging, get_logger from ypms.middleware import RequestIdMiddleware from ypms.settings import get_settings +from ypms.stubs import stub_routers from ypms.youth.router import router as registrant_router @@ -44,6 +45,8 @@ def create_app() -> FastAPI: app.include_router(auth_router) app.include_router(registrant_router) + for stub in stub_routers(): + app.include_router(stub) @app.get("/healthz", tags=["system"]) def healthz() -> dict[str, str]: diff --git a/src/ypms/stubs.py b/src/ypms/stubs.py new file mode 100644 index 0000000..306f348 --- /dev/null +++ b/src/ypms/stubs.py @@ -0,0 +1,61 @@ +"""Registry of Phase 0 stub entities (milestone 0.6). + +Every entity in the permissions matrix that has no real router yet gets a +501-stub router at the path below. When a domain phase builds an entity for +real, remove it from this table and register its real router in `main.py`; +`tests/test_stubs.py` fails if an entity is in neither place. + +Not stubbed: Registrant (real, 0.5); User, Role, AuditLog (framework tables, +admin surfaces arrive with the admin module). + +Paths are spelled out rather than pluralized by rule so they read cleanly in +the OpenAPI spec and auditors can grep them. +""" + +from __future__ import annotations + +from fastapi import APIRouter + +from ypms.framework.stub_router import make_stub_router + +STUB_ENTITIES: dict[str, str] = { + # parent/ + "Guardian": "/guardians", + "ConsentRecord": "/consent-records", + # host/ + "Host": "/hosts", + "HostContact": "/host-contacts", + "Worksite": "/worksites", + "Position": "/positions", + "JobPosting": "/job-postings", + # program/ + "Program": "/programs", + "Cohort": "/cohorts", + "EligibilityRule": "/eligibility-rules", + # application/ + "Application": "/applications", + "EligibilityDecision": "/eligibility-decisions", + "Certification": "/certifications", + # enrollment/, orientation/ + "Orientation": "/orientations", + "Enrollment": "/enrollments", + # timekeeping/ + "TimeEntry": "/time-entries", + "Timesheet": "/timesheets", + # payroll/ + "PayrollBatch": "/payroll-batches", + "TaxDocument": "/tax-documents", + "TransportationVoucher": "/transportation-vouchers", + # cross-cutting + "Attachment": "/attachments", + "Note": "/notes", + "IntegrationEvent": "/integration-events", + "Notification": "/notifications", + # integrations/ — FERPA (OSSE) and DC Code Title 4 (CFSA) scoped + "EducationRecord": "/education-records", + "FosterIndicator": "/foster-indicators", +} + + +def stub_routers() -> list[APIRouter]: + return [make_stub_router(entity=e, prefix=p) for e, p in STUB_ENTITIES.items()] diff --git a/src/ypms/youth/router.py b/src/ypms/youth/router.py index 5d40523..6763ec2 100644 --- a/src/ypms/youth/router.py +++ b/src/ypms/youth/router.py @@ -12,7 +12,8 @@ Registrant needs three things the generic factory doesn't provide: Guardian link lands in Phase 1. Every other route is staff-only. Not a generic scope layer — that's Phase 2+ (PHASE0_STATUS.md gotcha 7). 3. **History gated on `VersionHistory`**, not `View` — snapshots carry the - full PII record over time. + full PII record over time. Restore is gated on the matrix's `Restore` + action (the generic factory uses `Edit`). No bulk endpoints for Registrant in Phase 0. """ @@ -160,7 +161,7 @@ def delete_registrant( def restore_registrant( id: int, db: Session = Depends(get_db), - actor: User = Depends(require_permission("Edit", ENTITY)), + actor: User = Depends(require_permission("Restore", ENTITY)), ) -> dict[str, Any]: _require_staff(actor) try: diff --git a/tests/conftest.py b/tests/conftest.py index e360d3e..120dd6f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,7 +12,8 @@ imports `ypms.main` (which instantiates the FastAPI app at import time). from __future__ import annotations import os -from collections.abc import Iterator +from collections.abc import Callable, Iterator +from typing import Any os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") os.environ.setdefault("JWT_SECRET", "test-secret") @@ -70,3 +71,81 @@ def client() -> Iterator[TestClient]: app = create_app() with TestClient(app) as c: yield c + + +# ---------------- seeded RBAC + authenticated API client ---------------- + +Headers = dict[str, str] + + +@pytest.fixture +def roles(db: Session) -> dict[str, Any]: + """All seed roles with the real permissions matrix loaded.""" + from ypms.seed.permissions import seed_permissions + from ypms.seed.run import seed_roles + + r = seed_roles(db) + seed_permissions(db, r) + db.flush() + return r + + +_HASH_CACHE: list[str] = [] + + +def _unused_password_hash(hasher: Callable[[str], str]) -> str: + # PBKDF2 is deliberately slow; these users authenticate by minted token, + # so one hash shared across the session is enough. + if not _HASH_CACHE: + _HASH_CACHE.append(hasher("unused-in-these-tests")) + return _HASH_CACHE[0] + + +@pytest.fixture +def make_user(db: Session, roles: dict[str, Any]) -> Callable[..., tuple[Any, Headers]]: + """Factory: make_user("name", "ROLE_A", ...) -> (User, auth headers).""" + from ypms.auth.security import create_access_token, hash_password + from ypms.users.models import ASSURANCE_STANDARD, User, UserRole + + def _make(username: str, *role_codes: str) -> tuple[User, Headers]: + user = User( + username=username, + password_hash=_unused_password_hash(hash_password), + is_active=True, + assurance_level=ASSURANCE_STANDARD, + ) + db.add(user) + db.flush() + for code in role_codes: + db.add(UserRole(user_id=user.id, role_id=roles[code].id)) + db.flush() + db.refresh(user) + token = create_access_token( + sub=username, uid=user.id, roles=list(role_codes), + assurance_level=ASSURANCE_STANDARD, mfa_verified=False, + ) + return user, {"Authorization": f"Bearer {token}"} + + return _make + + +@pytest.fixture(scope="session") +def _shared_app() -> Any: + """One app per session: building ~240 routes costs ~0.1s per create_app().""" + from ypms.main import create_app + + return create_app() + + +@pytest.fixture +def api(db: Session, _shared_app: Any) -> Iterator[TestClient]: + """TestClient on the shared app, wired to the per-test SQLite session.""" + from ypms.db import get_db + + def _override() -> Iterator[Session]: + yield db + + _shared_app.dependency_overrides[get_db] = _override + with TestClient(_shared_app) as c: + yield c + _shared_app.dependency_overrides.clear() diff --git a/tests/test_stubs.py b/tests/test_stubs.py new file mode 100644 index 0000000..cc5e627 --- /dev/null +++ b/tests/test_stubs.py @@ -0,0 +1,104 @@ +"""Milestone 0.6 — every stub route enforces RBAC and audits before its 501. + +The permissions matrix JSON is the oracle: for every stub entity, every seeded +role, and every stub route, the expected status is 501 if the role holds the +action (or is the super-role) and 403 otherwise. Each 501 must leave exactly +one audit row; each 403 must leave none. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ypms.audit.models import AuditLog +from ypms.framework.stub_router import STUB_ROUTES, STUB_STATUS +from ypms.security.rbac import ADMIN_ROLE +from ypms.seed.permissions import load_matrix +from ypms.seed.roles import ROLES +from ypms.stubs import STUB_ENTITIES + +MATRIX = load_matrix() +REAL_ENTITIES = {"Registrant", "User", "Role", "AuditLog"} +STUB_ID = 7 + + +def _url(prefix: str, path: str) -> str: + return prefix + path.replace("{id}", str(STUB_ID)) + + +def test_every_matrix_entity_is_real_or_stubbed() -> None: + assert set(STUB_ENTITIES) == set(MATRIX["entities"]) - REAL_ENTITIES + + +def test_stub_routes_cover_every_matrix_action() -> None: + assert {action for _, _, action in STUB_ROUTES} == set(MATRIX["actions"]) + + +def test_stub_paths_are_unique() -> None: + assert len(set(STUB_ENTITIES.values())) == len(STUB_ENTITIES) + + +def test_stub_requires_authentication(api: TestClient) -> None: + assert api.get("/guardians").status_code == 401 + + +@pytest.mark.parametrize("entity", sorted(STUB_ENTITIES)) +def test_stub_rbac_and_audit_follow_matrix( + entity: str, + api: TestClient, + db: Session, + make_user: Callable[..., tuple[Any, dict[str, str]]], +) -> None: + prefix = STUB_ENTITIES[entity] + for role in (r["code"] for r in ROLES): + _, headers = make_user(f"u.{role.lower()}", role) + granted = set(MATRIX["grants"].get(role, {}).get(entity, [])) + for method, path, action in STUB_ROUTES: + allowed = role == ADMIN_ROLE or action in granted + before = db.execute(select(AuditLog.id)).all() + + r = api.request(method, _url(prefix, path), headers=headers) + + assert r.status_code == (501 if allowed else 403), (role, method, path) + new = list( + db.execute(select(AuditLog).where(AuditLog.id.notin_([b[0] for b in before]))).scalars() + ) + if not allowed: + assert new == [], (role, action) + continue + body = r.json() + assert body["status"] == "not_implemented" + assert (body["entity"], body["action"]) == (entity, action) + (row,) = new + assert (row.entity, row.action) == (entity, action) + assert row.actor_username == f"u.{role.lower()}" + assert row.after == {"status": STUB_STATUS} + assert row.entity_id == (STUB_ID if "{id}" in path else None) + + +def test_search_not_captured_by_id_route( + api: TestClient, make_user: Callable[..., tuple[Any, dict[str, str]]] +) -> None: + _, admin = make_user("admin", ADMIN_ROLE) + assert api.get("/programs/search", headers=admin).json()["action"] == "Search" + + +@pytest.mark.parametrize("entity", ["EducationRecord", "FosterIndicator"]) +@pytest.mark.parametrize("role", ["OYP_STAFF", "PAYROLL_ADMIN", "REPORTING_ANALYST", "HOST_USER"]) +def test_ferpa_and_cfsa_stubs_closed_to_non_examiners( + entity: str, + role: str, + api: TestClient, + make_user: Callable[..., tuple[Any, dict[str, str]]], +) -> None: + """Pinned separately from the matrix sweep: ARCH §6.2 guardrail.""" + _, headers = make_user("u", role) + prefix = STUB_ENTITIES[entity] + for method, path, _ in STUB_ROUTES: + assert api.request(method, _url(prefix, path), headers=headers).status_code == 403 diff --git a/tests/youth/test_registrant_api.py b/tests/youth/test_registrant_api.py index 51a809a..b81dc03 100644 --- a/tests/youth/test_registrant_api.py +++ b/tests/youth/test_registrant_api.py @@ -6,7 +6,7 @@ grants), so these tests also pin the matrix's Registrant rows. from __future__ import annotations -from collections.abc import Callable, Iterator +from collections.abc import Callable from datetime import date from typing import Any @@ -16,15 +16,10 @@ from sqlalchemy import select from sqlalchemy.orm import Session from ypms.audit.models import AuditLog -from ypms.auth.security import create_access_token, hash_password -from ypms.db import get_db from ypms.framework.crud import REDACTED from ypms.framework.pii_audit import PII_READ_ACTION from ypms.framework.version_history import VersionHistory -from ypms.main import create_app -from ypms.seed.permissions import seed_permissions -from ypms.seed.run import seed_roles -from ypms.users.models import ASSURANCE_STANDARD, Role, User, UserRole +from ypms.users.models import User from ypms.youth.models import Registrant SSN = "123-45-6789" @@ -42,56 +37,11 @@ PAYLOAD: dict[str, Any] = { Headers = dict[str, str] -@pytest.fixture -def roles(db: Session) -> dict[str, Role]: - r = seed_roles(db) - seed_permissions(db, r) - db.flush() - return r - - -@pytest.fixture -def make_user(db: Session, roles: dict[str, Role]) -> Callable[..., tuple[User, Headers]]: - def _make(username: str, *role_codes: str) -> tuple[User, Headers]: - user = User( - username=username, - password_hash=hash_password("unused-in-these-tests"), - is_active=True, - assurance_level=ASSURANCE_STANDARD, - ) - db.add(user) - db.flush() - for code in role_codes: - db.add(UserRole(user_id=user.id, role_id=roles[code].id)) - db.flush() - db.refresh(user) - token = create_access_token( - sub=username, uid=user.id, roles=list(role_codes), - assurance_level=ASSURANCE_STANDARD, mfa_verified=False, - ) - return user, {"Authorization": f"Bearer {token}"} - - return _make - - @pytest.fixture def staff(make_user: Callable[..., tuple[User, Headers]]) -> tuple[User, Headers]: return make_user("oyp.staff", "OYP_STAFF") -@pytest.fixture -def api(db: Session) -> Iterator[TestClient]: - app = create_app() - - def _override() -> Iterator[Session]: - yield db - - app.dependency_overrides[get_db] = _override - with TestClient(app) as c: - yield c - app.dependency_overrides.clear() - - def _create(api: TestClient, headers: Headers, **overrides: Any) -> dict[str, Any]: r = api.post("/registrants", json={**PAYLOAD, **overrides}, headers=headers) assert r.status_code == 201, r.text @@ -256,7 +206,9 @@ def test_delete_requires_delete_permission_then_restore( assert r.status_code == 200 and r.json()["deleted_at"] is not None assert api.get(f"/registrants/{rid}", headers=staff[1]).status_code == 404 - r = api.post(f"/registrants/{rid}/restore", headers=staff[1]) + # Restore is its own matrix action; only the super-role holds it today. + assert api.post(f"/registrants/{rid}/restore", headers=staff[1]).status_code == 403 + r = api.post(f"/registrants/{rid}/restore", headers=admin) assert r.status_code == 200 and r.json()["deleted_at"] is None assert [a.action for a in db.execute(select(AuditLog).order_by(AuditLog.id)).scalars()][-2:] == [ "delete", @@ -365,3 +317,21 @@ def test_host_user_denied( _, hh = make_user("host", "HOST_USER") rid = _create(api, staff[1])["id"] assert api.get(f"/registrants/{rid}", headers=hh).status_code == 403 + + +def test_staff_guard_holds_even_if_matrix_grants_portal_role_list( + api: TestClient, + db: Session, + roles: dict[str, Any], + make_user: Callable[..., tuple[User, Headers]], +) -> None: + """Defense in depth: a future matrix edit granting YOUTH_USER List must not + expose every registrant.""" + from ypms.security.models import Permission + + db.add(Permission(role_id=roles["YOUTH_USER"].id, entity="Registrant", action="List")) + db.flush() + _, yh = make_user("youth", "YOUTH_USER") + r = api.get("/registrants", headers=yh) + assert r.status_code == 403 + assert r.json()["detail"]["reason"] == "staff-only route"