"""PDP authorization utilities for project_transfer endpoints (PORT-69).""" import os from functools import wraps from typing import Optional import ddtrace from flask import g from oto import response from oto.adaptors.flask import flaskify from python_pdp_sdk.resource_getters import base from project_manager.api import authorization_backend from project_manager.constant import error_const from project_manager.constant import http_status_codes @ddtrace.tracer.wrap() def pdp_authorize_resource( resource_type: str, action: str = 'view', resource_id: Optional[int] = None, **attributes, ) -> bool: """Authorize a resource, forwarding any extra kwargs as resource attributes. Extra keyword arguments (e.g. ``tenant``, ``id_to_uuid_exchange_tenant``) are forwarded as resource attributes via ``ForwardKwargsGetter``. Args: resource_type: PDP resource type (e.g. ``project_transfer``). action: PDP action (e.g. ``view``, ``create``, ``execute_batch``). resource_id: Optional PDP resource id. When ``None`` the underlying SDK is called with ``0`` so policies that do not consult the resource id (e.g. ``execute_batch``) still get a valid call. **attributes: Extra resource attributes to forward to PDP. Returns: True when PDP authorizes the request, False otherwise. """ if os.getenv('PDP_DISABLED', '').lower() == 'true': return True # ForwardKwargsGetter passes the extra kwargs (e.g. tenant, # id_to_uuid_exchange_tenant) through to PDP as resource attributes, # instead of looking them up via a getter callable. sdk_resource_id = resource_id if resource_id is not None else 0 authorized = authorization_backend.is_authorized( action=action, resource_id=sdk_resource_id, resource_type=resource_type, resource_getter=base.ForwardKwargsGetter(), **attributes, ) if not authorized: log = getattr(g, 'log', None) if log is not None: log.warning( error_const.ERROR_MESSAGE_FORBIDDEN_USER, extra={ 'resource_id': resource_id, 'resource_type': resource_type, 'action': action, }, ) return False return True def pdp_authorize_project_transfer( action: str, job_id: Optional[int] = None, originating_vendor_id: Optional[int] = None, originating_subaccount_id: Optional[int] = None, ) -> bool: """Authorize a project_transfer action. Calls PDP with ``resource_type='project_transfer'``, ``resource_id=job_id`` and tenant attributes derived from the originating account. When ``originating_subaccount_id`` is set the tenant is modeled as a ``subaccount``; otherwise it is modeled as an ``account`` (vendor). When neither is set the call is made with no tenant attributes; the policy can still authorize via derived roles that do not require a tenant (e.g. ``execute_batch`` via the any-tenant transfer_operator). Args: action: one of ``view``, ``create``, ``execute_batch``. job_id: id of the project transfer job (the PDP resource id). Omit for actions that do not target a specific job (e.g. listing, ``execute_batch``). originating_vendor_id: id of the originating vendor. originating_subaccount_id: id of the originating subaccount, when the transfer originates from a subaccount rather than a vendor. Returns: True when PDP authorizes the request, False otherwise. """ attributes = {} if originating_subaccount_id is not None: attributes['tenant'] = {'tenant_type': 'subaccount'} attributes['id_to_uuid_exchange_tenant'] = { 'tenant_type': 'subaccount', 'tenant_id': originating_subaccount_id, } elif originating_vendor_id is not None: attributes['tenant'] = {'tenant_type': 'account'} attributes['id_to_uuid_exchange_tenant'] = { 'tenant_type': 'account', 'tenant_id': originating_vendor_id, } return pdp_authorize_resource( resource_id=job_id, resource_type='project_transfer', action=action, **attributes, ) def _forbidden_response(): """Build the 403 response used when PDP denies a project_transfer action.""" return response.create_error_response( code=error_const.ERROR_CODE_AUTHORIZATION, message=error_const.ERROR_MESSAGE_FORBIDDEN_USER, status=http_status_codes.FORBIDDEN) def _jwt_identity_or_error(): """Return (identity, None) or (None, error_response). Reads the JWT identity UUID from the request context. Falls back to identity_id when jwt_identity_id is absent (e.g. local dev where the upstream service does not forward a parseable JWT). The transfer-job tables store identity as CHAR(36), so a missing identity is rejected as 401 rather than silently falling back to a placeholder. """ context = getattr(g, 'request_context', None) identity = ( getattr(context, 'jwt_identity_id', None) or getattr(context, 'identity_id', None) ) if context else None if not identity: return None, response.create_error_response( code=error_const.ERROR_CODE_AUTHORIZATION, message='Missing or invalid JWT identity.', status=http_status_codes.UNAUTHORIZED) return identity, None def authorize_transfer_job( action: str, job_scoped: bool = True, defer_pdp: bool = False): """Decorator that authorizes a project_transfer action. Handles, in order: * JWT identity check (rejects missing identity with 401). * Job fetch via ``project_transfer.fetch_job_for_auth`` (when ``job_scoped`` is true). * PDP authorization (unless ``defer_pdp`` is true): for ``job_scoped`` routes, the job's originating account is passed as tenant; for list/batch routes, PDP is called without a tenant so any-tenant derived roles apply. The wrapped handler receives: * ``identity`` kwarg always (resolved JWT identity string). * ``job`` kwarg when ``job_scoped`` is true (the resolved job dict). Args: action: PDP action to authorize. job_scoped: True when the route targets a specific job (URL has ````). False for list/batch endpoints. defer_pdp: True to skip the PDP call inside the decorator. The handler is responsible for calling ``pdp_authorize_project_transfer`` once it has resolved the tenant (e.g. ``create_transfer_job`` reads originating_vendor_id from the request body). """ # Local import to avoid a circular import: project_transfer's tests # import this module, and we want to keep imports lazy at decoration # time so handlers.py's import order is unaffected. from project_manager.logic import project_transfer def decorator(handler): @wraps(handler) def wrapper(*args, **kwargs): identity, identity_err = _jwt_identity_or_error() if identity_err is not None: return flaskify(identity_err) if job_scoped: job_id = kwargs.get('job_id') job = project_transfer.fetch_job_for_auth(job_id) if isinstance(job, response.Response): return flaskify(job) if not defer_pdp and not pdp_authorize_project_transfer( action=action, job_id=job_id, originating_vendor_id=job['originating_vendor_id'], originating_subaccount_id=job.get('originating_subaccount_id')): return flaskify(_forbidden_response()) kwargs['job'] = job else: if not defer_pdp: if not pdp_authorize_project_transfer(action=action): return flaskify(_forbidden_response()) kwargs['identity'] = identity return handler(*args, **kwargs) return wrapper return decorator