from logging import getLogger from typing import Annotated, Any, Dict, List, Optional from uuid import UUID from pydantic import BaseModel, StringConstraints, TypeAdapter from pdp.constants.constants import DEFAULT_RESOURCE_ID, AuthEffect from pdp.fastapi.schemas.identity import CheckResourceAction, Resource, TenantRoles from pdp.fastapi.schemas.resource_attributes import ( IdentityAndTenantOwnedResourceAttributes, IdentityOwnedResourceAttributes, TenantOwnedResourceAttributes, ) from pdp.fastapi.schemas.tenant import Tenant logger = getLogger(__name__) class CreateCheckResourceActionsError(Exception): """Error for create_check_resource_actions.""" pass class CheckResourceTypeAction(BaseModel): """Representation for checking resource type action authorization.""" resource_type: Annotated[str, StringConstraints(min_length=1)] action: Annotated[str, StringConstraints(min_length=1)] def create_check_resource_actions( self, tenants: Dict[UUID, Tenant], identity_uuid: Optional[str] = None, ) -> List[CheckResourceAction]: """Creates list of CheckResourceAction(s). Turns a CheckResourceTypeAction into a CheckResourceAction, with attributes representing a tenant-owned, identity-owned, and shared resource. """ check_resource_action_list: List[CheckResourceAction] = [] if len(tenants): self._create_with_tenants( check_resource_action_list=check_resource_action_list, tenants=tenants, identity_uuid=identity_uuid, ) elif identity_uuid: self._create_with_identity_uuid( check_resource_action_list=check_resource_action_list, identity_uuid=identity_uuid, ) else: # The caller should handle this error and return the # appropriate response. raise CreateCheckResourceActionsError("No identity or tenants found") return check_resource_action_list def _create_with_tenants( self, check_resource_action_list: List[CheckResourceAction], tenants: Dict[UUID, Tenant], identity_uuid: Optional[str] = None, ) -> None: """Return a CheckResourceAction with tenant-owned resources.""" for tenant in tenants.values(): attributes: ( TenantOwnedResourceAttributes | IdentityAndTenantOwnedResourceAttributes # noqa: E501 ) = TenantOwnedResourceAttributes( tenant=tenant, ) if identity_uuid: attributes = IdentityAndTenantOwnedResourceAttributes( tenant=tenant, identity_uuid=identity_uuid, ) resource = CheckResourceAction( resource=Resource( resource_id=DEFAULT_RESOURCE_ID, resource_type=self.resource_type, attributes=attributes.model_dump(mode="json"), ), action=self.action, ) check_resource_action_list.append(resource) def _create_with_identity_uuid( self, check_resource_action_list: List[CheckResourceAction], identity_uuid: str, ) -> None: """Return a CheckResourceAction with IdentityOwnedResourceAttributes.""" attributes = IdentityOwnedResourceAttributes( identity_uuid=UUID(identity_uuid, version=4), ) resource = CheckResourceAction( resource=Resource( resource_id=DEFAULT_RESOURCE_ID, resource_type=self.resource_type, attributes=attributes.model_dump(mode="json"), ), action=self.action, ) check_resource_action_list.append(resource) class CheckResourceTypeActionResult(CheckResourceTypeAction): """Representation of result of checking resource type action authorization.""" effect: AuthEffect errors: Dict[str, Any] = {} ResourceTypeActionAdapter = TypeAdapter(List[CheckResourceTypeAction]) class CheckResourceTypeActionsRequest(BaseModel): """Request object for checking authorization on resource type - actions.""" resource_type_actions: List[CheckResourceTypeAction] def build_check_resource_actions( self, pdp_tenant_roles: Dict[UUID, TenantRoles], identity_uuid: Optional[str] = None, ) -> List[CheckResourceAction]: """Build a list of CheckResourceAction objects from a request.""" check_resource_type_action_list = self.resource_type_actions if not check_resource_type_action_list: return [] check_resource_action_list = [] for check_resource_type_action in check_resource_type_action_list: resources = check_resource_type_action.create_check_resource_actions( pdp_tenant_roles, # type: ignore[arg-type] identity_uuid=identity_uuid, ) check_resource_action_list.extend(resources) return check_resource_action_list class CheckResourceTypeActionsResponse(BaseModel): """Response object for checking authorization on resource type - actions.""" resource_type_actions: List[CheckResourceTypeActionResult]