fix(aws): scope connector permission simulation to its own role

aws-integration-setup/cloudformation/README.md

@@ -204,6 +204,12 @@ aws cloudformation update-stack \
 
 `UsePreviousValue=true` only works for parameters the stack's current template already has. A stack deployed from an older template version may lack `EKSOidcProviderURL`, `EnableEKSIntegration` or `NullifyKMSKeyArn`: for those, replace `UsePreviousValue=true` with `ParameterValue=...` (or drop the line to take the default), and keep `--template-body file://nullify-cloudformation-template.json` so the new template is used.
 
+### Connector permission preflight
+
+The read-only role can call `iam:SimulatePrincipalPolicy` only for its own role ARN. Nullify uses this read-only diagnostic to check collector permissions; it does not grant the simulated actions or permit changes to IAM. The Session Manager document grant remains restricted to `SSM-SessionManagerRunShell`.
+
+Existing installations need the updated template applied through their existing stack or StackSet. Deploy the resource-aware Nullify permission probe before updating the connector: an older probe evaluates `ssm:GetDocument` against `*` and can report a false gap for the restricted document grant. A successful role assumption alone does not prove complete collector coverage.
+
 ### 5. Collector upload target
 
 The Nullify configure page shows the upload target for the Helm chart's `collector.s3.bucket`. Which form it shows decides the setup:

aws-integration-setup/cloudformation/nullify-cloudformation-template.json

@@ -657,6 +657,14 @@
                 "Resource": "*",
                 "Condition": {}
               },
+              {
+                "Sid": "SimulateConnectorPolicy",
+                "Effect": "Allow",
+                "Action": ["iam:SimulatePrincipalPolicy"],
+                "Resource": {
+                  "Fn::Sub": "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/AWSIntegration-${CustomerName}-NullifyReadOnlyRole"
+                }
+              },
               {
                 "Effect": "Allow",
                 "Action": [

aws-integration-setup/terraform-v6/modules/nullify-aws-integration/data.tf

@@ -235,6 +235,13 @@ data "aws_iam_policy_document" "readonly_policy_part2" {
     resources = ["*"]
   }
 
+  statement {
+    sid       = "SimulateConnectorPolicy"
+    effect    = "Allow"
+    actions   = ["iam:SimulatePrincipalPolicy"]
+    resources = ["arn:${data.aws_partition.current.partition}:iam::${data.aws_caller_identity.current.account_id}:role/${local.role_name}"]
+  }
+
   statement {
     effect    = "Allow"
     actions   = ["ssm:GetDocument"]

aws-integration-setup/terraform/modules/nullify-aws-integration/data.tf

@@ -259,6 +259,13 @@ data "aws_iam_policy_document" "readonly_policy_part2" {
     resources = ["*"]
   }
 
+  statement {
+    sid       = "SimulateConnectorPolicy"
+    effect    = "Allow"
+    actions   = ["iam:SimulatePrincipalPolicy"]
+    resources = ["arn:${data.aws_partition.current.partition}:iam::${data.aws_caller_identity.current.account_id}:role/${local.role_name}"]
+  }
+
   statement {
     effect    = "Allow"
     actions   = ["ssm:GetDocument"]

scripts/check_iam_policy_drift.py

@@ -47,22 +47,67 @@
 _PARTITION_TOKEN = "${partition}"
 _CFN_PARTITION = re.compile(r"\$\{AWS::Partition\}")
 _TF_PARTITION = re.compile(r"\$\{data\.aws_partition\.current\.partition\}")
+_CFN_ACCOUNT = re.compile(r"\$\{AWS::AccountId\}")
+_TF_ACCOUNT = re.compile(r"\$\{data\.aws_caller_identity\.current\.account_id\}")
+_CFN_CUSTOMER = re.compile(r"\$\{CustomerName\}")
+_TF_CUSTOMER = re.compile(r"\$\{var\.customer_name\}")
+_TF_LOCAL = re.compile(r"\$\{local\.([A-Za-z_][A-Za-z0-9_]*)\}")
+
+
+def _norm_cfn_string(value: str) -> str:
+    value = _CFN_PARTITION.sub(_PARTITION_TOKEN, value)
+    value = _CFN_ACCOUNT.sub("${account_id}", value)
+    return _CFN_CUSTOMER.sub("${customer_name}", value)
 
 
 def _norm_cfn_value(value) -> str:
     if isinstance(value, str):
-        return _CFN_PARTITION.sub(_PARTITION_TOKEN, value)
+        return _norm_cfn_string(value)
     if isinstance(value, dict) and "Fn::Sub" in value:
         template = value["Fn::Sub"]
         if isinstance(template, list):
             template = template[0]
         if isinstance(template, str):
-            return _CFN_PARTITION.sub(_PARTITION_TOKEN, template)
+            return _norm_cfn_string(template)
     raise SystemExit(f"unsupported CloudFormation IAM value: {value!r}")
 
 
-def _norm_tf_value(value: str) -> str:
-    return _TF_PARTITION.sub(_PARTITION_TOKEN, value)
+def _norm_tf_value(value: str, local_strings=None) -> str:
+    local_strings = local_strings or {}
+
+    def expand(text, seen):
+        def replace(match):
+            name = match[1]
+            if name not in local_strings:
+                return match[0]
+            if name in seen:
+                raise SystemExit(f"cyclic Terraform string local: {name}")
+            return expand(local_strings[name], seen | {name})
+
+        return _TF_LOCAL.sub(replace, text)
+
+    value = expand(value, set())
+    value = _TF_PARTITION.sub(_PARTITION_TOKEN, value)
+    value = _TF_ACCOUNT.sub("${account_id}", value)
+    return _TF_CUSTOMER.sub("${customer_name}", value)
+
+
+def _terraform_role_locals(path: Path) -> dict:
+    skip = r"(?:\s|\#[^\n]*(?:\n|$)|//[^\n]*(?:\n|$))*"
+    quoted = r'"(?:\\.|[^"\\\r\n])*"'
+    line_end = r"[ \t]*(?:(?:\#|//)[^\n]*)?(?:\r?\n|(?=\}))"
+    pattern = (
+        r"\A" + skip + r"locals" + skip + r"\{" + skip
+        + r"role_name_prefix[ \t]*=[ \t]*(?P<prefix>" + quoted + r")" + line_end + skip
+        + r"role_name[ \t]*=[ \t]*(?P<name>" + quoted + r")" + line_end
+    )
+    match = re.match(pattern, path.read_text())
+    if match is None:
+        raise SystemExit(f"unsupported role locals in {path}: expected leading literal role_name_prefix and role_name assignments")
+    try:
+        return {"role_name_prefix": json.loads(match["prefix"]), "role_name": json.loads(match["name"])}
+    except json.JSONDecodeError as error:
+        raise SystemExit(f"unsupported role string in {path}: {error}") from error
 
 
 def _norm_resources(resources, negated: bool) -> tuple:
@@ -179,6 +224,8 @@ def _conditions(block: str) -> list:
 
 def terraform_entries(data_tf: Path, data_source_names: list) -> set:
     text = data_tf.read_text()
+    locals_path = data_tf.with_name("locals.tf")
+    local_strings = _terraform_role_locals(locals_path) if "${local.role_name}" in text else {}
     entries: set = set()
     for name in data_source_names:
         for block in _find_statement_blocks(text, name):
@@ -190,14 +237,14 @@ def terraform_entries(data_tf: Path, data_source_names: list) -> set:
                 actions, actions_negated = not_actions, True
             else:
                 actions, actions_negated = _string_list(block, "actions"), False
-            actions = [_norm_tf_value(action) for action in actions]
+            actions = [_norm_tf_value(action, local_strings) for action in actions]
 
             not_resources = _string_list(block, "not_resources")
             if not_resources:
                 resources, resources_negated = not_resources, True
             else:
                 resources, resources_negated = _string_list(block, "resources"), False
-            resources = [_norm_tf_value(resource) for resource in resources]
+            resources = [_norm_tf_value(resource, local_strings) for resource in resources]
 
             entries |= _flatten(effect, actions, actions_negated, resources, resources_negated, _conditions(block))
     return entries

scripts/test_collector_upload_scope.py

@@ -3,6 +3,7 @@
 import fnmatch
 import json
 import re
+import runpy
 import shutil
 import subprocess
 import tempfile
@@ -37,6 +38,78 @@ def render(value, parameters, conditions):
     return {key: render(item, parameters, conditions) for key, item in value.items()}
 
 
+class ConnectorPolicyProbeScope(unittest.TestCase):
+    def test_drift_parser_does_not_use_a_nested_role_name(self):
+        drift = runpy.run_path(str(ROOT / "scripts/check_iam_policy_drift.py"))
+        module = ROOT / "aws-integration-setup/terraform/modules/nullify-aws-integration"
+        expected = drift["cfn_entries"](drift["READONLY_RESOURCES"])
+        with tempfile.TemporaryDirectory() as directory:
+            target = Path(directory)
+            shutil.copy(module / "data.tf", target / "data.tf")
+            locals_text = (module / "locals.tf").read_text()
+            (target / "locals.tf").write_text(locals_text)
+            self.assertEqual(expected, drift["terraform_entries"](target / "data.tf", drift["READONLY_SOURCES"]))
+            wrong = re.sub(r'(?m)^(\s*role_name\s*=\s*).*$', r'\1"wrong-role"', locals_text)
+            wrong = wrong.rsplit("}", 1)[0] + '\n harmless_metadata = {\n role_name = "${local.role_name_prefix}-NullifyReadOnlyRole"\n }\n}\n'
+            (target / "locals.tf").write_text(wrong)
+            self.assertNotEqual(expected, drift["terraform_entries"](target / "data.tf", drift["READONLY_SOURCES"]))
+
+    def test_drift_parser_refuses_unsupported_role_expressions(self):
+        drift = runpy.run_path(str(ROOT / "scripts/check_iam_policy_drift.py"))
+        prefix = 'role_name_prefix = "AWSIntegration-${var.customer_name}"\n'
+        role = 'role_name = "${local.role_name_prefix}-NullifyReadOnlyRole"\n'
+        cases = {
+            "nested first": 'locals {\n metadata = {\n' + prefix + role + '}\n' + prefix + 'role_name = "wrong-role"\n}\n',
+            "block comment": 'locals {\n /*\n' + prefix + role + '*/\n' + prefix + 'role_name = "wrong-role"\n}\n',
+            "conditional": 'locals {\n' + prefix + role.rstrip() + ' != "" ? "wrong-role" : "other-role"\n}\n',
+            "heredoc": 'locals {\n' + prefix + 'role_name = <<-EOT\nwrong-role\nEOT\n}\n',
+        }
+        module = ROOT / "aws-integration-setup/terraform/modules/nullify-aws-integration"
+        with tempfile.TemporaryDirectory() as directory:
+            target = Path(directory)
+            shutil.copy(module / "data.tf", target / "data.tf")
+            for name, source in cases.items():
+                with self.subTest(name=name):
+                    (target / "locals.tf").write_text(source)
+                    with self.assertRaises(SystemExit):
+                        drift["terraform_entries"](target / "data.tf", drift["READONLY_SOURCES"])
+
+    def test_drift_normalization_preserves_role_and_account_boundaries(self):
+        drift = runpy.run_path(str(ROOT / "scripts/check_iam_policy_drift.py"))
+        cfn = drift["_norm_cfn_value"]({"Fn::Sub": "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/AWSIntegration-${CustomerName}-NullifyReadOnlyRole"})
+        resource = "arn:${data.aws_partition.current.partition}:iam::${data.aws_caller_identity.current.account_id}:role/${local.role_name}"
+        locals_ = {"role_name": "${local.role_name_prefix}-NullifyReadOnlyRole", "role_name_prefix": "AWSIntegration-${var.customer_name}"}
+        normalize = drift["_norm_tf_value"]
+        self.assertEqual(cfn, normalize(resource, locals_))
+        self.assertNotEqual(cfn, normalize(resource, {**locals_, "role_name": "another-role"}))
+        self.assertNotEqual(cfn, normalize(resource.replace("${data.aws_caller_identity.current.account_id}", "999999999999"), locals_))
+        with self.assertRaises(SystemExit):
+            normalize(resource, {"role_name": "${local.role_name}"})
+
+    def test_only_the_connector_role_can_be_simulated(self):
+        statements = TEMPLATE["Resources"]["ReadOnlyAccessPolicy2"]["Properties"]["PolicyDocument"]["Statement"]
+        grants = [item for item in statements if "iam:SimulatePrincipalPolicy" in item["Action"]]
+        self.assertEqual(len(grants), 1)
+        for partition in ("aws", "aws-us-gov", "aws-cn"):
+            parameters = {"AWS::Partition": partition, "AWS::AccountId": "123456789012", "CustomerName": "acme"}
+            grant = render(grants[0], parameters, {})
+            own = f"arn:{partition}:iam::123456789012:role/AWSIntegration-acme-NullifyReadOnlyRole"
+            self.assertEqual(grant["Resource"], own)
+            self.assertEqual(grant["Action"], ["iam:SimulatePrincipalPolicy"])
+            self.assertEqual(grant["Effect"], "Allow")
+            for foreign in (own.replace("acme", "other"), own.replace("123456789012", "999999999999")):
+                self.assertFalse(fnmatch.fnmatchcase(foreign, grant["Resource"]))
+
+    def test_terraform_keeps_the_same_self_scope(self):
+        for version in ("terraform", "terraform-v6"):
+            source = (ROOT / f"aws-integration-setup/{version}/modules/nullify-aws-integration/data.tf").read_text()
+            statements = re.findall(r"statement \{(.*?)\n  \}", source, re.S)
+            grants = [item for item in statements if '"iam:SimulatePrincipalPolicy"' in item]
+            self.assertEqual(len(grants), 1)
+            self.assertIn('resources = ["arn:${data.aws_partition.current.partition}:iam::${data.aws_caller_identity.current.account_id}:role/${local.role_name}"]', grants[0])
+            self.assertIn('actions   = ["iam:SimulatePrincipalPolicy"]', grants[0])
+
+
 class CollectorUploadScope(unittest.TestCase):
     def test_two_sources_never_grant_sibling_uploads(self):
         policy = TEMPLATE["Resources"]["S3AccessPolicy"]["Properties"]["PolicyDocument"]