How to connect FastAPI, Jinja2 and Keycloak?

Question

I work for a company and I have a Python FastAPI project. This is something like a multi-page website, where each endpoint is a page of the site. This was done using Jinja2 - TemplateResponse(). I know that this is not the best solution for similar projects, such as Flask or Django, but there is no way to change it.

I need to hide content using Keycloak authentication. I took the solution here: https://stackoverflow.com/a/77186511/21439459

Made same endpoint:

@app.get("/secure")
async def root(user: User = Depends(get_user_info)):
    return {"message": f"Hello {user.username} you have the following service: {user.realm_roles}"}

When I run the page I get (Not authenticated):

I don't understand how a user can authenticate. I expected a redirect to the Keycloak page like in other frameworks. Interesting point when running /docs

Here, after entering the data, a correct redirect to Keycloak occurs and after entering the login/password, I can receive a response from the secure endpoint.

I need help with a redirect when opening my website. I don't understand how to repeat authentication from the documentation.

I tried the fastapi_keycloak library, but as I understand it, it does not work with the company's version of keycloak.

I tried fastapi-keycloak-middleware, but I also received a 401 error and did not understand how to authenticate users.

Answer 1

Finally, my solution:

import jwt
import time
import json
import uvicorn
import requests
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.templating import Jinja2Templates
from fastapi.security.utils import get_authorization_scheme_param


AUTH_URL = "... KeyCloak Auth URL..." # with "...&redirect_uri=http://127.0.0.1:8080/auth"
TOKEN_URL = "...KeyCloak Token URL..."

templates = Jinja2Templates(directory="templates") # paste your directory

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)


@app.get("/secure_method")
def func(request: Request):
    authorization: str = request.cookies.get("Authorization")
    if not authorization:
        # You can add "&state=secure_method" to AUTH_URL for correct redirection after authentication
        return RedirectResponse(url=AUTH_URL)
    scheme, credentials = get_authorization_scheme_param(authorization)
    try:
        decoded = jwt.decode(jwt=credentials, options={'verify_signature': False})   
    except Exception as e
        return 0 # I send special error template
    # check expiration time
    if decoded['exp'] + 3600*24 < datetime.utcnow().timestamp():
        return RedirectResponse(url=AUTH_URL)

    # generate data or make something
  
    return templates.TemplateResponse('secure.html', {"request": request})
        

app.get("/auth")
def auth(code: str, state: str = "") -> RedirectResponse:
    payload = {
        'grant_type': 'authorization_code',
        'client_id': '...some client ID...',
        'code': code,
        'redirect_uri': 'http://127.0.0.1:8080/auth'
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    token_response = requests.request("POST", TOKEN_URL, data=payload, headers=headers)
    token_body = json.loads(token_response.content)
    access_token = token_body.get("access_token")
    if not access_token:
        return {"ERROR": "access_token"}
    response = RedirectResponse(url="/" + state)
    response.set_cookie("Authorization", value=f"Bearer {access_token}")
    return response


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8080)