Feat/issue 119 audit logs append only

app/core/exceptions.py

@@ -441,3 +441,19 @@ def __init__(
             payload=payload,
             code='INVALID_MFA_PASSWORD',
         )
+
+
+class AuditImmutabilityError(AppError):
+    """Raised when an operation attempts to modify or delete an audit log."""
+
+    def __init__(
+        self,
+        message: str = 'A tabela audit_logs é append-only. Operações de UPDATE ou DELETE são proibidas.',
+        payload: dict[str, Any] | None = None,
+    ):
+        super().__init__(
+            message=message,
+            status_code=403,
+            payload=payload,
+            code='AUDIT_LOG_IMMUTABLE',
+        )

app/core/security/__init__.py

@@ -1,3 +1,33 @@
-"""Security subpackage (hashing, JWT, rate limiting, security logging)."""
+"""Security subpackage (hashing, JWT, rate limiting, security logging, audit)."""
 
 from __future__ import annotations
+
+from app.core.security import audit, rate_limiter, security_logger
+from app.core.security.security import (
+    create_access_token,
+    create_refresh_token,
+    create_signed_token,
+    decode_access_token,
+    decode_refresh_token,
+    decode_token,
+    hash_password,
+    hash_password_async,
+    verify_password,
+    verify_password_async,
+)
+
+__all__ = [
+    'audit',
+    'create_access_token',
+    'create_refresh_token',
+    'create_signed_token',
+    'decode_access_token',
+    'decode_refresh_token',
+    'decode_token',
+    'hash_password',
+    'hash_password_async',
+    'rate_limiter',
+    'security_logger',
+    'verify_password',
+    'verify_password_async',
+]

app/core/security/audit.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+GENESIS_HASH: str = '0' * 64
+
+
+def _normalize_timestamp(dt: datetime) -> str:
+    """Normalize datetime to UTC ISO-8601 string across dialects."""
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    else:
+        dt = dt.astimezone(UTC)
+    return dt.isoformat()
+
+
+def compute_audit_hash(
+    *,
+    id: str,
+    actor_user_id: str | None,
+    action: str,
+    resource_type: str,
+    resource_id: str,
+    result: str,
+    details: dict[str, Any] | None,
+    created_at: datetime,
+    previous_hash: str | None,
+) -> str:
+    """Compute SHA-256 hash for an audit log record enforcing hash chaining."""
+    serialized_details = (
+        json.dumps(details, sort_keys=True, separators=(',', ':'), default=str)
+        if details is not None
+        else ''
+    )
+    timestamp_str = created_at.isoformat()
+    timestamp_str = _normalize_timestamp(created_at)
+    prev_hash_str = (
+        previous_hash if previous_hash is not None else GENESIS_HASH
+    )
+
+    payload = (
+        f'{id}|'
+        f'{actor_user_id or ""}|'
+        f'{action}|'
+        f'{resource_type}|'
+        f'{resource_id}|'
+        f'{result}|'
+        f'{serialized_details}|'
+        f'{timestamp_str}|'
+        f'{prev_hash_str}'
+    )
+    return hashlib.sha256(payload.encode('utf-8')).hexdigest()

app/models/audit_log.py

@@ -4,7 +4,16 @@
 from typing import Any
 from uuid import uuid4
 
-from sqlalchemy import JSON, DateTime, ForeignKey, Index, String, func
+from sqlalchemy import (
+    DDL,
+    JSON,
+    DateTime,
+    ForeignKey,
+    Index,
+    String,
+    event,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -50,10 +59,91 @@ class AuditLog(Base):
         nullable=False,
         server_default=func.now(),
     )
+    previous_hash: Mapped[str | None] = mapped_column(
+        String(64),
+        nullable=True,
+    )
+    hash: Mapped[str] = mapped_column(
+        String(64),
+        nullable=False,
+        index=True,
+    )
 
     def __repr__(self) -> str:  # pragma: no cover
         return (
             f'<AuditLog id={self.id!r} action={self.action!r} '
             f'resource={self.resource_type}:{self.resource_id} '
             f'result={self.result!r}>'
         )
+
+
+sqlite_prevent_update = DDL("""
+CREATE TRIGGER IF NOT EXISTS prevent_audit_log_update
+BEFORE UPDATE ON audit_logs
+BEGIN
+    SELECT RAISE(ABORT, 'A tabela audit_logs é append-only. Operações de UPDATE são proibidas.');
+END;
+""")
+
+sqlite_prevent_delete = DDL("""
+CREATE TRIGGER IF NOT EXISTS prevent_audit_log_delete
+BEFORE DELETE ON audit_logs
+BEGIN
+    SELECT RAISE(ABORT, 'A tabela audit_logs é append-only. Operações de DELETE são proibidas.');
+END;
+""")
+
+pg_prevent_mutation_func = DDL("""
+CREATE OR REPLACE FUNCTION block_audit_log_mutation()
+RETURNS TRIGGER AS $$
+BEGIN
+    RAISE EXCEPTION 'A tabela audit_logs é append-only. Operações de UPDATE ou DELETE são proibidas.';
+END;
+$$ LANGUAGE plpgsql;
+""")
+
+pg_prevent_mutation_trigger = DDL("""
+CREATE TRIGGER prevent_audit_log_mutation
+BEFORE UPDATE OR DELETE ON audit_logs
+FOR EACH ROW
+EXECUTE FUNCTION block_audit_log_mutation();
+""")
+
+event.listen(
+    AuditLog.__table__,
+    'after_create',
+    sqlite_prevent_update.execute_if(dialect='sqlite'),
+)
+event.listen(
+    AuditLog.__table__,
+    'after_create',
+    sqlite_prevent_delete.execute_if(dialect='sqlite'),
+)
+event.listen(
+    AuditLog.__table__,
+    'after_create',
+    pg_prevent_mutation_func.execute_if(dialect='postgresql'),
+)
+event.listen(
+    AuditLog.__table__,
+    'after_create',
+    pg_prevent_mutation_trigger.execute_if(dialect='postgresql'),
+)
+
+
+@event.listens_for(AuditLog, 'before_update')
+def _block_audit_log_orm_update(mapper, connection, target):
+    from app.core.exceptions import AuditImmutabilityError
+
+    raise AuditImmutabilityError(
+        'A tabela audit_logs é append-only. Operações de UPDATE são proibidas.'
+    )
+
+
+@event.listens_for(AuditLog, 'before_delete')
+def _block_audit_log_orm_delete(mapper, connection, target):
+    from app.core.exceptions import AuditImmutabilityError
+
+    raise AuditImmutabilityError(
+        'A tabela audit_logs é append-only. Operações de DELETE são proibidas.'
+    )

app/repositories/audit_repository.py

@@ -1,15 +1,34 @@
 from __future__ import annotations
 
+from datetime import UTC, datetime
 from typing import Any
+from uuid import uuid4
 
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exceptions import AuditImmutabilityError
+from app.core.security.audit import compute_audit_hash
 from app.models.audit_log import AuditLog
 from app.repositories.base import BaseRepository
 
 
 class AuditRepository(BaseRepository[AuditLog]):
-    def __init__(self, session):
+    def __init__(self, session: AsyncSession):
         super().__init__(session, AuditLog)
 
+    async def update(self, *args: Any, **kwargs: Any) -> AuditLog:
+        """Prevent updates to audit logs."""
+        raise AuditImmutabilityError(
+            'A tabela audit_logs é append-only. Operações de UPDATE são proibidas.'
+        )
+
+    async def delete(self, *args: Any, **kwargs: Any) -> None:
+        """Prevent deletion of audit logs."""
+        raise AuditImmutabilityError(
+            'A tabela audit_logs é append-only. Operações de DELETE são proibidas.'
+        )
+
     async def add_record(
         self,
         *,
@@ -21,14 +40,49 @@ async def add_record(
         details: dict[str, Any] | None = None,
     ) -> AuditLog:
         """Create an audit record within the caller's transaction."""
+        """Create an audit record with cryptographic hash chain."""
+        # Query the latest hash in the audit chain
+        last_hash_stmt = (
+            select(AuditLog.hash)
+            .order_by(AuditLog.created_at.desc(), AuditLog.id.desc())
+            .limit(1)
+        )
+        previous_hash = await self.session.scalar(last_hash_stmt)
+
+        record_id = str(uuid4())
+        created_at = datetime.now(UTC)
+        record_hash = compute_audit_hash(
+            id=record_id,
+            actor_user_id=actor_user_id,
+            action=action,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            result=result,
+            details=details,
+            created_at=created_at,
+            previous_hash=previous_hash,
+        )
+
         record = AuditLog(
+            id=record_id,
             actor_user_id=actor_user_id,
             action=action,
             resource_type=resource_type,
             resource_id=resource_id,
             result=result,
             details=details,
+            created_at=created_at,
+            previous_hash=previous_hash,
+            hash=record_hash,
         )
         self.session.add(record)
         await self.session.flush()
         return record
+
+    async def list_all_chronological(self) -> list[AuditLog]:
+        """Fetch all audit logs in chronological order for integrity verification."""
+        stmt = select(AuditLog).order_by(
+            AuditLog.created_at.asc(), AuditLog.id.asc()
+        )
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())

app/services/audit_service.py

@@ -4,11 +4,15 @@
 
 from typing import Any
 
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.security.audit import compute_audit_hash
+from app.models.audit_log import AuditLog
 from app.repositories.audit_repository import AuditRepository
 
 
 async def record_admin_action(
-    db,
+    db: AsyncSession,
     *,
     actor_user_id: str | None,
     action: str,
@@ -31,3 +35,48 @@ async def record_admin_action(
         result=result,
         details=details,
     )
+
+
+async def verify_audit_trail_integrity(
+    db: AsyncSession,
+    *,
+    records: list[AuditLog] | None = None,
+) -> tuple[bool, list[str]]:
+    """Verify cryptographic integrity and continuity of the audit trail.
+
+    Returns:
+        (True, []) if the hash chain is fully intact.
+        (False, errors) if any record was tampered with, deleted or out-of-order.
+    """
+    if records is None:
+        records = await AuditRepository(db).list_all_chronological()
+    errors: list[str] = []
+    expected_previous_hash: str | None = None
+
+    for idx, record in enumerate(records):
+        if record.previous_hash != expected_previous_hash:
+            errors.append(
+                f'Record {record.id} (index {idx}): broken previous_hash chain. '
+                f'Expected {expected_previous_hash!r}, found {record.previous_hash!r}.'
+            )
+
+        computed_hash = compute_audit_hash(
+            id=record.id,
+            actor_user_id=record.actor_user_id,
+            action=record.action,
+            resource_type=record.resource_type,
+            resource_id=record.resource_id,
+            result=record.result,
+            details=record.details,
+            created_at=record.created_at,
+            previous_hash=record.previous_hash,
+        )
+        if record.hash != computed_hash:
+            errors.append(
+                f'Record {record.id} (index {idx}): hash mismatch. '
+                f'Stored {record.hash!r}, computed {computed_hash!r}.'
+            )
+
+        expected_previous_hash = record.hash
+
+    return (len(errors) == 0, errors)