app/security/audit.py

from __future__ import annotations

from typing import Any

from sqlalchemy import select

from app.security.database import SecurityDatabase
from app.security.models import AuditEvent, AuditLog


class AuditService:
    def __init__(self, database: SecurityDatabase) -> None:
        self.database = database

    async def record(
        self,
        *,
        request_id: str,
        api_key_id: str,
        key_name: str,
        ip_address: str | None,
        user_agent: str | None,
        endpoint: str,
        http_method: str,
        response_code: int,
        processing_time_ms: int,
        bytes_uploaded: int,
        bytes_downloaded: int,
    ) -> None:
        async with self.database.session() as session:
            session.add(
                AuditLog(
                    request_id=request_id,
                    api_key_id=api_key_id,
                    key_name=key_name,
                    ip_address=ip_address,
                    user_agent=user_agent,
                    endpoint=endpoint,
                    http_method=http_method,
                    response_code=response_code,
                    processing_time_ms=processing_time_ms,
                    bytes_uploaded=bytes_uploaded,
                    bytes_downloaded=bytes_downloaded,
                )
            )
            await session.commit()

    async def list(self, *, offset: int = 0, limit: int = 100) -> list[AuditLog]:
        async with self.database.session() as session:
            return list(
                (
                    await session.scalars(
                        select(AuditLog)
                        .order_by(AuditLog.created_at.desc())
                        .offset(offset)
                        .limit(limit)
                    )
                ).all()
            )

    async def record_event(
        self,
        *,
        workspace_id: str,
        user_id: str,
        event_type: str,
        entity_type: str,
        entity_id: str,
        api_key_id: str | None = None,
        request_id: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """Persist a safe domain event through the existing audit boundary."""

        async with self.database.tenant_session(
            workspace_id=workspace_id, user_id=user_id
        ) as session:
            session.add(
                AuditEvent(
                    workspace_id=workspace_id,
                    actor_user_id=user_id,
                    api_key_id=api_key_id,
                    event_type=event_type[:100],
                    entity_type=entity_type[:64],
                    entity_id=entity_id,
                    request_id=request_id[:64] if request_id else None,
                    metadata_json=self._safe_metadata(metadata or {}),
                )
            )
            await session.commit()

    @staticmethod
    def _safe_metadata(metadata: dict[str, Any]) -> dict[str, object]:
        """Keep audit metadata flat, bounded, and free of credential-like keys."""

        blocked = ("token", "secret", "credential", "authorization", "password", "key")
        result: dict[str, object] = {}
        for raw_key, raw_value in list(metadata.items())[:32]:
            key = str(raw_key)[:64]
            if any(fragment in key.casefold() for fragment in blocked):
                continue
            if raw_value is None or isinstance(raw_value, (bool, int, float)):
                result[key] = raw_value
            elif isinstance(raw_value, str):
                result[key] = raw_value[:256]
            elif isinstance(raw_value, (list, tuple)):
                result[key] = [
                    value[:128] if isinstance(value, str) else value
                    for value in raw_value[:32]
                    if value is None or isinstance(value, (bool, int, float, str))
                ]
        return result