`mdsisclienttools.auth.TokenManager`

Module Contents

Classes

`StorageType`	str(object='') -> str
`BearerAuth`	Base class that all auth implementations derive from
`Stage`	str(object='') -> str
`Tokens`
`StageTokens`
`AuthFlow`	str(object='') -> str
`DeviceFlowManager`

Attributes

`LOCAL_STORAGE_DEFAULT`
`DEFAULT_CLIENT_ID`

class mdsisclienttools.auth.TokenManager.StorageType[source]

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

FILE = FILE

OBJECT = OBJECT

class mdsisclienttools.auth.TokenManager.BearerAuth(token: str)[source]

Bases: requests.auth.AuthBase

Base class that all auth implementations derive from

__call__(r: requests.PreparedRequest) → requests.PreparedRequest[source]

class mdsisclienttools.auth.TokenManager.Stage[source]

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

TEST = TEST

DEV = DEV

STAGE = STAGE

PROD = PROD

class mdsisclienttools.auth.TokenManager.Tokens[source]

Bases: pydantic.BaseModel

access_token :str

refresh_token :Optional[str]

class mdsisclienttools.auth.TokenManager.StageTokens[source]

Bases: pydantic.BaseModel

stages :Dict[Stage, Optional[Tokens]]

mdsisclienttools.auth.TokenManager.LOCAL_STORAGE_DEFAULT = .tokens.json

mdsisclienttools.auth.TokenManager.DEFAULT_CLIENT_ID = client-tools

class mdsisclienttools.auth.TokenManager.AuthFlow[source]

Bases: str, enum.Enum

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

DEVICE = DEVICE

OFFLINE = OFFLINE

class mdsisclienttools.auth.TokenManager.DeviceFlowManager(stage: str, keycloak_endpoint: str, auth_flow: AuthFlow = AuthFlow.DEVICE, offline_token: Optional[str] = None, client_id: str = DEFAULT_CLIENT_ID, local_storage_location: Optional[str] = None, local_storage_object: Optional[Dict[str, Any]] = None, scopes: List[str] = [], force_token_refresh: bool = False, silent: bool = False)[source]

optional_print(message: Optional[str] = None) → None[source]

Prints only if the silent value is not flagged.

Parameters: message (str) – The message to print.

retrieve_local_tokens(stage: Stage) → Optional[Tokens][source]

Retrieves credentials from a local cache file, if present. Credentials are on a per stage basis. If the creds are valid but expired, they will be refreshed. If this fails, then a failure is indicated by None.

Parameters: stage (Stage) – The stage to fetch creds for.
Returns: Tokens object if successful or None.
Return type: Optional[Tokens]

reset_storage() → None[source]: Resets the local storage by setting all values to None.

update_local_storage(stage: Stage) → None[source]

Pulls the current StageTokens object from cache storage, if present, then either updates the current stage token value in existing or new StageTokens object. Writes back to file.

Parameters: stage (Stage) – The stage to update

perform_offline_refresh() → Dict[str, Any][source]

perform_offline_refresh

Uses the current offline token to perform a token refresh.

Returns: The response from the token endpoint iff status code == 200
Return type: Dict[str, Any]
Raises: Exception – Exception if non 200 status code

get_tokens() → None[source]

Tries to get tokens. First attempts to pull from the local storage. Otherwise initiates a device auth flow then uses the token endpoint to generate the creds.

Raises

Exception – OAuth tokens not present in device auth flow
Exception – Tokens not present in keycloak token endpoint response

perform_token_refresh() → None[source]: Updates the current tokens by using the refresh token.

perform_refresh(tokens: Optional[Tokens] = None) → Dict[str, Any][source]

Helper function to perform refresh. This accepts tokens and other information from the class, calls the refresh endpoint, and responds with the keycloak token endpoint response.

Parameters: tokens (Optional[Tokens], optional) – The tokens object, by default None
Returns: The response from the keycloak endpoint as json dict.
Return type: Dict[str, Any]
Raises: Exception – Non 200 response code.

initiate_device_auth_flow() → Dict[str, Any][source]

Initiates OAuth device flow. This is triggered by a post request to the device endpoint of the keycloak server. The specified client (by id) must be public and have the device auth flow enabled.

Returns: The json response info from the device auth flow endpoint
Return type: Dict[str, Any]

get_token() → str[source]

Uses the current token - validates it, refreshes if necessary, and returns the valid token ready to be used.

Returns

The access token

Return type

str

Raises

Exception – Raises exception if tokens/public_key are not setup - make sure that the object is instantiated properly before calling this function
Exception – If the token is invalid and cannot be refreshed

get_auth() → BearerAuth[source]

A helper function which produces a BearerAuth object for use in the requests.xxx objects. For example:

manager = DeviceAuthFlowManager(…) auth = manager.get_auth requests.post(…, auth=auth)

Returns

The requests auth object.

Return type

BearerAuth

Raises

Exception – Tokens are not present
Exception – Token validation failed and refresh or device auth failed

retrieve_keycloak_public_key() → None[source]: Given the keycloak endpoint, retrieves the advertised public key. Based on https://github.com/nurgasemetey/fastapi-keycloak-oidc/blob/main/main.py

display_device_auth_flow(user_code: str, verification_url: str) → None[source]

Displays the current device auth flow challenge - first by trying to open a browser window - if this fails then prints suggestion to stdout to try using the URL manually.

Parameters

user_code (str) – The user code
verification_url (str) – The url which embeds challenge code

await_device_auth_flow_completion(device_code: str, interval: int, grant_type: str) → Optional[Dict[str, Any]][source]

Ping the token endpoint as specified in the OAuth standard at the advertised polling rate until response is positive or failure.

Parameters

device_code (str) – The device code
interval (int) – The polling interval (ms)
grant_type (str) – The OAuth grant type

Returns

If successful, the keycloak response

Return type

Optional[Dict[str, Any]]

validate_token(tokens: Optional[Tokens] = None) → None[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

mdsisclienttools.auth.TokenManager

Module Contents

Classes

Attributes

`mdsisclienttools.auth.TokenManager`