"""Utility Functions for Handlers.""" from owsrequest.constants import headers as header_constants from product_digital.constants import authorization from product_digital.constants import header from product_digital.constants.account import ACCOUNT_ID_KEY from product_digital.constants.account import SUBACCOUNT_ID_KEY from product_digital.models import ows_product def get_grass_headers(request): """Helper for extracted the grass headers. Args: request (Flask.request): the request. Returns: tuple: contains grass account type and grass account id. """ header_account_type = request.headers.get(header.GRASS_ACCOUNT_TYPE) header_account_id = request.headers.get(header.GRASS_ACCOUNT_ID) return (header_account_type, header_account_id) def both_grass_headers_present(request): """Helper to determind if both grass headers are present. Args: request (Flask.request): the request. Returns: Boolean (True/False) if both grass headers are present. """ header_account_type, header_account_id = get_grass_headers(request) return bool(header_account_type and header_account_id) def either_grass_headers_present(request): """Helper to determind if both grass headers are present. Args: request (Flask.request): the request. Returns: Boolean (True/False) if either grass headers are present. """ header_account_type, header_account_id = get_grass_headers(request) if header_account_type or header_account_id: return True return False def check_product_ownership(product_id, account_type, account_id): """Check if product is owned by the given account. Args: product_id (int): the id of the product to check account_type (str): type of account (vendor or subaccount) account_id (int): id of the account to check Returns: response.Response: status code 200 if owner, 403 if not owner """ return ows_product.check_ownership(product_id, account_type, account_id) def get_account_details(request): """Get account_id and account_type from the request.""" account_id = request.headers.get(header.GRASS_ACCOUNT_ID, '') account_type = request.headers.get(header.GRASS_ACCOUNT_TYPE, '') orchard_user_id = request.headers.get(header.ORCHARD_USER_ID, '') payload = request.get_json(silent=True) if orchard_user_id and orchard_user_id.startswith(header.OA_USER_PREFIX) and ( (payload.get(ACCOUNT_ID_KEY) or payload.get(SUBACCOUNT_ID_KEY))): account_id = payload.get(ACCOUNT_ID_KEY) subaccount_id = payload.get(SUBACCOUNT_ID_KEY) if subaccount_id: account_id = subaccount_id account_type = header.GRASS_ACCOUNT_TYPE_SUBACCOUNT else: account_id = account_id account_type = header.GRASS_ACCOUNT_TYPE_VENDOR return account_id, account_type def get_orchard_user_id(request): """Return an orchard-user-id value based on the available request headers. If orchard-user-id is defined, return that, otherwise munge one together based on OrchAdminProfile headers. Args: request (flask.Request): the request object Returns: string: representing the orchard-user-id value """ legacy = request.headers.get(header.ORCHARD_USER_ID) if legacy: return legacy profile_type = request.headers.get(header_constants.ORCHARD_PROFILE_TYPE) if profile_type == header_constants.PROFILE_TYPE_ORCH_ADMIN: return 'oa:%s' % request.headers.get(header_constants.ORCHARD_PROFILE_ID) def is_identity_authorized(identity_id): """Check if an identity has access to an endpoint. Args: jwt_identity_id (str): The identity UUID. Returns: bool: True if the identity is authorized """ return identity_id in authorization.AUTHORIZED_IDENTITIES