from __future__ import annotations import importlib.util import io import subprocess import sys import unittest from pathlib import Path from unittest.mock import patch SCRIPT = Path(__file__).parents[3] / "scripts" / "terraform_state_backend_preflight.py" SPEC = importlib.util.spec_from_file_location("terraform_state_backend_preflight", SCRIPT) assert SPEC and SPEC.loader preflight = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = preflight SPEC.loader.exec_module(preflight) class FakeAws: def __init__(self) -> None: self.versioning = "Enabled" self.encryption = ["aws:kms"] self.public_access = { "BlockPublicAcls": True, "IgnorePublicAcls": True, "BlockPublicPolicy": True, "RestrictPublicBuckets": True, } self.table = { "Status": "ACTIVE", "Keys": [{"AttributeName": "LockID", "KeyType": "HASH"}], } self.state_metadata = { "ETag": '"etag"', "ServerSideEncryption": "aws:kms", "SSEKMSKeyId": "arn:aws:kms:ap-southeast-1:123456789012:key/test", } self.head_buckets: list[str] = [] def bucket_versioning(self, _bucket: str) -> str: return self.versioning def bucket_encryption_algorithms(self, _bucket: str) -> list[str]: return self.encryption def public_access_block(self, _bucket: str) -> dict[str, object]: return self.public_access def lock_table(self, _table_name: str) -> dict[str, object]: return self.table def state_object_metadata(self, bucket: str) -> dict[str, object]: self.head_buckets.append(bucket) if not self.state_metadata: raise preflight.StateObjectAbsent(preflight.PROD_STATE_KEY) return self.state_metadata def execute(fake: FakeAws) -> tuple[int, str, str]: stdout = io.StringIO() stderr = io.StringIO() result = preflight.run_preflight( fake, bucket="state-bucket", lock_table="lock-table", out=stdout, err=stderr, ) return result, stdout.getvalue(), stderr.getvalue() class PreflightDecisionTests(unittest.TestCase): def test_safe_backend_with_exact_state_metadata_passes(self) -> None: fake = FakeAws() result, stdout, stderr = execute(fake) self.assertEqual(result, 0) self.assertEqual(fake.head_buckets, ["state-bucket"]) self.assertIn(preflight.PROD_STATE_KEY, stdout) self.assertEqual(stderr, "") def test_missing_exact_state_refuses_normal_plan_without_creating_state(self) -> None: fake = FakeAws() fake.state_metadata = {} result, stdout, stderr = execute(fake) self.assertEqual(result, preflight.EXIT_STATE_ABSENT) self.assertEqual(stdout, "") self.assertIn("canonical production state object is absent", stderr) self.assertIn("Refusing normal Terraform plan/apply", stderr) self.assertIn(preflight.ADOPTION_RUNBOOK, stderr) def test_disabled_versioning_fails_before_state_head(self) -> None: fake = FakeAws() fake.versioning = "Suspended" result, _stdout, stderr = execute(fake) self.assertEqual(result, preflight.EXIT_UNSAFE_BACKEND) self.assertEqual(fake.head_buckets, []) self.assertIn("bucket versioning is Enabled", stderr) def test_non_kms_encryption_is_rejected(self) -> None: fake = FakeAws() fake.encryption = ["AES256"] result, _stdout, stderr = execute(fake) self.assertEqual(result, preflight.EXIT_UNSAFE_BACKEND) self.assertIn("default encryption includes aws:kms", stderr) def test_state_object_must_itself_use_kms_encryption(self) -> None: fake = FakeAws() fake.state_metadata = { "ETag": '"etag"', "ServerSideEncryption": "AES256", } result, _stdout, stderr = execute(fake) self.assertEqual(result, preflight.EXIT_UNSAFE_BACKEND) self.assertIn("state object is not encrypted with a KMS key", stderr) def test_any_disabled_public_access_block_is_rejected(self) -> None: fake = FakeAws() fake.public_access["RestrictPublicBuckets"] = False result, _stdout, stderr = execute(fake) self.assertEqual(result, preflight.EXIT_UNSAFE_BACKEND) self.assertIn("all four S3 public-access blocks", stderr) def test_inactive_or_malformed_lock_table_is_rejected(self) -> None: fake = FakeAws() fake.table = { "Status": "ACTIVE", "Keys": [{"AttributeName": "WrongKey", "KeyType": "HASH"}], } result, _stdout, stderr = execute(fake) self.assertEqual(result, preflight.EXIT_UNSAFE_BACKEND) self.assertIn("LockID hash key", stderr) class AwsCliSafetyTests(unittest.TestCase): @patch.object(preflight.subprocess, "run") def test_state_check_uses_head_object_for_canonical_key_only(self, run_mock) -> None: run_mock.return_value = subprocess.CompletedProcess( [], 0, stdout=( '{"ETag":"etag","ServerSideEncryption":"aws:kms",' '"SSEKMSKeyId":"arn:aws:kms:region:account:key/test"}\n' ), stderr="", ) aws = preflight.AwsCli(region="ap-southeast-1") metadata = aws.state_object_metadata("state-bucket") command = run_mock.call_args.args[0] self.assertEqual(command[0:3], ["aws", "s3api", "head-object"]) self.assertEqual(command[command.index("--key") + 1], preflight.PROD_STATE_KEY) self.assertNotIn("get-object", command) self.assertEqual(metadata["ServerSideEncryption"], "aws:kms") @patch.object(preflight.subprocess, "run") def test_aws_error_details_are_not_exposed(self, run_mock) -> None: run_mock.return_value = subprocess.CompletedProcess( [], 255, stdout="", stderr="AccessDenied: account-specific-detail", ) aws = preflight.AwsCli(region="ap-southeast-1") with self.assertRaisesRegex( preflight.PreflightError, "AWS metadata check failed" ) as raised: aws.bucket_versioning("state-bucket") self.assertNotIn("account-specific-detail", str(raised.exception)) @patch.object(preflight.subprocess, "run", side_effect=FileNotFoundError("aws")) def test_missing_aws_cli_fails_closed_without_trace_details(self, _run_mock) -> None: aws = preflight.AwsCli(region="ap-southeast-1") with self.assertRaisesRegex(preflight.PreflightError, "could not run") as raised: aws.bucket_versioning("state-bucket") self.assertNotIn("FileNotFoundError", str(raised.exception)) @patch.object(preflight.subprocess, "run") def test_head_404_is_classified_without_exposing_aws_error(self, run_mock) -> None: run_mock.return_value = subprocess.CompletedProcess( [], 255, stdout="", stderr=( "An error occurred (404) when calling the HeadObject operation: Not Found" ), ) aws = preflight.AwsCli(region="ap-southeast-1") with self.assertRaises(preflight.StateObjectAbsent) as raised: aws.state_object_metadata("state-bucket") self.assertEqual(str(raised.exception), preflight.PROD_STATE_KEY) @patch.object(preflight.subprocess, "run") def test_missing_profile_is_not_misclassified_as_absent_state(self, run_mock) -> None: run_mock.return_value = subprocess.CompletedProcess( [], 255, stdout="", stderr="The config profile (404-not found) could not be found", ) aws = preflight.AwsCli(region="ap-southeast-1", profile="missing") with self.assertRaises(preflight.PreflightError) as raised: aws.state_object_metadata("state-bucket") self.assertNotIsInstance(raised.exception, preflight.StateObjectAbsent) @patch.object(preflight.subprocess, "run") def test_head_access_denied_is_not_misclassified_as_absent_state(self, run_mock) -> None: run_mock.return_value = subprocess.CompletedProcess( [], 255, stdout="", stderr=( "An error occurred (AccessDenied) when calling the HeadObject operation" ), ) aws = preflight.AwsCli(region="ap-southeast-1") with self.assertRaises(preflight.PreflightError) as raised: aws.state_object_metadata("state-bucket") self.assertNotIsInstance(raised.exception, preflight.StateObjectAbsent) @patch.object(preflight.subprocess, "run") def test_expired_credentials_are_not_misclassified_as_absent_state( self, run_mock ) -> None: run_mock.return_value = subprocess.CompletedProcess( [], 255, stdout="", stderr=( "An error occurred (ExpiredToken) when calling the HeadObject operation" ), ) aws = preflight.AwsCli(region="ap-southeast-1") with self.assertRaises(preflight.PreflightError) as raised: aws.state_object_metadata("state-bucket") self.assertNotIsInstance(raised.exception, preflight.StateObjectAbsent) if __name__ == "__main__": unittest.main()