"""Integration tests for POST /identity/{identity_uuid}/check/resources.""" from typing import Any, Dict from uuid import uuid4 import pytest import requests from mypy_boto3_dynamodb import DynamoDBClient from tests.integration import config, utils from tests.integration.conftest import seed_test_pp_identity @pytest.mark.parametrize( "body, expected_resource_response, identity_tenant_type", [ pytest.param( { "resources": [ { "resource": { "resource_id": "890", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 } }, }, "action": "view", } ] }, { "resource": { "resource_id": "890", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, "account", id="""A resource with a tenant that the identity is an audience_development_analyst for. Expect allow.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "890", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "c5bb194c-e53e-11ee-b603-4a2888760682", # noqa: E501 } }, }, "action": "view", } ] }, { "resource": { "resource_id": "890", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, "account", id="""A resource with a tenant that the identity is an audience_development_analyst for and tenant UUID is UUIDv1. Expect allow.""", # noqa: E501 ), pytest.param( { "resources": [ { "resource": { "resource_id": "456", "resource_type": "ad_campaign", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "327ea8cb-caef-43b5-ab60-9a425b3f344e", # noqa: E501 } }, }, "action": "view", } ] }, { "resource": { "resource_id": "456", "resource_type": "ad_campaign", "attributes": {}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""A resource with a tenant that that the identity does not have a role for. Expect deny.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123", "resource_type": "junk", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 } }, }, "action": "view", } ] }, { "resource": { "resource_id": "123", "resource_type": "junk", "attributes": {}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""A resource with a tenant that the identity is an audience_development_analyst for, but has a non-existent resource type. Expect deny.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123", "resource_type": "audience", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "21bb9c7f-6d47-4216-8442-b098f8c62e59", # noqa: E501 "tenant_hierarchy": [ "42879e8c-9f47-4214-b611-1e6feb0be6af" ], }, }, }, "action": "view", } ] }, { "resource": { "resource_id": "123", "resource_type": "audience", "attributes": {}, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, "company_brand", id="""A resource with a tenant and hierarchy. The identity is an audience_development_analyst for a company_brand in the resource's tenant hierarchy. Expect allow.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 "tenant_hierarchy": [], }, }, }, "action": "view", } ] }, { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, "account", id="""A fan data list resource owned by a tenant that the identity has access to. The identity is an audience_development_analyst. Expect allow.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "52879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 "tenant_hierarchy": [], }, }, }, "action": "view", } ] }, { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""A fan data list resource owned by a tenant that the identity does not have access to. Expect deny.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "joint_venture": { "tenants": [ { "tenant_type": "account", "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 "tenant_hierarchy": [], "is_provider": True, "is_consumer": True, } ] } }, }, "action": "view", } ] }, { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, "account", id="""A fan data list resource owned by a joint venture with a tenant that the identity has access to.The identity is an audience_development_analyst. Expect allow.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "joint_venture": { "tenants": [ { "tenant_type": "account", "tenant_uuid": "242fa66d-193d-48b9-aaae-cefe7255b799", # noqa: E501 "tenant_hierarchy": [ "42879e8c-9f47-4214-b611-1e6feb0be6af" ], "is_provider": True, "is_consumer": True, } ] } }, }, "action": "view", } ] }, { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, "account", id="""A fan data list resource owned by a joint venture with a tenant in its hierarchy that the identity has access to. The identity is an audience_development_analyst. Expect allow.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "joint_venture": { "tenants": [ { "tenant_type": "account", "tenant_uuid": "242fa66d-193d-48b9-aaae-cefe7255b799", # noqa: E501 "tenant_hierarchy": [ # Randomly generated uuid "e5ea7ed7-3e98-4ddc-a5ca-909aaa74f5f7" ], "is_provider": True, "is_consumer": True, } ] } }, }, "action": "view", } ] }, { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""A fan data list resource owned by a joint venture with a tenant in its hierarchy that the identity does not have access to. Expect deny.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "joint_venture": { "tenants": [ { "tenant_type": "account", "tenant_uuid": "52879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 "tenant_hierarchy": [], "is_provider": True, "is_consumer": True, } ] } }, }, "action": "view", } ] }, { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""A fan data list resource owned by a joint venture with a tenant that the identity does not have access to. Expect deny.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "integration test user's ad connection", "resource_type": "ad_connection", "attributes": { "identity_uuid": "4d5f24f5-83f9-4989-9f82-0924a5feaf88" }, }, "action": "connect", } ], }, { "resource": { "resource_id": "integration test user's ad connection", "resource_type": "ad_connection", "attributes": {}, }, "action": "connect", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""An ad_connection resource is owned by the integration test identity. However, the identity does not have the correct role. Expect deny.""", ), pytest.param( { "resources": [ { "resource": { "resource_id": "Maggie's ad connection", "resource_type": "ad_connection", "attributes": { "identity_uuid": "ca034907-1746-4c7b-9c42-502dc4002d6c", }, }, "action": "connect", } ], }, { "resource": { "resource_id": "Maggie's ad connection", "resource_type": "ad_connection", "attributes": {}, }, "action": "connect", "effect": "deny", "errors": {"validation_errors": None}, }, "account", id="""An ad_connection resource is not owned by the integration test identity. Expect deny.""", ), ], ) def test_check_identity_resources( body: Dict[str, Any], expected_resource_response: Dict[str, Any], default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, identity_tenant_type: str, ) -> None: """POST /identity/{identity_uuid}/check/resources/ endpoint.""" url = f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/check/resources/" # noqa: E501 assert_check_identity_resources( url, default_boto_client, bearer_token_pdptest_user, bearer_token_pdptest_user_identity_uuid, body=body, expected_resource_response=expected_resource_response, identity_tenant_type=identity_tenant_type, ) def assert_check_identity_resources( url: str, boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, body: Dict[str, Any], expected_resource_response: Dict[str, Any], identity_tenant_type: str = "account", ) -> None: """Verify response gives expected allow or deny dependent on principal's access.""" seed_test_pp_identity( boto_client, bearer_token_pdptest_user_identity_uuid, "42879e8c-9f47-4214-b611-1e6feb0be6af", "audience_development_analyst", tenant_type=identity_tenant_type, ) # UUIDv1 entry seed_test_pp_identity( boto_client, bearer_token_pdptest_user_identity_uuid, "c5bb194c-e53e-11ee-b603-4a2888760682", "audience_development_analyst", tenant_type=identity_tenant_type, ) response = requests.post( url, json=body, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"} ) assert expected_resource_response in response.json()["resources"], response.text def test_check_identity_malformed_resource( default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Verify 200 with deny/validation error for malformed resource.""" url = f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/check/resources/" # noqa: E501 # The Fan Data List resource schema expects a tenant OR joint venture, not both body = { "resources": [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "52879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 "tenant_hierarchy": [], }, "joint_venture": { "tenants": [ { "tenant_type": "account", "tenant_uuid": "52879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 "tenant_hierarchy": [], "is_provider": True, "is_consumer": True, } ] }, }, }, "action": "view", } ] } response = requests.post( url, json=body, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"} ) assert response.status_code == 200, f"Response: {response.text}" assert response.json()["resources"] == [ { "resource": { "resource_id": "123fandatalist", "resource_type": "fan_data_list", "attributes": {}, }, "action": "view", "effect": "deny", "errors": { "validation_errors": [ { "path": "/", "message": "valid against schemas at indexes 0 and 1", "source": "SOURCE_RESOURCE", } ] }, } ] def test_include_resource_attributes( default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Verify the endpoint returns resource attributes.""" body = { "resources": [ { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af", # noqa: E501 } }, }, "action": "view", }, { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "28b788a3-90f0-4416-9126-de19b084e9e4", # noqa: E501 } }, }, "action": "view", }, ], "include_resource_attributes_in_response": True, } expected_resource_response = [ { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af", } }, }, "action": "view", "effect": "allow", "errors": {"validation_errors": None}, }, { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "28b788a3-90f0-4416-9126-de19b084e9e4", } }, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, ] seed_test_pp_identity( default_boto_client, bearer_token_pdptest_user_identity_uuid, "42879e8c-9f47-4214-b611-1e6feb0be6af", "audience_development_analyst", tenant_type="account", ) response = requests.post( f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/check/resources/", json=body, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, ) assert expected_resource_response == response.json()["resources"] @pytest.mark.parametrize( "user_bearer_token", [ pytest.param( "bearer_token_pdptest_rap_admin_user", id="RAP admin user can check own resources", ), pytest.param( "bearer_token_pdptest_not_rap_admin_user", id="Non RAP admin user can check own resources", ), ], ) def test_check_resources_rap_admin_or_not( user_bearer_token: str, default_boto_client: DynamoDBClient, request: pytest.FixtureRequest, ) -> None: """Verify RAP admin and non-RAP admin can check their own resources.""" body = { "resources": [ { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": { "tenant": { "tenant_type": "account", "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", # noqa: E501 } }, }, "action": "view", }, ], "include_resource_attributes_in_response": True, } bearer_token = request.getfixturevalue(user_bearer_token) bearer_token_identity_uuid = utils.get_bearer_token_identity_uuid(bearer_token) response = requests.post( f"{config.QA_BASE_URL}/identity/{bearer_token_identity_uuid}/check/resources/", json=body, headers={"Authorization": f"Bearer {bearer_token}"}, ) assert response.status_code == 200, f"Response: {response.text}" other_identity_response = requests.post( f"{config.QA_BASE_URL}/identity/{uuid4()}/check/resources/", json=body, headers={"Authorization": f"Bearer {bearer_token}"}, ) assert other_identity_response.status_code == 403, ( "RAP admin/non-RAP admin cannot check other's resources" ) @pytest.mark.parametrize( "attributes, expected_attributes", [ pytest.param( { "id_to_uuid_exchange_tenant": { "tenant_type": "account", "tenant_id": 10, } }, { "id_to_uuid_exchange_tenant": { "tenant_type": "account", "tenant_id": 10, }, "tenant": { "tenant_type": "account", "tenant_uuid": "3391af7b-6596-4820-a21d-911f8b5f64b4", "tenant_hierarchy": [ "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", "d25a4cd1-e820-45f2-be5c-56edcfeb8298", ], }, }, id="account is exchanged.", ), pytest.param( { "id_to_uuid_exchange_tenant": { "tenant_type": "subaccount", "tenant_id": 10, } }, { "id_to_uuid_exchange_tenant": { "tenant_type": "subaccount", "tenant_id": 10, }, "tenant": { "tenant_type": "subaccount", "tenant_uuid": "0778592c-c232-45af-bdce-af7fbbc1728e", "tenant_hierarchy": [ "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "c14e18a9-2c9a-4d3d-acfc-e5897e202fd8", ], }, }, id="subaccount is exchanged.", ), pytest.param( { "id_to_uuid_exchange_tenant": { "tenant_type": "company_brand", "tenant_id": 10, } }, { "id_to_uuid_exchange_tenant": { "tenant_type": "company_brand", "tenant_id": 10, }, "tenant": { "tenant_type": "company_brand", "tenant_uuid": "2ed1077d-bcd3-4db6-9a47-c37da82aab18", "tenant_hierarchy": [ "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", ], }, }, id="company_brand is exchanged.", ), ], ) def test_uses_update_with_id_to_uuid_exchange( attributes: dict[str, Any], expected_attributes: dict[str, Any], default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Verify the endpoint handles id_to_uuid_exchange_tenant in attribute.""" body = { "resources": [ { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": attributes, }, "action": "view", }, ], "include_resource_attributes_in_response": True, } expected_resource_response = [ { "resource": { "resource_id": "0", "resource_type": "fan_data_list", "attributes": expected_attributes, }, "action": "view", "effect": "deny", "errors": {"validation_errors": None}, }, ] seed_test_pp_identity( default_boto_client, bearer_token_pdptest_user_identity_uuid, "42879e8c-9f47-4214-b611-1e6feb0be6af", "audience_development_analyst", tenant_type="account", ) response = requests.post( f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/check/resources/", json=body, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, ) assert response.status_code == 200, response.text assert expected_resource_response == response.json()["resources"]