"""Application. The API application is a `flask` application. It provides simple features such as registering a url for a specific handlers. """ from flask import Flask, request from owsclient import OwsClient from owslogger import flask_logger from owslogger import logger from owsrequest import flask_request from owsrequest.ows_client import correlation_id_getter, request_context_getter from python_pdp_sdk import ( AuthorizationBackend, MigrationAuthorizationBackend, OwsPdpClient, PdpAuthorizationBackend, ) from product import config from product import routing_converters app = Flask(config.SERVICE_NAME) # append extra route converters for e.g. to handle comma separated int app.url_map.converters.update(routing_converters.get_routing_converters()) flask_logger.setup( app, config.ENVIRONMENT, config.LOGGER_NAME, config.LOGGER_LEVEL, config.SERVICE_NAME, config.SERVICE_VERSION, exclude_paths=[config.HEALTH_CHECK], clear_handlers=False) logger.setup( config.ENVIRONMENT, 'owsrequest', config.LOGGER_LEVEL, config.SERVICE_NAME, config.SERVICE_VERSION, clear_handlers=True) flask_request.setup(app, config.ENVIRONMENT, add_request_context=True, label_profile=True, # disabling verify_access until we are # ready to flip access_log_only=False verify_access=False, rules_file='product/access_rules.yml', access_log_only=config.ONLY_LOG_ACCESS_ERRORS, exclude_paths=config.EXCLUDE_PATH, uwsgi_cache_enabled=True) def setup_ows_client() -> OwsClient: """Create and return a configured OwsClient instance.""" return OwsClient( environment=config.ENVIRONMENT, service_name=config.SERVICE_NAME, correlation_id_getter=correlation_id_getter, request_context_getter=request_context_getter, ) def setup_authorization_backend(ows_client: OwsClient) -> AuthorizationBackend: """Create and return a PDP-backed AuthorizationBackend instance.""" return PdpAuthorizationBackend(OwsPdpClient(ows_client)) def request_tags() -> list[str]: """Datadog tags describing the active request, for use as extra_tags_getter. Only invoked by MigrationAuthorizationBackend at metric-emit time, i.e. inside a Flask request context, so accessing `request` here is safe. """ return [ f"method:{request.method}", f"endpoint:{request.url_rule or request.path}", f"has_authorization_header:{str(bool(request.headers.get('Authorization'))).lower()}", f"profile_type:{(request.headers.get('Orchard-Profile-Type') or 'none').lower()}", ] def setup_migration_authorization_backend( authorization_backend: AuthorizationBackend, ) -> MigrationAuthorizationBackend: """Wrap an AuthorizationBackend in a shadow-mode MigrationAuthorizationBackend. During the PP auth migration rollout this backend always permits requests while shadowing the inner backend, emitting a ``pp_auth.rollout.would_deny`` Datadog metric (tagged via `request_tags`) whenever the inner backend would have denied. This lets us measure PP readiness per endpoint before flipping enforcement on. Metric emission is gated on `config.PP_MIGRATION_METRICS_ENABLED` (QA/prod only). """ return MigrationAuthorizationBackend( inner_backend=authorization_backend, service_name=config.SERVICE_NAME, environment=config.ENVIRONMENT, extra_tags_getter=request_tags, metrics_enabled=config.PP_MIGRATION_METRICS_ENABLED, ) ows_client: OwsClient = setup_ows_client() authorization_backend: AuthorizationBackend = setup_authorization_backend(ows_client) migration_authorization_backend: MigrationAuthorizationBackend = ( setup_migration_authorization_backend(authorization_backend) )