From 387909423549233a95a61be18963ec43d394a1fd Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Mon, 13 Jul 2026 14:08:45 -0400 Subject: [PATCH 1/4] Safely replace instrumentation policies --- .../attach_integration_permissions.py | 346 +++++++++++++++-- .../attach_integration_permissions_test.py | 212 ++++++++++- .../datadog_integration_permissions.yaml | 354 ++++++++++++++++-- 3 files changed, 842 insertions(+), 70 deletions(-) diff --git a/aws_quickstart/attach_integration_permissions.py b/aws_quickstart/attach_integration_permissions.py index 7102f889..c2a85932 100644 --- a/aws_quickstart/attach_integration_permissions.py +++ b/aws_quickstart/attach_integration_permissions.py @@ -33,13 +33,14 @@ STANDARD_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/standard" RESOURCE_COLLECTION_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/resource_collection?chunked=true" INSTRUMENTATION_PERMISSIONS_API_PATH = "/api/unstable/instrumenter/aws/iam_permissions" +MANAGED_POLICY_CHAR_LIMIT = 6144 class DatadogAPIError(Exception): pass -def fetch_permissions_from_datadog(api_url): +def fetch_permissions_attributes_from_datadog(api_url): headers = { "Dd-Aws-Api-Call-Source": API_CALL_SOURCE_HEADER_VALUE, } @@ -53,7 +54,36 @@ def fetch_permissions_from_datadog(api_url): error_message = error_body.get('errors', ['Unknown error'])[0] raise DatadogAPIError(f"Datadog API error: {error_message}") from e - return json.loads(response.read())["data"]["attributes"]["permissions"] + return json.loads(response.read())["data"]["attributes"] + + +def fetch_permissions_from_datadog(api_url): + return fetch_permissions_attributes_from_datadog(api_url)["permissions"] + + +def fetch_instrumentation_policy_documents(api_url): + attributes = fetch_permissions_attributes_from_datadog(api_url) + policy_documents = attributes.get("policy_documents") + if policy_documents is not None: + if not policy_documents: + raise DatadogAPIError("Datadog API returned no instrumentation policy documents") + for document in policy_documents: + _serialize_policy_document(document) + return policy_documents + + permission_chunks = attributes.get("permissions") + if not permission_chunks: + raise DatadogAPIError("Datadog API returned neither policy_documents nor permissions") + if isinstance(permission_chunks[0], str): + permission_chunks = [permission_chunks] + LOGGER.warning("Datadog API does not expose policy_documents yet; using legacy broad permissions") + return [ + { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": chunk, "Resource": "*"}], + } + for chunk in permission_chunks + ] def parse_resource_types(raw): @@ -65,9 +95,14 @@ def parse_resource_types(raw): return [t.strip() for t in items if t and t.strip()] -def build_instrumentation_permissions_url(datadog_site, resource_types): +def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( - [("resource_type", t) for t in resource_types] + [("chunked", "true")] + [("resource_type", t) for t in resource_types] + + [ + ("account_id", account_id), + ("partition", partition), + ("chunked", "true"), + ] ) return f"https://api.{datadog_site}{INSTRUMENTATION_PERMISSIONS_API_PATH}?{query}" @@ -75,12 +110,14 @@ def build_instrumentation_permissions_url(datadog_site, resource_types): def _detach_and_delete_policy(iam_client, role_name, policy_arn, policy_name): # Detach + delete are both no-ops if the entity is already gone, so callers can blindly # iterate the policy-name space without first checking what actually exists. + errors = [] try: iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy_arn) except iam_client.exceptions.NoSuchEntityException: pass except Exception as e: LOGGER.error(f"Error detaching policy {policy_name}: {str(e)}") + errors.append(f"detach {policy_name}: {e}") try: iam_client.delete_policy(PolicyArn=policy_arn) @@ -88,8 +125,11 @@ def _detach_and_delete_policy(iam_client, role_name, policy_arn, policy_name): pass except iam_client.exceptions.DeleteConflictException: LOGGER.warning(f"Policy {policy_name} still attached, skipping delete") + errors.append(f"delete {policy_name}: policy is still attached") except Exception as e: LOGGER.error(f"Error deleting policy {policy_name}: {str(e)}") + errors.append(f"delete {policy_name}: {e}") + return errors def _cleanup_chunked_policies(iam_client, role_name, account_id, partition, prefix, max_policies=10): @@ -138,17 +178,36 @@ def attach_standard_permissions(iam_client, role_name): ) -def _create_and_attach_policy(iam_client, role_name, policy_name, actions): - policy_json = json.dumps( - { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], - }, - separators=(',', ':'), - ) - LOGGER.info(f"Creating policy {policy_name} with {len(actions)} permissions ({len(policy_json)} characters)") +def _serialize_policy_document(policy_document): + if not isinstance(policy_document, dict): + raise DatadogAPIError("instrumentation policy document must be an object") + if policy_document.get("Version") != "2012-10-17": + raise DatadogAPIError("instrumentation policy document has an unsupported version") + if not isinstance(policy_document.get("Statement"), list) or not policy_document["Statement"]: + raise DatadogAPIError("instrumentation policy document must contain statements") + + policy_json = json.dumps(policy_document, separators=(',', ':')) + if len(policy_json) > MANAGED_POLICY_CHAR_LIMIT: + raise DatadogAPIError( + f"instrumentation policy document exceeds {MANAGED_POLICY_CHAR_LIMIT} characters" + ) + return policy_json + + +def _create_policy(iam_client, policy_name, policy_document): + policy_json = _serialize_policy_document(policy_document) + LOGGER.info(f"Creating policy {policy_name} ({len(policy_json)} characters)") policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=policy_json) - iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy['Policy']['Arn']) + return policy['Policy']['Arn'] + + +def _create_and_attach_policy(iam_client, role_name, policy_name, actions): + policy_document = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], + } + policy_arn = _create_policy(iam_client, policy_name, policy_document) + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) def attach_resource_collection_permissions(iam_client, role_name): @@ -162,13 +221,240 @@ def attach_resource_collection_permissions(iam_client, role_name): ) +def _list_attached_role_policies(iam_client, role_name): + policies = [] + marker = None + while True: + request = {"RoleName": role_name} + if marker: + request["Marker"] = marker + response = iam_client.list_attached_role_policies(**request) + policies.extend(response.get("AttachedPolicies", [])) + if not response.get("IsTruncated"): + return policies + marker = response["Marker"] + + +def _list_policy_versions(iam_client, policy_arn): + versions = [] + marker = None + while True: + request = {"PolicyArn": policy_arn} + if marker: + request["Marker"] = marker + response = iam_client.list_policy_versions(**request) + versions.extend(response.get("Versions", [])) + if not response.get("IsTruncated"): + return versions + marker = response["Marker"] + + +def _policy_quotas(iam_client): + summary = iam_client.get_account_summary().get("SummaryMap", {}) + attachment_quota = summary.get("AttachedPoliciesPerRoleQuota") + version_quota = summary.get("VersionsPerPolicyQuota") + if not isinstance(attachment_quota, int) or attachment_quota < 1: + raise RuntimeError("IAM account summary did not include AttachedPoliciesPerRoleQuota") + if not isinstance(version_quota, int) or version_quota < 2: + raise RuntimeError("IAM account summary did not include VersionsPerPolicyQuota") + return attachment_quota, version_quota + + +def _instrumentation_policy_name(role_name, index): + return f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{index}" + + +def _instrumentation_policy_index(policy_name, role_name): + prefix = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-" + if not policy_name.startswith(prefix): + return None + suffix = policy_name[len(prefix):] + if not suffix.isdigit() or int(suffix) < 1: + return None + return int(suffix) + + +def _stage_policy_version(iam_client, policy_arn, policy_document, version_quota): + versions = _list_policy_versions(iam_client, policy_arn) + previous_default = next( + (version["VersionId"] for version in versions if version.get("IsDefaultVersion")), + None, + ) + if previous_default is None: + raise RuntimeError(f"policy {policy_arn} has no default version") + + non_default_versions = sorted( + (version for version in versions if not version.get("IsDefaultVersion")), + key=lambda version: int(version["VersionId"].removeprefix("v")), + ) + while len(versions) >= version_quota: + if not non_default_versions: + raise RuntimeError(f"policy {policy_arn} has no removable non-default version") + version = non_default_versions.pop(0) + iam_client.delete_policy_version( + PolicyArn=policy_arn, + VersionId=version["VersionId"], + ) + versions.remove(version) + + response = iam_client.create_policy_version( + PolicyArn=policy_arn, + PolicyDocument=_serialize_policy_document(policy_document), + SetAsDefault=False, + ) + return { + "PolicyArn": policy_arn, + "PreviousVersionId": previous_default, + "StagedVersionId": response["PolicyVersion"]["VersionId"], + } + + +def _discard_staged_replacement(iam_client, role_name, staged_versions, created_policies): + cleanup_errors = [] + for policy in reversed(created_policies): + if policy.get("AttachAttempted"): + try: + iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) + except iam_client.exceptions.NoSuchEntityException: + pass + except Exception as e: + cleanup_errors.append(f"detach {policy['PolicyArn']}: {e}") + try: + iam_client.delete_policy(PolicyArn=policy["PolicyArn"]) + except Exception as e: + cleanup_errors.append(f"delete {policy['PolicyArn']}: {e}") + + for version in reversed(staged_versions): + if version.get("Activated"): + try: + iam_client.set_default_policy_version( + PolicyArn=version["PolicyArn"], + VersionId=version["PreviousVersionId"], + ) + except Exception as e: + cleanup_errors.append(f"restore {version['PolicyArn']}: {e}") + continue + try: + iam_client.delete_policy_version( + PolicyArn=version["PolicyArn"], + VersionId=version["StagedVersionId"], + ) + except Exception as e: + cleanup_errors.append(f"delete staged version for {version['PolicyArn']}: {e}") + return cleanup_errors + + +def replace_instrumentation_policies(iam_client, role_name, account_id, partition, policy_documents): + attached_policies = _list_attached_role_policies(iam_client, role_name) + attachment_quota, version_quota = _policy_quotas(iam_client) + existing_policies = {} + for policy in attached_policies: + index = _instrumentation_policy_index(policy["PolicyName"], role_name) + if index is not None: + existing_policies[index] = policy + + non_instrumentation_count = len(attached_policies) - len(existing_policies) + final_attachment_count = non_instrumentation_count + len(policy_documents) + if final_attachment_count > attachment_quota: + raise RuntimeError( + f"role {role_name} has room for only " + f"{attachment_quota - non_instrumentation_count} instrumentation policies, " + f"but Datadog returned {len(policy_documents)}" + ) + + LOGGER.info( + f"Replacing {len(existing_policies)} instrumentation policies with " + f"{len(policy_documents)} documents; final managed-policy usage " + f"{final_attachment_count}/{attachment_quota}" + ) + ordered_existing_policies = [ + policy for _, policy in sorted(existing_policies.items()) + ] + used_policy_indices = set(existing_policies) + next_policy_index = 1 + reused_policy_arns = set() + staged_versions = [] + created_policies = [] + try: + for document_index, policy_document in enumerate(policy_documents): + existing = ( + ordered_existing_policies[document_index] + if document_index < len(ordered_existing_policies) + else None + ) + if existing: + reused_policy_arns.add(existing["PolicyArn"]) + staged_versions.append( + _stage_policy_version( + iam_client, + existing["PolicyArn"], + policy_document, + version_quota, + ) + ) + continue + + while next_policy_index in used_policy_indices: + next_policy_index += 1 + policy_name = _instrumentation_policy_name(role_name, next_policy_index) + used_policy_indices.add(next_policy_index) + policy_arn = f"arn:{partition}:iam::{account_id}:policy/{policy_name}" + try: + iam_client.delete_policy(PolicyArn=policy_arn) + except iam_client.exceptions.NoSuchEntityException: + pass + created_policies.append({ + "PolicyArn": _create_policy(iam_client, policy_name, policy_document), + "AttachAttempted": False, + }) + + for version in staged_versions: + iam_client.set_default_policy_version( + PolicyArn=version["PolicyArn"], + VersionId=version["StagedVersionId"], + ) + version["Activated"] = True + for policy in created_policies: + policy["AttachAttempted"] = True + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) + except Exception as e: + cleanup_errors = _discard_staged_replacement( + iam_client, + role_name, + staged_versions, + created_policies, + ) + if cleanup_errors: + raise RuntimeError( + f"failed to replace instrumentation policies: {e}; " + f"rollback errors: {'; '.join(cleanup_errors)}" + ) from e + raise + + cleanup_errors = [] + for policy in existing_policies.values(): + if policy["PolicyArn"] in reused_policy_arns: + continue + cleanup_errors.extend(_detach_and_delete_policy( + iam_client, + role_name, + policy["PolicyArn"], + policy["PolicyName"], + )) + if cleanup_errors: + raise RuntimeError( + f"new instrumentation policies are active, but old policy cleanup failed: " + f"{'; '.join(cleanup_errors)}" + ) + + def attach_instrumentation_permissions(iam_client, role_name, account_id, partition, datadog_site, resource_types, previous_resource_types, fail_on_error=False): # Best-effort by default: instrumentation permissions are additive convenience on top of the # integration, so any failure is logged and swallowed rather than blocking install. The # post-setup add-on passes fail_on_error=True because attaching these policies is the stack's # whole purpose, so a silent SUCCESS that attached nothing would be worse than a visible failure. - # Fetch before cleanup so that a transient API failure on an Update leaves the - # previously-attached policies in place instead of silently revoking them. + # Fetch and stage replacements before activation so failures can restore the + # previously-attached policy versions. if not resource_types: # Only clean up if the previous Update had instrumentation enabled — avoids running # delete calls on stacks that never opted in to instrumentation in the first place. @@ -177,28 +463,30 @@ def attach_instrumentation_permissions(iam_client, role_name, account_id, partit return try: - url = build_instrumentation_permissions_url(datadog_site, resource_types) + url = build_instrumentation_permissions_url( + datadog_site, + resource_types, + account_id, + partition, + ) LOGGER.info(f"Fetching instrumentation permissions for {resource_types} from {url}") - permission_chunks = fetch_permissions_from_datadog(url) + policy_documents = fetch_instrumentation_policy_documents(url) + replace_instrumentation_policies( + iam_client, + role_name, + account_id, + partition, + policy_documents, + ) except Exception as e: if fail_on_error: raise LOGGER.warning( - f"Failed to fetch instrumentation permissions for {resource_types}: {e}. " + f"Failed to update instrumentation permissions for {resource_types}: {e}. " "Leaving any previously-attached instrumentation policies in place." ) return - cleanup_instrumentation_policies(iam_client, role_name, account_id, partition) - for i, chunk in enumerate(permission_chunks): - policy_name = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{i+1}" - try: - _create_and_attach_policy(iam_client, role_name, policy_name, chunk) - except Exception as e: - if fail_on_error: - raise - LOGGER.warning(f"Failed to create/attach instrumentation policy {policy_name}: {e}. Continuing.") - def handle_delete(event, context): props = event['ResourceProperties'] diff --git a/aws_quickstart/attach_integration_permissions_test.py b/aws_quickstart/attach_integration_permissions_test.py index 191ef671..659103b1 100644 --- a/aws_quickstart/attach_integration_permissions_test.py +++ b/aws_quickstart/attach_integration_permissions_test.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import json +from pathlib import Path import sys import unittest from unittest.mock import patch, Mock, MagicMock, call @@ -30,6 +31,23 @@ ) +class TestEmbeddedLambdaSource(unittest.TestCase): + def test_cloudformation_inline_lambda_matches_tested_source(self): + source_path = Path(__file__).with_name("attach_integration_permissions.py") + template_path = Path(__file__).with_name("datadog_integration_permissions.yaml") + source = source_path.read_text().rstrip("\n") + template = template_path.read_text() + start_marker = " Code:\n ZipFile: |\n" + end_marker = " DatadogAttachIntegrationPermissionsFunctionTrigger:\n" + start = template.index(start_marker) + len(start_marker) + end = template.index(end_marker, start) + embedded = "\n".join( + line[10:] if line else "" + for line in template[start:end].rstrip("\n").splitlines() + ) + self.assertEqual(embedded, source) + + def make_iam_mock(cleanup_side_effects=True): iam = MagicMock() iam.exceptions.NoSuchEntityException = type("NSE", (Exception,), {}) @@ -76,7 +94,12 @@ def _query_pairs(self, url): return parse_qsl(urlparse(url).query) def test_path_and_host(self): - url = build_instrumentation_permissions_url("datadoghq.eu", ["aws:ec2:instance"]) + url = build_instrumentation_permissions_url( + "datadoghq.eu", + ["aws:ec2:instance"], + "123456789012", + "aws", + ) parsed = urlparse(url) self.assertEqual(parsed.scheme, "https") self.assertEqual(parsed.netloc, "api.datadoghq.eu") @@ -86,6 +109,8 @@ def test_repeated_resource_type_and_chunked(self): url = build_instrumentation_permissions_url( "datadoghq.com", ["aws:ec2:instance", "aws:ecs:cluster", "aws:eks:cluster"], + "123456789012", + "aws-us-gov", ) pairs = self._query_pairs(url) resource_types = [v for k, v in pairs if k == "resource_type"] @@ -93,13 +118,22 @@ def test_repeated_resource_type_and_chunked(self): resource_types, ["aws:ec2:instance", "aws:ecs:cluster", "aws:eks:cluster"], ) + self.assertIn(("account_id", "123456789012"), pairs) + self.assertIn(("partition", "aws-us-gov"), pairs) self.assertIn(("chunked", "true"), pairs) class TestAttachInstrumentationPermissions(unittest.TestCase): def setUp(self): - self.iam = make_iam_mock() + self.iam = make_iam_mock(cleanup_side_effects=False) self.iam.create_policy.return_value = {"Policy": {"Arn": "arn:aws:iam::123:policy/X"}} + self.iam.list_attached_role_policies.return_value = {"AttachedPolicies": []} + self.iam.get_account_summary.return_value = { + "SummaryMap": { + "AttachedPoliciesPerRoleQuota": 10, + "VersionsPerPolicyQuota": 5, + } + } self.role_name = "DatadogIntegrationRole" self.account_id = "123456789012" self.partition = "aws" @@ -117,6 +151,14 @@ def _mock_chunks_response(self, chunks): resp.read.return_value = body return resp + def _mock_documents_response(self, documents): + body = json.dumps( + {"data": {"attributes": {"policy_documents": documents}}} + ).encode() + resp = Mock() + resp.read.return_value = body + return resp + def test_empty_resource_types_no_op_when_previously_empty(self): # Stack Create (or Update with no change) and no instrumentation requested: # don't touch IAM at all — there's nothing to clean up. @@ -134,15 +176,40 @@ def test_empty_resource_types_cleans_up_when_previously_set(self): self.assertGreater(self.iam.detach_role_policy.call_count, 0) @patch("attach_integration_permissions.urllib.request.urlopen") - def test_happy_path_attaches_each_chunk(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response( - [["ec2:Describe*"], ["ssm:SendCommand", "eks:DescribeCluster"]] - ) + def test_happy_path_attaches_each_policy_document(self, mock_urlopen): + documents = [ + { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": ["iam:PassRole"], + "Resource": ["arn:aws:iam::123456789012:role/datadog-*"], + } + ], + }, + { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": ["ec2:DescribeInstances"], + "Resource": ["*"], + } + ], + }, + ] + mock_urlopen.return_value = self._mock_documents_response(documents) self._attach(["aws:ec2:instance", "aws:eks:cluster"]) self.assertEqual(self.iam.create_policy.call_count, 2) self.assertEqual(self.iam.attach_role_policy.call_count, 2) + created_documents = [ + json.loads(c.kwargs["PolicyDocument"]) + for c in self.iam.create_policy.call_args_list + ] + self.assertEqual(created_documents, documents) names = [c.kwargs["PolicyName"] for c in self.iam.create_policy.call_args_list] self.assertEqual( @@ -155,6 +222,26 @@ def test_happy_path_attaches_each_chunk(self, mock_urlopen): sent_request = mock_urlopen.call_args[0][0] self.assertEqual(sent_request.headers.get("Dd-aws-api-call-source"), "cfn-quickstart") + query = parse_qsl(urlparse(sent_request.full_url).query) + self.assertIn(("account_id", self.account_id), query) + self.assertIn(("partition", self.partition), query) + + @patch("attach_integration_permissions.urllib.request.urlopen") + def test_legacy_response_is_wrapped_as_broad_policy_documents(self, mock_urlopen): + mock_urlopen.return_value = self._mock_chunks_response( + [["ec2:Describe*"], ["ssm:SendCommand"]] + ) + + self._attach(["aws:ec2:instance"]) + + created_documents = [ + json.loads(c.kwargs["PolicyDocument"]) + for c in self.iam.create_policy.call_args_list + ] + self.assertEqual( + [document["Statement"][0]["Action"] for document in created_documents], + [["ec2:Describe*"], ["ssm:SendCommand"]], + ) @patch("attach_integration_permissions.urllib.request.urlopen") def test_fetch_failure_preserves_existing_policies(self, mock_urlopen): @@ -173,20 +260,20 @@ def test_fetch_failure_preserves_existing_policies(self, mock_urlopen): self.iam.delete_policy.assert_not_called() @patch("attach_integration_permissions.urllib.request.urlopen") - def test_per_chunk_failure_is_swallowed_and_others_continue(self, mock_urlopen): + def test_create_failure_rolls_back_without_attaching_partial_documents(self, mock_urlopen): mock_urlopen.return_value = self._mock_chunks_response( - [["chunk1:Action"], ["chunk2:Action"], ["chunk3:Action"]] + [["chunk1:Action"], ["chunk2:Action"]] ) self.iam.create_policy.side_effect = [ {"Policy": {"Arn": "arn:aws:iam::123:policy/A"}}, Exception("EntityAlreadyExists"), - {"Policy": {"Arn": "arn:aws:iam::123:policy/C"}}, ] self._attach(["aws:ec2:instance"]) - self.assertEqual(self.iam.create_policy.call_count, 3) - self.assertEqual(self.iam.attach_role_policy.call_count, 2) + self.assertEqual(self.iam.create_policy.call_count, 2) + self.iam.attach_role_policy.assert_not_called() + self.iam.delete_policy.assert_any_call(PolicyArn="arn:aws:iam::123:policy/A") @patch("attach_integration_permissions.urllib.request.urlopen") def test_fail_on_error_raises_on_fetch_failure(self, mock_urlopen): @@ -204,12 +291,113 @@ def test_fail_on_error_raises_on_fetch_failure(self, mock_urlopen): @patch("attach_integration_permissions.urllib.request.urlopen") def test_fail_on_error_raises_on_attach_failure(self, mock_urlopen): mock_urlopen.return_value = self._mock_chunks_response([["chunk1:Action"]]) - self.iam.create_policy.side_effect = Exception("AccessDenied") + self.iam.attach_role_policy.side_effect = Exception("AccessDenied") with self.assertRaises(Exception): attach_instrumentation_permissions( self.iam, self.role_name, self.account_id, self.partition, self.site, ["aws:ec2:instance"], (), fail_on_error=True, ) + self.iam.delete_policy.assert_any_call(PolicyArn="arn:aws:iam::123:policy/X") + + @patch("attach_integration_permissions.urllib.request.urlopen") + def test_quota_failure_happens_before_policy_mutation(self, mock_urlopen): + mock_urlopen.return_value = self._mock_chunks_response( + [["chunk1:Action"], ["chunk2:Action"]] + ) + self.iam.list_attached_role_policies.return_value = { + "AttachedPolicies": [ + { + "PolicyName": f"CustomerPolicy{index}", + "PolicyArn": f"arn:aws:iam::123:policy/CustomerPolicy{index}", + } + for index in range(9) + ] + } + + with self.assertRaisesRegex(RuntimeError, "room for only 1 instrumentation policies"): + attach_instrumentation_permissions( + self.iam, self.role_name, self.account_id, self.partition, self.site, + ["aws:ec2:instance"], (), fail_on_error=True, + ) + + self.iam.create_policy.assert_not_called() + self.iam.create_policy_version.assert_not_called() + self.iam.attach_role_policy.assert_not_called() + self.iam.detach_role_policy.assert_not_called() + + @patch("attach_integration_permissions.urllib.request.urlopen") + def test_existing_policy_is_staged_as_a_new_version(self, mock_urlopen): + mock_urlopen.return_value = self._mock_chunks_response([["new:Action"]]) + policy_arn = ( + f"arn:aws:iam::{self.account_id}:policy/" + f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1" + ) + self.iam.list_attached_role_policies.return_value = { + "AttachedPolicies": [{ + "PolicyName": f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1", + "PolicyArn": policy_arn, + }] + } + self.iam.list_policy_versions.return_value = { + "Versions": [{"VersionId": "v1", "IsDefaultVersion": True}] + } + self.iam.create_policy_version.return_value = { + "PolicyVersion": {"VersionId": "v2"} + } + + self._attach(["aws:ec2:instance"]) + + self.iam.create_policy.assert_not_called() + self.iam.create_policy_version.assert_called_once() + self.iam.set_default_policy_version.assert_called_once_with( + PolicyArn=policy_arn, + VersionId="v2", + ) + self.iam.detach_role_policy.assert_not_called() + + @patch("attach_integration_permissions.urllib.request.urlopen") + def test_attach_failure_restores_previous_policy_version(self, mock_urlopen): + mock_urlopen.return_value = self._mock_chunks_response( + [["updated:Action"], ["new:Action"]] + ) + existing_arn = ( + f"arn:aws:iam::{self.account_id}:policy/" + f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1" + ) + self.iam.list_attached_role_policies.return_value = { + "AttachedPolicies": [{ + "PolicyName": f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1", + "PolicyArn": existing_arn, + }] + } + self.iam.list_policy_versions.return_value = { + "Versions": [{"VersionId": "v1", "IsDefaultVersion": True}] + } + self.iam.create_policy_version.return_value = { + "PolicyVersion": {"VersionId": "v2"} + } + self.iam.create_policy.return_value = { + "Policy": {"Arn": "arn:aws:iam::123:policy/New"} + } + self.iam.attach_role_policy.side_effect = Exception("LimitExceeded") + + self._attach(["aws:ec2:instance"]) + + self.assertEqual( + self.iam.set_default_policy_version.call_args_list, + [ + call(PolicyArn=existing_arn, VersionId="v2"), + call(PolicyArn=existing_arn, VersionId="v1"), + ], + ) + self.iam.delete_policy_version.assert_called_with( + PolicyArn=existing_arn, + VersionId="v2", + ) + self.iam.detach_role_policy.assert_called_once_with( + RoleName=self.role_name, + PolicyArn="arn:aws:iam::123:policy/New", + ) class TestCleanup(unittest.TestCase): diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 5bf11c80..2e01027c 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -71,11 +71,16 @@ Resources: - Effect: Allow Action: - iam:CreatePolicy + - iam:CreatePolicyVersion - iam:DeletePolicy + - iam:DeletePolicyVersion - iam:DeleteRolePolicy - iam:AttachRolePolicy - iam:DetachRolePolicy + - iam:ListAttachedRolePolicies + - iam:ListPolicyVersions - iam:PutRolePolicy + - iam:SetDefaultPolicyVersion Resource: # Wildcards cover both the v2 names this template creates and the un-suffixed legacy # names it cleans up on an in-place upgrade. @@ -83,6 +88,9 @@ Resources: - !Sub arn:${AWS::Partition}:iam::${AWS::AccountId}:policy/datadog-aws-integration-resource-collection-permissions-* - !Sub arn:${AWS::Partition}:iam::${AWS::AccountId}:policy/datadog-aws-integration-instrumentation-permissions-* - !Sub "arn:${AWS::Partition}:iam::aws:policy/SecurityAudit" + - Effect: Allow + Action: iam:GetAccountSummary + Resource: "*" DatadogAttachIntegrationPermissionsFunction: Type: AWS::Lambda::Function Properties: @@ -131,13 +139,14 @@ Resources: STANDARD_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/standard" RESOURCE_COLLECTION_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/resource_collection?chunked=true" INSTRUMENTATION_PERMISSIONS_API_PATH = "/api/unstable/instrumenter/aws/iam_permissions" + MANAGED_POLICY_CHAR_LIMIT = 6144 class DatadogAPIError(Exception): pass - def fetch_permissions_from_datadog(api_url): + def fetch_permissions_attributes_from_datadog(api_url): headers = { "Dd-Aws-Api-Call-Source": API_CALL_SOURCE_HEADER_VALUE, } @@ -151,7 +160,36 @@ Resources: error_message = error_body.get('errors', ['Unknown error'])[0] raise DatadogAPIError(f"Datadog API error: {error_message}") from e - return json.loads(response.read())["data"]["attributes"]["permissions"] + return json.loads(response.read())["data"]["attributes"] + + + def fetch_permissions_from_datadog(api_url): + return fetch_permissions_attributes_from_datadog(api_url)["permissions"] + + + def fetch_instrumentation_policy_documents(api_url): + attributes = fetch_permissions_attributes_from_datadog(api_url) + policy_documents = attributes.get("policy_documents") + if policy_documents is not None: + if not policy_documents: + raise DatadogAPIError("Datadog API returned no instrumentation policy documents") + for document in policy_documents: + _serialize_policy_document(document) + return policy_documents + + permission_chunks = attributes.get("permissions") + if not permission_chunks: + raise DatadogAPIError("Datadog API returned neither policy_documents nor permissions") + if isinstance(permission_chunks[0], str): + permission_chunks = [permission_chunks] + LOGGER.warning("Datadog API does not expose policy_documents yet; using legacy broad permissions") + return [ + { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": chunk, "Resource": "*"}], + } + for chunk in permission_chunks + ] def parse_resource_types(raw): @@ -163,9 +201,14 @@ Resources: return [t.strip() for t in items if t and t.strip()] - def build_instrumentation_permissions_url(datadog_site, resource_types): + def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( - [("resource_type", t) for t in resource_types] + [("chunked", "true")] + [("resource_type", t) for t in resource_types] + + [ + ("account_id", account_id), + ("partition", partition), + ("chunked", "true"), + ] ) return f"https://api.{datadog_site}{INSTRUMENTATION_PERMISSIONS_API_PATH}?{query}" @@ -173,12 +216,14 @@ Resources: def _detach_and_delete_policy(iam_client, role_name, policy_arn, policy_name): # Detach + delete are both no-ops if the entity is already gone, so callers can blindly # iterate the policy-name space without first checking what actually exists. + errors = [] try: iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy_arn) except iam_client.exceptions.NoSuchEntityException: pass except Exception as e: LOGGER.error(f"Error detaching policy {policy_name}: {str(e)}") + errors.append(f"detach {policy_name}: {e}") try: iam_client.delete_policy(PolicyArn=policy_arn) @@ -186,8 +231,11 @@ Resources: pass except iam_client.exceptions.DeleteConflictException: LOGGER.warning(f"Policy {policy_name} still attached, skipping delete") + errors.append(f"delete {policy_name}: policy is still attached") except Exception as e: LOGGER.error(f"Error deleting policy {policy_name}: {str(e)}") + errors.append(f"delete {policy_name}: {e}") + return errors def _cleanup_chunked_policies(iam_client, role_name, account_id, partition, prefix, max_policies=10): @@ -236,17 +284,36 @@ Resources: ) - def _create_and_attach_policy(iam_client, role_name, policy_name, actions): - policy_json = json.dumps( - { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], - }, - separators=(',', ':'), - ) - LOGGER.info(f"Creating policy {policy_name} with {len(actions)} permissions ({len(policy_json)} characters)") + def _serialize_policy_document(policy_document): + if not isinstance(policy_document, dict): + raise DatadogAPIError("instrumentation policy document must be an object") + if policy_document.get("Version") != "2012-10-17": + raise DatadogAPIError("instrumentation policy document has an unsupported version") + if not isinstance(policy_document.get("Statement"), list) or not policy_document["Statement"]: + raise DatadogAPIError("instrumentation policy document must contain statements") + + policy_json = json.dumps(policy_document, separators=(',', ':')) + if len(policy_json) > MANAGED_POLICY_CHAR_LIMIT: + raise DatadogAPIError( + f"instrumentation policy document exceeds {MANAGED_POLICY_CHAR_LIMIT} characters" + ) + return policy_json + + + def _create_policy(iam_client, policy_name, policy_document): + policy_json = _serialize_policy_document(policy_document) + LOGGER.info(f"Creating policy {policy_name} ({len(policy_json)} characters)") policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=policy_json) - iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy['Policy']['Arn']) + return policy['Policy']['Arn'] + + + def _create_and_attach_policy(iam_client, role_name, policy_name, actions): + policy_document = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], + } + policy_arn = _create_policy(iam_client, policy_name, policy_document) + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) def attach_resource_collection_permissions(iam_client, role_name): @@ -260,13 +327,240 @@ Resources: ) + def _list_attached_role_policies(iam_client, role_name): + policies = [] + marker = None + while True: + request = {"RoleName": role_name} + if marker: + request["Marker"] = marker + response = iam_client.list_attached_role_policies(**request) + policies.extend(response.get("AttachedPolicies", [])) + if not response.get("IsTruncated"): + return policies + marker = response["Marker"] + + + def _list_policy_versions(iam_client, policy_arn): + versions = [] + marker = None + while True: + request = {"PolicyArn": policy_arn} + if marker: + request["Marker"] = marker + response = iam_client.list_policy_versions(**request) + versions.extend(response.get("Versions", [])) + if not response.get("IsTruncated"): + return versions + marker = response["Marker"] + + + def _policy_quotas(iam_client): + summary = iam_client.get_account_summary().get("SummaryMap", {}) + attachment_quota = summary.get("AttachedPoliciesPerRoleQuota") + version_quota = summary.get("VersionsPerPolicyQuota") + if not isinstance(attachment_quota, int) or attachment_quota < 1: + raise RuntimeError("IAM account summary did not include AttachedPoliciesPerRoleQuota") + if not isinstance(version_quota, int) or version_quota < 2: + raise RuntimeError("IAM account summary did not include VersionsPerPolicyQuota") + return attachment_quota, version_quota + + + def _instrumentation_policy_name(role_name, index): + return f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{index}" + + + def _instrumentation_policy_index(policy_name, role_name): + prefix = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-" + if not policy_name.startswith(prefix): + return None + suffix = policy_name[len(prefix):] + if not suffix.isdigit() or int(suffix) < 1: + return None + return int(suffix) + + + def _stage_policy_version(iam_client, policy_arn, policy_document, version_quota): + versions = _list_policy_versions(iam_client, policy_arn) + previous_default = next( + (version["VersionId"] for version in versions if version.get("IsDefaultVersion")), + None, + ) + if previous_default is None: + raise RuntimeError(f"policy {policy_arn} has no default version") + + non_default_versions = sorted( + (version for version in versions if not version.get("IsDefaultVersion")), + key=lambda version: int(version["VersionId"].removeprefix("v")), + ) + while len(versions) >= version_quota: + if not non_default_versions: + raise RuntimeError(f"policy {policy_arn} has no removable non-default version") + version = non_default_versions.pop(0) + iam_client.delete_policy_version( + PolicyArn=policy_arn, + VersionId=version["VersionId"], + ) + versions.remove(version) + + response = iam_client.create_policy_version( + PolicyArn=policy_arn, + PolicyDocument=_serialize_policy_document(policy_document), + SetAsDefault=False, + ) + return { + "PolicyArn": policy_arn, + "PreviousVersionId": previous_default, + "StagedVersionId": response["PolicyVersion"]["VersionId"], + } + + + def _discard_staged_replacement(iam_client, role_name, staged_versions, created_policies): + cleanup_errors = [] + for policy in reversed(created_policies): + if policy.get("AttachAttempted"): + try: + iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) + except iam_client.exceptions.NoSuchEntityException: + pass + except Exception as e: + cleanup_errors.append(f"detach {policy['PolicyArn']}: {e}") + try: + iam_client.delete_policy(PolicyArn=policy["PolicyArn"]) + except Exception as e: + cleanup_errors.append(f"delete {policy['PolicyArn']}: {e}") + + for version in reversed(staged_versions): + if version.get("Activated"): + try: + iam_client.set_default_policy_version( + PolicyArn=version["PolicyArn"], + VersionId=version["PreviousVersionId"], + ) + except Exception as e: + cleanup_errors.append(f"restore {version['PolicyArn']}: {e}") + continue + try: + iam_client.delete_policy_version( + PolicyArn=version["PolicyArn"], + VersionId=version["StagedVersionId"], + ) + except Exception as e: + cleanup_errors.append(f"delete staged version for {version['PolicyArn']}: {e}") + return cleanup_errors + + + def replace_instrumentation_policies(iam_client, role_name, account_id, partition, policy_documents): + attached_policies = _list_attached_role_policies(iam_client, role_name) + attachment_quota, version_quota = _policy_quotas(iam_client) + existing_policies = {} + for policy in attached_policies: + index = _instrumentation_policy_index(policy["PolicyName"], role_name) + if index is not None: + existing_policies[index] = policy + + non_instrumentation_count = len(attached_policies) - len(existing_policies) + final_attachment_count = non_instrumentation_count + len(policy_documents) + if final_attachment_count > attachment_quota: + raise RuntimeError( + f"role {role_name} has room for only " + f"{attachment_quota - non_instrumentation_count} instrumentation policies, " + f"but Datadog returned {len(policy_documents)}" + ) + + LOGGER.info( + f"Replacing {len(existing_policies)} instrumentation policies with " + f"{len(policy_documents)} documents; final managed-policy usage " + f"{final_attachment_count}/{attachment_quota}" + ) + ordered_existing_policies = [ + policy for _, policy in sorted(existing_policies.items()) + ] + used_policy_indices = set(existing_policies) + next_policy_index = 1 + reused_policy_arns = set() + staged_versions = [] + created_policies = [] + try: + for document_index, policy_document in enumerate(policy_documents): + existing = ( + ordered_existing_policies[document_index] + if document_index < len(ordered_existing_policies) + else None + ) + if existing: + reused_policy_arns.add(existing["PolicyArn"]) + staged_versions.append( + _stage_policy_version( + iam_client, + existing["PolicyArn"], + policy_document, + version_quota, + ) + ) + continue + + while next_policy_index in used_policy_indices: + next_policy_index += 1 + policy_name = _instrumentation_policy_name(role_name, next_policy_index) + used_policy_indices.add(next_policy_index) + policy_arn = f"arn:{partition}:iam::{account_id}:policy/{policy_name}" + try: + iam_client.delete_policy(PolicyArn=policy_arn) + except iam_client.exceptions.NoSuchEntityException: + pass + created_policies.append({ + "PolicyArn": _create_policy(iam_client, policy_name, policy_document), + "AttachAttempted": False, + }) + + for version in staged_versions: + iam_client.set_default_policy_version( + PolicyArn=version["PolicyArn"], + VersionId=version["StagedVersionId"], + ) + version["Activated"] = True + for policy in created_policies: + policy["AttachAttempted"] = True + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) + except Exception as e: + cleanup_errors = _discard_staged_replacement( + iam_client, + role_name, + staged_versions, + created_policies, + ) + if cleanup_errors: + raise RuntimeError( + f"failed to replace instrumentation policies: {e}; " + f"rollback errors: {'; '.join(cleanup_errors)}" + ) from e + raise + + cleanup_errors = [] + for policy in existing_policies.values(): + if policy["PolicyArn"] in reused_policy_arns: + continue + cleanup_errors.extend(_detach_and_delete_policy( + iam_client, + role_name, + policy["PolicyArn"], + policy["PolicyName"], + )) + if cleanup_errors: + raise RuntimeError( + f"new instrumentation policies are active, but old policy cleanup failed: " + f"{'; '.join(cleanup_errors)}" + ) + + def attach_instrumentation_permissions(iam_client, role_name, account_id, partition, datadog_site, resource_types, previous_resource_types, fail_on_error=False): # Best-effort by default: instrumentation permissions are additive convenience on top of the # integration, so any failure is logged and swallowed rather than blocking install. The # post-setup add-on passes fail_on_error=True because attaching these policies is the stack's # whole purpose, so a silent SUCCESS that attached nothing would be worse than a visible failure. - # Fetch before cleanup so that a transient API failure on an Update leaves the - # previously-attached policies in place instead of silently revoking them. + # Fetch and stage replacements before activation so failures can restore the + # previously-attached policy versions. if not resource_types: # Only clean up if the previous Update had instrumentation enabled — avoids running # delete calls on stacks that never opted in to instrumentation in the first place. @@ -275,28 +569,30 @@ Resources: return try: - url = build_instrumentation_permissions_url(datadog_site, resource_types) + url = build_instrumentation_permissions_url( + datadog_site, + resource_types, + account_id, + partition, + ) LOGGER.info(f"Fetching instrumentation permissions for {resource_types} from {url}") - permission_chunks = fetch_permissions_from_datadog(url) + policy_documents = fetch_instrumentation_policy_documents(url) + replace_instrumentation_policies( + iam_client, + role_name, + account_id, + partition, + policy_documents, + ) except Exception as e: if fail_on_error: raise LOGGER.warning( - f"Failed to fetch instrumentation permissions for {resource_types}: {e}. " + f"Failed to update instrumentation permissions for {resource_types}: {e}. " "Leaving any previously-attached instrumentation policies in place." ) return - cleanup_instrumentation_policies(iam_client, role_name, account_id, partition) - for i, chunk in enumerate(permission_chunks): - policy_name = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{i+1}" - try: - _create_and_attach_policy(iam_client, role_name, policy_name, chunk) - except Exception as e: - if fail_on_error: - raise - LOGGER.warning(f"Failed to create/attach instrumentation policy {policy_name}: {e}. Continuing.") - def handle_delete(event, context): props = event['ResourceProperties'] From 227e6b04daba43cc55f3024fe34872e6d7be3d3b Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Mon, 13 Jul 2026 14:51:06 -0400 Subject: [PATCH 2/4] Bump quickstart template version --- aws_quickstart/version.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws_quickstart/version.txt b/aws_quickstart/version.txt index 1597f02b..6112fedd 100644 --- a/aws_quickstart/version.txt +++ b/aws_quickstart/version.txt @@ -1 +1 @@ -v4.15.1 +v4.16.0 From 1db3a3ab824cedf162add207d67ff0dc28f94b70 Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Mon, 13 Jul 2026 17:12:25 -0400 Subject: [PATCH 3/4] Simplify integration policy handling --- .../attach_integration_permissions.py | 75 ++++------ .../attach_integration_permissions_test.py | 133 +++++++----------- .../datadog_integration_permissions.yaml | 75 ++++------ 3 files changed, 112 insertions(+), 171 deletions(-) diff --git a/aws_quickstart/attach_integration_permissions.py b/aws_quickstart/attach_integration_permissions.py index c2a85932..4ef19ff5 100644 --- a/aws_quickstart/attach_integration_permissions.py +++ b/aws_quickstart/attach_integration_permissions.py @@ -10,30 +10,20 @@ LOGGER = logging.getLogger() LOGGER.setLevel(logging.INFO) API_CALL_SOURCE_HEADER_VALUE = "cfn-quickstart" -# The "-v2" suffix on these policy names is load-bearing, not cosmetic. The pre-extraction -# inline trigger (<= v4.13) deletes policies by their un-suffixed names on teardown, and that -# teardown runs whenever the old trigger is removed — i.e. when a role stack is upgraded off -# <= v4.13. Distinct v2 names ensure that destructive delete can never hit the policies this -# template attaches: -# - standard / resource-collection: an in-place role-stack upgrade removes the old trigger -# after this nested stack has re-attached them; v2 names keep them from being wiped. -# - instrumentation: the add-on attaches these against an existing role; if that role's stack -# is later upgraded off <= v4.13, the old trigger's unconditional instrumentation cleanup -# would wipe them unless they sit under a name it doesn't know. +# The <= v4.13 inline trigger deletes un-suffixed policies when removed during an upgrade. +# Keep replacement policies on v2 names so that teardown cannot delete newly attached policies. POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicyV2" BASE_POLICY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions-v2" BASE_POLICY_PREFIX_INSTRUMENTATION = "datadog-aws-integration-instrumentation-permissions-v2" -# Un-suffixed standard/resource-collection names created by the pre-extraction inline trigger -# (<= v4.13). The role-creation path cleans these up before attaching the v2 policies so the two -# generations never sit attached at once (IAM caps managed policies per role, default 10); the -# old trigger's own Delete handler then no-ops against names that are already gone. Legacy -# instrumentation policies need no such cleanup — that feature is unreleased, so none exist. +# Remove legacy base policies before attaching v2 policies to avoid exceeding the role quota. +# Instrumentation was not released under legacy names, so it needs no equivalent cleanup list. LEGACY_POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicy" LEGACY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions" STANDARD_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/standard" RESOURCE_COLLECTION_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/resource_collection?chunked=true" INSTRUMENTATION_PERMISSIONS_API_PATH = "/api/unstable/instrumenter/aws/iam_permissions" MANAGED_POLICY_CHAR_LIMIT = 6144 +POLICY_DOCUMENT_VERSION = "2012-10-17" class DatadogAPIError(Exception): @@ -61,6 +51,13 @@ def fetch_permissions_from_datadog(api_url): return fetch_permissions_attributes_from_datadog(api_url)["permissions"] +def _broad_policy_document(actions): + return { + "Version": POLICY_DOCUMENT_VERSION, + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], + } + + def fetch_instrumentation_policy_documents(api_url): attributes = fetch_permissions_attributes_from_datadog(api_url) policy_documents = attributes.get("policy_documents") @@ -77,13 +74,7 @@ def fetch_instrumentation_policy_documents(api_url): if isinstance(permission_chunks[0], str): permission_chunks = [permission_chunks] LOGGER.warning("Datadog API does not expose policy_documents yet; using legacy broad permissions") - return [ - { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": chunk, "Resource": "*"}], - } - for chunk in permission_chunks - ] + return [_broad_policy_document(chunk) for chunk in permission_chunks] def parse_resource_types(raw): @@ -95,6 +86,10 @@ def parse_resource_types(raw): return [t.strip() for t in items if t and t.strip()] +def _is_true(value): + return str(value).lower() == "true" + + def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( [("resource_type", t) for t in resource_types] @@ -167,21 +162,17 @@ def cleanup_legacy_base_policies(iam_client, role_name, account_id, partition, m def attach_standard_permissions(iam_client, role_name): permissions = fetch_permissions_from_datadog(STANDARD_PERMISSIONS_API_URL) - policy_document = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": permissions, "Resource": "*"}], - } iam_client.put_role_policy( RoleName=role_name, PolicyName=POLICY_NAME_STANDARD, - PolicyDocument=json.dumps(policy_document, separators=(',', ':')), + PolicyDocument=json.dumps(_broad_policy_document(permissions), separators=(',', ':')), ) def _serialize_policy_document(policy_document): if not isinstance(policy_document, dict): raise DatadogAPIError("instrumentation policy document must be an object") - if policy_document.get("Version") != "2012-10-17": + if policy_document.get("Version") != POLICY_DOCUMENT_VERSION: raise DatadogAPIError("instrumentation policy document has an unsupported version") if not isinstance(policy_document.get("Statement"), list) or not policy_document["Statement"]: raise DatadogAPIError("instrumentation policy document must contain statements") @@ -202,11 +193,7 @@ def _create_policy(iam_client, policy_name, policy_document): def _create_and_attach_policy(iam_client, role_name, policy_name, actions): - policy_document = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], - } - policy_arn = _create_policy(iam_client, policy_name, policy_document) + policy_arn = _create_policy(iam_client, policy_name, _broad_policy_document(actions)) iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) @@ -269,9 +256,10 @@ def _instrumentation_policy_index(policy_name, role_name): if not policy_name.startswith(prefix): return None suffix = policy_name[len(prefix):] - if not suffix.isdigit() or int(suffix) < 1: + if not suffix.isdigit(): return None - return int(suffix) + index = int(suffix) + return index if index >= 1 else None def _stage_policy_version(iam_client, policy_arn, policy_document, version_quota): @@ -449,12 +437,8 @@ def replace_instrumentation_policies(iam_client, role_name, account_id, partitio def attach_instrumentation_permissions(iam_client, role_name, account_id, partition, datadog_site, resource_types, previous_resource_types, fail_on_error=False): - # Best-effort by default: instrumentation permissions are additive convenience on top of the - # integration, so any failure is logged and swallowed rather than blocking install. The - # post-setup add-on passes fail_on_error=True because attaching these policies is the stack's - # whole purpose, so a silent SUCCESS that attached nothing would be worse than a visible failure. - # Fetch and stage replacements before activation so failures can restore the - # previously-attached policy versions. + # Role creation treats instrumentation as best-effort; the add-on fails when its only task + # cannot complete. Replacement stages policy versions before activation so it can roll back. if not resource_types: # Only clean up if the previous Update had instrumentation enabled — avoids running # delete calls on stacks that never opted in to instrumentation in the first place. @@ -485,7 +469,6 @@ def attach_instrumentation_permissions(iam_client, role_name, account_id, partit f"Failed to update instrumentation permissions for {resource_types}: {e}. " "Leaving any previously-attached instrumentation policies in place." ) - return def handle_delete(event, context): @@ -493,7 +476,7 @@ def handle_delete(event, context): role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' + manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) iam_client = boto3.client('iam') try: if manage_base_permissions: @@ -510,9 +493,9 @@ def handle_create_update(event, context): role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' - fail_on_instrumentation_error = str(props.get('FailOnInstrumentationError', 'false')).lower() == 'true' - should_install_security_audit_policy = str(props['ResourceCollectionPermissions']).lower() == 'true' + manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) + fail_on_instrumentation_error = _is_true(props.get('FailOnInstrumentationError', 'false')) + should_install_security_audit_policy = _is_true(props['ResourceCollectionPermissions']) datadog_site = props.get('DatadogSite') or 'datadoghq.com' instrumentation_resource_types = parse_resource_types(props.get('InstrumentationResourceTypes')) previous_instrumentation_resource_types = parse_resource_types( diff --git a/aws_quickstart/attach_integration_permissions_test.py b/aws_quickstart/attach_integration_permissions_test.py index 659103b1..fff6e474 100644 --- a/aws_quickstart/attach_integration_permissions_test.py +++ b/aws_quickstart/attach_integration_permissions_test.py @@ -139,29 +139,31 @@ def setUp(self): self.partition = "aws" self.site = "datadoghq.com" - def _attach(self, resource_types, previous_resource_types=()): + def _attach(self, resource_types, previous_resource_types=(), fail_on_error=False): attach_instrumentation_permissions( - self.iam, self.role_name, self.account_id, self.partition, self.site, - resource_types, previous_resource_types, + self.iam, + self.role_name, + self.account_id, + self.partition, + self.site, + resource_types, + previous_resource_types, + fail_on_error=fail_on_error, ) - def _mock_chunks_response(self, chunks): - body = json.dumps({"data": {"attributes": {"permissions": chunks}}}).encode() - resp = Mock() - resp.read.return_value = body - return resp + def _mock_response(self, **attributes): + body = json.dumps({"data": {"attributes": attributes}}).encode() + response = Mock() + response.read.return_value = body + return response - def _mock_documents_response(self, documents): - body = json.dumps( - {"data": {"attributes": {"policy_documents": documents}}} - ).encode() - resp = Mock() - resp.read.return_value = body - return resp + def _created_policy_documents(self): + return [ + json.loads(request.kwargs["PolicyDocument"]) + for request in self.iam.create_policy.call_args_list + ] def test_empty_resource_types_no_op_when_previously_empty(self): - # Stack Create (or Update with no change) and no instrumentation requested: - # don't touch IAM at all — there's nothing to clean up. self._attach([], previous_resource_types=[]) self.iam.create_policy.assert_not_called() self.iam.attach_role_policy.assert_not_called() @@ -169,7 +171,6 @@ def test_empty_resource_types_no_op_when_previously_empty(self): self.iam.delete_policy.assert_not_called() def test_empty_resource_types_cleans_up_when_previously_set(self): - # Toggling instrumentation off on an Update should remove the previously-attached policies. self._attach([], previous_resource_types=["aws:ec2:instance"]) self.iam.create_policy.assert_not_called() self.iam.attach_role_policy.assert_not_called() @@ -199,19 +200,18 @@ def test_happy_path_attaches_each_policy_document(self, mock_urlopen): ], }, ] - mock_urlopen.return_value = self._mock_documents_response(documents) + mock_urlopen.return_value = self._mock_response(policy_documents=documents) self._attach(["aws:ec2:instance", "aws:eks:cluster"]) self.assertEqual(self.iam.create_policy.call_count, 2) self.assertEqual(self.iam.attach_role_policy.call_count, 2) - created_documents = [ - json.loads(c.kwargs["PolicyDocument"]) - for c in self.iam.create_policy.call_args_list - ] - self.assertEqual(created_documents, documents) + self.assertEqual(self._created_policy_documents(), documents) - names = [c.kwargs["PolicyName"] for c in self.iam.create_policy.call_args_list] + names = [ + request.kwargs["PolicyName"] + for request in self.iam.create_policy.call_args_list + ] self.assertEqual( names, [ @@ -228,26 +228,23 @@ def test_happy_path_attaches_each_policy_document(self, mock_urlopen): @patch("attach_integration_permissions.urllib.request.urlopen") def test_legacy_response_is_wrapped_as_broad_policy_documents(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response( - [["ec2:Describe*"], ["ssm:SendCommand"]] + mock_urlopen.return_value = self._mock_response( + permissions=[["ec2:Describe*"], ["ssm:SendCommand"]] ) self._attach(["aws:ec2:instance"]) - created_documents = [ - json.loads(c.kwargs["PolicyDocument"]) - for c in self.iam.create_policy.call_args_list - ] self.assertEqual( - [document["Statement"][0]["Action"] for document in created_documents], + [ + document["Statement"][0]["Action"] + for document in self._created_policy_documents() + ], [["ec2:Describe*"], ["ssm:SendCommand"]], ) @patch("attach_integration_permissions.urllib.request.urlopen") def test_fetch_failure_preserves_existing_policies(self, mock_urlopen): - # Regression: a transient API failure on Update must not silently revoke the - # previously-attached instrumentation policies. The function must neither - # call detach_role_policy / delete_policy nor raise. + # A transient update failure must not revoke previously attached policies. mock_urlopen.side_effect = HTTPError( "u", 500, "boom", {}, BytesIO(b'{"errors":["upstream down"]}') ) @@ -261,8 +258,8 @@ def test_fetch_failure_preserves_existing_policies(self, mock_urlopen): @patch("attach_integration_permissions.urllib.request.urlopen") def test_create_failure_rolls_back_without_attaching_partial_documents(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response( - [["chunk1:Action"], ["chunk2:Action"]] + mock_urlopen.return_value = self._mock_response( + permissions=[["chunk1:Action"], ["chunk2:Action"]] ) self.iam.create_policy.side_effect = [ {"Policy": {"Arn": "arn:aws:iam::123:policy/A"}}, @@ -277,32 +274,24 @@ def test_create_failure_rolls_back_without_attaching_partial_documents(self, moc @patch("attach_integration_permissions.urllib.request.urlopen") def test_fail_on_error_raises_on_fetch_failure(self, mock_urlopen): - # Add-on mode (fail_on_error=True): a fetch failure must propagate so the stack fails - # instead of silently reporting SUCCESS with nothing attached. mock_urlopen.side_effect = HTTPError( "u", 500, "boom", {}, BytesIO(b'{"errors":["upstream down"]}') ) with self.assertRaises(Exception): - attach_instrumentation_permissions( - self.iam, self.role_name, self.account_id, self.partition, self.site, - ["aws:ec2:instance"], (), fail_on_error=True, - ) + self._attach(["aws:ec2:instance"], fail_on_error=True) @patch("attach_integration_permissions.urllib.request.urlopen") def test_fail_on_error_raises_on_attach_failure(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response([["chunk1:Action"]]) + mock_urlopen.return_value = self._mock_response(permissions=[["chunk1:Action"]]) self.iam.attach_role_policy.side_effect = Exception("AccessDenied") with self.assertRaises(Exception): - attach_instrumentation_permissions( - self.iam, self.role_name, self.account_id, self.partition, self.site, - ["aws:ec2:instance"], (), fail_on_error=True, - ) + self._attach(["aws:ec2:instance"], fail_on_error=True) self.iam.delete_policy.assert_any_call(PolicyArn="arn:aws:iam::123:policy/X") @patch("attach_integration_permissions.urllib.request.urlopen") def test_quota_failure_happens_before_policy_mutation(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response( - [["chunk1:Action"], ["chunk2:Action"]] + mock_urlopen.return_value = self._mock_response( + permissions=[["chunk1:Action"], ["chunk2:Action"]] ) self.iam.list_attached_role_policies.return_value = { "AttachedPolicies": [ @@ -315,10 +304,7 @@ def test_quota_failure_happens_before_policy_mutation(self, mock_urlopen): } with self.assertRaisesRegex(RuntimeError, "room for only 1 instrumentation policies"): - attach_instrumentation_permissions( - self.iam, self.role_name, self.account_id, self.partition, self.site, - ["aws:ec2:instance"], (), fail_on_error=True, - ) + self._attach(["aws:ec2:instance"], fail_on_error=True) self.iam.create_policy.assert_not_called() self.iam.create_policy_version.assert_not_called() @@ -327,7 +313,7 @@ def test_quota_failure_happens_before_policy_mutation(self, mock_urlopen): @patch("attach_integration_permissions.urllib.request.urlopen") def test_existing_policy_is_staged_as_a_new_version(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response([["new:Action"]]) + mock_urlopen.return_value = self._mock_response(permissions=[["new:Action"]]) policy_arn = ( f"arn:aws:iam::{self.account_id}:policy/" f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1" @@ -357,8 +343,8 @@ def test_existing_policy_is_staged_as_a_new_version(self, mock_urlopen): @patch("attach_integration_permissions.urllib.request.urlopen") def test_attach_failure_restores_previous_policy_version(self, mock_urlopen): - mock_urlopen.return_value = self._mock_chunks_response( - [["updated:Action"], ["new:Action"]] + mock_urlopen.return_value = self._mock_response( + permissions=[["updated:Action"], ["new:Action"]] ) existing_arn = ( f"arn:aws:iam::{self.account_id}:policy/" @@ -420,15 +406,13 @@ def test_cleanup_instrumentation_targets_only_instrumentation_prefix(self): class TestCleanupLegacyBasePolicies(unittest.TestCase): - # Removing the old un-suffixed base policies before attaching the v2 ones is what keeps both - # generations from sitting attached at once during an in-place upgrade (IAM managed-policy limit). + # Removing legacy base policies first prevents both generations consuming the role quota. def setUp(self): self.iam = make_iam_mock() def test_only_targets_legacy_names_not_v2(self): cleanup_legacy_base_policies(self.iam, "MyRole", "123456789012", "aws", max_policies=3) for arn in detached_arns(self.iam): - # Legacy managed-policy ARNs must never carry the -v2 generation segment. self.assertNotIn("-permissions-v2-", arn) def test_cleans_legacy_resource_collection_and_standard(self): @@ -447,10 +431,7 @@ def test_does_not_touch_instrumentation(self): class TestManageBasePermissions(unittest.TestCase): - # ManageBasePermissions gates the standard + resource-collection policies. The role-creation - # path sets it true (manage everything); the post-setup add-on sets it false so it manages only - # the instrumentation policies and never touches the standard/resource-collection policies the - # role stack owns. + # The add-on disables base management so it cannot modify role-stack-owned policies. def setUp(self): self.iam = make_iam_mock(cleanup_side_effects=False) @@ -498,7 +479,6 @@ def test_create_manage_base_false_only_instrumentation( mock_standard.assert_not_called() mock_rc.assert_not_called() mock_instr.assert_called_once() - # Add-on mode must not touch the role stack's standard/resource-collection policies. mock_legacy.assert_not_called() @patch("attach_integration_permissions.boto3.client") @@ -542,7 +522,6 @@ def test_create_threads_fail_on_instrumentation_error(self, mock_instr, mock_cli def test_create_reports_failed_when_instrumentation_raises( self, mock_instr, mock_client, mock_cfn ): - # Add-on mode: a propagated instrumentation failure must surface as a FAILED response. mock_client.return_value = self.iam mock_instr.side_effect = Exception("AccessDenied") handle_create_update( @@ -552,14 +531,8 @@ def test_create_reports_failed_when_instrumentation_raises( class TestUpgradeSafePolicyNames(unittest.TestCase): - # Guards the invariant that makes the inline-trigger era safe: every policy name this template - # attaches must be disjoint from the un-suffixed names the legacy (<= v4.13) Delete handler removes, - # so the old handler can never wipe a policy this stack attached. This covers instrumentation too — - # the add-on attaches instrumentation policies against an existing role, and a later upgrade of that - # role's stack removes the old trigger, whose unconditional instrumentation cleanup would otherwise - # delete them. + # V2 names must be disjoint from names the <= v4.13 Delete handler removes. role = "DatadogIntegrationRole" - # Un-suffixed prefix the legacy trigger deletes instrumentation policies by. LEGACY_PREFIX_INSTRUMENTATION = "datadog-aws-integration-instrumentation-permissions" def _names(self, prefix): @@ -569,15 +542,17 @@ def test_standard_policy_name_differs_from_legacy(self): self.assertNotEqual(POLICY_NAME_STANDARD, LEGACY_POLICY_NAME_STANDARD) def test_resource_collection_names_disjoint_from_legacy(self): - self.assertEqual( - self._names(BASE_POLICY_PREFIX_RESOURCE_COLLECTION) & self._names(LEGACY_PREFIX_RESOURCE_COLLECTION), - set(), + self.assertTrue( + self._names(BASE_POLICY_PREFIX_RESOURCE_COLLECTION).isdisjoint( + self._names(LEGACY_PREFIX_RESOURCE_COLLECTION) + ) ) def test_instrumentation_names_disjoint_from_legacy(self): - self.assertEqual( - self._names(BASE_POLICY_PREFIX_INSTRUMENTATION) & self._names(self.LEGACY_PREFIX_INSTRUMENTATION), - set(), + self.assertTrue( + self._names(BASE_POLICY_PREFIX_INSTRUMENTATION).isdisjoint( + self._names(self.LEGACY_PREFIX_INSTRUMENTATION) + ) ) diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 2e01027c..13d6968b 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -116,30 +116,20 @@ Resources: LOGGER = logging.getLogger() LOGGER.setLevel(logging.INFO) API_CALL_SOURCE_HEADER_VALUE = "cfn-quickstart" - # The "-v2" suffix on these policy names is load-bearing, not cosmetic. The pre-extraction - # inline trigger (<= v4.13) deletes policies by their un-suffixed names on teardown, and that - # teardown runs whenever the old trigger is removed — i.e. when a role stack is upgraded off - # <= v4.13. Distinct v2 names ensure that destructive delete can never hit the policies this - # template attaches: - # - standard / resource-collection: an in-place role-stack upgrade removes the old trigger - # after this nested stack has re-attached them; v2 names keep them from being wiped. - # - instrumentation: the add-on attaches these against an existing role; if that role's stack - # is later upgraded off <= v4.13, the old trigger's unconditional instrumentation cleanup - # would wipe them unless they sit under a name it doesn't know. + # The <= v4.13 inline trigger deletes un-suffixed policies when removed during an upgrade. + # Keep replacement policies on v2 names so that teardown cannot delete newly attached policies. POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicyV2" BASE_POLICY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions-v2" BASE_POLICY_PREFIX_INSTRUMENTATION = "datadog-aws-integration-instrumentation-permissions-v2" - # Un-suffixed standard/resource-collection names created by the pre-extraction inline trigger - # (<= v4.13). The role-creation path cleans these up before attaching the v2 policies so the two - # generations never sit attached at once (IAM caps managed policies per role, default 10); the - # old trigger's own Delete handler then no-ops against names that are already gone. Legacy - # instrumentation policies need no such cleanup — that feature is unreleased, so none exist. + # Remove legacy base policies before attaching v2 policies to avoid exceeding the role quota. + # Instrumentation was not released under legacy names, so it needs no equivalent cleanup list. LEGACY_POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicy" LEGACY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions" STANDARD_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/standard" RESOURCE_COLLECTION_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/resource_collection?chunked=true" INSTRUMENTATION_PERMISSIONS_API_PATH = "/api/unstable/instrumenter/aws/iam_permissions" MANAGED_POLICY_CHAR_LIMIT = 6144 + POLICY_DOCUMENT_VERSION = "2012-10-17" class DatadogAPIError(Exception): @@ -167,6 +157,13 @@ Resources: return fetch_permissions_attributes_from_datadog(api_url)["permissions"] + def _broad_policy_document(actions): + return { + "Version": POLICY_DOCUMENT_VERSION, + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], + } + + def fetch_instrumentation_policy_documents(api_url): attributes = fetch_permissions_attributes_from_datadog(api_url) policy_documents = attributes.get("policy_documents") @@ -183,13 +180,7 @@ Resources: if isinstance(permission_chunks[0], str): permission_chunks = [permission_chunks] LOGGER.warning("Datadog API does not expose policy_documents yet; using legacy broad permissions") - return [ - { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": chunk, "Resource": "*"}], - } - for chunk in permission_chunks - ] + return [_broad_policy_document(chunk) for chunk in permission_chunks] def parse_resource_types(raw): @@ -201,6 +192,10 @@ Resources: return [t.strip() for t in items if t and t.strip()] + def _is_true(value): + return str(value).lower() == "true" + + def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( [("resource_type", t) for t in resource_types] @@ -273,21 +268,17 @@ Resources: def attach_standard_permissions(iam_client, role_name): permissions = fetch_permissions_from_datadog(STANDARD_PERMISSIONS_API_URL) - policy_document = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": permissions, "Resource": "*"}], - } iam_client.put_role_policy( RoleName=role_name, PolicyName=POLICY_NAME_STANDARD, - PolicyDocument=json.dumps(policy_document, separators=(',', ':')), + PolicyDocument=json.dumps(_broad_policy_document(permissions), separators=(',', ':')), ) def _serialize_policy_document(policy_document): if not isinstance(policy_document, dict): raise DatadogAPIError("instrumentation policy document must be an object") - if policy_document.get("Version") != "2012-10-17": + if policy_document.get("Version") != POLICY_DOCUMENT_VERSION: raise DatadogAPIError("instrumentation policy document has an unsupported version") if not isinstance(policy_document.get("Statement"), list) or not policy_document["Statement"]: raise DatadogAPIError("instrumentation policy document must contain statements") @@ -308,11 +299,7 @@ Resources: def _create_and_attach_policy(iam_client, role_name, policy_name, actions): - policy_document = { - "Version": "2012-10-17", - "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], - } - policy_arn = _create_policy(iam_client, policy_name, policy_document) + policy_arn = _create_policy(iam_client, policy_name, _broad_policy_document(actions)) iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) @@ -375,9 +362,10 @@ Resources: if not policy_name.startswith(prefix): return None suffix = policy_name[len(prefix):] - if not suffix.isdigit() or int(suffix) < 1: + if not suffix.isdigit(): return None - return int(suffix) + index = int(suffix) + return index if index >= 1 else None def _stage_policy_version(iam_client, policy_arn, policy_document, version_quota): @@ -555,12 +543,8 @@ Resources: def attach_instrumentation_permissions(iam_client, role_name, account_id, partition, datadog_site, resource_types, previous_resource_types, fail_on_error=False): - # Best-effort by default: instrumentation permissions are additive convenience on top of the - # integration, so any failure is logged and swallowed rather than blocking install. The - # post-setup add-on passes fail_on_error=True because attaching these policies is the stack's - # whole purpose, so a silent SUCCESS that attached nothing would be worse than a visible failure. - # Fetch and stage replacements before activation so failures can restore the - # previously-attached policy versions. + # Role creation treats instrumentation as best-effort; the add-on fails when its only task + # cannot complete. Replacement stages policy versions before activation so it can roll back. if not resource_types: # Only clean up if the previous Update had instrumentation enabled — avoids running # delete calls on stacks that never opted in to instrumentation in the first place. @@ -591,7 +575,6 @@ Resources: f"Failed to update instrumentation permissions for {resource_types}: {e}. " "Leaving any previously-attached instrumentation policies in place." ) - return def handle_delete(event, context): @@ -599,7 +582,7 @@ Resources: role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' + manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) iam_client = boto3.client('iam') try: if manage_base_permissions: @@ -616,9 +599,9 @@ Resources: role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' - fail_on_instrumentation_error = str(props.get('FailOnInstrumentationError', 'false')).lower() == 'true' - should_install_security_audit_policy = str(props['ResourceCollectionPermissions']).lower() == 'true' + manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) + fail_on_instrumentation_error = _is_true(props.get('FailOnInstrumentationError', 'false')) + should_install_security_audit_policy = _is_true(props['ResourceCollectionPermissions']) datadog_site = props.get('DatadogSite') or 'datadoghq.com' instrumentation_resource_types = parse_resource_types(props.get('InstrumentationResourceTypes')) previous_instrumentation_resource_types = parse_resource_types( From 40db3d0beb68ae5a284794aa716f3e99a0163a7c Mon Sep 17 00:00:00 2001 From: Fanny Jiang Date: Tue, 14 Jul 2026 13:49:03 -0400 Subject: [PATCH 4/4] Reduce integration policy attachment logic --- .../attach_integration_permissions.py | 358 ++++------------- .../attach_integration_permissions_test.py | 203 ++++------ .../datadog_integration_permissions.yaml | 364 ++++-------------- 3 files changed, 251 insertions(+), 674 deletions(-) diff --git a/aws_quickstart/attach_integration_permissions.py b/aws_quickstart/attach_integration_permissions.py index 4ef19ff5..94e0514a 100644 --- a/aws_quickstart/attach_integration_permissions.py +++ b/aws_quickstart/attach_integration_permissions.py @@ -10,20 +10,30 @@ LOGGER = logging.getLogger() LOGGER.setLevel(logging.INFO) API_CALL_SOURCE_HEADER_VALUE = "cfn-quickstart" -# The <= v4.13 inline trigger deletes un-suffixed policies when removed during an upgrade. -# Keep replacement policies on v2 names so that teardown cannot delete newly attached policies. +# The "-v2" suffix on these policy names is load-bearing, not cosmetic. The pre-extraction +# inline trigger (<= v4.13) deletes policies by their un-suffixed names on teardown, and that +# teardown runs whenever the old trigger is removed — i.e. when a role stack is upgraded off +# <= v4.13. Distinct v2 names ensure that destructive delete can never hit the policies this +# template attaches: +# - standard / resource-collection: an in-place role-stack upgrade removes the old trigger +# after this nested stack has re-attached them; v2 names keep them from being wiped. +# - instrumentation: the add-on attaches these against an existing role; if that role's stack +# is later upgraded off <= v4.13, the old trigger's unconditional instrumentation cleanup +# would wipe them unless they sit under a name it doesn't know. POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicyV2" BASE_POLICY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions-v2" BASE_POLICY_PREFIX_INSTRUMENTATION = "datadog-aws-integration-instrumentation-permissions-v2" -# Remove legacy base policies before attaching v2 policies to avoid exceeding the role quota. -# Instrumentation was not released under legacy names, so it needs no equivalent cleanup list. +# Un-suffixed standard/resource-collection names created by the pre-extraction inline trigger +# (<= v4.13). The role-creation path cleans these up before attaching the v2 policies so the two +# generations never sit attached at once (IAM caps managed policies per role, default 10); the +# old trigger's own Delete handler then no-ops against names that are already gone. Legacy +# instrumentation policies need no such cleanup — that feature is unreleased, so none exist. LEGACY_POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicy" LEGACY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions" STANDARD_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/standard" RESOURCE_COLLECTION_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/resource_collection?chunked=true" INSTRUMENTATION_PERMISSIONS_API_PATH = "/api/unstable/instrumenter/aws/iam_permissions" -MANAGED_POLICY_CHAR_LIMIT = 6144 -POLICY_DOCUMENT_VERSION = "2012-10-17" +MAX_POLICY_DOCUMENTS = 10 class DatadogAPIError(Exception): @@ -51,30 +61,16 @@ def fetch_permissions_from_datadog(api_url): return fetch_permissions_attributes_from_datadog(api_url)["permissions"] -def _broad_policy_document(actions): - return { - "Version": POLICY_DOCUMENT_VERSION, - "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], - } - - def fetch_instrumentation_policy_documents(api_url): - attributes = fetch_permissions_attributes_from_datadog(api_url) - policy_documents = attributes.get("policy_documents") - if policy_documents is not None: - if not policy_documents: - raise DatadogAPIError("Datadog API returned no instrumentation policy documents") - for document in policy_documents: - _serialize_policy_document(document) - return policy_documents - - permission_chunks = attributes.get("permissions") - if not permission_chunks: - raise DatadogAPIError("Datadog API returned neither policy_documents nor permissions") - if isinstance(permission_chunks[0], str): - permission_chunks = [permission_chunks] - LOGGER.warning("Datadog API does not expose policy_documents yet; using legacy broad permissions") - return [_broad_policy_document(chunk) for chunk in permission_chunks] + policy_documents = fetch_permissions_attributes_from_datadog(api_url)["policy_documents"] + if not policy_documents: + raise DatadogAPIError("Datadog API returned no instrumentation policy documents") + if len(policy_documents) > MAX_POLICY_DOCUMENTS: + raise DatadogAPIError( + f"Datadog API returned {len(policy_documents)} instrumentation policy documents; " + f"at most {MAX_POLICY_DOCUMENTS} are supported" + ) + return policy_documents def parse_resource_types(raw): @@ -86,10 +82,6 @@ def parse_resource_types(raw): return [t.strip() for t in items if t and t.strip()] -def _is_true(value): - return str(value).lower() == "true" - - def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( [("resource_type", t) for t in resource_types] @@ -105,14 +97,12 @@ def build_instrumentation_permissions_url(datadog_site, resource_types, account_ def _detach_and_delete_policy(iam_client, role_name, policy_arn, policy_name): # Detach + delete are both no-ops if the entity is already gone, so callers can blindly # iterate the policy-name space without first checking what actually exists. - errors = [] try: iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy_arn) except iam_client.exceptions.NoSuchEntityException: pass except Exception as e: LOGGER.error(f"Error detaching policy {policy_name}: {str(e)}") - errors.append(f"detach {policy_name}: {e}") try: iam_client.delete_policy(PolicyArn=policy_arn) @@ -120,11 +110,8 @@ def _detach_and_delete_policy(iam_client, role_name, policy_arn, policy_name): pass except iam_client.exceptions.DeleteConflictException: LOGGER.warning(f"Policy {policy_name} still attached, skipping delete") - errors.append(f"delete {policy_name}: policy is still attached") except Exception as e: LOGGER.error(f"Error deleting policy {policy_name}: {str(e)}") - errors.append(f"delete {policy_name}: {e}") - return errors def _cleanup_chunked_policies(iam_client, role_name, account_id, partition, prefix, max_policies=10): @@ -162,39 +149,35 @@ def cleanup_legacy_base_policies(iam_client, role_name, account_id, partition, m def attach_standard_permissions(iam_client, role_name): permissions = fetch_permissions_from_datadog(STANDARD_PERMISSIONS_API_URL) + policy_document = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": permissions, "Resource": "*"}], + } iam_client.put_role_policy( RoleName=role_name, PolicyName=POLICY_NAME_STANDARD, - PolicyDocument=json.dumps(_broad_policy_document(permissions), separators=(',', ':')), + PolicyDocument=json.dumps(policy_document, separators=(',', ':')), ) -def _serialize_policy_document(policy_document): - if not isinstance(policy_document, dict): - raise DatadogAPIError("instrumentation policy document must be an object") - if policy_document.get("Version") != POLICY_DOCUMENT_VERSION: - raise DatadogAPIError("instrumentation policy document has an unsupported version") - if not isinstance(policy_document.get("Statement"), list) or not policy_document["Statement"]: - raise DatadogAPIError("instrumentation policy document must contain statements") - - policy_json = json.dumps(policy_document, separators=(',', ':')) - if len(policy_json) > MANAGED_POLICY_CHAR_LIMIT: - raise DatadogAPIError( - f"instrumentation policy document exceeds {MANAGED_POLICY_CHAR_LIMIT} characters" - ) - return policy_json +def _create_and_attach_policy(iam_client, role_name, policy_name, actions): + policy_json = json.dumps( + { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], + }, + separators=(',', ':'), + ) + LOGGER.info(f"Creating policy {policy_name} with {len(actions)} permissions ({len(policy_json)} characters)") + policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=policy_json) + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy['Policy']['Arn']) -def _create_policy(iam_client, policy_name, policy_document): - policy_json = _serialize_policy_document(policy_document) +def _create_and_attach_policy_document(iam_client, role_name, policy_name, policy_document): + policy_json = json.dumps(policy_document, separators=(',', ':')) LOGGER.info(f"Creating policy {policy_name} ({len(policy_json)} characters)") policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=policy_json) - return policy['Policy']['Arn'] - - -def _create_and_attach_policy(iam_client, role_name, policy_name, actions): - policy_arn = _create_policy(iam_client, policy_name, _broad_policy_document(actions)) - iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy['Policy']['Arn']) def attach_resource_collection_permissions(iam_client, role_name): @@ -222,226 +205,37 @@ def _list_attached_role_policies(iam_client, role_name): marker = response["Marker"] -def _list_policy_versions(iam_client, policy_arn): - versions = [] - marker = None - while True: - request = {"PolicyArn": policy_arn} - if marker: - request["Marker"] = marker - response = iam_client.list_policy_versions(**request) - versions.extend(response.get("Versions", [])) - if not response.get("IsTruncated"): - return versions - marker = response["Marker"] - - -def _policy_quotas(iam_client): - summary = iam_client.get_account_summary().get("SummaryMap", {}) - attachment_quota = summary.get("AttachedPoliciesPerRoleQuota") - version_quota = summary.get("VersionsPerPolicyQuota") - if not isinstance(attachment_quota, int) or attachment_quota < 1: - raise RuntimeError("IAM account summary did not include AttachedPoliciesPerRoleQuota") - if not isinstance(version_quota, int) or version_quota < 2: - raise RuntimeError("IAM account summary did not include VersionsPerPolicyQuota") - return attachment_quota, version_quota - - -def _instrumentation_policy_name(role_name, index): - return f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{index}" - - -def _instrumentation_policy_index(policy_name, role_name): +def _is_instrumentation_policy(policy_name, role_name): prefix = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-" if not policy_name.startswith(prefix): - return None + return False suffix = policy_name[len(prefix):] - if not suffix.isdigit(): - return None - index = int(suffix) - return index if index >= 1 else None + return suffix.isdigit() and 1 <= int(suffix) <= MAX_POLICY_DOCUMENTS -def _stage_policy_version(iam_client, policy_arn, policy_document, version_quota): - versions = _list_policy_versions(iam_client, policy_arn) - previous_default = next( - (version["VersionId"] for version in versions if version.get("IsDefaultVersion")), - None, +def _validate_instrumentation_policy_capacity(iam_client, role_name, policy_document_count): + attached_policies = _list_attached_role_policies(iam_client, role_name) + existing_instrumentation_count = sum( + _is_instrumentation_policy(policy["PolicyName"], role_name) + for policy in attached_policies ) - if previous_default is None: - raise RuntimeError(f"policy {policy_arn} has no default version") - - non_default_versions = sorted( - (version for version in versions if not version.get("IsDefaultVersion")), - key=lambda version: int(version["VersionId"].removeprefix("v")), + non_instrumentation_count = len(attached_policies) - existing_instrumentation_count + attachment_quota = iam_client.get_account_summary().get("SummaryMap", {}).get( + "AttachedPoliciesPerRoleQuota" ) - while len(versions) >= version_quota: - if not non_default_versions: - raise RuntimeError(f"policy {policy_arn} has no removable non-default version") - version = non_default_versions.pop(0) - iam_client.delete_policy_version( - PolicyArn=policy_arn, - VersionId=version["VersionId"], - ) - versions.remove(version) - - response = iam_client.create_policy_version( - PolicyArn=policy_arn, - PolicyDocument=_serialize_policy_document(policy_document), - SetAsDefault=False, - ) - return { - "PolicyArn": policy_arn, - "PreviousVersionId": previous_default, - "StagedVersionId": response["PolicyVersion"]["VersionId"], - } - - -def _discard_staged_replacement(iam_client, role_name, staged_versions, created_policies): - cleanup_errors = [] - for policy in reversed(created_policies): - if policy.get("AttachAttempted"): - try: - iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) - except iam_client.exceptions.NoSuchEntityException: - pass - except Exception as e: - cleanup_errors.append(f"detach {policy['PolicyArn']}: {e}") - try: - iam_client.delete_policy(PolicyArn=policy["PolicyArn"]) - except Exception as e: - cleanup_errors.append(f"delete {policy['PolicyArn']}: {e}") - - for version in reversed(staged_versions): - if version.get("Activated"): - try: - iam_client.set_default_policy_version( - PolicyArn=version["PolicyArn"], - VersionId=version["PreviousVersionId"], - ) - except Exception as e: - cleanup_errors.append(f"restore {version['PolicyArn']}: {e}") - continue - try: - iam_client.delete_policy_version( - PolicyArn=version["PolicyArn"], - VersionId=version["StagedVersionId"], - ) - except Exception as e: - cleanup_errors.append(f"delete staged version for {version['PolicyArn']}: {e}") - return cleanup_errors - - -def replace_instrumentation_policies(iam_client, role_name, account_id, partition, policy_documents): - attached_policies = _list_attached_role_policies(iam_client, role_name) - attachment_quota, version_quota = _policy_quotas(iam_client) - existing_policies = {} - for policy in attached_policies: - index = _instrumentation_policy_index(policy["PolicyName"], role_name) - if index is not None: - existing_policies[index] = policy - - non_instrumentation_count = len(attached_policies) - len(existing_policies) - final_attachment_count = non_instrumentation_count + len(policy_documents) - if final_attachment_count > attachment_quota: + if not isinstance(attachment_quota, int) or attachment_quota < 1: + raise RuntimeError("IAM account summary did not include AttachedPoliciesPerRoleQuota") + if non_instrumentation_count + policy_document_count > attachment_quota: raise RuntimeError( f"role {role_name} has room for only " f"{attachment_quota - non_instrumentation_count} instrumentation policies, " - f"but Datadog returned {len(policy_documents)}" - ) - - LOGGER.info( - f"Replacing {len(existing_policies)} instrumentation policies with " - f"{len(policy_documents)} documents; final managed-policy usage " - f"{final_attachment_count}/{attachment_quota}" - ) - ordered_existing_policies = [ - policy for _, policy in sorted(existing_policies.items()) - ] - used_policy_indices = set(existing_policies) - next_policy_index = 1 - reused_policy_arns = set() - staged_versions = [] - created_policies = [] - try: - for document_index, policy_document in enumerate(policy_documents): - existing = ( - ordered_existing_policies[document_index] - if document_index < len(ordered_existing_policies) - else None - ) - if existing: - reused_policy_arns.add(existing["PolicyArn"]) - staged_versions.append( - _stage_policy_version( - iam_client, - existing["PolicyArn"], - policy_document, - version_quota, - ) - ) - continue - - while next_policy_index in used_policy_indices: - next_policy_index += 1 - policy_name = _instrumentation_policy_name(role_name, next_policy_index) - used_policy_indices.add(next_policy_index) - policy_arn = f"arn:{partition}:iam::{account_id}:policy/{policy_name}" - try: - iam_client.delete_policy(PolicyArn=policy_arn) - except iam_client.exceptions.NoSuchEntityException: - pass - created_policies.append({ - "PolicyArn": _create_policy(iam_client, policy_name, policy_document), - "AttachAttempted": False, - }) - - for version in staged_versions: - iam_client.set_default_policy_version( - PolicyArn=version["PolicyArn"], - VersionId=version["StagedVersionId"], - ) - version["Activated"] = True - for policy in created_policies: - policy["AttachAttempted"] = True - iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) - except Exception as e: - cleanup_errors = _discard_staged_replacement( - iam_client, - role_name, - staged_versions, - created_policies, - ) - if cleanup_errors: - raise RuntimeError( - f"failed to replace instrumentation policies: {e}; " - f"rollback errors: {'; '.join(cleanup_errors)}" - ) from e - raise - - cleanup_errors = [] - for policy in existing_policies.values(): - if policy["PolicyArn"] in reused_policy_arns: - continue - cleanup_errors.extend(_detach_and_delete_policy( - iam_client, - role_name, - policy["PolicyArn"], - policy["PolicyName"], - )) - if cleanup_errors: - raise RuntimeError( - f"new instrumentation policies are active, but old policy cleanup failed: " - f"{'; '.join(cleanup_errors)}" + f"but Datadog returned {policy_document_count}" ) def attach_instrumentation_permissions(iam_client, role_name, account_id, partition, datadog_site, resource_types, previous_resource_types, fail_on_error=False): - # Role creation treats instrumentation as best-effort; the add-on fails when its only task - # cannot complete. Replacement stages policy versions before activation so it can roll back. + # Fetch and validate capacity before cleanup so a transient failure leaves existing policies. if not resource_types: - # Only clean up if the previous Update had instrumentation enabled — avoids running - # delete calls on stacks that never opted in to instrumentation in the first place. if previous_resource_types: cleanup_instrumentation_policies(iam_client, role_name, account_id, partition) return @@ -455,20 +249,34 @@ def attach_instrumentation_permissions(iam_client, role_name, account_id, partit ) LOGGER.info(f"Fetching instrumentation permissions for {resource_types} from {url}") policy_documents = fetch_instrumentation_policy_documents(url) - replace_instrumentation_policies( + _validate_instrumentation_policy_capacity( iam_client, role_name, - account_id, - partition, - policy_documents, + len(policy_documents), ) except Exception as e: if fail_on_error: raise LOGGER.warning( - f"Failed to update instrumentation permissions for {resource_types}: {e}. " + f"Failed to prepare instrumentation permissions for {resource_types}: {e}. " "Leaving any previously-attached instrumentation policies in place." ) + return + + cleanup_instrumentation_policies(iam_client, role_name, account_id, partition) + for i, policy_document in enumerate(policy_documents): + policy_name = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{i+1}" + try: + _create_and_attach_policy_document( + iam_client, + role_name, + policy_name, + policy_document, + ) + except Exception as e: + if fail_on_error: + raise + LOGGER.warning(f"Failed to create/attach instrumentation policy {policy_name}: {e}. Continuing.") def handle_delete(event, context): @@ -476,7 +284,7 @@ def handle_delete(event, context): role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) + manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' iam_client = boto3.client('iam') try: if manage_base_permissions: @@ -493,9 +301,9 @@ def handle_create_update(event, context): role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) - fail_on_instrumentation_error = _is_true(props.get('FailOnInstrumentationError', 'false')) - should_install_security_audit_policy = _is_true(props['ResourceCollectionPermissions']) + manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' + fail_on_instrumentation_error = str(props.get('FailOnInstrumentationError', 'false')).lower() == 'true' + should_install_security_audit_policy = str(props['ResourceCollectionPermissions']).lower() == 'true' datadog_site = props.get('DatadogSite') or 'datadoghq.com' instrumentation_resource_types = parse_resource_types(props.get('InstrumentationResourceTypes')) previous_instrumentation_resource_types = parse_resource_types( diff --git a/aws_quickstart/attach_integration_permissions_test.py b/aws_quickstart/attach_integration_permissions_test.py index fff6e474..4624eeb2 100644 --- a/aws_quickstart/attach_integration_permissions_test.py +++ b/aws_quickstart/attach_integration_permissions_test.py @@ -4,7 +4,7 @@ from pathlib import Path import sys import unittest -from unittest.mock import patch, Mock, MagicMock, call +from unittest.mock import patch, Mock, MagicMock from urllib.error import HTTPError from urllib.parse import urlparse, parse_qsl from io import BytesIO @@ -28,6 +28,7 @@ BASE_POLICY_PREFIX_RESOURCE_COLLECTION, LEGACY_POLICY_NAME_STANDARD, LEGACY_PREFIX_RESOURCE_COLLECTION, + MAX_POLICY_DOCUMENTS, ) @@ -47,6 +48,10 @@ def test_cloudformation_inline_lambda_matches_tested_source(self): ) self.assertEqual(embedded, source) + def test_custom_resource_has_policy_attachment_schema_version(self): + template_path = Path(__file__).with_name("datadog_integration_permissions.yaml") + self.assertIn(' PolicyAttachmentSchemaVersion: "2"', template_path.read_text()) + def make_iam_mock(cleanup_side_effects=True): iam = MagicMock() @@ -59,7 +64,7 @@ def make_iam_mock(cleanup_side_effects=True): def detached_arns(iam): - return [c.kwargs["PolicyArn"] for c in iam.detach_role_policy.call_args_list] + return [request.kwargs["PolicyArn"] for request in iam.detach_role_policy.call_args_list] class TestParseResourceTypes(unittest.TestCase): @@ -79,7 +84,6 @@ def test_multiple_with_whitespace(self): ) def test_list_input(self): - # CFN may forward a CommaDelimitedList as a JSON array self.assertEqual( parse_resource_types(["aws:ec2:instance", " aws:ecs:cluster "]), ["aws:ec2:instance", "aws:ecs:cluster"], @@ -105,7 +109,7 @@ def test_path_and_host(self): self.assertEqual(parsed.netloc, "api.datadoghq.eu") self.assertEqual(parsed.path, "/api/unstable/instrumenter/aws/iam_permissions") - def test_repeated_resource_type_and_chunked(self): + def test_query_parameters(self): url = build_instrumentation_permissions_url( "datadoghq.com", ["aws:ec2:instance", "aws:ecs:cluster", "aws:eks:cluster"], @@ -113,7 +117,7 @@ def test_repeated_resource_type_and_chunked(self): "aws-us-gov", ) pairs = self._query_pairs(url) - resource_types = [v for k, v in pairs if k == "resource_type"] + resource_types = [value for key, value in pairs if key == "resource_type"] self.assertEqual( resource_types, ["aws:ec2:instance", "aws:ecs:cluster", "aws:eks:cluster"], @@ -125,14 +129,11 @@ def test_repeated_resource_type_and_chunked(self): class TestAttachInstrumentationPermissions(unittest.TestCase): def setUp(self): - self.iam = make_iam_mock(cleanup_side_effects=False) + self.iam = make_iam_mock() self.iam.create_policy.return_value = {"Policy": {"Arn": "arn:aws:iam::123:policy/X"}} self.iam.list_attached_role_policies.return_value = {"AttachedPolicies": []} self.iam.get_account_summary.return_value = { - "SummaryMap": { - "AttachedPoliciesPerRoleQuota": 10, - "VersionsPerPolicyQuota": 5, - } + "SummaryMap": {"AttachedPoliciesPerRoleQuota": 10} } self.role_name = "DatadogIntegrationRole" self.account_id = "123456789012" @@ -207,13 +208,11 @@ def test_happy_path_attaches_each_policy_document(self, mock_urlopen): self.assertEqual(self.iam.create_policy.call_count, 2) self.assertEqual(self.iam.attach_role_policy.call_count, 2) self.assertEqual(self._created_policy_documents(), documents) - - names = [ - request.kwargs["PolicyName"] - for request in self.iam.create_policy.call_args_list - ] self.assertEqual( - names, + [ + request.kwargs["PolicyName"] + for request in self.iam.create_policy.call_args_list + ], [ f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1", f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-2", @@ -227,50 +226,59 @@ def test_happy_path_attaches_each_policy_document(self, mock_urlopen): self.assertIn(("partition", self.partition), query) @patch("attach_integration_permissions.urllib.request.urlopen") - def test_legacy_response_is_wrapped_as_broad_policy_documents(self, mock_urlopen): - mock_urlopen.return_value = self._mock_response( - permissions=[["ec2:Describe*"], ["ssm:SendCommand"]] + def test_fetch_failure_preserves_existing_policies(self, mock_urlopen): + mock_urlopen.side_effect = HTTPError( + "u", 500, "boom", {}, BytesIO(b'{"errors":["upstream down"]}') ) self._attach(["aws:ec2:instance"]) - self.assertEqual( - [ - document["Statement"][0]["Action"] - for document in self._created_policy_documents() - ], - [["ec2:Describe*"], ["ssm:SendCommand"]], - ) + self.iam.create_policy.assert_not_called() + self.iam.attach_role_policy.assert_not_called() + self.iam.detach_role_policy.assert_not_called() + self.iam.delete_policy.assert_not_called() @patch("attach_integration_permissions.urllib.request.urlopen") - def test_fetch_failure_preserves_existing_policies(self, mock_urlopen): - # A transient update failure must not revoke previously attached policies. - mock_urlopen.side_effect = HTTPError( - "u", 500, "boom", {}, BytesIO(b'{"errors":["upstream down"]}') + def test_missing_policy_documents_preserves_existing_policies(self, mock_urlopen): + mock_urlopen.return_value = self._mock_response( + permissions=[["ec2:DescribeInstances"]] ) self._attach(["aws:ec2:instance"]) self.iam.create_policy.assert_not_called() - self.iam.attach_role_policy.assert_not_called() self.iam.detach_role_policy.assert_not_called() - self.iam.delete_policy.assert_not_called() @patch("attach_integration_permissions.urllib.request.urlopen") - def test_create_failure_rolls_back_without_attaching_partial_documents(self, mock_urlopen): + def test_rejects_more_than_ten_policy_documents(self, mock_urlopen): + document = {"Version": "2012-10-17", "Statement": []} mock_urlopen.return_value = self._mock_response( - permissions=[["chunk1:Action"], ["chunk2:Action"]] + policy_documents=[document] * (MAX_POLICY_DOCUMENTS + 1) ) + + with self.assertRaisesRegex(Exception, "at most 10"): + self._attach(["aws:ec2:instance"], fail_on_error=True) + + self.iam.create_policy.assert_not_called() + self.iam.detach_role_policy.assert_not_called() + + @patch("attach_integration_permissions.urllib.request.urlopen") + def test_per_document_failure_is_swallowed_and_others_continue(self, mock_urlopen): + documents = [ + {"Version": "2012-10-17", "Statement": [{"Action": [f"chunk{i}:Action"]}]} + for i in range(3) + ] + mock_urlopen.return_value = self._mock_response(policy_documents=documents) self.iam.create_policy.side_effect = [ {"Policy": {"Arn": "arn:aws:iam::123:policy/A"}}, Exception("EntityAlreadyExists"), + {"Policy": {"Arn": "arn:aws:iam::123:policy/C"}}, ] self._attach(["aws:ec2:instance"]) - self.assertEqual(self.iam.create_policy.call_count, 2) - self.iam.attach_role_policy.assert_not_called() - self.iam.delete_policy.assert_any_call(PolicyArn="arn:aws:iam::123:policy/A") + self.assertEqual(self.iam.create_policy.call_count, 3) + self.assertEqual(self.iam.attach_role_policy.call_count, 2) @patch("attach_integration_permissions.urllib.request.urlopen") def test_fail_on_error_raises_on_fetch_failure(self, mock_urlopen): @@ -282,17 +290,20 @@ def test_fail_on_error_raises_on_fetch_failure(self, mock_urlopen): @patch("attach_integration_permissions.urllib.request.urlopen") def test_fail_on_error_raises_on_attach_failure(self, mock_urlopen): - mock_urlopen.return_value = self._mock_response(permissions=[["chunk1:Action"]]) - self.iam.attach_role_policy.side_effect = Exception("AccessDenied") + mock_urlopen.return_value = self._mock_response( + policy_documents=[{"Version": "2012-10-17", "Statement": []}] + ) + self.iam.create_policy.side_effect = Exception("AccessDenied") with self.assertRaises(Exception): self._attach(["aws:ec2:instance"], fail_on_error=True) - self.iam.delete_policy.assert_any_call(PolicyArn="arn:aws:iam::123:policy/X") @patch("attach_integration_permissions.urllib.request.urlopen") def test_quota_failure_happens_before_policy_mutation(self, mock_urlopen): - mock_urlopen.return_value = self._mock_response( - permissions=[["chunk1:Action"], ["chunk2:Action"]] - ) + documents = [ + {"Version": "2012-10-17", "Statement": [{"Action": [f"chunk{i}:Action"]}]} + for i in range(2) + ] + mock_urlopen.return_value = self._mock_response(policy_documents=documents) self.iam.list_attached_role_policies.return_value = { "AttachedPolicies": [ { @@ -307,83 +318,37 @@ def test_quota_failure_happens_before_policy_mutation(self, mock_urlopen): self._attach(["aws:ec2:instance"], fail_on_error=True) self.iam.create_policy.assert_not_called() - self.iam.create_policy_version.assert_not_called() self.iam.attach_role_policy.assert_not_called() self.iam.detach_role_policy.assert_not_called() @patch("attach_integration_permissions.urllib.request.urlopen") - def test_existing_policy_is_staged_as_a_new_version(self, mock_urlopen): - mock_urlopen.return_value = self._mock_response(permissions=[["new:Action"]]) - policy_arn = ( - f"arn:aws:iam::{self.account_id}:policy/" - f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1" - ) - self.iam.list_attached_role_policies.return_value = { - "AttachedPolicies": [{ - "PolicyName": f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1", - "PolicyArn": policy_arn, - }] - } - self.iam.list_policy_versions.return_value = { - "Versions": [{"VersionId": "v1", "IsDefaultVersion": True}] - } - self.iam.create_policy_version.return_value = { - "PolicyVersion": {"VersionId": "v2"} - } - - self._attach(["aws:ec2:instance"]) - - self.iam.create_policy.assert_not_called() - self.iam.create_policy_version.assert_called_once() - self.iam.set_default_policy_version.assert_called_once_with( - PolicyArn=policy_arn, - VersionId="v2", - ) - self.iam.detach_role_policy.assert_not_called() - - @patch("attach_integration_permissions.urllib.request.urlopen") - def test_attach_failure_restores_previous_policy_version(self, mock_urlopen): - mock_urlopen.return_value = self._mock_response( - permissions=[["updated:Action"], ["new:Action"]] - ) - existing_arn = ( - f"arn:aws:iam::{self.account_id}:policy/" - f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1" - ) + def test_quota_preflight_excludes_policies_that_cleanup_replaces(self, mock_urlopen): + documents = [ + {"Version": "2012-10-17", "Statement": [{"Action": [f"chunk{i}:Action"]}]} + for i in range(2) + ] + mock_urlopen.return_value = self._mock_response(policy_documents=documents) + existing_name = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1" self.iam.list_attached_role_policies.return_value = { - "AttachedPolicies": [{ - "PolicyName": f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{self.role_name}-1", - "PolicyArn": existing_arn, - }] - } - self.iam.list_policy_versions.return_value = { - "Versions": [{"VersionId": "v1", "IsDefaultVersion": True}] - } - self.iam.create_policy_version.return_value = { - "PolicyVersion": {"VersionId": "v2"} - } - self.iam.create_policy.return_value = { - "Policy": {"Arn": "arn:aws:iam::123:policy/New"} + "AttachedPolicies": [ + { + "PolicyName": f"CustomerPolicy{index}", + "PolicyArn": f"arn:aws:iam::123:policy/CustomerPolicy{index}", + } + for index in range(8) + ] + + [ + { + "PolicyName": existing_name, + "PolicyArn": f"arn:aws:iam::123:policy/{existing_name}", + } + ] } - self.iam.attach_role_policy.side_effect = Exception("LimitExceeded") - self._attach(["aws:ec2:instance"]) + self._attach(["aws:ec2:instance"], fail_on_error=True) - self.assertEqual( - self.iam.set_default_policy_version.call_args_list, - [ - call(PolicyArn=existing_arn, VersionId="v2"), - call(PolicyArn=existing_arn, VersionId="v1"), - ], - ) - self.iam.delete_policy_version.assert_called_with( - PolicyArn=existing_arn, - VersionId="v2", - ) - self.iam.detach_role_policy.assert_called_once_with( - RoleName=self.role_name, - PolicyArn="arn:aws:iam::123:policy/New", - ) + self.assertEqual(self.iam.create_policy.call_count, 2) + self.assertEqual(self.iam.attach_role_policy.call_count, 2) class TestCleanup(unittest.TestCase): @@ -406,7 +371,6 @@ def test_cleanup_instrumentation_targets_only_instrumentation_prefix(self): class TestCleanupLegacyBasePolicies(unittest.TestCase): - # Removing legacy base policies first prevents both generations consuming the role quota. def setUp(self): self.iam = make_iam_mock() @@ -418,20 +382,18 @@ def test_only_targets_legacy_names_not_v2(self): def test_cleans_legacy_resource_collection_and_standard(self): cleanup_legacy_base_policies(self.iam, "MyRole", "123456789012", "aws", max_policies=3) arns = detached_arns(self.iam) - self.assertTrue(any(LEGACY_PREFIX_RESOURCE_COLLECTION + "-MyRole" in a for a in arns)) + self.assertTrue(any(LEGACY_PREFIX_RESOURCE_COLLECTION + "-MyRole" in arn for arn in arns)) self.iam.delete_role_policy.assert_called_once_with( RoleName="MyRole", PolicyName=LEGACY_POLICY_NAME_STANDARD ) def test_does_not_touch_instrumentation(self): - # Base cleanup only handles standard/resource-collection; instrumentation is managed separately. cleanup_legacy_base_policies(self.iam, "MyRole", "123456789012", "aws", max_policies=3) arns = detached_arns(self.iam) - self.assertTrue(all("instrumentation" not in a for a in arns)) + self.assertTrue(all("instrumentation" not in arn for arn in arns)) class TestManageBasePermissions(unittest.TestCase): - # The add-on disables base management so it cannot modify role-stack-owned policies. def setUp(self): self.iam = make_iam_mock(cleanup_side_effects=False) @@ -512,7 +474,8 @@ def test_delete_manage_base_true_cleans_both( def test_create_threads_fail_on_instrumentation_error(self, mock_instr, mock_client): mock_client.return_value = self.iam handle_create_update( - self._props(ManageBasePermissions="false", FailOnInstrumentationError="true"), None + self._props(ManageBasePermissions="false", FailOnInstrumentationError="true"), + None, ) self.assertTrue(mock_instr.call_args.kwargs["fail_on_error"]) @@ -525,13 +488,13 @@ def test_create_reports_failed_when_instrumentation_raises( mock_client.return_value = self.iam mock_instr.side_effect = Exception("AccessDenied") handle_create_update( - self._props(ManageBasePermissions="false", FailOnInstrumentationError="true"), None + self._props(ManageBasePermissions="false", FailOnInstrumentationError="true"), + None, ) self.assertEqual(mock_cfn.send.call_args.args[2], mock_cfn.FAILED) class TestUpgradeSafePolicyNames(unittest.TestCase): - # V2 names must be disjoint from names the <= v4.13 Delete handler removes. role = "DatadogIntegrationRole" LEGACY_PREFIX_INSTRUMENTATION = "datadog-aws-integration-instrumentation-permissions" diff --git a/aws_quickstart/datadog_integration_permissions.yaml b/aws_quickstart/datadog_integration_permissions.yaml index 13d6968b..8ef97fba 100644 --- a/aws_quickstart/datadog_integration_permissions.yaml +++ b/aws_quickstart/datadog_integration_permissions.yaml @@ -71,16 +71,12 @@ Resources: - Effect: Allow Action: - iam:CreatePolicy - - iam:CreatePolicyVersion - iam:DeletePolicy - - iam:DeletePolicyVersion - iam:DeleteRolePolicy - iam:AttachRolePolicy - iam:DetachRolePolicy - iam:ListAttachedRolePolicies - - iam:ListPolicyVersions - iam:PutRolePolicy - - iam:SetDefaultPolicyVersion Resource: # Wildcards cover both the v2 names this template creates and the un-suffixed legacy # names it cleans up on an in-place upgrade. @@ -116,20 +112,30 @@ Resources: LOGGER = logging.getLogger() LOGGER.setLevel(logging.INFO) API_CALL_SOURCE_HEADER_VALUE = "cfn-quickstart" - # The <= v4.13 inline trigger deletes un-suffixed policies when removed during an upgrade. - # Keep replacement policies on v2 names so that teardown cannot delete newly attached policies. + # The "-v2" suffix on these policy names is load-bearing, not cosmetic. The pre-extraction + # inline trigger (<= v4.13) deletes policies by their un-suffixed names on teardown, and that + # teardown runs whenever the old trigger is removed — i.e. when a role stack is upgraded off + # <= v4.13. Distinct v2 names ensure that destructive delete can never hit the policies this + # template attaches: + # - standard / resource-collection: an in-place role-stack upgrade removes the old trigger + # after this nested stack has re-attached them; v2 names keep them from being wiped. + # - instrumentation: the add-on attaches these against an existing role; if that role's stack + # is later upgraded off <= v4.13, the old trigger's unconditional instrumentation cleanup + # would wipe them unless they sit under a name it doesn't know. POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicyV2" BASE_POLICY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions-v2" BASE_POLICY_PREFIX_INSTRUMENTATION = "datadog-aws-integration-instrumentation-permissions-v2" - # Remove legacy base policies before attaching v2 policies to avoid exceeding the role quota. - # Instrumentation was not released under legacy names, so it needs no equivalent cleanup list. + # Un-suffixed standard/resource-collection names created by the pre-extraction inline trigger + # (<= v4.13). The role-creation path cleans these up before attaching the v2 policies so the two + # generations never sit attached at once (IAM caps managed policies per role, default 10); the + # old trigger's own Delete handler then no-ops against names that are already gone. Legacy + # instrumentation policies need no such cleanup — that feature is unreleased, so none exist. LEGACY_POLICY_NAME_STANDARD = "DatadogAWSIntegrationPolicy" LEGACY_PREFIX_RESOURCE_COLLECTION = "datadog-aws-integration-resource-collection-permissions" STANDARD_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/standard" RESOURCE_COLLECTION_PERMISSIONS_API_URL = "https://api.datadoghq.com/api/v2/integration/aws/iam_permissions/resource_collection?chunked=true" INSTRUMENTATION_PERMISSIONS_API_PATH = "/api/unstable/instrumenter/aws/iam_permissions" - MANAGED_POLICY_CHAR_LIMIT = 6144 - POLICY_DOCUMENT_VERSION = "2012-10-17" + MAX_POLICY_DOCUMENTS = 10 class DatadogAPIError(Exception): @@ -157,30 +163,16 @@ Resources: return fetch_permissions_attributes_from_datadog(api_url)["permissions"] - def _broad_policy_document(actions): - return { - "Version": POLICY_DOCUMENT_VERSION, - "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], - } - - def fetch_instrumentation_policy_documents(api_url): - attributes = fetch_permissions_attributes_from_datadog(api_url) - policy_documents = attributes.get("policy_documents") - if policy_documents is not None: - if not policy_documents: - raise DatadogAPIError("Datadog API returned no instrumentation policy documents") - for document in policy_documents: - _serialize_policy_document(document) - return policy_documents - - permission_chunks = attributes.get("permissions") - if not permission_chunks: - raise DatadogAPIError("Datadog API returned neither policy_documents nor permissions") - if isinstance(permission_chunks[0], str): - permission_chunks = [permission_chunks] - LOGGER.warning("Datadog API does not expose policy_documents yet; using legacy broad permissions") - return [_broad_policy_document(chunk) for chunk in permission_chunks] + policy_documents = fetch_permissions_attributes_from_datadog(api_url)["policy_documents"] + if not policy_documents: + raise DatadogAPIError("Datadog API returned no instrumentation policy documents") + if len(policy_documents) > MAX_POLICY_DOCUMENTS: + raise DatadogAPIError( + f"Datadog API returned {len(policy_documents)} instrumentation policy documents; " + f"at most {MAX_POLICY_DOCUMENTS} are supported" + ) + return policy_documents def parse_resource_types(raw): @@ -192,10 +184,6 @@ Resources: return [t.strip() for t in items if t and t.strip()] - def _is_true(value): - return str(value).lower() == "true" - - def build_instrumentation_permissions_url(datadog_site, resource_types, account_id, partition): query = urllib.parse.urlencode( [("resource_type", t) for t in resource_types] @@ -211,14 +199,12 @@ Resources: def _detach_and_delete_policy(iam_client, role_name, policy_arn, policy_name): # Detach + delete are both no-ops if the entity is already gone, so callers can blindly # iterate the policy-name space without first checking what actually exists. - errors = [] try: iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy_arn) except iam_client.exceptions.NoSuchEntityException: pass except Exception as e: LOGGER.error(f"Error detaching policy {policy_name}: {str(e)}") - errors.append(f"detach {policy_name}: {e}") try: iam_client.delete_policy(PolicyArn=policy_arn) @@ -226,11 +212,8 @@ Resources: pass except iam_client.exceptions.DeleteConflictException: LOGGER.warning(f"Policy {policy_name} still attached, skipping delete") - errors.append(f"delete {policy_name}: policy is still attached") except Exception as e: LOGGER.error(f"Error deleting policy {policy_name}: {str(e)}") - errors.append(f"delete {policy_name}: {e}") - return errors def _cleanup_chunked_policies(iam_client, role_name, account_id, partition, prefix, max_policies=10): @@ -268,39 +251,35 @@ Resources: def attach_standard_permissions(iam_client, role_name): permissions = fetch_permissions_from_datadog(STANDARD_PERMISSIONS_API_URL) + policy_document = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": permissions, "Resource": "*"}], + } iam_client.put_role_policy( RoleName=role_name, PolicyName=POLICY_NAME_STANDARD, - PolicyDocument=json.dumps(_broad_policy_document(permissions), separators=(',', ':')), + PolicyDocument=json.dumps(policy_document, separators=(',', ':')), ) - def _serialize_policy_document(policy_document): - if not isinstance(policy_document, dict): - raise DatadogAPIError("instrumentation policy document must be an object") - if policy_document.get("Version") != POLICY_DOCUMENT_VERSION: - raise DatadogAPIError("instrumentation policy document has an unsupported version") - if not isinstance(policy_document.get("Statement"), list) or not policy_document["Statement"]: - raise DatadogAPIError("instrumentation policy document must contain statements") - - policy_json = json.dumps(policy_document, separators=(',', ':')) - if len(policy_json) > MANAGED_POLICY_CHAR_LIMIT: - raise DatadogAPIError( - f"instrumentation policy document exceeds {MANAGED_POLICY_CHAR_LIMIT} characters" - ) - return policy_json + def _create_and_attach_policy(iam_client, role_name, policy_name, actions): + policy_json = json.dumps( + { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": actions, "Resource": "*"}], + }, + separators=(',', ':'), + ) + LOGGER.info(f"Creating policy {policy_name} with {len(actions)} permissions ({len(policy_json)} characters)") + policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=policy_json) + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy['Policy']['Arn']) - def _create_policy(iam_client, policy_name, policy_document): - policy_json = _serialize_policy_document(policy_document) + def _create_and_attach_policy_document(iam_client, role_name, policy_name, policy_document): + policy_json = json.dumps(policy_document, separators=(',', ':')) LOGGER.info(f"Creating policy {policy_name} ({len(policy_json)} characters)") policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=policy_json) - return policy['Policy']['Arn'] - - - def _create_and_attach_policy(iam_client, role_name, policy_name, actions): - policy_arn = _create_policy(iam_client, policy_name, _broad_policy_document(actions)) - iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy_arn) + iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy['Policy']['Arn']) def attach_resource_collection_permissions(iam_client, role_name): @@ -328,226 +307,37 @@ Resources: marker = response["Marker"] - def _list_policy_versions(iam_client, policy_arn): - versions = [] - marker = None - while True: - request = {"PolicyArn": policy_arn} - if marker: - request["Marker"] = marker - response = iam_client.list_policy_versions(**request) - versions.extend(response.get("Versions", [])) - if not response.get("IsTruncated"): - return versions - marker = response["Marker"] - - - def _policy_quotas(iam_client): - summary = iam_client.get_account_summary().get("SummaryMap", {}) - attachment_quota = summary.get("AttachedPoliciesPerRoleQuota") - version_quota = summary.get("VersionsPerPolicyQuota") - if not isinstance(attachment_quota, int) or attachment_quota < 1: - raise RuntimeError("IAM account summary did not include AttachedPoliciesPerRoleQuota") - if not isinstance(version_quota, int) or version_quota < 2: - raise RuntimeError("IAM account summary did not include VersionsPerPolicyQuota") - return attachment_quota, version_quota - - - def _instrumentation_policy_name(role_name, index): - return f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{index}" - - - def _instrumentation_policy_index(policy_name, role_name): + def _is_instrumentation_policy(policy_name, role_name): prefix = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-" if not policy_name.startswith(prefix): - return None + return False suffix = policy_name[len(prefix):] - if not suffix.isdigit(): - return None - index = int(suffix) - return index if index >= 1 else None + return suffix.isdigit() and 1 <= int(suffix) <= MAX_POLICY_DOCUMENTS - def _stage_policy_version(iam_client, policy_arn, policy_document, version_quota): - versions = _list_policy_versions(iam_client, policy_arn) - previous_default = next( - (version["VersionId"] for version in versions if version.get("IsDefaultVersion")), - None, + def _validate_instrumentation_policy_capacity(iam_client, role_name, policy_document_count): + attached_policies = _list_attached_role_policies(iam_client, role_name) + existing_instrumentation_count = sum( + _is_instrumentation_policy(policy["PolicyName"], role_name) + for policy in attached_policies ) - if previous_default is None: - raise RuntimeError(f"policy {policy_arn} has no default version") - - non_default_versions = sorted( - (version for version in versions if not version.get("IsDefaultVersion")), - key=lambda version: int(version["VersionId"].removeprefix("v")), + non_instrumentation_count = len(attached_policies) - existing_instrumentation_count + attachment_quota = iam_client.get_account_summary().get("SummaryMap", {}).get( + "AttachedPoliciesPerRoleQuota" ) - while len(versions) >= version_quota: - if not non_default_versions: - raise RuntimeError(f"policy {policy_arn} has no removable non-default version") - version = non_default_versions.pop(0) - iam_client.delete_policy_version( - PolicyArn=policy_arn, - VersionId=version["VersionId"], - ) - versions.remove(version) - - response = iam_client.create_policy_version( - PolicyArn=policy_arn, - PolicyDocument=_serialize_policy_document(policy_document), - SetAsDefault=False, - ) - return { - "PolicyArn": policy_arn, - "PreviousVersionId": previous_default, - "StagedVersionId": response["PolicyVersion"]["VersionId"], - } - - - def _discard_staged_replacement(iam_client, role_name, staged_versions, created_policies): - cleanup_errors = [] - for policy in reversed(created_policies): - if policy.get("AttachAttempted"): - try: - iam_client.detach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) - except iam_client.exceptions.NoSuchEntityException: - pass - except Exception as e: - cleanup_errors.append(f"detach {policy['PolicyArn']}: {e}") - try: - iam_client.delete_policy(PolicyArn=policy["PolicyArn"]) - except Exception as e: - cleanup_errors.append(f"delete {policy['PolicyArn']}: {e}") - - for version in reversed(staged_versions): - if version.get("Activated"): - try: - iam_client.set_default_policy_version( - PolicyArn=version["PolicyArn"], - VersionId=version["PreviousVersionId"], - ) - except Exception as e: - cleanup_errors.append(f"restore {version['PolicyArn']}: {e}") - continue - try: - iam_client.delete_policy_version( - PolicyArn=version["PolicyArn"], - VersionId=version["StagedVersionId"], - ) - except Exception as e: - cleanup_errors.append(f"delete staged version for {version['PolicyArn']}: {e}") - return cleanup_errors - - - def replace_instrumentation_policies(iam_client, role_name, account_id, partition, policy_documents): - attached_policies = _list_attached_role_policies(iam_client, role_name) - attachment_quota, version_quota = _policy_quotas(iam_client) - existing_policies = {} - for policy in attached_policies: - index = _instrumentation_policy_index(policy["PolicyName"], role_name) - if index is not None: - existing_policies[index] = policy - - non_instrumentation_count = len(attached_policies) - len(existing_policies) - final_attachment_count = non_instrumentation_count + len(policy_documents) - if final_attachment_count > attachment_quota: + if not isinstance(attachment_quota, int) or attachment_quota < 1: + raise RuntimeError("IAM account summary did not include AttachedPoliciesPerRoleQuota") + if non_instrumentation_count + policy_document_count > attachment_quota: raise RuntimeError( f"role {role_name} has room for only " f"{attachment_quota - non_instrumentation_count} instrumentation policies, " - f"but Datadog returned {len(policy_documents)}" - ) - - LOGGER.info( - f"Replacing {len(existing_policies)} instrumentation policies with " - f"{len(policy_documents)} documents; final managed-policy usage " - f"{final_attachment_count}/{attachment_quota}" - ) - ordered_existing_policies = [ - policy for _, policy in sorted(existing_policies.items()) - ] - used_policy_indices = set(existing_policies) - next_policy_index = 1 - reused_policy_arns = set() - staged_versions = [] - created_policies = [] - try: - for document_index, policy_document in enumerate(policy_documents): - existing = ( - ordered_existing_policies[document_index] - if document_index < len(ordered_existing_policies) - else None - ) - if existing: - reused_policy_arns.add(existing["PolicyArn"]) - staged_versions.append( - _stage_policy_version( - iam_client, - existing["PolicyArn"], - policy_document, - version_quota, - ) - ) - continue - - while next_policy_index in used_policy_indices: - next_policy_index += 1 - policy_name = _instrumentation_policy_name(role_name, next_policy_index) - used_policy_indices.add(next_policy_index) - policy_arn = f"arn:{partition}:iam::{account_id}:policy/{policy_name}" - try: - iam_client.delete_policy(PolicyArn=policy_arn) - except iam_client.exceptions.NoSuchEntityException: - pass - created_policies.append({ - "PolicyArn": _create_policy(iam_client, policy_name, policy_document), - "AttachAttempted": False, - }) - - for version in staged_versions: - iam_client.set_default_policy_version( - PolicyArn=version["PolicyArn"], - VersionId=version["StagedVersionId"], - ) - version["Activated"] = True - for policy in created_policies: - policy["AttachAttempted"] = True - iam_client.attach_role_policy(RoleName=role_name, PolicyArn=policy["PolicyArn"]) - except Exception as e: - cleanup_errors = _discard_staged_replacement( - iam_client, - role_name, - staged_versions, - created_policies, - ) - if cleanup_errors: - raise RuntimeError( - f"failed to replace instrumentation policies: {e}; " - f"rollback errors: {'; '.join(cleanup_errors)}" - ) from e - raise - - cleanup_errors = [] - for policy in existing_policies.values(): - if policy["PolicyArn"] in reused_policy_arns: - continue - cleanup_errors.extend(_detach_and_delete_policy( - iam_client, - role_name, - policy["PolicyArn"], - policy["PolicyName"], - )) - if cleanup_errors: - raise RuntimeError( - f"new instrumentation policies are active, but old policy cleanup failed: " - f"{'; '.join(cleanup_errors)}" + f"but Datadog returned {policy_document_count}" ) def attach_instrumentation_permissions(iam_client, role_name, account_id, partition, datadog_site, resource_types, previous_resource_types, fail_on_error=False): - # Role creation treats instrumentation as best-effort; the add-on fails when its only task - # cannot complete. Replacement stages policy versions before activation so it can roll back. + # Fetch and validate capacity before cleanup so a transient failure leaves existing policies. if not resource_types: - # Only clean up if the previous Update had instrumentation enabled — avoids running - # delete calls on stacks that never opted in to instrumentation in the first place. if previous_resource_types: cleanup_instrumentation_policies(iam_client, role_name, account_id, partition) return @@ -561,20 +351,34 @@ Resources: ) LOGGER.info(f"Fetching instrumentation permissions for {resource_types} from {url}") policy_documents = fetch_instrumentation_policy_documents(url) - replace_instrumentation_policies( + _validate_instrumentation_policy_capacity( iam_client, role_name, - account_id, - partition, - policy_documents, + len(policy_documents), ) except Exception as e: if fail_on_error: raise LOGGER.warning( - f"Failed to update instrumentation permissions for {resource_types}: {e}. " + f"Failed to prepare instrumentation permissions for {resource_types}: {e}. " "Leaving any previously-attached instrumentation policies in place." ) + return + + cleanup_instrumentation_policies(iam_client, role_name, account_id, partition) + for i, policy_document in enumerate(policy_documents): + policy_name = f"{BASE_POLICY_PREFIX_INSTRUMENTATION}-{role_name}-{i+1}" + try: + _create_and_attach_policy_document( + iam_client, + role_name, + policy_name, + policy_document, + ) + except Exception as e: + if fail_on_error: + raise + LOGGER.warning(f"Failed to create/attach instrumentation policy {policy_name}: {e}. Continuing.") def handle_delete(event, context): @@ -582,7 +386,7 @@ Resources: role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) + manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' iam_client = boto3.client('iam') try: if manage_base_permissions: @@ -599,9 +403,9 @@ Resources: role_name = props['DatadogIntegrationRole'] account_id = props['AccountId'] partition = props.get('Partition', 'aws') - manage_base_permissions = _is_true(props.get('ManageBasePermissions', 'true')) - fail_on_instrumentation_error = _is_true(props.get('FailOnInstrumentationError', 'false')) - should_install_security_audit_policy = _is_true(props['ResourceCollectionPermissions']) + manage_base_permissions = str(props.get('ManageBasePermissions', 'true')).lower() == 'true' + fail_on_instrumentation_error = str(props.get('FailOnInstrumentationError', 'false')).lower() == 'true' + should_install_security_audit_policy = str(props['ResourceCollectionPermissions']).lower() == 'true' datadog_site = props.get('DatadogSite') or 'datadoghq.com' instrumentation_resource_types = parse_resource_types(props.get('InstrumentationResourceTypes')) previous_instrumentation_resource_types = parse_resource_types( @@ -637,6 +441,8 @@ Resources: Type: Custom::DatadogAttachIntegrationPermissionsFunctionTrigger Properties: ServiceToken: !GetAtt DatadogAttachIntegrationPermissionsFunction.Arn + # Bump this value when Lambda behavior changes so existing stacks invoke the custom resource. + PolicyAttachmentSchemaVersion: "2" DatadogIntegrationRole: !Ref IAMRoleName AccountId: !Ref AWS::AccountId Partition: !Sub "${AWS::Partition}"