diff --git a/core/output_publisher.py b/core/output_publisher.py index 144a2aa..124089b 100644 --- a/core/output_publisher.py +++ b/core/output_publisher.py @@ -52,7 +52,23 @@ def _ssm_client(): def _kms_key_id(): - return os.environ.get(KMS_KEY_ID_ENV, "alias/aws/ssm") + """Return the KMS key ID for SSM SecureString encryption. + + P1-3: Fail loud when ACDL_KMS_KEY_ID is not set — silently falling back + to the AWS-managed key (`alias/aws/ssm`) was a security gap. The platform + CMK must be explicitly configured. Set ACDL_ALLOW_DEFAULT_KMS=1 to use + the AWS-managed key as an escape hatch for local testing. + """ + key_id = os.environ.get(KMS_KEY_ID_ENV) + if key_id: + return key_id + if os.environ.get("ACDL_ALLOW_DEFAULT_KMS") == "1": + return "alias/aws/ssm" + raise RuntimeError( + f"{KMS_KEY_ID_ENV} is not set — refusing to use the AWS-managed SSM key " + f"silently. Set {KMS_KEY_ID_ENV} to your platform CMK ARN, or set " + f"ACDL_ALLOW_DEFAULT_KMS=1 to use alias/aws/ssm (escape hatch for local testing)." + ) def publish_to_ssm(outputs, environment, contract_id): diff --git a/terraform/platform/consumer_invoke_policy.json b/terraform/platform/consumer_invoke_policy.json index 7c24c59..5ba4afe 100644 --- a/terraform/platform/consumer_invoke_policy.json +++ b/terraform/platform/consumer_invoke_policy.json @@ -4,7 +4,7 @@ { "Effect": "Allow", "Action": "lambda:InvokeFunctionUrl", - "Resource": "arn:aws:lambda:us-east-1:000000000000:function:acdl-contract-ingestor", + "Resource": "arn:aws:lambda:${region}:${account_id}:function:acdl-contract-ingestor", "Condition": { "StringEquals": { "aws:PrincipalTag/acdl:owner": "${consumerRepo}" diff --git a/terraform/platform/main.tf b/terraform/platform/main.tf index 8148a73..68fd043 100644 --- a/terraform/platform/main.tf +++ b/terraform/platform/main.tf @@ -161,4 +161,25 @@ resource "aws_lambda_function" "contract_ingestor" { resource "aws_lambda_function_url" "contract_ingestor" { function_name = aws_lambda_function.contract_ingestor.function_name authorization_type = "AWS_IAM" +} + +# P1-6: Render the consumer invoke policy with the live AWS account ID. +# The JSON template (consumer_invoke_policy.json) uses ${account_id} and +# ${region} placeholders. Terraform renders them at apply time using the +# caller's live account ID — no hardcoded placeholder account IDs. +data "aws_caller_identity" "current" {} + +data "aws_region" "current" {} + +locals { + invoke_policy_template = file("${path.module}/consumer_invoke_policy.json") + rendered_invoke_policy = replace( + replace(local.invoke_policy_template, "${account_id}", data.aws_caller_identity.current.account_id), + "${region}", data.aws_region.current.name + ) +} + +output "consumer_invoke_policy_rendered" { + value = local.rendered_invoke_policy + description = "The consumer invoke policy JSON with the live account ID rendered. Distribute this to consumer accounts during onboarding." } \ No newline at end of file diff --git a/tests/test_output_publisher.py b/tests/test_output_publisher.py index cad1385..2c7f4f4 100644 --- a/tests/test_output_publisher.py +++ b/tests/test_output_publisher.py @@ -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() \ No newline at end of file + 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 \ No newline at end of file