Validate tokens with Introspect Token Validation in FastAPI

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 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

        if key.startswith("base64:"):
            base64_key = key.split("base64:")[1]
            self.key = base64.b64decode(base64_key)
        else:
            self.key = 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
            )

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 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

        if key.startswith("base64:"):
            base64_key = key.split("base64:")[1]
            self.key = base64.b64decode(base64_key)
        else:
            self.key = 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

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