Provider-Agnostic OIDC Auth Flow for Your Apps (PyJWT + FastAPI)
Learn about OIDC auth flow and how to secure your app by incorporating OIDC flow that can work with any IdP of your choice like Okta, Auth0, MS AD, etc.
Join the DZone community and get the full member experience.
Join For FreeWhen building web applications, handling authentication securely and reliably is critical. That's where OpenID Connect (OIDC) comes in. OIDC is a thin identity layer built on top of OAuth 2.0, and it gives your app the ability to verify who a user is and get some basic info about them, without the developer having to store passwords or build their own login systems from scratch. Things like passwords and access control will be managed by the Identity provider (IdP) thereby giving us a clear separation of responsibilities.
In this article, we will:
- Go over the general OIDC and understand the authorization code flow.
- Examine a JSON Web Token (JWT)
- Use PyJWT python library to decode and validate tokens.
- Secure routes on a FastAPI backend server (agnostic to the IdP being used)
A Quick Primer on OIDC
OIDC extends OAuth 2.0 by introducing an ID token—a JWT (JSON Web Token) that contains claims about the authenticated user. While OAuth 2.0 is great for authorization (e.g., "Can this user access the documents folder in the dashboard?"), it doesn't actually authenticate users ("Is the user who they claim they are?"). OIDC fills this gap. It also standardizes things like login, logout, and user info retrieval, making it easy for the developer to extend it for use cases like Single Sign-On (SSO) and federated identity setups.
The Authorization Code Flow (With PKCE)
For most web apps and single page apps, the Authorization Code Flow with PKCE is the go-to. This flow diagram shows the basic flow and we will briefly go through the steps below:
+-------------+
| |
| Browser |
| (User) |
| |
+-------------+
|
| 1. Redirect to Authorization Endpoint
|------------------------------------------------------>
| |
| 2a. User logs in |
| --------------------------> |
| |
| 2b. Redirect back with Authorization Code |
|<------------------------------------------------------
| |
| |
+-------------+ +--------------------+
| | | |
| Frontend | | Identity Provider |
| App | | (IdP) |
| | | |
+-------------+ +--------------------+
| |
| 3a. Backend sends code to Token Endpoint |
|----------------------------------------------------->|
| |
| 3b. IdP returns ID Token + Access Token |
|<-----------------------------------------------------|
| |
| 3c. App validates ID Token and logs user in |
|-----------------------------------------------------> (Actual app logic starts)
Redirect to the Authorization Server
Your frontend redirects the user to the identity provider (IdP) with a request like its shown below. Effectively this follows from the "sign-in" button on your app's UI.
GET https://auth.customdomain.com/authorize?
response_type=code&client_id=my-client-id&redirect_uri=https://myapp.customdomain.com/callback
&scope=openid profile email&state=random123&code_challenge=abc123
&code_challenge_method=S256
A few things of note here:
client_id=my-client-id
here is the client-id of the frontend app. This is what identifies the app on the IdP's settings interface- The
redirect_uri=https://myapp.customdomain.com/callback
specifies the app's callback url on a successful login. This is the uri where the IdP should send the user back to if logged in successfully. - The scope parameter must include
openid
to signify use of OIDC
User Logs In
The user authenticates on the IdP (e.g., Okta, Auth0, MS Active directory, your custom IdP). If successful, the IdP redirects them back to your app with an authorization code like so:
https://myapp.customdomain.com/callback?code=xyz789&state=random123
Exchange Code for Tokens
As soon as the user logs in fine, the app sends a POST request from the backend to the token endpoint on the IdP to exchange the code for an access and id token:
POST /token
Host: auth.customdomain.com
Content-Type: application/x-www-form-urlencoded
with the payload:
{
"grant_type": "authorization_code",
"code": "xyz789",
"redirect_uri": "https://myapp.customdomain.com/callback",
"client_id": "my-client-id",
"code_verifier": "original-code-verifier"
}
The response will look something like:
{
"id_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"access_token": "eyJz93a...k4laUWw",
"expires_in": 3600,
"token_type": "Bearer"
}
This response contains an access token, an id token and it denotes that this will expire in 1 hour. These tokens are encoded and carry embedded claims, including details like the user's identity, the issuing authority, and other metadata. This is indeed the JSON Web Token (JWT) and the app needs to decode + validate it before serving the pages/views intended for that user. In the next section, we will look at an example of a decoded JWT.
A Decoded JWT
JWTs can be decoded using a variety of libraries that parse the token and extract its claims for inspection or validation. In Node.js, one could use jsonwebtoken, in python PyJWT is the popular library. In this section we will look at an example of a decoded JWT. Note a JWT can have a lot more attributes than those shown below, for instance "roles"
and "scp"
(scope) have not been set on this one.
In later sections we will also look at some real tokens and decode them using python code.
{
"iss": "https://auth.customdomain.com",
"sub": "1134113669",
"aud": "my-client-id",
"exp": 1711924860,
"iat": 1711921260,
"name": "Shane Durant",
"email": "[email protected]"
}
This tells us:
Field | Meaning |
---|---|
sub |
The user’s unique ID - subject |
iss |
Who issued the token - issuer |
aud |
Who the token is for - audience |
exp |
When it expires - expiry |
iat | When the token was issued - issued at |
name/email | Optional profile info - this is useful for apps to use when audit logging which user/email ran what queries. In Microsoft AD based IdP, this is sent as the "upn" field - User Principal Name |
Validating the Token
The backend app on receiving the token from the IdP, must validate these three things:
- The token signature (using the IdP’s public key)
- The audience
- The expiration time
If you’re using a popular IdP like Auth0, Okta, Azure AD, or AWS Cognito, they all provide a "JSON Web Key Set" endpoint your app can use to fetch and cache public keys for signature validation (Step 1 above).
Decoding a Token Using PyJWT
PyJWT is the most common python library to decode JSON Web Tokens. Here we will look at the python code to see decoding in action.
- To decode and verify a JWT’s signature, the identity provider’s (IdP) public signing key is required. This key ensures the token was issued by a trusted source and hasn’t been tampered with. The JWT comes with headers which specify what the key id
'kid'
is for the public key that the IdP used to sign the token. The public keys (full set of keys) are obtained by calling the IdP's JWKS endpoint. - The header also has info on what cryptographic algorithm was used. Specified as
'alg'
- The backend (of the app) where the decoding will happen, will need to know the audience if the token as the
'aud'
claim - By default the
decode()
method provided by the PyJWT library will return the decoded token if valid, or throw an exception if invalid (eg: expired, cant verify signature, etc). However for development purposes, you can passoptions={"verify_signature": False}
to the call and still read the token. We will explore both ways.
>> import jwt
>>> public_keys
{'230498151c214b788dd97f22b85410a5': '-----BEGIN PUBLIC KEY-----\nMIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAzemPWaeAPbFKZ1L/Vjkp\nwiAo697Ve/k9iKHpgegVJvwIBEBC8aX3P88hHPkL1/nfI4IMUp6KkeuFkRaLsXRs\nJ53Yn5/OUoSTNyiy74zL4viVKy+mzEnUvcu098K27KgyDxqZhsJmAur7qoN1GZLg\neBu5lXxIVZtqW2vCWy2cUNROwgfm/cFqWxZnvJiZxEhtYMKwub+LaHPqA8R7i2CK\nugqni1vTrMhvukQ00OiXPdZ6PMy1JcRWRW75DKZLWA+79SUUoEqxyNuv/sKw+jta\nh3UNkGzj45A3hvpG74YNHWO/w6Jk+A5XbCcGJCSALq+l7yNhqNRAPwz+F9CeTJNZ\n9E7mnL1+/jSg0tnR+CZ8wKB0fVC1IXDBsCjMZ+Zm1lAMs3ufJ4xPvzuzlPRLrbm8\nW0/UMQdIHq6CXUAoLSwQfkWSJ/9qwa8Xj4ggtF8fu0FivTdcrLVqS9S36Bz9NQck\nKSrXOIRs71HdZGsUxbyRap/Ye1OHGLfcPxozLloKQhw3f8lyNA9xSn4zNEtSdPWq\n/YTl7f3I5ANqjosopWVj1fXTU4CU2wRAFBSEKdo+BsHuhY8a209MQVtE6xBhD2aD\nOqAVb5of7MqV7zEsvQKkS4vFe0yHEQRvP6RHpSzxg9SJEehzNO7uENT+MNIR/Ydd\np1HCHWNPTIhC0MsxD+eXzAsCAwEAAQ==\n-----END PUBLIC KEY-----\n'}
>>> audience = ["WestCoast"]
>>> jwt_token
'eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6IjIzMDQ5ODE1MWMyMTRiNzg4ZGQ5N2YyMmI4NTQxMGE1In0.eyJzb21lIjoicGF5bG9hZCIsImF1ZCI6Ildlc3RDb2FzdCJ9.kxqQP-L6LhPYYrG7LRW0I-9LSlKR0JskGIxejqOVp6EjkiQ1qVYhFBSEiYdu4SwhwIKBJqIoEbbAzeQOp6Jnj9lUh0GhNu6m3gOcXuCD5ebhb_rHgQDFaGsX0YwdG6wZQGif8sgoIC_zuXDeR7ieqlv-4mrD1Bqp2ethAQHCskzpA81ilKEfDDdtKNzJpe9anYtdOuNAjICJLN0WQlTmgvacNF4tjQQrkCFtk4sDc0kj1sertAcA1g89cgEyFkUh65sIA5IGga-sm3A9mORF82pR6-EXWnVD8cOb0_YW3GJiN4DZOz5gy_Hnpg5whApXYszj0Agk8DB03tblL_qPyOEaFjqKWSKxJzSBoXXVEtgrXP0mZR_H0GVtK_uXkIJi1P75-A1kwHF1dZ9Zm61gcnu9SyDJMMTJe6D95uY3FkCfMcv7zBXfx9sT6OLXIVVoc8oVNHkrJ5kgE22CJ0JSslBUSgky6QqPzZBbNYBCgnt7ovaS21rtt4k4lwXAF2WMpDMiF7mQkTnUUoDGsB_itELQhjWnZ5-Bu8d_NTSz6iqCNNPOsFYpfQIPVH2kLZ0fYM1C02YYx9_I9fbbfd6ZFVs4mcg_Gs-VNKkXSLp-jUV0jjBS2V0g6YimzknOXFR9RfDtB0RqVI_nRxSl6dX5RNvZkaTCOpVL_F5D1WkEtdQ'
>>> jwt_headers = jwt.get_unverified_header(jwt_token)
>>> jwt_headers
{'typ': 'JWT', 'alg': 'RS256', 'kid': '230498151c214b788dd97f22b85410a5'}
>>> signing_key = public_keys[jwt_headers['kid']]
>>> jwt.decode(jwt_token, signing_key, audience=audience, algorithms=jwt_headers['alg'])
{'some': 'payload', 'aud': 'WestCoast'}
>>> expired_jwt_token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6IjIzMDQ5ODE1MWMyMTRiNzg4ZGQ5N2YyMmI4NTQxMGE1In0.eyJzb21lIjoicGF5bG9hZCIsImV4cCI6MTIzLCJhdWQiOiJXZXN0Q29hc3QifQ.EBJE2W1SNkVAAPPETulqf9InrOYmGoVheQzJhkE4_oViX3zuHisJ0dL8d__fD1gwsDBDMTDnSQQSIgcO2VPGtPvi6a2OUEQV1xwa4jST_TLZIfPHZrqGeAmPxxr7_VR-HlDXEZUsN30Rr1VMdikYepTDRtpMHSSrrERWgPER4ezOxlqeIqQ21Tid3T8VZRpdpEVd6LwnERQEtr6R2sdnPRUl_YZXU-Vvs-qsA1a_7s4j89_QxHMKwGx6rPQtRQ7tb6HNuJUReyvzK0iJaqvwZfYSl6PILLQHwYdLfQ14TbmEgVBrTVksEpKjVn3aUPLTgGigLzQF2HwG7dDJiwshVu-BUnB6145QKb4IxnKaoXeeW7VDmopB7vncOIbKeJVdAVknwlTAoG_wi7HfnUqhAVYDLhDHIDBiAwx5eb19vyQCmevtcqMtCafeNVwEEXOBqIkbLIv-LE9ox0RlFtrv22C2TQ-NHiThtcVzJLQEyz12NmDtyfqyXU1hrXv9VqifDzVe-UTnNFO71SuqYKYAMVRCDMrdPou_8xPfWgkMT2_X--gmkDf5GRAVBNRgJ2K01wUU3YCkCgsSAaQ1At-WjsLllPnouDhEGshOYARrXWRnfk3NP1nxfhyTm7oa9tz1COmLszdRZRuwkTOaZ8DDpa1FA8YOS2pdgzTnhKXrvsY"
>>> jwt.decode(expired_jwt_token, signing_key, audience=audience, algorithms=jwt_headers['alg'])
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/src/product/common/.venv/lib/python3.11/site-packages/jwt/api_jwt.py", line 168, in decode
decoded = self.decode_complete(
^^^^^^^^^^^^^^^^^^^^^
File "/src/product/common/.venv/lib/python3.11/site-packages/jwt/api_jwt.py", line 136, in decode_complete
self._validate_claims(
File "/src/product/common/.venv/lib/python3.11/site-packages/jwt/api_jwt.py", line 199, in _validate_claims
self._validate_exp(payload, now, leeway)
File "/src/product/common/.venv/lib/python3.11/site-packages/jwt/api_jwt.py", line 234, in _validate_exp
raise ExpiredSignatureError("Signature has expired")
jwt.exceptions.ExpiredSignatureError: Signature has expired
>>> jwt.decode(expired_jwt_token, signing_key, audience=audience, algorithms=jwt_headers['alg'], options={"verify_signature": False})
{'some': 'payload', 'exp': 123, 'aud': 'WestCoast'}
- In the backend server implementation, the App will have to maintain the dictionary of signing keys (
public_keys
in this example). While this can be static, it is preferred that the backend polls the IdP's JWKS endpoint every so often to keep this mapping up to date. - The JWT will part of the Authorization header in the incoming API request from your frontend. The backend code will need to parse the request header and pluck out this token. In FastAPI this is done via dependency injection. But under the hood, this is what is being done:
import re
authorization_header = request.headers.get("Authorization")
jwt_token = re.search("Bearer (.*)$", authorization_header).group(1)
In the above example we see 2 keys.
- Both signed by the same signing key specified by the
kid = 230498151c214b788dd97f22b85410a5
- using
RS256
algorithm and having the sameaud
claim.
The first is a valid token. The second is an expired token. The library throws the jwt.exceptions.ExpiredSignatureError: Signature has expired
exception which can be caught and logged by your backend server while rejecting the API request.
To show that it is indeed expired, we also tried decoding without verifying the signature and we can see in this example, the exp
value is indeed in the past (compared to current epoch time).
Securing Your FastAPI Routes With OIDC Auth
In this section, we will explore a FastAPI webserver implementation that authorizes user requests from authenticated users when using an OAuth2.0 identity provider. This section only focuses on the auth aspects and assumes the reader is familiar with FastAPI and Python webserver concepts.
The Well-Known URL
A well-known URL in the context of OpenID Connect (OIDC) typically looks like this (for most if not all Identity providers):https://<your-idp-domain>/.well-known/openid-configuration
This endpoint returns a JSON document describing the identity provider's configuration, including URLs for authorization, token, user info, supported claims, etc. This is an unauthenticated endpoint providing all the details needed for your app's backend and frontend to function. This is what the app can use at system startup/bootstrap before declaring itself ready to serve routes.
An example output from this endpoint from a Azure Active directory OIDC server:
{
"issuer": "https://login.microsoftonline.com/common/v2.0",
"authorization_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/authorize",
"token_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/token",
"userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo",
"jwks_uri": "https://login.microsoftonline.com/common/discovery/v2.0/keys",
"response_types_supported": ["code", "id_token", "code id_token"],
"scopes_supported": ["openid", "profile", "email", "offline_access"],
"id_token_signing_alg_values_supported": ["RS256"],
"end_session_endpoint": "https://login.microsoftonline.com/common/oauth2/v2.0/logout"
}
As we can see from above, this has the url for the frontend to pull up on logout (end_session_endpoint
). This also has the url the backend uses to exchange the code for a token on successful login (token_endpoint
) and the url to pull the set of signing keys (JWKS) that we talked about earlier (jwks_uri
). We can also see the issuer
, and the supported algorithms which we talked about earlier.
Incorporating Auth into Your FastAPI Routes
In this section, we look at a real webserver backend implementation that works with a frontend built with the PKCE flow against any OIDC IdP:
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import jwt
import httpx
from jwt import PyJWKClient
app = FastAPI()
security = HTTPBearer() # Extracts Bearer token from Authorization header
OIDC_ISSUER = "https://login.microsoftonline.com/common/v2.0"
CLIENT_ID = "your-client-id" # Replace with your client ID
REDIRECT_URI = "http://localhost:3000/callback" # Must match your app's redirect URI..this assumes the app is running locally
AUDIENCE = "your-audience" # This is the 'aud' claim set by your IdP. In microsoft/azure AD servers, audience is the same as the client_id
async def get_signing_key(token: str):
async with httpx.AsyncClient() as client:
# Get JWKS URI from OIDC metadata
oidc_config = await client.get(f"{OIDC_ISSUER}/.well-known/openid-configuration")
jwks_uri = oidc_config.json()["jwks_uri"]
async with httpx.AsyncClient() as client:
keys_resp = await client.get(jwks_uri)
jwk_client = PyJWKClient.from_jwks_data(keys_resp.json())
# this unpacks the token headers to figure out the 'kid' and returns the corresponding signing key
return jwk_client.get_signing_key_from_jwt(token).key
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
try:
signing_key = await get_signing_key(token)
# this is the exact decode call we talked in the previous section
payload = jwt.decode(
token,
signing_key,
algorithms=["RS256"],
audience=AUDIENCE,
issuer=OIDC_ISSUER,
)
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expired")
except jwt.InvalidTokenError as e:
raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
# ---- Secure API Routes ----
@app.get("/secure-area")
async def private_route(user=Depends(verify_token)):
return {"message": "Hello from /secure-area!", "user": user}
@app.get("/data")
async def data_route(user=Depends(verify_token)):
return {"message": "data access granted", "email": user.get("email")}
# ---- Token Exchange Endpoint for Frontend (client/frontent to hit during PKCE) ----
class TokenExchangeRequest(BaseModel):
code: str
code_verifier: str
# When the user first logs-in via the IdP portal, the frontend gets a code
# This code can be passed on to THIS backend route, which will talk to the IdP
# and exchange the code for the token, which the frontend can then use for all
# API requests from thereon.
# Steps 3a, 3b, 3c of the flow diagram.
@app.post("/token")
async def exchange_code(req: TokenExchangeRequest):
token_url = f"{OIDC_ISSUER}/token" # this can also be gotten from the well-known url's response
async with httpx.AsyncClient() as client:
try:
response = await client.post(
token_url,
data={
"client_id": CLIENT_ID,
"grant_type": "authorization_code",
"code": req.code,
"redirect_uri": REDIRECT_URI,
"code_verifier": req.code_verifier,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail=response.text)
return response.json()
except Exception as e:
raise HTTPException(status_code=500, detail=f"Token exchange failed: {str(e)}")
- The above webserver offers two routes both secured by OIDC auth.
- This server can be wired to talk to any OIDC IdP such as Azure, Okta, Auth0, Cognito and so on. All you'd have to change is the
OIDC_ISSUER
andAUDIENCE
value to denote the IdP URL and what audience claim the issuer is setting. - Concepts such as "roles" and "scopes" have not been covered in this implementation but can easily be incorporated into the code above to implement Role Based Access Control (RBAC). This allows for fine grained access to resources. Eg: Certain users of your app might be designated "admins" and can have more privileges than regular users. Certain users could be "read-only" and so on.
- We have also focused purely on the backend implementation using FastAPI (one of the most popular backend web frameworks in python). The frontend implementation is left as an exercise for the reader.
Conclusion
The article covered the fundamentals of OIDC auth flow with hands-on examples of decoding real-life JWTs. We also implemented the backend server in FastAPI that presents routes secured by authorization. The implementation is agnostic of your Identity provider (IdP) as long as they implement OIDC on top of OAuth2.0. This way of implementing auth takes away the burden of maintaining passwords, user DBs, ACL's from the app and the app dev thereby making the application more secure and also faster to develop. Its a stateless and scalable implementation that is uniform across frontends, backends, and mobile apps and can also be extended to support SSO and federated scenarios.
Opinions expressed by DZone contributors are their own.
Comments