fix(monitors,synthetics_tests): drop stale restricted_roles/principals under new flag (3/4)

datadog_sync/model/monitors.py

@@ -4,14 +4,15 @@
 # Copyright 2019 Datadog, Inc.
 
 from __future__ import annotations
+from collections import defaultdict
 import logging
 import re
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, cast
 
 from datadog_sync.constants import LOGGER_NAME
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, TaggingConfig
 from datadog_sync.utils.custom_client import PaginationConfig
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource
+from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource, find_attr
 
 _log = logging.getLogger(LOGGER_NAME)
 
@@ -238,6 +239,45 @@ async def delete_resource(self, _id: str) -> None:
             params={"force": "true"},
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> None:
+        """Drop-aware override.
+
+        query / composite-monitor / slo-alert connections keep the generic
+        find_attr/connect_id path. The two access-control shapes -- the flat
+        `restricted_roles` list and the `restriction_policy.bindings.principals` composites
+        -- go through the shared drop-aware filters so permanently-stale references can be
+        dropped (under --drop-unresolvable-principals) while an emptied binding/list still
+        hard-fails as an access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection in ("restricted_roles", "restriction_policy.bindings.principals"):
+                    continue  # handled by the drop-aware filters below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        empty_risk = False
+        restriction_policy = resource.get("restriction_policy")
+        if restriction_policy:
+            principal_failed, binding_risk = self._filter_stale_binding_principals(
+                _id, restriction_policy.get("bindings")
+            )
+            for rt, ids in principal_failed.items():
+                failed_connections_dict[rt].extend(ids)
+            empty_risk = empty_risk or binding_risk
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource, "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+        empty_risk = empty_risk or roles_risk
+
+        self._raise_connection_error_if_any(_id, failed_connections_dict, empty_risk)
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         monitors = self.config.state.destination[resource_to_connect]
 
@@ -308,6 +348,9 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         # Mirror of connect_id -- keep in sync when connect_id changes.
+        # Intentionally UNAFFECTED by --drop-unresolvable-principals: must keep returning
+        # every referenced id (including ones connect_resources will later drop) so
+        # --minimize-reads lazy-loading can look them up in source to make the drop decision.
         if key == "query" and r_obj.get("type") == "composite" and resource_to_connect != "service_level_objectives":
             return re.findall("[0-9]+", r_obj[key])
         elif key == "query" and resource_to_connect == "service_level_objectives" and r_obj.get("type") == "slo alert":

datadog_sync/model/synthetics_tests.py

@@ -9,11 +9,13 @@
 import certifi
 import json
 import ssl
+from collections import defaultdict
 from copy import deepcopy
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, cast
 from yarl import URL
 
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, TaggingConfig
+from datadog_sync.utils.resource_utils import find_attr
 from datadog_sync.model.synthetics_mobile_applications_versions import SyntheticsMobileApplicationsVersions
 
 if TYPE_CHECKING:
@@ -422,6 +424,45 @@ async def delete_resource(self, _id: str) -> None:
         dest_resource = self.config.state.destination[self.resource_type][_id]
         await self._delete_test(destination_client, dest_resource.get("type"), dest_resource["public_id"])
 
+    def connect_resources(self, _id: str, resource: Dict) -> None:
+        """Drop-aware override.
+
+        All non-access-control connections (private locations, subtests, global variables,
+        rum/mobile apps) keep the generic find_attr/connect_id path. The flat
+        `options.restricted_roles` list and the `restriction_policy.bindings.principals`
+        composites go through the shared drop-aware filters so permanently-stale references
+        can be dropped (under --drop-unresolvable-principals) while an emptied binding/list
+        still hard-fails as an access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return
+
+        failed_connections_dict = defaultdict(list)
+        for resource_to_connect, attrs in self.resource_config.resource_connections.items():
+            for attr_connection in attrs:
+                if attr_connection in ("options.restricted_roles", "restriction_policy.bindings.principals"):
+                    continue  # handled by the drop-aware filters below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        empty_risk = False
+        restriction_policy = resource.get("restriction_policy")
+        if restriction_policy:
+            principal_failed, binding_risk = self._filter_stale_binding_principals(
+                _id, restriction_policy.get("bindings")
+            )
+            for rt, ids in principal_failed.items():
+                failed_connections_dict[rt].extend(ids)
+            empty_risk = empty_risk or binding_risk
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource.get("options"), "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+        empty_risk = empty_risk or roles_risk
+
+        self._raise_connection_error_if_any(_id, failed_connections_dict, empty_risk)
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         failed_connections: List[str] = []
         if resource_to_connect == "synthetics_private_locations":
@@ -504,6 +545,9 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         # Mirror of connect_id -- keep in sync when connect_id changes.
+        # Intentionally UNAFFECTED by --drop-unresolvable-principals: must keep returning
+        # every referenced id (including ones connect_resources will later drop) so
+        # --minimize-reads lazy-loading can look them up in source to make the drop decision.
         # Only synthetics_private_locations and mobile application versions need special handling.
         # rum_applications, synthetics_tests (subtests), synthetics_global_variables, roles, and
         # synthetics_mobile_applications (applicationId key) all use plain IDs at the leaf —

tests/unit/test_monitors.py

@@ -11,11 +11,13 @@
 """
 
 import asyncio
+from collections import defaultdict
 import pytest
 from unittest.mock import MagicMock
 
 from datadog_sync.model.monitors import Monitors
-from datadog_sync.utils.resource_utils import SkipResource
+from datadog_sync.utils.resource_utils import SkipResource, ResourceConnectionError
+from datadog_sync.utils.workers import Counter
 
 
 class TestMonitorsPreResourceActionHook:
@@ -635,3 +637,95 @@ def test_pre_apply_hook_skips_current_user_when_source_empty(self):
         asyncio.run(monitors.pre_apply_hook())
         assert monitors.org_principal is None
         mock_client.get.assert_not_awaited()
+
+
+class TestMonitorsConnectResourcesDrop:
+    """connect_resources drop/keep/hard-fail for monitors' access-control shapes:
+    the flat `restricted_roles` list and `restriction_policy.bindings.principals` composites.
+    'stale' == absent from destination AND source; 'valid' == present in destination.
+    """
+
+    def _make_monitors(self, drop=False, skip_failed=False):
+        config = MagicMock()
+        config.state = MagicMock()
+        config.state.source = defaultdict(dict)
+        config.state.destination = defaultdict(dict)
+        config.state.ensure_resource_loaded = MagicMock()
+        config.drop_unresolvable_principals = drop
+        config.skip_failed_resource_connections = skip_failed
+        config.counter = Counter()
+        config.logger = MagicMock()
+        return Monitors(config)
+
+    def _seed_valid_role(self, m, src="role-good", dst="role-good-dst"):
+        m.config.state.destination["roles"][src] = {"id": dst}
+
+    def test_restriction_policy_flag_off_stale_hard_fails(self):
+        m = self._make_monitors(drop=False)
+        resource = {
+            "id": 1,
+            "query": "avg(last_5m):sum:x{*} > 1",
+            "restriction_policy": {"bindings": [{"principals": ["role:role-gone"], "relation": "editor"}]},
+        }
+        with pytest.raises(ResourceConnectionError):
+            m.connect_resources("1", resource)
+
+    def test_restriction_policy_flag_on_drops_stale(self):
+        m = self._make_monitors(drop=True)
+        self._seed_valid_role(m)
+        resource = {
+            "id": 1,
+            "query": "avg(last_5m):sum:x{*} > 1",
+            "restriction_policy": {
+                "bindings": [{"principals": ["role:role-good", "role:role-gone"], "relation": "editor"}]
+            },
+        }
+        m.connect_resources("1", resource)  # no raise
+        assert resource["restriction_policy"]["bindings"][0]["principals"] == ["role:role-good-dst"]
+        assert m.config.counter.stale_principals_dropped_by_type["monitors"] == ["1"]
+
+    def test_restriction_policy_empty_binding_raises_risk(self):
+        m = self._make_monitors(drop=True)
+        resource = {
+            "id": 1,
+            "restriction_policy": {"bindings": [{"principals": ["role:role-gone"], "relation": "editor"}]},
+        }
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            m.connect_resources("1", resource)
+        assert exc_info.value.empty_binding_risk is True
+
+    def test_restricted_roles_flat_flag_on_drops_stale(self):
+        m = self._make_monitors(drop=True)
+        self._seed_valid_role(m, src="role-good", dst="role-good-dst")
+        resource = {"id": 1, "restricted_roles": ["role-good", "role-gone"]}
+        m.connect_resources("1", resource)  # no raise
+        assert resource["restricted_roles"] == ["role-good-dst"]
+
+    def test_restricted_roles_flat_all_stale_empty_risk(self):
+        m = self._make_monitors(drop=True)
+        resource = {"id": 1, "restricted_roles": ["role-gone"]}
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            m.connect_resources("1", resource)
+        assert exc_info.value.empty_binding_risk is True
+        assert resource["restricted_roles"] == []
+
+    def test_restricted_roles_flat_middle_drop_keeps_neighbors(self):
+        m = self._make_monitors(drop=True)
+        self._seed_valid_role(m, src="ra", dst="ra-dst")
+        self._seed_valid_role(m, src="rc", dst="rc-dst")
+        resource = {"id": 1, "restricted_roles": ["ra", "role-gone", "rc"]}
+        m.connect_resources("1", resource)  # no raise
+        assert resource["restricted_roles"] == ["ra-dst", "rc-dst"]
+
+    def test_flag_off_restricted_roles_unchanged_behavior(self):
+        m = self._make_monitors(drop=False)
+        resource = {"id": 1, "restricted_roles": ["role-gone"]}
+        with pytest.raises(ResourceConnectionError):
+            m.connect_resources("1", resource)
+        # Flag off => not dropped
+        assert resource["restricted_roles"] == ["role-gone"]
+
+    def test_extract_source_ids_principals_unaffected(self):
+        m = self._make_monitors(drop=True)
+        binding = {"principals": ["role:role-gone"], "relation": "editor"}
+        assert m.extract_source_ids("principals", binding, "roles") == ["role-gone"]

tests/unit/test_synthetics_tests.py

@@ -15,11 +15,14 @@
 
 import asyncio
 import copy
+from collections import defaultdict
 
 import pytest
 from unittest.mock import AsyncMock, MagicMock
 from datadog_sync.model.synthetics_tests import SyntheticsTests
 from datadog_sync.utils.configuration import Configuration
+from datadog_sync.utils.resource_utils import ResourceConnectionError
+from datadog_sync.utils.workers import Counter
 
 
 class TestSyntheticsTestsStatusBehavior:
@@ -476,3 +479,84 @@ def test_extract_source_ids_org_excluded(self):
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
+
+
+class TestSyntheticsTestsConnectResourcesDrop:
+    """connect_resources drop/keep/hard-fail for synthetics_tests' access-control shapes:
+    flat `options.restricted_roles` and `restriction_policy.bindings.principals` composites.
+    """
+
+    def _make_test(self, drop=False, skip_failed=False):
+        config = MagicMock()
+        config.state = MagicMock()
+        config.state.source = defaultdict(dict)
+        config.state.destination = defaultdict(dict)
+        config.state.ensure_resource_loaded = MagicMock()
+        config.drop_unresolvable_principals = drop
+        config.skip_failed_resource_connections = skip_failed
+        config.counter = Counter()
+        config.logger = MagicMock()
+        return SyntheticsTests(config)
+
+    def _seed_valid_role(self, t, src="role-good", dst="role-good-dst"):
+        t.config.state.destination["roles"][src] = {"id": dst}
+
+    def test_restriction_policy_flag_on_drops_stale(self):
+        t = self._make_test(drop=True)
+        self._seed_valid_role(t)
+        resource = {
+            "public_id": "abc-def-ghi",
+            "restriction_policy": {
+                "bindings": [{"principals": ["role:role-good", "role:role-gone"], "relation": "editor"}]
+            },
+        }
+        t.connect_resources("abc-def-ghi", resource)  # no raise
+        assert resource["restriction_policy"]["bindings"][0]["principals"] == ["role:role-good-dst"]
+        assert t.config.counter.stale_principals_dropped_by_type["synthetics_tests"] == ["abc-def-ghi"]
+
+    def test_restriction_policy_flag_off_stale_hard_fails(self):
+        t = self._make_test(drop=False)
+        resource = {
+            "public_id": "abc-def-ghi",
+            "restriction_policy": {"bindings": [{"principals": ["role:role-gone"], "relation": "editor"}]},
+        }
+        with pytest.raises(ResourceConnectionError):
+            t.connect_resources("abc-def-ghi", resource)
+
+    def test_restriction_policy_empty_binding_raises_risk(self):
+        t = self._make_test(drop=True)
+        resource = {
+            "public_id": "abc-def-ghi",
+            "restriction_policy": {"bindings": [{"principals": ["role:role-gone"], "relation": "editor"}]},
+        }
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            t.connect_resources("abc-def-ghi", resource)
+        assert exc_info.value.empty_binding_risk is True
+
+    def test_options_restricted_roles_flat_drops_stale(self):
+        t = self._make_test(drop=True)
+        self._seed_valid_role(t, src="role-good", dst="role-good-dst")
+        resource = {"public_id": "abc", "options": {"restricted_roles": ["role-good", "role-gone"]}}
+        t.connect_resources("abc", resource)  # no raise
+        assert resource["options"]["restricted_roles"] == ["role-good-dst"]
+
+    def test_options_restricted_roles_all_stale_empty_risk(self):
+        t = self._make_test(drop=True)
+        resource = {"public_id": "abc", "options": {"restricted_roles": ["role-gone"]}}
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            t.connect_resources("abc", resource)
+        assert exc_info.value.empty_binding_risk is True
+        assert resource["options"]["restricted_roles"] == []
+
+    def test_options_restricted_roles_middle_drop_keeps_neighbors(self):
+        t = self._make_test(drop=True)
+        self._seed_valid_role(t, src="ra", dst="ra-dst")
+        self._seed_valid_role(t, src="rc", dst="rc-dst")
+        resource = {"public_id": "abc", "options": {"restricted_roles": ["ra", "role-gone", "rc"]}}
+        t.connect_resources("abc", resource)  # no raise
+        assert resource["options"]["restricted_roles"] == ["ra-dst", "rc-dst"]
+
+    def test_extract_source_ids_principals_unaffected(self):
+        t = self._make_test(drop=True)
+        binding = {"principals": ["role:role-gone"], "relation": "editor"}
+        assert t.extract_source_ids("principals", binding, "roles") == ["role-gone"]