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 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 .schemas import AuditAdminDeltaResponse, AuditAdminItem, AuditAdminListResponse, AuditLogItemResponse, AuditLogListResponse 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 _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: 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, ) 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: 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, ) 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])