"""PdpAuthorizationBackend tests.""" import logging import uuid from typing import Any from unittest.mock import MagicMock import pytest from pytest_mock import MockerFixture from python_pdp_sdk import ( PdpAuthorizationBackend, ResourceWithAttributes, ) from python_pdp_sdk.backends.authorization_backend import ( AuthorizationBackend, ResourceAction, ) from python_pdp_sdk.backends.exceptions import ( AllowedTenantsException, AttributesException, InvalidRequestException, PdpAuthenticationError, UnauthenticatedException, UnauthorizedException, ) from python_pdp_sdk.connectors.ows_pdp.models.allowed_tenant import AllowedTenant from python_pdp_sdk.connectors.ows_pdp.models.auth_effect import AuthEffect from python_pdp_sdk.connectors.ows_pdp.models.check_resource_action import ( CheckResourceAction, ) from python_pdp_sdk.connectors.ows_pdp.models.check_resource_action_result import ( CheckResourceActionResult, ) from python_pdp_sdk.connectors.ows_pdp.models.check_resources_request import ( CheckResourcesRequest, ) from python_pdp_sdk.connectors.ows_pdp.models.check_resources_response import ( CheckResourcesResponse, ) from python_pdp_sdk.connectors.ows_pdp.models.get_allowed_tenants_response import ( GetAllowedTenantsResponse, ) from python_pdp_sdk.connectors.ows_pdp.models.resource import Resource from python_pdp_sdk.connectors.ows_pdp.models.resource_id import ResourceId from python_pdp_sdk.connectors.ows_pdp.models.tenant_type import TenantType from python_pdp_sdk.connectors.ows_pdp.ows_pdp import OwsPdpClient @pytest.fixture() def mock_ows_pdp_client() -> MagicMock: """Return mocked OwsPdpClient.""" return MagicMock(spec=OwsPdpClient) @pytest.fixture() def pdp_backend(mock_ows_pdp_client: MagicMock) -> PdpAuthorizationBackend: """Return reusable PdpAuthorizationBackend.""" return PdpAuthorizationBackend(mock_ows_pdp_client) def make_check_my_resources_response( is_allowed: bool, action: str = "eat", resource_type: str = "food", ) -> CheckResourcesResponse: """Return a configured CheckResourcesResponse.""" return CheckResourcesResponse( request_id="some request", resources=[ CheckResourceActionResult( action=action, resource=Resource( resource_id=ResourceId(0), resource_type=resource_type ), effect=(AuthEffect.ALLOW if is_allowed else AuthEffect.DENY), ) ], ) @pytest.mark.parametrize( "get_attributes_side_effect, raise_when_unauthorized, check_my_resources_response, expected_exception, expected", # noqa: E501 [ pytest.param( {"some": "attributes"}, False, make_check_my_resources_response(True), None, True, id="PDP allows", ), pytest.param( {"some": "attributes"}, True, make_check_my_resources_response(True), None, True, id="PDP allows, no exception raised", ), pytest.param( {"some": "attributes"}, False, make_check_my_resources_response(False), None, False, id="PDP denies", ), pytest.param( {"some": "attributes"}, True, make_check_my_resources_response(False), UnauthorizedException, False, id="PDP denies, exception is raised", ), pytest.param( {"some": "attributes"}, True, Exception("some non 200 from PDP"), UnauthorizedException, False, id="PDP has error, exception is raised", ), pytest.param( Exception("couldn't get attributes"), True, None, AttributesException, False, id="get_attributes fails, exception is raised", ), ], ) def test_is_authorized( get_attributes_side_effect: dict[str, Any] | Exception, raise_when_unauthorized: bool, check_my_resources_response: Any, expected_exception: Exception | None, expected: bool, mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, ) -> None: """Test is_authorized.""" mock_resource_getter = MagicMock() mock_resource_getter.get_attributes.side_effect = [get_attributes_side_effect] mock_ows_pdp_client.check_my_resources.side_effect = [check_my_resources_response] if expected_exception: with pytest.raises(Exception) as e: pdp_backend.is_authorized( "eat", "pho", "food", mock_resource_getter, raise_when_unauthorized, "some", "args", have="all", the="fun", ) assert isinstance(e, type(expected_exception)) else: result = pdp_backend.is_authorized( "eat", "pho", "food", mock_resource_getter, raise_when_unauthorized, "some", "args", have="all", the="fun", ) assert result == expected mock_resource_getter.get_attributes.assert_called_once_with( "some", "args", have="all", the="fun" ) if isinstance(get_attributes_side_effect, Exception): mock_ows_pdp_client.check_my_resources.assert_not_called() else: mock_ows_pdp_client.check_my_resources.assert_called_once_with( CheckResourcesRequest( resources=[ CheckResourceAction( action="eat", resource=Resource( resource_id=ResourceId("pho"), resource_type="food", attributes=get_attributes_side_effect, ), ) ] ), ) def test_get_authorized_tenants( pdp_backend: PdpAuthorizationBackend, mock_ows_pdp_client: MagicMock ) -> None: """Test get_authorized_tenants returns list of Tenants.""" tenant_uuid = str(uuid.uuid4()) mock_ows_pdp_client.get_allowed_tenants.return_value = GetAllowedTenantsResponse( resource_type="food", action="eat", tenants=[ AllowedTenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.SUBACCOUNT) ], ) result = pdp_backend.get_authorized_tenants("eat", "food") assert result == [ AllowedTenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.SUBACCOUNT) ] def test_get_authorized_tenants_failure( pdp_backend: PdpAuthorizationBackend, mock_ows_pdp_client: MagicMock, caplog: Any, ) -> None: """Test get_authorized_tenants failure.""" mock_ows_pdp_client.get_allowed_tenants.side_effect = Exception("some error") with pytest.raises(AllowedTenantsException): pdp_backend.get_authorized_tenants("fail", "test") with caplog.at_level("WARNING"): assert "Error when getting allowed tenants" in caplog.text def test_build_check_resources_request__valid_input( pdp_backend: PdpAuthorizationBackend, ) -> None: """Test the _build_check_resources_request method with valid input.""" resources_with_attributes = [ ResourceWithAttributes( resource_id="123", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ResourceWithAttributes( resource_id="456", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "7c6e068b-73cc-4917-a626-b47b3c35b517", } }, ), ] check_resources_request = pdp_backend._build_check_resources_request( action="view", resource_type="test-resource", resources_with_attributes=resources_with_attributes, ) assert check_resources_request == CheckResourcesRequest( resources=[ CheckResourceAction( action="view", resource=Resource( resource_id=ResourceId("123"), resource_type="test-resource", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ), CheckResourceAction( action="view", resource=Resource( resource_id=ResourceId("456"), resource_type="test-resource", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "7c6e068b-73cc-4917-a626-b47b3c35b517", } }, ), ), ] ) def test_build_check_resources_request__invalid_input( pdp_backend: PdpAuthorizationBackend, ) -> None: """Test the _build_check_resources_request method with invalid input.""" with pytest.raises(InvalidRequestException) as e: pdp_backend._build_check_resources_request( action="view", resource_type="test-resource", resources_with_attributes=[] ) assert "Received a check-resources request without any resources." in str(e.value) def test_is_authorized_many( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, mocker: MockerFixture, ) -> None: """Test is_authorized_many.""" request_id = "testing" action = "view" resource_type = "test-resource" resources_with_attributes = [ ResourceWithAttributes( resource_id="123", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ResourceWithAttributes( resource_id="456", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "7c6e068b-73cc-4917-a626-b47b3c35b517", } }, ), ] # Spy on this helper method. build_check_resources_request_spy = mocker.spy( pdp_backend, "_build_check_resources_request" ) # response from pdp mock_ows_pdp_client.check_my_resources.return_value = CheckResourcesResponse( request_id=request_id, resources=[ CheckResourceActionResult( action=action, resource=Resource( resource_id=ResourceId("123"), resource_type=resource_type ), effect=AuthEffect.ALLOW, ), CheckResourceActionResult( action=action, resource=Resource( resource_id=ResourceId("456"), resource_type=resource_type ), effect=AuthEffect.DENY, ), ], ) results = pdp_backend.is_authorized_many( action=action, resource_type=resource_type, resources_with_attributes=resources_with_attributes, ) # [ALLOW, DENY] effects should return [True, False] booleans assert results == [True, False] mock_ows_pdp_client.check_my_resources.assert_called_once_with( check_resources_request=CheckResourcesRequest( resources=[ CheckResourceAction( action=action, resource=Resource( resource_id=ResourceId("123"), resource_type=resource_type, attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ), CheckResourceAction( action=action, resource=Resource( resource_id=ResourceId("456"), resource_type=resource_type, attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "7c6e068b-73cc-4917-a626-b47b3c35b517", } }, ), ), ], ), ) # Verify that the helper function received the expected arguments. build_check_resources_request_spy.assert_called_once_with( action=action, resource_type=resource_type, resources_with_attributes=resources_with_attributes, ) @pytest.mark.parametrize("raise_when_unauthorized", [(True), (False)]) def test_is_authorized_many__error( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, caplog: pytest.LogCaptureFixture, raise_when_unauthorized: bool, ) -> None: """Test is_authorized_many PDP lookup failure case.""" caplog.set_level(logging.WARNING) action = "view" resource_type = "test-resource" mock_ows_pdp_client.check_my_resources.side_effect = RuntimeError( "M.E.T.H.O.D. Man..." ) if raise_when_unauthorized: with pytest.raises(Exception) as e: _ = pdp_backend.is_authorized_many( action=action, resource_type=resource_type, resources_with_attributes=[ ResourceWithAttributes( resource_id="123", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ) ], raise_when_unauthorized=raise_when_unauthorized, ) assert "Implicitly denied by an upstream exception" in str(e.value) else: _ = pdp_backend.is_authorized_many( action=action, resource_type=resource_type, resources_with_attributes=[ ResourceWithAttributes( resource_id="123", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ) ], raise_when_unauthorized=raise_when_unauthorized, ) assert "M.E.T.H.O.D. Man..." in caplog.text mock_ows_pdp_client.check_my_resources.assert_called_once_with( check_resources_request=CheckResourcesRequest( resources=[ CheckResourceAction( action=action, resource=Resource( resource_id=ResourceId("123"), resource_type=resource_type, attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ), ] ) ) def test_is_authorized_many__raise_when_unauthorized_false( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, ) -> None: """Test is_authorized_many PDP lookup failure case when raise flag is False.""" action = "view" resource_type = "test-resource" mock_ows_pdp_client.check_my_resources.side_effect = RuntimeError( "C.R.E.A.M. get the money..." ) result = pdp_backend.is_authorized_many( action=action, resource_type=resource_type, resources_with_attributes=[ ResourceWithAttributes( resource_id="123", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ResourceWithAttributes( resource_id="456", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "4474b51d-615b-48f0-ba5e-e1c00be330d3", } }, ), ], raise_when_unauthorized=False, ) assert result == [False, False] def test_is_authorized_many__raise_when_unauthorized_true( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, ) -> None: """Test is_authorized_many PDP lookup when raise flag is True and DENYs are received.""" request_id = "testing" action = "view" resource_type = "test-resource" # response from pdp mock_ows_pdp_client.check_my_resources.return_value = CheckResourcesResponse( request_id=request_id, resources=[ CheckResourceActionResult( action=action, resource=Resource( resource_id=ResourceId("123"), resource_type=resource_type ), effect=AuthEffect.ALLOW, ), CheckResourceActionResult( action=action, resource=Resource( resource_id=ResourceId("456"), resource_type=resource_type ), effect=AuthEffect.DENY, ), ], ) with pytest.raises(UnauthorizedException) as e: _ = pdp_backend.is_authorized_many( action=action, resource_type=resource_type, resources_with_attributes=[ ResourceWithAttributes( resource_id="123", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ResourceWithAttributes( resource_id="456", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "4474b51d-615b-48f0-ba5e-e1c00be330d3", } }, ), ], raise_when_unauthorized=True, ) assert "Unauthorized for at least one resource." in str(e.value) @pytest.fixture() def list_of_resource_actions() -> list[ResourceAction]: """Reusable list of ResourceAction.""" return [ ResourceAction( resource_id=1, attributes={"what": "no tenant here"}, action="eat", resource_type="meal", ), ResourceAction( resource_id="1", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", }, }, action="walk", resource_type="dog", ), ] @pytest.fixture() def list_of_resource_actions_as_check_resources_request() -> CheckResourcesRequest: """Reusable fixture for the list of resource actions as a CheckResourcesRequest.""" return CheckResourcesRequest( resources=[ CheckResourceAction( action="eat", resource=Resource( resource_id=ResourceId(1), resource_type="meal", attributes={"what": "no tenant here"}, ), ), CheckResourceAction( action="walk", resource=Resource( resource_id=ResourceId("1"), resource_type="dog", attributes={ "tenant": { "tenant_type": "account", "tenant_uuid": "9b73a995-8b6e-41ae-81d9-2b0822d436ba", } }, ), ), ], ) def test_is_authorized_many_resources_and_actions( list_of_resource_actions: list[ResourceAction], list_of_resource_actions_as_check_resources_request: CheckResourcesRequest, mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, ) -> None: """Test is_authorized_many_resources_and_actions happy path.""" mock_ows_pdp_client.check_my_resources.return_value = CheckResourcesResponse( request_id="testing", resources=[ CheckResourceActionResult( action="eat", resource=Resource(resource_id=ResourceId(1), resource_type="meal"), effect=AuthEffect.ALLOW, ), CheckResourceActionResult( action="walk", resource=Resource(resource_id=ResourceId("1"), resource_type="dog"), effect=AuthEffect.DENY, ), ], ) result = pdp_backend.is_authorized_many_resources_and_actions( resource_actions=list_of_resource_actions, ) assert result == [True, False] mock_ows_pdp_client.check_my_resources.assert_called_once_with( check_resources_request=list_of_resource_actions_as_check_resources_request, ) def test_is_authorized_many_resources_and_actions_raise_when_unauthorized( list_of_resource_actions: list[ResourceAction], list_of_resource_actions_as_check_resources_request: CheckResourcesRequest, mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, ) -> None: """Test is_authorized_many_resources_and_actions happy path when raise_when_unauthorized is True.""" mock_ows_pdp_client.check_my_resources.return_value = CheckResourcesResponse( request_id="testing", resources=[ CheckResourceActionResult( action="eat", resource=Resource(resource_id=ResourceId(1), resource_type="meal"), effect=AuthEffect.ALLOW, ), CheckResourceActionResult( action="walk", resource=Resource(resource_id=ResourceId("1"), resource_type="dog"), effect=AuthEffect.DENY, ), ], ) with pytest.raises(Exception) as e: pdp_backend.is_authorized_many_resources_and_actions( resource_actions=list_of_resource_actions, raise_when_unauthorized=True, ) assert "Unauthorized for at least one resource." in str(e.value) mock_ows_pdp_client.check_my_resources.assert_called_once_with( check_resources_request=list_of_resource_actions_as_check_resources_request, ) @pytest.mark.parametrize("raise_when_unauthorized", [(True), (False)]) def test_is_authorized_many_resources_and_actions__error( list_of_resource_actions: list[ResourceAction], list_of_resource_actions_as_check_resources_request: CheckResourcesRequest, mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, caplog: pytest.LogCaptureFixture, raise_when_unauthorized: bool, ) -> None: """Test is_authorized_many_resources_and_actions PDP lookup failure case.""" caplog.set_level(logging.WARNING) mock_ows_pdp_client.check_my_resources.side_effect = RuntimeError( "In a big country..." ) if raise_when_unauthorized: with pytest.raises(Exception) as e: _ = pdp_backend.is_authorized_many_resources_and_actions( resource_actions=list_of_resource_actions, raise_when_unauthorized=raise_when_unauthorized, ) assert "Implicitly denied by an upstream exception" in str(e.value) else: result = pdp_backend.is_authorized_many_resources_and_actions( resource_actions=list_of_resource_actions, raise_when_unauthorized=raise_when_unauthorized, ) assert result == [False, False] assert "In a big country..." in caplog.text mock_ows_pdp_client.check_my_resources.assert_called_once_with( check_resources_request=list_of_resource_actions_as_check_resources_request, ) @pytest.mark.parametrize("raise_when_unauthorized", [True, False]) def test_is_authorized_many_resources_and_actions__empty_resources( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, raise_when_unauthorized: bool, ) -> None: """Test is_authorized_many_resources_and_actions PDP lookup failure case.""" with pytest.raises(InvalidRequestException) as e: pdp_backend.is_authorized_many_resources_and_actions( resource_actions=[], raise_when_unauthorized=raise_when_unauthorized, ) assert "resource_actions must be a non-empty list." in str(e.value) mock_ows_pdp_client.check_my_resources.assert_not_called() @pytest.mark.parametrize( "raise_when_unauthorized", [ pytest.param(False, id="raise_when_unauthorized=False"), pytest.param(True, id="raise_when_unauthorized=True"), ], ) def test_is_authorized_propagates_authentication_error( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, raise_when_unauthorized: bool, ) -> None: """Test that PdpAuthenticationError propagates regardless of raise_when_unauthorized flag.""" mock_ows_pdp_client.check_my_resources.side_effect = PdpAuthenticationError( "Authentication failed when checking resources: invalid token" ) mock_resource_getter = MagicMock() mock_resource_getter.get_attributes.return_value = {"some": "attributes"} with pytest.raises(UnauthenticatedException) as e: pdp_backend.is_authorized( action="view", resource_id="123", resource_type="audience", resource_getter=mock_resource_getter, raise_when_unauthorized=raise_when_unauthorized, ) assert "Authentication Error" in str(e.value) def test_get_authorized_tenants_propagates_authentication_error( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, ) -> None: """Test that PdpAuthenticationError propagates in get_authorized_tenants.""" mock_ows_pdp_client.get_allowed_tenants.side_effect = PdpAuthenticationError( "Authentication failed when getting allowed tenants: invalid token" ) with pytest.raises(UnauthenticatedException) as e: pdp_backend.get_authorized_tenants( action="view", resource_type="audience", ) assert "Authentication Error" in str(e.value) @pytest.mark.parametrize( "raise_when_unauthorized", [ pytest.param(False, id="raise_when_unauthorized=False"), pytest.param(True, id="raise_when_unauthorized=True"), ], ) def test_is_authorized_many_propagates_authentication_error( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, raise_when_unauthorized: bool, ) -> None: """Test that PdpAuthenticationError propagates regardless of raise_when_unauthorized flag.""" mock_ows_pdp_client.check_my_resources.side_effect = PdpAuthenticationError( "Authentication failed when checking resources: invalid token" ) with pytest.raises(UnauthenticatedException) as e: pdp_backend.is_authorized_many( action="view", resource_type="audience", resources_with_attributes=[ ResourceWithAttributes( resource_id="123", attributes={"tenant": {"tenant_type": "account"}}, ), ], raise_when_unauthorized=raise_when_unauthorized, ) assert "Authentication Error" in str(e.value) @pytest.mark.parametrize( "raise_when_unauthorized", [ pytest.param(False, id="raise_when_unauthorized=False"), pytest.param(True, id="raise_when_unauthorized=True"), ], ) def test_is_authorized_many_resources_and_actions_propagates_authentication_error( mock_ows_pdp_client: MagicMock, pdp_backend: PdpAuthorizationBackend, raise_when_unauthorized: bool, ) -> None: """Test that PdpAuthenticationError propagates regardless of raise_when_unauthorized flag.""" mock_ows_pdp_client.check_my_resources.side_effect = PdpAuthenticationError( "Authentication failed when checking resources: invalid token" ) with pytest.raises(UnauthenticatedException) as e: pdp_backend.is_authorized_many_resources_and_actions( resource_actions=[ ResourceAction( resource_id="123", resource_type="audience", action="view", attributes={"tenant": {"tenant_type": "account"}}, ), ], raise_when_unauthorized=raise_when_unauthorized, ) assert "Authentication Error" in str(e.value) def test_authorization_backend_protocol_is_runtime_checkable( pdp_backend: PdpAuthorizationBackend, ) -> None: """PdpAuthorizationBackend satisfies AuthorizationBackend via isinstance.""" assert isinstance(pdp_backend, AuthorizationBackend) def test_authorization_backend_protocol_rejects_objects_missing_methods() -> None: """isinstance(...) returns False for objects that don't implement the interface.""" assert not isinstance(object(), AuthorizationBackend)