diff --git a/datadog_sync/commands/shared/options.py b/datadog_sync/commands/shared/options.py index df944fcc..a519b6ec 100644 --- a/datadog_sync/commands/shared/options.py +++ b/datadog_sync/commands/shared/options.py @@ -499,6 +499,21 @@ 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 connection fails normally. --skip-failed-resource-connections may suppress " + "that failure and continue syncing; an ERROR log and risk metric explicitly warn that " + "the destination resource may be unrestricted. Off by default.", + cls=CustomOptionClass, + ), option( "--cleanup", default="False", diff --git a/datadog_sync/model/dashboards.py b/datadog_sync/model/dashboards.py index 101d4246..a2a660ab 100644 --- a/datadog_sync/model/dashboards.py +++ b/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) diff --git a/datadog_sync/model/monitors.py b/datadog_sync/model/monitors.py index 754c74e8..e8e86c27 100644 --- a/datadog_sync/model/monitors.py +++ b/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": diff --git a/datadog_sync/model/restriction_policies.py b/datadog_sync/model/restriction_policies.py index 392c4733..992cb43f 100644 --- a/datadog_sync/model/restriction_policies.py +++ b/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"} diff --git a/datadog_sync/model/synthetics_private_locations.py b/datadog_sync/model/synthetics_private_locations.py index dadca4a2..a2544b68 100644 --- a/datadog_sync/model/synthetics_private_locations.py +++ b/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) diff --git a/datadog_sync/model/synthetics_tests.py b/datadog_sync/model/synthetics_tests.py index 39c65a5c..81d6485c 100644 --- a/datadog_sync/model/synthetics_tests.py +++ b/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 — diff --git a/datadog_sync/utils/base_resource.py b/datadog_sync/utils/base_resource.py index 88867594..8e6ed486 100644 --- a/datadog_sync/utils/base_resource.py +++ b/datadog_sync/utils/base_resource.py @@ -141,6 +141,13 @@ def build_excluded_attributes(self) -> None: self.excluded_attributes[i] = "root" + "".join(["['{}']".format(v) for v in attr.split(".")]) +@dataclass(frozen=True) +class ResourceConnectionResult: + """Outcome metadata from resolving a resource's cross-resource connections.""" + + empty_binding_escalation: bool = False + + class BaseResource(abc.ABC): resource_type: str resource_config: ResourceConfig @@ -437,6 +444,191 @@ 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 rechecks destination because the lazy load + may have populated its mapping, and only then checks source. + - 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) + destination = self.config.state.destination[resource_to_connect] + if plain_id in destination: + return destination[plain_id]["id"], False + + 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 + ) -> bool: + """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 skipped-resource metrics. When + --skip-failed-resource-connections suppresses the exception, returns True for an + empty-binding risk so the apply path can tag the successful action and include it in + the end-of-run escalation summary. Otherwise returns False. + """ + if not failed_connections_dict and not empty_binding_risk: + return False + e = ResourceConnectionError( + failed_connections_dict=failed_connections_dict, empty_binding_risk=empty_binding_risk + ) + if empty_binding_risk: + if self.config.skip_failed_resource_connections: + self.config.logger.error( + "access-elevation risk: a binding/list whose source had principals became " + "empty after dropping unresolvable references; " + "--skip-failed-resource-connections is enabled, continuing sync. " + "DESTINATION RESOURCE MAY BE UNRESTRICTED", + resource_type=self.resource_type, + _id=_id, + ) + else: + 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) + return empty_binding_risk + 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. @@ -451,9 +643,9 @@ def extract_source_ids(self, key: str, r_obj: Dict, resource_to_connect: str) -> return [str(v) for v in r_obj[key]] return [str(r_obj[key])] - def connect_resources(self, _id: str, resource: Dict) -> None: + def connect_resources(self, _id: str, resource: Dict) -> ResourceConnectionResult: if not self.resource_config.resource_connections: - return + return ResourceConnectionResult() failed_connections_dict = defaultdict(list) for resource_to_connect, v in self.resource_config.resource_connections.items(): @@ -475,6 +667,8 @@ def connect_resources(self, _id: str, resource: Dict) -> None: else: self.config.logger.debug(f"{str(e)}", _id=_id, resource_type=self.resource_type) + return ResourceConnectionResult() + def filter(self, resource: Dict) -> bool: if not self.config.filters or self.resource_type not in self.config.filters: return True diff --git a/datadog_sync/utils/configuration.py b/datadog_sync/utils/configuration.py index f91e9591..75589760 100644 --- a/datadog_sync/utils/configuration.py +++ b/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. The shared BaseResource drop-aware helpers consume this + # flag; model-specific connection wiring is intentionally separate. + 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, diff --git a/datadog_sync/utils/resource_utils.py b/datadog_sync/utils/resource_utils.py index 800d2dbb..aa550b78 100644 --- a/datadog_sync/utils/resource_utils.py +++ b/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): diff --git a/datadog_sync/utils/resources_handler.py b/datadog_sync/utils/resources_handler.py index 752b7392..ad6bde84 100644 --- a/datadog_sync/utils/resources_handler.py +++ b/datadog_sync/utils/resources_handler.py @@ -78,6 +78,48 @@ 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), + ) + for rt in sorted(getattr(counter, "empty_binding_escalation_by_type", {}).keys()): + ids = counter.empty_binding_escalation_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 synced %d resource(s) after an empty-binding " + "connection failure was suppressed [%d-%d of %d]: %s. " + "DESTINATION RESOURCE MAY BE UNRESTRICTED", + rt, + total, + start + 1, + end, + total, + ", ".join(chunk), + ) def _list_time_filter_passes(r_class, config, resource) -> bool: @@ -451,7 +493,12 @@ async def _apply_resource_cb(self, q_item: List) -> None: # Run hooks await r_class._pre_resource_action_hook(_id, resource) - r_class.connect_resources(_id, resource) + connection_result = r_class.connect_resources(_id, resource) + empty_binding_escalation = connection_result.empty_binding_escalation + + success_metric_tags = [] + if empty_binding_escalation: + success_metric_tags.append("risk:empty_restriction_policy") prep_resource(r_class.resource_config, resource) if _id in self.config.state.destination[resource_type]: @@ -463,16 +510,26 @@ async def _apply_resource_cb(self, q_item: List) -> None: self.config.logger.debug(f"Running update for {resource_type} with {_id}") await r_class._update_resource(_id, resource) + if empty_binding_escalation: + self.worker.counter.record_empty_binding_escalation(resource_type=resource_type, _id=_id) await r_class._send_action_metrics( - Command.SYNC.value, _id, Status.SUCCESS.value, tags=["action_sub_type:update"] + Command.SYNC.value, + _id, + Status.SUCCESS.value, + tags=["action_sub_type:update", *success_metric_tags], ) self.config.logger.debug(f"Finished update for {resource_type} with {_id}") self._emit(resource_type, _id, "sync", "success", "update") else: self.config.logger.debug(f"Running create for {resource_type} with id: {_id}") await r_class._create_resource(_id, resource) + if empty_binding_escalation: + self.worker.counter.record_empty_binding_escalation(resource_type=resource_type, _id=_id) await r_class._send_action_metrics( - Command.SYNC.value, _id, Status.SUCCESS.value, tags=["action_sub_type:create"] + Command.SYNC.value, + _id, + Status.SUCCESS.value, + tags=["action_sub_type:create", *success_metric_tags], ) self.config.logger.debug(f"finished create for {resource_type} with id: {_id}") self._emit(resource_type, _id, "sync", "success", "create") @@ -502,9 +559,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 diff --git a/datadog_sync/utils/workers.py b/datadog_sync/utils/workers.py index f67dea66..35704899 100644 --- a/datadog_sync/utils/workers.py +++ b/datadog_sync/utils/workers.py @@ -23,6 +23,10 @@ def __init__(self, config: Configuration) -> None: self.workers: List[Task] = [] self.work_queue: Queue = Queue() self.counter: Counter = Counter() + # Back-reference so resource models (which only hold self.config) can record + # dropped stale principals into the counter from inside connect_id/connect_resources. + # Last-writer-wins is fine: only one Workers drives a given command's apply pass. + config.counter = self.counter self.pbar: Optional[tqdm] = None self._running_workers_count: int = 0 self._loop: AbstractEventLoop = get_event_loop() @@ -119,6 +123,13 @@ class Counter: # actionable if the roles-sync run logged that X, Y, Z failed. failed_ids_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) skipped_missing_deps_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) + # Per-resource-type source IDs whose stale (source-absent) principal/role + # references were dropped from an access-control list, and resource IDs skipped + # because dropping left an empty binding/list (access-elevation risk). Populated + # only when --drop-unresolvable-principals is set. Surfaced by _emit_apply_summary. + stale_principals_dropped_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) + empty_binding_risk_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) + empty_binding_escalation_by_type: Dict[str, List[str]] = field(default_factory=lambda: defaultdict(list)) def __str__(self): return ( @@ -129,6 +140,9 @@ def reset_counter(self) -> None: self.successes = self.failure = self.skipped = self.filtered = 0 self.failed_ids_by_type = defaultdict(list) self.skipped_missing_deps_by_type = defaultdict(list) + self.stale_principals_dropped_by_type = defaultdict(list) + self.empty_binding_risk_by_type = defaultdict(list) + self.empty_binding_escalation_by_type = defaultdict(list) def increment_success(self) -> None: self.successes += 1 @@ -141,11 +155,32 @@ def increment_failure(self, resource_type: Optional[str] = None, _id: Optional[s if resource_type is not None and _id is not None: self.failed_ids_by_type[resource_type].append(str(_id)) - def increment_skipped(self, resource_type: Optional[str] = None, _id: Optional[str] = None, - missing_deps: bool = False) -> None: + def increment_skipped( + self, resource_type: Optional[str] = None, _id: Optional[str] = None, missing_deps: bool = False + ) -> None: self.skipped += 1 if missing_deps and resource_type is not None and _id is not None: self.skipped_missing_deps_by_type[resource_type].append(str(_id)) def increment_filtered(self) -> None: self.filtered += 1 + + # Two dedicated methods rather than the single-multi-purpose-method convention used by + # increment_skipped/increment_failure: the "dropped a stale principal but kept syncing" + # and "skipped for empty-binding access-elevation risk" cases are semantically distinct + # enough (one is a non-fatal drop, the other a hard-fail security signal) to warrant + # separate names. This deliberate deviation is called out here so it isn't flagged as + # an inconsistency. + def record_stale_principal_dropped(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None: + if resource_type is not None and _id is not None: + self.stale_principals_dropped_by_type[resource_type].append(str(_id)) + + def record_empty_binding_risk(self, resource_type: Optional[str] = None, _id: Optional[str] = None) -> None: + if resource_type is not None and _id is not None: + self.empty_binding_risk_by_type[resource_type].append(str(_id)) + + def record_empty_binding_escalation( + self, resource_type: Optional[str] = None, _id: Optional[str] = None + ) -> None: + if resource_type is not None and _id is not None: + self.empty_binding_escalation_by_type[resource_type].append(str(_id)) diff --git a/tests/unit/test_apply_resource_connection_error.py b/tests/unit/test_apply_resource_connection_error.py index 535c7b2d..2ecef0b2 100644 --- a/tests/unit/test_apply_resource_connection_error.py +++ b/tests/unit/test_apply_resource_connection_error.py @@ -14,6 +14,7 @@ import asyncio from unittest.mock import AsyncMock, MagicMock +from datadog_sync.utils.base_resource import ResourceConfig, ResourceConnectionResult from datadog_sync.utils.resource_utils import ResourceConnectionError from datadog_sync.utils.resources_handler import ResourcesHandler @@ -68,3 +69,65 @@ def test_resource_connection_error_still_increments_skipped(mock_config): emit_args, emit_kwargs = handler._emit.call_args assert emit_args[:4] == ("dashboards", "dash-1", "sync", "skipped") assert "reason" in emit_kwargs + + +def _metric_tags(r_class): + """Extract the tags passed to the (async) _send_action_metrics call.""" + _args, kwargs = r_class._send_action_metrics.call_args + return kwargs.get("tags", []) + + +def test_empty_binding_risk_adds_metric_tag_and_records(mock_config): + """empty_binding_risk=True must add the risk tag AND bump the dedicated counter.""" + exc = ResourceConnectionError({"roles": ["role-gone"]}, empty_binding_risk=True) + + handler = _drive_apply_cb(mock_config, "restriction_policies", "dashboard:dash-1", exc) + + r_class = mock_config.resources["restriction_policies"] + tags = _metric_tags(r_class) + assert "reason:connection_error" in tags + assert "risk:empty_restriction_policy" in tags + handler.worker.counter.record_empty_binding_risk.assert_called_once_with( + resource_type="restriction_policies", _id="dashboard:dash-1" + ) + + +def test_ordinary_connection_error_has_no_risk_tag(mock_config): + """Without empty_binding_risk the risk tag is absent and the counter is untouched.""" + exc = ResourceConnectionError({"roles": ["role-not-yet-synced"]}) + + handler = _drive_apply_cb(mock_config, "restriction_policies", "dashboard:xyz", exc) + + r_class = mock_config.resources["restriction_policies"] + tags = _metric_tags(r_class) + assert "reason:connection_error" in tags + assert "risk:empty_restriction_policy" not in tags + handler.worker.counter.record_empty_binding_risk.assert_not_called() + + +def test_successful_update_after_suppressed_empty_binding_risk_is_tagged_and_recorded(mock_config): + resource_type = "restriction_policies" + _id = "dashboard:dash-1" + r_class = MagicMock() + r_class.resource_config = ResourceConfig(base_path="", skip_resource_mapping=True) + r_class.connect_resources.return_value = ResourceConnectionResult(empty_binding_escalation=True) + r_class._pre_resource_action_hook = AsyncMock() + r_class._update_resource = AsyncMock() + r_class._send_action_metrics = AsyncMock() + mock_config.resources = {resource_type: r_class} + mock_config.state.source[resource_type][_id] = {"id": _id, "bindings": []} + mock_config.state.destination[resource_type][_id] = {"id": _id, "bindings": [{"relation": "editor"}]} + + handler = ResourcesHandler(mock_config) + handler.worker = MagicMock() + handler.worker.counter = MagicMock() + handler.sorter = MagicMock() + handler._emit = MagicMock() + + asyncio.run(handler._apply_resource_cb([resource_type, _id])) + + assert "risk:empty_restriction_policy" in _metric_tags(r_class) + handler.worker.counter.record_empty_binding_escalation.assert_called_once_with( + resource_type=resource_type, _id=_id + ) + handler.worker.counter.record_empty_binding_risk.assert_not_called() diff --git a/tests/unit/test_apply_summary_handler.py b/tests/unit/test_apply_summary_handler.py index 45ea3b7e..1290e58f 100644 --- a/tests/unit/test_apply_summary_handler.py +++ b/tests/unit/test_apply_summary_handler.py @@ -36,6 +36,19 @@ def _rendered_warnings(logger): return out +def _rendered_errors(logger): + out = [] + for call in logger.error.call_args_list: + args = call.args + if not args: + continue + try: + out.append(args[0] % args[1:]) + except Exception: + out.append(str(args)) + return out + + def test_summary_emits_failed_ids_by_type(): counter = Counter() counter.increment_failure(resource_type="roles", _id="uuid-abc") @@ -123,6 +136,55 @@ def test_summary_uses_join_not_repr_for_greppability(): assert "sync summary" in joined +def test_summary_emits_stale_principals_dropped_at_warning(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="role:role-uuid-1") + + logger = _make_logger() + _emit_apply_summary(logger, counter) + + joined = "\n".join(_rendered_warnings(logger)) + assert "restriction_policies" in joined + assert "role:role-uuid-1" in joined + assert "dropped" in joined and "stale principal" in joined + + +def test_summary_emits_empty_binding_risk_at_error(): + counter = Counter() + counter.record_empty_binding_risk(resource_type="restriction_policies", _id="dashboard:dash-1") + + logger = _make_logger() + _emit_apply_summary(logger, counter) + + joined_err = "\n".join(_rendered_errors(logger)) + assert "restriction_policies" in joined_err + assert "dashboard:dash-1" in joined_err + assert "empty-binding" in joined_err and "access-elevation" in joined_err + + +def test_summary_emits_applied_empty_binding_escalation_at_error(): + counter = Counter() + counter.record_empty_binding_escalation(resource_type="restriction_policies", _id="dashboard:dash-1") + + logger = _make_logger() + _emit_apply_summary(logger, counter) + + joined_err = "\n".join(_rendered_errors(logger)) + assert "restriction_policies" in joined_err + assert "dashboard:dash-1" in joined_err + assert "connection failure was suppressed" in joined_err + assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" in joined_err + + +def test_summary_silent_for_new_buckets_when_empty(): + counter = Counter() + logger = _make_logger() + _emit_apply_summary(logger, counter) + # No stale/empty-binding activity -> neither WARNING nor ERROR lines emitted. + assert logger.warning.call_count == 0 + assert logger.error.call_count == 0 + + def test_chunk_constant_is_reasonable(): assert 20 <= _SUMMARY_ID_CHUNK <= 200 diff --git a/tests/unit/test_base_resource.py b/tests/unit/test_base_resource.py new file mode 100644 index 00000000..e3bf0446 --- /dev/null +++ b/tests/unit/test_base_resource.py @@ -0,0 +1,160 @@ +# Unless explicitly stated otherwise all files in this repository are licensed +# under the 3-clause BSD style license (see LICENSE). +# This product includes software developed at Datadog (https://www.datadoghq.com/). +# Copyright 2019 Datadog, Inc. + +"""Unit tests for BaseResource._resolve_or_drop. + +_resolve_or_drop is the shared three-way resolver behind --drop-unresolvable-principals: +it decides whether a plain dependency id resolves against destination state, is a +"not yet synced" source-present miss, or is a permanently-gone (source-absent) stale +reference. It reads only self.config, so we exercise it as an unbound method with a +stub self rather than instantiating the abstract BaseResource. +""" + +from collections import defaultdict +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from datadog_sync.utils.base_resource import BaseResource, ResourceConnectionResult + + +def _make_config(): + config = MagicMock() + state = MagicMock() + state.source = defaultdict(dict) + state.destination = defaultdict(dict) + state.ensure_resource_loaded = MagicMock() + config.state = state + return config + + +def _call(config, plain_id, resource_to_connect): + stub = SimpleNamespace(config=config) + return BaseResource._resolve_or_drop(stub, plain_id, resource_to_connect) + + +def test_connect_resources_returns_named_default_result_when_no_connections(): + stub = SimpleNamespace(resource_config=SimpleNamespace(resource_connections=None)) + + result = BaseResource.connect_resources(stub, "resource-id", {}) + + assert result == ResourceConnectionResult() + assert result.empty_binding_escalation is False + + +def test_resolve_or_drop_destination_present_returns_dest_id_not_stale(): + config = _make_config() + config.state.destination["roles"]["src-role"] = {"id": "dst-role"} + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved == "dst-role" + assert stale is False + # No need to consult source when destination already has it. + config.state.ensure_resource_loaded.assert_not_called() + + +def test_resolve_or_drop_rechecks_destination_after_lazy_load_populates_both_states(): + config = _make_config() + + def load_role(resource_type, plain_id): + config.state.source[resource_type][plain_id] = {"id": plain_id} + config.state.destination[resource_type][plain_id] = {"id": "dst-role"} + + config.state.ensure_resource_loaded.side_effect = load_role + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved == "dst-role" + assert stale is False + config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role") + + +def test_resolve_or_drop_rechecks_destination_after_lazy_load_populates_destination_only(): + config = _make_config() + + def load_role(resource_type, plain_id): + config.state.destination[resource_type][plain_id] = {"id": "dst-role"} + + config.state.ensure_resource_loaded.side_effect = load_role + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved == "dst-role" + assert stale is False + config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role") + + +def test_resolve_or_drop_source_present_not_synced_returns_none_not_stale(): + config = _make_config() + config.state.source["roles"]["src-role"] = {"id": "src-role"} + + resolved, stale = _call(config, "src-role", "roles") + + assert resolved is None + assert stale is False # legitimate "not yet synced" -> caller hard-fails, does NOT drop + config.state.ensure_resource_loaded.assert_called_once_with("roles", "src-role") + + +def test_resolve_or_drop_absent_from_both_is_permanently_stale(): + config = _make_config() + + resolved, stale = _call(config, "ghost-role", "roles") + + assert resolved is None + assert stale is True # absent from destination AND source -> permanently gone + config.state.ensure_resource_loaded.assert_called_once_with("roles", "ghost-role") + + +def test_resolve_or_drop_unknown_resource_type_does_not_raise(): + # state.destination/source are defaultdict(dict); an unknown type yields {} not KeyError. + config = _make_config() + + resolved, stale = _call(config, "whatever", "never_seen_type") + + assert resolved is None + assert stale is True + + +def test_resolve_or_drop_propagates_ensure_resource_loaded_exception(): + config = _make_config() + config.state.ensure_resource_loaded.side_effect = RuntimeError("storage down") + + # Not in destination -> ensure_resource_loaded is called -> its error must propagate, + # matching every other ensure_resource_loaded call site (none catch it). + with pytest.raises(RuntimeError, match="storage down"): + _call(config, "src-role", "roles") + + +def test_empty_binding_risk_continues_with_explicit_escalation_warning_when_connection_failures_are_skipped(): + config = _make_config() + config.skip_failed_resource_connections = True + config.logger = MagicMock() + stub = SimpleNamespace(config=config, resource_type="restriction_policies") + + escalation_risk = BaseResource._raise_connection_error_if_any(stub, "dashboard:abc", {}, True) + + assert escalation_risk is True + message = 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_empty_binding_risk_refuses_to_sync_when_connection_failures_are_not_skipped(): + from datadog_sync.utils.resource_utils import ResourceConnectionError + + config = _make_config() + config.skip_failed_resource_connections = False + config.logger = MagicMock() + stub = SimpleNamespace(config=config, resource_type="restriction_policies") + + with pytest.raises(ResourceConnectionError): + BaseResource._raise_connection_error_if_any(stub, "dashboard:abc", {}, True) + + message = config.logger.error.call_args.args[0] + assert "refusing to sync" in message + assert "DESTINATION RESOURCE MAY BE UNRESTRICTED" not in message diff --git a/tests/unit/test_dashboards.py b/tests/unit/test_dashboards.py index a0932065..c087870d 100644 --- a/tests/unit/test_dashboards.py +++ b/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 diff --git a/tests/unit/test_drop_unresolvable_principals_cli.py b/tests/unit/test_drop_unresolvable_principals_cli.py new file mode 100644 index 00000000..1dc11f69 --- /dev/null +++ b/tests/unit/test_drop_unresolvable_principals_cli.py @@ -0,0 +1,43 @@ +# Unless explicitly stated otherwise all files in this repository are licensed +# under the 3-clause BSD style license (see LICENSE). +# This product includes software developed at Datadog (https://www.datadoghq.com/). +# Copyright 2019 Datadog, Inc. + +"""CLI round-trip test for --drop-unresolvable-principals. + +Mirrors test_force_missing_dependencies_cli.py. The flag lives in _diffs_options, +which is attached to sync, diffs, and migrate (the three commands that run +connect_resources). exit_code == 2 is click's "no such option" usage error, so +exit_code != 2 proves the option is recognized by each command. +""" + +import pytest +from click.testing import CliRunner + +from datadog_sync.cli import cli + + +@pytest.fixture +def runner(): + return CliRunner(mix_stderr=False) + + +def test_sync_accepts_drop_unresolvable_principals(runner): + result = runner.invoke(cli, ["sync", "--drop-unresolvable-principals", "--validate=false"]) + assert result.exit_code != 2 + + +def test_diffs_accepts_drop_unresolvable_principals(runner): + result = runner.invoke(cli, ["diffs", "--drop-unresolvable-principals"]) + assert result.exit_code != 2 + + +def test_migrate_accepts_drop_unresolvable_principals(runner): + result = runner.invoke(cli, ["migrate", "--drop-unresolvable-principals", "--validate=false"]) + assert result.exit_code != 2 + + +def test_import_rejects_drop_unresolvable_principals(runner): + # The flag is NOT attached to import (import does not run connect_resources). + result = runner.invoke(cli, ["import", "--drop-unresolvable-principals"]) + assert result.exit_code == 2 diff --git a/tests/unit/test_monitors.py b/tests/unit/test_monitors.py index c951246e..bdc7dc0a 100644 --- a/tests/unit/test_monitors.py +++ b/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,102 @@ 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_empty_risk_is_returned_when_connection_failure_is_suppressed(self): + m = self._make_monitors(drop=True, skip_failed=True) + resource = {"id": 1, "restricted_roles": ["role-gone"]} + + assert m.connect_resources("1", resource).empty_binding_escalation 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"] diff --git a/tests/unit/test_resource_utils.py b/tests/unit/test_resource_utils.py index c2780356..ca243115 100644 --- a/tests/unit/test_resource_utils.py +++ b/tests/unit/test_resource_utils.py @@ -7,7 +7,19 @@ from unittest.mock import MagicMock, call from datadog_sync import models -from datadog_sync.utils.resource_utils import find_attr +from datadog_sync.utils.resource_utils import find_attr, ResourceConnectionError + + +def test_resource_connection_error_empty_binding_risk_defaults_false(): + exc = ResourceConnectionError({"roles": ["r1"]}) + assert exc.empty_binding_risk is False + + +def test_resource_connection_error_empty_binding_risk_settable(): + exc = ResourceConnectionError({"roles": ["r1"]}, empty_binding_risk=True) + assert exc.empty_binding_risk is True + # Message formatting is unchanged by the new kwarg. + assert "Failed to connect resource." in str(exc) @pytest.fixture(scope="class") diff --git a/tests/unit/test_restriction_policies.py b/tests/unit/test_restriction_policies.py index ebb42fdd..b916ba14 100644 --- a/tests/unit/test_restriction_policies.py +++ b/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"] == [] diff --git a/tests/unit/test_synthetics_private_locations.py b/tests/unit/test_synthetics_private_locations.py new file mode 100644 index 00000000..3e216d36 --- /dev/null +++ b/tests/unit/test_synthetics_private_locations.py @@ -0,0 +1,86 @@ +# Unless explicitly stated otherwise all files in this repository are licensed +# under the 3-clause BSD style license (see LICENSE). +# This product includes software developed at Datadog (https://www.datadoghq.com/). +# Copyright 2019 Datadog, Inc. + +"""Unit tests for synthetics_private_locations connect_resources drop behavior. + +The only access-control connection is the flat `metadata.restricted_roles` list. Under +--drop-unresolvable-principals a permanently-stale role is dropped; an emptied list still +hard-fails as an access-elevation guard. +""" + +from collections import defaultdict +from unittest.mock import MagicMock + +import pytest + +from datadog_sync.model.synthetics_private_locations import SyntheticsPrivateLocations +from datadog_sync.utils.resource_utils import ResourceConnectionError +from datadog_sync.utils.workers import Counter + + +def _make_spl(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 SyntheticsPrivateLocations(config) + + +def _seed_valid_role(spl, src="role-good", dst="role-good-dst"): + spl.config.state.destination["roles"][src] = {"id": dst} + + +def test_flag_off_stale_role_hard_fails(): + spl = _make_spl(drop=False) + resource = {"id": "pl:abc", "metadata": {"restricted_roles": ["role-gone"]}} + with pytest.raises(ResourceConnectionError): + spl.connect_resources("pl:abc", resource) + assert resource["metadata"]["restricted_roles"] == ["role-gone"] # not dropped when off + + +def test_flag_on_drops_stale_keeps_valid(): + spl = _make_spl(drop=True) + _seed_valid_role(spl) + resource = {"id": "pl:abc", "metadata": {"restricted_roles": ["role-good", "role-gone"]}} + spl.connect_resources("pl:abc", resource) # no raise + assert resource["metadata"]["restricted_roles"] == ["role-good-dst"] + assert spl.config.counter.stale_principals_dropped_by_type["synthetics_private_locations"] == ["pl:abc"] + + +def test_flag_on_all_stale_empty_list_raises_risk(): + spl = _make_spl(drop=True) + resource = {"id": "pl:abc", "metadata": {"restricted_roles": ["role-gone"]}} + with pytest.raises(ResourceConnectionError) as exc_info: + spl.connect_resources("pl:abc", resource) + assert exc_info.value.empty_binding_risk is True + assert resource["metadata"]["restricted_roles"] == [] + + +def test_empty_list_risk_is_returned_when_connection_failure_is_suppressed(): + spl = _make_spl(drop=True, skip_failed=True) + resource = {"id": "pl:abc", "metadata": {"restricted_roles": ["role-gone"]}} + + assert spl.connect_resources("pl:abc", resource).empty_binding_escalation is True + assert resource["metadata"]["restricted_roles"] == [] + + +def test_flag_on_middle_drop_keeps_neighbors(): + spl = _make_spl(drop=True) + _seed_valid_role(spl, src="ra", dst="ra-dst") + _seed_valid_role(spl, src="rc", dst="rc-dst") + resource = {"id": "pl:abc", "metadata": {"restricted_roles": ["ra", "role-gone", "rc"]}} + spl.connect_resources("pl:abc", resource) # no raise + assert resource["metadata"]["restricted_roles"] == ["ra-dst", "rc-dst"] + + +def test_no_metadata_is_noop(): + spl = _make_spl(drop=True) + resource = {"id": "pl:abc"} # no metadata/restricted_roles + spl.connect_resources("pl:abc", resource) # no raise diff --git a/tests/unit/test_synthetics_tests.py b/tests/unit/test_synthetics_tests.py index f8a1d3f8..fce54a11 100644 --- a/tests/unit/test_synthetics_tests.py +++ b/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: @@ -474,5 +477,89 @@ def test_extract_source_ids_org_excluded(self): assert synthetics_tests.extract_source_ids("principals", binding, "teams") == [] -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_empty_risk_is_returned_when_connection_failure_is_suppressed(self): + t = self._make_test(drop=True, skip_failed=True) + resource = {"public_id": "abc", "options": {"restricted_roles": ["role-gone"]}} + + assert t.connect_resources("abc", resource).empty_binding_escalation 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"] diff --git a/tests/unit/test_workers.py b/tests/unit/test_workers.py new file mode 100644 index 00000000..bf923ac4 --- /dev/null +++ b/tests/unit/test_workers.py @@ -0,0 +1,72 @@ +# Unless explicitly stated otherwise all files in this repository are licensed +# under the 3-clause BSD style license (see LICENSE). +# This product includes software developed at Datadog (https://www.datadoghq.com/). +# Copyright 2019 Datadog, Inc. + +"""Unit tests for the Counter buckets added for --drop-unresolvable-principals. + +Guards two things: (1) the new fields default empty and record correctly, and +(2) reset_counter() zeroes them so repeated import/sync invocations in one process +don't double-count. +""" + +from datadog_sync.utils.workers import Counter + + +def test_new_counter_fields_default_empty(): + counter = Counter() + assert counter.stale_principals_dropped_by_type == {} + assert counter.empty_binding_risk_by_type == {} + assert counter.empty_binding_escalation_by_type == {} + + +def test_record_stale_principal_dropped_appends_by_type(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="dashboard:abc") + counter.record_stale_principal_dropped(resource_type="restriction_policies", _id="dashboard:def") + counter.record_stale_principal_dropped(resource_type="monitors", _id="mon-1") + + assert counter.stale_principals_dropped_by_type["restriction_policies"] == [ + "dashboard:abc", + "dashboard:def", + ] + assert counter.stale_principals_dropped_by_type["monitors"] == ["mon-1"] + + +def test_record_empty_binding_risk_appends_by_type(): + counter = Counter() + counter.record_empty_binding_risk(resource_type="restriction_policies", _id="dashboard:dash-1") + + assert counter.empty_binding_risk_by_type["restriction_policies"] == ["dashboard:dash-1"] + + +def test_record_empty_binding_escalation_appends_by_type(): + counter = Counter() + counter.record_empty_binding_escalation(resource_type="restriction_policies", _id="dashboard:dash-1") + + assert counter.empty_binding_escalation_by_type["restriction_policies"] == ["dashboard:dash-1"] + + +def test_record_methods_ignore_none_args(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type=None, _id="x") + counter.record_stale_principal_dropped(resource_type="roles", _id=None) + counter.record_empty_binding_risk(resource_type=None, _id=None) + counter.record_empty_binding_escalation(resource_type=None, _id=None) + + assert counter.stale_principals_dropped_by_type == {} + assert counter.empty_binding_risk_by_type == {} + assert counter.empty_binding_escalation_by_type == {} + + +def test_reset_counter_clears_new_fields(): + counter = Counter() + counter.record_stale_principal_dropped(resource_type="roles", _id="r1") + counter.record_empty_binding_risk(resource_type="roles", _id="r2") + counter.record_empty_binding_escalation(resource_type="roles", _id="r3") + + counter.reset_counter() + + assert counter.stale_principals_dropped_by_type == {} + assert counter.empty_binding_risk_by_type == {} + assert counter.empty_binding_escalation_by_type == {}