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

datadog_sync/model/dashboards.py

@@ -4,11 +4,18 @@
 # Copyright 2019 Datadog, Inc.
 
 from __future__ import annotations
+from collections import defaultdict
 from copy import deepcopy
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple, cast
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource, check_diff, prep_resource
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult
+from datadog_sync.utils.resource_utils import (
+    CustomClientHTTPError,
+    SkipResource,
+    check_diff,
+    find_attr,
+    prep_resource,
+)
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
@@ -172,5 +179,36 @@ async def delete_resource(self, _id: str) -> None:
             self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override.
+
+        Widget connections (monitor alert_ids, powerpacks, slos) keep the generic
+        find_attr/connect_id path. The flat `restricted_roles` list goes through the shared
+        drop-aware filter so a permanently-stale role can be dropped (under
+        --drop-unresolvable-principals) while an emptied list still hard-fails as an
+        access-elevation guard.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        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 == "restricted_roles":
+                    continue  # handled by the drop-aware filter below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource, "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, roles_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         return super(Dashboards, self).connect_id(key, r_obj, resource_to_connect)

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.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult, 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,49 @@ async def delete_resource(self, _id: str) -> None:
             params={"force": "true"},
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """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 ResourceConnectionResult()
+
+        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
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=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 +352,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_private_locations.py

@@ -6,11 +6,12 @@
 from __future__ import annotations
 import json
 import re
+from collections import defaultdict
 
 from typing import TYPE_CHECKING, List, Dict, Optional, Tuple
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, TaggingConfig
-from datadog_sync.utils.resource_utils import SkipResource
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult, TaggingConfig
+from datadog_sync.utils.resource_utils import SkipResource, find_attr
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
@@ -102,5 +103,34 @@ async def delete_resource(self, _id: str) -> None:
             self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
         )
 
+    def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult:
+        """Drop-aware override for the flat `metadata.restricted_roles` list.
+
+        A permanently-stale role can be dropped (under --drop-unresolvable-principals) while
+        an emptied list still hard-fails as an access-elevation guard. Any future
+        non-restricted_roles connection keeps the generic find_attr/connect_id path.
+        """
+        if not self.resource_config.resource_connections:
+            return ResourceConnectionResult()
+
+        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 == "metadata.restricted_roles":
+                    continue  # handled by the drop-aware filter below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        role_failed, roles_risk = self._filter_stale_flat_roles(_id, resource.get("metadata"), "restricted_roles")
+        if role_failed:
+            failed_connections_dict["roles"].extend(role_failed)
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, roles_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         return super(SyntheticsPrivateLocations, self).connect_id(key, r_obj, resource_to_connect)

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.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult, 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,49 @@ 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) -> ResourceConnectionResult:
+        """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 ResourceConnectionResult()
+
+        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
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=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 +549,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_dashboards.py

@@ -12,12 +12,14 @@
 """
 
 import asyncio
+from collections import defaultdict
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from datadog_sync.model.dashboards import Dashboards
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource
+from datadog_sync.utils.resource_utils import CustomClientHTTPError, ResourceConnectionError, SkipResource
+from datadog_sync.utils.workers import Counter
 
 
 def _read_only_error() -> CustomClientHTTPError:
@@ -285,3 +287,65 @@ def test_non_403_error_reraised(self):
         with pytest.raises(CustomClientHTTPError):
             asyncio.run(dashboards.update_resource("abc-123", {"title": "src"}))
         dashboards.config.destination_client.post.assert_not_awaited()
+
+
+class TestDashboardsConnectResourcesDrop:
+    """connect_resources drop/keep/hard-fail for dashboards' flat `restricted_roles`."""
+
+    def _make_dashboard(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 Dashboards(config)
+
+    def _seed_valid_role(self, d, src="role-good", dst="role-good-dst"):
+        d.config.state.destination["roles"][src] = {"id": dst}
+
+    def test_flag_off_stale_role_hard_fails(self):
+        d = self._make_dashboard(drop=False)
+        resource = {"id": "abc-def-ghi", "restricted_roles": ["role-gone"]}
+        with pytest.raises(ResourceConnectionError):
+            d.connect_resources("abc-def-ghi", resource)
+        assert resource["restricted_roles"] == ["role-gone"]  # not dropped when flag off
+
+    def test_flag_on_drops_stale_keeps_valid(self):
+        d = self._make_dashboard(drop=True)
+        self._seed_valid_role(d)
+        resource = {"id": "abc-def-ghi", "restricted_roles": ["role-good", "role-gone"]}
+        d.connect_resources("abc-def-ghi", resource)  # no raise
+        assert resource["restricted_roles"] == ["role-good-dst"]
+        assert d.config.counter.stale_principals_dropped_by_type["dashboards"] == ["abc-def-ghi"]
+
+    def test_flag_on_all_stale_empty_list_raises_risk(self):
+        d = self._make_dashboard(drop=True)
+        resource = {"id": "abc-def-ghi", "restricted_roles": ["role-gone"]}
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            d.connect_resources("abc-def-ghi", resource)
+        assert exc_info.value.empty_binding_risk is True
+        assert resource["restricted_roles"] == []
+
+    def test_empty_list_risk_is_returned_when_connection_failure_is_suppressed(self):
+        d = self._make_dashboard(drop=True, skip_failed=True)
+        resource = {"id": "abc-def-ghi", "restricted_roles": ["role-gone"]}
+
+        assert d.connect_resources("abc-def-ghi", resource).empty_binding_escalation is True
+        assert resource["restricted_roles"] == []
+
+    def test_flag_on_middle_drop_keeps_neighbors(self):
+        d = self._make_dashboard(drop=True)
+        self._seed_valid_role(d, src="ra", dst="ra-dst")
+        self._seed_valid_role(d, src="rc", dst="rc-dst")
+        resource = {"id": "abc", "restricted_roles": ["ra", "role-gone", "rc"]}
+        d.connect_resources("abc", resource)  # no raise
+        assert resource["restricted_roles"] == ["ra-dst", "rc-dst"]
+
+    def test_no_restricted_roles_is_noop(self):
+        d = self._make_dashboard(drop=True)
+        resource = {"id": "abc"}  # no restricted_roles at all
+        d.connect_resources("abc", resource)  # no raise, nothing to do