"""MigrationAuthorizationBackend for shadow-mode PP auth migration rollout.""" from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable from python_pdp_sdk.backends.authorization_backend import ( AuthorizationBackend, ResourceAction, ResourceWithAttributes, ) from python_pdp_sdk.backends.exceptions import UnauthenticatedException from python_pdp_sdk.connectors.ows_pdp.models.allowed_tenant import AllowedTenant from python_pdp_sdk.connectors.secretsmanager import DatadogSecretsManager from python_pdp_sdk.logging_adapter import get_logger from python_pdp_sdk.protocols import ResourceGetter try: import datadog _DATADOG_AVAILABLE = True except ImportError: datadog = None # type: ignore[assignment] _DATADOG_AVAILABLE = False logger = get_logger(__name__) @dataclass(frozen=True) class WouldDenyMetadata: """Structured info passed to ``on_would_deny`` callbacks. Mirrors the tags emitted with the ``pp_auth.rollout.would_deny`` Datadog metric. """ action: str resource_type: str reason: str environment: str service_name: str extra_tags: tuple[str, ...] = () def as_datadog_tags(self) -> list[str]: """Return this metadata as the flat ``key:value`` tag list sent to Datadog.""" return [ f"environment:{self.environment}", f"service_name:{self.service_name}", f"action:{self.action}", f"resource_type:{self.resource_type}", f"reason:{self.reason}", *self.extra_tags, ] class MigrationAuthorizationBackend(AuthorizationBackend): """Wrapper backend that always permits requests but shadows the inner backend. Used during PP auth migration rollout. Delegates every authorization check to an inner ``AuthorizationBackend`` without raising to the caller. When the inner backend would deny, emit a ``pp_auth.rollout.would_deny`` Datadog metric so teams can measure readiness before switching enforcement on. Note: ``get_authorized_tenants`` is not covered by the always-allow guarantee. On ``UnauthenticatedException`` or any other exception it emits a metric and returns ``[]`` to keep the service running rather than propagating the error. """ def __init__( self, inner_backend: AuthorizationBackend, service_name: str, environment: str, dd_api_key: str | None = None, *, extra_tags_getter: Callable[[], list[str]] | None = None, metric_name: str = "pp_auth.rollout.would_deny", metrics_enabled: bool = True, ) -> None: """Create a MigrationAuthorizationBackend. Args: inner_backend: Authorization backend to shadow. service_name: Value for the ``service_name`` metric tag. environment: Value for the ``environment`` metric tag. dd_api_key: Datadog API key. Metrics are disabled when absent. extra_tags_getter: Optional callable returning additional metric tags. Exceptions from this callable are swallowed; the metric is still sent with the base tags. metric_name: Datadog metric name to emit on would-deny events. metrics_enabled: Set to ``False`` to explicitly disable Datadog metrics, skipping DD API key resolution and ``datadog.initialize`` entirely. Enabled by default. """ self._inner = inner_backend self._service_name = service_name self._environment = environment self._extra_tags_getter = extra_tags_getter self._metric_name = metric_name self._metrics_enabled = metrics_enabled and self._try_enable_metrics(dd_api_key) def _try_enable_metrics( self, dd_api_key: str | None, ) -> bool: """Resolve DD credentials, initialize Datadog, and return whether metrics are enabled.""" api_key = self._use_provided_dd_api_key_or_fetch(dd_api_key) if not api_key: logger.warning( "Datadog API key not provided; metrics disabled for MigrationAuthorizationBackend." ) return False if not _DATADOG_AVAILABLE: logger.warning( "datadog package is not installed; metrics disabled for " "MigrationAuthorizationBackend. " "Install with: pip install 'python-pdp-sdk[migration]'" ) return False try: datadog.initialize(api_key=api_key) return True except Exception: logger.warning( "Failed to initialize Datadog; metrics disabled for MigrationAuthorizationBackend.", exc_info=True, ) return False def _use_provided_dd_api_key_or_fetch( self, dd_api_key: str | None = None, ) -> str | None: """Return the DD API key from caller-supplied value, falling back to Secrets Manager. Never raises — AWS / boto3 failures are logged as warnings and result in ``None`` so that metrics are simply disabled. """ if dd_api_key: logger.info( "[MigrationAuthorizationBackend] Using user provided `dd_api_key`" ) return dd_api_key try: secrets_client = DatadogSecretsManager(environment=self._environment) dd_api_key = secrets_client.get_secret("DD_API_KEY") logger.info( "[MigrationAuthorizationBackend] Using SecretsManager provided `dd_api_key`" ) except Exception: logger.warning( "Failed to fetch Datadog API key from Secrets Manager; " "metrics disabled for MigrationAuthorizationBackend.", exc_info=True, ) return dd_api_key def _call_would_deny( self, on_would_deny: Callable[[WouldDenyMetadata], None] | None, metadata: WouldDenyMetadata, ) -> None: """Invoke ``on_would_deny`` with the deny metadata, swallowing any exceptions.""" if not on_would_deny: return try: on_would_deny(metadata) except Exception: logger.debug("on_would_deny callback raised; ignoring.", exc_info=True) def _build_would_deny_metadata( self, action: str, resource_type: str, reason: str ) -> WouldDenyMetadata: """Build the ``WouldDenyMetadata`` shared by the Datadog metric and ``on_would_deny``.""" extra_tags: tuple[str, ...] = () if self._extra_tags_getter: try: extra_tags = tuple(self._extra_tags_getter()) except Exception: logger.debug( "extra_tags_getter raised; omitting extra tags from " "would-deny metadata.", exc_info=True, ) return WouldDenyMetadata( action=action, resource_type=resource_type, reason=reason, environment=self._environment, service_name=self._service_name, extra_tags=extra_tags, ) def _emit(self, metadata: WouldDenyMetadata) -> None: """Emit a would-deny metric to Datadog. Swallows all exceptions so metric failures never reach the caller. """ if not self._metrics_enabled: return try: datadog.api.Metric.send( # type: ignore[attr-defined, no-untyped-call] metric=self._metric_name, points=1, tags=metadata.as_datadog_tags() ) except Exception: logger.debug("Failed to emit Datadog metric.", exc_info=True) def is_authorized( self, action: str, resource_id: int | str, resource_type: str, resource_getter: ResourceGetter, raise_when_unauthorized: bool = False, *args: Any, on_would_deny: Callable[[WouldDenyMetadata], None] | None = None, **kwargs: Any, ) -> bool: """Return True; emit metric if inner backend would deny. Args: action: Operation being authorized. resource_id: Resource identifier. resource_type: Resource type being authorized. resource_getter: Used with args and kwargs to fetch resource attributes. raise_when_unauthorized: Accepted for interface compatibility; never raises. *args: Forwarded to inner backend (and on to resource_getter). on_would_deny: Optional callback invoked with a ``WouldDenyMetadata`` (mirroring the Datadog metric tags) whenever the inner backend would deny the request. Not forwarded to the inner backend. Exceptions raised by the callback are logged and swallowed. **kwargs: Forwarded to inner backend (and on to resource_getter). """ try: authorized = self._inner.is_authorized( action, resource_id, resource_type, resource_getter, False, *args, **kwargs, ) except UnauthenticatedException: metadata = self._build_would_deny_metadata( action, resource_type, "unauthenticated" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return True except Exception: metadata = self._build_would_deny_metadata( action, resource_type, "exception" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return True if not authorized: metadata = self._build_would_deny_metadata( action, resource_type, "pp_denied" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return True def is_authorized_many( self, action: str, resource_type: str, resources_with_attributes: list[ResourceWithAttributes], raise_when_unauthorized: bool = False, *, on_would_deny: Callable[[WouldDenyMetadata], None] | None = None, ) -> list[bool]: """Return all-True; emit one metric per resource that would be denied. Args: action: Operation being authorized. resource_type: Resource type being authorized. resources_with_attributes: Resources to check. raise_when_unauthorized: Accepted for interface compatibility; never raises. on_would_deny: Optional callback invoked once per would-deny event with a ``WouldDenyMetadata`` (mirroring the Datadog metric tags). Exceptions raised by the callback are logged and swallowed. """ try: results = self._inner.is_authorized_many( action, resource_type, resources_with_attributes, raise_when_unauthorized=False, ) except UnauthenticatedException: for _ in resources_with_attributes: metadata = self._build_would_deny_metadata( action, resource_type, "unauthenticated" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return [True] * len(resources_with_attributes) except Exception: for _ in resources_with_attributes: metadata = self._build_would_deny_metadata( action, resource_type, "exception" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return [True] * len(resources_with_attributes) for authorized in results: if not authorized: metadata = self._build_would_deny_metadata( action, resource_type, "pp_denied" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return [True] * len(resources_with_attributes) def is_authorized_many_resources_and_actions( self, resource_actions: list[ResourceAction], raise_when_unauthorized: bool = False, *, on_would_deny: Callable[[WouldDenyMetadata], None] | None = None, ) -> list[bool]: """Return all-True; emit per-resource-action metric when inner would deny. Args: resource_actions: Resource-action pairs to check. raise_when_unauthorized: Accepted for interface compatibility; never raises. on_would_deny: Optional callback invoked once per would-deny event with a ``WouldDenyMetadata`` (mirroring the Datadog metric tags). Exceptions raised by the callback are logged and swallowed. """ try: results = self._inner.is_authorized_many_resources_and_actions( resource_actions, raise_when_unauthorized=False, ) except UnauthenticatedException: for ra in resource_actions: metadata = self._build_would_deny_metadata( ra.action, ra.resource_type, "unauthenticated" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return [True] * len(resource_actions) except Exception: for ra in resource_actions: metadata = self._build_would_deny_metadata( ra.action, ra.resource_type, "exception" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) return [True] * len(resource_actions) if len(resource_actions) == len(results): for ra, authorized in zip(resource_actions, results): if not authorized: metadata = self._build_would_deny_metadata( ra.action, ra.resource_type, "pp_denied" ) self._emit(metadata) self._call_would_deny(on_would_deny, metadata) else: logger.warning( "MIGRATION_BACKEND_ERROR: Unexpected result %s resource_actions but SDK returned %s results", len(resource_actions), len(results), ) return [True] * len(resource_actions) def get_authorized_tenants( self, action: str, resource_type: str, ) -> list[AllowedTenant]: """Delegate to inner backend; emits metric and returns allowed_tenants when no error occurs. Args: action: Operation being authorized. resource_type: Resource type being authorized. On error, emits a metric and returns ``[]`` rather than propagating the exception, keeping the service running in a degraded state during rollout. """ try: return self._inner.get_authorized_tenants(action, resource_type) except UnauthenticatedException: self._emit( self._build_would_deny_metadata( action, resource_type, "unauthenticated" ) ) return [] except Exception: self._emit( self._build_would_deny_metadata(action, resource_type, "exception") ) return []