diff --git a/core/lambda/contract_ingestor.py b/core/lambda/contract_ingestor.py index 5132d70..0b975e8 100644 --- a/core/lambda/contract_ingestor.py +++ b/core/lambda/contract_ingestor.py @@ -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)})} \ No newline at end of file diff --git a/core/local_emulators.py b/core/local_emulators.py index 13b3745..805c586 100644 --- a/core/local_emulators.py +++ b/core/local_emulators.py @@ -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 diff --git a/tests/test_contract_ingestor.py b/tests/test_contract_ingestor.py index ec95ec2..cc435c6 100644 --- a/tests/test_contract_ingestor.py +++ b/tests/test_contract_ingestor.py @@ -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