Keycloak and FastAPI can protect your Python APIs using OpenID Connect (OIDC), but the most surprising part is how little actual OIDC protocol work you end up doing yourself.
Let’s see it in action. Imagine a simple FastAPI app:
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
# This is where the magic will happen
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
# In a real app, you'd validate the token here and extract user info
return {"message": "This is a protected resource!"}
@app.get("/public/")
async def read_public():
return {"message": "This is a public resource."}
And here’s a minimal Keycloak setup (using Docker for simplicity):
docker run -p 8080:8080 -e KEYCLOAK_ADMIN=admin -e KEYCLOAK_ADMIN_PASSWORD=admin quay.io/keycloak/keycloak:24.0
Once Keycloak is running, you’d create a Realm (e.g., myrealm), then a Client (e.g., myapi), set its access type to "confidential", and enable "Standard flow" and "Implicit flow". You’d also configure a valid Redirect URI, like http://localhost:8000/auth.
Now, when a user requests /items/ from your FastAPI app, the Depends(oauth2_scheme) will trigger Keycloak’s login flow. The user gets redirected to Keycloak, logs in, and Keycloak redirects them back to your app with an access token. FastAPI’s OAuth2PasswordBearer is designed to extract this token from the Authorization: Bearer <token> header.
The real protection isn’t just checking if a token exists; it’s verifying that the token is valid and issued by your trusted Keycloak server for your specific API client. This is where the python-jose library comes in.
Here’s how you’d integrate token validation into your FastAPI app:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from jose.exceptions import ExpiredSignatureError, JWTClaimsError
import requests
app = FastAPI()
# Keycloak details
KEYCLOAK_URL = "http://localhost:8080/realms/myrealm"
KEYCLOAK_CLIENT_ID = "myapi"
KEYCLOAK_CLIENT_SECRET = "YOUR_CLIENT_SECRET" # Get this from Keycloak client config
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# Fetch Keycloak's public keys for token verification
def get_public_key():
try:
response = requests.get(f"{KEYCLOAK_URL}/protocol/openid-connect/certs")
response.raise_for_status() # Raise an exception for bad status codes
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error fetching Keycloak public keys: {e}")
return None
# Cache public keys to avoid frequent requests
public_keys = get_public_key()
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# Decode the JWT token without verification first to get the key ID (kid)
unverified_header = jwt.get_unverified_header(token)
rsa_key = None
if public_keys and 'keys' in public_keys:
for key in public_keys['keys']:
if key['kid'] == unverified_header['kid']:
rsa_key = {
"kty": key["kty"],
"kid": key["kid"],
"use": key["use"],
"n": key["n"],
"e": key["e"],
}
break
if rsa_key is None:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="No matching public key found")
# Verify the token using the retrieved public key
payload = jwt.decode(
token,
rsa_key,
algorithms=["RS256"],
audience=KEYCLOAK_CLIENT_ID,
issuer=f"{KEYCLOAK_URL}/protocol/openid-connect/token"
)
# You can extract user info from the payload, e.g., payload.get("sub")
return payload
except ExpiredSignatureError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTClaimsError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect claims, please check the audience and issuer",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTError:
raise credentials_exception
except Exception as e:
print(f"An unexpected error occurred: {e}")
raise credentials_exception
@app.get("/items/")
async def read_items(payload: dict = Depends(get_current_user)):
# The payload contains verified user information
user_id = payload.get("sub")
return {"message": f"Hello user {user_id}! This is a protected resource!"}
@app.get("/public/")
async def read_public():
return {"message": "This is a public resource."}
The core of this is fetching Keycloak’s public signing keys from the /protocol/openid-connect/certs endpoint. These keys are used by python-jose to verify the signature of the JWT (JSON Web Token) access token. The decode function checks that the token was indeed signed by Keycloak (using the public key) and that it hasn’t been tampered with. We also verify the audience (ensuring the token is for our myapi client) and the issuer (ensuring it came from the correct Keycloak realm).
The get_current_user dependency function is where all the heavy lifting happens. It takes the token provided by OAuth2PasswordBearer, finds the correct public key from Keycloak’s certificate endpoint based on the kid (Key ID) in the token’s header, and then uses that key to decode and verify the token. If verification fails for any reason (expired, wrong audience, wrong issuer, invalid signature), an HTTPException is raised, preventing access to the protected endpoint.
The most common pitfall here is not correctly fetching and using Keycloak’s public keys for verification. If you try to verify with a hardcoded key or a key that’s not the current signing key, your tokens will always fail validation. Keycloak rotates its signing keys, so dynamically fetching them from the /certs endpoint is crucial.
Another subtle point is ensuring your FastAPI application’s audience claim in the JWT matches your Keycloak client ID exactly. If they don’t match, the jwt.decode call will fail with a JWTClaimsError.
The underlying mechanism relies on asymmetric cryptography. Keycloak signs tokens with its private key, and anyone can verify those signatures using Keycloak’s corresponding public key. This allows your FastAPI app to trust tokens issued by Keycloak without needing to share any secrets with Keycloak beyond the initial client secret for obtaining tokens.
The next concept you’ll likely encounter is handling different token scopes and permissions, which are also encoded within the JWT payload.