diff --git a/api/deps_audit.py b/api/deps_audit.py new file mode 100644 index 0000000..07097dd --- /dev/null +++ b/api/deps_audit.py @@ -0,0 +1,32 @@ +"""Authorization context for tenant-safe audit reads.""" +from dataclasses import dataclass + +from fastapi import Header, HTTPException + +from audit.jwt_verify import TokenInvalid, verify_token + +PLATFORM_WIDE_PERMISSION = "manage_all_orgs" +ORGANIZATION_READ_ROLE = "org_admin" + + +@dataclass(frozen=True) +class AuditAccess: + platform_wide: bool + organization_id: str | None + + +def require_audit_read_access(authorization: str | None = Header(default=None)) -> AuditAccess: + if not authorization or not authorization.lower().startswith("bearer "): + raise HTTPException(401, "Missing or malformed Authorization header") + try: + payload = verify_token(authorization.split(" ", 1)[1].strip()) + except TokenInvalid as exc: + raise HTTPException(401, str(exc)) from exc + if PLATFORM_WIDE_PERMISSION in (payload.get("permissions") or []): + return AuditAccess(platform_wide=True, organization_id=None) + if ORGANIZATION_READ_ROLE not in (payload.get("org_role") or []): + raise HTTPException(403, "Audit read permission required") + org_id = payload.get("org_id") + if isinstance(org_id, (dict, list, tuple, set)) or org_id in (None, ""): + raise HTTPException(403, "Verified organization scope required") + return AuditAccess(platform_wide=False, organization_id=str(org_id)) diff --git a/api/main.py b/api/main.py index 56faf32..d5d278d 100644 --- a/api/main.py +++ b/api/main.py @@ -2,9 +2,11 @@ from api.routes_audit import router from api.routes_audit_events import router as audit_events_router +from api.routes_audit_safe import router as audit_safe_router from audit.config import AuditConfig app = FastAPI(title=f"OmniBioAI Security Audit — {AuditConfig.SERVICE_NAME}") app.include_router(router) app.include_router(audit_events_router) +app.include_router(audit_safe_router) diff --git a/api/routes_audit_events.py b/api/routes_audit_events.py index 3e434cb..8db3fde 100644 --- a/api/routes_audit_events.py +++ b/api/routes_audit_events.py @@ -1,11 +1,19 @@ from datetime import datetime from fastapi import APIRouter, Depends, Query +from fastapi.responses import JSONResponse +from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session from api.deps import require_platform_admin +from audit.source_semantics import available_query_evidence, unavailable_query_evidence from db.session import get_db -from schemas.audit import AuditEventListResponse +from schemas.audit import ( + AuditEventListResponse, + AuditEventOut, + FreshnessOut, + RetentionOut, +) from services import audit_query_service # Deliberately a separate router/module from routes_audit.py (/health, @@ -36,23 +44,39 @@ def list_audit_events( db: Session = Depends(get_db), # noqa: B008 -- FastAPI's own documented dependency-injection pattern, not a mutable-default bug _admin: dict = Depends(require_platform_admin), # noqa: B008 -- FastAPI's own documented dependency-injection pattern, not a mutable-default bug ) -> AuditEventListResponse: - rows, total = audit_query_service.list_audit_events( - db, - page=page, - page_size=page_size, - user_id=user_id, - service=service, - event_type=event_type, - decision=decision, - from_timestamp=from_timestamp, - to_timestamp=to_timestamp, - integrity_status=integrity_status, - ) + try: + rows, total = audit_query_service.list_audit_events( + db, page=page, page_size=page_size, user_id=user_id, service=service, + event_type=event_type, decision=decision, from_timestamp=from_timestamp, + to_timestamp=to_timestamp, integrity_status=integrity_status, + ) + except SQLAlchemyError: + evidence = unavailable_query_evidence() + return JSONResponse(status_code=503, content={ + "error": "AUDIT_SOURCE_UNAVAILABLE", "source": "security_audit", + "source_availability": evidence.availability.value, + "generated_at": evidence.generated_at.isoformat(), + "source_checked_at": evidence.source_checked_at.isoformat(), + "warnings": list(evidence.warnings), + }) total_pages = (total + page_size - 1) // page_size if total else 0 + evidence = available_query_evidence() + items = [] + for row in rows: + item = AuditEventOut.model_validate(row) + item.context = audit_query_service.project_safe_metadata(row) + items.append(item) return AuditEventListResponse( - items=rows, + source="security_audit", + items=items, total=total, page=page, page_size=page_size, total_pages=total_pages, + source_availability=evidence.availability, + generated_at=evidence.generated_at, + source_checked_at=evidence.source_checked_at, + freshness=FreshnessOut(status=evidence.freshness.status), + retention=RetentionOut(status=evidence.retention.status), + warnings=list(evidence.warnings), ) diff --git a/api/routes_audit_safe.py b/api/routes_audit_safe.py new file mode 100644 index 0000000..4bc6561 --- /dev/null +++ b/api/routes_audit_safe.py @@ -0,0 +1,72 @@ +from datetime import datetime +from typing import Literal + +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import JSONResponse +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session + +from api.deps_audit import AuditAccess, require_audit_read_access +from audit.source_semantics import available_query_evidence, unavailable_query_evidence +from db.session import get_db +from schemas.audit import ( + FreshnessOut, + RetentionOut, + SafeAuditEventListResponse, + SafeAuditEventOut, +) +from services import audit_query_service + +router = APIRouter() + + +@router.get("/audit/events/safe", response_model=SafeAuditEventListResponse) +def list_safe_audit_events( + page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), + user_id: str | None = Query(None, max_length=255), service: str | None = Query(None, max_length=255), + event_type: str | None = Query(None, max_length=255), decision: str | None = Query(None, max_length=64), + from_timestamp: datetime | None = Query(None), # noqa: B008 + to_timestamp: datetime | None = Query(None), # noqa: B008 + integrity_status: Literal["valid", "invalid", "unsigned", "unknown"] | None = Query(None), + organization_id: str | None = Query(None, max_length=255), + access: AuditAccess = Depends(require_audit_read_access), # noqa: B008 + db: Session = Depends(get_db), # noqa: B008 +) -> SafeAuditEventListResponse | JSONResponse: + if from_timestamp and to_timestamp and from_timestamp > to_timestamp: + raise HTTPException(422, "from_timestamp must not be after to_timestamp") + if not access.platform_wide and organization_id not in (None, access.organization_id): + raise HTTPException(403, "organization_id is outside verified scope") + effective_org = organization_id if access.platform_wide else access.organization_id + try: + rows, total = audit_query_service.list_safe_audit_events( + db, page=page, page_size=page_size, organization_id=effective_org, + platform_wide=access.platform_wide, user_id=user_id, service=service, + event_type=event_type, decision=decision, from_timestamp=from_timestamp, + to_timestamp=to_timestamp, integrity_status=integrity_status, + ) + except SQLAlchemyError: + evidence = unavailable_query_evidence() + return JSONResponse(status_code=503, content={ + "error": "AUDIT_SOURCE_UNAVAILABLE", "source": "security_audit", + "source_availability": evidence.availability.value, + "generated_at": evidence.generated_at.isoformat(), + "source_checked_at": evidence.source_checked_at.isoformat(), + "warnings": list(evidence.warnings), + }) + evidence = available_query_evidence() + items = [SafeAuditEventOut( + event_id=row.event_id, timestamp=row.timestamp, organization_id=row.organization_id, + tenant_scope=row.tenant_scope or "unknown", actor=row.user_id, event_type=row.event_type, + action=row.action, decision=row.decision, + integrity=row.integrity_status if row.integrity_status in {"valid", "invalid", "unsigned", "unknown"} else "unknown", + metadata=audit_query_service.project_safe_metadata(row), + ) for row in rows] + return SafeAuditEventListResponse( + source="security_audit", items=items, total=total, page=page, page_size=page_size, + total_pages=(total + page_size - 1) // page_size if total else 0, + source_availability=evidence.availability, generated_at=evidence.generated_at, + source_checked_at=evidence.source_checked_at, + freshness=FreshnessOut(**evidence.freshness.__dict__), + retention=RetentionOut(**evidence.retention.__dict__), + warnings=list(evidence.warnings), + ) diff --git a/audit/source_semantics.py b/audit/source_semantics.py new file mode 100644 index 0000000..90c05a0 --- /dev/null +++ b/audit/source_semantics.py @@ -0,0 +1,59 @@ +"""Explicit source evidence semantics; unknown evidence is never health.""" +from dataclasses import dataclass +from datetime import datetime, timezone +from enum import StrEnum + + +class SourceAvailability(StrEnum): + AVAILABLE = "AVAILABLE" + UNAVAILABLE = "UNAVAILABLE" + PARTIAL = "PARTIAL" + UNKNOWN = "UNKNOWN" + + +class FreshnessStatus(StrEnum): + CURRENT = "CURRENT" + STALE = "STALE" + UNKNOWN = "UNKNOWN" + + +class RetentionStatus(StrEnum): + KNOWN = "KNOWN" + UNKNOWN = "UNKNOWN" + + +@dataclass(frozen=True) +class RetentionEvidence: + status: RetentionStatus = RetentionStatus.UNKNOWN + retention_days: int | None = None + oldest_available_event_at: datetime | None = None + + +@dataclass(frozen=True) +class FreshnessEvidence: + status: FreshnessStatus = FreshnessStatus.UNKNOWN + last_persisted_event_at: datetime | None = None + ingestion_lag_seconds: float | None = None + + +@dataclass(frozen=True) +class SourceEvidence: + availability: SourceAvailability + generated_at: datetime + source_checked_at: datetime + freshness: FreshnessEvidence + retention: RetentionEvidence + warnings: tuple[str, ...] = () + + +def _evidence(availability: SourceAvailability, warnings: tuple[str, ...]) -> SourceEvidence: + now = datetime.now(timezone.utc) + return SourceEvidence(availability, now, now, FreshnessEvidence(), RetentionEvidence(), warnings) + + +def available_query_evidence() -> SourceEvidence: + return _evidence(SourceAvailability.AVAILABLE, ("freshness_unknown", "retention_unknown", "ingestion_lag_unknown")) + + +def unavailable_query_evidence() -> SourceEvidence: + return _evidence(SourceAvailability.UNAVAILABLE, ("durable_query_unavailable",)) diff --git a/docs/AUDIT_SOURCE_SEMANTICS_SAT4.md b/docs/AUDIT_SOURCE_SEMANTICS_SAT4.md new file mode 100644 index 0000000..50a495a --- /dev/null +++ b/docs/AUDIT_SOURCE_SEMANTICS_SAT4.md @@ -0,0 +1,43 @@ +# SAT-4 source evidence semantics + +`GET /audit/events/safe` reports durable SQL query evidence separately from +ingestion health. A completed SQL query is `source_availability: AVAILABLE`, +including an empty result. SQL failure is `UNAVAILABLE` with a normalized +`AUDIT_SOURCE_UNAVAILABLE` HTTP 503. + +Durable retention is `UNKNOWN`: `retention_days` and +`oldest_available_event_at` are null. `AUDIT_MAXLEN` is a Redis backlog cap, +not SQL retention. Freshness is also `UNKNOWN`; this service has no +authoritative heartbeat, as-of marker, lag metric, or stale threshold. It +never emits `CURRENT`, `STALE`, a retention duration, or inferred lag. + +Responses include `source`, `source_availability`, `generated_at`, +`source_checked_at`, `freshness`, `retention`, and safe warning codes. The two +timestamps are timezone-aware UTC values. Redis/consumer health does not make +a successful SQL query unavailable. + +SAT-3 owns `/audit/events/safe` authentication, verified tenant scope, SQL +tenant filtering, pagination, and allowlisted metadata. Organization callers +cannot see GLOBAL or UNKNOWN events; platform callers require +`manage_all_orgs`. SAT-4 supplies the evidence fields and failure behavior; +there is one safe response contract. + +No raw context, SQL, connection details, credentials, tokens, or stack traces +are returned. Stronger freshness and retention claims require authoritative +upstream evidence from the worker/deployment contract. SAT-2 producer changes +remain outside this worktree. + +## Live certification + +The deployed `GET /audit/events/safe` contract was live-certified with a +supported organization-owner identity. The owner received HTTP 200 and only +organization-scoped events for the verified organization; legacy UNKNOWN +events were present in storage but excluded. An ordinary authenticated +identity received 403, unauthenticated access received 401, and an explicit +cross-organization override was rejected with 403. Read-only method behavior +remained enforced. + +The live evidence does not establish CURRENT freshness, retention duration, or +GLOBAL visibility because no legitimate GLOBAL event was available. SAT-2 +producer limitations for TES and Workflow Bundles live fixtures remain +separate and are not represented as ecosystem-wide producer completeness. diff --git a/schemas/audit.py b/schemas/audit.py index 566bbc1..863c38d 100644 --- a/schemas/audit.py +++ b/schemas/audit.py @@ -3,6 +3,8 @@ from pydantic import BaseModel, ConfigDict +from audit.source_semantics import FreshnessStatus, RetentionStatus, SourceAvailability + class AuditEventOut(BaseModel): """Read-model for one audit_events row (db/models.py::AuditEventRecord). @@ -46,3 +48,50 @@ class AuditEventListResponse(BaseModel): page: int page_size: int total_pages: int + source: str + source_availability: SourceAvailability + generated_at: datetime + source_checked_at: datetime + freshness: "FreshnessOut" + retention: "RetentionOut" + warnings: list[str] + + +class FreshnessOut(BaseModel): + status: FreshnessStatus + last_persisted_event_at: datetime | None = None + ingestion_lag_seconds: float | None = None + + +class RetentionOut(BaseModel): + status: RetentionStatus + retention_days: int | None = None + oldest_available_event_at: datetime | None = None + + +class SafeAuditEventOut(BaseModel): + event_id: str + timestamp: datetime + organization_id: str | None = None + tenant_scope: str + actor: str | None = None + event_type: str + action: str + decision: str | None = None + integrity: str + metadata: dict[str, Any] + + +class SafeAuditEventListResponse(BaseModel): + source: str + items: list[SafeAuditEventOut] + total: int + page: int + page_size: int + total_pages: int + source_availability: SourceAvailability + generated_at: datetime + source_checked_at: datetime + freshness: FreshnessOut + retention: RetentionOut + warnings: list[str] diff --git a/services/audit_query_service.py b/services/audit_query_service.py index 70e308b..80281ea 100644 --- a/services/audit_query_service.py +++ b/services/audit_query_service.py @@ -60,3 +60,45 @@ def list_audit_events( ) return rows, total + + +SAFE_METADATA_KEYS = frozenset({"trace_id", "request_id", "workflow_id", "run_id", "resource_type", "resource_id", "backend"}) + + +def project_safe_metadata(row: AuditEventRecord) -> dict[str, object]: + context = row.context if isinstance(row.context, dict) else {} + return {key: context[key] for key in SAFE_METADATA_KEYS if key in context} + + +def list_safe_audit_events( + db: Session, *, page: int, page_size: int, organization_id: str | None, + platform_wide: bool, user_id: str | None = None, service: str | None = None, + event_type: str | None = None, decision: str | None = None, + from_timestamp: datetime | None = None, to_timestamp: datetime | None = None, + integrity_status: str | None = None, +) -> tuple[list[AuditEventRecord], int]: + """Tenant-safe SQL query; scope is applied before count and pagination.""" + query = db.query(AuditEventRecord) + if not platform_wide: + query = query.filter( + AuditEventRecord.organization_id == organization_id, + AuditEventRecord.tenant_scope == "organization", + ) + elif organization_id is not None: + query = query.filter(AuditEventRecord.organization_id == organization_id) + for column, value in ( + (AuditEventRecord.user_id, user_id), (AuditEventRecord.service, service), + (AuditEventRecord.event_type, event_type), (AuditEventRecord.decision, decision), + (AuditEventRecord.integrity_status, integrity_status), + ): + if value is not None: + query = query.filter(column == value) + if from_timestamp is not None: + query = query.filter(AuditEventRecord.timestamp >= from_timestamp) + if to_timestamp is not None: + query = query.filter(AuditEventRecord.timestamp <= to_timestamp) + total = query.count() + rows = query.order_by( + AuditEventRecord.timestamp.desc(), AuditEventRecord.event_id.desc() + ).offset((page - 1) * page_size).limit(page_size).all() + return rows, total diff --git a/tests/conftest.py b/tests/conftest.py index d0e88b6..94d891f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,7 @@ -import pytest from unittest.mock import AsyncMock, MagicMock, patch +import pytest + @pytest.fixture def mock_async_redis(): @@ -40,8 +41,8 @@ def db_session(): from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker - from db.base import Base import db.models # noqa: F401 -- registers AuditEventRecord on Base.metadata + from db.base import Base engine = create_engine( "sqlite:///:memory:", connect_args={"check_same_thread": False} @@ -66,16 +67,16 @@ def audit_events_client(monkeypatch): Yields (TestClient, SessionLocal) so tests can seed rows directly via the same SessionLocal the app's dependency override uses. """ + from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool - from fastapi.testclient import TestClient - from db.base import Base import db.models # noqa: F401 - from db.session import get_db - from audit import jwt_verify as jwt_verify_module from api.main import app + from audit import jwt_verify as jwt_verify_module + from db.base import Base + from db.session import get_db engine = create_engine( "sqlite:///:memory:", @@ -88,6 +89,7 @@ def audit_events_client(monkeypatch): # SSO Phase 2 PR3: decoding now happens in audit.jwt_verify, not # api.deps -- api.deps no longer has its own JWT_SECRET to patch. monkeypatch.setattr(jwt_verify_module, "JWT_SECRET", "test-secret") + monkeypatch.setattr(jwt_verify_module._blacklist, "exists", lambda _key: 0) def override_get_db(): db = TestingSessionLocal() diff --git a/tests/test_routes_audit_events.py b/tests/test_routes_audit_events.py index 4a608ad..a605e49 100644 --- a/tests/test_routes_audit_events.py +++ b/tests/test_routes_audit_events.py @@ -97,7 +97,11 @@ def test_response_contains_expected_fields(audit_events_client): resp = client.get("/audit/events", headers=_auth_headers()) body = resp.json() - assert set(body.keys()) == {"items", "total", "page", "page_size", "total_pages"} + assert set(body.keys()) == { + "source", "items", "total", "page", "page_size", "total_pages", + "source_availability", "generated_at", "source_checked_at", + "freshness", "retention", "warnings", + } item = body["items"][0] assert set(item.keys()) == { "event_id", "timestamp", "service", "event_type", "user_id", @@ -106,7 +110,7 @@ def test_response_contains_expected_fields(audit_events_client): "created_at", "integrity_status", } assert item["event_id"] == "evt-0" - assert item["context"] == {"i": 0} + assert item["context"] == {} # _seed() rows are constructed without an explicit integrity_status -- # the DB column's own server_default="unsigned" (0002_integrity_status) # applies, same as every real historical event before any producer diff --git a/tests/test_routes_audit_safe.py b/tests/test_routes_audit_safe.py new file mode 100644 index 0000000..4e6f23f --- /dev/null +++ b/tests/test_routes_audit_safe.py @@ -0,0 +1,97 @@ +from datetime import datetime, timedelta +from unittest.mock import patch + +import jwt +from sqlalchemy.exc import OperationalError + +from api.deps_audit import AuditAccess +from api.routes_audit_safe import list_safe_audit_events +from db.models import AuditEventRecord + +SECRET = "test-secret" + + +def _headers(*, org_id=None, org_role=None, permissions=None): + claims = {"sub": "actor", "roles": [], "org_role": org_role or [], "permissions": permissions or []} + if org_id is not None: + claims["org_id"] = org_id + return {"Authorization": f"Bearer {jwt.encode(claims, SECRET, algorithm='HS256')}"} + + +def _seed(factory): + db = factory() + base = datetime(2026, 1, 1, 12, 0, 0) # noqa: DTZ001 + for i, (scope, org) in enumerate((("organization", "1"), ("organization", "2"), ("global", None), ("unknown", None))): + db.add(AuditEventRecord( + event_id=f"safe-{i}", timestamp=base + timedelta(minutes=i), service="auth", + event_type="login", action="login", user_id=f"u-{i}", organization_id=org, + tenant_scope=scope, context={"request_id": f"r-{i}", "Authorization": "secret", "drop": "x"}, + )) + db.commit() + db.close() + + +def test_org_scope_excludes_other_global_and_unknown_and_is_sql_paginated(audit_events_client): + client, sessions = audit_events_client + _seed(sessions) + response = client.get("/audit/events/safe", headers=_headers(org_id="1", org_role=["org_admin"]), params={"page_size": 1}) + assert response.status_code == 200 + body = response.json() + assert body["source_availability"] == "AVAILABLE" + assert body["total"] == 1 + assert [item["event_id"] for item in body["items"]] == ["safe-0"] + assert body["items"][0]["metadata"] == {"request_id": "r-0"} + assert body["freshness"]["status"] == "UNKNOWN" + assert body["retention"] == {"status": "UNKNOWN", "retention_days": None, "oldest_available_event_at": None} + + +def test_platform_scope_sees_global_and_unknown(audit_events_client): + client, sessions = audit_events_client + _seed(sessions) + response = client.get("/audit/events/safe", headers=_headers(permissions=["manage_all_orgs"])) + assert response.status_code == 200 + assert response.json()["total"] == 4 + assert [x["event_id"] for x in response.json()["items"]] == ["safe-3", "safe-2", "safe-1", "safe-0"] + + +def test_org_query_override_cannot_widen(audit_events_client): + client, _ = audit_events_client + response = client.get("/audit/events/safe", headers=_headers(org_id="1", org_role=["org_admin"]), params={"organization_id": "2"}) + assert response.status_code == 403 + + +def test_empty_safe_result_is_available(audit_events_client): + client, _ = audit_events_client + response = client.get("/audit/events/safe", headers=_headers(org_id="1", org_role=["org_admin"])) + body = response.json() + assert response.status_code == 200 + assert body["items"] == [] + assert body["total"] == 0 + assert body["source_availability"] == "AVAILABLE" + + +def test_safe_auth_failures_and_validation(audit_events_client): + client, _ = audit_events_client + assert client.get("/audit/events/safe").status_code == 401 + assert client.get("/audit/events/safe", headers=_headers()).status_code == 403 + assert client.get("/audit/events/safe", headers=_headers(org_role=["org_admin"])).status_code == 403 + assert client.get("/audit/events/safe", headers=_headers(permissions=["manage_all_orgs"]), params={"page_size": 101}).status_code == 422 + + +def test_safe_database_failure_is_normalized_without_internal_details(): + with patch( + "api.routes_audit_safe.audit_query_service.list_safe_audit_events", + side_effect=OperationalError("SELECT audit_events", {}, Exception("db.internal")), + ): + response = list_safe_audit_events( + page=1, page_size=20, user_id=None, service=None, event_type=None, + decision=None, from_timestamp=None, to_timestamp=None, + integrity_status=None, organization_id=None, + access=AuditAccess(True, None), db=object() + ) + + assert response.status_code == 503 + body = response.body.decode() + assert "AUDIT_SOURCE_UNAVAILABLE" in body + assert "SELECT" not in body + assert "db.internal" not in body diff --git a/tests/test_source_semantics_sat4.py b/tests/test_source_semantics_sat4.py new file mode 100644 index 0000000..3e1e646 --- /dev/null +++ b/tests/test_source_semantics_sat4.py @@ -0,0 +1,20 @@ +from datetime import timezone + +from audit.source_semantics import ( + FreshnessStatus, + RetentionStatus, + SourceAvailability, + available_query_evidence, +) + + +def test_available_evidence_keeps_unknown_dimensions_unknown(): + evidence = available_query_evidence() + assert evidence.availability is SourceAvailability.AVAILABLE + assert evidence.freshness.status is FreshnessStatus.UNKNOWN + assert evidence.retention.status is RetentionStatus.UNKNOWN + assert evidence.retention.retention_days is None + assert evidence.retention.oldest_available_event_at is None + assert evidence.freshness.ingestion_lag_seconds is None + assert evidence.generated_at.tzinfo == timezone.utc + assert evidence.source_checked_at.tzinfo == timezone.utc