audit log A09

Author: makara4code

backend/app/alembic/versions/ffb35978c176_change_audit_log_details_to_jsonb.py

@@ -0,0 +1,45 @@
+"""change_audit_log_details_to_jsonb
+
+Revision ID: ffb35978c176
+Revises: 405b1d47dec0
+Create Date: 2025-12-14 23:30:20.938714
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = 'ffb35978c176'
+down_revision = '405b1d47dec0'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    # First, convert existing string data to JSON format
+    op.execute("""
+        UPDATE audit_log
+        SET details = jsonb_build_object('message', details)
+        WHERE details IS NOT NULL
+          AND details != ''
+          AND NOT (details::text LIKE '{%' OR details::text LIKE '[%')
+    """)
+    # Change column type
+    op.alter_column('audit_log', 'details',
+               existing_type=sa.VARCHAR(length=5000),
+               type_=postgresql.JSONB(astext_type=sa.Text()),
+               existing_nullable=True,
+               postgresql_using='details::jsonb')
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('audit_log', 'details',
+               existing_type=postgresql.JSONB(astext_type=sa.Text()),
+               type_=sa.VARCHAR(length=5000),
+               existing_nullable=True)
+    # ### end Alembic commands ###

backend/app/modules/owasp_demo/models.py

@@ -5,6 +5,8 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, Optional
 
+from sqlalchemy import Column
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlmodel import Field, Relationship, SQLModel
 
 if TYPE_CHECKING:
@@ -51,6 +53,25 @@ class AuditLog(SQLModel, table=True):
     id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
     action: str = Field(max_length=100)
     user_id: uuid.UUID | None = Field(default=None)
-    details: str | None = Field(default=None, max_length=5000)
+    details: dict | None = Field(default=None, sa_column=Column(JSONB))
     ip_address: str | None = Field(default=None, max_length=45)
     timestamp: datetime = Field(default_factory=datetime.utcnow)
+
+
+class AuditLogPublic(SQLModel):
+    """Public schema for audit log responses."""
+
+    id: str
+    action: str
+    user_id: str | None
+    user_email: str | None
+    details: dict | None
+    ip_address: str | None
+    timestamp: str | None
+
+
+class AuditLogsPublic(SQLModel):
+    """Paginated list of audit logs response."""
+
+    data: list[AuditLogPublic]
+    count: int

backend/app/modules/owasp_demo/routers/a07_authentication.py

@@ -35,16 +35,42 @@ def timing_attack_vulnerable(
     """
     from app.modules.users.repository import UserRepository
 
+    start_time = time.time()
     repo = UserRepository(session)
     user = repo.get_by_email(email=email)
 
     if not user:
-        return {"success": False, "message": "Invalid credentials"}
-
+        # FAST response - no password check needed
+        # (Adding small artificial delay to make the difference more visible in demo)
+        time.sleep(0.05)  # 50ms base delay
+        elapsed = time.time() - start_time
+        return {
+            "success": False,
+            "message": "Invalid credentials",
+            "response_time_ms": round(elapsed * 1000, 2),
+            "timing_leak": "FAST (~50ms) - Email does NOT exist!",
+            "security_issue": "Attacker now knows this email is not registered",
+        }
+
+    # SLOW response - password verification takes time (bcrypt is intentionally slow)
+    # Adding artificial delay to make timing difference obvious for demo
+    time.sleep(0.5)  # 500ms delay to simulate bcrypt
     if not verify_password(password, user.hashed_password or ""):
-        return {"success": False, "message": "Invalid credentials"}
-
-    return {"success": True, "message": "Login successful"}
+        elapsed = time.time() - start_time
+        return {
+            "success": False,
+            "message": "Invalid credentials",
+            "response_time_ms": round(elapsed * 1000, 2),
+            "timing_leak": "SLOW (~500ms) - Email EXISTS! Password check was performed",
+            "security_issue": "Attacker now knows this email is registered",
+        }
+
+    elapsed = time.time() - start_time
+    return {
+        "success": True,
+        "message": "Login successful",
+        "response_time_ms": round(elapsed * 1000, 2),
+    }
 
 
 @router.post("/secure/login-timing")
@@ -60,18 +86,35 @@ def timing_attack_secure(
     """
     from app.modules.users.repository import UserRepository
 
+    start_time = time.time()
     repo = UserRepository(session)
     user = repo.get_by_email(email=email)
 
+    # Always perform password check - use dummy hash if user doesn't exist
     dummy_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.x6S5h0r9h6F5Hy"
     stored_hash = user.hashed_password if user else dummy_hash
 
+    # Always perform password verification (constant time)
+    # Adding same artificial delay as vulnerable endpoint to demonstrate
+    time.sleep(0.5)  # 500ms - same delay whether email exists or not
     is_valid = verify_password(password, stored_hash or dummy_hash)
+    elapsed = time.time() - start_time
 
     if not user or not is_valid:
-        return {"success": False, "message": "Invalid credentials"}
+        return {
+            "success": False,
+            "message": "Invalid credentials",
+            "response_time_ms": round(elapsed * 1000, 2),
+            "timing_protection": "CONSTANT (~500ms) - Same time for all requests",
+            "security_benefit": "Attacker cannot determine if email exists",
+        }
 
-    return {"success": True, "message": "Login successful"}
+    return {
+        "success": True,
+        "message": "Login successful",
+        "response_time_ms": round(elapsed * 1000, 2),
+        "timing_protection": "CONSTANT (~500ms) - Same time for all requests",
+    }
 
 
 @router.get("/vulnerable/session")

backend/app/modules/owasp_demo/routers/a09_logging.py

@@ -4,14 +4,31 @@
 Demonstrates missing audit logging vulnerabilities.
 """
 
+import logging
 from datetime import datetime
+from typing import Any
 
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, Request
+from sqlmodel import select
 
-from app.modules.owasp_demo.models import AuditLog
-from app.modules.rbac.service import UserRoleService
+from app.modules.owasp_demo.models import AuditLog, AuditLogPublic, AuditLogsPublic
 from app.modules.shared import CurrentUser, SessionDep
 
+# Configure security audit logger
+# In production, this would output JSON to a SIEM system (Splunk, ELK, Datadog)
+audit_logger = logging.getLogger("security.audit")
+audit_logger.setLevel(logging.INFO)
+
+# Add console handler if not already configured
+if not audit_logger.handlers:
+    handler = logging.StreamHandler()
+    handler.setFormatter(
+        logging.Formatter(
+            "%(asctime)s | %(levelname)s | %(name)s | %(message)s | %(extra_data)s"
+        )
+    )
+    audit_logger.addHandler(handler)
+
 router = APIRouter(
     prefix="/a09",
     tags=["A09 - Logging Failures"],
@@ -35,40 +52,156 @@ def no_audit_logging(
     return {"status": "Action completed", "action": action}
 
 
+class LogExtraAdapter(logging.LoggerAdapter):
+    """Adapter to format extra data for logging."""
+
+    def process(
+        self, msg: str, kwargs: dict[str, Any]
+    ) -> tuple[str, dict[str, Any]]:
+        extra = kwargs.get("extra", {})
+        # Format extra data as key=value pairs for structured logging
+        extra_str = " ".join(f"{k}={v}" for k, v in extra.items())
+        kwargs["extra"] = {"extra_data": extra_str}
+        return msg, kwargs
+
+
+# Create adapter for structured logging
+security_logger = LogExtraAdapter(audit_logger, {})
+
+
 @router.post("/secure/action")
 def with_audit_logging(
     session: SessionDep,
     request: Request,
     current_user: CurrentUser,
     action: str,
     target_id: str,
-) -> dict[str, str]:
+) -> dict[str, Any]:
     """
-    SECURE: Comprehensive audit logging with RBAC
+    SECURE: Comprehensive audit logging
 
     Fix: Log all security-relevant events with context.
-    Requires 'audit:write' permission to perform auditable actions.
+    - Console/SIEM logging for real-time monitoring
+    - Database audit trail for compliance and queries
     """
-    if not current_user.is_superuser:
-        user_role_service = UserRoleService(session)
-        if not user_role_service.user_has_permission(current_user.id, "audit:write"):
-            raise HTTPException(
-                status_code=403, detail="Permission 'audit:write' required"
-            )
-
     client_ip = request.client.host if request.client else "unknown"
+    user_agent = request.headers.get("User-Agent", "unknown")
+    timestamp = datetime.utcnow()
+
+    # Structured log context for all log entries
+    log_context = {
+        "event_type": "security_action",
+        "action": action,
+        "target_id": target_id,
+        "user_id": str(current_user.id),
+        "user_email": current_user.email,
+        "ip_address": client_ip,
+        "user_agent": user_agent[:50] if user_agent else "unknown",
+        "timestamp": timestamp.isoformat(),
+    }
 
+    # Log action attempt
+    security_logger.info("Action attempted", extra=log_context)
+
+    # Store in database for audit trail (queryable, compliance)
+    # Using JSON for rich, structured details that can be queried
     log_entry = AuditLog(
         action=action,
         user_id=current_user.id,
-        details=f"Target: {target_id}",
+        details={
+            "target_id": target_id,
+            "user_agent": user_agent[:100] if user_agent else "unknown",
+            "request_path": str(request.url.path),
+            "request_method": request.method,
+        },
         ip_address=client_ip,
-        timestamp=datetime.utcnow(),
+        timestamp=timestamp,
     )
     session.add(log_entry)
     session.commit()
 
+    # Log successful completion
+    security_logger.info(
+        "Action completed",
+        extra={**log_context, "status": "success", "log_id": str(log_entry.id)},
+    )
+
     return {
         "status": "Action completed and logged",
         "log_id": str(log_entry.id),
+        "audit_log": log_context,
+        "logging_destinations": {
+            "console": "Real-time monitoring (check server terminal)",
+            "database": f"Audit trail stored (ID: {log_entry.id})",
+        },
     }
+
+
+@router.delete("/audit-logs")
+def clear_audit_logs(
+    session: SessionDep,
+    current_user: CurrentUser,  # noqa: ARG001
+) -> dict[str, str]:
+    """
+    Clear all audit logs from the database.
+
+    This is for demo purposes to reset the audit trail.
+    In production, audit logs should be immutable and archived, not deleted.
+    """
+    statement = select(AuditLog)
+    logs = session.exec(statement).all()
+    for log in logs:
+        session.delete(log)
+    session.commit()
+    return {"status": "Audit logs cleared", "deleted_count": str(len(logs))}
+
+
+@router.get("/audit-logs")
+def get_audit_logs(
+    session: SessionDep,
+    current_user: CurrentUser,  # noqa: ARG001
+    skip: int = 0,
+    limit: int = 10,
+) -> AuditLogsPublic:
+    """
+    Fetch recent audit logs from the database with pagination.
+
+    Returns the most recent audit log entries for display.
+    Requires authentication.
+    """
+    from app.modules.users.models import User
+
+    # Get total count
+    count_statement = select(AuditLog)
+    total_count = len(session.exec(count_statement).all())
+
+    # Get paginated logs
+    statement = (
+        select(AuditLog)
+        .order_by(AuditLog.timestamp.desc())
+        .offset(skip)
+        .limit(limit)
+    )
+    logs = session.exec(statement).all()
+
+    # Collect unique user IDs and fetch users in batch
+    user_ids = {log.user_id for log in logs if log.user_id}
+    users_map: dict[str, str] = {}
+    if user_ids:
+        users = session.exec(select(User).where(User.id.in_(user_ids))).all()  # type: ignore[attr-defined]
+        users_map = {str(user.id): user.email for user in users}
+
+    data = [
+        AuditLogPublic(
+            id=str(log.id),
+            action=log.action,
+            user_id=str(log.user_id) if log.user_id else None,
+            user_email=users_map.get(str(log.user_id)) if log.user_id else None,
+            details=log.details,
+            ip_address=log.ip_address,
+            timestamp=log.timestamp.isoformat() if log.timestamp else None,
+        )
+        for log in logs
+    ]
+
+    return AuditLogsPublic(data=data, count=total_count)