from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import and_, false, func, or_ from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal, has_scope, require_any_scope from govoplan_audit.backend.db.models import AuditLog from govoplan_core.audit.logging import ( AUDIT_MODULE_ID, AUDIT_SYSTEM_EVENTS_COLLECTION, AUDIT_TENANT_EVENTS_COLLECTION, audit_from_principal, ) from govoplan_core.core.access import CAPABILITY_ACCESS_ADMINISTRATION, AccessAdministration from govoplan_core.core.change_sequence import decode_sequence_watermark, encode_sequence_watermark, max_sequence_id, sequence_entries_since, sequence_watermark_is_expired from govoplan_core.core.pagination import KeysetCursorError, decode_keyset_cursor, encode_keyset_cursor, keyset_query_fingerprint from govoplan_core.core.runtime import get_registry from govoplan_core.db.session import get_session from govoplan_core.tenancy.scope import Tenant from govoplan_core.core.events import platform_event_outbox from govoplan_audit.backend.db.models import AuditEvidenceBundle from govoplan_audit.backend.evidence_bundles import ( EvidenceBundleError, build_evidence_bundle, canonical_sha256, configured_signing_key, normalize_evidence_reference, ) from govoplan_audit.backend.permissions import ( AUDIT_EVIDENCE_EXPORT_SCOPE, AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE, ) from .schemas import ( AuditAdminDeltaResponse, AuditAdminItem, AuditAdminListResponse, AuditLogItemResponse, AuditLogListResponse, EventDeliveryMetricsResponse, EventDeliveryReplayRequest, EventDeliveryReplayResponse, EvidenceBundleDownloadResponse, EvidenceBundleExportRequest, EvidenceBundleResponse, ) router = APIRouter(tags=["audit"]) AUDIT_ADMIN_CURSOR_SCOPE = "audit.admin" @dataclass(slots=True) class AuditAdminQueryContext: query: Any access_admin: AccessAdministration effective_scope: str resolved_tenant_id: str | None sort_column: Any order: Any total: int effective_page_size: int pages: int fingerprint: str def _access_administration() -> AccessAdministration: registry = get_registry() if registry is None or not registry.has_capability(CAPABILITY_ACCESS_ADMINISTRATION): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Access administration capability is not configured") capability = registry.require_capability(CAPABILITY_ACCESS_ADMINISTRATION) if not isinstance(capability, AccessAdministration): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Access administration capability is invalid") return capability def _resolve_tenant( session: Session, principal: ApiPrincipal, tenant_id: str | None, ) -> Tenant: target_id = tenant_id or principal.tenant_id if target_id != principal.tenant_id: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Switch to the target tenant before using tenant-administration endpoints.", ) tenant = session.get(Tenant, target_id) if tenant is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found") return tenant def _parse_audit_filter(value: str | None) -> tuple[str, str]: if not value: return "contains", "" if ":" not in value: return "contains", value.strip() operator, raw = value.split(":", 1) if operator not in {"contains", "eq", "before", "after", "gt", "gte", "lt", "lte"}: return "contains", value.strip() return operator, raw.strip() def _audit_text_filter(column, raw: str | None): operator, value = _parse_audit_filter(raw) if not value: return None normalized = value.casefold() text = func.lower(func.coalesce(column, "")) return text == normalized if operator == "eq" else text.contains(normalized) def _text_matches(candidate: str, *, operator: str, value: str) -> bool: normalized = value.casefold() text = candidate.casefold() return text == normalized if operator == "eq" else normalized in text def _audit_actor_filter(access_admin: AccessAdministration, session: Session, raw: str | None): operator, value = _parse_audit_filter(raw) if not value: return None user_ids = access_admin.user_ids_for_actor_filter(session, operator=operator, value=value) conditions = [] if user_ids: conditions.append(AuditLog.user_id.in_(user_ids)) if _text_matches("System", operator=operator, value=value): conditions.append(AuditLog.user_id.is_(None)) return or_(*conditions) if conditions else false() def _audit_time_filter(raw: str | None): operator, value = _parse_audit_filter(raw) if not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Invalid audit date filter.") from exc if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) else: parsed = parsed.astimezone(timezone.utc) if operator == "eq": if "T" not in value and " " not in value: return (AuditLog.created_at >= parsed) & (AuditLog.created_at < parsed + timedelta(days=1)) return AuditLog.created_at == parsed if operator in {"before", "lt"}: return AuditLog.created_at < parsed if operator == "lte": return AuditLog.created_at <= parsed if operator in {"after", "gt"}: return AuditLog.created_at > parsed if operator == "gte": return AuditLog.created_at >= parsed return AuditLog.created_at == parsed def _audit_delta_collections(effective_scope: str) -> tuple[str, ...]: if effective_scope == "system": return (AUDIT_SYSTEM_EVENTS_COLLECTION,) if effective_scope == "tenant": return (AUDIT_TENANT_EVENTS_COLLECTION,) return (AUDIT_TENANT_EVENTS_COLLECTION, AUDIT_SYSTEM_EVENTS_COLLECTION) def _audit_delta_watermark(session: Session, *, effective_scope: str, tenant_id: str | None) -> str: return encode_sequence_watermark( max_sequence_id( session, tenant_id=tenant_id if effective_scope == "tenant" else None, module_id=AUDIT_MODULE_ID, collections=_audit_delta_collections(effective_scope), ) ) def _audit_delta_entries( session: Session, *, effective_scope: str, tenant_id: str | None, since: str, limit: int, ): try: since_sequence = decode_sequence_watermark(since) except ValueError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc scoped_tenant_id = tenant_id if effective_scope == "tenant" else None collections = _audit_delta_collections(effective_scope) if sequence_watermark_is_expired( session, since=since_sequence, tenant_id=scoped_tenant_id, module_id=AUDIT_MODULE_ID, collections=collections, ): return None, False entries_plus_one = sequence_entries_since( session, since=since_sequence, tenant_id=scoped_tenant_id, module_id=AUDIT_MODULE_ID, collections=collections, limit=limit + 1, ) has_more = len(entries_plus_one) > limit return entries_plus_one[:limit], has_more def _audit_delta_response_watermark( session: Session, *, effective_scope: str, tenant_id: str | None, entries, has_more: bool, ) -> str: return encode_sequence_watermark(entries[-1].id) if has_more and entries else _audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=tenant_id) def _full_audit_delta_response( session: Session, *, context: AuditAdminQueryContext, page_query: Any, start_cursor: str | None, sort_by: str, sort_direction: str, ) -> AuditAdminDeltaResponse: rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all() rows = rows_plus_one[:context.effective_page_size] next_cursor = ( _audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint) if len(rows_plus_one) > context.effective_page_size and rows else None ) return AuditAdminDeltaResponse( total=context.total, page=1, page_size=context.effective_page_size, pages=context.pages, cursor=start_cursor, next_cursor=next_cursor, items=_audit_items(session, rows, context.access_admin), deleted=[], watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id), has_more=False, full=True, ) def _audit_items(session: Session, rows: list[AuditLog], access_admin: AccessAdministration) -> list[AuditAdminItem]: actor_email_by_user_id = access_admin.actor_email_by_user_id(session, {row.user_id for row in rows if row.user_id}) return [ AuditAdminItem( id=row.id, scope=row.scope, tenant_id=row.tenant_id, actor_email=actor_email_by_user_id.get(row.user_id) if row.user_id else None, action=row.action, object_type=row.object_type, object_id=row.object_id, details=row.details or {}, created_at=row.created_at, ) for row in rows ] def _audit_filter_params( *, filter_time: str | None, filter_actor: str | None, filter_action: str | None, filter_object: str | None, filter_tenant: str | None, ) -> dict[str, str]: return { "time": filter_time or "", "actor": filter_actor or "", "action": filter_action or "", "object": filter_object or "", "tenant": filter_tenant or "", } def _audit_cursor_fingerprint( *, effective_scope: str, tenant_id: str | None, page_size: int, sort_by: str, sort_direction: str, filters: dict[str, str], ) -> str: return keyset_query_fingerprint( AUDIT_ADMIN_CURSOR_SCOPE, { "scope": effective_scope, "tenant_id": tenant_id or "", "page_size": page_size, "sort_by": sort_by, "sort_direction": sort_direction, "filters": filters, }, ) def _audit_sort_value(row: AuditLog, sort_by: str): if sort_by == "time": return row.created_at if sort_by == "actor": return row.user_id or "System" if sort_by == "action": return row.action if sort_by == "object": return f"{row.object_type or ''} {row.object_id or ''}" if sort_by == "tenant": return row.tenant_id or "" raise KeysetCursorError("Unsupported audit sort column") def _audit_cursor_for_row(row: AuditLog, *, sort_by: str, sort_direction: str, fingerprint: str) -> str: return encode_keyset_cursor( AUDIT_ADMIN_CURSOR_SCOPE, fingerprint=fingerprint, values={ "id": row.id, "sort_by": sort_by, "sort_direction": sort_direction, "sort_value": _audit_sort_value(row, sort_by), }, ) def _audit_decode_sort_value(sort_by: str, value): if sort_by == "time": if not isinstance(value, str): raise KeysetCursorError("Invalid pagination cursor") try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise KeysetCursorError("Invalid pagination cursor") from exc if value is None: return "" return str(value) def _audit_cursor_condition(sort_column, *, sort_by: str, sort_direction: str, cursor_values: dict[str, object]): cursor_id = cursor_values.get("id") if not isinstance(cursor_id, str) or not cursor_id: raise KeysetCursorError("Invalid pagination cursor") sort_value = _audit_decode_sort_value(sort_by, cursor_values.get("sort_value")) primary_after = sort_column > sort_value if sort_direction == "asc" else sort_column < sort_value return or_(primary_after, and_(sort_column == sort_value, AuditLog.id < cursor_id)) def _prepare_audit_admin_query( session: Session, principal: ApiPrincipal, *, tenant_id: str | None, all_tenants: bool, audit_scope: str | None, limit: int, page_size: int | None, sort_by: str, sort_direction: str, filter_time: str | None, filter_actor: str | None, filter_action: str | None, filter_object: str | None, filter_tenant: str | None, ) -> AuditAdminQueryContext: effective_scope = audit_scope or ("all" if all_tenants else "tenant") if effective_scope not in {"tenant", "system", "all"}: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.") if sort_by not in {"time", "actor", "action", "object", "tenant"}: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Unsupported audit sort column.") if sort_direction not in {"asc", "desc"}: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit sort direction must be asc or desc.") query = session.query(AuditLog) resolved_tenant_id: str | None = None if effective_scope != "all": query = query.filter(AuditLog.scope == effective_scope) if effective_scope == "system": if not has_scope(principal, "system:audit:read"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read") elif effective_scope == "all" or all_tenants: if not has_scope(principal, "system:audit:read"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read") else: if not has_scope(principal, "audit:read"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: audit:read") tenant = _resolve_tenant(session, principal, tenant_id) resolved_tenant_id = tenant.id query = query.filter(AuditLog.tenant_id == tenant.id) object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "") access_admin = _access_administration() for condition in ( _audit_time_filter(filter_time), _audit_actor_filter(access_admin, session, filter_actor), _audit_text_filter(AuditLog.action, filter_action), _audit_text_filter(object_text, filter_object), _audit_text_filter(AuditLog.tenant_id, filter_tenant), ): if condition is not None: query = query.filter(condition) sort_columns = { "time": AuditLog.created_at, "actor": func.coalesce(AuditLog.user_id, "System"), "action": AuditLog.action, "object": object_text, "tenant": func.coalesce(AuditLog.tenant_id, ""), } sort_column = sort_columns[sort_by] order = sort_column.asc() if sort_direction == "asc" else sort_column.desc() total = query.count() effective_page_size = page_size or limit pages = max(1, (total + effective_page_size - 1) // effective_page_size) filters = _audit_filter_params( filter_time=filter_time, filter_actor=filter_actor, filter_action=filter_action, filter_object=filter_object, filter_tenant=filter_tenant, ) fingerprint = _audit_cursor_fingerprint( effective_scope=effective_scope, tenant_id=resolved_tenant_id, page_size=effective_page_size, sort_by=sort_by, sort_direction=sort_direction, filters=filters, ) return AuditAdminQueryContext( query=query, access_admin=access_admin, effective_scope=effective_scope, resolved_tenant_id=resolved_tenant_id, sort_column=sort_column, order=order, total=total, effective_page_size=effective_page_size, pages=pages, fingerprint=fingerprint, ) @router.get("/admin/audit", response_model=AuditAdminListResponse) def list_admin_audit( tenant_id: str | None = Query(default=None), all_tenants: bool = Query(default=False), audit_scope: str | None = Query(default=None, alias="scope"), limit: int = Query(default=100, ge=1, le=500), offset: int = Query(default=0, ge=0), page: int | None = Query(default=None, ge=1), page_size: int | None = Query(default=None, ge=1, le=500), cursor: str | None = Query(default=None), sort_by: str = Query(default="time"), sort_direction: str = Query(default="desc"), filter_time: str | None = Query(default=None), filter_actor: str | None = Query(default=None), filter_action: str | None = Query(default=None), filter_object: str | None = Query(default=None), filter_tenant: str | None = Query(default=None), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")), ): context = _prepare_audit_admin_query( session, principal, tenant_id=tenant_id, all_tenants=all_tenants, audit_scope=audit_scope, limit=limit, page_size=page_size, sort_by=sort_by, sort_direction=sort_direction, filter_time=filter_time, filter_actor=filter_actor, filter_action=filter_action, filter_object=filter_object, filter_tenant=filter_tenant, ) ordered_query = context.query.order_by(context.order, AuditLog.id.desc()) start_cursor: str | None = None if cursor: try: cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=context.fingerprint) if cursor_values is None: raise KeysetCursorError("Invalid pagination cursor") page_query = context.query.filter( _audit_cursor_condition(context.sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values) ) except KeysetCursorError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc effective_page = page or (offset // context.effective_page_size + 1) effective_offset = 0 start_cursor = cursor else: if page is not None or page_size is not None: effective_page = min(page or 1, context.pages) effective_offset = (effective_page - 1) * context.effective_page_size else: effective_page = offset // context.effective_page_size + 1 effective_offset = offset page_query = context.query if effective_offset > 0: previous_row = ordered_query.offset(effective_offset - 1).limit(1).first() if previous_row is not None: start_cursor = _audit_cursor_for_row(previous_row, sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint) rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).offset(effective_offset).limit(context.effective_page_size + 1).all() rows = rows_plus_one[:context.effective_page_size] next_cursor = ( _audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint) if len(rows_plus_one) > context.effective_page_size and rows else None ) return AuditAdminListResponse( total=context.total, page=effective_page, page_size=context.effective_page_size, pages=context.pages, cursor=start_cursor, next_cursor=next_cursor, items=_audit_items(session, rows, context.access_admin), ) @router.get("/admin/audit/delta", response_model=AuditAdminDeltaResponse) def list_admin_audit_delta( tenant_id: str | None = Query(default=None), all_tenants: bool = Query(default=False), audit_scope: str | None = Query(default=None, alias="scope"), limit: int = Query(default=100, ge=1, le=500), page_size: int | None = Query(default=None, ge=1, le=500), cursor: str | None = Query(default=None), sort_by: str = Query(default="time"), sort_direction: str = Query(default="desc"), filter_time: str | None = Query(default=None), filter_actor: str | None = Query(default=None), filter_action: str | None = Query(default=None), filter_object: str | None = Query(default=None), filter_tenant: str | None = Query(default=None), since: str | None = Query(default=None), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")), ): context = _prepare_audit_admin_query( session, principal, tenant_id=tenant_id, all_tenants=all_tenants, audit_scope=audit_scope, limit=limit, page_size=page_size, sort_by=sort_by, sort_direction=sort_direction, filter_time=filter_time, filter_actor=filter_actor, filter_action=filter_action, filter_object=filter_object, filter_tenant=filter_tenant, ) start_cursor: str | None = None page_query = context.query if cursor: try: cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=context.fingerprint) if cursor_values is None: raise KeysetCursorError("Invalid pagination cursor") page_query = context.query.filter( _audit_cursor_condition(context.sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values) ) start_cursor = cursor except KeysetCursorError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc if since is None: return _full_audit_delta_response( session, context=context, page_query=page_query, start_cursor=start_cursor, sort_by=sort_by, sort_direction=sort_direction, ) entries, has_more = _audit_delta_entries( session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id, since=since, limit=context.effective_page_size, ) if entries is None: return _full_audit_delta_response( session, context=context, page_query=page_query, start_cursor=start_cursor, sort_by=sort_by, sort_direction=sort_direction, ) changed_ids = [entry.resource_id for entry in entries if entry.resource_type == "audit_log"] rows = ( page_query.filter(AuditLog.id.in_(changed_ids)).order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size).all() if changed_ids else [] ) return AuditAdminDeltaResponse( total=context.total, page=1, page_size=context.effective_page_size, pages=context.pages, cursor=start_cursor, next_cursor=None, items=_audit_items(session, rows, context.access_admin), deleted=[], watermark=_audit_delta_response_watermark( session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id, entries=entries, has_more=has_more, ), has_more=has_more, full=False, ) @router.get("/audit", response_model=AuditLogListResponse) def list_audit_log( limit: int = Query(default=100, ge=1, le=500), offset: int = Query(default=0, ge=0), action: str | None = None, object_type: str | None = None, object_id: str | None = None, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_any_scope("audit:read")), ): if not has_scope(principal, "audit:read"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: audit:read") query = session.query(AuditLog).filter(AuditLog.tenant_id == principal.tenant_id) if action: query = query.filter(AuditLog.action == action) if object_type: query = query.filter(AuditLog.object_type == object_type) if object_id: query = query.filter(AuditLog.object_id == object_id) items = query.order_by(AuditLog.created_at.desc()).offset(offset).limit(limit).all() return AuditLogListResponse(items=[AuditLogItemResponse.model_validate(item) for item in items]) @router.get( "/admin/audit/event-delivery/metrics", response_model=EventDeliveryMetricsResponse, ) def event_delivery_metrics( session: Session = Depends(get_session), _principal: ApiPrincipal = Depends( require_any_scope("system:audit:read") ), ): outbox = platform_event_outbox(get_registry()) if outbox is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Durable platform event delivery is not configured", ) return EventDeliveryMetricsResponse.model_validate( outbox.delivery_metrics(session) ) @router.post( "/admin/audit/event-deliveries/{event_id}/{consumer_id}/replay", response_model=EventDeliveryReplayResponse, ) def replay_event_delivery( event_id: str, consumer_id: str, payload: EventDeliveryReplayRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends( require_any_scope("system:governance:write") ), ): outbox = platform_event_outbox(get_registry()) if outbox is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Durable platform event delivery is not configured", ) try: result = outbox.replay_delivery( session, event_id=event_id, consumer_id=consumer_id, operator_id=principal.account_id, reason=payload.reason, ) except LookupError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc), ) from exc except ValueError as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=str(exc), ) from exc audit_from_principal( session, principal, action="platform_event.delivery_replayed", scope="system", object_type="platform_event_delivery", object_id=f"{event_id}:{consumer_id}", details={ "event_id": event_id, "consumer_id": consumer_id, "reason": payload.reason, }, ) session.commit() return EventDeliveryReplayResponse.model_validate(result) @router.post( "/admin/audit/evidence-bundles", response_model=EvidenceBundleResponse, status_code=status.HTTP_201_CREATED, ) def export_evidence_bundle( payload: EvidenceBundleExportRequest, session: Session = Depends(get_session), principal: ApiPrincipal = Depends( require_any_scope( AUDIT_EVIDENCE_EXPORT_SCOPE, AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE, ) ), ): scope, tenant_id = _resolve_evidence_bundle_scope(session, principal, payload) records = _evidence_bundle_records( session, payload=payload, scope=scope, tenant_id=tenant_id, ) try: normalized_references = [ normalize_evidence_reference(item.model_dump(mode="json")) for item in payload.references ] except EvidenceBundleError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc request_payload = payload.model_dump(mode="json") request_payload["resolved_tenant_id"] = tenant_id request_payload["selection_complete"] = True row = AuditEvidenceBundle( scope=scope, tenant_id=tenant_id, requested_by=principal.account_id, status="pending", request_payload=request_payload, ) session.add(row) session.flush() generated_at = datetime.now(timezone.utc) try: key_id, key_path = ( configured_signing_key(required=True) if payload.sign else (None, None) ) bundle = build_evidence_bundle( records, bundle_id=row.id, generated_at=generated_at, scope={"kind": scope, "tenant_id": tenant_id}, request=_evidence_manifest_request(request_payload), references=normalized_references, signing_key_id=key_id if payload.sign else None, signing_private_key_path=key_path if payload.sign else None, ) except EvidenceBundleError as exc: row.status = "failed" row.error_code = "evidence_bundle_generation_failed" session.add(row) session.commit() raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc row.status = "ready" row.bundle_payload = bundle row.bundle_sha256 = canonical_sha256(bundle) row.record_count = len(records) row.reference_count = len(payload.references) row.generated_at = generated_at session.add(row) audit_from_principal( session, principal, action="audit.evidence_bundle.generated", scope="system" if scope in {"system", "all"} else "tenant", object_type="audit_evidence_bundle", object_id=row.id, details={ "bundle_sha256": row.bundle_sha256, "record_count": row.record_count, "reference_count": row.reference_count, "scope": scope, }, ) session.commit() return _evidence_bundle_response(row) @router.get( "/admin/audit/evidence-bundles/{bundle_id}", response_model=EvidenceBundleResponse, ) def get_evidence_bundle( bundle_id: str, session: Session = Depends(get_session), principal: ApiPrincipal = Depends( require_any_scope( AUDIT_EVIDENCE_EXPORT_SCOPE, AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE, ) ), ): row = _authorized_evidence_bundle(session, principal, bundle_id) return _evidence_bundle_response(row) @router.get( "/admin/audit/evidence-bundles/{bundle_id}/download", response_model=EvidenceBundleDownloadResponse, ) def download_evidence_bundle( bundle_id: str, session: Session = Depends(get_session), principal: ApiPrincipal = Depends( require_any_scope( AUDIT_EVIDENCE_EXPORT_SCOPE, AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE, ) ), ): row = _authorized_evidence_bundle(session, principal, bundle_id) if row.status != "ready" or not isinstance(row.bundle_payload, dict): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Evidence bundle is not ready for download.", ) if not row.bundle_sha256 or canonical_sha256(row.bundle_payload) != row.bundle_sha256: row.status = "failed" row.error_code = "evidence_bundle_storage_integrity_failed" session.add(row) audit_from_principal( session, principal, action="audit.evidence_bundle.integrity_failed", scope="system" if row.scope in {"system", "all"} else "tenant", object_type="audit_evidence_bundle", object_id=row.id, details={"expected_bundle_sha256": row.bundle_sha256}, ) session.commit() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Stored evidence bundle failed its canonical integrity check.", ) row.downloaded_at = datetime.now(timezone.utc) session.add(row) audit_from_principal( session, principal, action="audit.evidence_bundle.downloaded", scope="system" if row.scope in {"system", "all"} else "tenant", object_type="audit_evidence_bundle", object_id=row.id, details={"bundle_sha256": row.bundle_sha256}, ) session.commit() return EvidenceBundleDownloadResponse(bundle=row.bundle_payload) def _resolve_evidence_bundle_scope( session: Session, principal: ApiPrincipal, payload: EvidenceBundleExportRequest, ) -> tuple[str, str | None]: if payload.scope in {"system", "all"}: if not has_scope(principal, AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE}", ) if payload.tenant_id is not None: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="System and all-scope evidence bundles do not accept a tenant id.", ) return payload.scope, None if not has_scope(principal, AUDIT_EVIDENCE_EXPORT_SCOPE): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {AUDIT_EVIDENCE_EXPORT_SCOPE}", ) tenant = _resolve_tenant(session, principal, payload.tenant_id) return "tenant", tenant.id def _evidence_bundle_records( session: Session, *, payload: EvidenceBundleExportRequest, scope: str, tenant_id: str | None, ) -> list[AuditLog]: query = session.query(AuditLog) if scope == "tenant": query = query.filter(AuditLog.scope == "tenant", AuditLog.tenant_id == tenant_id) elif scope == "system": query = query.filter(AuditLog.scope == "system") if payload.since is not None: query = query.filter(AuditLog.created_at >= payload.since) if payload.until is not None: query = query.filter(AuditLog.created_at <= payload.until) if payload.record_ids: query = query.filter(AuditLog.id.in_(payload.record_ids)) if payload.action: query = query.filter(AuditLog.action == payload.action) if payload.object_type: query = query.filter(AuditLog.object_type == payload.object_type) if payload.object_id: query = query.filter(AuditLog.object_id == payload.object_id) records = ( query.order_by(AuditLog.created_at.asc(), AuditLog.id.asc()) .limit(payload.max_records + 1) .all() ) if len(records) > payload.max_records: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Evidence selection exceeds the bounded record limit; narrow the requested scope.", ) if payload.record_ids and {item.id for item in records} != set(payload.record_ids): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="One or more requested audit records are unavailable in the authorized scope.", ) return records def _authorized_evidence_bundle( session: Session, principal: ApiPrincipal, bundle_id: str, ) -> AuditEvidenceBundle: row = session.get(AuditEvidenceBundle, bundle_id) if row is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Evidence bundle not found.") if row.scope in {"system", "all"}: allowed = has_scope(principal, AUDIT_SYSTEM_EVIDENCE_EXPORT_SCOPE) else: allowed = ( has_scope(principal, AUDIT_EVIDENCE_EXPORT_SCOPE) and row.tenant_id == principal.tenant_id ) if not allowed: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Evidence bundle not found.") return row def _evidence_manifest_request(request_payload: dict[str, Any]) -> dict[str, Any]: result = dict(request_payload) references = result.pop("references", []) result["reference_ids"] = [ item.get("reference_id") for item in references if isinstance(item, dict) and item.get("reference_id") ] return result def _evidence_bundle_response(row: AuditEvidenceBundle) -> EvidenceBundleResponse: return EvidenceBundleResponse( id=row.id, scope=row.scope, tenant_id=row.tenant_id, status=row.status, bundle_sha256=row.bundle_sha256, record_count=row.record_count, reference_count=row.reference_count, generated_at=row.generated_at, downloaded_at=row.downloaded_at, error_code=row.error_code, created_at=row.created_at, download_url=( f"/api/v1/admin/audit/evidence-bundles/{row.id}/download" if row.status == "ready" else None ), )