feat: --drop-unresolvable-principals flag + drop-aware connection scaffolding (1/4)

datadog_sync/commands/shared/options.py

@@ -499,6 +499,19 @@ def click_config_file_provider(ctx: Context, opts: CustomOptionClass, value: Non
         help="Skip resource if resource connection fails.",
         cls=CustomOptionClass,
     ),
+    option(
+        "--drop-unresolvable-principals",
+        required=False,
+        is_flag=True,
+        default=False,
+        show_default=True,
+        help="For restriction policies and restricted_roles lists: drop principal/role "
+        "references that are absent from BOTH destination and source state (permanently "
+        "gone, e.g. deleted before the org's first import) instead of skipping the whole "
+        "resource. If dropping empties a binding/list that had entries at the source, the "
+        "resource is still skipped (access-elevation guard). Off by default.",
+        cls=CustomOptionClass,
+    ),
     option(
         "--cleanup",
         default="False",

datadog_sync/utils/base_resource.py

@@ -437,6 +437,173 @@ def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optiona
 
         return failed_connections
 
+    def _resolve_or_drop(self, plain_id: str, resource_to_connect: str) -> Tuple[Optional[str], bool]:
+        """Resolve plain_id (no "type:" prefix) against destination state.
+
+        Returns (resolved_destination_id, is_permanently_stale):
+          - Present in state.destination[resource_to_connect]: returns
+            (destination_id, False) -- success path, caller keeps/remaps this entry.
+          - Absent from destination: calls ensure_resource_loaded() (for
+            --minimize-reads correctness) then checks state.source[resource_to_connect].
+              - Present in source ("not yet synced"): returns (None, False) -- caller
+                must treat this as today's hard-fail (add plain_id to failed_connections);
+                NOT a drop. This is the legitimate "retry on a later sync" case.
+              - Absent from source (permanently gone -- deleted before this org's
+                first-ever import, or never existed): returns (None, True) -- caller
+                drops this entry ONLY if self.config.drop_unresolvable_principals is True;
+                if the flag is False, caller must treat this identically to the "not yet
+                synced" case (today's unchanged hard-fail).
+
+        Callers own: parsing "type:id" composites to plain_id before calling, reassembling
+        "type:id" on success, deciding what "the list" means for their shape
+        (binding.principals vs. flat restricted_roles), and the "list is now empty"
+        access-elevation check.
+
+        state.destination/state.source are defaultdict(dict), so indexing an unknown
+        resource_to_connect returns {} rather than raising -- plain `in` membership checks
+        are safe and no KeyError guard is needed. An exception raised by
+        ensure_resource_loaded (e.g. a storage error) propagates uncaught, matching every
+        other ensure_resource_loaded call site.
+        """
+        destination = self.config.state.destination[resource_to_connect]
+        if plain_id in destination:
+            return destination[plain_id]["id"], False
+
+        self.config.state.ensure_resource_loaded(resource_to_connect, plain_id)
+        if plain_id in self.config.state.source[resource_to_connect]:
+            return None, False
+        return None, True
+
+    # Composite "type:id" principal type prefixes -> the resource_to_connect they map to.
+    _PRINCIPAL_TYPE_MAP: ClassVar[Dict[str, str]] = {"user": "users", "role": "roles", "team": "teams"}
+
+    def _filter_stale_binding_principals(
+        self, _id: str, bindings: Optional[List[Dict]]
+    ) -> Tuple[Dict[str, List[str]], bool]:
+        """Resolve/drop/hard-fail composite "type:id" principals across restriction-policy
+        bindings (shared by restriction_policies / monitors / synthetics_tests).
+
+        For each principal in each binding: resolve against destination (remap in place),
+        else consult source. Source-present ("not yet synced") or flag-off -> hard-fail
+        (add to failed_connections). Source-absent ("permanently gone") AND
+        --drop-unresolvable-principals -> drop it, WARN, and count it.
+
+        Rebuilds each binding's "principals" as a NEW list (never del/pop/index-assign
+        during iteration) to avoid the enumerate/index-shift skip bug. Returns
+        (failed_connections_dict, empty_binding_risk) where empty_binding_risk is True if
+        any binding whose source list was non-empty became empty after dropping.
+        """
+        failed_connections_dict: Dict[str, List[str]] = defaultdict(list)
+        empty_binding_risk = False
+        for binding in bindings or []:
+            principals = binding.get("principals")
+            if not principals:
+                continue
+            had_principals = len(principals) > 0
+            kept: List[str] = []
+            for policy_id in principals:
+                parts = policy_id.split(":", 1)
+                if len(parts) != 2:
+                    kept.append(policy_id)
+                    continue
+                _type, plain_id = parts
+                resource_to_connect = self._PRINCIPAL_TYPE_MAP.get(_type)
+                if resource_to_connect is None:
+                    # org: (already remapped by pre_resource_action_hook) or any other
+                    # non-user/role/team principal -> pass through untouched.
+                    kept.append(policy_id)
+                    continue
+                resolved, stale = self._resolve_or_drop(plain_id, resource_to_connect)
+                if resolved is not None:
+                    kept.append(f"{_type}:{resolved}")
+                elif stale and self.config.drop_unresolvable_principals:
+                    self.config.logger.warning(
+                        f"dropping stale principal '{policy_id}': absent from source and "
+                        "destination (likely deleted before the org's first sync)",
+                        resource_type=self.resource_type,
+                        _id=_id,
+                    )
+                    if self.config.counter is not None:
+                        self.config.counter.record_stale_principal_dropped(resource_type=self.resource_type, _id=_id)
+                else:
+                    # Source-present ("not yet synced", retry later) or flag off:
+                    # unchanged hard-fail semantics -- keep the original id and record it.
+                    kept.append(policy_id)
+                    failed_connections_dict[resource_to_connect].append(plain_id)
+            binding["principals"] = kept
+            if had_principals and not kept:
+                empty_binding_risk = True
+        return failed_connections_dict, empty_binding_risk
+
+    def _filter_stale_flat_roles(self, _id: str, container: Optional[Dict], key: str) -> Tuple[List[str], bool]:
+        """Resolve/drop/hard-fail a flat list of role ids at container[key]
+        (restricted_roles / options.restricted_roles / metadata.restricted_roles).
+
+        Same three-way logic as _filter_stale_binding_principals but for plain role ids
+        (no "type:" prefix). Rebuilds the list as a NEW list. Returns
+        (failed_role_ids, empty_list_risk) where empty_list_risk is True if a non-empty
+        source list became empty after dropping.
+        """
+        failed: List[str] = []
+        if not container:
+            return failed, False
+        roles = container.get(key)
+        if not roles:
+            return failed, False
+        had_roles = len(roles) > 0
+        kept: List[str] = []
+        for role in roles:
+            plain_id = str(role)
+            resolved, stale = self._resolve_or_drop(plain_id, "roles")
+            if resolved is not None:
+                kept.append(type(role)(resolved))
+            elif stale and self.config.drop_unresolvable_principals:
+                self.config.logger.warning(
+                    f"dropping stale role '{plain_id}' from {key}: absent from source and "
+                    "destination (likely deleted before the org's first sync)",
+                    resource_type=self.resource_type,
+                    _id=_id,
+                )
+                if self.config.counter is not None:
+                    self.config.counter.record_stale_principal_dropped(resource_type=self.resource_type, _id=_id)
+            else:
+                kept.append(role)
+                failed.append(plain_id)
+        container[key] = kept
+        return failed, (had_roles and not kept)
+
+    def _raise_connection_error_if_any(
+        self, _id: str, failed_connections_dict: Dict[str, List[str]], empty_binding_risk: bool = False
+    ) -> None:
+        """Shared terminal step for the drop-aware connect_resources overrides.
+
+        Mirrors the base connect_resources raise/skip behavior: raise unless
+        --skip-failed-resource-connections is set. empty_binding_risk is threaded onto the
+        exception so _apply_resource_cb can tag the metric, and is logged at ERROR because
+        an empty binding is an access-elevation risk.
+        """
+        if not failed_connections_dict and not empty_binding_risk:
+            return
+        e = ResourceConnectionError(
+            failed_connections_dict=failed_connections_dict, empty_binding_risk=empty_binding_risk
+        )
+        if empty_binding_risk:
+            self.config.logger.error(
+                "access-elevation risk: a binding/list whose source had principals became "
+                "empty after dropping unresolvable references; refusing to sync",
+                resource_type=self.resource_type,
+                _id=_id,
+            )
+        if not self.config.skip_failed_resource_connections:
+            self.config.logger.info(
+                f"skipping resource: {str(e)}",
+                _id=_id,
+                resource_type=self.resource_type,
+            )
+            raise e
+        else:
+            self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type)
+
     def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         """Extract dependency IDs referenced at r_obj[key] for resource_to_connect.
 

datadog_sync/utils/configuration.py

@@ -8,7 +8,7 @@
 import logging
 import sys
 import time
-from typing import Any, Optional, Union, Dict, List
+from typing import Any, Optional, TYPE_CHECKING, Union, Dict, List
 
 import click
 
@@ -47,6 +47,9 @@
 from datadog_sync.utils.state import State
 from datadog_sync.utils.storage.storage_types import StorageType
 
+if TYPE_CHECKING:
+    from datadog_sync.utils.workers import Counter
+
 
 @dataclass
 class Configuration(object):
@@ -77,6 +80,16 @@ class Configuration(object):
     allow_self_lockout: bool
     datadog_host_override: Optional[str] = None
     emit_json: bool = False
+    # Opt-in: drop principal/role references that are absent from BOTH destination and
+    # source state (permanently gone -- e.g. deleted before this org's first-ever import)
+    # instead of hard-failing the whole resource. When False (default) behavior is
+    # byte-for-byte unchanged. Consumed by the restriction_policies/monitors/
+    # synthetics_tests/dashboards/synthetics_private_locations models' connect logic.
+    drop_unresolvable_principals: bool = False
+    # Set at runtime by Workers.__init__ so resource models (which only hold self.config)
+    # can record dropped stale principals into the apply-run Counter from inside
+    # connect_id/connect_resources. None until a Workers is constructed for the command.
+    counter: Optional["Counter"] = None
     # Opt-in refresh of state.destination from storage before apply_resources
     # dispatches workers. Motivating case: an external orchestrator runs
     # sync-cli once per resource type as separate processes reading a shared
@@ -407,6 +420,7 @@ def build_config(cmd: Command, **kwargs: Optional[Any]) -> Configuration:
     # Additional settings
     force_missing_dependencies = kwargs.get("force_missing_dependencies") or False
     skip_failed_resource_connections = kwargs.get("skip_failed_resource_connections")
+    drop_unresolvable_principals = kwargs.get("drop_unresolvable_principals") or False
     refresh_destination_state_before_apply = kwargs.get("refresh_destination_state_before_apply") or False
     max_workers = kwargs.get("max_workers")
     max_workers_per_type_raw = kwargs.get("max_workers_per_type")
@@ -660,6 +674,7 @@ def build_config(cmd: Command, **kwargs: Optional[Any]) -> Configuration:
         filter_operator=filter_operator,
         force_missing_dependencies=force_missing_dependencies,
         skip_failed_resource_connections=skip_failed_resource_connections,
+        drop_unresolvable_principals=drop_unresolvable_principals,
         refresh_destination_state_before_apply=refresh_destination_state_before_apply,
         max_workers=max_workers,
         max_workers_per_type=max_workers_per_type,

datadog_sync/utils/resource_utils.py

@@ -78,8 +78,13 @@ def __init__(self, _id: str, _type: str):
 
 
 class ResourceConnectionError(Exception):
-    def __init__(self, failed_connections_dict):
+    def __init__(self, failed_connections_dict, empty_binding_risk: bool = False):
         super(ResourceConnectionError, self).__init__(f"Failed to connect resource. {dict(failed_connections_dict)}")
+        # empty_binding_risk flags the access-elevation case: a restriction-policy
+        # binding (or a restricted_roles list) whose source-side list was non-empty
+        # but became empty after dropping unresolvable principals. Read by
+        # ResourcesHandler._apply_resource_cb to add a distinguishing metric tag.
+        self.empty_binding_risk = empty_binding_risk
 
 
 class CustomClientHTTPError(Exception):

datadog_sync/utils/resources_handler.py

@@ -78,6 +78,30 @@ def _chunked_emit(rt: str, action_desc: str, ids: List[str]) -> None:
         ids = counter.skipped_missing_deps_by_type[rt]
         if ids:
             _chunked_emit(rt, "skipped for missing dependencies", ids)
+    # --drop-unresolvable-principals aggregate signals. Both dicts stay empty when the flag
+    # is off, so these lines never appear in the default path. getattr guards Counters
+    # constructed before these buckets existed.
+    for rt in sorted(getattr(counter, "stale_principals_dropped_by_type", {}).keys()):
+        ids = counter.stale_principals_dropped_by_type[rt]
+        if ids:
+            _chunked_emit(rt, "dropped stale principals", ids)
+    for rt in sorted(getattr(counter, "empty_binding_risk_by_type", {}).keys()):
+        ids = counter.empty_binding_risk_by_type[rt]
+        if ids:
+            total = len(ids)
+            for start in range(0, total, _SUMMARY_ID_CHUNK):
+                chunk = ids[start:start + _SUMMARY_ID_CHUNK]
+                end = min(start + _SUMMARY_ID_CHUNK, total)
+                logger.error(
+                    "sync summary: %s skipped %d resource(s) for empty-binding "
+                    "access-elevation risk [%d-%d of %d]: %s",
+                    rt,
+                    total,
+                    start + 1,
+                    end,
+                    total,
+                    ", ".join(chunk),
+                )
 
 
 def _list_time_filter_passes(r_class, config, resource) -> bool:
@@ -502,9 +526,15 @@ async def _apply_resource_cb(self, q_item: List) -> None:
             )
             _reason, _fc = self._sanitize_reason(e)
             self._emit(resource_type, _id, "sync", "skipped", reason=_reason, failure_class=_fc)
-            await r_class._send_action_metrics(
-                Command.SYNC.value, _id, Status.SKIPPED.value, tags=["reason:connection_error"]
-            )
+            # Distinguish the access-elevation case (a restriction-policy binding /
+            # restricted_roles list that emptied out after dropping stale principals) with
+            # a dedicated metric tag and a separate counter bucket for the end-of-run
+            # summary. getattr guards ResourceConnectionError instances without the attr.
+            extra_tags = ["reason:connection_error"]
+            if getattr(e, "empty_binding_risk", False):
+                extra_tags.append("risk:empty_restriction_policy")
+                self.worker.counter.record_empty_binding_risk(resource_type=resource_type, _id=_id)
+            await r_class._send_action_metrics(Command.SYNC.value, _id, Status.SKIPPED.value, tags=extra_tags)
         except Exception as e:
             # Track the source id in the counter's failed-ids bucket so
             # apply_resources can emit a targeted per-type summary at the