Seeking Feedback: FastAPI with Kratos Authentication + Keto Authorisation

[0xliam]

Hi there,

I am looking for feedback on a FastAPI integration I've written to secure API endpoints and authenticate against Kratos.

My FastAPI application currently just consists of a few CRUD functions, with no authentication.

Another application that runs on customer premises needs to interact with this FastAPI service, and I wanted to add a layer of authentication and authorisation.

I plan on using Keto for authorisation, to ensure users of a customer/tenant can only run CRUD functions on their own objects, and will likely adapt this guide for Flask to FastAPI's API.

Hopefully that makes sense!

Here is my authorisation code - feedback is much appreciated - I haven't implemented any authorisation yet - just checking that a user has a valid session token and then using that session to allow access to a resource.

I've implemented two endpoints:

• token: initialises and submits a flow using username + password, and returns the session token.
• secret: uses the session token to unlock a secret message

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi import APIRouter, HTTPException, Depends
import ory_kratos_client
from ory_kratos_client.api import v0alpha2_api
from ory_kratos_client.model.submit_self_service_login_flow_body import SubmitSelfServiceLoginFlowBody
from ory_kratos_client.exceptions import UnauthorizedException

KRATOS_API_READ_URL = "http://localhost:4433"

configuration = ory_kratos_client.Configuration(host=KRATOS_API_READ_URL)

router = APIRouter(prefix="/auth", tags=["auth"])

kratos = ory_kratos_client.ApiClient(configuration)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_current_active_user(token: str = Depends(oauth2_scheme)):
    with kratos as api_client:
        api_instance = v0alpha2_api.V0alpha2Api(api_client)
        try:
            session = api_instance.to_session(x_session_token=token)
        except UnauthorizedException as e:
            raise HTTPException(status_code=401)
    if not session.active:
        raise HTTPException(status_code=401)
    return session

@router.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    with kratos as api_client:
        api_instance = v0alpha2_api.V0alpha2Api(api_client)
        refresh = True
        try:
            flow = (api_instance.initialize_self_service_login_flow_without_browser(async_req=True, refresh=refresh))
        except ory_kratos_client.ApiException as e:
            raise HTTPException(status_code=403, detail=e) 

        body = SubmitSelfServiceLoginFlowBody(identifier=form_data.username, password=form_data.password, method='password')

        try:
            api_response = api_instance.submit_self_service_login_flow(async_req=True, flow=flow.id, submit_self_service_login_flow_body=body)
            return {"access_token": api_response.session_token, "token_type": "bearer"}
        except ory_kratos_client.ApiException as e:
            raise HTTPException(status_code=403, detail=e) 

@router.get("/secret")
async def authenticated_route(session = Depends(get_current_active_user)):
    return {'message': 'secret'}

Feedback would be much appreciated.

[0xliam]

Update: Implemented a very basic Keto permissions check:

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi import APIRouter, HTTPException, Depends
import ory_kratos_client
from ory_kratos_client.api import v0alpha2_api
from ory_kratos_client.model.submit_self_service_login_flow_body import (
    SubmitSelfServiceLoginFlowBody,
)
from ory_kratos_client.exceptions import UnauthorizedException
import requests

KRATOS_API_READ_URL = "http://localhost:4433"
KETO_API_READ_URL = "http://localhost:4466"

KRATOS_CONFIGURATION = ory_kratos_client.Configuration(host=KRATOS_API_READ_URL)
KETO_CONFIGURATION = ory_kratos_client.Configuration(host=KETO_API_READ_URL)

router = APIRouter(prefix="/auth", tags=["auth"])
kratos = ory_kratos_client.ApiClient(KRATOS_CONFIGURATION)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


async def get_current_active_user(token: str = Depends(oauth2_scheme)):
    with kratos as api_client:
        api_instance = v0alpha2_api.V0alpha2Api(api_client)
        try:
            session = api_instance.to_session(x_session_token=token)
        except UnauthorizedException as e:
            raise HTTPException(status_code=401)
    if not session.active:
        raise HTTPException(status_code=401)
    return session


class KetoChecker:
    def __init__(self, namespace: str, object: str, relation):
        self.namespace = namespace
        self.object = object
        self.relation = relation

    def __call__(self, session=Depends(get_current_active_user)):
        # with httpx.AsyncClient() as client:
        data = {
            "namespace": self.namespace,
            "object": self.object,
            "relation": self.relation,
            "subject_id": session.identity.id,
        }
        headers = {"content-type": "application/json"}
        response = requests.post(
            f"{KETO_API_READ_URL}/relation-tuples/check",
            json=data,
            headers=headers,
        )
        if not response.json().get("allowed"):
            raise HTTPException(status_code=403)


@router.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    with kratos as api_client:
        api_instance = v0alpha2_api.V0alpha2Api(api_client)
        refresh = True
        try:
            flow = api_instance.initialize_self_service_login_flow_without_browser(
                refresh=refresh
            )
        except ory_kratos_client.ApiException as e:
            raise HTTPException(status_code=403, detail=e)

        body = SubmitSelfServiceLoginFlowBody(
            identifier=form_data.username,
            password=form_data.password,
            method="password",
        )

        try:
            api_response = api_instance.submit_self_service_login_flow(
                flow=flow.id, submit_self_service_login_flow_body=body
            )
            return {"access_token": api_response.session_token, "token_type": "bearer"}
        except ory_kratos_client.ApiException as e:
            raise HTTPException(status_code=403, detail=e)


secret_auth = KetoChecker(namespace="auth", object="secret", relation="read")


@router.get("/secret")
async def authenticated_route(auth: secret_auth = Depends()):
    return {"secret": "message"}

[meltingrock]

@0xliam I am doing research on integrating FastAPI and Orty. Your post is the first example I have found between the two. How are things going with your integration?

[0xliam]

@0xliam I am doing research on integrating FastAPI and Orty. Your post is the first example I have found between the two. How are things going with your integration?

Unfortunately, we never proceeded with the project, so I haven't touched it since this post. I didn't find any issues with the integration when testing, but it might need a review by a more seasoned developer to identify any security flaws with my example code.

[meltingrock]

Ok, cool, thanks for letting me know :-)