Suggest middlwares for oauth2 against external providers

[jorgecarleitao]

Is your feature request related to a problem

Is your feature request related to a problem? Yes

It is unclear how to integrate an external oauth provider such as Microsoft, Google, Auth0 with FastAPI.

The solution you would like

A section on the documentation describing how to achieve this, or which libraries do we recommend to do so.

Two examples include the client from authlib and starlette-oauth2-api.

Describe alternatives you've considered

PR an implementation to FastAPI. The reason I avoided this is because OAuth is not a thing of FastAPI only, but of a web app in general.

Additional context

I am the author of starlette-oauth2-api, which we have been using to secure an API of ours against access tokens signed by external providers (multi-tenancy).

[dmontagu]

I personally would welcome a docs PR, and I think the discussion at the end of #797 may be relevant. I think @tiangolo might already be cooking up something along these lines.

[GeorgeFischhof]

Hi All,

I would need it too ;-)
I found a good doc for flask here:
https://realpython.com/flask-google-login/

BR,
George

[meteoDaniel]

How to add google auth is explained in this article:
https://medium.com/data-rebels/fastapi-google-as-an-external-authentication-provider-3a527672cf33

[meteoDaniel]

Did not know if this helps but I think this code should run for auth0 + FastAPI (actually I am not able to test it, but i will do asap). I merged the suggestion by auth0 from this site https://auth0.com/docs/quickstart/backend/python/01-authorization#validate-scopes with my implementation from FastAPI that just checks a list of valid tokens.

First I define all authentication stuff in one python script. Please note that I am just enabled query authentication.

""" functions to connect token validation with auth0 """
import os
import json
from urllib.request import urlopen

from jose import jwt
from typing import Dict
from fastapi.security.api_key import APIKeyQuery
from fastapi import Security

AUTH0_DOMAIN = os.environ['AUTH0_DOMAIN']
API_AUDIENCE = os.environ['API_AUDIENCE']
ALGORITHMS = ["RS256"]

API_KEY_NAME = "access_token"
api_key_query = APIKeyQuery(name=API_KEY_NAME, auto_error=False)


class AuthError(Exception):
    """ Error handling object """
    def __init__(self, error: Dict[str, str], status_code: int):
        self.error = error
        self.status_code = status_code


def auth0_token_authentication(
    token: str = Security(api_key_query)
):
    """ authenticates token parsed via query with auth0 API """
    jsonurl = urlopen("https://" + AUTH0_DOMAIN + "/.well-known/jwks.json")
    jwks = json.loads(jsonurl.read())
    unverified_header = jwt.get_unverified_header(token)
    rsa_key = {}
    for key in jwks["keys"]:
        if key["kid"] == unverified_header["kid"]:
            rsa_key = {
                "kty": key["kty"],
                "kid": key["kid"],
                "use": key["use"],
                "n": key["n"],
                "e": key["e"]
            }
    if rsa_key:
        try:
            payload = jwt.decode(
                token,
                rsa_key,
                algorithms=ALGORITHMS,
                audience=API_AUDIENCE,
                issuer="https://" + AUTH0_DOMAIN + "/"
            )
        except jwt.ExpiredSignatureError:
            raise AuthError({"code": "token_expired",
                             "description": "token is expired"}, 401)
        except jwt.JWTClaimsError:
            raise AuthError({"code": "invalid_claims",
                             "description":
                                 "incorrect claims,"
                                 "please check the audience and issuer"}, 401)
        except Exception:
            raise AuthError({"code": "invalid_header",
                             "description":
                                 "Unable to parse authentication"
                                 " token."}, 401)


        return payload
    raise AuthError({"code": "invalid_header",
                     "description": "Unable to find appropriate key"}, 401)

In your app you have to declare an exception handler and add the well known Depends hint with the authentication function :

@app.exception_handler(AuthError)
async def unicorn_exception_handler(request: Request, exc: AuthError):
    return JSONResponse(
        status_code=exc.status_code,
        content=exc.error,
    ) 

@app.get("/documentation", tags=["documentation"])
async def get_documentation(api_key: APIKey = Depends(auth0_token_authentication)):
    response = get_swagger_ui_html(openapi_url="/openapi.json", title="docs")
    response.set_cookie(
        API_KEY_NAME,
        value=api_key,
        domain=COOKIE_DOMAIN,
        httponly=True,
        max_age=1800,
        expires=1800,
    )
    return response

I hope this can help you.

[jfmrm]

@meteoDaniel Thanks for your post! It helped me a lot. And I actually iterated through it a little, making a dependency injection version, and gone a little further. Implemented a couple of automated tests on it. Unfortunately I have not found a way of testing it without hitting Auth0 API. If anyone would have some solution for that I would be pleased.

Here follows my code:

import json
from urllib.request import urlopen
from jose import jwt
from typing import Dict
from fastapi import Security
from fastapi.security import OAuth2AuthorizationCodeBearer
from typing import List


class AuthError(Exception):
    """ Error handling object """
    def __init__(self, error: Dict[str, str], status_code: int):
        self.error = error
        self.status_code = status_code


def get_jwks(auth0_domain: str):
    json_url = urlopen(f"https://{auth0_domain}/.well-known/jwks.json")
    return json.loads(json_url.read())


def auth0_token_authenticator_builder(
        oauth2_scheme: OAuth2AuthorizationCodeBearer,
        auth0_domain: str,
        api_audience: str,
        algorithms: List[str],
        jwks: dict
):
    def auth0_token_authentication(
        token: str = Security(oauth2_scheme)
    ):
        unverified_header = jwt.get_unverified_header(token)
        rsa_key = {}
        for key in jwks["keys"]:
            if key["kid"] == unverified_header["kid"]:
                rsa_key = {
                    "kty": key["kty"],
                    "kid": key["kid"],
                    "use": key["use"],
                    "n": key["n"],
                    "e": key["e"]
                }
        if rsa_key:
            try:
                payload = jwt.decode(
                    token,
                    rsa_key,
                    algorithms=algorithms,
                    audience=api_audience,
                    issuer=f"https://{auth0_domain}/"
                )
            except jwt.ExpiredSignatureError:
                raise AuthError({"code": "token_expired",
                                 "description": "token is expired"}, 401)
            except jwt.JWTClaimsError:
                raise AuthError({"code": "invalid_claims",
                                 "description":
                                     "incorrect claims,"
                                     "please check the audience and issuer"}, 401)
            except Exception:
                raise AuthError({"code": "invalid_header",
                                 "description":
                                     "Unable to parse authentication"
                                     " token."}, 401)

            return payload
        raise AuthError({"code": "invalid_header",
                         "description": "Unable to find appropriate key"}, 401)

    return auth0_token_authentication


def scope_verifier_builder(oauth2_scheme: OAuth2AuthorizationCodeBearer, required_scope):
    def scope_verifier(token: str = Security(oauth2_scheme)):
        unverified_claims = jwt.get_unverified_claims(token)
        if unverified_claims.get("scope"):
            token_scopes = unverified_claims["scope"].split()
            for token_scope in token_scopes:
                if token_scope == required_scope:
                    return True
        raise AuthError({
            "code": "no_required_permissions",
            "description": "user does not have the required permissions"
        }, 403)

    return scope_verifier

here follows the tests:

import unittest
from fastapi import FastAPI, Depends, status
from fastapi.testclient import TestClient
from fastapi.security import OAuth2AuthorizationCodeBearer
from utils import authentication


class TestAuthentication(unittest.TestCase):
    def setUp(self) -> None:
        # mocked env variables
        AUTH0_DOMAIN = "your_auth0_domain"
        API_AUDIENCE = "your_audience_endpoint"
        ALGORITHMS = ["RS256"]

        jwks = authentication.get_jwks(AUTH0_DOMAIN)

        # mocked auth scheme
        oauth2_scheme = OAuth2AuthorizationCodeBearer(
            authorizationUrl=AUTH0_DOMAIN,
            tokenUrl=API_AUDIENCE
        )

        # build auth method
        auth = authentication.auth0_token_authenticator_builder(
            oauth2_scheme=oauth2_scheme,
            auth0_domain=AUTH0_DOMAIN,
            api_audience=API_AUDIENCE,
            algorithms=ALGORITHMS,
            jwks=jwks
        )

        # build scope verifier
        scope_verifier = authentication.scope_verifier_builder(
            oauth2_scheme=oauth2_scheme,
            required_scope='read:all'
        )

        # build mocked protected API
        api = FastAPI()

        @api.get("/test")
        def test_endpoint(api_key=Depends(auth)):
            return api_key

        @api.get("/test/scope")
        def test_scope(scope: bool = Depends(scope_verifier)):
            return scope

        self.client = TestClient(api)

        # mocked token
        self.token = "generate_a_token_and_put_it_here:)"

    def test_authenticate(self):
        response = self.client.get(
            "/test",
            headers={"authorization": f"Bearer {self.token}"}
        )
        self.assertEqual(status.HTTP_200_OK, response.status_code)

    def test_has_required_scope(self):
        response = self.client.get(
            "/test/scope",
            headers={"authorization": f"Bearer {self.token}"}
        )
        self.assertEqual(status.HTTP_200_OK, response.status_code)


if __name__ == '__main__':
    unittest.main()

[Vivalldi]

@jfmrm Obrigado pelo exemplo! I reached a very similar solution that doesn't require the full "builder" aspect.

auth0.py

import json

from fastapi import Depends
from fastapi.security import OAuth2AuthorizationCodeBearer, SecurityScopes
from jose import jwt
from six.moves.urllib.request import urlopen

AUTH0_DOMAIN = "<auth0-domain>"
API_AUDIENCE = "<auth0-api-audience>"
ALGORITHMS = ["RS256"]

class AuthError(Exception):
    def __init__(self, error, status_code):
        self.error = error
        self.status_code = status_code

# Define a Authorization scheme specific to our Auth0 config
auth0_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl=AUTH0_DOMAIN, tokenUrl=API_AUDIENCE
)

async def get_current_user(security_scopes: SecurityScopes, token: str = Depends(auth0_scheme)):
    # This down to `END` comment is from https://auth0.com/docs/quickstart/backend/python/01-authorization#create-the-jwt-validation-decorator
    jsonurl = urlopen("https://" + AUTH0_DOMAIN + "/.well-known/jwks.json")
    jwks = json.loads(jsonurl.read())
    unverified_header = jwt.get_unverified_header(token)

    rsa_key = {}
    for key in jwks["keys"]:
        if key["kid"] == unverified_header["kid"]:
            rsa_key = {
                "kty": key["kty"],
                "kid": key["kid"],
                "use": key["use"],
                "n": key["n"],
                "e": key["e"],
            }

    if rsa_key:
        try:
            payload = jwt.decode(
                token,
                rsa_key,
                algorithms=ALGORITHMS,
                audience=API_AUDIENCE,
                issuer=f"https://{AUTH0_DOMAIN}/",
            )
        except jwt.ExpiredSignatureError:
            raise AuthError(
                {"code": "token_expired", "description": "token is expired"}, 401
            )
        except jwt.JWTClaimsError:
            raise AuthError(
                {
                    "code": "invalid_claims",
                    "description": "incorrect claims,"
                    "please check the audience and issuer",
                },
                401,
            )
        except Exception:
            raise AuthError(
                {
                    "code": "invalid_header",
                    "description": "Unable to parse authentication" " token.",
                },
                401,
            )
       # END from Auth0

        # token.scope is represented as a string of scopes space seperated
        token_scopes = payload.get("scope", "").split()

        # Check that we all scopes are present
        for scope in security_scopes.scopes:
            if scope not in token_scopes:
                raise AuthError(
                    {
                        "code": "Unauthorized",
                        "description": f"You don't have access to this resource. `{' '.join(security_scopes.scopes)}` scopes required",
                    },
                    403,
                )

        return payload

Using this is now as simple as
app.py

from fastapi import FastAPI, Security

from .auth0 import AuthError, get_current_user

app = FastAPI()

@app.exception_handler(AuthError)
def handle_auth_error(request: Request, ex: AuthError):
    return JSONResponse(status_code=ex.status_code, content=ex.error)

@app.get("/private")
def private(user=Security(get_current_user)):
    return {"message": "You're an authorized user"}

@app.get("/private-with-scopes")
def privateScopes(user=Security(get_current_user, scopes["read:example"])):
    return {"message": "You're authorized with scopes!"}

[btotharye]

@Vivalldi do you have this code anywhere on github or somewhere? I am trying to do something similar but I'm hitting issues on this code.

Thanks

[btotharye]

Actually @Vivalldi I got your example working, my only question now is how do you auth on the /docs route to have it send you to Auth0 to login and back or is that even possible? Was just curious how users would use the api via the docs.

[Vivalldi]

The OpenAPI spec doesn't include JWT bearer tokens yet. There's no defined standard as to where those tokens live and as such there's no OpenAPI config for it yet. I might be able to contrive an example where you set the token in the username field of basic auth (without pass) similar to stripe https://stripe.com/docs/api/authentication

[btotharye]

@Vivalldi how are you handling your user relations this way? I'm guessing the sub data from the user object is basically the userid so would I just use this when I'm creating records that would have an ownerId for example or how have you been doing that?

[Vivalldi]

This isn't directly related to this issue but in short, yes, you should use the user identifier.

https://auth0.com/docs/users/normalized/auth0/store-user-data

[rsitro4]

Here is a working Auth0 and FastAPI code snippet.
main.py

from fastapi import Depends, FastAPI
from routers import items
from utils.auth import require_auth

app = FastAPI()

app.include_router(
    items.router,
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(require_auth)],
    responses={404: {"description": "Not found"}},
)

utils.auth.py

import json
import logging
import os
from fastapi import HTTPException, Header
from six.moves.urllib.request import urlopen
from jose import jwt
from jose import exceptions as JoseExceptions


logging.basicConfig(
    format='%(asctime)s %(message)s', 
    datefmt='%m/%d/%Y %I:%M:%S %p',
    level=logging.os.environ.get('LOGGING_LEVEL', 'INFO'))

AUTH0_DOMAIN = os.environ.get(
    'AUTH0_DOMAIN', '')

AUTH0_API_AUDIENCE = os.environ.get(
    'AUTH0_API_AUDIENCE', '')

AUTH0_ALGORITHMS = os.environ.get(
    'AUTH0_ALGORITHMS', 'RS256')


def get_token_auth_header(authorization):
    parts = authorization.split()

    if parts[0].lower() != "bearer":
        raise HTTPException(
            status_code=401, 
            detail='Authorization header must start with Bearer')
    elif len(parts) == 1:
        raise HTTPException(
            status_code=401, 
            detail='Authorization token not found')
    elif len(parts) > 2:
        raise HTTPException(
            status_code=401, 
            detail='Authorization header be Bearer token')
    
    token = parts[1]

    return token


def get_payload(jwks, unverified_header, token):
    rsa_key = {}
    payload = None
    for key in jwks["keys"]:
        if key["kid"] == unverified_header["kid"]:
            rsa_key = {
                "kty": key["kty"],
                "kid": key["kid"],
                "use": key["use"],
                "n": key["n"],
                "e": key["e"]
            }
        if rsa_key:
            try:
                payload = jwt.decode(
                    token,
                    rsa_key,
                    algorithms=AUTH0_ALGORITHMS,
                    audience=AUTH0_API_AUDIENCE,
                    issuer="https://"+AUTH0_DOMAIN+"/"
                )
            except jwt.ExpiredSignatureError:
                raise HTTPException(
                    status_code=401, 
                    detail='Authorization token expired')
            except jwt.JWTClaimsError:
                raise HTTPException(
                    status_code=401, 
                    detail='Incorrect claims, check the audience and issuer.')
            except Exception as e:
                logging.warning(e, exc_info=True)
                raise HTTPException(
                    status_code=401, 
                    detail='Unable to parse authentication token')

    return payload


async def require_auth(authorization: str = Header(...)):
    token = get_token_auth_header(authorization)
    jsonurl = urlopen("https://"+AUTH0_DOMAIN+"/.well-known/jwks.json")
    jwks = json.loads(jsonurl.read())

    try:
        unverified_header = jwt.get_unverified_header(token)
    except JoseExceptions.JWTError:
        raise HTTPException(
                    status_code=401, 
                    detail='Unable to decode authorization token headers')
    
    payload = get_payload(jwks, unverified_header, token)
    
    if not payload:
        raise HTTPException(
                    status_code=401, 
                    detail='Invalid authorization token')

    return payload

routers.items.py

from fastapi import APIRouter, HTTPException

router = APIRouter()


@router.get("/")
async def read_items():
    return [{"name": "Item Foo"}, {"name": "item Bar"}]


@router.get("/{item_id}")
async def read_item(item_id: str):
    return {"name": "Fake Specific Item", "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "foo":
        raise HTTPException(status_code=403, detail="You can only update the item: foo")
    return {"item_id": item_id, "name": "The Fighters"}

[LindezaGrey]

@rsitro4 Nice example. How do you use swagger-ui (/docs) with this?

[LuisHernandez1611681]

@LindezaGrey did you know how to use it with swagger-ui (/docs)?

[LindezaGrey]

@LuisHernandez1611681 well sort of. In auth0 i added another authorization flow and internally used OAuth2PasswordBearer for the docs page. This means that you need to provide also client_secret and client_id. In my use case only admins use the /docs page for data entry and retrieval so it's not a big deal. if the docs page is faces to customers/clients, of course you don't want to provide those credentials...

[ycd]

@dorinclisu you might want publish it to PyPI with a package manager like poetry, flit etc

[dorinclisu]

I'm looking into it, meanwhile it can be just as easily installed with pip from github archive. Locking against a specific release can be done like this pip install https://github.com/dorinclisu/fastapi-auth0/archive/0.1.0.zip

[inspiralpatterns]

user=Security(get_current_user, scopes["read:example"]

@Vivalldi do you think there might be a way to have this as dependency injection in the router?
I am thinking something like this:

@router.post(
    "/predict",
    callbacks=callback_router.routes,
    response_model=models.Margin,
    dependencies=[Depends(Security(required_auth, scopes["role:scope"])]
)
def predict(predict_request: models.PredictRequest):
    # whatever the function does

although it does not work as Depends expects a callable.

[dorinclisu]

@inspiralpatterns
It should work by changing to this dependencies=[Security(required_auth, scopes["role:scope"])].
Security is just a subclass of Depends so any place that expects a Depends can also take Security.

[dorinclisu]

Anyway, here's a minimal example using fastapi_auth0:

from fastapi import FastAPI
from fastapi_auth0 import Auth0, Auth0User

app = FastAPI()
auth = Auth0(domain='your-tenant.auth0.com', api_audience='your-api-identifier', scopes={'read:blabla': ''})

@app.get("/secure", dependencies=[Depends(auth.implicit_scheme)])
def get_secure(user: Auth0User = Security(auth.get_user, scopes=['read:blabla'])):
    return {"message": f"{user}"}

Bear in mind that auth.implicit_scheme (along with auth.password_scheme and auth.oidc_scheme) as a dependency doesn't secure anything, it just provides the swagger documentation.

If you want to secure the path (or whole router) and are not interested in user details, you could do this:

@app.get("/secure", dependencies=[
    Depends(auth.implicit_scheme),
    Security(auth.get_user, scopes=['read:blabla'])
])
def get_secure():
    return {"message": "I don't care who you are, but you're authorized"}

[dorinclisu]

@dorinclisu you might want publish it to PyPI with a package manager like poetry, flit etc

https://pypi.org/project/fastapi-auth0/

[JHBalaji]

oauth2 api seems to be neat
https://pypi.org/project/starlette-oauth2-api/

I have a question @jorgecarleitao, instead of hitting the jwks endpoint for every singe request through middleware is it possible to set a refresh interval with a cache.

Also here is another one for AWS.

[dorinclisu]

I have a question @jorgecarleitao, instead of hitting the jwks endpoint for every singe request through middleware is it possible to set a refresh interval with a cache.

It's not needed to use a cache, you only need to read it once at api startup. The jwks never changes except when you rotate the tenant secret, which means it's not big deal to restart the api also.

[jorgecarleitao]

@JHBalaji, thanks :D

I have a question @jorgecarleitao, instead of hitting the jwks endpoint for every singe request through middleware is it possible to set a refresh interval with a cache.

Currently the midlleware fetches then once when they are first required and cached in-memory. You can inject the keys yourself if your API for some reason does not have egress (e.g. corp stuff).

There is a valid use-case for IdP that rotate keys; I've added an issue for that use-case.

[JHBalaji]

@jorgecarleitao thanks for talking it. Yes my IdP rotate every 30 min as well as JWKS url can also rotate which is fetched from another URL (/openid-configuration) with key being jwks_uri.

[jorgecarleitao]

@JHBalaji , the refreshing of the key is now available, on version 0.2.5.

On a separate note: I have never heard of a refreshing key URL. Do you know which concern the rotating URL is addressing? Which OAuth2 provider have you seen that behavior on?

[JHBalaji]

@jorgecarleitao I dont use a public Idp provider but I have added another hook to address this in mine. But the flow is retrieve openid-configuration and read jwks_uri and call jwks_uri to get refresh tokens every X mins and visit openid-configuration and get updated jwks_uri. The jwks_uri changes in our dev env.
Thanks for the release.

[tiangolo]

Hey all! Thanks for the discussion here! ☕

Having more docs about auth and integration with OAuth2 providers, and how to create OAuth2 servers is one of the things I've been wanting to do for a long time. But it's a relatively complex topic that needs a long time of dedication to write the whole tutorial.

And yes, the way to go would be quite similar to @Vivalldi's example: #840 (comment)

That one is specific to Auth0, with their own AUTH0_DOMAIN. I want to make a tutorial that doesn't specifically depend on/promote one of these external providers that are not involved with FastAPI. I would hope they would make tutorials to integrate their auth with FastAPI, but anyway. 🤷

An initial way to go is, if you are using some third party provider and are integrating it with FastAPI, you could write a blog post about it, and you could add it to the external links section in the docs. As this is useful to others using your same provider, I'm pretty sure it would be very well received.

Now, about the original issue posted, I wouldn't recommend using middlewares for this, but dependencies. Because dependencies are what can be integrated with FastAPI, with OpenAPI, the automatic docs, respect types, etc. So I would try to move away from middlewares. They also have several issues that have been improving in Starlette, but still, it's not the best way to go with FastAPI.

But anyway, although I can't make promises about timelines (I'm now trying to at least clean issues first) at some point OAuth2 will be in the docs and in the tutorials for sure. 🤓 📝

Thanks for the discussion and interest everyone! 🍰

Sorry for the long delay! 🙈 I wanted to personally address each issue and they piled up through time, but now I'm checking each one in order.

[github-actions[bot]]

Assuming the original need was handled, this will be automatically closed now. But feel free to add more comments or create new issues or PRs.

[yordis]

For those who may be interested in Zitadel, here a version, please share your thoughts:

import time
import json
import requests
from typing import Optional
from fastapi.security import OAuth2AuthorizationCodeBearer, SecurityScopes
from fastapi.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED
from authlib.jose import jwt
import base64

def decode_value(value: str):
    if value.startswith("base64:"):
        value = value.split("base64:")[1]
        return base64.b64decode(value)
    return value

def match_token_scopes(token, or_scopes):
    if or_scopes is None:
        return True
    scopes = token.get("scope", "").split()
    for and_scopes in or_scopes.values():
        if all(key in scopes for key in and_scopes.split()):
            return True
    return False

class ApiPrivateKey:
    def __init__(
        self,
        client_id: str,
        key_id: str,
        key: str,
        aud: str,
        exp_in=60 * 60,
        alg="RS256",
        type="application",
        app_id=None
    ):
        self.client_id = client_id
        self.key_id = key_id
        self.aud = aud
        self.app_id = app_id
        self.type = type

        self.exp_in = exp_in
        self.alg = alg
        self.key = decode_value(key)

    @staticmethod
    def from_file(file_path: str, aud: str):
        with open(file_path, "r") as f:
            data = json.load(f)
            ApiPrivateKey(
                type=data["type"],
                key_id=data["keyId"],
                key=data["key"],
                app_id=data["appId"],
                client_id=data["clientId"],
                aud=aud
            )

class ZitadelIntrospectToken(OAuth2AuthorizationCodeBearer):
    def __init__(
        self,
        base_url: str,
        private_key: ApiPrivateKey,
        scopes: dict[str, str] = None,
    ):
        super().__init__(
            authorizationUrl=f"#{base_url}/oauth/v2/authorize",
            tokenUrl=f"#{base_url}/oauth/v2/token",
            refreshUrl=f"#{base_url}/oauth/v2/token",
            scopes=scopes,
            auto_error=True
        )
        self.base_url = base_url
        self.private_key = private_key
        self.scopes = scopes

    async def __call__(self, request: Request) -> Optional[str]:
        token = await super().__call__(request)
        if not token:
            return None

        now = time.time()
        resp = requests.post(
            url=f'{self.base_url}/oauth/v2/introspect',
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            data={
                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
                "client_assertion": jwt.encode(
                    header={"alg": self.private_key.alg, "kid": self.private_key.key_id},
                    payload={
                        "iss": self.private_key.client_id,
                        "sub": self.private_key.client_id,
                        "aud": self.private_key.aud,
                        "exp": int(now) + self.private_key.exp_in,
                        "iat": int(now),
                    },
                    key=self.private_key.key,
                ),
                "token": token,
            }
        )

        return self.process_response(resp)

    def process_response(self, resp):
        if resp.status_code != 200:
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail="Not authenticated",
                headers={"WWW-Authenticate": "Bearer"},
            )

        introspected_token = resp.json()

        if not introspected_token:
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail="Token was revoked.",
                headers={"WWW-Authenticate": "Bearer"},
            )
        if not introspected_token.get("active"):
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail="Token is inactive.",
                headers={"WWW-Authenticate": "Bearer"},
            )

        now = int(time.time())
        if introspected_token["exp"] < now:
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail="Token has expired.",
                headers={"WWW-Authenticate": "Bearer"},
            )

        if not match_token_scopes(introspected_token, self.scopes):
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail=f"Token has insufficient scope. Route requires: {self.scopes.values()}",
                headers={"WWW-Authenticate": "Bearer"},
            )

        return introspected_token

[davidhuser]

hey @yordis thanks again, can you post your complete example on how you use it with FastAPI (and Swagger) and setup steps in Zitadel? One bit of feedback would be to use httpx.AsyncClient() instead of sync requests, and maybe use pydantic.BaseModel for the ApiPrivateKey.

[GeorgeFischhof]

Hi @davidhuser @yordis few months ago I checked available HTTP libraries, and found that it seems niquests is better (has more features, and active development) than httpx. Right now did a search on the net, and some says niquests is far beyond httpx. Even if niquests is not so popular right now.

BR,
George

[GeorgeFischhof]

Hi @davidhuser @yordis few months ago I checked available HTTP libraries, and found that it seems niquests is better (has more features, and active development) than httpx. Right now did a search on the net, and some says niquests is far beyond httpx. Even if niquests is not so popular right now.

BR,
George

[yordis]

@davidhuser, keep in mind I am not a professional python-dev, so I am not familiar with the ecosystem; I am hoping that you make your package as good as it gets

introspect_token = ZitadelIntrospectToken(
    base_url=os.getenv("ZITADEL_BASE_URL"),
    private_key=ApiPrivateKey(
        client_id=os.getenv("ZITADEL_CLIENT_ID"),
        key_id=os.getenv("ZITADEL_KEY_ID"),
        key=os.getenv("ZITADEL_PRIVATE_KEY"),
        aud=os.getenv("ZITADEL_AUD"),
    ),
    scopes={
        "openid": "openid",
        "email": "email",
        "profile": "profile",
    },
)

async def current_principal(introspected_token: Annotated[dict, Depends(introspect_token)]):
    zitadel_id = introspected_token.get("sub")

    if not introspected_token.get("email_verified"):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Email not verified")

    user = find_or_create_user(zitadel_id=zitadel_id)
    return Principal(identity=user)

app = FastAPI(
    swagger_ui_parameters={"deepLinking": False},
    swagger_ui_init_oauth={
        "clientId": os.getenv("ZITADEL_DOCS_CLIENT_ID"),
        "usePkceWithAuthorizationCodeGrant": True,
        "scopeSeparator": " ",
        "scopes": "openid profile email",
    }
)

@app.get("/me")
def get_me(principal: Principal = Depends(current_principal)) -> User:
    return User(id=principal.id)