"""Validates the usage or non-usage of Grass headers, validate access.""" from functools import wraps from ddtrace import tracer from flask import request from oto.adaptors.flask import flaskify from owsrequest import flask_request from owsrequest.constants import headers as owsrequest_headers from sound_recordings.constants import error as error_constants from sound_recordings.constants.access import ACCESS_ANALYTICS from sound_recordings.permissions.analytics import has_analytics from sound_recordings.validation.dev import get_dev_auth, get_profile_dev_auth @tracer.wrap(name="_get_account_type_account_id_user_id") def _get_account_type_account_id_user_id(current_request): """Return values from grass headers.""" dev_auth = get_dev_auth() if dev_auth is not None: return ( dev_auth[owsrequest_headers.GRASS_ACCOUNT_TYPE], dev_auth[owsrequest_headers.GRASS_ACCOUNT_ID], dev_auth[owsrequest_headers.ORCHARD_USER_ID], ) account_type, account_id = flask_request.get_grass_headers(current_request) user_id = request.headers.get(owsrequest_headers.ORCHARD_USER_ID) return account_type, account_id, user_id @tracer.wrap(name="_get_profile_type_profile_id") def _get_profile_type_profile_id(current_request): """Return values from profile headers.""" profile_dev_auth = get_profile_dev_auth() if profile_dev_auth is not None: return ( profile_dev_auth[owsrequest_headers.ORCHARD_PROFILE_TYPE], profile_dev_auth[owsrequest_headers.ORCHARD_PROFILE_ID], ) return flask_request.get_profile_headers(current_request) @tracer.wrap(name="_verify_grass_headers") def _verify_grass_headers(current_request): """Return True if all the required headers are present and valid. Return error response otherwise. """ # if in a development environment, check for dev overrides if get_dev_auth() is not None: return True # validate Grass-Account-Type and Grass-Account-Id grass_validation = flask_request.verify_grass_headers( current_request, required=True ) if not grass_validation: return flaskify(grass_validation) # check if Orchard-User-Id is present user_id = request.headers.get(owsrequest_headers.ORCHARD_USER_ID) if not user_id: return error_constants.RESPONSE_INCOMPLETE_GRASS_HEADERS return True @tracer.wrap(name="_verify_profile_headers") def _verify_profile_headers(current_request): """Return True if all the required headers are present and valid. Return error response otherwise. """ # if in a development environment, check for dev overrides if get_profile_dev_auth() is not None: return True # validate Profiles profile_validation = flask_request.verify_profile_headers(current_request) if not profile_validation: return flaskify(profile_validation) return True def verify(grass_headers=True, access=ACCESS_ANALYTICS): """Verify grass headers (including Orchard-User-Id) and access to Analytics. For Analytics check feature on account level and to Analytics or Administrator role on user level. Args: grass_headers (bool): If True, verify that grass headers are present. access (bool): If True, verify access to the mobile app. Return wrapped function if checks passed, otherwise render grass validation or access denied error. """ def inner_wrapper(orig_func): @wraps(orig_func) def wrapper(*args, **kwargs): if grass_headers: grass_validation_result = _verify_grass_headers(request) if ( hasattr(grass_validation_result, "status_code") and grass_validation_result.status_code != 200 ): return grass_validation_result if access == ACCESS_ANALYTICS: ( account_type, account_id, user_id, ) = _get_account_type_account_id_user_id(request) profile_type, profile_id = _get_profile_type_profile_id(request) if not has_analytics( account_type, account_id, user_id, profile_type, profile_id ): return error_constants.RESPONSE_FORBIDDEN_USER account_type, account_id, user_id = _get_account_type_account_id_user_id( request ) return orig_func( *args, **kwargs, account_type=account_type, account_id=account_id, user_id=user_id ) return wrapper return inner_wrapper @tracer.wrap(name="verify") def verify_profile(access=ACCESS_ANALYTICS): """ Verify grass headers, profile headers and access to Analytics. For Analytics check feature on account level and to Analytics or Administrator role on user level. Args: access (bool): If True, verify access to the mobile app. Return wrapped function if checks passed, otherwise render grass validation or access denied error. """ def inner_wrapper(orig_func): @wraps(orig_func) def wrapper(*args, **kwargs): grass_validation_result = _verify_grass_headers(request) profile_validation_result = _verify_profile_headers(request) if ( hasattr(grass_validation_result, "status_code") and grass_validation_result.status_code != 200 and hasattr(profile_validation_result, "status_code") and profile_validation_result.status_code != 200 ): return grass_validation_result if access == ACCESS_ANALYTICS: ( account_type, account_id, user_id, ) = _get_account_type_account_id_user_id(request) profile_type, profile_id = _get_profile_type_profile_id(request) if not has_analytics( account_type, account_id, user_id, profile_type, profile_id ): return error_constants.RESPONSE_FORBIDDEN_USER return orig_func(*args, **kwargs) return wrapper return inner_wrapper