231 lines
8.7 KiB
Python
231 lines
8.7 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""Compliance & Audit API endpoints.
|
|
|
|
Provides three views:
|
|
- AI Data-Flow Log (Tab A)
|
|
- Security/GDPR Audit Log (Tab B)
|
|
- Aggregated Audit Statistics (Tab C)
|
|
|
|
RBAC: mandate-admin or compliance-viewer role required.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Path, status
|
|
from starlette.requests import Request
|
|
|
|
from modules.auth import limiter, getCurrentUser, getRequestContext, RequestContext
|
|
from modules.datamodels.datamodelUam import User
|
|
from modules.shared.i18nRegistry import apiRouteContext
|
|
|
|
logger = logging.getLogger(__name__)
|
|
routeApiMsg = apiRouteContext("routeAudit")
|
|
|
|
router = APIRouter(prefix="/api/audit", tags=["Audit"])
|
|
|
|
|
|
def _requireAuditAccess(context: RequestContext):
|
|
"""Raise 403 unless user has mandate-admin or compliance-viewer access."""
|
|
from modules.auth.authentication import _hasSysAdminRole
|
|
if _hasSysAdminRole(str(context.user.id)):
|
|
return
|
|
|
|
from modules.interfaces.interfaceDbApp import getInterface
|
|
appIf = getInterface(context.user, mandateId=str(context.mandateId) if context.mandateId else None)
|
|
mandateId = str(context.mandateId) if context.mandateId else None
|
|
if not mandateId:
|
|
raise HTTPException(status_code=403, detail=routeApiMsg("Mandanten-Kontext erforderlich"))
|
|
|
|
from modules.datamodels.datamodelRbac import AccessRuleContext
|
|
permissions = appIf.rbac.getUserPermissions(
|
|
context.user, AccessRuleContext.UI, "ui.system.complianceAudit",
|
|
mandateId=mandateId,
|
|
)
|
|
if not permissions or not permissions.view:
|
|
raise HTTPException(status_code=403, detail=routeApiMsg("Kein Zugriff auf Audit-Daten"))
|
|
|
|
|
|
# ── Tab A: AI Data-Flow Log ──
|
|
|
|
@router.get("/ai-log")
|
|
@limiter.limit("120/minute")
|
|
async def getAiAuditLog(
|
|
request: Request,
|
|
context: RequestContext = Depends(getRequestContext),
|
|
userId: Optional[str] = Query(None),
|
|
featureInstanceId: Optional[str] = Query(None),
|
|
aiModel: Optional[str] = Query(None),
|
|
dateFrom: Optional[float] = Query(None, description="UTC epoch seconds"),
|
|
dateTo: Optional[float] = Query(None, description="UTC epoch seconds"),
|
|
limit: int = Query(50, ge=1, le=500),
|
|
offset: int = Query(0, ge=0),
|
|
):
|
|
_requireAuditAccess(context)
|
|
mandateId = str(context.mandateId) if context.mandateId else ""
|
|
if not mandateId:
|
|
raise HTTPException(status_code=400, detail=routeApiMsg("Mandanten-ID erforderlich"))
|
|
|
|
from modules.shared.aiAuditLogger import aiAuditLogger
|
|
return aiAuditLogger.getAiAuditLogs(
|
|
mandateId,
|
|
userId=userId,
|
|
featureInstanceId=featureInstanceId,
|
|
aiModel=aiModel,
|
|
fromTimestamp=dateFrom,
|
|
toTimestamp=dateTo,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
|
|
|
|
@router.get("/ai-log/{entryId}/content")
|
|
@limiter.limit("60/minute")
|
|
async def getAiAuditEntryContent(
|
|
request: Request,
|
|
entryId: str = Path(...),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
):
|
|
_requireAuditAccess(context)
|
|
mandateId = str(context.mandateId) if context.mandateId else ""
|
|
|
|
from modules.shared.aiAuditLogger import aiAuditLogger
|
|
result = aiAuditLogger.getAiAuditEntryContent(entryId, mandateId)
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail=routeApiMsg("Audit-Eintrag nicht gefunden"))
|
|
|
|
_phRx = re.compile(r"\[([a-z]+)\.([a-f0-9-]{36})\]")
|
|
combinedText = (result.get("contentInputFull") or "") + (result.get("contentOutputFull") or "")
|
|
phIds = set(m.group(2) for m in _phRx.finditer(combinedText))
|
|
mappings = []
|
|
|
|
if phIds:
|
|
try:
|
|
from modules.features.neutralization.interfaceFeatureNeutralizer import getInterface as getNeutIf
|
|
neutIf = getNeutIf(context.user, mandateId=mandateId, featureInstanceId=None)
|
|
for phId in phIds:
|
|
attr = neutIf.getAttributeById(phId)
|
|
if attr:
|
|
mappings.append({
|
|
"id": attr.get("id", ""),
|
|
"originalText": attr.get("originalText", ""),
|
|
"patternType": attr.get("patternType", ""),
|
|
})
|
|
except Exception as mapErr:
|
|
logger.warning(f"Could not resolve neutralization mappings for audit entry {entryId}: {mapErr}")
|
|
|
|
result["neutralizationMappings"] = mappings
|
|
return result
|
|
|
|
|
|
# ── Tab B: Security / GDPR Audit Log ──
|
|
|
|
@router.get("/log")
|
|
@limiter.limit("120/minute")
|
|
async def getAuditLog(
|
|
request: Request,
|
|
context: RequestContext = Depends(getRequestContext),
|
|
userId: Optional[str] = Query(None),
|
|
category: Optional[str] = Query(None),
|
|
action: Optional[str] = Query(None),
|
|
dateFrom: Optional[float] = Query(None),
|
|
dateTo: Optional[float] = Query(None),
|
|
limit: int = Query(100, ge=1, le=500),
|
|
offset: int = Query(0, ge=0),
|
|
):
|
|
_requireAuditAccess(context)
|
|
mandateId = str(context.mandateId) if context.mandateId else None
|
|
|
|
from modules.shared.auditLogger import audit_logger
|
|
records = audit_logger.getAuditLogs(
|
|
userId=userId,
|
|
mandateId=mandateId,
|
|
category=category,
|
|
action=action,
|
|
fromTimestamp=dateFrom,
|
|
toTimestamp=dateTo,
|
|
limit=limit + offset + 1,
|
|
)
|
|
totalItems = len(records)
|
|
page = records[offset: offset + limit]
|
|
return {"items": page, "totalItems": totalItems}
|
|
|
|
|
|
# ── Tab C: Audit Statistics ──
|
|
|
|
@router.get("/stats")
|
|
@limiter.limit("60/minute")
|
|
async def getAuditStats(
|
|
request: Request,
|
|
context: RequestContext = Depends(getRequestContext),
|
|
timeRange: int = Query(30, ge=1, le=365, description="Days to aggregate"),
|
|
groupBy: str = Query("model", description="Grouping: model, user, feature, day"),
|
|
):
|
|
_requireAuditAccess(context)
|
|
mandateId = str(context.mandateId) if context.mandateId else ""
|
|
if not mandateId:
|
|
raise HTTPException(status_code=400, detail=routeApiMsg("Mandanten-ID erforderlich"))
|
|
|
|
from modules.shared.aiAuditLogger import aiAuditLogger
|
|
return aiAuditLogger.getAiAuditStats(mandateId, timeRangeDays=timeRange, groupBy=groupBy)
|
|
|
|
|
|
# ── Tab D: Neutralization Mappings ──
|
|
|
|
@router.get("/neutralization-mappings")
|
|
@limiter.limit("120/minute")
|
|
async def getNeutralizationMappings(
|
|
request: Request,
|
|
context: RequestContext = Depends(getRequestContext),
|
|
limit: int = Query(200, ge=1, le=2000),
|
|
offset: int = Query(0, ge=0),
|
|
):
|
|
_requireAuditAccess(context)
|
|
mandateId = str(context.mandateId) if context.mandateId else ""
|
|
if not mandateId:
|
|
raise HTTPException(status_code=400, detail=routeApiMsg("Mandanten-ID erforderlich"))
|
|
|
|
try:
|
|
from modules.features.neutralization.interfaceFeatureNeutralizer import getInterface as getNeutIf
|
|
neutIf = getNeutIf(context.user, mandateId=mandateId, featureInstanceId=None)
|
|
attrs = neutIf.getNeutralizationAttributes()
|
|
items = [a.model_dump() if hasattr(a, "model_dump") else dict(a) for a in attrs]
|
|
for item in items:
|
|
pType = item.get("patternType", "")
|
|
uid = item.get("id", "")
|
|
item["placeholder"] = f"[{pType}.{uid}]" if pType and uid else uid
|
|
items.sort(key=lambda r: (r.get("patternType", ""), r.get("originalText", "")))
|
|
totalItems = len(items)
|
|
page = items[offset: offset + limit]
|
|
return {"items": page, "totalItems": totalItems}
|
|
except Exception as e:
|
|
logger.error(f"Failed to load neutralization mappings: {e}")
|
|
raise HTTPException(status_code=500, detail=routeApiMsg("Fehler beim Laden der Neutralisierungs-Zuordnungen"))
|
|
|
|
|
|
@router.delete("/neutralization-mappings/{mappingId}")
|
|
@limiter.limit("60/minute")
|
|
async def deleteNeutralizationMapping(
|
|
request: Request,
|
|
mappingId: str = Path(...),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
):
|
|
_requireAuditAccess(context)
|
|
mandateId = str(context.mandateId) if context.mandateId else ""
|
|
if not mandateId:
|
|
raise HTTPException(status_code=400, detail=routeApiMsg("Mandanten-ID erforderlich"))
|
|
|
|
try:
|
|
from modules.features.neutralization.interfaceFeatureNeutralizer import getInterface as getNeutIf
|
|
neutIf = getNeutIf(context.user, mandateId=mandateId, featureInstanceId=None)
|
|
deleted = neutIf.deleteAttributeById(mappingId)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail=routeApiMsg("Zuordnung nicht gefunden"))
|
|
return {"success": True, "id": mappingId}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete neutralization mapping {mappingId}: {e}")
|
|
raise HTTPException(status_code=500, detail=routeApiMsg("Fehler beim Löschen der Zuordnung"))
|