dal.classes.utils package

Submodules

dal.classes.utils.local_robot_info module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

Usage:

a basic class for getting robot information.

Developers: - Ofer Katz (ofer@mov.ai) - 2022

class dal.classes.utils.local_robot_info.LocalRobotInfo

Bases: object

Helper class for retrieving the robot id and robot name from the local Redis DB

static generate_identity_string(component: str) → str

A string with with the robot info.

Parameters:

component (str) – The component who needs that identity.

Returns:

The identity string.

Return type:

str

static get_id()

Returns the robot ID

static get_name()

Returns the robot name

robot_id = None
robot_name = None

dal.classes.utils.secretkey module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

Developers: - Erez Zomer (erez@mov.ai) - 2022

class dal.classes.utils.secretkey.SecretKey

Bases: object

classmethod create(fleet_name: str, length: int = 32) → None

creates a new key in the DB

Parameters:

  • fleet_name (str) – The name of the fleet which owns the key.

  • length (int, optional) – The length of the key.. Defaults to 32.

Returns:

True if succesfull, False otherwise.

Return type:

bool

classmethod db()
classmethod get_secret(fleet_name: str) → str

returns the secret content.

Parameters:

fleet_name (str) – The name of the fleet which owns the key.

Returns:

The secret.

Return type:

str

classmethod is_exist(fleet_name: str) → bool

checks if a specified key exist in DB.

Parameters:

fleet_name (str) – The name of the fleet which owns the key.

Returns:

True if exist, False otherwise.

Return type:

bool

log = <Logger dal.classes.utils.secretkey (INFO)>
classmethod remove(fleet_name: str) → None

Removes a key from the DB.

Parameters:

fleet_name (str) – The name of the fleet which owns the key.

Returns:

True if succesfull, False otherwise.

Return type:

bool

secret_key_dict = {'KeyLength': '', 'LastUpdate': '', 'Secret': ''}
type_name = 'SecretKey'
classmethod update(fleet_name: str, length: int = 32) → None

updates an existing key in the DB.

Parameters:

  • fleet_name (str) – The name of the fleet which owns the key.

  • length (int, optional) – The length of the key.. Defaults to 32.

Returns:

True if succesfull, False otherwise.

Return type:

bool

dal.classes.utils.token module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

Developers: - Erez Zomer (erez@mov.ai) - 2022

class dal.classes.utils.token.DBToken(token: TokenObject, token_type: str = 'Token')

Bases: dict

class dal.classes.utils.token.EmptyDBToken(token_id: str | None = None, token_type: str = 'Token')

Bases: dict

class dal.classes.utils.token.Token

Bases: object

allowed_algorithms = ['HS256', 'RS256', 'ES256']
classmethod decode_and_verify_token(token: str, secret_key='551vDXYyMSDkRZphGdC7jq2HyXIAHMKQbcoXDwLWbx5tLMul-jlp2wYucld_2mvpTK4wsmWNaNfQ8VUHBL6p9g') → dict

This function verifies and decodes the token, the decoded token is returned in a dictionary with meaninigfull keys. if verification fails an Exception would be returned.

Parameters:

  • token (str) – a string representing the token (encoded JWT).

  • secret_key (str) – The key used to encrypt the data. Defaults to JWT_SECRET_KEY.

Raises:

wt.exceptions.InvalidTokenError – In case token verification fails.

Returns:

a dicitionary with all the token’s payload info.

Return type:

dict

classmethod decode_no_verify_token(token: str, secret_key='551vDXYyMSDkRZphGdC7jq2HyXIAHMKQbcoXDwLWbx5tLMul-jlp2wYucld_2mvpTK4wsmWNaNfQ8VUHBL6p9g') → dict

This function decodes the token withoud verifying it, the decoded token is returned in a dictionary with meaninigfull keys.

Parameters:

  • token (str) – a string representing the token (encoded JWT).

  • secret_key (str) – The key used to verify token signature. Defaults to JWT_SECRET_KEY.

Returns:

A dicitionary with all the token’s payload info.

Return type:

dict

classmethod encode_token(token_payload: dict, secret_key: str = '551vDXYyMSDkRZphGdC7jq2HyXIAHMKQbcoXDwLWbx5tLMul-jlp2wYucld_2mvpTK4wsmWNaNfQ8VUHBL6p9g', algorithm: str = 'HS256') → str

This function generates Json Web Token (JWT) that will use the client to access system resources.

Parameters:

  • token_payload (dict) – A dictionary containing the payload values.

  • secret_key (str) – The key used to sign token signature. Defaults to JWT_SECRET_KEY.

  • algorithm (str, optional) – The signing algorithm to use. Defaults to ‘HS256’.

Raises:

  • TokenError – In case the requested algorithm is not found on the allowed algorithms set.

  • TokenError – In case the one of the required keys is not found in the token.

Returns:

The encoded token.

Return type:

str

classmethod get_token_id(token: str)
classmethod get_token_obj(token: str) → TokenObject

Extracts the token and returns a TokenObject. :param token: the token to extract. :type token: str

Returns:

The token info wrapped in an object.

Return type:

TokenObject

classmethod init_payload(subject: str, expiration_delta: timedelta) → dict

initializes a dictionary which will be used as a payload for token (JWT) generation.

Parameters:

  • subject (str) – The subject of the token (‘Access’, ‘Refresh’)

  • expiration_delta (timedelta) – the time delta from now.

Returns:

the payload of the token.

Return type:

(dict)

log = <Logger Token (INFO)>
required_keys = {'exp', 'iat', 'iss', 'jti', 'sub'}
classmethod revoke_token(token: str) → None

Removes the token id from the allowed token list. If the token has a refresh_id token it will remove the refresh_id token as well.

Parameters:

token (str) – The token to revoke.

classmethod verify_token(token: str) → None

verifies the token validity

Parameters:

token (str) – The string representation of the token.

Raises:

  • TokenRevoked – In case the token was revoked from the DB.

  • TokenExpired – In case the token expiration time has expired.

  • InvalidToken – In case there is decode error

  • InvalidToken – In case the token is invalid.

class dal.classes.utils.token.TokenManager

Bases: object

A general class for managing tokens in DB.

classmethod db()
classmethod is_token_exist(token_id: str) → bool

Checks whether a specific token exist in the DB.

Parameters:

token (TokenObject) – The token to look for.

Returns:

True if exist, False otherwise.

Return type:

bool

log = <Logger TokenManger (INFO)>
async classmethod remove_all_expired_tokens()

Removes all the tokens that their expiration time has passed.

classmethod remove_all_tokens()

Removes all token from db.

classmethod remove_token(token_id: str) → None

Removes token scheme from DB by the Token id.

Parameters:

token (TokenObject) – The token to remove.

classmethod store_token(token: TokenObject) → None

Stors a token in the DB.

Parameters:

token (TokenObject) – The token to store.

token_type = 'Token'
class dal.classes.utils.token.TokenObject(token: dict)

Bases: object

An object for refercing token info with object attributes.

class dal.classes.utils.token.UserToken

Bases: Token

classmethod generate_access_token(user: BaseUser, refresh_id) → str

This function encapsulates the generate_token function to generate the access token.

Parameters:

  • user (BaseUser) – The user for whom the token is generated.

  • refresh_id (str) – The id of the associated refresh token. Defaults to empty string.

Returns:

The generated token decoded in utf-8.

Return type:

str

classmethod generate_refresh_token(user: BaseUser) → str

This function encapsulates the generate_token function to generate the refresh token.

Parameters:

user (BaseUser) – The user for whom the token is generated.

Returns:

The generated token decoded in utf-8.

Return type:

str

classmethod get_refresh_id(token: str) → str

Returns the refresh_id of the token if exist.

Parameters:

token (str) – The token where refresh_id is specified.

Returns:

The refresh_id if exists, None otherwise.

Return type:

str

classmethod get_token_obj(token: str) → UserTokenObject

Extracts the token and returns a UserTokenObject.

Parameters:

token (str) – the token to extract.

Returns:

The token info wrapped in an object.

Return type:

UserTokenObject

classmethod init_payload(user: BaseUser, subject: str, expiration_delta: timedelta, refresh_id: str)

initializes a dictionary which will be used as a payload for token (JWT) generation.

Parameters:

  • user – (BaseUser): The user to for whom the token will be generated.

  • expiration_delta (timedelta) – the time delta from now.

  • refresh_id (str) – An id of the assciated refresh token.

Returns:

The dictionary to use as a payload when encoding the token.

Return type:

(dict)

class dal.classes.utils.token.UserTokenObject(token: dict)

Bases: TokenObject

Extends the Token object with more attributes relating to user tokens.

Module contents