"""Test cerbos Logic.""" import logging import uuid from json import JSONDecodeError from typing import Any, Dict, Optional from unittest.mock import AsyncMock, MagicMock, call, patch import pytest from cerbos.sdk.client import AsyncCerbosClient from cerbos.sdk.model import APIError as CerbosAPIError from cerbos.sdk.model import CheckResourcesResponse as CerbosCheckResourcesResponse from cerbos.sdk.model import CheckResourcesResult as CerbosCheckResourcesResult from cerbos.sdk.model import Effect as CerbosEffect from cerbos.sdk.model import Principal as CerbosPrincipal from cerbos.sdk.model import Resource as CerbosResource from cerbos.sdk.model import ResourceAction as CerbosResourceAction from cerbos.sdk.model import ResourceList as CerbosResourceList from cerbos.sdk.model import Source as CerbosSource from cerbos.sdk.model import ValidationError as CerbosValidationError from fastapi import HTTPException from httpcore import ConnectError from pdp.constants.constants import TenantType from pdp.constants.features import FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP from pdp.fastapi.schemas.check_resources import CheckResourcesRequest from pdp.fastapi.schemas.identity import ( CheckResourceAction, CheckResourceActionResult, CheckResourcesResponse, Resource, RolesResponse, TenantRoles, ) from pdp.fastapi.schemas.principal import Principal from pdp.fastapi.schemas.tenant import Tenant, TenantHierarchy from pdp.logic.cerbos import ( CERBOS_ATTRIBUTES_ERROR_MESSAGE, _build_principal, _build_resource_list, _hydrate_resources_with_hierarchy_as_needed, _merge_cerbos_responses, _paginated_resource_check, check_resources, ) @pytest.fixture() def mock_resource_response() -> CerbosCheckResourcesResponse: """Mock CheckResourcesResponse.""" return CerbosCheckResourcesResponse( request_id="meeeeee-123", results=[ CerbosCheckResourcesResult( resource=CerbosResource(id="1234", kind="fan_data_list"), actions={"delete": CerbosEffect.DENY}, validation_errors=[ CerbosValidationError( path="/tenant", message="missing properties: 'tenant_uuid'", source=CerbosSource.RESOURCE, ) ], ), CerbosCheckResourcesResult( resource=CerbosResource(id="7890", kind="fan_data_list"), actions={"delete": CerbosEffect.ALLOW}, ), ], ) @pytest.mark.parametrize("include_resource_attributes", [True, False]) @patch("pdp.logic.cerbos.get_cerbos_policy_metadata") @patch("pdp.logic.cerbos.BooleanFeature") @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.logic.cerbos._paginated_resource_check") @patch("pdp.logic.cerbos._build_principal") @patch("pdp.logic.cerbos._hydrate_resources_with_hierarchy_as_needed") async def test_check_resources( mock_hydrate_resources_with_hierarchy_as_needed: AsyncMock, mock_build_principal: MagicMock, mock_paginated_resource_check: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, mock_boolean_feature: MagicMock, mock_get_cerbos_policy_metadata: AsyncMock, include_resource_attributes: bool, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, ) -> None: """Test check resources.""" mock_resources_request = CheckResourcesRequest.model_validate( { "resources": [ { "resource": { "resource_id": "5678", "resource_type": "campaign", "attributes": {"test": "value"}, }, "action": "connect", }, { "resource": { "resource_id": "1234", "resource_type": "campaign", "attributes": {"test": "value2"}, }, "action": "connect", }, ], } ) mock_hydrated_request = MagicMock(spec=CheckResourcesRequest) mock_hydrate_resources_with_hierarchy_as_needed.return_value = mock_hydrated_request mock_cerbos_principal = MagicMock(spec=CerbosPrincipal) mock_build_principal.return_value = mock_cerbos_principal mock_splitio_client = MagicMock() mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_enabled.return_value = True result = await check_resources( identity_uuid, mock_resources_request, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, authenticated_identity_uuid=uuid.UUID(identity_uuid), ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, include_resource_attributes=include_resource_attributes, splitio_client=mock_splitio_client, ) mock_paginated_resource_check.assert_called_once_with( check_resources_request=mock_hydrated_request, cerbos_client=mock_async_cerbos_client, principal=mock_cerbos_principal, include_resource_attributes=include_resource_attributes, ) assert result == mock_paginated_resource_check.return_value mock_hydrate_resources_with_hierarchy_as_needed.assert_called_once_with( check_resources_request=mock_resources_request, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) # Assert the pp_vendor_features_lookup flag was checked so this test fails # (and forces an update) when the flag and its FF lookup are torn down. mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name=FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP, ) mock_boolean_feature_instance.is_enabled.assert_called_once_with() mock_get_cerbos_policy_metadata.assert_awaited_once_with(mock_redis_connector) @patch("pdp.logic.cerbos.get_cerbos_policy_metadata") @patch("pdp.logic.cerbos.BooleanFeature") @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.logic.cerbos._hydrate_resources_with_hierarchy_as_needed") @patch("pdp.logic.cerbos._paginated_resource_check") @patch("pdp.logic.cerbos._build_principal") @patch("pdp.logic.cerbos.get_correlation_id") async def test_check_resources_uses_ows_permissions_tenant_roles( mock_get_correlation_id: MagicMock, mock_build_principal: MagicMock, mock_paginated_resource_check: MagicMock, mock_hydrate_resources_with_hierarchy_as_needed: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, mock_boolean_feature: MagicMock, mock_get_cerbos_policy_metadata: AsyncMock, identity_uuid: str, mock_resource_response: CerbosCheckResourcesResponse, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncMock, ) -> None: """Test check resources passes ows_permissions_tenant_roles to _build_principal.""" mock_get_correlation_id.return_value = "follow me in the logs" mock_async_cerbos_client.check_resources.return_value = mock_resource_response mock_cerbos_principal = MagicMock(spec=CerbosPrincipal) mock_build_principal.return_value = mock_cerbos_principal mock_resources_request = MagicMock() mock_hydrate_resources_with_hierarchy_as_needed.return_value = ( mock_resources_request ) mock_splitio_client = MagicMock() mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_enabled.return_value = True await check_resources( identity_uuid, mock_resources_request, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, ) mock_build_principal.assert_called_once_with( identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, ) mock_hydrate_resources_with_hierarchy_as_needed.assert_awaited_once_with( check_resources_request=mock_resources_request, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) mock_paginated_resource_check.assert_called_once_with( check_resources_request=mock_resources_request, cerbos_client=mock_async_cerbos_client, principal=mock_cerbos_principal, include_resource_attributes=False, ) # Assert the pp_vendor_features_lookup flag was checked so this test fails # (and forces an update) when the flag and its FF lookup are torn down. mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name=FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP, ) mock_boolean_feature_instance.is_enabled.assert_called_once_with() mock_get_cerbos_policy_metadata.assert_awaited_once_with(mock_redis_connector) @patch("pdp.logic.cerbos.get_cerbos_policy_metadata") @patch("pdp.logic.cerbos.BooleanFeature") @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.logic.cerbos._paginated_resource_check") @patch("pdp.logic.cerbos._build_principal") @patch("pdp.logic.cerbos._hydrate_resources_with_hierarchy_as_needed") async def test_check_resources_uses_principal_obj_when_present( mock_hydrate_resources_with_hierarchy_as_needed: AsyncMock, mock_build_principal: MagicMock, mock_paginated_resource_check: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, mock_boolean_feature: MagicMock, mock_get_cerbos_policy_metadata: AsyncMock, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, ) -> None: """Test cerbos.check_resources uses Principal when present.""" include_resource_attributes = True principal = MagicMock(spec=Principal) mock_cerbos_principal = MagicMock(spec=CerbosPrincipal) principal.get_cerbos_principal.return_value = mock_cerbos_principal mock_resources_request = CheckResourcesRequest.model_validate( { "resources": [ { "resource": { "resource_id": "5678", "resource_type": "campaign", "attributes": {"test": "value"}, }, "action": "connect", }, { "resource": { "resource_id": "1234", "resource_type": "campaign", "attributes": {"test": "value2"}, }, "action": "connect", }, ], } ) mock_hydrated_request = MagicMock(spec=CheckResourcesRequest) mock_hydrate_resources_with_hierarchy_as_needed.return_value = mock_hydrated_request mock_splitio_client = MagicMock() mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_enabled.return_value = True result = await check_resources( identity_uuid, mock_resources_request, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, authenticated_identity_uuid=uuid.UUID(identity_uuid), ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, include_resource_attributes=include_resource_attributes, principal=principal, splitio_client=mock_splitio_client, ) assert result == mock_paginated_resource_check.return_value mock_build_principal.assert_not_called() principal.get_cerbos_principal.assert_called_once() mock_paginated_resource_check.assert_called_once_with( check_resources_request=mock_hydrated_request, cerbos_client=mock_async_cerbos_client, principal=mock_cerbos_principal, include_resource_attributes=include_resource_attributes, ) mock_hydrate_resources_with_hierarchy_as_needed.assert_called_once_with( check_resources_request=mock_resources_request, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) # Assert the pp_vendor_features_lookup flag was checked so this test fails # (and forces an update) when the flag and its FF lookup are torn down. mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name=FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP, ) mock_boolean_feature_instance.is_enabled.assert_called_once_with() mock_get_cerbos_policy_metadata.assert_awaited_once_with(mock_redis_connector) async def test_build_resource_list() -> None: """Test build resource list.""" check_resources_request = { "resources": [ { "action": "view", "resource": { "resource_id": "first", "resource_type": "job", "attributes": {"pay_schedule": "biweekly"}, }, }, { "action": "delete", "resource": { "resource_id": "second", "resource_type": "job", "attributes": {"pay_schedule": "bonus"}, }, }, ] } request = CheckResourcesRequest.model_validate(check_resources_request) result = await _build_resource_list(request) assert result == CerbosResourceList( resources=[ CerbosResourceAction( actions={"view"}, resource=CerbosResource( id="first", kind="job", attr={"pay_schedule": "biweekly"} ), ), CerbosResourceAction( actions={"delete"}, resource=CerbosResource( id="second", kind="job", attr={"pay_schedule": "bonus"} ), ), ] ) def test_build_principal( mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], ) -> None: """Test _build_principal.""" mock_pdp_tenant_roles = RolesResponse.model_validate( { "tenants": { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], } }, "cursor": {"cursor": None}, } ).tenants identity_uuid = uuid.uuid4() result = _build_principal( str(identity_uuid), pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=None, ) assert result.id == str(identity_uuid) assert result.roles == {"user"} assert result.attr == { "type": "human", "tenants": { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": {"dog_whisperer": {"role": "dog_whisperer"}}, }, }, } def test_build_principal_adds_in_ows_permissions_tenant_roles( mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], ) -> None: """Test _build_principal adds in new tenants from ows-permissions tenant roles.""" mock_pdp_tenant_roles = RolesResponse.model_validate( { "tenants": { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], } }, "cursor": {"cursor": None}, } ).tenants identity_uuid = uuid.uuid4() result = _build_principal( str(identity_uuid), pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, ) assert result.id == str(identity_uuid) assert result.roles == {"user"} assert result.attr == { "type": "human", "tenants": { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": {"dog_whisperer": {"role": "dog_whisperer"}}, }, "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "tenant_type": "account", "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "roles": { "ows_permissions_rap_admin": {"role": "ows_permissions_rap_admin"} }, }, }, } def test_build_principal_merges_in_ows_permissions_tenant_roles( mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], ) -> None: """Test _build_principal merges in roles from ows-permissions tenant roles.""" mock_pdp_tenant_roles = RolesResponse.model_validate( { "tenants": { "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "tenant_type": "account", "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "roles": [{"role": "dog_whisperer"}], } }, "cursor": {"cursor": None}, } ).tenants identity_uuid = uuid.uuid4() result = _build_principal( str(identity_uuid), pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, ) assert result.id == str(identity_uuid) assert result.roles == {"user"} assert result.attr == { "type": "human", "tenants": { "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "tenant_type": "account", "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "roles": { "dog_whisperer": {"role": "dog_whisperer"}, "ows_permissions_rap_admin": {"role": "ows_permissions_rap_admin"}, }, } }, } COMPANY_BRAND_TENANT_UUID = uuid.UUID("c2254421-2acb-4ca9-af17-23e95d99f79f") @pytest.mark.parametrize( "description, resource_request, batch_responses, expected_result, expect_exception", [ ( "Response objects from multiple Cerbos responses should be merged.", CheckResourcesRequest.model_validate( { "resources": [ { "resource": { "resource_id": "1", "resource_type": "bunny", "attributes": {"cuteness": "very"}, }, "action": "view", }, { "resource": { "resource_id": "2", "resource_type": "dog", "attributes": {"cuteness": "sort of"}, }, "action": "walk", }, { "resource": { "resource_id": "3", "resource_type": "cat", "attributes": {"cuteness": "no"}, }, "action": "play", }, ] } ), [ CerbosCheckResourcesResponse( request_id="some-sort-of-id", results=[ CerbosCheckResourcesResult( resource=CerbosResource( id="first", kind="bunny", attr={}, policy_version="default", scope="", ), actions={"view": CerbosEffect.DENY}, validation_errors=None, outputs=None, ), CerbosCheckResourcesResult( resource=CerbosResource( id="second", kind="dog", attr={}, policy_version="default", scope="", ), actions={"walk": CerbosEffect.ALLOW}, validation_errors=None, outputs=None, ), ], status_code=200, status_msg=None, ), CerbosCheckResourcesResponse( request_id="some-other-id-we-will-not-use", results=[ CerbosCheckResourcesResult( resource=CerbosResource( id="third", kind="cat", attr={}, policy_version="default", scope="", ), actions={"play": CerbosEffect.DENY}, validation_errors=None, outputs=None, ) ], status_code=200, status_msg=None, ), ], CheckResourcesResponse.model_validate( { "request_id": "some-sort-of-id", "resources": [ { "action": "view", "effect": "deny", "resource": { "resource_id": "first", "resource_type": "bunny", "attributes": {"cuteness": "very"}, }, "errors": {"validation_errors": None}, }, { "action": "walk", "effect": "allow", "resource": { "resource_id": "second", "resource_type": "dog", "attributes": {"cuteness": "sort of"}, }, "errors": {"validation_errors": None}, }, { "action": "play", "effect": "deny", "resource": { "resource_id": "third", "resource_type": "cat", "attributes": {"cuteness": "no"}, }, "errors": {"validation_errors": None}, }, ], } ), False, ), ( "Unequal number of request and response objects should cause a 500.", CheckResourcesRequest.model_validate( { "resources": [ { "resource": { "resource_id": "1", "resource_type": "bunny", "attributes": {"cuteness": "very"}, }, "action": "view", }, { "resource": { "resource_id": "2", "resource_type": "dog", "attributes": {"cuteness": "sort of"}, }, "action": "walk", }, ] } ), [ CerbosCheckResourcesResponse( request_id="some-sort-of-id", results=[ CerbosCheckResourcesResult( resource=CerbosResource( id="first", kind="bunny", attr={}, policy_version="default", scope="", ), actions={"view": CerbosEffect.DENY}, validation_errors=None, outputs=None, ), ], status_code=200, status_msg=None, ) ], None, True, ), ], ) def test_merge_cerbos_responses( description: str, resource_request: CheckResourcesRequest, batch_responses: list[CerbosCheckResourcesResponse], expected_result: Optional[CheckResourcesResponse], expect_exception: bool, ) -> None: """Test _merge_cerbos_responses.""" if expect_exception: with pytest.raises(HTTPException) as actual_exception: _merge_cerbos_responses( check_resources_request=resource_request, batch_responses=batch_responses, include_resource_attributes=True, ) assert ( actual_exception.value.detail == "Batched Cerbos responses does not match request length." ), description else: merge_result = _merge_cerbos_responses( check_resources_request=resource_request, batch_responses=batch_responses, include_resource_attributes=True, ) assert merge_result == expected_result, description @pytest.mark.parametrize( "description, side_effect, expect_exception", [ ( "Multiple Cerbos responses should be combined into one response.", [ CerbosCheckResourcesResponse( request_id="test-cerbos-request-id", results=[ CerbosCheckResourcesResult( resource=CerbosResource( id="some-id", kind="a-resource-type" ), actions={"view": CerbosEffect.ALLOW}, validation_errors=[], ) ], ), CerbosCheckResourcesResponse( request_id="test-cerbos-request-id", results=[ CerbosCheckResourcesResult( resource=CerbosResource( id="some-other-id", kind="an-exciting-resource-type" ), actions={"edit": CerbosEffect.ALLOW}, validation_errors=[], ) ], ), ], False, ), ( "A Cerbos error should result in a raised http exception.", Exception("bad news"), True, ), ( "A valid response and a Cerbos error should result" " in a raised http exception.", [ CerbosCheckResourcesResponse( request_id="test-cerbos-request-id", results=[ CerbosCheckResourcesResult( resource=CerbosResource( id="some-id", kind="a-resource-type" ), actions={"view": CerbosEffect.ALLOW}, validation_errors=[], ) ], ), Exception("bad news"), ], True, ), ( "Empty cerbos results should result in a raised http exception.", [ CerbosCheckResourcesResponse( request_id="test-cerbos-request-id", results=None, status_msg=CerbosAPIError(code=3, message="bad news"), ), ], True, ), ], ) @patch("pdp.logic.cerbos.CERBOS_BATCH_SIZE", 1) @patch("pdp.logic.cerbos.AsyncCerbosClient") @patch("pdp.logic.cerbos._build_resource_list") async def test_paginated_resource_check( mock_build_resource_list: MagicMock, mock_async_cerbos_client: AsyncMock, identity_uuid: str, description: str, side_effect: Any, expect_exception: bool, ) -> None: """Test paginated resource check.""" resource_request = CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_id="some-id", resource_type="a-resource-type", attributes={}, ), action="view", ), CheckResourceAction( resource=Resource( resource_id="some-other-id", resource_type="an-exciting-resource-type", attributes={}, ), action="edit", ), ] ) principal = CerbosPrincipal( id=identity_uuid, roles={"user"}, attr={}, policy_version="default", scope="", ) mock_async_cerbos_client.check_resources = AsyncMock(side_effect=side_effect) if expect_exception: with pytest.raises(Exception) as actual_exception: await _paginated_resource_check( check_resources_request=resource_request, cerbos_client=mock_async_cerbos_client, principal=principal, ) assert "bad news" in str(actual_exception.value) else: result = await _paginated_resource_check( check_resources_request=resource_request, cerbos_client=mock_async_cerbos_client, principal=principal, ) assert mock_build_resource_list.call_count == 2 resource_request_1 = dict( resource_request.model_dump(), resources=[resource_request.resources[0]] ) resource_request_2 = dict( resource_request.model_dump(), resources=[resource_request.resources[1]] ) assert mock_build_resource_list.call_args_list == [ call( CheckResourcesRequest(**resource_request_1), ), call( CheckResourcesRequest(**resource_request_2), ), ] assert result == CheckResourcesResponse( request_id="test-cerbos-request-id", resources=[ CheckResourceActionResult( resource=Resource( resource_id="some-id", resource_type="a-resource-type", attributes={}, ), action="view", effect="allow", errors={"validation_errors": []}, ), CheckResourceActionResult( resource=Resource( resource_id="some-other-id", resource_type="an-exciting-resource-type", attributes={}, ), action="edit", effect="allow", errors={"validation_errors": []}, ), ], ), description @patch("pdp.logic.cerbos.CERBOS_BATCH_SIZE", 1) @patch("pdp.logic.cerbos._merge_cerbos_responses") @patch("pdp.logic.cerbos.AsyncCerbosClient") @patch("pdp.logic.cerbos._build_resource_list") async def test_paginated_resource_check_makes_expected_calls( mock_build_resource_list: AsyncMock, mock_async_cerbos_client: AsyncMock, mock_merge_cerbos_responses: MagicMock, identity_uuid: str, ) -> None: """Test paginated resource check.""" mock_async_cerbos_client.check_resources = AsyncMock() resource_request = CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_id="some-id", resource_type="a-resource-type", attributes={}, ), action="view", ), CheckResourceAction( resource=Resource( resource_id="some-other-id", resource_type="an-exciting-resource-type", attributes={}, ), action="edit", ), ] ) principal = CerbosPrincipal( id=identity_uuid, roles={"user"}, attr={}, policy_version="default", scope="", ) mock_build_resource_list.side_effect = [ CerbosResourceList( resources=[ CerbosResourceAction( actions={"view"}, resource=CerbosResource( id="some-id", kind="a-resource-type", attr={}, ), ) ] ), CerbosResourceList( resources=[ CerbosResourceAction( actions={"edit"}, resource=CerbosResource( id="some-other-id", kind="an-exciting-resource-type", attr={}, ), ) ] ), ] await _paginated_resource_check( check_resources_request=resource_request, cerbos_client=mock_async_cerbos_client, principal=principal, ) mock_async_cerbos_client.check_resources.assert_has_calls( [ call( principal, CerbosResourceList( resources=[ CerbosResourceAction( actions={"view"}, resource=CerbosResource( id="some-id", kind="a-resource-type", attr={}, ), ) ] ), None, ), call( principal, CerbosResourceList( resources=[ CerbosResourceAction( actions={"edit"}, resource=CerbosResource( id="some-other-id", kind="an-exciting-resource-type", attr={}, ), ) ] ), None, ), ] ) @pytest.mark.parametrize( "exception_class, raised_exception, expected_status_code, expected_message, expected_log", # noqa: E501 [ ( Exception, Exception("bad news"), 500, ("bad news",), "Unhandled Cerbos error", ), ( ConnectError, ConnectError(), 500, "Cerbos is unavailable", "Cerbos is unavailable", ), ( HTTPException, KeyError("what did you do"), 400, f"400: {CERBOS_ATTRIBUTES_ERROR_MESSAGE} Invalid ('what did you do',)", CERBOS_ATTRIBUTES_ERROR_MESSAGE, ), ], ) @patch("pdp.logic.cerbos._merge_cerbos_responses") @patch("pdp.logic.cerbos._build_resource_list") @patch("pdp.logic.cerbos.get_correlation_id") async def test_paginated_resource_check_raises_exceptions( mock_get_correlation_id: MagicMock, mock_build_resource_list: AsyncMock, mock_merge_cerbos_responses: MagicMock, exception_class: type[BaseException], raised_exception: BaseException, expected_status_code: int, expected_message: str, expected_log: str, mock_async_cerbos_client: AsyncMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test _paginated_resource_check raises http exceptions.""" mock_get_correlation_id.return_value = "down the rabbit hole" mock_async_cerbos_client.check_resources.side_effect = raised_exception mock_principal = MagicMock() mock_resources_request = CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_id="123", resource_type="dog", attributes={}, ), action="feed", ), ] ) mock_resource_list = MagicMock() mock_build_resource_list.return_value = mock_resource_list with pytest.raises(exception_class) as actual_exception: await _paginated_resource_check( mock_resources_request, cerbos_client=mock_async_cerbos_client, principal=mock_principal, ) if exception_class == HTTPException: assert str(actual_exception.value) == expected_message mock_build_resource_list.assert_called_once() mock_merge_cerbos_responses.assert_not_called() with caplog.at_level(logging.WARNING): assert expected_log in caplog.text @patch("pdp.logic.cerbos.MultiTenantProxy") @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.logic.cerbos.OwsParticipantClient") async def test__hydrate_resources_with_hierarchy_as_needed( mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, mock_multi_tenant_proxy: MagicMock, ) -> None: """Test CheckResourcesRequest is hydrated with hierarchy, as needed.""" missing_hierarchy_uuid = uuid.uuid4() found_hierarchy_uuid = uuid.uuid4() already_has_hierarchy_uuid = uuid.uuid4() already_has_hierarchy_uuid2 = uuid.uuid4() def mock_get_tenant_hierarchy(uuid: uuid.UUID) -> Optional[TenantHierarchy]: assert uuid not in [already_has_hierarchy_uuid, already_has_hierarchy_uuid2] if uuid in [found_hierarchy_uuid, already_has_hierarchy_uuid]: return TenantHierarchy( company_brand=Tenant( tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, tenant_uuid=COMPANY_BRAND_TENANT_UUID, ) ) if uuid == already_has_hierarchy_uuid2: return TenantHierarchy( company_brand=Tenant( tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, tenant_uuid=missing_hierarchy_uuid, ) ) return None mock_multi_tenant_proxy.return_value.gather_tenant_hierarchies = AsyncMock() mock_multi_tenant_proxy.return_value.get_tenant_hierarchy = MagicMock( side_effect=mock_get_tenant_hierarchy ) check_resources_request = CheckResourcesRequest( resources=[ CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=1, resource_type="no_tenant", ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=2, resource_type="bad_tenant", attributes={ "tenant": { "tenant_uuid": "bad_uuid", "tenant_type": "bad_account", } }, ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=3, resource_type="missing_hierarchy", attributes={ "tenant": { "tenant_uuid": missing_hierarchy_uuid, "tenant_type": "account", } }, ), ), CheckResourceAction( action="will_hydrate_hierarchy", resource=Resource( resource_id=4, resource_type="audience", attributes={ "tenant": { "tenant_uuid": found_hierarchy_uuid, "tenant_type": "account", } }, ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=5, resource_type="audience", attributes={ "tenant": { "tenant_uuid": already_has_hierarchy_uuid, "tenant_type": "account", "tenant_hierarchy": [], } }, ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=6, resource_type="audience", attributes={ "tenant": { "tenant_uuid": already_has_hierarchy_uuid2, "tenant_type": "account", "tenant_hierarchy": [str(COMPANY_BRAND_TENANT_UUID)], } }, ), ), ] ) result = await _hydrate_resources_with_hierarchy_as_needed( check_resources_request=check_resources_request, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) mock_multi_tenant_proxy.assert_called_once_with( tenants=[ Tenant( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=missing_hierarchy_uuid, ), Tenant( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=found_hierarchy_uuid, ), ], redis_client=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) mock_multi_tenant_proxy.return_value.gather_tenant_hierarchies.assert_called_once() mock_multi_tenant_proxy.return_value.get_tenant_hierarchy.assert_has_calls( [ call(missing_hierarchy_uuid), call(found_hierarchy_uuid), ] ) assert result == CheckResourcesRequest( resources=[ CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=1, resource_type="no_tenant", ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=2, resource_type="bad_tenant", attributes={ "tenant": { "tenant_uuid": "bad_uuid", "tenant_type": "bad_account", } }, ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=3, resource_type="missing_hierarchy", attributes={ "tenant": { "tenant_uuid": missing_hierarchy_uuid, "tenant_type": "account", }, }, ), ), CheckResourceAction( action="will_hydrate_hierarchy", resource=Resource( resource_id=4, resource_type="audience", attributes={ "tenant": { "tenant_uuid": found_hierarchy_uuid, "tenant_type": "account", "tenant_hierarchy": [str(COMPANY_BRAND_TENANT_UUID)], }, }, ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=5, resource_type="audience", attributes={ "tenant": { "tenant_uuid": already_has_hierarchy_uuid, "tenant_type": "account", "tenant_hierarchy": [], } }, ), ), CheckResourceAction( action="will_not_hydrate_hierarchy", resource=Resource( resource_id=6, resource_type="audience", attributes={ "tenant": { "tenant_uuid": already_has_hierarchy_uuid2, "tenant_type": "account", "tenant_hierarchy": [str(COMPANY_BRAND_TENANT_UUID)], } }, ), ), ] ) @patch("pdp.logic.cerbos.AsyncCerbosClient") async def test___paginated_resource_check__jsondecoder_error( mock_async_cerbos_client: AsyncMock, identity_uuid: str, ) -> None: """Test paginated resource check returns a 500 error for JSONDecoder errors.""" mock_async_cerbos_client.check_resources = AsyncMock( side_effect=JSONDecodeError(msg="error", doc="", pos=0) ) resource_request = CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_id="some-id", resource_type="a-resource-type", attributes={}, ), action="view", ), ] ) principal = CerbosPrincipal( id=identity_uuid, roles={"user"}, attr={}, policy_version="default", scope="", ) with pytest.raises(HTTPException) as actual_exception: await _paginated_resource_check( check_resources_request=resource_request, cerbos_client=mock_async_cerbos_client, principal=principal, ) assert "AWS WAF or ELB error" in str(actual_exception.value) assert actual_exception.value.status_code == 500 @patch("pdp.logic.cerbos.get_cerbos_policy_metadata") @patch("pdp.logic.cerbos._hydrate_resources_with_hierarchy_as_needed") @patch("pdp.logic.cerbos._paginated_resource_check") @patch("pdp.logic.cerbos._build_principal") async def test_check_resources_reads_policy_metadata_when_flag_on( mock_build_principal: MagicMock, mock_paginated_resource_check: AsyncMock, mock_hydrate: AsyncMock, mock_get_cerbos_policy_metadata: AsyncMock, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_redis_connector: MagicMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, ) -> None: """get_cerbos_policy_metadata is called when pp_vendor_features_lookup is on.""" mock_splitio_client = MagicMock() mock_hydrate.return_value = MagicMock() with patch("pdp.logic.cerbos.BooleanFeature") as mock_boolean_feature_cls: mock_flag = MagicMock() mock_flag.is_enabled.return_value = True mock_boolean_feature_cls.return_value = mock_flag await check_resources( identity_uuid, MagicMock(), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, ) mock_get_cerbos_policy_metadata.assert_awaited_once_with(mock_redis_connector) @patch("pdp.logic.cerbos.get_cerbos_policy_metadata") @patch("pdp.logic.cerbos._hydrate_resources_with_hierarchy_as_needed") @patch("pdp.logic.cerbos._paginated_resource_check") @patch("pdp.logic.cerbos._build_principal") async def test_check_resources_skips_policy_metadata_when_flag_off( mock_build_principal: MagicMock, mock_paginated_resource_check: AsyncMock, mock_hydrate: AsyncMock, mock_get_cerbos_policy_metadata: AsyncMock, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_redis_connector: MagicMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, ) -> None: """get_cerbos_policy_metadata is skipped when pp_vendor_features_lookup is off.""" mock_splitio_client = MagicMock() mock_hydrate.return_value = MagicMock() with patch("pdp.logic.cerbos.BooleanFeature") as mock_boolean_feature_cls: mock_flag = MagicMock() mock_flag.is_enabled.return_value = False mock_boolean_feature_cls.return_value = mock_flag await check_resources( identity_uuid, MagicMock(), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, ) mock_get_cerbos_policy_metadata.assert_not_called() @patch("pdp.logic.cerbos.get_cerbos_policy_metadata") @patch("pdp.logic.cerbos._hydrate_resources_with_hierarchy_as_needed") @patch("pdp.logic.cerbos._paginated_resource_check") @patch("pdp.logic.cerbos._build_principal") async def test_check_resources_skips_policy_metadata_when_no_splitio_client( mock_build_principal: MagicMock, mock_paginated_resource_check: AsyncMock, mock_hydrate: AsyncMock, mock_get_cerbos_policy_metadata: AsyncMock, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_redis_connector: MagicMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, ) -> None: """get_cerbos_policy_metadata is not called when no splitio_client is provided.""" mock_hydrate.return_value = MagicMock() await check_resources( identity_uuid, MagicMock(), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=None, ) mock_get_cerbos_policy_metadata.assert_not_called()