Middleware Pipeline
How CoreSDK intercepts requests in FastAPI, Flask, and Django. Real middleware classes, exclude_paths, and how to read authenticated claims in handlers.
Middleware Pipeline
CoreSDK provides first-class middleware for FastAPI, Flask, and Django. The middleware verifies JWTs, evaluates Rego policies, and populates the request context with authenticated claims before your handler runs.
FastAPI: CoreSDKMiddleware
Import from coresdk.middleware.fastapi:
from fastapi import FastAPI
from coresdk import CoreSDKClient, SDKConfig
from coresdk.middleware.fastapi import CoreSDKMiddleware
_sdk = CoreSDKClient(SDKConfig(
sidecar_addr="127.0.0.1:50051",
tenant_id="acme",
service_name="orders-api",
fail_mode="open",
))
app = FastAPI()
app.add_middleware(CoreSDKMiddleware, sdk=_sdk)exclude_paths
Pass exclude_paths to skip auth and policy checks for specific routes (e.g., health checks, public endpoints):
app.add_middleware(
CoreSDKMiddleware,
sdk=_sdk,
exclude_paths=["/healthz", "/readyz", "/metrics"],
)Reading claims in handlers
After the middleware runs, authenticated claims are on request.state.coresdk_user:
from fastapi import FastAPI, Request
@app.get("/api/orders")
async def get_orders(request: Request):
user = request.state.coresdk_user
# user.sub, user.email, user.role, user.tenant_id, user.claims
return await db.fetch_orders(tenant_id=user.tenant_id, user_id=user.sub)Flask: CoreSDKFlask
Import from coresdk.middleware.flask:
from flask import Flask, g
from coresdk import CoreSDKClient, SDKConfig
from coresdk.middleware.flask import CoreSDKFlask
_sdk = CoreSDKClient(SDKConfig(
sidecar_addr="127.0.0.1:50051",
tenant_id="acme",
service_name="orders-api",
fail_mode="open",
))
app = Flask(__name__)
CoreSDKFlask(_sdk, app)exclude_paths with Flask
CoreSDKFlask(_sdk, app, exclude_paths=["/healthz", "/readyz"])Reading claims in Flask handlers
Claims are on g.claims after the middleware runs:
from flask import g, jsonify
@app.route("/api/orders")
def get_orders():
claims = g.claims
# claims["sub"], claims["email"], claims["role"], claims["tenant_id"]
orders = db.fetch_orders(tenant_id=claims["tenant_id"], user_id=claims["sub"])
return jsonify(orders)Django: CoreSDKMiddleware
Add "coresdk.django.CoreSDKMiddleware" to the MIDDLEWARE list in settings.py:
MIDDLEWARE = [
"django.middleware.security.SecurityMiddleware",
"coresdk.django.CoreSDKMiddleware", # add after SecurityMiddleware
"django.contrib.sessions.middleware.SessionMiddleware",
# ...
]
CORESDK = {
"SIDECAR_ADDR": "127.0.0.1:50051",
"TENANT_ID": "acme",
"SERVICE_NAME": "orders-api",
"FAIL_MODE": "open",
"EXCLUDE_PATHS": ["/healthz/", "/readyz/"],
}Reading claims in Django views
Claims are on request.coresdk_claims:
from django.http import JsonResponse
def get_orders(request):
claims = request.coresdk_claims
# claims["sub"], claims["email"], claims["role"], claims["tenant_id"]
orders = list(Order.objects.filter(tenant_id=claims["tenant_id"]))
return JsonResponse({"orders": orders})Default pipeline order
All three frameworks follow the same pipeline:
Incoming request
↓
[1] Auth check — verifies JWT via sidecar, populates claims
↓
[2] Policy check — evaluates Rego policy, short-circuits on deny
↓
[3] Tracing — opens OTel span, attaches trace/span IDs
↓
[4] Handler — your application code
↓
[4] Tracing — closes span, records status code
↓
ResponseAudit logging happens asynchronously off the hot path after the response is sent.
Tower middleware (Rust / axum)
For Rust services built on axum, use Tower middleware via middleware::from_fn_with_state:
use axum::{middleware, Router};
use coresdk_engine::Engine;
let engine = Engine::from_env()?;
let app = Router::new()
.route("/api/orders", axum::routing::get(get_orders_handler))
.layer(middleware::from_fn_with_state(
engine.clone(),
coresdk_engine::middleware::auth_and_policy,
));Next steps
- Request Context — full reference for claims and context fields
- Error Handling — how denied requests become RFC 9457 responses
- Intent Annotations — adding semantic labels to traced handlers