continuous-intelligence/acdl
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 128 Wiki Activity

verify(P10): contract-ingestor-defense-in-depth — 4-layer verify PASS + ship

Browse Source 
VERIFY: structural — fail-closed + env discovery; behavioral — 49 tests + CI PASS; security — defense-in-depth on IAM identity.

---ci---
project: acdl
phase: 10
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-174]
  partial: []
---/ci---
This commit is contained in:
Jon Chery
2026-08-01 12:58:23 +00:00
parent f12f6edd23
commit 76714bebc4
3 changed files with 93 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_contract_ingestor.py
+38 -10
View File
@@ -37,6 +37,15 @@ _spec.loader.exec_module(ingestor)
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _local_lambda_bypass(monkeypatch):
"""P10 (REQ-174): set NOVA_LAMBDA_LOCAL_BYPASS for all ingestor tests
so the fail-closed identity check doesn't block handler-routing tests.
Tests that explicitly exercise the identity check (TestCallerIdentity
Validation) override this per-test."""
monkeypatch.setenv("NOVA_LAMBDA_LOCAL_BYPASS", "1")
@pytest.fixture
def sample_payload():
return {
@@ -50,7 +59,8 @@ def sample_payload():
@pytest.fixture
def function_url_event(sample_payload):
return {"body": json.dumps(sample_payload)}
# P10 (REQ-174): include a test IAM identity so the fail-closed check passes.
return {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/nova-deploy/test"}}}
@pytest.fixture
@@ -361,9 +371,21 @@ class TestLambdaHandler:
class TestCallerIdentityValidation:
"""P1-2: the Lambda validates consumerRepo against the invoking principal."""
def test_no_identity_skips_check(self, moto_contracts_table, function_url_event):
# No requestContext.identity in the event — check is skipped (relies on IAM ABAC).
resp = ingestor.lambda_handler(function_url_event, None)
def test_no_identity_fails_closed(self, moto_contracts_table, sample_payload, monkeypatch):
# P10 (REQ-174): no requestContext.identity → fail closed (defense-in-
# depth). The old behavior (silent pass) is replaced with a 401.
monkeypatch.delenv("NOVA_LAMBDA_LOCAL_BYPASS", raising=False)
event = {"body": json.dumps(sample_payload), "requestContext": {}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 401
assert "missing IAM caller identity" in json.loads(resp["body"])["error"]
def test_no_identity_passes_with_local_bypass(self, moto_contracts_table, sample_payload, monkeypatch):
# P10 (REQ-174): the NOVA_LAMBDA_LOCAL_BYPASS env allows local/stub
# testing without an IAM identity (the LocalLambdaStub sets it).
monkeypatch.setenv("NOVA_LAMBDA_LOCAL_BYPASS", "1")
event = {"body": json.dumps(sample_payload), "requestContext": {}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200
def test_invalid_consumer_repo_format_rejected(self, moto_contracts_table, sample_payload):
@@ -496,7 +518,7 @@ class TestValidateChangeRequest:
"action": "validate_change_request",
"changeRequestId": "CHG0678912",
"consumerRepo": "acdl/consumer-a",
})}
}), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/nova-deploy/test"}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200
body = json.loads(resp["body"])
@@ -505,33 +527,39 @@ class TestValidateChangeRequest:
class TestV14IdentityValidation:
"""v1.14 (REQ-144): contractId format, environment enum, error length
validation + spoofing resistance."""
validation + spoofing resistance.
P10 (REQ-174): these tests supply a valid userArn so the fail-closed
identity check passes and the field validation is reached."""
_ARN = "arn:aws:sts::000:assumed-role/nova-deploy/test-session"
def test_invalid_contract_id_rejected(self, moto_contracts_table, sample_payload):
sample_payload["contractId"] = "bad contract!@#"
event = {"body": json.dumps(sample_payload), "requestContext": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": self._ARN}}}
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": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": self._ARN}}}
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": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": self._ARN}}}
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):
arn = "arn:aws:sts::000:assumed-role/nova-deploy/test"
for env in ["dev", "qa", "prod", "dr"]:
sample_payload["environment"] = env
event = {"body": json.dumps(sample_payload), "requestContext": {}}
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": arn}}}
resp = ingestor.lambda_handler(event, None)
assert resp["statusCode"] == 200