"""Application. The API application is a `Tornado` application. It provides simple features such as registering a url for a specific handlers, access control, login a user, and performing simple tasks. """ import logging import time import urllib import uuid from functools import partialmethod, wraps import sentry_sdk import tornado.httpclient from ddtrace import tracer from owslogger import logger as ows_logger from owsrequest.constants import headers as owsheaders from toolz import dicttoolz from tornado import web from grass import access, config from grass.connectors import datadog from grass.consts import headers from grass.logger import CustomAdapter from grass.logic import user class Orch404Handler(web.RequestHandler): def prepare(self): self.set_status(404) self.clear_header('Server') self.finish('404: Not Found') app = web.Application(default_handler_class=Orch404Handler, compress_response=True) Base = (web.RequestHandler,) # Exception handling. if config.SENTRY: sentry_sdk.init(dsn=config.SENTRY) sentry_sdk.integrations.logging.ignore_logger('ddtrace.*') # Log handling. ows_logger.setup( config.environment, None, # add all loggers (ROOT and down) to owslogger logging.INFO, config.SERVICE_NAME, config.SERVICE_VERSION, ) # disable generic access logging logging.getLogger('tornado.access').disabled = True logger = logging.getLogger(config.LOGGER_NAME) logger.setLevel(config.LOGGER_LEVEL) class BaseHandler(*Base): """Base Handler for Requests. Provides basic mechanisms for all requests on Grass (such as ACL verifications). Please note: the user information cannot be accessed until the ACL has been confirmed. """ def initialize(self): """Initialization of the handler.""" self.authorized = False self.correlation_id = None self.unique_call_count = 0 @tracer.wrap() def prepare(self): """Prepare the request and check requirements. Requirements include access permissions. If the access permissions are not valid, a 403 is thrown, otherwise the request continue its way. Raises: web.HTTPError: if the validation has failed, the ACL returns a 403 (since not all endpoints requires the user to be logged in.) """ self.get_correlation_id() access_confirmed = access.confirm(self, self._options.get('access')) if not access_confirmed: reason = None if config.environment != config.PROD_ENVIRONMENT: reason = headers.GRASS_RESPONSE_UNAUTHORIZED_REASON raise web.HTTPError(403, reason=reason) self.authorized = True @property def logger(self): """Get a logger with a context. The context provides additional information (such as the correlation id) which is later """ context = dict(correlation_id=self.get_correlation_id()) return CustomAdapter(logger, context) def get_unique_call_id(self): """Get a call id. Create a call id. This call id is incremental and will be appended to the received (or created) correlation id. Returns: str: the id of the subcall. """ self.unique_call_count += 1 return str(self.unique_call_count) def get_correlation_id(self): """Get the correlation id. If the correlation id already exists, we use it. If it does not exist We create it and log its creation. Returns: str: the correlation id. """ if self.correlation_id: return self.correlation_id self.correlation_id = str(uuid.uuid1()) correlation_id = self.request.headers.get('Correlation-Id', '') if not correlation_id: message = f'Correlation-Id ({self.correlation_id}) created by Grass.' else: message = ( f'Correlation-Id ({correlation_id}) received by Grass and ignored. ' f'Correlation-Id ({self.correlation_id}) created by Grass.' ) if self.request.path not in [config.HEALTH_CHECK]: self.logger.info(message) return self.correlation_id def write_error(self, status_code, **kargs): self.clear_header('Server') super().write_error(status_code, **kargs) @tracer.wrap() def get_current_user(self): """Get the current user. Note: ALERT: This method may appear as not being called. This is an override of an internal method that tornado will call by default during it's process. The current user can only be fetched after the ACL has been performed, which happens in the preparation step. Returns: User: the User object. """ if not self.authorized: return None user_id = self.get_argument('user', default='') if not user_id: return None current_user = user.get_user_by_id(user_id) if current_user.success: return current_user.message return None @tracer.wrap() def add_request_information(self): """Provide request information. For security reasons, request information are only available in developer and QA environments. Args: request_handler (RequestHandler): the TornadoWeb request handler. """ self.clear_header('Server') self.set_header(headers.GRASS_HEADER, config.environment) if config.environment == config.PROD_ENVIRONMENT: return self.set_header( headers.GRASS_RESPONSE_AUTHORIZE_HEADER, headers.GRASS_RESPONSE_AUTHORIZE_HEADER_APPROVED, ) if self.current_user: self.set_header( headers.GRASS_RESPONSE_USER_HEADER, self.current_user.user_id ) class ServiceHandler(BaseHandler): """Service Handler. The service handler is responsible for proxying the current request and sending it to the designated service (listed in grass/services.py). The service handler provides support for all basic http verbs (get, post, delete, put, patch, head, and options.) TODO(mortali): add file support. Based on the tech design, the files are going to be sent to s3 and a token will be added to the request. """ @property def request_params(self): """Get the params for the request. Clean up the params: some of those params are reserved keywords and will be removed from the url (the service called should not know about the session id for instance, the token and the client.) Args: request (Request): the http request tied to this execution. Returns: dict: the params. """ params = urllib.parse.parse_qs(self.request.query) allow_reserved_params = getattr(self, 'allow_reserved_params', set()) reserved_params = config.RESERVED_KEYS - allow_reserved_params for reserved_param in reserved_params: params.pop(reserved_param, '') return params @property def request_path(self): """Get the request path. The current path contains the relative path to the service – which needs to be removed. So /analytics/all/?... turns into: `/all/?...`. Returns: str: the path for the request. """ return self.request.path.replace(self.rule, '', 1) @property def request_headers(self): """Get the request headers. The request headers are based off the incoming request headers (Grass is just an authentication proxy). For security reasons, Grass will remove information that conflicts with its behavior (headers such as Grass-Account-Id and Grass-Account-Type). Returns: dict: the headers of the request. """ request_headers = dicttoolz.dissoc( self.request.headers, headers.GRASS_HEADER_ACCOUNT_ID, headers.GRASS_HEADER_ACCOUNT_TYPE, headers.GRASS_HEADER_USER_ID, ) request_headers.update( {owsheaders.ORCHARD_REQUESTOR_SERVICE: config.SERVICE_NAME} ) if not self.current_user: return request_headers # set ORCHARD_IDENTITY_ID from JWT / User object. identity_from_jwt = None if ( hasattr(self.current_user, 'orchard_identity_id') and self.current_user.orchard_identity_id ): identity_from_jwt = self.current_user.orchard_identity_id request_headers.update({headers.ORCHARD_IDENTITY_ID: identity_from_jwt}) if ( hasattr(self.current_user, 'orchard_identity_uuid') and self.current_user.orchard_identity_uuid ): identity_uuid_from_jwt = self.current_user.orchard_identity_uuid request_headers.update( {headers.ORCHARD_IDENTITY_UUID: identity_uuid_from_jwt} ) if hasattr(self.current_user, 'roles') and self.current_user.roles: request_headers.update( {headers.ORCHARD_ROLES: ','.join(self.current_user.roles)} ) # set Profile Headers for OA # TODO : set profile headers from JWT if ( hasattr(self.current_user, 'profile_type') and self.current_user.profile_type and hasattr(self.current_user, 'profile_id') and self.current_user.profile_id ): request_headers.update( { headers.ORCHARD_PROFILE_TYPE: self.current_user.profile_type, headers.ORCHARD_PROFILE_ID: self.current_user.profile_id, } ) # if this is a profile user with JWT, don't remove profile headers. if self.current_user.user_group == user.UserGroups.PROFILE: client_name = self.request.headers.get(headers.APOLLOGRAPHQL_CLIENT_NAME) user_id = self.request.headers.get(headers.GRASS_HEADER_USER_ID) if ( client_name in ['orchard-suite-oa-applications', 'workstation'] and user_id ): request_headers.update({headers.GRASS_HEADER_USER_ID: user_id}) self.logger.info( f'PROFILE USER: with headers ' f'Identity: {self.current_user.orchard_identity_id} ' f'Profile: {request_headers.get(headers.ORCHARD_PROFILE_TYPE)}' f' {request_headers.get(headers.ORCHARD_PROFILE_ID)}' ) return request_headers user_header = self.current_user.user_id # If Authorization header exists we are using an Auth0 JWT # and wish for the user header to remain in tact if self.request.headers.get('Authorization'): user_header = self.request.headers.get( headers.GRASS_HEADER_USER_ID, user_header ) request_headers.update({headers.GRASS_HEADER_USER_ID: user_header}) # For users who have an account under management (such as a label, # distributor or subaccount), we provide those information in addition. if self.current_user.account_id: request_headers.update( { headers.GRASS_HEADER_ACCOUNT_ID: ( str(self.current_user.account_id) ), headers.GRASS_HEADER_ACCOUNT_TYPE: (self.current_user.account_type), } ) if identity_from_jwt: self.logger.info( f'LEGACY USER: with headers ' f'Identity: {identity_from_jwt} ' f'Profile: {request_headers.get(headers.ORCHARD_PROFILE_TYPE)}' f' {request_headers.get(headers.ORCHARD_PROFILE_ID)}' ) return request_headers async def process(self, method, *args, **kwargs): """Trigger the request. The initial request is forwarded to the service after some cleanup has been performed (for instance: removing all grass specific params and adding new headers.) Args: method (str): http verb (get, post, put, patch, delete, head, options). Returns: Future: the future that contains the request. """ start_time = time.time() * 1000 request = self.request body = None if request.body: body = request.body headers = self.request_headers cookies = request.cookies files = request.files process = getattr(self.service.protocol, method) await process( self, self.service, path=self.request_path, params=self.request_params, body=body, headers=headers, cookies=cookies, files=files, ) end_time = time.time() * 1000 if getattr(self, 'metric', None): datadog.send_timer( datadog.key(f'{self.metric}.latency'), delta_ms=end_time - start_time, logger=logger, ) delete = partialmethod(process, 'delete') get = partialmethod(process, 'get') head = partialmethod(process, 'head') # options is not supported, we always return 200 to support CORS post = partialmethod(process, 'post') put = partialmethod(process, 'put') patch = partialmethod(process, 'patch') def options(self, *args, **kwargs): """Handle We always return 200 with an empty body to support CORS preflight checks. haproxy takes care of adding the appropriate CORS headers. We want to skip hitting the microservice to check if the route exists to decreate network traffic / latency since and give the benefit of the doubt that our frontends are making good API calls. https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS#Preflighted_requests """ self.set_status(200) self.finish() def route(rule, **route_options): """A decorator to register a view function for a specific url. This route decorator introduces the ability to apply specific response format and the granular access (ACL), along with argument validators. The usage is similar to the Flask route, with some extra options: * **args** (`list`): list of all the arguments, their format, and the argument validator (field validator). * **format** (``ResponseFormat``): Representation of the response format, by default: ``format_json``. Example: Add a new endpoint ``/user/create`` only allowed for POST requests:: @api.route('/user/create', methods=['POST']) def add_user(): pass Args: rule (string): The rule that corresponds to the resource. It will be converted into a regex. options (dict): The remaining options. They are mapping the flask options for the route (e.g. endpoint). Returns: `callable`: The function decorator. """ def route_decorator(fn): # TODO(mortali): extract the route handler, similar to the Service # Handler. class RouteHandler(BaseHandler): _options = route_options metric = route_options.get('metric') @wraps(fn) def wrapper(handler, *args, **kwargs): """Wrap the handler. The wrapper orchestrate the different steps of the request in our system: perform requirements checks, call the handler, and return the response. If the flag `metric` is set, send the metrics to datadog. Args: handler (RequestHandler): the request handler. args (list): the list of arguments. kwargs (dict): the dictionary of arguments Raises: HTTPError: If an exception has been raised by using abort, this method will catch it. If the metric flag is set, we also count the number of failures. """ try: response = process(handler, *args, **kwargs) route_metric(response, metric, logger=logger) handler.set_status(response.status) handler.write(response.message) handler.add_request_information() handler.finish() except tornado.httpclient.HTTPError as e: route_metric(e.get_response(), metric, logger=logger) raise e def process(request, *args, **kwargs): """Process the request. This method processes the requests and format the response appropriately based on the response of the handler. Args: args (list): the list of arguments to pass to the handler. kwargs (dict): the dictionary of arguments to pass to the handler. Returns: Response: the response object. """ response = fn(request, *args, **kwargs) return response for method in route_options.get('methods', ['GET']): setattr(RouteHandler, method.lower(), wrapper) app.add_handlers('.*$', [web.url(rule, RouteHandler)]) return wrapper return route_decorator def register(name, rule, metric, service, access, allow_reserved_params=None): """Register services to the application. Args: name (str): the name of the service. rule (str): the rule of the service. metric (str): gather information about the service (perf, req counter) service (Service): the service information. access (list): the access list. allowed_reserved_keys (list): list of all reserved keys that are allowed to be provided to the service. """ handler_name = name handler_rule = rule handler_service = service handler_metric = metric handler_allow_reserved_params = allow_reserved_params or set() if handler_allow_reserved_params: assert isinstance( handler_allow_reserved_params, set ), 'The field "allow_reserved_params" must be a set.' class Handler(ServiceHandler): name = handler_name _options = dict(access=access) metric = handler_metric rule = handler_rule service = handler_service allow_reserved_params = handler_allow_reserved_params app.add_handlers('.*$', [web.url(f'{rule}(.*)', Handler)]) def route_metric(response, key, count=1, logger=None): """Special metric handler for response. The response metrics need to be split by status (so we know how many times an endpoint returns a 200, 403, 404, etc.) Args: response (Response): the response object. key (str): the key. count (int): the count to send. """ if not response or not key: return metric_key = datadog.key(f'handler.{key}.{response.status}') datadog.publish_metric(metric_key, count=count, logger=logger) def run(debug=False): """Run the application. Args: debug (bool): If the application needs to be (or not) in debug mode. """ app.debug = debug app.run()