Spaces:
Sleeping
Sleeping
File size: 3,922 Bytes
c91c7db 3493993 c91c7db 3493993 c91c7db 3493993 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | 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
|