Different OAuth2 flows per route ?

[Abdelgha-4]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Description

The OAuth2 scopes section in FastAPI Docs clearly explains how it is possible to use multiple SecurityScopes per routes as necessary. However, for my use-case I need to define different OAuth2 flows as well, basically something like this:

oauth2_scheme = OAuth2(
    flows=OAuthFlows(
        implicit=OAuthFlowImplicit(
            authorizationUrl="",
            scopes={"scope0": "Common scope", "scope1": "Implicit only scope"},
        ),
        clientCredentials=OAuthFlowClientCredentials(
            tokenUrl="",
            scopes={"scope0": "Common scope", "scope2": "ClientCredentials only scope"},
        ),
    )
)

async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    ...

async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["scope1"], flows=["implicit"])],
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

Security Does not support a way to specify authorized flows. The scenario above for instance would enable only authorizing "implicit" users with the scope "scope1", and the general feature I'm looking for is specifying any combination of scopes and flows per route.

I thinks that one way of achieving this is building oauth2_scheme for each one of the desired combinations but that defeats the intended flexibly offered by Security and SecurityScopes so I hope there is a better solution.

Operating System

Windows

FastAPI Version

0.111.0

Pydantic Version

2.7.1

Python Version

3.11

[YuriiMotov]

Hi!
I think that there is no easy way (standard way) to do what you want.
I have one idea, but I need time to try it out. I will try to do it this weekend if nobody helps you

[YuriiMotov]

This is an explanation of my idea. I didn't manage to check it. So, it's just pseudo-code (not a working example):

from typing import Annotated

from fastapi import Depends, HTTPException, Security, status
from fastapi.openapi.models import OAuthFlowClientCredentials, OAuthFlowImplicit
from fastapi.security import OAuth2, OAuthFlows, SecurityScopes
from pydantic import BaseModel

IMPLICIT_FLOW_SCOPES = {
    "scope0": "Common scope",
    "implicit-scope1": "Implicit only scope",
}
CLIENT_CREDENTIALS_FLOW_SCOPES = {
    "scope0": "Common scope",
    "clientcredentials-scope2": "ClientCredentials only scope",
}

oauth2_scheme = OAuth2(
    flows=OAuthFlows(
        implicit=OAuthFlowImplicit(
            authorizationUrl="",
            scopes=IMPLICIT_FLOW_SCOPES,
        ),
        clientCredentials=OAuthFlowClientCredentials(
            tokenUrl="",
            scopes=CLIENT_CREDENTIALS_FLOW_SCOPES,
        ),
    )
)

class User(BaseModel):
    name: str

def decode_token(token: str):
    pass

async def get_user_from_db(user_id: str):
    pass

async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    implicit_scopes_required = [
        scope for scope in security_scopes if scope in IMPLICIT_FLOW_SCOPES
    ]
    client_credentials_scopes_required = [
        scope for scope in security_scopes if scope in CLIENT_CREDENTIALS_FLOW_SCOPES
    ]

    token = decode_token(token)

    if (not implicit_scopes_required) and (not client_credentials_scopes_required):
        return await get_user_from_db(token.sub)

    if implicit_scopes_required and set(implicit_scopes_required).issubset(
        token.scopes
    ):
        return await get_user_from_db(token.sub)

    if client_credentials_scopes_required and set(
        client_credentials_scopes_required
    ).issubset(token.scopes):
        return await get_user_from_db(token.sub)

    if implicit_scopes_required:
        implicit_scopes_required_str = (
            f"scopes = \"{' '.join(implicit_scopes_required)}\""
        )
    else:
        implicit_scopes_required_str = ""

    if client_credentials_scopes_required:
        client_credentials_scopes_required_str = (
            f"scopes = \"{' '.join(client_credentials_scopes_required)}\""
        )
    else:
        client_credentials_scopes_required_str = ""

    # Not sure about correctness of this,
    # but https://lists.w3.org/Archives/Public/ietf-http-wg/2012JanMar/0411.html
    www_authenticate_header = (
        f"Bearer {implicit_scopes_required_str}"
        f", , Bearer {client_credentials_scopes_required_str}"
    )

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Not enough rights",
        headers={"WWW-Authenticate": www_authenticate_header},
    )

async def get_current_active_user(
    current_user: Annotated[
        User, Security(get_current_user, scopes=["implicit-scope1"])
    ],
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

The idea is:

1. include flow name into scope name (e.g. implicit-scope1)

2. before checking scopes, separate scopes of different flows (implicit_scopes_required and client_credentials_scopes_required)

3. user is authorized for operation if token's scopes match the required scopes of at least one of flows

Hope you will get the idea!