""" Access Validation ================= Validates the usage or non-usage of Grass headers. """ from functools import wraps import re from flask import request from ows_accounting import response from ows_accounting.constants import error from ows_accounting.constants import header from ows_accounting.models import period def verify_grass_header(required=False): """Verify grass headers. Applying this decorator to check if grass headers are there. Example usage: @app.route('/report', methods=['POST']) @verify_grass_header(request) def post_report(): Args: required (bool): if True, both id and type grass headers are required. Returns: func: reference to function which return decorated results. """ def inner_wrapper(func): @wraps(func) def wrapper(*args, **kwargs): account_id = request.headers.get(header.GRASS_ACCOUNT_ID) account_type = request.headers.get(header.GRASS_ACCOUNT_TYPE) params_account_id = request.args.get('account_id') params_account_type = request.args.get('account_type') vendor = header.GRASS_ACCOUNT_TYPE_VENDOR subaccount = header.GRASS_ACCOUNT_TYPE_SUBACCOUNT wrong_account_type = account_type and account_type not in [ vendor, subaccount] wrong_account_id = account_id and not account_id.isnumeric() accounts_dont_match = params_account_id and params_account_type \ and ( params_account_id != account_id or params_account_type != account_type) if required: if not account_id or not account_type: return response.flaskify(response.create_error_response( error.ERROR_CODE_BAD_GRASS_REQUEST, error.ERROR_MESSAGE_INCOMPLETE_GRASS_HEADERS)) # Additional validation only required when grass headers are # required or when required flag is False and there is grass header if account_id or account_type: # When account id exists, account type must also exists if # required flag is False. If both of them do not exists, this # validation will be skipped. if not account_id or not account_type: return response.flaskify(response.create_error_response( error.ERROR_CODE_BAD_GRASS_REQUEST, error.ERROR_MESSAGE_INCOMPLETE_GRASS_HEADERS)) if wrong_account_type: return response.flaskify(response.create_error_response( error.ERROR_CODE_BAD_GRASS_REQUEST, error.ERROR_MESSAGE_INVALID_GRASS_ACCOUNT_TYPE)) if wrong_account_id: return response.flaskify(response.create_error_response( error.ERROR_CODE_BAD_GRASS_REQUEST, error.ERROR_MESSAGE_INVALID_GRASS_ACCOUNT_ID)) if accounts_dont_match: return response.flaskify(response.create_error_response( error.ERROR_CODE_BAD_GRASS_REQUEST, error.ERROR_MESSAGE_INVALID_GRASS_ACCOUNT_ID)) return func(*args, **kwargs) return wrapper return inner_wrapper def verify_account_params(func): """Verify account id and account type. There are only 3 cases that are valid. Case one is when both account_id and account_type are provided only from grass headers; case two is when both account_id and account_type are provided only from user request; and finally, the third case is when id and type are provided from both grass headers and user params AND the account_id from grass header is equal to account id in user request and account_type from grass header is equal to account_type in user_request. Applying this decorator to check if grass headers are there. Example usage: @app.route('/report', methods=['GET']) @verify_account_params def get_report(): Args: func (func): reference to decorated function. Returns: func|flask response: reference to function which return decorated results. Flask error response if error. """ @wraps(func) def wrapper(*args, **kwargs): grass_account_id = request.headers.get(header.GRASS_ACCOUNT_ID) grass_account_type = request.headers.get(header.GRASS_ACCOUNT_TYPE) request_account_id = request.args.get('account_id') request_account_type = request.args.get('account_type') grass_headers_exist = grass_account_id and grass_account_type request_params_exist = request_account_id and request_account_type # Case when only grass account_id and account_type are provided. case1 = grass_account_id and grass_account_type case1 = case1 and not request_params_exist # Case when account_id and account_type are provided only in # request object. case2 = request_account_id and request_account_type case2 = case2 and not grass_headers_exist # Case when account_id and account_type are provided from both # the request object and grass headers. same_account_id = request_account_id == grass_account_id same_account_type = request_account_type == grass_account_type case3 = same_account_id and same_account_type case3 = case3 and grass_headers_exist and request_params_exist if case1 or case2 or case3: return func(*args, **kwargs) else: status = 400 if grass_headers_exist and request_params_exist and not case3: status = 403 return response.flaskify(response.create_error_response( error.ERROR_CODE_AUTHORIZATION, error.ERROR_MESSAGE_FORBIDDEN_USER, status)) return wrapper def verify_periods_params(func): """Verify if only the periods that have sales are passed in. Args: func (func): reference to decorated function. Returns: func|flask response: reference to function which return decorated results. Flask error response if error. """ @wraps(func) def wrapper(*args, **kwargs): account_id = request.headers.get( header.GRASS_ACCOUNT_ID) or request.args.get('account_id') account_type = request.headers.get( header.GRASS_ACCOUNT_TYPE) or request.args.get('account_type') available_periods_resp = period.get_available_accounting_periods( account_id, account_type) available_periods = sorted(available_periods_resp.message) periods = [] if request.method == 'POST': data = request.get_json() periods = data.get('periods') else: periods = request.args.get('periods', '') periods = ','.join(sorted(periods.split(','))) if available_periods and periods in ','.join( map(str, available_periods)): return func(*args, **kwargs) else: return response.flaskify(response.create_error_response( error.ERROR_CODE_INVALID_REQUEST, error.ERROR_MESSAGE_INVALID_PERIOD, 404)) return wrapper def verify_orchard_user_header(func): """Verify all headers required for payment hold request are set. Args: func (func): reference to decorated function. Returns: func: reference to function which return decorated results. """ @wraps(func) def inner_wrapper(*args, **kwargs): user_id = request.headers.get('Orchard-User-Id') if not user_id or not re.match(r'^(oa:|alw:)\d+$', user_id): return response.flaskify(response.create_error_response( error.ERROR_CODE_INVALID_REQUEST, error.ERROR_MESSAGE_INVALID_ORCHARD_USER_HEADERS)) return func(*args, **kwargs) return inner_wrapper