feat(security-audit): add tenant-safe audit query API

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))

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)

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),
     )

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),
+    )

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",))

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.

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]