FastAPI Integration

FastAPI Integration

This guide covers how to integrate Skycloak authentication into FastAPI applications using authlib for the OIDC authorization-code flow and a get_current_user dependency that validates bearer tokens against the realm’s JWKS endpoint.

ℹ️

Looking for the Keycloak admin API instead?

This page focuses on FastAPI-specific dependency-injection patterns for authenticating requests. If you need to manage users, roles, or realm configuration from Python code (not just validate incoming tokens), see Python-Keycloak Library, which covers KeycloakOpenID/KeycloakAdmin in depth. The two are complementary: use python-keycloak for admin/provisioning tasks, and the pattern below for validating requests inside your API.

Prerequisites

  • FastAPI 0.110+ application
  • Python 3.10+
  • Skycloak cluster with configured realm and client
  • Basic understanding of FastAPI dependency injection

Quick Start

1. Create an Application in Skycloak

  1. In Skycloak, navigate to your cluster → ApplicationsCreate Application
  2. Set Application Type to Confidential if FastAPI also drives the login redirect (server-rendered flow), or Public if a separate SPA/mobile client obtains tokens and only calls this API as a resource server
  3. Add Redirect URIs if using the authorization-code flow directly from FastAPI: http://localhost:8000/auth/callback for local dev, plus your production URL
  4. Copy the Client ID (and Client Secret, for confidential clients) from the credentials tab

2. Install Dependencies

pip install fastapi authlib httpx python-jose[cryptography] itsdangerous

3. Configure Settings

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    skycloak_issuer: str = "https://your-cluster-id.app.skycloak.io/realms/your-realm"
    skycloak_client_id: str
    skycloak_client_secret: str
    session_secret: str

    class Config:
        env_file = ".env"

settings = Settings()
# .env
SKYCLOAK_CLIENT_ID=your-fastapi-app
SKYCLOAK_CLIENT_SECRET=your-client-secret
SESSION_SECRET=a-random-secret-for-signing-session-cookies

4. Register the OIDC Client with Authlib (Login Flow)

# auth.py
from authlib.integrations.starlette_client import OAuth
from config import settings

oauth = OAuth()
oauth.register(
    name="skycloak",
    server_metadata_url=f"{settings.skycloak_issuer}/.well-known/openid-configuration",
    client_id=settings.skycloak_client_id,
    client_secret=settings.skycloak_client_secret,
    client_kwargs={"scope": "openid profile email"},
)
# main.py
from fastapi import FastAPI, Request
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import RedirectResponse
from auth import oauth
from config import settings

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key=settings.session_secret)

@app.get("/auth/login")
async def login(request: Request):
    redirect_uri = request.url_for("auth_callback")
    return await oauth.skycloak.authorize_redirect(request, redirect_uri)

@app.get("/auth/callback", name="auth_callback")
async def auth_callback(request: Request):
    token = await oauth.skycloak.authorize_access_token(request)
    request.session["user"] = token.get("userinfo")
    request.session["access_token"] = token.get("access_token")
    return RedirectResponse(url="/")

@app.get("/auth/logout")
async def logout(request: Request):
    request.session.clear()
    logout_url = (
        f"{settings.skycloak_issuer}/protocol/openid-connect/logout"
        f"?client_id={settings.skycloak_client_id}"
    )
    return RedirectResponse(url=logout_url)

5. Build a get_current_user Dependency (Bearer Token Validation)

This is the pattern most FastAPI services use for their actual API routes — validating a bearer token issued by any client, not just the one FastAPI itself logged in with.

# security.py
import time
from functools import lru_cache

import httpx
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError

from config import settings

bearer_scheme = HTTPBearer()

@lru_cache(maxsize=1)
def _jwks_cache() -> dict:
    return {"keys": None, "fetched_at": 0.0}

def get_jwks() -> dict:
    cache = _jwks_cache()
    if cache["keys"] is None or time.time() - cache["fetched_at"] > 3600:
        with httpx.Client(timeout=5.0) as client:
            metadata = client.get(
                f"{settings.skycloak_issuer}/.well-known/openid-configuration"
            ).json()
            jwks = client.get(metadata["jwks_uri"]).json()
        cache["keys"] = jwks
        cache["fetched_at"] = time.time()
    return cache["keys"]

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
) -> dict:
    token = credentials.credentials
    try:
        claims = jwt.decode(
            token,
            get_jwks(),
            algorithms=["RS256"],
            audience=settings.skycloak_client_id,
            issuer=settings.skycloak_issuer,
            options={"verify_aud": False},  # see Troubleshooting below
        )
    except JWTError as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid token: {exc}",
        )

    return claims
# routers/users.py
from fastapi import APIRouter, Depends
from security import get_current_user

router = APIRouter()

@router.get("/api/me")
async def read_current_user(user: dict = Depends(get_current_user)):
    return {
        "sub": user["sub"],
        "username": user.get("preferred_username"),
        "email": user.get("email"),
        "roles": user.get("realm_access", {}).get("roles", []),
    }

Role-Based Access Control

# security.py (continued)
from fastapi import Depends, HTTPException, status

def require_realm_role(*roles: str):
    async def checker(user: dict = Depends(get_current_user)) -> dict:
        user_roles = user.get("realm_access", {}).get("roles", [])
        if not any(role in user_roles for role in roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Requires one of: {', '.join(roles)}",
            )
        return user
    return checker
# routers/admin.py
from fastapi import APIRouter, Depends
from security import require_realm_role

router = APIRouter()

@router.get("/api/admin/reports")
async def admin_reports(user: dict = Depends(require_realm_role("admin"))):
    return {"generated_by": user["sub"]}

Testing

# tests/test_security.py
from unittest.mock import patch
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def _fake_claims(roles=None):
    return {
        "sub": "user-123",
        "preferred_username": "testuser",
        "email": "[email protected]",
        "realm_access": {"roles": roles or ["user"]},
    }

def test_me_requires_token():
    response = client.get("/api/me")
    assert response.status_code == 403  # HTTPBearer rejects missing header

@patch("security.jwt.decode")
def test_me_with_valid_token(mock_decode):
    mock_decode.return_value = _fake_claims(roles=["user", "admin"])
    response = client.get("/api/me", headers={"Authorization": "Bearer fake-token"})
    assert response.status_code == 200
    assert "admin" in response.json()["roles"]

@patch("security.jwt.decode")
def test_admin_route_forbidden_without_role(mock_decode):
    mock_decode.return_value = _fake_claims(roles=["user"])
    response = client.get(
        "/api/admin/reports", headers={"Authorization": "Bearer fake-token"}
    )
    assert response.status_code == 403

Production Considerations

  • Cache the JWKS response (as shown above) instead of fetching on every request — Skycloak’s jwks_uri rotates keys infrequently, but hitting it per-request adds unnecessary latency and load.
  • Set verify_aud=False only if your client doesn’t have an audience mapper configured; the safer long-term fix is adding an Audience mapper on the client’s dedicated scope so verify_aud can stay True. See Python-Keycloak Library for the mapper configuration steps (identical for any Skycloak client, not just Python-Keycloak).
  • Run FastAPI behind HTTPS in production; set SessionMiddleware(..., https_only=True) if you use the cookie-session login flow from Step 4.
  • Keep SESSION_SECRET and SKYCLOAK_CLIENT_SECRET in your platform’s secret manager, never in .env committed to source control.

Troubleshooting

  1. Invalid audience on jwt.decode — Skycloak doesn’t include your client_id in the token’s aud claim by default. Add an Audience mapper on the client (see link above) or keep verify_aud=False if you trust the issuer check alone.
  2. 401 on every request despite a valid-looking token — check issuer matches exactly, including trailing slash differences between your SKYCLOAK_ISSUER setting and the token’s iss claim.
  3. JWKS fetch failing intermittently — wrap get_jwks() in retry logic if your network path to the cluster has transient failures; a stale cached key set from the last successful fetch is safer than blocking requests.

Next Steps

Last updated on