provenaclient.auth.helpers

Created Date: Tuesday June 18th 2024 +1000 Author: peter —– Last Modified: Tuesday June 18th 2024 12:44:03 pm +1000 Modified By: peter —– Description: Miscellaneous helper functions assisting with implementation of auth flows —– HISTORY: Date By Comments ———- — ———————————————————

Attributes

JWT_DEFAULT_WINDOW

Classes

`AccessToken`
`Tokens`
`HttpxBearerAuth`	Base class for all authentication schemes.

Functions

`check_token_expiry_window`(→ bool)	This helper function checks if the current JWT token will
`retrieve_keycloak_public_key`(→ str)	Given the keycloak endpoint, retrieves the advertised
`keycloak_refresh_token_request`(→ Dict[str, Any])	Performs the token refresh by making an HTTP post request to the token endpoint
`validate_access_token`(→ bool)	Uses the python-jose library to validate current creds.

Module Contents

provenaclient.auth.helpers.JWT_DEFAULT_WINDOW = 30

class provenaclient.auth.helpers.AccessToken[source]

Bases: pydantic.BaseModel

access_token: str

class provenaclient.auth.helpers.Tokens[source]

Bases: pydantic.BaseModel

access_token: str

refresh_token: str | None

class provenaclient.auth.helpers.HttpxBearerAuth(token: str)[source]

Bases: httpx.Auth

Base class for all authentication schemes.

To implement a custom authentication scheme, subclass Auth and override the .auth_flow() method.

If the authentication scheme does I/O such as disk access or network calls, or uses synchronization primitives such as locks, you should override .sync_auth_flow() and/or .async_auth_flow() instead of .auth_flow() to provide specialized implementations that will be used by Client and AsyncClient respectively.

token

auth_flow(request: httpx.Request) → Generator[httpx.Request, httpx.Response, None][source]

Execute the authentication flow.

To dispatch a request, yield it:

` yield request `

The client will .send() the response back into the flow generator. You can access it like so:

` response = yield request `

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.

provenaclient.auth.helpers.check_token_expiry_window(jwt_data: dict[str, Any], logger: logging.Logger, jwt_token_expiry_window: int = JWT_DEFAULT_WINDOW) → bool[source]

This helper function checks if the current JWT token will expire or not expire either within the provided or default (30sec) window. If the token is going to be expired in less than or within the provided or default (30sec). expiry window they will be refreshed.

Parameters:

jwt_data (dict[str,Any]) – A dictionary containing the token validation results.
jwt_token_expiry_window (Optional[int]) – A potential integer value containing your desired JWT expiry window.

Returns:

True: The current token will not expire within 30 seconds False: The current token will expire within 30 seconds.

Return type:

bool

provenaclient.auth.helpers.retrieve_keycloak_public_key(keycloak_endpoint: str, logger: logging.Logger) → str[source]: Given the keycloak endpoint, retrieves the advertised public key. Based on https://github.com/nurgasemetey/fastapi-keycloak-oidc/blob/main/main.py

provenaclient.auth.helpers.keycloak_refresh_token_request(token_endpoint: str, client_id: str, scopes: List[str], refresh_token: str, logger: logging.Logger) → Dict[str, Any][source]

Performs the token refresh by making an HTTP post request to the token endpoint to obtain new access and refresh tokens.

Parameters:

tokens (Optional[Tokens], optional) –

An optional Tokens object containing the refresh token. If not provided, the method will use the class variable stored tokens.

By default this parameter is None.

Returns:

A dictionary containing the new access and refresh tokens if the refresh is successful.

Return type:

Dict[str, Any]

Raises:

ValueError – If no refresh token is provided or found in the class token variable.
Exception – If the HTTP request fails a message is displayed with the HTTP status code. Can occur if the refresh token has expired.

provenaclient.auth.helpers.validate_access_token(public_key: str, access_token: str, logger: logging.Logger) → bool[source]

Uses the python-jose library to validate current creds.

In this context, it is basically just checking signature and expiry. The tokens are enforced at the API side as well.

Parameters:: tokens (Optional[Tokens], optional) – The tokens object to validate, by default None