from __future__ import annotations from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import false, func, or_ from sqlalchemy.orm import Session from govoplan_access.backend.auth.dependencies import ApiPrincipal, has_scope, require_any_scope from govoplan_audit.backend.db.models import AuditLog from govoplan_core.core.access import CAPABILITY_ACCESS_ADMINISTRATION, AccessAdministration from govoplan_core.core.runtime import get_registry from govoplan_core.db.session import get_session from govoplan_tenancy.backend.db.models import Tenant from .schemas import AuditAdminItem, AuditAdminListResponse, AuditLogItemResponse, AuditLogListResponse router = APIRouter(tags=["audit"]) 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_ENTITY, 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 @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), 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")), ): 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_ENTITY, 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_ENTITY, detail="Unsupported audit sort column.") if sort_direction not in {"asc", "desc"}: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Audit sort direction must be asc or desc.") query = session.query(AuditLog) 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) query = query.filter(AuditLog.tenant_id == tenant.id) object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "") access_admin = _access_administration() filters = [ _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), ] for condition in filters: if condition is not None: query = query.filter(condition) total = query.count() effective_page_size = page_size or limit pages = max(1, (total + effective_page_size - 1) // effective_page_size) if page is not None or page_size is not None: effective_page = min(page or 1, pages) effective_offset = (effective_page - 1) * effective_page_size else: effective_page = offset // effective_page_size + 1 effective_offset = offset 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() rows = query.order_by(order, AuditLog.id.desc()).offset(effective_offset).limit(effective_page_size).all() 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 AuditAdminListResponse( total=total, page=effective_page, page_size=effective_page_size, pages=pages, items=[ 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 ], ) @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])