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 98ff20e61d
3 changed files with 93 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/lambda/contract_ingestor.py
+43 -11
View File
@@ -34,6 +34,21 @@ _dynamodb = None
_secrets_client = None
def _discover_environments():
"""P10 (REQ-174): derive the valid environment names from
core/environments/*.json (the directory is the single source of truth,
not a hardcoded set). Falls back to {'dev','qa','prod','dr'} if the
directory is not readable (e.g. packaged Lambda without the dir).
"""
env_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(
os.path.abspath(__file__)))), "core", "environments")
try:
names = {f[:-5] for f in os.listdir(env_dir) if f.endswith(".json")}
return names or {"dev", "qa", "prod", "dr"}
except OSError:
return {"dev", "qa", "prod", "dr"}
def _get_dynamodb():
global _dynamodb
if _dynamodb is None:
@@ -236,20 +251,33 @@ def _validate_caller_identity(event, payload):
in the payload matches the principal's ARN-derived source identity, preventing
one consumer from impersonating another.
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).
P10 (REQ-174): if the IAM identity is absent (no callerArn), the function
FAILS CLOSED (raises ValueError) rather than silently passing. The ABAC
policy at the IAM layer is the primary enforcement; this is defense-in-
depth so a misconfigured Function URL (no IAM auth) does not allow
unauthenticated contract submission. Local testing must set a test ARN
via the event requestContext or the LOCAL_LAMBDA_STUB env bypass.
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/nova:owner). This function validates format only, not
ownership.
error length. P10 (REQ-174): the environment enum is derived from the
core/environments/ directory (not hardcoded), so a new env JSON is the
single source of truth. 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/nova:owner). This function validates format only,
not ownership.
"""
identity = event.get("requestContext", {}).get("identity", {})
caller_arn = identity.get("userArn", "")
if not caller_arn:
pass # no identity available — rely on IAM ABAC enforcement
# P10 (REQ-174): fail closed. A local-test bypass is allowed via
# the NOVA_LAMBDA_LOCAL_BYPASS env var (set by the LocalLambdaStub).
import os as _os
if not _os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS"):
raise ValueError(
"missing IAM caller identity (requestContext.identity.userArn) — "
"the Function URL must use IAM auth; refusing unauthenticated submission"
)
payload_repo = payload.get("consumerRepo", "")
if payload_repo:
# consumerRepo must be org/repo format, <=128 chars
@@ -263,12 +291,13 @@ def _validate_caller_identity(event, payload):
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
# P10 (REQ-174): environment enum derived from core/environments/ (not
# hardcoded) — the directory is the single source of truth.
environment = payload.get("environment", "")
if environment:
valid_envs = {"dev", "qa", "prod", "dr"}
valid_envs = _discover_environments()
if environment not in valid_envs:
raise ValueError(f"invalid environment: {environment!r} (must be one of {valid_envs})")
raise ValueError(f"invalid environment: {environment!r} (must be one of {sorted(valid_envs)})")
# v1.14 (REQ-144): error length cap (for report_error action)
error_msg = payload.get("error", "")
@@ -357,6 +386,9 @@ def lambda_handler(event, context):
}
return {"statusCode": 200, "body": json.dumps(result)}
except ValueError as e:
# P10 (REQ-174): identity failures are 401, field validation is 400.
if "missing IAM caller identity" in str(e):
return {"statusCode": 401, "body": json.dumps({"error": str(e)})}
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
except Exception as e: # pragma: no cover - defensive top-level guard
return {"statusCode": 500, "body": json.dumps({"error": str(e)})}
core/local_emulators.py
+12
View File
@@ -392,12 +392,24 @@ class LocalLambdaStub:
"httpContext": {"authorizer": {"iam": {"userId": "local-stub"}}}
},
}
# P10 (REQ-174): the local stub has no real IAM identity; set
# the bypass so the fail-closed identity check passes for local
# tier testing. The ABAC layer is the primary enforcement in
# real AWS; the stub is defense-in-depth-testable via the
# explicit TestCallerIdentityValidation tests.
import os as _os
_prev_bypass = _os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS")
_os.environ["NOVA_LAMBDA_LOCAL_BYPASS"] = "1"
result = ci.lambda_handler(event, None)
finally:
ci._get_dynamodb = original_get
if original_urlopen is not None:
import urllib.request
urllib.request.urlopen = original_urlopen
if _prev_bypass is None:
_os.environ.pop("NOVA_LAMBDA_LOCAL_BYPASS", None)
else:
_os.environ["NOVA_LAMBDA_LOCAL_BYPASS"] = _prev_bypass
return result
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