"""Logic for Identities.""" import datetime import logging from typing import Dict, List, Optional from uuid import UUID from cerbos.sdk.client import AsyncCerbosClient from ddtrace.trace import tracer from fastapi import HTTPException from pdp.connectors.dynamo import DynamoDbConnector from pdp.connectors.features import SplitioClient from pdp.connectors.ows_account import OwsAccountClient from pdp.connectors.ows_participant import OwsParticipantClient from pdp.connectors.redis_client import RedisConnector from pdp.constants.constants import AuthEffect, TenantType from pdp.constants.roles import ROLES_PARENT_CHILD_RELATIONSHIPS, ROLES_SUPPORTED from pdp.fastapi.schemas.check_resource_type_actions import ( CheckResourceTypeActionResult, CheckResourceTypeActionsRequest, CheckResourceTypeActionsResponse, CreateCheckResourceActionsError, ResourceTypeActionAdapter, ) from pdp.fastapi.schemas.check_resources import CheckResourcesRequest from pdp.fastapi.schemas.deactivation import DeactivationSummary from pdp.fastapi.schemas.get_allowed_tenants import ( AllowedTenant, GetAllowedTenantsRequest, GetAllowedTenantsResponse, ) from pdp.fastapi.schemas.identity import ( IdentityTenant, Role, RolesResponse, TenantRoles, ) from pdp.fastapi.schemas.principal import Principal from pdp.logic import cerbos from pdp.models.identity import Identity from pdp.utils.dynamo import deserialize_dynamo_item_array from pdp.utils.resource_type_actions import ( _build_resource_type_action_results_list, ) from pdp.utils.roles import RolesListHelper, is_new_list_different logger = logging.getLogger(__name__) class AttachDetachRolesError(Exception): """Error class for attach and detach roles.""" pass @tracer.wrap() def is_tenant_assigned( identity_uuid: str, tenant_uuid: UUID, identity_ddb_connector: DynamoDbConnector, ) -> bool: """Check an identity has any/some permissions for a single tenant.""" identity = Identity(identity_uuid, identity_ddb_connector) tenant_state = identity.get_permissions_by_tenant(tenant_uuid) return bool(tenant_state) @tracer.wrap() async def attach_and_detach_roles( identity_uuid: str, tenant_uuid: UUID, tenant_type: TenantType, roles_to_attach: List[Role], roles_to_detach: List[Role], authenticated_identity_uuid: str, identity_ddb_connector: DynamoDbConnector, principal: Principal | None = None, ) -> IdentityTenant: """Upsert an identity's permissions for a single tenant.""" identity = Identity(identity_uuid, identity_ddb_connector) tenant_state = identity.get_permissions_by_tenant(tenant_uuid) date_time = datetime.datetime.now(datetime.timezone.utc).isoformat() if not tenant_state: # The identity does not have any permissions set for this tenant # Create an empty tenant_state if principal: tenant_state = identity.get_empty_tenant_permissions_state( tenant_uuid=tenant_uuid, tenant_type=tenant_type, created_at=date_time, created_by=str(principal.identity_uuid), created_impersonated_by=str(principal.impersonated_by_identity_uuid) if principal.impersonated_by_identity_uuid else None, ) else: tenant_state = identity.get_empty_tenant_permissions_state( tenant_uuid=tenant_uuid, tenant_type=tenant_type, created_at=date_time, created_by=str(authenticated_identity_uuid), ) if tenant_type != tenant_state.tenant_type: message = f"Tenant_type does not match. Identity: {identity_uuid}, Tenant: {tenant_uuid}, Given tenant_type: {tenant_type.value}, Existing tenant_type: {tenant_state.tenant_type}" # noqa: E501 logger.warning(message) raise HTTPException(status_code=400, detail=message) roles_to_detach = _extend_with_child_roles_to_detach(roles_to_detach) roles_to_attach = _keep_supported_roles(roles_to_attach) existing_roles = tenant_state.roles # Determine the final desired set of roles roles_after_attach_and_detach = ( RolesListHelper(existing_roles) .attach(roles_to_attach) .detach(roles_to_detach) .get_roles_list() ) is_new_tenant_state = tenant_state.version == "0" final_roles_list_is_empty = len(roles_after_attach_and_detach) == 0 # If there are no roles left and it's a new tenant state, do nothing if final_roles_list_is_empty and is_new_tenant_state: logger.warning( f"Not creating identity/tenant pair identity: {identity_uuid} and tenant: {tenant_uuid} as there are no roles being added." # noqa: E501 ) return tenant_state # Else if there are no roles left and it's NOT a new tenant state, # delete the item elif final_roles_list_is_empty and not is_new_tenant_state: return await _deactivate_row_with_empty_roles_list( identity_uuid=identity_uuid, identity_ddb_connector=identity_ddb_connector, authenticated_identity_uuid=authenticated_identity_uuid, identity=identity, tenant_state=tenant_state, principal=principal, ) # Check if the new roles are different, and apply if is_new_list_different(existing_roles, roles_after_attach_and_detach): tenant_state.roles = roles_after_attach_and_detach if principal: return _commit_tenant_state( identity=identity, tenant_state=tenant_state, updated_by=str(principal.identity_uuid), updated_impersonated_by=str(principal.impersonated_by_identity_uuid) if principal.impersonated_by_identity_uuid else None, updated_at=date_time, ) return _commit_tenant_state( identity=identity, tenant_state=tenant_state, updated_by=authenticated_identity_uuid, updated_at=date_time, ) # Nothing to change, just return. return tenant_state async def _deactivate_row_with_empty_roles_list( identity_uuid: str, identity_ddb_connector: DynamoDbConnector, authenticated_identity_uuid: str, identity: Identity, tenant_state: IdentityTenant, principal: Principal | None = None, ) -> IdentityTenant: """Deactivate a tenant row if it has an empty roles list.""" tenant_uuid = tenant_state.tenant_uuid tenant_type = tenant_state.tenant_type logger.info( f"Deactivating tenant {tenant_uuid} for identity {identity_uuid} as it has no roles." # noqa: E501 ) await deactivate_many( identity_uuid=identity_uuid, tenants={ tenant_uuid: TenantRoles( roles=[], tenant_type=tenant_type, tenant_uuid=tenant_uuid ) }, identity_ddb_connector=identity_ddb_connector, authenticated_identity_uuid=str(authenticated_identity_uuid), principal=principal, ) logger.info( f"Deactivated identity/tenant pair identity: {identity_uuid} and tenant: {tenant_uuid} as there are no roles left." # noqa: E501 ) created_at = ( tenant_state.created_at or datetime.datetime.now(datetime.timezone.utc).isoformat() ) if principal: return identity.get_empty_tenant_permissions_state( tenant_uuid=tenant_uuid, tenant_type=tenant_type, created_at=created_at, created_by=str(principal.identity_uuid), created_impersonated_by=str(principal.impersonated_by_identity_uuid) if principal.impersonated_by_identity_uuid else None, ) return identity.get_empty_tenant_permissions_state( tenant_uuid=tenant_uuid, tenant_type=tenant_type, created_at=created_at, created_by=str(authenticated_identity_uuid), ) @tracer.wrap() def _commit_tenant_state( identity: Identity, tenant_state: IdentityTenant, updated_by: str, updated_at: str, updated_impersonated_by: str | None = None, ) -> IdentityTenant: """Update IdentityTenant metadata fields and commit change.""" tenant_state.increment_version() tenant_state.updated_at = updated_at tenant_state.updated_by = updated_by tenant_state.updated_impersonated_by = updated_impersonated_by return identity.update_tenant_permissions(tenant_state.tenant_uuid, tenant_state) @tracer.wrap() def get_roles_by_identity( identity_uuid: str, identity_ddb_connector: DynamoDbConnector, cursor: Optional[str] = None, ) -> RolesResponse: """Get paginated list of roles from DynamoDB for an identity_uuid. Args: identity_uuid (str): UUID to match identity_ddb_connector (DynamoDbConnector): dynamodb connection cursor (str): Pagination cursor Returns: Response (Dict): Response object containing list of roles and pagination cursor """ identity = Identity(identity_uuid, identity_ddb_connector=identity_ddb_connector) try: result = identity.get_roles(cursor=cursor) tenant_roles = result.pop("items") tenant_list = deserialize_dynamo_item_array(tenant_roles) result["tenants"] = { tenant["tenant_uuid"]: TenantRoles.model_validate(tenant) for tenant in tenant_list } except Exception: logger.error("Unable to get roles from DynamoDB.") # default_error_handler will log the trace raise return RolesResponse.model_validate(result) def _extend_with_child_roles_to_detach(roles_to_detach: List[Role]) -> List[Role]: """ Extend the roles to detach when a base/parent role is being detached. For example, `fansifter_can_view_fan_data` is a base role. If it is detached, every other `fansifter_` role should be detached. """ if not roles_to_detach: return [] child_roles_to_detach: List[str] = [] for role in roles_to_detach: if role.role in ROLES_PARENT_CHILD_RELATIONSHIPS: child_roles_to_detach.extend(ROLES_PARENT_CHILD_RELATIONSHIPS[role.role]) roles_to_detach.extend([Role(role=role) for role in child_roles_to_detach]) return roles_to_detach def _keep_supported_roles(roles: List[Role]) -> List[Role]: """Return/keep the supported roles, discard anything else""" if not roles: return [] supported_roles = list(filter(lambda x: x.role in ROLES_SUPPORTED, roles)) if len(supported_roles) != len(roles): logger.warning( f"Unsupported roles found in the list, discarding them: {[role.role for role in roles if role.role not in ROLES_SUPPORTED]}" # noqa: E501 ) return supported_roles @tracer.wrap() async def check_resource_type_actions( request: CheckResourceTypeActionsRequest, identity_uuid: str, pdp_tenant_roles: Dict[UUID, TenantRoles], cerbos_client: AsyncCerbosClient, ows_account_client: OwsAccountClient, ows_participant_client: OwsParticipantClient, redis_connector: RedisConnector, splitio_client: SplitioClient = None, authenticated_identity_uuid: Optional[UUID] = None, principal: Optional[Principal] = None, ) -> CheckResourceTypeActionsResponse: """Check resource type actions. Args: request (CheckResourceTypeActionsRequest): Request object identity_uuid (str): UUID of identity pdp_tenant_roles (dict): Dict with identity's tenant roles cerbos_client (AsyncCerbosClient): Cerbos client Returns: Response (CheckResourceTypeActionsResponse): Response object """ if not request.resource_type_actions: return CheckResourceTypeActionsResponse(resource_type_actions=[]) try: check_resource_action_list = request.build_check_resource_actions( pdp_tenant_roles, identity_uuid=identity_uuid, ) # Build a dict of resource_type, action tuples with default AuthEffect.DENY check_resource_action_result_dict: Dict[tuple[str, str], AuthEffect] = {} for item in check_resource_action_list: check_resource_action_result_dict[ (item.resource.resource_type, item.action) ] = AuthEffect.AUTH_EFFECT_DENY check_response = await cerbos.check_resources( str(identity_uuid), check_resources_request=CheckResourcesRequest( resources=check_resource_action_list ), pdp_tenant_roles=pdp_tenant_roles, cerbos_client=cerbos_client, ows_account_client=ows_account_client, ows_participant_client=ows_participant_client, redis_connector=redis_connector, authenticated_identity_uuid=authenticated_identity_uuid, principal=principal, splitio_client=splitio_client, ) resource_type_action_responses = _build_resource_type_action_results_list( check_resource_action_result_dict, check_response ) except CreateCheckResourceActionsError as ex: # Log when this error occurs, so we can investigate logger.error( "create_check_resource_actions error: '%s'", ex, extra={ "pdp": { "resource_type_actions": ResourceTypeActionAdapter.dump_json( # noqa: E501 request.resource_type_actions ), "identity_uuid": str(identity_uuid), "authenticated_identity_uuid": str(authenticated_identity_uuid), # noqa: E501 }, }, ) # Return a 200 status code with DENYs after logging the error. return CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( action=resource_type_action.action, resource_type=resource_type_action.resource_type, effect=AuthEffect.AUTH_EFFECT_DENY, ) for resource_type_action in request.resource_type_actions ] ) except Exception: logger.error("Unable to check resource type actions.") # default_error_handler will log the trace raise return CheckResourceTypeActionsResponse( resource_type_actions=resource_type_action_responses ) @tracer.wrap() async def get_allowed_tenants( request: GetAllowedTenantsRequest, identity_uuid: str, pdp_tenant_roles: Dict[UUID, TenantRoles], cerbos_client: AsyncCerbosClient, ows_account_client: OwsAccountClient, ows_participant_client: OwsParticipantClient, redis_connector: RedisConnector, authenticated_identity_uuid: Optional[UUID] = None, splitio_client: SplitioClient = None, principal: Optional[Principal] = None, ) -> GetAllowedTenantsResponse: """Get allowed tenants for resource type - action. Args: request: Request object identity_uuid: UUID of identity pdp_tenant_roles: Dict with identity's tenant roles cerbos_client: Cerbos client connection ows_account_client: OwsAccount client connection ows_participant_client: OwsParticipant client connection redis_connector: Redis client connection authenticated_identity_uuid: Used for identity-based feature flagging behavior """ try: check_resource_action_list = request.create_check_resource_actions( pdp_tenant_roles, # type: ignore[arg-type] ) check_response = await cerbos.check_resources( str(identity_uuid), check_resources_request=CheckResourcesRequest( resources=check_resource_action_list ), pdp_tenant_roles=pdp_tenant_roles, cerbos_client=cerbos_client, ows_account_client=ows_account_client, ows_participant_client=ows_participant_client, redis_connector=redis_connector, authenticated_identity_uuid=authenticated_identity_uuid, include_resource_attributes=True, # Need attr w/Tenant for filtering principal=principal, splitio_client=splitio_client, ) except CreateCheckResourceActionsError as ex: # create_check_resource_actions will raise this error # when the identity has 0 pdp_tenant_roles. # # This isn't an error, so log it at INFO level. logger.info( "create_check_resource_actions error: '%s'", ex, extra={ "pdp": { "resource_type_actions": request.model_dump(), "identity_uuid": str(identity_uuid), "authenticated_identity_uuid": str(authenticated_identity_uuid), # noqa: E501 "function": "get_allowed_tenants", }, }, ) return GetAllowedTenantsResponse( action=request.action, resource_type=request.resource_type, tenants=[], ) except Exception: logger.error("Unable to check resources for getting allowed tenants.") raise filtered_tenants = check_response.filter_for_allowed_tenants() return GetAllowedTenantsResponse( action=request.action, resource_type=request.resource_type, tenants=[AllowedTenant(**tenant.model_dump()) for tenant in filtered_tenants], ) @tracer.wrap() async def deactivate_many( identity_uuid: str, tenants: Dict[UUID, TenantRoles], identity_ddb_connector: DynamoDbConnector, authenticated_identity_uuid: str, principal: Principal | None = None, ) -> DeactivationSummary: """Deactivate access to multiple tenants from an Identity.""" identity = Identity(identity_uuid, identity_ddb_connector) unprocessed_items = await identity.deactivate_many(tenants) # This method returns the unprocessed tombstone records, but # we're ignoring them. Each failure will log an error and # generate a sentry alert for eng intervention. _ = await identity.write_tombstone_records_for_deactivations( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=tenants, unprocessed_items=unprocessed_items, principal=principal, ) unprocessed_items_count = len(unprocessed_items) if unprocessed_items_count > 0: return DeactivationSummary( deleted=len(tenants) - unprocessed_items_count, remaining=unprocessed_items_count, ) return DeactivationSummary(deleted=len(tenants), remaining=0)