Guide: Microservice JWT¶
Issue RS256 tokens in an auth service, verify them in a separate resource service using only the public key from /.well-known/jwks.json. See examples/jwt-microservice/.
Install¶
pip install "sreekarnv-fastauth[standard]" # auth service
pip install "joserfc" "httpx" # resource service
Generate RSA keys¶
generate_keys.py produces a 2048-bit RSA pair in PEM format. Commit the public key, never the private one.
Auth service (auth_service.py)¶
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastauth import FastAuth, FastAuthConfig, JWTConfig
from fastauth.adapters.sqlalchemy import SQLAlchemyAdapter
from fastauth.providers.credentials import CredentialsProvider
_PRIVATE_KEY = Path("private_key.pem").read_text()
_PUBLIC_KEY = Path("public_key.pem").read_text()
adapter = SQLAlchemyAdapter(engine_url="sqlite+aiosqlite:///./auth.db")
config = FastAuthConfig(
secret="unused-for-rs256-but-required", # required field, unused by RS256
providers=[CredentialsProvider()],
adapter=adapter.user,
token_adapter=adapter.token,
jwt=JWTConfig(
algorithm="RS256",
private_key=_PRIVATE_KEY,
public_key=_PUBLIC_KEY,
jwks_enabled=True,
access_token_ttl=900,
refresh_token_ttl=2_592_000,
),
base_url="http://localhost:8000",
)
auth = FastAuth(config)
@asynccontextmanager
async def lifespan(app: FastAPI):
await adapter.create_tables()
await auth.initialize_jwks() # required before issuing RS256 tokens
yield
app = FastAPI(lifespan=lifespan)
auth.mount(app)
Run with uvicorn auth_service:app --port 8000. Public keys are served at http://localhost:8000/.well-known/jwks.json.
Resource service (resource_service.py)¶
A separate FastAPI app that verifies tokens against the auth service's JWKS endpoint. No FastAuth, no DB.
import httpx
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from joserfc import jwt
from joserfc.jwk import KeySet
app = FastAPI()
bearer = HTTPBearer()
# Cache the KeySet; refresh on key rotation.
_KEYSET: KeySet | None = None
def get_keyset() -> KeySet:
global _KEYSET
if _KEYSET is None:
resp = httpx.get("http://localhost:8000/.well-known/jwks.json")
_KEYSET = KeySet.import_key_set(resp.json())
return _KEYSET
async def current_user(credentials: HTTPAuthorizationCredentials = Depends(bearer)):
try:
claims = jwt.decode(credentials.credentials, get_keyset())
except Exception as e:
raise HTTPException(401, "Invalid token") from e
if claims.claims.get("type") != "access":
raise HTTPException(401, "Not an access token")
return claims.claims
@app.get("/me")
async def me(user=Depends(current_user)):
return {"sub": user["sub"], "email": user.get("email")}
@app.get("/health")
async def health():
return {"status": "ok"}
Run with uvicorn resource_service:app --port 8001.
Try it¶
# Get an access token from the auth service
ACCESS=$(curl -s -X POST http://localhost:8000/auth/login \
-H "Content-Type: application/json" \
-d '{"email":"a@b.com","password":"s3cur3!"}' | python -c "import json,sys; print(json.load(sys.stdin)['access_token'])")
# Use it on the resource service
curl http://localhost:8001/me -H "Authorization: Bearer $ACCESS"
# -> {"sub":"...","email":"a@b.com"}
Notes¶
- The resource service needs no shared secret — only the public key.
- Cache the
KeySetin-process; refresh when key rotation may have happened (e.g. periodic warm-refresh). - If you supply your own PEM
private_key/public_key, they survive restarts. Auto-generated keys are ephemeral — see JWT feature.