"""Util to parse headers. Additional methods to work with HTTP headers. """ from ddtrace import tracer from grass.consts import headers from grass.consts.headers import JWT_USER_METADATA from grass.consts.user import UUID_USERS @tracer.wrap() def extract_authorization_token(header): """Extract the authorization token from a header. Args: header (str): the header to get the value from. Returns: str: the token (if found) """ if not header: return '' header = header.strip().split(' ') if len(header) != 2 or header[0] != 'Bearer': return '' return header[1] @tracer.wrap() def profile_exist_in_payload(payload, profile_type, profile_id): """Verifies if profile exist in payload. Args: payload (list): List of profile dicts. profile_type (str): Type of profile. profile_id (str): Profile identifier. Returns: bool: True if profile exist in payload, else false. """ if not payload.get(headers.JWT_PROFILES): return False for profile in payload.get(headers.JWT_PROFILES): if profile.get('profile_type').lower() == profile_type.lower() and str( profile.get('profile_id') ) == str(profile_id): return True return False @tracer.wrap() def profile_uuid_exist_in_payload(payload, profile_uuid): """Verifies if profile exist in payload. Args: payload (list): List of profile dicts. profile_uuid (str): Profile uuid. Returns: bool: True if profile exist in payload, else false. """ if not payload.get(headers.JWT_PROFILES): return False for profile in payload.get(headers.JWT_PROFILES): if profile.get('uuid').lower() == profile_uuid.lower(): return True return False @tracer.wrap() def get_orchard_identity_id(payload): """Get orchard_identity_id from payload. Args: payload (list): List of profile dicts. Returns: str: orchard_identity_id. """ if not payload: raise Exception('Invalid JWT payload.') neo4j_identity_id = payload.get(JWT_USER_METADATA, {}).get('orchardIdentityId') auth0_sub = payload.get('sub', '') if 'auth0|' in auth0_sub: if auth0_sub in UUID_USERS and neo4j_identity_id: return neo4j_identity_id return auth0_sub.lstrip('auth0|') if 'google-apps|' in auth0_sub: if neo4j_identity_id: return neo4j_identity_id return auth0_sub if 'waad|' in auth0_sub: if neo4j_identity_id: return neo4j_identity_id return auth0_sub raise Exception('Invalid sub info in JWT .') @tracer.wrap() def get_orchard_identity_uuid(payload): """Get orchard_identity_uuid from payload. Args: payload (list): List of profile dicts. Returns: str: orchard_identity_uuid """ if not payload: raise Exception('Invalid JWT payload.') return payload.get(JWT_USER_METADATA, {}).get('orchardIdentityId')