"""PDP authorization utilities.""" from ddtrace import tracer from flask import g from python_pdp_sdk import ( ForwardKwargsGetter, ) from python_pdp_sdk.backends.exceptions import AttributesException from ledger.config import authorization_backend from ledger.constants import error @tracer.wrap() def pdp_authorize_resource( account_id: int | str, resource_type: str = 'account', action: str = 'view_abacus_account_info', tenant_type: str = 'account', ) -> bool: """Authorize a resource using PDP.""" g.log.debug( 'PP authorize account', resources={ 'identity_id': g.request_context.jwt_identity_id, 'account_id': account_id, }, ) try: auth_response = authorization_backend.is_authorized( action, account_id, resource_type, ForwardKwargsGetter(), tenant={ 'tenant_type': tenant_type, }, id_to_uuid_exchange_tenant={ 'tenant_type': tenant_type, 'tenant_id': account_id, }, ) except AttributesException as e: g.log.warn( f'PP raised an AttributesException: {e}', resources={ 'identity_id': g.request_context.jwt_identity_id, 'account_id': account_id, }, ) return False if not auth_response: g.log.warn( error.ERROR_CODE_UNAUTHORIZED_ACCOUNT, resources={ 'identity_id': g.request_context.jwt_identity_id, 'account_id': account_id, 'auth_response': auth_response, }, ) return auth_response