Protecting your API with middleware

Each server SDK has a thin middleware that protects your API with KeenAuth access tokens: net/http in Go, Express and Fastify in Node.js, and Django, Flask and FastAPI in Python. It sits on top of the token verifier and adds no token logic of its own.

What the middleware does

• Reads the token from the Authorization header with the Bearer or DPoP scheme. A token in the URL query (?access_token=) is always refused, because URLs end up in logs and Referer headers. A request with more than one Authorization header or more than one credential is refused too.
• Checks DPoP. With the DPoP scheme, the DPoP proof header, the request method and the absolute request URL (the configured public base URL followed by the request path) are passed to the verifier. A DPoP-bound token sent with the Bearer scheme is rejected, and so is the DPoP scheme with a token that is not DPoP-bound.
• Checks the route's requirements: required scopes (all of them) and, optionally, step-up: acceptable acr values and a maximum age of the sign-in (RFC 9470).
• Asks the revocation cache, if you configure one, whether the token was revoked.
• Attaches the verified claims to the request: protect.ClaimsFrom(ctx) in Go, req.auth in Express and Fastify, request.auth in Django, g.auth in Flask, and the return value of the dependency in FastAPI.
• Answers failures with the standard WWW-Authenticate challenge and a small application/problem+json body. The response never contains the token or any claim value, and the middleware never logs them.

Go (net/http)

The package protect uses only the standard library and works with any router that accepts an http.Handler.

import (
	"github.com/keenauth/keenauth-go/protect"
	"github.com/keenauth/keenauth-go/verify"
)

v, err := verify.New(verify.Config{
	Issuer:   "https://acme.eu.keenauth.app",
	Audience: "https://api.acme.example",
})
if err != nil {
	return err
}
m, err := protect.New(v, protect.Options{
	PublicBaseURL:       "https://api.acme.example",
	ResourceMetadataURL: "https://api.acme.example/.well-known/oauth-protected-resource",
})
if err != nil {
	return err
}

mux := http.NewServeMux()
mux.Handle("GET /orders", m.Require(protect.Requirement{Scopes: []string{"orders:read"}})(http.HandlerFunc(
	func(w http.ResponseWriter, r *http.Request) {
		claims, _ := protect.ClaimsFrom(r.Context())
		fmt.Fprintln(w, claims.Subject)
	})))

age := 10 * time.Minute
mux.Handle("POST /payouts", m.Require(protect.Requirement{
	Scopes:        []string{"payouts:write"},
	AcceptableACR: []string{"urn:idp:acr:mfa"},
	MaxAuthAge:    &age,
})(payoutsHandler))

m.Handler(next) protects a handler with a valid token and no further requirement. Require panics at startup if a scope is not a valid scope token, as http.ServeMux does for an invalid pattern.

Node.js (Express and Fastify)

The middleware is in the @keenauth/node/middleware subpath. Express and Fastify are not dependencies of the SDK.

import { createVerifier } from "@keenauth/node";
import { createProtector, expressAuth, fastifyAuth } from "@keenauth/node/middleware";

const verifier = createVerifier({ issuer: "https://acme.eu.keenauth.app", audience: "https://api.acme.example" });
const protector = createProtector(verifier, { publicBaseUrl: "https://api.acme.example" });

// Express
app.get("/orders", expressAuth(protector, { scopes: ["orders:read"] }), (req, res) => {
  res.json({ sub: req.auth.subject });
});

// Fastify
fastify.decorateRequest("auth", null);
fastify.get("/orders", { preHandler: fastifyAuth(protector, { scopes: ["orders:read"] }) }, async (request) => {
  return { sub: request.auth.subject };
});

Step-up uses { acceptableAcr: ["urn:idp:acr:mfa"], maxAuthAgeSeconds: 600 }. In TypeScript, type the request as AuthenticatedRequest (exported by the subpath) to read req.auth.

Python (Django, Flask, FastAPI)

Install the extra for your framework: pip install "keenauth-python[django]", "keenauth-python[flask]" or "keenauth-python[fastapi]". The framework is imported only by its own adapter module.

from keenauth.middleware import Protector
from keenauth.verify import Verifier, VerifierConfig

verifier = Verifier(VerifierConfig(issuer="https://acme.eu.keenauth.app", audience="https://api.acme.example"))
protector = Protector(verifier, public_base_url="https://api.acme.example")

Django:

from django.http import JsonResponse
from keenauth.middleware.django import protect

@protect(protector, scopes=["orders:read"])
def orders(request):
    return JsonResponse({"sub": request.auth.subject})

Flask:

from flask import g
from keenauth.middleware.flask import protect

@app.get("/orders")
@protect(protector, scopes=["orders:read"])
def orders():
    return {"sub": g.auth.subject}

FastAPI:

from fastapi import Depends, FastAPI
from keenauth.middleware.fastapi import RequireAuth, install
from keenauth.verify import AccessClaims

app = FastAPI()
install(app)

@app.get("/orders")
def orders(claims: AccessClaims = Depends(RequireAuth(protector, scopes=["orders:read"]))):
    return {"sub": claims.subject}

install(app) registers the handler that writes the problem body and one WWW-Authenticate header per challenge. Step-up uses acceptable_acr=["urn:idp:acr:mfa"], max_auth_age=timedelta(minutes=10).

The public base URL (required)

A DPoP proof names the URL the client called (htu). The middleware compares it with the public base URL you configure (PublicBaseURL, publicBaseUrl, public_base_url) followed by the request path. Set it to the address clients use, for example https://api.acme.example or https://acme.example/api. It is required: creating the middleware without it fails at startup with a configuration error, even if your clients send only Bearer tokens.

The middleware never builds this URL from the request's Host header or from X-Forwarded-* headers, and in Python it does not use the framework's own request URL, even if the framework or server is configured to trust forwarded headers. The client chooses these values: if your API is served under two host names, a proof made for one of them could otherwise be replayed to the other. Behind a load balancer or reverse proxy, the public base URL is also what tells the middleware the scheme, host and path prefix that clients see. The public base URL must use https, except for localhost, 127.0.0.1 and [::1] (RFC 6750 §5.3); the scheme and host are compared case-insensitively and a default port (:443, :80) is ignored, both in the base URL and in the proof's htu.

Responses

The challenge uses the scheme the client used. When you set the resource metadata URL, every 401 challenge carries resource_metadata (RFC 9728). The body has type, title, status, code and a fixed detail; it never says why a token was invalid. For your logs, pass an error callback (OnError, onError, on_error): it receives the verifier's error, which contains no token or claim value. Do not log the Authorization or DPoP headers yourself.

Revocation cache (token.revoked)

A JWT access token stays valid until it expires. To reject revoked tokens sooner, handle the token.revoked webhook and record its selector in a revocation cache that the middleware consults. The SDKs include an in-memory cache: protect.NewMemoryRevocationCache (Go), createMemoryRevocationCache (Node.js) and MemoryRevocationCache (Python).

A selector names a token ID (jti), a session (sid), a user (sub), an organization (org_id), an application (client_id), or the pairs organization + user and application + user. A jti selector revokes that token. The other selectors revoke every token issued before the event's not_before; tokens issued later are accepted. Markers are kept for 24 hours and 1 minute by default, the longest possible access token lifetime plus clock skew.

cache := protect.NewMemoryRevocationCache(0, nil)
m, _ := protect.New(v, protect.Options{PublicBaseURL: "https://api.acme.example", Revocation: cache})

// In your verified token.revoked webhook handler:
err := cache.Revoke(protect.Selector{SessionID: "ses_…"}, notBefore)

The in-memory cache belongs to one process. If your API runs on several instances, implement the revocation checker interface over a shared store. If the cache cannot answer, the middleware refuses the request with 503.

Testing

Use the offline test kit to mint tokens for your middleware tests. The kit has no DPoP proof helper; sign a proof in your test with the language's crypto library.