feat(audit): implement audit logging functionality with middleware

api/app/audit/__init__.py

@@ -0,0 +1 @@
+# Audit logging module - exports events to external SIEMs

api/app/audit/core/__init__.py

@@ -0,0 +1 @@
+# Audit core - config, event schema, sender

api/app/audit/core/audit_config.py

@@ -0,0 +1,26 @@
+"""Audit logging configuration loaded from environment variables."""
+
+import os
+
+
+def is_audit_enabled() -> bool:
+    """Return True if audit logging is enabled via AUDIT_LOG_ENABLED."""
+    return os.getenv("AUDIT_LOG_ENABLED", "false").lower() == "true"
+
+
+def get_siem_url() -> str | None:
+    """Return the SIEM endpoint URL or None if not configured."""
+    return os.getenv("AUDIT_SIEM_URL") or None
+
+
+def get_siem_token() -> str | None:
+    """Return the SIEM Bearer token or None if not configured."""
+    return os.getenv("AUDIT_SIEM_TOKEN") or None
+
+
+def get_siem_timeout() -> float:
+    """Return the SIEM HTTP timeout in seconds (default: 5)."""
+    try:
+        return float(os.getenv("AUDIT_SIEM_TIMEOUT", "5"))
+    except ValueError:
+        return 5.0

api/app/audit/core/audit_event.py

@@ -0,0 +1,82 @@
+"""Audit event schema and HTTP method to action mapping."""
+
+from datetime import datetime, timezone
+
+
+def map_method_to_action(method: str, path: str) -> str:
+    """
+    Map HTTP method and path to audit action.
+    GET=READ, POST=CREATE/EXEC, PUT/PATCH=UPDATE, DELETE=DELETE.
+    For /exec endpoints, use EXEC.
+    """
+    method_upper = method.upper()
+    path_lower = path.lower()
+
+    if "/exec" in path_lower:
+        return "EXEC"
+
+    mapping = {
+        "GET": "READ",
+        "HEAD": "READ",
+        "OPTIONS": "READ",
+        "POST": "CREATE",
+        "PUT": "UPDATE",
+        "PATCH": "UPDATE",
+        "DELETE": "DELETE",
+    }
+    return mapping.get(method_upper, "UNKNOWN")
+
+
+def normalize_resource(path: str) -> str:
+    """
+    Normalize request path to resource identifier.
+    Strips leading slash and collapses repeated slashes.
+    Keeps path structure (e.g., organizations/123/instances/456).
+    """
+    if not path:
+        return "unknown"
+    # Remove leading slash, strip trailing slash
+    normalized = path.strip("/")
+    # Collapse multiple slashes
+    while "//" in normalized:
+        normalized = normalized.replace("//", "/")
+    return normalized or "unknown"
+
+
+def _truncate_for_audit(value: str, max_length: int = 4096) -> str:
+    """Truncate string for audit logging to avoid huge payloads."""
+    if not value or len(value) <= max_length:
+        return value
+    return value[:max_length] + f"... [truncated, total {len(value)} chars]"
+
+
+def build_audit_event(
+    actor: str,
+    method: str,
+    path: str,
+    status_code: int,
+    source_ip: str,
+    exec_payload: dict | None = None,
+) -> dict:
+    """Build an audit event dict from request/response data."""
+    event = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "actor": actor,
+        "action": map_method_to_action(method, path),
+        "resource": normalize_resource(path),
+        "status": f"{'success' if 200 <= status_code < 400 else 'failure'}:{status_code}",
+        "source_ip": source_ip,
+    }
+    if exec_payload:
+        event["exec_request"] = exec_payload.get("request")
+        response_data = exec_payload.get("response")
+        if response_data and isinstance(response_data, dict):
+            truncated = response_data.copy()
+            if "stdout" in truncated and isinstance(truncated["stdout"], str):
+                truncated["stdout"] = _truncate_for_audit(truncated["stdout"])
+            if "stderr" in truncated and isinstance(truncated["stderr"], str):
+                truncated["stderr"] = _truncate_for_audit(truncated["stderr"])
+            event["exec_response"] = truncated
+        else:
+            event["exec_response"] = response_data
+    return event

api/app/audit/core/audit_sender.py

@@ -0,0 +1,46 @@
+"""Asynchronous audit event sender with fail-open behavior."""
+
+import logging
+from typing import Any
+
+import httpx
+from httpx import ConnectError, TimeoutException
+
+from app.audit.core.audit_config import get_siem_token, get_siem_timeout
+
+logger = logging.getLogger(__name__)
+
+
+async def send_audit_event(siem_url: str, payload: dict[str, Any]) -> None:
+    """
+    Send audit event to SIEM endpoint asynchronously.
+    Fail-open: logs errors but never propagates exceptions.
+    """
+    timeout = get_siem_timeout()
+    headers: dict[str, str] = {"Content-Type": "application/json"}
+    token = get_siem_token()
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                siem_url,
+                json=payload,
+                headers=headers,
+                timeout=timeout,
+            )
+            if response.status_code >= 400:
+                logger.warning(
+                    "Audit SIEM returned %s for event: %s",
+                    response.status_code,
+                    payload.get("resource", "unknown"),
+                )
+    except TimeoutException as e:
+        logger.warning("Audit SIEM request timed out: %s", e)
+    except ConnectError as e:
+        logger.warning("Audit SIEM request failed: %s", e)
+    except httpx.RequestError as e:
+        logger.warning("Audit SIEM request failed: %s", e)
+    except Exception as e:
+        logger.warning("Audit send unexpected error: %s", e)

api/app/main.py

@@ -4,7 +4,9 @@
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.openapi.docs import get_redoc_html
 
+from app.audit.core.audit_config import get_siem_url, is_audit_enabled
 from app.shared.database.database import Base, engine
+from app.shared.middleware.audit_middleware import AuditMiddleware
 
 # Also import Base from old database to ensure compatibility
 from app.database import Base as OldBase
@@ -95,6 +97,10 @@
 ).split(",")
 CORS_ALLOW_HEADERS = [header.strip() for header in CORS_ALLOW_HEADERS if header.strip()]
 
+# Audit logging - only add when enabled and SIEM URL is configured
+if is_audit_enabled() and get_siem_url():
+    app.add_middleware(AuditMiddleware)
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=CORS_ORIGINS,

api/app/shared/dependencies/auth.py

@@ -1,4 +1,4 @@
-from fastapi import Depends, HTTPException, status, Header
+from fastapi import Depends, HTTPException, Request, status, Header
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from sqlalchemy.orm import Session
 from typing import Optional, Union
@@ -15,6 +15,7 @@
 
 
 async def get_current_user_or_token(
+    request: Request,
     x_tron_token: Optional[str] = Header(None, alias="x-tron-token"),
     credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
     db: Session = Depends(get_db),
@@ -46,6 +47,7 @@ async def get_current_user_or_token(
                 status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expirado"
             )
 
+        request.state.audit_actor = f"token:{token.name}"
         return token
 
     # Fallback para JWT
@@ -82,6 +84,7 @@ async def get_current_user_or_token(
             status_code=status.HTTP_401_UNAUTHORIZED, detail="User inactive"
         )
 
+    request.state.audit_actor = user.email
     return user
 
 

api/app/shared/middleware/__init__.py

@@ -0,0 +1 @@
+# Shared middleware

api/app/shared/middleware/audit_middleware.py

@@ -0,0 +1,70 @@
+"""Audit logging middleware - captures request/response and sends to SIEM."""
+
+import asyncio
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+
+from app.audit.core.audit_config import get_siem_url
+from app.audit.core.audit_event import build_audit_event
+from app.audit.core.audit_sender import send_audit_event
+
+SKIP_PATHS = {"/health", "/docs", "/redoc", "/openapi.json"}
+
+
+def _get_client_ip(request: Request) -> str:
+    """Extract client IP, supporting X-Forwarded-For and X-Real-IP proxies."""
+    forwarded = request.headers.get("X-Forwarded-For")
+    if forwarded:
+        return forwarded.split(",")[0].strip()
+    real_ip = request.headers.get("X-Real-IP")
+    if real_ip:
+        return real_ip.strip()
+    if request.client:
+        return request.client.host
+    return "unknown"
+
+
+class AuditMiddleware(BaseHTTPMiddleware):
+    """Middleware that logs audit events to configured SIEM endpoint."""
+
+    async def dispatch(self, request: Request, call_next):
+        response = await call_next(request)
+
+        path = request.url.path
+        if path in SKIP_PATHS:
+            return response
+
+        siem_url = get_siem_url()
+        if not siem_url:
+            return response
+
+        actor = getattr(request.state, "audit_actor", None) or "anonymous"
+        source_ip = _get_client_ip(request)
+        exec_payload = getattr(request.state, "audit_exec_payload", None)
+
+        event = build_audit_event(
+            actor=actor,
+            method=request.method,
+            path=path,
+            status_code=response.status_code,
+            source_ip=source_ip,
+            exec_payload=exec_payload,
+        )
+
+        task = asyncio.create_task(send_audit_event(siem_url, event))
+
+        def _log_task_error(t):
+            if t.cancelled():
+                return
+            exc = t.exception()
+            if exc:
+                import logging
+
+                logging.getLogger(__name__).warning(
+                    "Audit task failed: %s", exc, exc_info=True
+                )
+
+        task.add_done_callback(_log_task_error)
+
+        return response

api/app/webapps/api/webapp_handlers.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.orm import Session
 from uuid import UUID
 
@@ -505,6 +505,7 @@ def get_webapp_pod_describe(
 
 @router.post("/{uuid}/pods/{pod_name}/exec", response_model=PodCommandResponse)
 def exec_webapp_pod_command(
+    http_request: Request,
     uuid: UUID,
     pod_name: str,
     request: PodCommandRequest,
@@ -513,6 +514,12 @@ def exec_webapp_pod_command(
     current_user: User = Depends(get_current_user),
 ):
     """Execute a command in a pod."""
+    http_request.state.audit_exec_payload = {
+        "request": {
+            "command": request.command,
+            "container_name": request.container_name,
+        }
+    }
     repository = WebappRepository(database_session)
     webapp = repository.find_by_uuid(uuid, load_relations=True)
 
@@ -544,8 +551,14 @@ def exec_webapp_pod_command(
         result = exec_webapp_pod_command_from_cluster(
             cluster, namespace, pod_name, request.command, request.container_name
         )
+        http_request.state.audit_exec_payload["response"] = {
+            "stdout": result.get("stdout", ""),
+            "stderr": result.get("stderr", ""),
+            "return_code": result.get("return_code", -1),
+        }
         return result
     except Exception as e:
+        http_request.state.audit_exec_payload["response"] = {"error": str(e)}
         raise HTTPException(
             status_code=500,
             detail=f"Failed to execute command in pod {pod_name}: {str(e)}",

api/app/workers/api/worker_handlers.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.orm import Session
 from uuid import UUID
 
@@ -461,6 +461,7 @@ def get_worker_pod_describe(
 
 @router.post("/{uuid}/pods/{pod_name}/exec", response_model=PodCommandResponse)
 def exec_worker_pod_command(
+    http_request: Request,
     uuid: UUID,
     pod_name: str,
     request: PodCommandRequest,
@@ -469,6 +470,12 @@ def exec_worker_pod_command(
     current_user: User = Depends(get_current_user),
 ):
     """Execute a command in a pod."""
+    http_request.state.audit_exec_payload = {
+        "request": {
+            "command": request.command,
+            "container_name": request.container_name,
+        }
+    }
     repository = WorkerRepository(database_session)
     worker = repository.find_by_uuid(uuid, load_relations=True)
 
@@ -503,8 +510,14 @@ def exec_worker_pod_command(
         result = exec_worker_pod_command_from_cluster(
             cluster, namespace, pod_name, request.command, request.container_name
         )
+        http_request.state.audit_exec_payload["response"] = {
+            "stdout": result.get("stdout", ""),
+            "stderr": result.get("stderr", ""),
+            "return_code": result.get("return_code", -1),
+        }
         return result
     except Exception as e:
+        http_request.state.audit_exec_payload["response"] = {"error": str(e)}
         raise HTTPException(
             status_code=500,
             detail=f"Failed to execute command in pod {pod_name}: {str(e)}",

api/pyrightconfig.json

@@ -0,0 +1,5 @@
+{
+  "venvPath": ".",
+  "venv": ".venv",
+  "extraPaths": ["."]
+}