"""RequestContext class and helper function. Determine "profile" or "account" context from request headers (this module is not tied to flask in any way). """ from random import randint from typing import Optional from jwtauth import JWTAuth from jwtauth.utils import jwt_auth_from_environment from requests.structures import CaseInsensitiveDict from owsrequest import auth from owsrequest import config from owsrequest.config import logger from owsrequest.constants import headers from owsrequest.constants.environment import QA_ENVIRONMENT from owsrequest.constants.headers import DEFAULT_BRAND class RequestContext: """ Class to determine "profile" vs "account" context from HTTP headers. The internal attributes will be referenced using "profile" nomenclature but will be overloaded for both "account" and "profile" contexts. Attributes: brand (str): 'brand' associated with the request context_type: (str): 'profile' or 'account' profile_type (str): Orchard-Profile-Type or Grass-Account-Type profile_id (str): Orchard-Profile-Id or Grass-Account-Id identity_id (str): Auth0 user id orchard_user_id (str): Orchard-User-Id (only present when context_type is "account") label_profile (bool): if True, treat Orchard-User-Id as profile id """ def __init__( self, request_headers: CaseInsensitiveDict, label_profile=False, jwt_auth_client: Optional[JWTAuth] = None, ): """Initialize class. Args: self (RequestContext): self request_headers (CaseInsensitiveDict): request headers "universal id" - Orchard-Identity-Uuid "account" based headers: - Grass-Account-Type - Grass-Account-Id - Orchard-User-Id "profile" based headers: - Orchard-Profile-Type - Orchard-Profile-Id - Orchard-Profile-UUID - Orchard-Identity-Id label_profile (bool) jwt_auth_client (Optional[JWTAuth]): client to use to decode JWT Tokens """ if type(request_headers).__name__ != 'CaseInsensitiveDict': raise TypeError( 'request_headers must be instance of CaseInsensitiveDict') # account context headers grass_account_type = request_headers.get( headers.GRASS_ACCOUNT_TYPE, None) grass_account_id = request_headers.get( headers.GRASS_ACCOUNT_ID, None) orchard_user_id = request_headers.get( headers.ORCHARD_USER_ID, None) # profile context headers profile_type = request_headers.get( headers.ORCHARD_PROFILE_TYPE, None) profile_id = request_headers.get( headers.ORCHARD_PROFILE_ID, None) profile_uuid = request_headers.get( headers.ORCHARD_PROFILE_UUID, None) # identity identity_id = request_headers.get( headers.ORCHARD_IDENTITY_ID, None) identity_uuid = request_headers.get( headers.ORCHARD_IDENTITY_UUID, None) roles = request_headers.get(headers.ORCHARD_ROLES, []) # add the requesting microservice name self.requestor_service_name = request_headers.get( headers.ORCHARD_REQUESTOR_SERVICE, None) # determine context type based on headers and set class attributes # default to context type to error self.context_type = headers.CONTEXT_TYPE_ERROR # always set identity id and UUID self.identity_id = identity_id self.identity_uuid = identity_uuid # initialize profile attributes self.profile_type = None self.profile_id = None self.profile_uuid = None # Initialize things from the JWT self.brand = DEFAULT_BRAND self.jwt_identity_id = None # both Profiles and Grass headers should have roles self.roles = [] if roles: self.roles = roles.split(',') # profile based context takes precedence if (profile_type and profile_id) or profile_uuid: self.context_type = headers.CONTEXT_TYPE_PROFILE self.profile_type = profile_type self.profile_id = profile_id self.profile_uuid = profile_uuid # set orchard_user_id attribute for OA with profile context if orchard_user_id and orchard_user_id.startswith('oa'): self.orchard_user_id = orchard_user_id # account based context elif ( (grass_account_type and grass_account_id) or orchard_user_id ): if orchard_user_id and label_profile: if orchard_user_id.startswith('alw'): self.profile_id = int(orchard_user_id.lstrip('alw:')) self.profile_type = headers.PROFILE_TYPE_LABEL self.context_type = headers.CONTEXT_TYPE_PROFILE if orchard_user_id.startswith('oa'): self.profile_id = int(orchard_user_id.lstrip('oa:')) self.profile_type = headers.PROFILE_TYPE_ORCH_ADMIN self.context_type = headers.CONTEXT_TYPE_PROFILE else: self.context_type = headers.CONTEXT_TYPE_ACCOUNT self.profile_type = grass_account_type self.profile_id = grass_account_id # this attribute only exists in this context self.orchard_user_id = orchard_user_id # only identity id or identity_uuid elif ( (self.context_type != headers.CONTEXT_TYPE_ACCOUNT) and (identity_id or identity_uuid) ): self.context_type = headers.CONTEXT_TYPE_PROFILE # no context elif ( not grass_account_type and not grass_account_id and not orchard_user_id and not profile_type and not profile_id and not identity_id and not identity_uuid ): self.context_type = headers.CONTEXT_TYPE_NONE authorization = request_headers.get(headers.AUTHORIZATION, None) self.authorization = authorization if authorization and \ authorization.startswith('Bearer ') else None debug = config.ENVIRONMENT == QA_ENVIRONMENT and randint(0, 100) == 1 if self.authorization: if not jwt_auth_client: jwt_auth_client = jwt_auth_from_environment(config.ENVIRONMENT) try: [_, token] = self.authorization.split(' ') decoded_response = auth.validate_and_decode_jwt_token( token, jwt_auth_client=jwt_auth_client) decoded = decoded_response.message self.brand = \ decoded.get(headers.JWT_BRAND_FIELD, DEFAULT_BRAND) if debug: logger.info('Valid authorization') self.jwt_identity_id = _get_jwt_identity_id(decoded) except Exception as e: logger.info( f'Failed to decode brand from jwt ' f'{config.ENVIRONMENT} Exception: {str(e)}') if debug: logger.info('Invalid authorization') def _get_jwt_identity_id(decoded_jwt) -> Optional[str]: """Get orchardIdentityId from a decoded JWT.""" identity_uuid = None grass_data = decoded_jwt.get(headers.JWT_USER_METADATA_FIELD) if grass_data: identity_uuid = grass_data.get(headers.JWT_CLAIM_ORCHARD_IDENTITY_ID) return identity_uuid def get_request_context_from_headers( headers, label_profile: bool = False, jwt_auth_client: Optional[JWTAuth] = None, ): """Given a list of header key-values, determine request context type. Args: headers (list or dict): request headers "account" based headers: - Grass-Account-Type - Grass-Account-Id - Orchard-User-Id "profile" based headers: - Orchard-Profile-Type - Orchard-Profile-Id - Orchard-Profile-UUID - Orchard-Identity-Id label_profile (bool) jwt_auth_client (Optional[JWTAuth]): Client to decode jwt tokens Returns: (RequestContext): RequestContext class with attribute context_type """ return RequestContext( headers, label_profile=label_profile, jwt_auth_client=jwt_auth_client, )