FastAPI Integration
Integrate ImmutableLog into any FastAPI application with an async middleware based on Starlette. Automatically captures all HTTP requests — errors, successes, and unhandled exceptions — without modifying your routes.
Installation
The middleware only depends on `requests` (for synchronous sending in a separate thread) and Starlette, already included in FastAPI. Copy the `middleware.py` file into your project.
pip install requests fastapiMiddleware Code
Paste the code below into `middleware.py` in your project. The middleware inherits from Starlette's `BaseHTTPMiddleware` and asynchronously intercepts the complete lifecycle of each request.
# middleware.py
import hashlib
import json
import logging
import time
import uuid
from datetime import datetime, timezone
import requests
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.types import ASGIApp
logger = logging.getLogger(__name__)
MAX_PAYLOAD_BYTES = 12_000
# Chaves cujos valores sao mascarados antes do envio (PII/segredos).
SENSITIVE_KEYS = {
"password", "pwd", "passwd", "pass", "senha",
"token", "access_token", "refresh_token", "id_token", "api_key", "apikey",
"secret", "client_secret", "authorization", "auth",
"credit_card", "card_number", "card", "cvv", "cvc",
"pin", "ssn", "cpf", "rg", "otp", "code",
}
REDACTED = "***REDACTED***"
def redact(obj):
if isinstance(obj, dict):
return {k: (REDACTED if isinstance(k, str) and k.lower() in SENSITIVE_KEYS else redact(v)) for k, v in obj.items()}
if isinstance(obj, list):
return [redact(v) for v in obj]
return obj
def sanitize_trail(value):
"""Normaliza o immutable_trail (trim, nao-vazio, max 256, sem ':').
O servidor responde 400 invalid_immutable_trail se a regra for violada."""
if not isinstance(value, str):
return None
v = value.strip()
if not v:
return None
if ":" in v:
v = v.replace(":", "-")
if len(v) > 256:
v = v[:256]
return v
def clamp_payload(payload: dict) -> str:
s = json.dumps(payload, ensure_ascii=False)
if len(s.encode("utf-8")) <= MAX_PAYLOAD_BYTES:
return s
payload = dict(payload)
payload.pop("error", None)
payload.pop("request_body", None)
s2 = json.dumps(payload, ensure_ascii=False)
if len(s2.encode("utf-8")) <= MAX_PAYLOAD_BYTES:
return s2
return json.dumps({
"id": payload.get("id"),
"kind": payload.get("kind"),
"message": (payload.get("message") or "event")[:500],
"timestamp": payload.get("timestamp"),
}, ensure_ascii=False)
def hash_body(body: bytes) -> str:
return hashlib.sha256(body).hexdigest()
def severity_from(status_code: int | None, exc: Exception | None) -> str:
if exc is not None:
return "error"
if status_code is None:
return "info"
if status_code >= 400:
return "error"
if status_code >= 300:
return "info"
if status_code >= 200:
return "success"
return "info"
def get_severity_level(kind: str, status_code: int | None) -> str:
if kind == "error":
return "high"
elif kind == "warning":
return "medium"
return "low"
class ImmutableLogAuditMiddleware(BaseHTTPMiddleware):
def __init__(
self,
app: ASGIApp,
api_key: str,
api_url: str,
service_name: str = "fastapi-service",
env: str = "production",
skip_paths: list[str] | None = None,
):
super().__init__(app)
self.api_key = api_key
self.api_url = api_url
self.service_name = service_name
self.env = env
self.skip_paths = set(skip_paths or ["/health", "/healthz"])
async def dispatch(self, request: Request, call_next):
if request.url.path in self.skip_paths:
return await call_next(request)
started_at = time.time()
request_id = request.headers.get("x-request-id") or str(uuid.uuid4())
body = await request.body()
exc_captured = None
response = None
try:
response = await call_next(request)
except Exception as exc:
exc_captured = exc
self._emit(request, body, None, exc, started_at, request_id)
raise
self._emit(request, body, response, None, started_at, request_id)
response.headers["x-request-id"] = request_id
return response
def _emit(self, request: Request, body: bytes, response, exc, started_at: float, request_id: str):
try:
latency_ms = int((time.time() - started_at) * 1000)
status_code = response.status_code if response else (500 if exc else None)
kind = severity_from(status_code, exc)
event_name = getattr(request.state, "imtbl_event_name", None) or f"http.{request.method}.{request.url.path}"
user_context = {
"ip": request.client.host if request.client else "unknown",
"user_agent": request.headers.get("user-agent", "unknown"),
}
payload = {
"id": str(uuid.uuid4()),
"kind": kind,
"message": self._generate_message(request, status_code, exc, kind),
"timestamp": datetime.now(timezone.utc).isoformat(),
"context": user_context,
"request": {
"request_id": request_id,
"method": request.method,
"path": str(request.url.path),
# query params passam por redact() — nunca enviar segredos em claro.
"query_params": redact(dict(request.query_params)) or None,
},
"metrics": {"latency_ms": latency_ms, "status_code": status_code},
"severity": get_severity_level(kind, status_code),
}
if kind == "error":
error_details = {
"status_code": status_code,
"retryable": status_code in [408, 429, 500, 502, 503, 504] if status_code else False,
}
if exc:
error_details["exception"] = type(exc).__name__
error_details["exception_message"] = str(exc)
if body:
error_details["request_body_hash"] = hash_body(body)
if len(body) < 2000:
# Tenta redatar campos sensiveis se o body for JSON.
text = body.decode("utf-8", errors="replace")
try:
error_details["request_body"] = json.dumps(redact(json.loads(text)), ensure_ascii=False)
except (ValueError, TypeError):
error_details["request_body"] = text
payload["error"] = error_details
elif kind == "success":
payload["success"] = {"status_code": status_code, "result": "ok" if status_code == 200 else "processed"}
elif kind == "info":
payload["info"] = {"status_code": status_code, "action": request.method.lower()}
# Trilha imutavel: request.state.imtbl_trail -> header X-Imtbl-Trail.
trail = sanitize_trail(getattr(request.state, "imtbl_trail", None)) or sanitize_trail(request.headers.get("x-imtbl-trail"))
meta = {
"type": kind,
"event_name": event_name,
"service": self.service_name,
"request_id": request_id,
"env": self.env,
}
if trail:
meta["immutable_trail"] = trail
event = {"payload": clamp_payload(payload), "meta": meta}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
# Idempotency-Key e OBRIGATORIO (sem ele a API responde 400).
"Idempotency-Key": f"{event_name}-{request_id}",
"Request-Id": request_id,
"X-Client-TZ": "UTC",
"X-Client-Offset-Minutes": "0",
}
if trail:
headers["X-Imtbl-Trail"] = trail
requests.post(f"{self.api_url}/v1/events", json=event, headers=headers, timeout=5)
except Exception as e:
logger.warning("ImmutableLogAuditMiddleware _emit failed: %s", e)
def _generate_message(self, request: Request, status_code, exc, kind: str) -> str:
method = request.method
path = str(request.url.path)
if exc:
return f"{method} {path} failed with exception: {type(exc).__name__}"
if kind == "error":
return f"{method} {path} returned error {status_code}"
elif kind == "success":
return f"{method} {path} completed successfully"
return f"{method} {path} processed"Limit: Payload limit: 12KB per event. Large fields are automatically removed if the limit is exceeded.
Registration and Configuration
Unlike Django, FastAPI does not use a centralized `settings.py` file. The middleware is configured directly via parameters at registration time with `app.add_middleware()`. Use environment variables to load credentials securely.
import os
from fastapi import FastAPI
from middleware import ImmutableLogAuditMiddleware
app = FastAPI()
app.add_middleware(
ImmutableLogAuditMiddleware,
api_key=os.environ.get("IMTBL_API_KEY", ""),
api_url=os.environ.get("IMTBL_URL", "https://api.immutablelog.com"),
service_name=os.environ.get("IMTBL_SERVICE_NAME", "fastapi-service"),
env=os.environ.get("IMTBL_ENV", "production"),
skip_paths=["/health", "/healthz", "/metrics"],
)Never pass the token directly in code. Use environment variables (.env) and load them via os.environ.get() or libraries like python-dotenv.
How it works
The `dispatch` method intercepts each request. The body is read before passing to the handler to allow inclusion in error events. The response passes through the handler and the event is emitted to ImmutableLog. Unhandled exceptions are captured, the event is emitted, and the exception is re-raised for FastAPI's default handling.
await request.body()
Body read before handler for capture in error events
await call_next(request)
Handler executed; latency calculated; event emitted on exit
except Exception → re-raise
Exceptions captured, error event emitted, and exception re-raised
The x-request-id is injected into the response for end-to-end traceability across systems.
Custom event names
Use `request.state.imtbl_event_name` inside any route to override the automatically generated event name. This enables granular traceability by business event in the dashboard.
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post("/checkout")
async def process_checkout(request: Request):
# Sobrescreve o nome do evento para esta rota
# Override the event name for this route
request.state.imtbl_event_name = "payment.checkout.initiated"
# ... logica de negocio / business logic ...
return JSONResponse({"status": "ok"})If not set, the default is http.METHOD.path (e.g., http.POST./checkout).
Immutable trail (immutable_trail)
A trail groups related events into a single auditable timeline. Set `request.state.imtbl_trail` in a route or propagate it across services with the `X-Imtbl-Trail` header. The value goes through `sanitize_trail` before sending.
from fastapi import FastAPI, Request
@app.post("/flows/{flow_id}/run")
async def run_flow(flow_id: str, request: Request):
request.state.imtbl_trail = f"flow-{flow_id}"
# ... logica de negocio / business logic ...
return {"status": "ok"}The trail cannot be empty, exceed 256 characters, or contain `:` — violations return `400 invalid_immutable_trail`. `sanitize_trail` fixes the value before sending.
Path exclusion
Pass a list of paths in `skip_paths` when registering the middleware to ignore specific routes (e.g., health checks). By default, `/health` and `/healthz` are automatically ignored.
app.add_middleware(
ImmutableLogAuditMiddleware,
api_key=os.environ["IMTBL_API_KEY"],
api_url="https://api.immutablelog.com",
# Rotas ignoradas / Ignored routes
skip_paths=["/health", "/healthz", "/metrics", "/readyz"],
)Note on async sending
Sending to ImmutableLog uses `requests` (synchronous) in an async context. For high-load production, we recommend replacing it with `httpx.AsyncClient` to avoid blocking the event loop. The example below shows both approaches.
# Alternativa async com httpx (recomendada para alta carga)
# Async alternative with httpx (recommended for high load)
import httpx
import asyncio
# Na classe ImmutableLogAuditMiddleware, substitua requests.post por:
# In ImmutableLogAuditMiddleware class, replace requests.post with:
async def _emit_async(self, ...):
async with httpx.AsyncClient() as client:
await client.post(
f"{self.api_url}/v1/events",
json=event,
headers=headers,
timeout=5,
)
# Ou use asyncio.get_event_loop().run_in_executor() para manter requests:
# Or use asyncio.get_event_loop().run_in_executor() to keep requests:
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None,
lambda: requests.post(url, json=event, headers=headers, timeout=5)
)To install httpx: pip install httpx
meta fields
Beyond the payload, the meta object carries the fields the SIEM uses to normalize (ECS) and enrich the event. All meta values are strings.
| Field in meta | What it does in the SIEM |
|---|---|
event_name | The event label (e.g. http.GET.auth-me-user, Payment error) — the identifier shown in the listing. Becomes event.action and derives event.category (e.g. a name with login → authentication). |
client_ip | Becomes source.ip with maximum precedence and provenance client_asserted. See the section below. |
immutable_trail | Becomes imtbl.immutable_trail — groups related events for investigation (e.g. one order, one user). |
type | Event severity: error / warning / info / success. Colors the badge in the listing. (event_type is accepted as a synonym.) |
service, env, request_id | Metadata: indexed and queryable in the dashboard (search/filters). |
client_ip — end-user IP
The problem it solves
Between the browser and the core there are proxies/ALB. The IP the core sees on the connection is the previous hop (the client's backend or an AWS proxy), not the user's. Without action, source.ip would be the proxy IP — useless for geo, threat and IP-based detection. (Real example: XFF arrived as 44.192.13.3 (AWS) while the user was 179.110.4.205.)
The solution
The client's backend is the only one that sees the browser's real IP — so it forwards that IP in meta.client_ip when calling POST /v1/events.
How the SIEM handles it
- Validates it is an IP (v4/v6). A non-IP value is ignored — it does not drop the event.
- Writes to
source.ipwith maximum precedence (see order below). - Stamps
imtbl.source_ip_origin = "client_asserted"— the UI shows the “User IP” badge. - Feeds GeoIP, threat intel and IP-based detection rules.
client_ipX-Forwarded-For (1º hop)X-Real-IPconnection IPProvenance and trust
client_ip is asserted by the tenant — the client's backend claims the IP, and it is spoofable by whoever controls that backend. However it is self-contained to the tenant (it never crosses another tenant's boundary). A threat/geo hit on a client_asserted value is a claim, not an edge observation.
How to obtain the IP in your backend
Take the first IP from X-Forwarded-For (closest to the user); as a fallback, use the connection IP.
def get_client_ip(request):
xff = request.META.get("HTTP_X_FORWARDED_FOR", "")
if xff:
return xff.split(",")[0].strip() # primeiro hop = IP do usuario
return request.META.get("REMOTE_ADDR")
# ao montar o evento:
meta = {"event_name": "user.login", "client_ip": get_client_ip(request)}Payload example
{
"payload": "{\"user\":\"bob\",\"action\":\"login\"}",
"meta": {
"event_name": "user.login",
"client_ip": "179.110.4.205",
"immutable_trail": "user-bob"
}
}PII note: client_ip is PII and gets sealed into the immutable block (permanent). Decide consciously between forensic evidence and “right to be forgotten” before sending it.
This documentation reflects the current integration behavior. For questions or advanced integrations, contact the support team.