docs(P11): complete schema-input-validation-hardening phase (v1.13.14)

---ci---
project: acdl
phase: 11
milestone: v1.14
status: complete
requirements:
  covered: [REQ-145]
  partial: []
---/ci---

core/lambda/contract_ingestor.py

@@ -238,21 +238,42 @@ def _validate_caller_identity(event, payload):
 
     If the identity is not available (e.g. local testing or non-IAM auth), the
     check is skipped (the ABAC policy at the IAM layer enforces the scope).
+
+    v1.14 (REQ-144): also validates contractId format, environment enum, and
+    error length. The ABAC reliance is documented here: the Function URL IAM
+    identity does not expose principal tags in the event, so full enforcement
+    of consumerRepo ownership is at the IAM layer (ABAC via
+    aws:PrincipalTag/acdl:owner). This function validates format only, not
+    ownership.
     """
     identity = event.get("requestContext", {}).get("identity", {})
     caller_arn = identity.get("userArn", "")
     if not caller_arn:
-        return  # no identity available — rely on IAM ABAC enforcement
+        pass  # no identity available — rely on IAM ABAC enforcement
     payload_repo = payload.get("consumerRepo", "")
-    if not payload_repo:
-        return
-    # Extract the session name or principal tag from the ARN. The ABAC policy
-    # scopes via aws:PrincipalTag/acdl:owner = <consumerRepo>. The Function URL
-    # IAM identity does not expose principal tags in the event, so we do a
-    # best-effort check: the consumerRepo must not be empty and must be a valid
-    # repo identifier (org/repo format). Full enforcement is at the IAM layer.
-    if "/" not in payload_repo or len(payload_repo) > 128:
-        raise ValueError(f"invalid consumerRepo format: {payload_repo!r}")
+    if payload_repo:
+        # consumerRepo must be org/repo format, <=128 chars
+        if "/" not in payload_repo or len(payload_repo) > 128:
+            raise ValueError(f"invalid consumerRepo format: {payload_repo!r}")
+
+    # v1.14 (REQ-144): contractId format validation
+    contract_id = payload.get("contractId", "")
+    if contract_id:
+        import re
+        if not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$', contract_id):
+            raise ValueError(f"invalid contractId format: {contract_id!r} (alphanumeric, hyphen, underscore; max 64 chars)")
+
+    # v1.14 (REQ-144): environment enum validation
+    environment = payload.get("environment", "")
+    if environment:
+        valid_envs = {"dev", "qa", "prod", "dr"}
+        if environment not in valid_envs:
+            raise ValueError(f"invalid environment: {environment!r} (must be one of {valid_envs})")
+
+    # v1.14 (REQ-144): error length cap (for report_error action)
+    error_msg = payload.get("error", "")
+    if error_msg and len(str(error_msg)) > 10000:
+        payload["error"] = str(error_msg)[:10000]
 
 
 def _validate_change_request(payload):

modules/l2/microservice/composition.json

@@ -18,7 +18,7 @@
   "wires": [
     {"from": "contract.inputs.name", "to": "alb.inputs.name", "default": "app"},
     {"from": "contract.inputs.name", "to": "ecr.inputs.name", "default": "app-repo"},
-    {"from": "contract.inputs.name", "to": "roles.inputs.role_name", "default": "app-role"},
+    {"from": "contract.inputs.name", "to": "roles.inputs.role_name", "default": "acdl-app-role"},
     {"from": "contract.inputs.region", "to": "cluster.inputs.region"},
     {"from": "contract.inputs.region", "to": "ecr.inputs.region"},
     {"from": "contract.inputs.region", "to": "roles.inputs.region"},

schemas/environment.schema.json

@@ -27,20 +27,23 @@
       "type": "object",
       "required": ["bucket", "lock_table"],
       "properties": {
-        "bucket": {"type": "string", "description": "S3 state bucket name."},
+        "bucket": {"type": "string", "pattern": "^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$", "description": "S3 state bucket name (lowercase, 3-63 chars, dots/hyphens)."},
         "lock_table": {"type": "string", "description": "DynamoDB lock table name."}
-      }
+      },
+      "additionalProperties": false
     },
     "network": {
       "type": "object",
       "required": ["vpc_cidr", "azs"],
       "properties": {
-        "vpc_cidr": {"type": "string", "description": "VPC CIDR block."},
-        "azs": {"type": "array", "items": {"type": "string"}, "description": "Availability zones."}
-      }
+        "vpc_cidr": {"type": "string", "pattern": "^[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}/[0-9]{1,2}$", "description": "VPC CIDR block (e.g. 10.0.0.0/16)."},
+        "azs": {"type": "array", "items": {"type": "string"}, "maxItems": 6, "description": "Availability zones (max 6)."}
+      },
+      "additionalProperties": false
     },
     "runner_role_arn": {
       "type": "string",
+      "pattern": "^arn:aws:iam::[0-9]{12}:role/.+$",
       "description": "The IAM role ARN surfaced to the consumer's repo via ABAC."
     },
     "autonomy": {
@@ -54,5 +57,6 @@
       "maximum": 1,
       "description": "The confidence gate threshold for this environment (dev 0.50, qa 0.75, prod 0.90, dr 0.95)."
     }
-  }
+  },
+  "additionalProperties": false
 }

terraform/bootstrap/spike_runner_policy.json

@@ -215,7 +215,10 @@
         "kms:TagResource",
         "kms:UntagResource"
       ],
-      "Resource": "*"
+      "Resource": [
+        "arn:aws:kms:*:*:key/*",
+        "arn:aws:kms:*:*:alias/acdl-*"
+      ]
     },
     {
       "Effect": "Allow",
@@ -233,7 +236,7 @@
         "iam:TagRole",
         "iam:UntagRole"
       ],
-      "Resource": "*"
+      "Resource": "arn:aws:iam::*:role/acdl-*"
     }
   ]
 }

tests/test_contract_ingestor.py

@@ -500,4 +500,43 @@ class TestValidateChangeRequest:
         resp = ingestor.lambda_handler(event, None)
         assert resp["statusCode"] == 200
         body = json.loads(resp["body"])
-        assert body["action"] == "validate_change_request"
+        assert body["action"] == "validate_change_request"
+
+
+class TestV14IdentityValidation:
+    """v1.14 (REQ-144): contractId format, environment enum, error length
+    validation + spoofing resistance."""
+
+    def test_invalid_contract_id_rejected(self, moto_contracts_table, sample_payload):
+        sample_payload["contractId"] = "bad contract!@#"
+        event = {"body": json.dumps(sample_payload), "requestContext": {}}
+        resp = ingestor.lambda_handler(event, None)
+        assert resp["statusCode"] == 400
+        assert "invalid contractId" in resp["body"]
+
+    def test_contract_id_too_long_rejected(self, moto_contracts_table, sample_payload):
+        sample_payload["contractId"] = "a" * 65
+        event = {"body": json.dumps(sample_payload), "requestContext": {}}
+        resp = ingestor.lambda_handler(event, None)
+        assert resp["statusCode"] == 400
+        assert "invalid contractId" in resp["body"]
+
+    def test_invalid_environment_rejected(self, moto_contracts_table, sample_payload):
+        sample_payload["environment"] = "staging"
+        event = {"body": json.dumps(sample_payload), "requestContext": {}}
+        resp = ingestor.lambda_handler(event, None)
+        assert resp["statusCode"] == 400
+        assert "invalid environment" in resp["body"]
+
+    def test_valid_environments_accepted(self, moto_contracts_table, sample_payload):
+        for env in ["dev", "qa", "prod", "dr"]:
+            sample_payload["environment"] = env
+            event = {"body": json.dumps(sample_payload), "requestContext": {}}
+            resp = ingestor.lambda_handler(event, None)
+            assert resp["statusCode"] == 200
+
+    def test_abac_reliance_documented(self):
+        """The _validate_caller_identity docstring documents the ABAC reliance."""
+        docstring = ingestor._validate_caller_identity.__doc__
+        assert "ABAC" in docstring
+        assert "PrincipalTag" in docstring

tests/test_environment_schema.py

@@ -96,4 +96,64 @@ def test_account_id_is_12_digits():
     for env_file in ENV_FILES:
         env = json.loads((ENV_DIR / env_file).read_text())
         assert len(env["account_id"]) == 12
-        assert env["account_id"].isdigit()
+        assert env["account_id"].isdigit()
+
+
+def test_v14_schema_rejects_undocumented_fields():
+    """v1.14 (REQ-145): additionalProperties: false rejects unknown fields."""
+    schema = json.loads(SCHEMA.read_text())
+    bad_env = {
+        "name": "dev",
+        "account_id": "123456789012",
+        "region": "us-east-1",
+        "state_backend": {"bucket": "test", "lock_table": "test"},
+        "network": {"vpc_cidr": "10.0.0.0/16", "azs": ["us-east-1a"]},
+        "runner_role_arn": "arn:aws:iam::123456789012:role/test",
+        "autonomy": "full",
+        "confidence_threshold": 0.5,
+        "rogue_field": "should be rejected"
+    }
+    with pytest.raises(jsonschema.ValidationError, match="Additional properties are not allowed"):
+        jsonschema.validate(bad_env, schema)
+
+
+def test_v14_schema_validates_bucket_name_format():
+    """v1.14 (REQ-145): state_backend.bucket must match S3 naming rules."""
+    schema = json.loads(SCHEMA.read_text())
+    bad_env = {
+        "name": "dev", "account_id": "123456789012", "region": "us-east-1",
+        "state_backend": {"bucket": "Invalid_Bucket!", "lock_table": "test"},
+        "network": {"vpc_cidr": "10.0.0.0/16", "azs": ["us-east-1a"]},
+        "runner_role_arn": "arn:aws:iam::123456789012:role/test",
+        "autonomy": "full", "confidence_threshold": 0.5
+    }
+    with pytest.raises(jsonschema.ValidationError, match="does not match"):
+        jsonschema.validate(bad_env, schema)
+
+
+def test_v14_schema_validates_arn_format():
+    """v1.14 (REQ-145): runner_role_arn must match ARN format."""
+    schema = json.loads(SCHEMA.read_text())
+    bad_env = {
+        "name": "dev", "account_id": "123456789012", "region": "us-east-1",
+        "state_backend": {"bucket": "test", "lock_table": "test"},
+        "network": {"vpc_cidr": "10.0.0.0/16", "azs": ["us-east-1a"]},
+        "runner_role_arn": "not-an-arn",
+        "autonomy": "full", "confidence_threshold": 0.5
+    }
+    with pytest.raises(jsonschema.ValidationError, match="does not match"):
+        jsonschema.validate(bad_env, schema)
+
+
+def test_v14_schema_validates_cidr_format():
+    """v1.14 (REQ-145): vpc_cidr must match CIDR format."""
+    schema = json.loads(SCHEMA.read_text())
+    bad_env = {
+        "name": "dev", "account_id": "123456789012", "region": "us-east-1",
+        "state_backend": {"bucket": "test", "lock_table": "test"},
+        "network": {"vpc_cidr": "not-a-cidr", "azs": ["us-east-1a"]},
+        "runner_role_arn": "arn:aws:iam::123456789012:role/test",
+        "autonomy": "full", "confidence_threshold": 0.5
+    }
+    with pytest.raises(jsonschema.ValidationError, match="does not match"):
+        jsonschema.validate(bad_env, schema)

tests/test_iam_policy_baseline.py

@@ -176,4 +176,43 @@ class TestIAMPolicyBaseline:
                 res = s.get("Resource", "")
                 if isinstance(res, list):
                     res = " ".join(res)
-                assert res != "*", "iam:PassRole must not be granted to Resource: *"
+                assert res != "*", "iam:PassRole must not be granted to Resource: *"
+
+    def test_iam_role_creation_scoped_to_acdl_prefix(self, policy):
+        """G-104: iam:CreateRole must be scoped to role/acdl-* (not Resource: *)."""
+        for s in policy["Statement"]:
+            acts = s.get("Action", [])
+            if isinstance(acts, str):
+                acts = [acts]
+            if "iam:CreateRole" in acts:
+                res = s.get("Resource", "")
+                if isinstance(res, list):
+                    res = " ".join(res)
+                assert "acdl-*" in res, f"iam:CreateRole must be scoped to acdl-* (got: {res})"
+
+    def test_kms_scoped_to_acdl_alias(self, policy):
+        """G-104: kms:CreateKey etc. must be scoped to alias/acdl-* (not Resource: *)."""
+        for s in policy["Statement"]:
+            acts = s.get("Action", [])
+            if isinstance(acts, str):
+                acts = [acts]
+            if any(a.startswith("kms:") for a in acts):
+                res = s.get("Resource", "")
+                if isinstance(res, list):
+                    res = " ".join(res)
+                assert "acdl-*" in res, f"kms actions must be scoped to acdl-* (got: {res})"
+
+    def test_cloudfront_waf_remain_global(self, policy):
+        """G-104: CloudFront + WAFv2 (CloudFront scope) ARNs are global;
+        Resource: * is acceptable here (documented constraint, not a defect)."""
+        global_actions = {"cloudfront:", "wafv2:"}
+        for s in policy["Statement"]:
+            acts = s.get("Action", [])
+            if isinstance(acts, str):
+                acts = [acts]
+            if any(any(a.startswith(g) for g in global_actions) for a in acts):
+                res = s.get("Resource", "")
+                if isinstance(res, list):
+                    res = res[0] if res else ""
+                # CloudFront/WAFv2 are allowed to be * (global ARNs)
+                assert res == "*" or "acdl" in res