fix(terraform): validate exact pre-adoption policies

scripts/check-terraform-import-plan.py

@@ -177,6 +177,53 @@ def _distribution_id(
     return None
 
 
+def _expected_pre_adoption_bucket_policy(
+    environment: str,
+    distribution_id: str,
+) -> dict[str, Any]:
+    config = ENVIRONMENT_CONFIG[environment]
+    bucket_arn = f"arn:aws:s3:::{config['bucket_name']}"
+    distribution_arn = (
+        f"arn:aws:cloudfront::396287094661:distribution/{distribution_id}"
+    )
+    return _canonical(
+        {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {
+                        "AWS": config["bucket_auto_delete_helper_role_arn"]
+                    },
+                    "Action": [
+                        "s3:DeleteObject*",
+                        "s3:GetBucket*",
+                        "s3:List*",
+                        "s3:PutBucketPolicy",
+                    ],
+                    "Resource": [bucket_arn, f"{bucket_arn}/*"],
+                },
+                {
+                    "Effect": "Allow",
+                    "Principal": {"Service": "cloudfront.amazonaws.com"},
+                    "Action": "s3:GetObject",
+                    "Resource": f"{bucket_arn}/*",
+                    "Condition": {
+                        "StringEquals": {"AWS:SourceArn": distribution_arn}
+                    },
+                },
+                {
+                    "Effect": "Deny",
+                    "Principal": {"AWS": "*"},
+                    "Action": "s3:*",
+                    "Resource": [bucket_arn, f"{bucket_arn}/*"],
+                    "Condition": {"Bool": {"aws:SecureTransport": "false"}},
+                },
+            ],
+        }
+    )
+
+
 def _expected_bucket_policy(environment: str, distribution_id: str) -> dict[str, Any]:
     bucket = ENVIRONMENT_CONFIG[environment]["bucket_name"]
     bucket_arn = f"arn:aws:s3:::{bucket}"
@@ -208,6 +255,66 @@ def _expected_bucket_policy(environment: str, distribution_id: str) -> dict[str,
     )
 
 
+def _expected_pre_adoption_deploy_policy(
+    environment: str,
+    distribution_id: str,
+) -> dict[str, Any]:
+    config = ENVIRONMENT_CONFIG[environment]
+    bucket_arn = f"arn:aws:s3:::{config['bucket_name']}"
+    distribution_arn = (
+        f"arn:aws:cloudfront::396287094661:distribution/{distribution_id}"
+    )
+    statements: list[dict[str, Any]] = []
+    if environment == "dev":
+        statements.append(
+            {
+                "Sid": "AssumeCdkBootstrapRoles",
+                "Effect": "Allow",
+                "Action": "sts:AssumeRole",
+                "Resource": "arn:aws:iam::396287094661:role/cdk-hnb659fds-*",
+            }
+        )
+    statements.extend(
+        [
+            {
+                "Sid": "DescribeStack",
+                "Effect": "Allow",
+                "Action": "cloudformation:DescribeStacks",
+                "Resource": (
+                    "arn:aws:cloudformation:us-east-1:396287094661:stack/"
+                    f"{config['cloudformation_stack_name']}/*"
+                ),
+            },
+            {
+                "Effect": "Allow",
+                "Action": [
+                    "s3:Abort*",
+                    "s3:DeleteObject*",
+                    "s3:GetBucket*",
+                    "s3:GetObject*",
+                    "s3:List*",
+                    "s3:PutObject",
+                    "s3:PutObjectLegalHold",
+                    "s3:PutObjectRetention",
+                    "s3:PutObjectTagging",
+                    "s3:PutObjectVersionTagging",
+                ],
+                "Resource": [bucket_arn, f"{bucket_arn}/*"],
+            },
+            {
+                "Sid": "InvalidateDistribution",
+                "Effect": "Allow",
+                "Action": [
+                    "cloudfront:CreateInvalidation",
+                    "cloudfront:GetInvalidation",
+                ],
+                "Resource": distribution_arn,
+            },
+        ]
+    )
+    return _canonical({"Version": "2012-10-17", "Statement": statements})
+
+
 def _expected_deploy_policy(environment: str, distribution_id: str) -> dict[str, Any]:
     bucket = ENVIRONMENT_CONFIG[environment]["bucket_name"]
     bucket_arn = f"arn:aws:s3:::{bucket}"
@@ -325,12 +432,19 @@ def _validate_policy_update(
             f"{address}: cannot verify policy without the pinned distribution ID"
         )
         return violations
-    expected = (
+    expected_before = (
+        _expected_pre_adoption_bucket_policy(environment, distribution_id)
+        if address == BUCKET_POLICY_ADDRESS
+        else _expected_pre_adoption_deploy_policy(environment, distribution_id)
+    )
+    expected_after = (
         _expected_bucket_policy(environment, distribution_id)
         if address == BUCKET_POLICY_ADDRESS
         else _expected_deploy_policy(environment, distribution_id)
     )
-    if after_policy is not None and after_policy != expected:
+    if before_policy is not None and before_policy != expected_before:
+        violations.append(f"{address}: pre-adoption policy semantics are not exact")
+    if after_policy is not None and after_policy != expected_after:
         violations.append(f"{address}: post-adoption policy semantics are not exact")
     return violations
 

scripts/terraform_import_plan_resources.py

@@ -47,16 +47,31 @@ CONTROLLED_UPDATE_ADDRESSES = frozenset(
 ENVIRONMENT_CONFIG = {
     "dev": {
         "bucket_name": "seahaven-shoc-frontend-dev",
+        "bucket_auto_delete_helper_role_arn": (
+            "arn:aws:iam::396287094661:role/"
+            "shoc-frontend-dev-CustomS3AutoDeleteObjectsCustomRe-dmSDIY8EH7KV"
+        ),
+        "cloudformation_stack_name": "shoc-frontend-dev",
         "distribution_id": "E2CWLM1AFB964P",
         "workspace_name": "shoc-frontend-new-dev",
     },
     "staging": {
         "bucket_name": "seahaven-shoc-frontend-staging",
+        "bucket_auto_delete_helper_role_arn": (
+            "arn:aws:iam::396287094661:role/"
+            "shoc-frontend-staging-CustomS3AutoDeleteObjectsCust-QbMDqZbl7YQ3"
+        ),
+        "cloudformation_stack_name": "shoc-frontend-staging",
         "distribution_id": "E2JDVEZ6EGD49J",
         "workspace_name": "shoc-frontend-new-staging",
     },
     "tf-poc": {
         "bucket_name": "seahaven-shoc-frontend-tf-poc",
+        "bucket_auto_delete_helper_role_arn": (
+            "arn:aws:iam::396287094661:role/"
+            "shoc-frontend-tf-poc-CustomS3AutoDeleteObjectsCusto-GBCGk2k2ZORz"
+        ),
+        "cloudformation_stack_name": "shoc-frontend-tf-poc",
         "distribution_id": None,
         "workspace_name": "shoc-frontend-new-tf-poc",
     },

scripts/test-terraform-import-plan-check.py

@@ -43,6 +43,47 @@ def distribution_id(environment: str) -> str:
     return configured if isinstance(configured, str) else "ETFPOCGENERATED123"
 
 
+def pre_adoption_bucket_policy(environment: str) -> dict[str, Any]:
+    config = ENVIRONMENT_CONFIG[environment]
+    bucket_arn = f"arn:aws:s3:::{config['bucket_name']}"
+    source = (
+        "arn:aws:cloudfront::396287094661:distribution/"
+        f"{distribution_id(environment)}"
+    )
+    return {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Principal": {
+                    "AWS": config["bucket_auto_delete_helper_role_arn"]
+                },
+                "Action": [
+                    "s3:DeleteObject*",
+                    "s3:GetBucket*",
+                    "s3:List*",
+                    "s3:PutBucketPolicy",
+                ],
+                "Resource": [bucket_arn, f"{bucket_arn}/*"],
+            },
+            {
+                "Effect": "Allow",
+                "Principal": {"Service": "cloudfront.amazonaws.com"},
+                "Action": "s3:GetObject",
+                "Resource": f"{bucket_arn}/*",
+                "Condition": {"StringEquals": {"AWS:SourceArn": source}},
+            },
+            {
+                "Effect": "Deny",
+                "Principal": {"AWS": "*"},
+                "Action": "s3:*",
+                "Resource": [bucket_arn, f"{bucket_arn}/*"],
+                "Condition": {"Bool": {"aws:SecureTransport": "false"}},
+            },
+        ],
+    }
+
+
 def bucket_policy(environment: str) -> dict[str, Any]:
     bucket = ENVIRONMENT_CONFIG[environment]["bucket_name"]
     bucket_arn = f"arn:aws:s3:::{bucket}"
@@ -71,6 +112,64 @@ def bucket_policy(environment: str) -> dict[str, Any]:
     }
 
 
+def pre_adoption_deploy_policy(environment: str) -> dict[str, Any]:
+    config = ENVIRONMENT_CONFIG[environment]
+    bucket_arn = f"arn:aws:s3:::{config['bucket_name']}"
+    distribution_arn = (
+        "arn:aws:cloudfront::396287094661:distribution/"
+        f"{distribution_id(environment)}"
+    )
+    statements: list[dict[str, Any]] = []
+    if environment == "dev":
+        statements.append(
+            {
+                "Sid": "AssumeCdkBootstrapRoles",
+                "Effect": "Allow",
+                "Action": "sts:AssumeRole",
+                "Resource": "arn:aws:iam::396287094661:role/cdk-hnb659fds-*",
+            }
+        )
+    statements.extend(
+        [
+            {
+                "Sid": "DescribeStack",
+                "Effect": "Allow",
+                "Action": "cloudformation:DescribeStacks",
+                "Resource": (
+                    "arn:aws:cloudformation:us-east-1:396287094661:stack/"
+                    f"{config['cloudformation_stack_name']}/*"
+                ),
+            },
+            {
+                "Effect": "Allow",
+                "Action": [
+                    "s3:Abort*",
+                    "s3:DeleteObject*",
+                    "s3:GetBucket*",
+                    "s3:GetObject*",
+                    "s3:List*",
+                    "s3:PutObject",
+                    "s3:PutObjectLegalHold",
+                    "s3:PutObjectRetention",
+                    "s3:PutObjectTagging",
+                    "s3:PutObjectVersionTagging",
+                ],
+                "Resource": [bucket_arn, f"{bucket_arn}/*"],
+            },
+            {
+                "Sid": "InvalidateDistribution",
+                "Effect": "Allow",
+                "Action": [
+                    "cloudfront:CreateInvalidation",
+                    "cloudfront:GetInvalidation",
+                ],
+                "Resource": distribution_arn,
+            },
+        ]
+    )
+    return {"Version": "2012-10-17", "Statement": statements}
+
+
 def deploy_policy(environment: str) -> dict[str, Any]:
     bucket = ENVIRONMENT_CONFIG[environment]["bucket_name"]
     bucket_arn = f"arn:aws:s3:::{bucket}"
@@ -152,12 +251,16 @@ def tag_change(environment: str, address: str) -> dict[str, Any]:
 
 
 def policy_change(environment: str, address: str) -> dict[str, Any]:
+    before_policy = (
+        pre_adoption_bucket_policy(environment)
+        if address == BUCKET_POLICY
+        else pre_adoption_deploy_policy(environment)
+    )
     after_policy = (
         bucket_policy(environment)
         if address == BUCKET_POLICY
         else deploy_policy(environment)
     )
-    before_policy = {"Version": "2012-10-17", "Statement": []}
     return {
         "actions": ["update"],
         "before": {"policy": json.dumps(before_policy)},
@@ -384,17 +487,18 @@ class ImportPlanCheckerTests(unittest.TestCase):
         self.assert_fails(plan, "dev", post_import=True)
 
     def test_every_allowed_controlled_diff_passes(self) -> None:
-        for address in CONTROLLED_UPDATE_ADDRESSES:
+        for environment in REQUIRED_RESOURCES:
-            with self.subTest(address=address):
+            for address in CONTROLLED_UPDATE_ADDRESSES:
-                self.assert_passes(
+                with self.subTest(environment=environment, address=address):
-                    make_plan(
+                    self.assert_passes(
-                        "tf-poc",
+                        make_plan(
-                        mode="controlled",
+                            environment,
-                        controlled_updates={address},
+                            mode="controlled",
-                    ),
+                            controlled_updates={address},
-                    "tf-poc",
+                        ),
-                    address,
+                        environment,
-                )
+                        address,
+                    )
 
     def test_full_exact_controlled_allowlist_passes(self) -> None:
         addresses = tuple(sorted(CONTROLLED_UPDATE_ADDRESSES))
@@ -480,6 +584,28 @@ class ImportPlanCheckerTests(unittest.TestCase):
             with self.subTest(mutation=mutation):
                 self.assert_fails(plan, "staging", DEPLOY_POLICY)
 
+    def test_policy_updates_require_exact_pre_adoption_state(self) -> None:
+        for environment in REQUIRED_RESOURCES:
+            for address in (BUCKET_POLICY, DEPLOY_POLICY):
+                plan = make_plan(
+                    environment,
+                    mode="controlled",
+                    controlled_updates={address},
+                )
+                change = resource(plan, address)["change"]
+                before = json.loads(change["before"]["policy"])
+                before["Statement"].append(
+                    {
+                        "Sid": "UnexpectedDrift",
+                        "Effect": "Deny",
+                        "Action": "*",
+                        "Resource": "*",
+                    }
+                )
+                change["before"]["policy"] = json.dumps(before)
+                with self.subTest(environment=environment, address=address):
+                    self.assert_fails(plan, environment, address)
+
     def test_controlled_update_rejects_unknown_and_replace_paths(self) -> None:
         for field, value in (
             ("after_unknown", {"tags": {"ManagedBy": True}}),