FlaskPythonWSGI

Flask Integration

Integrate ImmutableLog into any Flask application with an extension that uses Flask's native hooks. Automatically captures all HTTP requests — errors, successes, and exceptions — with support for the application factory pattern via `init_app`.

Installation

The extension only depends on Flask and `requests`. Copy the `middleware.py` file into your application directory.

bash

pip install flask requests

Extension Code

Paste the code below into `middleware.py`. The extension follows Flask's extension pattern, with `init_app` support for the application factory pattern, and uses `flask.g` to store per-request state.

python

# middleware.py
import hashlib
import json
import logging
import time
import uuid
from datetime import datetime, timezone

import requests as req_lib
from flask import Flask, g, request

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 ImmutableLogAudit:
    """Flask extension for ImmutableLog audit logging."""

    def __init__(
        self,
        app: Flask | None = None,
        api_key: str = "",
        api_url: str = "",
        service_name: str = "flask-service",
        env: str = "production",
        skip_paths: list[str] | None = None,
    ):
        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"])

        if app is not None:
            self.init_app(app)

    def init_app(self, app: Flask):
        app.extensions["immutablelog"] = self
        app.before_request(self._before_request)
        app.after_request(self._after_request)
        app.teardown_request(self._teardown_request)

    def _before_request(self):
        if request.path in self.skip_paths:
            return
        g._imtbl_started_at = time.time()
        g._imtbl_request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
        g._imtbl_body = request.get_data(cache=True)
        g._imtbl_exc_handled = False

    def _after_request(self, response):
        if request.path in self.skip_paths:
            return response
        if not getattr(g, "_imtbl_exc_handled", False):
            self._emit(response=response, exc=None)
        response.headers["X-Request-Id"] = getattr(g, "_imtbl_request_id", "")
        return response

    def _teardown_request(self, exc):
        if exc is not None and not getattr(g, "_imtbl_exc_handled", False):
            g._imtbl_exc_handled = True
            self._emit(response=None, exc=exc)

    def _emit(self, response=None, exc=None):
        try:
            started = getattr(g, "_imtbl_started_at", None)
            latency_ms = int((time.time() - started) * 1000) if started else None
            status_code = response.status_code if response else (500 if exc else None)
            request_id = getattr(g, "_imtbl_request_id", str(uuid.uuid4()))
            body = getattr(g, "_imtbl_body", b"")
            kind = severity_from(status_code, exc)

            event_name = getattr(g, "imtbl_event_name", None) or f"http.{request.method}.{request.path}"

            ip = request.headers.get("X-Forwarded-For", request.remote_addr or "unknown")
            ip = ip.split(",")[0].strip()

            user_context = {
                "ip": ip,
                "user_agent": request.user_agent.string or "unknown",
            }

            payload = {
                "id": str(uuid.uuid4()),
                "kind": kind,
                "message": self._generate_message(status_code, exc, kind),
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "context": user_context,
                "request": {
                    "request_id": request_id,
                    "method": request.method,
                    "path": request.path,
                    # query params passam por redact() — nunca enviar segredos em claro.
                    "query_params": redact(dict(request.args)) 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: g.imtbl_trail (view) -> header X-Imtbl-Trail.
            trail = sanitize_trail(getattr(g, "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

            req_lib.post(f"{self.api_url}/v1/events", json=event, headers=headers, timeout=5)

        except Exception as e:
            logger.warning("ImmutableLogAudit _emit failed: %s", e)

    def _generate_message(self, status_code, exc, kind: str) -> str:
        method = request.method
        path = request.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.

Configuration and Registration

The extension supports two usage modes: direct instantiation with the Flask app, or the application factory pattern with `init_app`. Use the `init_app` pattern in larger projects to facilitate testing and modularization.

Mode 1: Direct instantiation

python

import os
from flask import Flask
from middleware import ImmutableLogAudit

app = Flask(__name__)

ImmutableLogAudit(
    app,
    api_key=os.environ.get("IMTBL_API_KEY", ""),
    api_url=os.environ.get("IMTBL_URL", "https://api.immutablelog.com"),
    service_name="meu-servico-flask",
    env="production",
)

Mode 2: Application Factory (recommended)

python

import os
from flask import Flask
from middleware import ImmutableLogAudit

# Instancia sem app — registra depois com init_app
# Instantiate without app — register later with init_app
immutablelog = ImmutableLogAudit(
    api_key=os.environ.get("IMTBL_API_KEY", ""),
    api_url=os.environ.get("IMTBL_URL", "https://api.immutablelog.com"),
    service_name="meu-servico-flask",
    env="production",
)

def create_app():
    app = Flask(__name__)
    immutablelog.init_app(app)
    # ... registrar blueprints, etc.
    return app

Never hardcode the token. Use environment variables and load via os.environ.get() or python-dotenv.

How it works

The extension registers three hooks in Flask. `before_request` captures the start timestamp, generates the `request_id`, and reads the body. `after_request` calculates latency and emits the success/info/warning event. `teardown_request` captures unhandled exceptions and emits the error event with exception details.

before_request

Captures timestamp, generates request_id, and reads body via flask.g

after_request

Calculates latency, emits event, and injects X-Request-Id into response

teardown_request

Captures exceptions, emits error event, and prevents double-emit

The g._imtbl_exc_handled prevents double-emit when both `after_request` and `teardown_request` are triggered.

Custom event names

Use `g.imtbl_event_name` inside any view function or blueprint to override the automatically generated event name. Flask's `g` ensures the value is isolated per request.

python

from flask import g, jsonify

@app.post("/checkout")
def process_checkout():
    # Sobrescreve o nome do evento para esta view
    # Override the event name for this view
    g.imtbl_event_name = "payment.checkout.initiated"

    # ... logica de negocio / business logic ...

    return jsonify({"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 `g.imtbl_trail` in a view or propagate it across services with the `X-Imtbl-Trail` header. The value goes through `sanitize_trail` before sending.

python

from flask import g, jsonify

@app.post("/flows/<flow_id>/run")
def run_flow(flow_id):
    g.imtbl_trail = f"flow-{flow_id}"
    # ... logica de negocio / business logic ...
    return jsonify({"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 instantiating the extension to ignore specific routes. Ideal for load balancer health checks and metrics endpoints.

python

ImmutableLogAudit(
    app,
    api_key=os.environ["IMTBL_API_KEY"],
    api_url="https://api.immutablelog.com",
    # Rotas ignoradas / Ignored routes
    skip_paths=["/health", "/healthz", "/metrics", "/ping"],
)

Usage with Blueprints

The extension works automatically with blueprints, as the `before_request` and `after_request` hooks are registered on the main application. No additional configuration is needed in blueprints.

python

from flask import Blueprint, g, jsonify

payments_bp = Blueprint("payments", __name__, url_prefix="/payments")

@payments_bp.post("/checkout")
def checkout():
    # g.imtbl_event_name funciona normalmente em blueprints
    # g.imtbl_event_name works normally in blueprints
    g.imtbl_event_name = "payment.checkout.initiated"
    return jsonify({"status": "ok"})

# app.py
from flask import Flask
from middleware import ImmutableLogAudit
from payments import payments_bp

def create_app():
    app = Flask(__name__)
    ImmutableLogAudit(app, api_key=..., api_url=...)
    app.register_blueprint(payments_bp)
    return app

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.ip with 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.

source.ip precedence:client_ipX-Forwarded-For (1º hop)X-Real-IPconnection IP

Provenance 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.

python

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

json

{
  "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.