Since fastapi-sso handles the communication with the login provider but does not manage session state, you must implement your own authentication mechanism (e.g., using JWTs) to protect endpoints.
To add the lock 🔒 icon to your Swagger/OpenAPI documentation and enforce authentication via cookies, use FastAPI's Security and APIKeyCookie classes.
Requirements
fastapifastapi-ssopython-jose[cryptography] (to sign and verify JWTs)
Implementation Pattern
- Login: Use
sso.get_login_redirect() to send users to the provider. - Callback: Use
sso.verify_and_process(request) to get the OpenID object, then generate a signed JWT and store it in a cookie. - Protection: Create a dependency function using
Security(APIKeyCookie(name="token")) that decodes the JWT and returns the user data. Use this dependency in your protected routes via Depends().
import datetime
from fastapi import FastAPI, Depends, HTTPException, Security, Request
from fastapi.responses import RedirectResponse
from fastapi.security import APIKeyCookie
from fastapi_sso.sso.google import GoogleSSO
from fastapi_sso.sso.base import OpenID
from jose import jwt
SECRET_KEY = "this-is-very-secret"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
sso = GoogleSSO(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, redirect_uri="http://127.0.0.1:5000/auth/callback")
app = FastAPI()
async def get_logged_user(cookie: str = Security(APIKeyCookie(name="token"))) -> OpenID:
"""Get user's JWT stored in cookie 'token', parse it and return the user's OpenID."""
try:
claims = jwt.decode(cookie, key=SECRET_KEY, algorithms=["HS256"])
return OpenID(**claims["pld"])
except Exception as error:
raise HTTPException(status_code=401, detail="Invalid authentication credentials") from error
@app.get("/protected")
async def protected_endpoint(user: OpenID = Depends(get_logged_user)):
return {"message": f"You are very welcome, {user.email}!"}
@app.get("/auth/login")
async def login():
async with sso:
return await sso.get_login_redirect()
@app.get("/auth/logout")
async def logout():
response = RedirectResponse(url="/protected")
response.delete_cookie(key="token")
return response
@app.get("/auth/callback")
async def login_callback(request: Request):
async with sso:
openid = await sso.verify_and_process(request)
if not openid:
raise HTTPException(status_code=401, detail="Authentication failed")
expiration = datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(days=1)
token = jwt.encode({"pld": openid.dict(), "exp": expiration, "sub": openid.id}, key=SECRET_KEY, algorithm="HS256")
response = RedirectResponse(url="/protected")
response.set_cookie(key="token", value=token, expires=expiration)
return response
