fix(P29): SSM fail-loud without CMK + Terraform-rendered invoke policy (P1-3, P1-6)

---ci---
project: acdl
phase: 29
milestone: v1.8
status: execute
---/ci---

P1-3: SSM publisher now raises RuntimeError when ACDL_KMS_KEY_ID is
unset. ACDL_ALLOW_DEFAULT_KMS=1 escape hatch for local testing.
P1-6: consumer_invoke_policy.json now uses ${account_id} and ${region}
placeholders. Terraform renders them via data.aws_caller_identity +
data.aws_region + replace() at apply time. No more hardcoded 000000000000.

Tests: +7 (285 -> 292). All pass.
This commit is contained in:
Jon Chery
2026-07-22 22:05:40 +00:00
parent 0eb578c606
commit 843cd17b97
4 changed files with 136 additions and 3 deletions
+97 -1
View File
@@ -82,6 +82,7 @@ class TestPublishToSsm:
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("ACDL_KMS_KEY_ID", "alias/aws/ssm")
with mock_aws():
ssm = boto3.client("ssm", region_name="us-east-1")
@@ -104,6 +105,7 @@ class TestPublishToSsm:
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("ACDL_KMS_KEY_ID", "alias/aws/ssm")
with mock_aws():
ssm = boto3.client("ssm", region_name="us-east-1")
@@ -121,6 +123,7 @@ class TestPublishToSsm:
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("ACDL_KMS_KEY_ID", "alias/aws/ssm")
with mock_aws():
ssm = boto3.client("ssm", region_name="us-east-1")
@@ -365,4 +368,97 @@ class TestCli:
capture_output=True, text=True, cwd=str(Path(__file__).resolve().parent.parent),
)
assert result.returncode == 2
assert "usage" in result.stderr.lower()
assert "usage" in result.stderr.lower()
# ---------------------------------------------------------------------------
# P1-3: KMS fail-loud tests
# ---------------------------------------------------------------------------
class TestKmsFailLoud:
"""P1-3: SSM publisher must fail loud when ACDL_KMS_KEY_ID is unset."""
def test_kms_unset_raises(self, monkeypatch):
from moto import mock_aws
import boto3
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.delenv("ACDL_KMS_KEY_ID", raising=False)
monkeypatch.delenv("ACDL_ALLOW_DEFAULT_KMS", raising=False)
with mock_aws():
with pytest.raises(RuntimeError, match="ACDL_KMS_KEY_ID is not set"):
publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
def test_kms_unset_allow_default_kms_escape_hatch(self, monkeypatch):
from moto import mock_aws
import boto3
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.delenv("ACDL_KMS_KEY_ID", raising=False)
monkeypatch.setenv("ACDL_ALLOW_DEFAULT_KMS", "1")
with mock_aws():
ssm = boto3.client("ssm", region_name="us-east-1")
results = publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
assert results["vpc_id"] == "/acdl/dev/c-1/vpc_id"
param = ssm.get_parameter(Name="/acdl/dev/c-1/vpc_id", WithDecryption=True)
assert param["Parameter"]["Type"] == "SecureString"
def test_kms_set_takes_precedence_over_allow_default(self, monkeypatch):
from moto import mock_aws
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("ACDL_KMS_KEY_ID", "arn:aws:kms:us-east-1:123:key/abc")
monkeypatch.setenv("ACDL_ALLOW_DEFAULT_KMS", "1")
from core.output_publisher import _kms_key_id
assert _kms_key_id() == "arn:aws:kms:us-east-1:123:key/abc"
# ---------------------------------------------------------------------------
# P1-6: Invoke policy template tests
# ---------------------------------------------------------------------------
class TestInvokePolicyTemplate:
"""P1-6: consumer_invoke_policy.json must use placeholders, not hardcoded account ID."""
def test_policy_has_no_hardcoded_account_id(self):
policy_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "consumer_invoke_policy.json"
policy = json.load(open(policy_path))
resource_arn = policy["Statement"][0]["Resource"]
assert "000000000000" not in resource_arn
assert "${account_id}" in resource_arn
def test_policy_has_region_placeholder(self):
policy_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "consumer_invoke_policy.json"
policy = json.load(open(policy_path))
resource_arn = policy["Statement"][0]["Resource"]
assert "${region}" in resource_arn
def test_policy_renders_with_real_account_id(self):
"""Simulate the Terraform rendering: replace ${account_id} and ${region}."""
policy_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "consumer_invoke_policy.json"
template = open(policy_path).read()
rendered = template.replace("${account_id}", "123456789012").replace("${region}", "us-east-1")
policy = json.loads(rendered)
resource_arn = policy["Statement"][0]["Resource"]
assert resource_arn == "arn:aws:lambda:us-east-1:123456789012:function:acdl-contract-ingestor"
assert "000000000000" not in resource_arn
# ${consumerRepo} is a runtime placeholder (not a Terraform variable) — it stays.
assert "${account_id}" not in rendered
assert "${region}" not in rendered
def test_main_tf_has_caller_identity_data_source(self):
"""P1-6: main.tf must have data.aws_caller_identity for rendering."""
main_tf_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "main.tf"
main_tf = open(main_tf_path).read()
assert "data \"aws_caller_identity\" \"current\"" in main_tf
assert "rendered_invoke_policy" in main_tf
assert "consumer_invoke_policy_rendered" in main_tf