fix(aws): grant configuration-only Lambda reads to connector roles

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

@@ -614,6 +614,7 @@
                   "kms:GetKeyRotationStatus",
                   "kms:List*",
                   "lambda:GetFunction",
+                  "lambda:GetFunctionConfiguration",
                   "lambda:GetFunctionUrlConfig",
                   "lambda:GetPolicy",
                   "lambda:List*",

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

@@ -192,6 +192,7 @@ data "aws_iam_policy_document" "readonly_policy_part2" {
       "kms:GetKeyRotationStatus",
       "kms:List*",
       "lambda:GetFunction",
+      "lambda:GetFunctionConfiguration",
       "lambda:GetFunctionUrlConfig",
       "lambda:GetPolicy",
       "lambda:List*",

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

@@ -216,6 +216,7 @@ data "aws_iam_policy_document" "readonly_policy_part2" {
       "kms:GetKeyRotationStatus",
       "kms:List*",
       "lambda:GetFunction",
+      "lambda:GetFunctionConfiguration",
       "lambda:GetFunctionUrlConfig",
       "lambda:GetPolicy",
       "lambda:List*",

scripts/check_iam_policy_drift.py

@@ -21,6 +21,7 @@
 Exits non-zero and prints the symmetric difference on any mismatch.
 """
 
+import fnmatch
 import json
 import re
 import sys
@@ -203,6 +204,17 @@ def terraform_entries(data_tf: Path, data_source_names: list) -> set:
     return entries
 
 
+def grants_cloud_configuration_read(readonly: set, denied: set) -> bool:
+    action = "lambda:GetFunctionConfiguration"
+    required = ("Allow", "action", action, ("resources", ("*",)), ())
+    if required not in readonly:
+        return False
+    return not any(
+        effect == "Deny" and (key == "not_action" or fnmatch.fnmatchcase(action.lower(), pattern.lower()))
+        for effect, key, pattern, _, _ in denied
+    )
+
+
 def compare(label: str, cfn: set, terraform: set, reference: str = "CloudFormation") -> bool:
     if cfn == terraform:
         print(f"OK: {label} matches ({len(cfn)} entries)")
@@ -270,12 +282,17 @@ def main() -> int:
     cfn_readonly = cfn_entries(READONLY_RESOURCES)
     cfn_deny = cfn_entries(DENY_RESOURCES)
 
-    ok = True
+    ok = grants_cloud_configuration_read(cfn_readonly, cfn_deny)
+    if not ok:
+        print("MISSING: exact Lambda configuration-read grant in CloudFormation")
     for tree in TERRAFORM_TREES:
         data_tf = REPO_ROOT / tree / DATA_TF_RELATIVE
         tf_readonly = terraform_entries(data_tf, READONLY_SOURCES)
         tf_deny = terraform_entries(data_tf, [DENY_SOURCE])
 
+        if not grants_cloud_configuration_read(tf_readonly, tf_deny):
+            print(f"MISSING: exact Lambda configuration-read grant in {tree}")
+            ok = False
         ok = compare(f"Allow statements: CloudFormation vs {tree}", cfn_readonly, tf_readonly) and ok
         ok = compare(f"Deny statements: CloudFormation vs {tree}", cfn_deny, tf_deny) and ok
 

scripts/test_collector_upload_scope.py

@@ -9,6 +9,8 @@
 import unittest
 from pathlib import Path
 
+import check_iam_policy_drift as policy_drift
+
 ROOT = Path(__file__).resolve().parents[1]
 TEMPLATE = json.loads((ROOT / "aws-integration-setup/cloudformation/nullify-cloudformation-template.json").read_text())
 
@@ -37,6 +39,36 @@ def render(value, parameters, conditions):
     return {key: render(item, parameters, conditions) for key, item in value.items()}
 
 
+class CloudConfigurationReadPermissions(unittest.TestCase):
+    def policy_sources(self):
+        yield "CloudFormation", policy_drift.cfn_entries(policy_drift.READONLY_RESOURCES), policy_drift.cfn_entries(policy_drift.DENY_RESOURCES)
+        for tree in policy_drift.TERRAFORM_TREES:
+            source = ROOT / tree / policy_drift.DATA_TF_RELATIVE
+            yield tree, policy_drift.terraform_entries(source, policy_drift.READONLY_SOURCES), policy_drift.terraform_entries(source, [policy_drift.DENY_SOURCE])
+
+    def test_actual_templates_grant_configuration_reads(self):
+        for source, allowed, denied in self.policy_sources():
+            with self.subTest(source=source):
+                self.assertTrue(policy_drift.grants_cloud_configuration_read(allowed, denied))
+
+    def test_missing_broad_scoped_or_denied_grants_do_not_pass(self):
+        required = ("Allow", "action", "lambda:GetFunctionConfiguration", ("resources", ("*",)), ())
+        for source, allowed, denied in self.policy_sources():
+            self.assertIn(required, allowed)
+            without = allowed - {required}
+            cases = {
+                "missing": without,
+                "wildcard": without | {("Allow", "action", "lambda:Get*", ("resources", ("*",)), ())},
+                "scoped": without | {("Allow", "action", "lambda:GetFunctionConfiguration", ("resources", ("arn:aws:lambda:us-east-1:123456789012:function:one",)), ())},
+                "wrong effect": without | {("Deny", "action", "lambda:GetFunctionConfiguration", ("resources", ("*",)), ())},
+            }
+            for case, changed in cases.items():
+                with self.subTest(source=source, case=case):
+                    self.assertFalse(policy_drift.grants_cloud_configuration_read(changed, denied))
+            with self.subTest(source=source, case="explicit deny"):
+                self.assertFalse(policy_drift.grants_cloud_configuration_read(allowed, denied | {("Deny", "action", "lambda:*", ("resources", ("*",)), ())}))
+
+
 class CollectorUploadScope(unittest.TestCase):
     def test_two_sources_never_grant_sibling_uploads(self):
         policy = TEMPLATE["Resources"]["S3AccessPolicy"]["Properties"]["PolicyDocument"]