Skip to content

Auth Validator

AuthValidator 

AuthValidator()

Dependency that validates the access token and returns the AuthInfo.

Source code in bitonicai/internal/authentication/auth_validator.py 
20
21
22
23
def __init__(self):
    self.logto_config = logto_config
    self.well_known_url = f"{self.logto_config.uri}/oidc/.well-known/openid-configuration"
    self.jwks_client = PyJWKClient(f"{self.logto_config.uri}/oidc/jwks")

Functions

get_well_known_config async 
get_well_known_config()

Gets the well-known configuration from the Logto server.

Source code in bitonicai/internal/authentication/auth_validator.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def get_well_known_config(self) -> dict:
    """
    Gets the well-known configuration from the Logto server.
    """

    try:
        async with httpx.AsyncClient(verify=False) as client:
            response = await client.get(
                self.well_known_url, headers={"Accept": "application/json"}
            )
            if response.status_code != 200:
                raise HTTPException(
                    status_code=response.status_code,
                    detail=f"HTTP error: {response.status_code} {response.text}",
                )
            return response.json()
    except httpx.HTTPStatusError as e:
        raise HTTPException(
            status_code=e.response.status_code,
            detail=f"HTTP error: {e.response.status_code} {e.response.text}",
        ) from e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") from e
introspect_token async 
introspect_token(token)

Introspects the token and returns the payload.

Source code in bitonicai/internal/authentication/auth_validator.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def introspect_token(self, token: str) -> dict:
    """
    Introspects the token and returns the payload.
    """

    try:
        well_known_config = await self.get_well_known_config()
        introspection_url = well_known_config["introspection_endpoint"]
        async with httpx.AsyncClient(verify=False) as client:
            response = await client.post(
                introspection_url,
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                data={
                    "token": token,
                    "client_id": self.logto_config.client_id,
                    "client_secret": self.logto_config.client_secret,
                },
            )
            return response.json()
    except httpx.HTTPStatusError as e:
        raise HTTPException(
            status_code=e.response.status_code,
            detail=f"HTTP error: {e.response.status_code} {e.response.text}",
        ) from e
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")