"""Session. Sessions are short lived connections between the user and the API without the need of a client. The session contains information about the client that will consume the data and the user id. The main use case for this: when the UI needs to perform direct calls to the API to fetch additional data. """ import random from datetime import datetime from ddtrace import tracer from grass import api from grass.logic import auth from grass.models import oa_user, session, workstation_user from grass.utils import headers @tracer.wrap() def create_token(client_id, user_id): """Create a session token. Tokens have two parts: a hash and an expiration date. The expiration date conditions the duration of the token (how long will this be valid.) Args: client_id (str): the client id. user_id (int): the user id. Returns: `tuple`: the dictionary that contains the token and the expiration date, and a HTTP status. """ token = create_random_token() roles = [] roles_by_name = [] identity_uuid = None auth0_user_id = None if user_id.startswith('alw'): roles_resp = workstation_user.get_roles_for_user(user_id) if not roles_resp: return roles_resp roles = roles_resp.message['role_ids'] roles_by_name = roles_resp.message['role_names'] identity_resp = workstation_user.get_user_identity(user_id) if identity_resp.success: identity_uuid = identity_resp.message.get('id') auth0_user_id = identity_resp.message.get('auth0_user_id') elif user_id.startswith('oa'): roles_resp = oa_user.get_roles_for_user(user_id) if not roles_resp: return roles_resp roles = roles_resp.message['role_ids'] roles_by_name = roles_resp.message['role_names'] identity_resp = oa_user.get_user_identity(user_id) if identity_resp.success: identity_uuid = identity_resp.message.get('id') auth0_user_id = identity_resp.message.get('auth0_user_id') return session.create( client_id, user_id, token, roles=roles, roles_by_name=roles_by_name, identity_uuid=identity_uuid, auth0_user_id=auth0_user_id, ) @tracer.wrap() def get_token(token): """Get a token. Args: token (string): the token information. Returns: `tuple`: response of the token, which contains the data of the token (user id, client id) and the http status (404 and 200.) """ return session.get_token(token) @tracer.wrap() def create_random_token(): """Create a random session token. Returns: `str`: the random token. """ token = hex(random.getrandbits(128)) return token.replace('0x', '') @tracer.wrap() def delete_token(session_token): """Delete a session token. Args: session_token (str): the session token. Returns: 200 Session Deleted if success, else 404 Session does not exist. """ return session.delete(session_token) @tracer.wrap() def calculate_time_to_live(expire_time): """Calculate time difference. Args: expire_time (int): expire_time. Returns: time to live in seconds. """ current_time_utc = int(datetime.now().strftime('%s')) time_difference = expire_time - current_time_utc return time_difference @tracer.wrap() def store_auth_token_on_logout(auth_token): """Use to store auth token in redis cache on logout Args: token (str): authorization token. Returns: message in string. """ token = headers.extract_authorization_token(auth_token) if token: message, status, payload = auth.validate_auth0_token(token) if status != 200: api.logger.info(f'{status}:{message}') return 'Token not cached' else: expire_time = payload['exp'] time_to_live = calculate_time_to_live(expire_time) if time_to_live > 0 and time_to_live <= 1200: session.store_auth_on_logout(token, time_to_live) return 'Token cached' else: return 'Token not provided'