fix(restriction_policies): drop stale principals instead of skipping the whole policy (2/4)

datadog_sync/model/restriction_policies.py

@@ -4,10 +4,15 @@
 # Copyright 2019 Datadog, Inc.
 
 from __future__ import annotations
+from collections import defaultdict
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple
 
-from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
-from datadog_sync.utils.resource_utils import CustomClientHTTPError, SkipResource
+from datadog_sync.utils.base_resource import BaseResource, ResourceConfig, ResourceConnectionResult
+from datadog_sync.utils.resource_utils import (
+    CustomClientHTTPError,
+    SkipResource,
+    find_attr,
+)
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
@@ -147,6 +152,38 @@ 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.
+
+        The "id" connections (dashboards/slos/notebooks) keep the generic
+        find_attr/connect_id hard-fail path. The bindings' composite principals go through
+        the shared per-binding filter so that principals permanently absent from source can
+        be dropped (under --drop-unresolvable-principals) instead of failing the whole
+        policy, while an emptied binding 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 == "attributes.bindings.principals":
+                    continue  # handled per-binding below
+                c = find_attr(attr_connection, resource_to_connect, resource, self.connect_id)
+                if c:
+                    failed_connections_dict[resource_to_connect].extend(c)
+
+        bindings = (resource.get("attributes") or {}).get("bindings")
+        principal_failed, empty_binding_risk = self._filter_stale_binding_principals(_id, bindings)
+        for rt, ids in principal_failed.items():
+            failed_connections_dict[rt].extend(ids)
+
+        return ResourceConnectionResult(
+            empty_binding_escalation=self._raise_connection_error_if_any(
+                _id, failed_connections_dict, empty_binding_risk
+            )
+        )
+
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         dashboards = self.config.state.destination["dashboards"]
         slos = self.config.state.destination["service_level_objectives"]
@@ -198,6 +235,11 @@ 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: this must keep
+        # returning every referenced id (including ones connect_resources will later drop),
+        # because --minimize-reads lazy-loading relies on it to decide WHICH principals to
+        # look up in source in order to make the drop decision. Do NOT "fix" this to match
+        # connect_resources' filtering.
         if key == "id":
             _type, _id = r_obj[key].split(":", 1)
             type_map = {"dashboard": "dashboards", "slo": "service_level_objectives", "notebook": "notebooks"}

tests/unit/test_restriction_policies.py

@@ -11,10 +11,13 @@
 """
 
 import asyncio
+from collections import defaultdict
 import pytest
 from unittest.mock import AsyncMock, MagicMock
 
 from datadog_sync.model.restriction_policies import RestrictionPolicies
+from datadog_sync.utils.resource_utils import ResourceConnectionError
+from datadog_sync.utils.workers import Counter
 
 
 class TestRestrictionPoliciesOrgPrincipal:
@@ -99,3 +102,199 @@ def test_pre_resource_action_hook_skips_when_org_principal_none(self):
 
         assert r["attributes"]["bindings"][0]["principals"][0] == "org:source-pub-id"
         assert r["attributes"]["bindings"][0]["principals"][1] == "user:some-user"
+
+
+class TestRestrictionPoliciesConnectResources:
+    """connect_resources drop/keep/hard-fail behavior for --drop-unresolvable-principals.
+
+    'stale' == absent from BOTH destination and source (permanently gone).
+    'pending' == present in source, absent from destination ('not yet synced').
+    'valid' == present in destination (resolves and remaps).
+    """
+
+    def _make_resource(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()
+        resource = RestrictionPolicies(config)
+        # Seed the primary "id" connection so it always resolves; keeps these tests
+        # focused on principal handling rather than the dashboard id link.
+        config.state.destination["dashboards"]["dash-src"] = {"id": "dash-dst"}
+        return resource
+
+    def _seed_valid_role(self, resource, src="role-good", dst="role-good-dst"):
+        resource.config.state.destination["roles"][src] = {"id": dst}
+
+    def _seed_pending_role(self, resource, src="role-pending"):
+        resource.config.state.source["roles"][src] = {"id": src}
+
+    def _policy(self, bindings):
+        return {"id": "dashboard:dash-src", "attributes": {"bindings": bindings}}
+
+    # --- flag OFF: byte-for-byte unchanged behavior --------------------------------
+
+    def test_flag_off_stale_principal_hard_fails(self):
+        resource = self._make_resource(drop=False)
+        policy = self._policy([{"principals": ["role:role-gone"], "relation": "editor"}])
+
+        with pytest.raises(ResourceConnectionError):
+            resource.connect_resources("dashboard:dash-src", policy)
+
+        resource.config.counter  # no drop recorded
+        assert resource.config.counter.stale_principals_dropped_by_type == {}
+
+    def test_flag_off_valid_principal_remaps_and_succeeds(self):
+        resource = self._make_resource(drop=False)
+        self._seed_valid_role(resource)
+        policy = self._policy([{"principals": ["role:role-good"], "relation": "editor"}])
+
+        resource.connect_resources("dashboard:dash-src", policy)  # no raise
+
+        assert policy["attributes"]["bindings"][0]["principals"] == ["role:role-good-dst"]
+
+    # --- flag ON: drop stale, keep syncing -----------------------------------------
+
+    def test_flag_on_drops_stale_keeps_valid_and_syncs(self):
+        resource = self._make_resource(drop=True)
+        self._seed_valid_role(resource)
+        policy = self._policy([{"principals": ["role:role-good", "role:role-gone"], "relation": "editor"}])
+
+        resource.connect_resources("dashboard:dash-src", policy)  # no raise
+
+        assert policy["attributes"]["bindings"][0]["principals"] == ["role:role-good-dst"]
+        resource.config.logger.warning.assert_called()
+        assert resource.config.counter.stale_principals_dropped_by_type["restriction_policies"] == [
+            "dashboard:dash-src"
+        ]
+
+    def test_flag_on_pending_principal_still_hard_fails(self):
+        resource = self._make_resource(drop=True)
+        self._seed_pending_role(resource)  # in source, not destination
+        policy = self._policy([{"principals": ["role:role-pending"], "relation": "editor"}])
+
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            resource.connect_resources("dashboard:dash-src", policy)
+
+        # Not an access-elevation case -- it's the legitimate retry-later path.
+        assert exc_info.value.empty_binding_risk is False
+        assert resource.config.counter.stale_principals_dropped_by_type == {}
+
+    def test_flag_on_binding_emptied_raises_empty_binding_risk(self):
+        resource = self._make_resource(drop=True)
+        policy = self._policy([{"principals": ["role:role-gone"], "relation": "editor"}])
+
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            resource.connect_resources("dashboard:dash-src", policy)
+
+        assert exc_info.value.empty_binding_risk is True
+        resource.config.logger.error.assert_called()
+        assert policy["attributes"]["bindings"][0]["principals"] == []
+
+    def test_multiple_bindings_only_one_empties_still_raises(self):
+        resource = self._make_resource(drop=True)
+        self._seed_valid_role(resource)
+        policy = self._policy(
+            [
+                {"principals": ["role:role-good"], "relation": "viewer"},
+                {"principals": ["role:role-gone"], "relation": "editor"},
+            ]
+        )
+
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            resource.connect_resources("dashboard:dash-src", policy)
+
+        assert exc_info.value.empty_binding_risk is True
+        # Resource-level skip: the surviving binding was still filtered/remapped in place.
+        assert policy["attributes"]["bindings"][0]["principals"] == ["role:role-good-dst"]
+        assert policy["attributes"]["bindings"][1]["principals"] == []
+
+    def test_skip_failed_resource_connections_suppresses_empty_binding_raise(self):
+        resource = self._make_resource(drop=True, skip_failed=True)
+        policy = self._policy([{"principals": ["role:role-gone"], "relation": "editor"}])
+
+        # skip_failed_resource_connections stays authoritative: no raise even for the
+        # empty-binding risk case.
+        connection_result = resource.connect_resources("dashboard:dash-src", policy)
+
+        assert policy["attributes"]["bindings"][0]["principals"] == []
+        assert connection_result.empty_binding_escalation is True
+        message = resource.config.logger.error.call_args.args[0]
+        assert "continuing sync" in message
+        assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" in message
+        assert "refusing to sync" not in message
+
+    def test_middle_principal_drop_does_not_skip_neighbors(self):
+        # Enumerate/index-shift regression: stale entry in the MIDDLE of 3.
+        resource = self._make_resource(drop=True)
+        self._seed_valid_role(resource, src="role-a", dst="role-a-dst")
+        self._seed_valid_role(resource, src="role-c", dst="role-c-dst")
+        policy = self._policy([{"principals": ["role:role-a", "role:role-gone", "role:role-c"], "relation": "editor"}])
+
+        resource.connect_resources("dashboard:dash-src", policy)  # no raise
+
+        # Both neighbors survive, in order; the middle stale one is gone.
+        assert policy["attributes"]["bindings"][0]["principals"] == [
+            "role:role-a-dst",
+            "role:role-c-dst",
+        ]
+
+    def test_extract_source_ids_unaffected_by_drop_logic(self):
+        # extract_source_ids must still surface an id that connect_resources would drop.
+        resource = self._make_resource(drop=True)
+        binding = {"principals": ["role:role-gone"], "relation": "editor"}
+
+        assert resource.extract_source_ids("principals", binding, "roles") == ["role-gone"]
+
+    def test_non_principal_id_connection_still_hard_fails(self):
+        # A dangling dashboard "id" link fails via the generic path regardless of the flag.
+        resource = self._make_resource(drop=True)
+        self._seed_valid_role(resource)
+        # Point at a dashboard id that is NOT in destination state.
+        policy = {
+            "id": "dashboard:missing-dash",
+            "attributes": {"bindings": [{"principals": ["role:role-good"], "relation": "editor"}]},
+        }
+
+        with pytest.raises(ResourceConnectionError) as exc_info:
+            resource.connect_resources("dashboard:missing-dash", policy)
+
+        assert exc_info.value.empty_binding_risk is False
+
+    def test_inert_when_off_end_to_end_raises_like_today(self):
+        # Full connect_resources with the flag off + a would-be-dropped principal must
+        # raise exactly as today (no silent drop).
+        resource = self._make_resource(drop=False)
+        policy = self._policy([{"principals": ["role:role-gone"], "relation": "editor"}])
+
+        with pytest.raises(ResourceConnectionError):
+            resource.connect_resources("dashboard:dash-src", policy)
+
+        # Flag off => principal is NOT dropped; it stays in the (rebuilt) list.
+        assert policy["attributes"]["bindings"][0]["principals"] == ["role:role-gone"]
+
+    def test_org_and_non_composite_principals_pass_through(self):
+        # org: principals (already remapped by the hook) and any non-user/role/team or
+        # colon-less token must pass through untouched; an empty-principals binding is skipped.
+        resource = self._make_resource(drop=True)
+        self._seed_valid_role(resource)
+        policy = self._policy(
+            [
+                {"principals": ["org:dest-org", "weirdnoprefix", "role:role-good"], "relation": "editor"},
+                {"principals": [], "relation": "viewer"},
+            ]
+        )
+
+        resource.connect_resources("dashboard:dash-src", policy)  # no raise
+
+        assert policy["attributes"]["bindings"][0]["principals"] == [
+            "org:dest-org",
+            "weirdnoprefix",
+            "role:role-good-dst",
+        ]
+        assert policy["attributes"]["bindings"][1]["principals"] == []