teal.web_listener.utils – Utilities for the TeaL web listener#

class teal.web_listener.utils.Base64JSONEncodedCallbackState(*, final_redirect_url: AnyHttpUrl)#

Bases: BaseModel

Base64-encoded JSON callback state handling.

Such states mainly use the TeaL Web Listener as a bouncer. They are base64-encoded UTF-8 JSON payloads with the following keys:

  • redirect_uri: The final redirect URL.

  • state: The state to include in the final redirect URL.

Note that while the initially encoded state may have had base64 padding, they may have been trimmed by authorization servers through which the state has transited, and as such, we must support this case as well.

classmethod decode(value: str, /) Base64JSONEncodedCallbackStateType#

Decode a base64 JSON encoded callback state.

Parameters:

value – The raw value to decode.

Returns:

The decoded state.

Raises:

ValueError – An invalid format was detected.

encode() str#

Encode the state into a base64 JSON-encoded callback state.

Returns:

The encoded version of the state information.

final_redirect_url: AnyHttpUrl#

The final redirect URL.

class teal.web_listener.utils.CallbackStateInformation(*, state: ConstrainedStrValue, final_redirect_url: AnyHttpUrl | None = None, with_fragment: bool = False)#

Bases: BaseModel

State information, to be exploited in the router.

final_redirect_url: AnyHttpUrl | None#

The final redirect URL.

classmethod from_powens_redirect_data(*, state: str, final_redirect_url: str, with_fragment: bool = False, full_url: str, request: Request) CallbackStateInformationType#

Get callback state information from redirect data.

Parameters:
  • state – The state for which to redirect.

  • final_redirect_url – The final redirect URL to redirect to.

  • with_fragment – Whether to force getting the fragment or not.

  • full_url – The full URL from which to gather the original parameters.

  • request – The request in which the callback URL was submitted.

async classmethod from_url(full_url: str, /, *, request: Request, settings: Settings) CallbackStateInformationType#

Get state information regarding a state provided in an URL.

Parameters:
  • full_url – The full URL from which to gather a state.

  • settings – The settings applying to the current request.

Returns:

The callback information obtained from the state present in the callback URL.

Raises:
state: str#

The callback state.

with_fragment: bool#

Whether to force gathering the fragment or not.

class teal.web_listener.utils.OpenIDCIBAPingPayload(*, auth_req_id: str)#

Bases: BaseModel

OpenID Connect CIBA Ping payload.

This model is taken from OpenID Connect Client-Initiated Backchannel Authentication Flow (Core 1.0) specification, section 10.2.

class Config#

Bases: object

Model configuration.

extra = 'forbid'#
auth_req_id: str#

The authentication request identifier.

class teal.web_listener.utils.OpenIDCIBAPushErrorPayload(*, auth_req_id: str, error: str, error_description: str | None = None)#

Bases: BaseModel

OpenID Connect CIBA Push Error payload.

This model is taken from OpenID Connect Client-Initiated Backchannel Authentication Flow (Core 1.0) specification, section 12.

class Config#

Bases: object

Model configuration.

extra = 'forbid'#
auth_req_id: str#

The authentication request identifier.

error: str#

The error code, usually among:

access_denied

The end-user denied the authorization request.

expired_token

The authentication request identifier has expired. The Client will need to make a new Authentication Request.

transaction_failed

The OpenID Provider encountered an unexpected condition that prevented it from successfully completing the transaction.

error_description: str | None#

The human-readable text providing additional information.

class teal.web_listener.utils.OpenIDCIBAPushTokenPayload(*, auth_req_id: str, access_token: str, token_type: str, refresh_token: str, expires_in: int, id_token: str)#

Bases: BaseModel

OpenID Connect CIBA Push Successful Token payload.

This model is taken from OpenID Connect Client-Initiated Backchannel Authentication Flow (Core 1.0) specification, section 10.3.1.

class Config#

Bases: object

Model configuration.

extra = 'forbid'#
access_token: str#

The obtained access token.

auth_req_id: str#

The authentication request identifier.

expires_in: int#

The number of seconds in which the token expires.

id_token: str#

The OpenID token.

refresh_token: str#

The refresh token.

token_type: str#

The token type.

class teal.web_listener.utils.PowensPackedCallbackState(*, final_redirect_url: AnyHttpUrl, with_fragment: bool = False)#

Bases: BaseModel

Packed callback state format handler from Powens.

The format for the state is the following:

B1<config flags>[port]<host suffix><path index><host>[_<state>]

Where:

  • The configuration flags are represented as a single character encoded unsigned integer (up to 63), with each bit representing a flag.

  • (only present if PORT_SET flag is set) The port, as a three-character (18-bit) big-endian encoded value for the port.

  • The host suffix, as a one-character integer.

  • The path code, as a one-character integer.

  • The host, stripped of its suffix.

  • The state to provide to the final URL.

Note that if there is no state to transmit, the underscore does not need to be present.

The allowed configuration flags are the following:

  • PORT_SET (0x01): whether the port is the non-standard port for the provided protocol.

  • PORT_DEFAULT (0x02): whether the non-standard port is the default non-standard port 3158 (1) or is provided in the three characters following the configuration flags. This flag is a no-op if PORT_SET is not set.

  • HAS_FRAGMENT (0x04): whether the fragment should be forcefully gathered by the callback.

  • IS_HTTP (0x08): whether the protocol of the final redirect is ‘http’ (1) or ‘https’ (0).

Other configuration flags (0x10, 0x20) are reserved and should be set to 0.

Example packed URL states are the following:

B1001www.example.org

Redirect to https://www.example.org/ (no port, no fragment, use HTTPS, empty host suffix, / path).

B1021budgea_abc

Redirect to https://budgea-sandbox.biapi.pro/?state=abc (no port, no fragment, use HTTPS, -sandbox.biapi.pro host suffix, / path).

B1033budgea

Redirect to https://budgea.biapi.pro/2.0/webauth/resume (no port, no fragment, use HTTPS, .biapi.pro host suffix, /2.0/webauth/resume path).

B1b02localhost

Redirect to http://localhost:3158/webauth/resume (port set to default non-standard port 3158, use HTTP, empty host suffix, /webauth/resume path).

B110ji01localhost

Redirect to https://localhost:1234/ (port set to non-standard port 1234, use HTTPS, empty host suffix, / path).

Note that 0ji is the encoded version of 1234, since:

  • The offset of i is 18.

  • The offset of j is 19.

  • The offset of 0 is 0.

  • The result of 0 * 64 ** 2 + 19 * 64 + 18 is 1234.

CHARACTER_SET: ClassVar[str] = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-.'#

Characters served for encoding an integer.

Note that this set is 64-characters long, which means an integer encoded using this character set is 6-bit wide (0 to 63).

DECODE_PATTERN: ClassVar[Pattern] = re.compile('B1(?:([159dhlptxBFJNRVZ])([^_]{3})|([^159dhlptxBFJNRVZ]))([^_])([^_])([^_]+)(?:_(.*))?')#

The pattern for decoding the state.

Note that 159dhlptxBFJNRVZ includes all of the characters for which PORT_SET is set but PORT_DEFAULT is unset, which constitutes the condition in which the port is present.

DEFAULT_NONSTANDARD_PORT: ClassVar[int] = 3158#

Default port.

HAS_FRAGMENT: ClassVar[int] = 4#

Flag set when the fragment should systematically be gathered.

IS_HTTP: ClassVar[int] = 8#

Flag set when the scheme is ‘http’ instead of ‘https’.

KNOWN_HOST_SUFFIXES: ClassVar[dict[int, str]] = {0: '', 2: '-sandbox.biapi.pro', 3: '.biapi.pro'}#

Known host endings.

KNOWN_PATHS: ClassVar[dict[int, str]] = {1: '/', 2: '/webauth/resume', 3: '/2.0/webauth/resume'}#

Known path prefixes.

PORT_DEFAULT: ClassVar[int] = 3#

Flag set when the port is the default non-standard port 3158.

Note that this is a combination of 0x01 and 0x02, as 0x02 on its own is ineffective.

PORT_SET: ClassVar[int] = 1#

Flag set when the port is the non-standard port for the scheme.

classmethod decode(value: str, /) PowensPackedCallbackStateType#

Decode a packed URL state type.

Parameters:

value – The raw value to decode.

Returns:

The state to decode.

Raises:

ValueError – An invalid format was detected.

encode() str#

Encode the state data into the packed URL state format.

Returns:

The encoded version of the stored state information.

Raises:

ValueError – The data from the URL state is invalid.

final_redirect_url: AnyHttpUrl#

The final redirect URL.

with_fragment: bool#

Whether to force gathering fragments for the provided state.

class teal.web_listener.utils.QueryStringMiddleware(app: FastAPI, *, delimiter: str)#

Bases: object

Middleware for supporting URLs with a different query string marker.

This is useful for APIs with weird query string management, such as https://example.org/callback&state=abc&code=def, where the query string is actually marked with a first ampersand rather than a question mark.

app: FastAPI#

The application on which the middleware should act.

delimiter: str#

The delimiter used as an alternative to ?.

For example, if using & as the delimiter here, the query string determined from the https://example.org/callback&state=abc&code=def will be state=abc&code=def.

teal.web_listener.utils.find_state_in_url(url: str, /) str | None#

Find the state in a given URL.

Note that this will look for query parameters first, then fragment if necessary.

Parameters:

url – The URL to look for a state in.

Returns:

The found state, or None if no state could be found.

teal.web_listener.utils.fix_url_query_separator(url: str, /) str#

Replace alternative query separators with ‘?’ if necessary.

This helps mirroring the behaviour we have with QueryStringMiddleware with the /raw-callback ASGI endpoint.

Parameters:

url – The URL to fix.

Returns:

The fixed URL.

teal.web_listener.utils.get_powens_hmac_signature(request: Request) PowensHMACSignature | None#

Get the Powens HMAC signature from a fastapi request.

For more information on the expected headers or header format, see Authentication with a HMAC signature header.

Parameters:

request – The request from which to gather the HMAC signature data.

Returns:

The HMAC signature, or None if no such signature could be found.

teal.web_listener.utils.get_powens_user_token(request: Request) str | None#

Get the Powens user-scoped token if available.

For more information, see Authentication with user-scoped token.

Parameters:

request – The request from which to gather a user-scoped token.

Returns:

The user-scoped token, or None if no such token could be found.