"""Handlers. Minimal handlers for health check, token generation, and vendor agreements. """ import json from time import time from flask import g, request import jwt from neo4j.exceptions import Neo4jError from owsrequest import flask_request from owsresponse import response from owsresponse.adaptors.flask import flaskify from users import config, constants from users.app import app from users.connectors import aws_secrets_manager from users.logic import user_info, vendor_agreement from users.utils.basic_utils import sign_token @app.route(config.HEALTH_CHECK) def hello_world(): """Health check.""" return 'Hello World', 200 @app.errorhandler(500) def exception_handler(error): """Handle error when uncaught exception is raised. Default exception handler. Note: Exception will also be sent to Sentry if config.SENTRY is set. Returns: flask.Response: A 500 response with JSON 'code' & 'message' payload. """ message = 'The server encountered an internal error and was unable to complete your request.' g.log.exception(error) return flaskify(response.create_fatal_response(message)) @app.errorhandler(Neo4jError) def neo4j_error_handler(error): """Handle uncaught Neo4j driver errors. A query that fails inside a transaction leaves the transaction unusable; the connector rolls it back on session exit. Capture the driver error here and return a clean response instead of leaking the raw exception (or a confusing downstream error from the connector). Returns: flask.Response: A 500 response with JSON 'code' & 'message' payload. """ message = 'The server encountered a database error and was unable to complete your request.' g.log.exception(error) return flaskify(response.create_fatal_response(message)) @app.route('/ws/generate-token/', methods=['GET']) def generate_token(token_type): """Generate token based on a token type. The query params are: timestamp, label_id, user_id and subaccount_id if we are requesting a token for a subaccount. Args: token_type (str): the type of the token e.g. feature-fm Returns: flask.Response: containing the token. """ account_type = request.headers.get(constants.GRASS_ACCOUNT_TYPE) account_id = request.headers.get(constants.GRASS_ACCOUNT_ID) user_id = request.headers.get(constants.ORCHARD_USER_ID) query_string = 'timestamp={0}&label_id={1}&user_id={2}' timestamp = int(time()) base_path = constants.URL_PATHS[token_type] if not timestamp or not account_id or not user_id: return flaskify( response.create_error_response( code=constants.BAD_PARAMS_ERROR_CODE, message=constants.BAD_PARAMS_ERROR_message, status=400, ) ) access_response = flask_request.verify_grass_access( request, required=False, subaccount=account_id, vendor=account_id ) if not access_response: return response.flaskify(access_response) if account_type == constants.GRASS_ACCOUNT_TYPE_VENDOR: query_string = query_string.format(timestamp, account_id, user_id) unassigned_token = bytes('{0}?{1}'.format(base_path, query_string).encode('utf-8')) if account_type == constants.GRASS_ACCOUNT_TYPE_SUBACCOUNT: query_string = 'timestamp={0}&subaccount_id={1}&user_id={2}'.format( timestamp, account_id, user_id ) unassigned_token = bytes('{0}?{1}'.format(base_path, query_string).encode('utf-8')) secret_token = aws_secrets_manager.get_secret(token_type) secret_token_value = bytes(secret_token.get(config.FEATURE_FM_SECRET_TOKEN_KEY).encode('utf-8')) signed_token = sign_token(unassigned_token, secret_token_value).decode('utf-8') url_path = {'url_path': '{0}?{1}&token={2}'.format(base_path, query_string, signed_token)} return flaskify(response.Response(url_path)) @app.route('/ws/generate-jwt-token/feature-fm', methods=['GET']) def generate_jwt_token(): """Generate jwt token signed for feature-fm. Returns: flask.Response: containing the jwt token. """ account_type = request.headers.get(constants.GRASS_ACCOUNT_TYPE) account_id = request.headers.get(constants.GRASS_ACCOUNT_ID) user_id = request.headers.get(constants.ORCHARD_USER_ID) correlation_id = g.correlation_id if not account_id or not user_id: return flaskify( response.create_error_response( code=constants.BAD_PARAMS_ERROR_CODE, message=constants.BAD_PARAMS_ERROR_message, status=400, ) ) if 'alw' in user_id: user_id = user_id.split('alw:')[1] access_response = flask_request.verify_grass_headers(request, True) if not access_response: return response.flaskify(access_response) secret = aws_secrets_manager.get_secret('feature-fm').get(config.FEATURE_FM_SECRET_TOKEN_KEY) user_response = user_info.get_user_info_feature_fm( user_id, account_type, int(account_id), correlation_id ) if not user_response: return flaskify(user_response) result = user_response.message encoded = jwt.encode(result, secret, algorithm='HS256') encrypted_data = {'jwt': encoded} return flaskify(response.Response(encrypted_data)) @app.route('/ws/agreements/vendor/', methods=['POST']) def create_vendor_agreement(permission_type_id): """Create a new vendor agreement. Args: permission_type_id (int): the permission type id as defined in opt_in_preference table Returns: flask.Response: JSON object representing vendor agreement data. """ account_id = request.headers.get(constants.GRASS_ACCOUNT_ID) user_id = request.headers.get(constants.ORCHARD_USER_ID) impersonator_user_id = None if request.data: impersonator_user_id = json.loads(request.data.decode('utf-8')).get( 'impersonator_user_id', None ) if not account_id and not user_id: return flaskify( response.create_error_response( code=constants.BAD_PARAMS_ERROR_CODE, message=constants.BAD_PARAMS_ERROR_message, status=400, ) ) access_response = flask_request.verify_grass_headers(request, True) if not access_response: return flaskify(access_response) if 'alw:' in user_id: user_id = user_id.split('alw:')[1] data = { 'vendor_id': int(account_id), 'opt_in_preference_id': permission_type_id, 'user_id': int(user_id), } if impersonator_user_id: data.update({'impersonator_user_id': impersonator_user_id}) return flaskify(vendor_agreement.create_vendor_agreement(data)) @app.route('/ws/agreements/vendor/', methods=['GET']) def get_vendor_agreement(permission_type_id): """Fetch a vendor agreement. Args: permission_type_id (int): the permission type id as defined in opt_in_preference table exclude_impersonator (string): if "true" results with an impersonator user ID will be excluded Returns: flask.Response: JSON object representing vendor agreement data. """ account_id = request.headers.get(constants.GRASS_ACCOUNT_ID) if not account_id: return flaskify( response.create_error_response( code=constants.BAD_PARAMS_ERROR_CODE, message=constants.BAD_PARAMS_ERROR_message, status=400, ) ) access_response = flask_request.verify_grass_headers(request, True) if not access_response: return flaskify(access_response) exclude_impersonator = request.args.get('exclude_impersonator') == 'true' return flaskify( vendor_agreement.get_vendor_agreement( int(account_id), permission_type_id, exclude_impersonator=exclude_impersonator ) ) @app.route('/ws/agreements/vendor', methods=['DELETE']) def delete_vendor_agreement(): """Delete a vendor agreement. Intended to by called manually to aid in testing. Args: id (int): id of the agreement to delete Returns: flask.Response: Representing the result of the operation """ try: vendor_agreement_id = int(request.args.get('id')) except Exception: return flaskify( response.create_error_response( code=constants.BAD_PARAMS_ERROR_CODE, message=constants.BAD_PARAMS_ERROR_message, status=400, ) ) return flaskify(vendor_agreement.delete_vendor_agreement(vendor_agreement_id))