test(P04): CAP-037 KMS round-trip + CAP-038 PAT revocation SLO (REQ-350/351, security-engineer)

---ci--- project: acdl phase: 4 milestone: v1.28 status: execute persona: security-engineer ---
2026-08-19 23:11:43 +00:00
parent 14809327fb
commit 7dab9d5756
2 changed files with 211 additions and 0 deletions
@@ -0,0 +1,84 @@
+"""CAP-037 KMS round-trip test (REQ-350).
+
+Sign a test JWT via ``core.kms_signing.sign_jwt()`` (mock KMS with a
+test keypair) → fetch JWKS via ``nova_idp_jwks.lambda_handler()`` (mock
+KMS) → verify the JWT with ``pyjwt`` using the JWKS key. Round-trip
+succeeds — proves the DER→raw conversion + JWK export are mutually
+consistent (the #1 gotcha from RESEARCH §5).
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+os.environ.setdefault("NOVA_LAMBDA_LOCAL_BYPASS", "1")
+
+_JWKS_PATH = Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_jwks.py"
+_spec = importlib.util.spec_from_file_location("nova_idp_jwks_rt", _JWKS_PATH)
+jwks_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(jwks_mod)
+
+import core.kms_signing as kms_signing
+import jwt as pyjwt
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives import hashes, serialization
+
+
+class _MockKms:
+    def __init__(self, priv, pub_der):
+        self._priv = priv
+        self._pub_der = pub_der
+
+    def sign(self, KeyId, Message, MessageType, SigningAlgorithm):
+        return {"Signature": self._priv.sign(Message, ec.ECDSA(hashes.SHA256()))}
+
+    def get_public_key(self, KeyId):
+        return {"PublicKey": self._pub_der}
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    yield
+    kms_signing.set_kms_client_for_testing(None)
+
+
+def test_cap037_kms_roundtrip():
+    """Sign JWT → JWKS → pyjwt verify. The full KMS round-trip (REQ-350)."""
+    priv = ec.generate_private_key(ec.SECP256R1())
+    pub = priv.public_key()
+    pub_der = pub.public_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    kms_signing.set_kms_client_for_testing(_MockKms(priv, pub_der))
+
+    # 1. Sign a JWT via kms_signing.sign_jwt (uses DER→raw conversion).
+    claims = {
+        "sub": "roundtrip-user", "aud": "nova-cli", "iss": "nova-idp",
+        "exp": 9999999999, "iat": 1700000000, "jti": "rt-jti",
+        "roles": ["developer"], "typ": "nova_oidc_token",
+    }
+    token = kms_signing.sign_jwt(claims, key_id="alias/nova-oidc-signing")
+
+    # 2. Fetch the JWKS via the JWKS Lambda (mock KMS get_public_key).
+    resp = jwks_mod.lambda_handler({}, None)
+    assert resp["statusCode"] == 200
+    jwks_body = json.loads(resp["body"])
+    jwk = jwks_body["keys"][0]
+    assert jwk["kty"] == "EC" and jwk["crv"] == "P-256"
+
+    # 3. Verify the JWT with pyjwt using the JWKS key.
+    key = pyjwt.PyJWK(jwk).key
+    decoded = pyjwt.decode(token, key, algorithms=["ES256"], audience="nova-cli")
+    assert decoded["sub"] == "roundtrip-user"
+    assert decoded["jti"] == "rt-jti"
+    assert decoded["roles"] == ["developer"]
+    assert decoded["typ"] == "nova_oidc_token"
@@ -0,0 +1,127 @@
+"""CAP-038 PAT revocation SLO test (REQ-351).
+
+Issue a PAT → vend a token (succeeds) → revoke the PAT → vend a token
+(403, reason ``pat_revoked``). Asserts the denial happens immediately
+(D-229: the strong read on the main table is synchronous — the 60s SLO
+is for propagation, which with strong reads is instant; assert <1s
+locally). Uses moto for DynamoDB.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
+os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
+os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")
+os.environ.setdefault("NOVA_LAMBDA_LOCAL_BYPASS", "1")
+
+_TV_PATH = Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_token_vend.py"
+_spec = importlib.util.spec_from_file_location("nova_idp_token_vend_rev", _TV_PATH)
+tv = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(tv)
+
+import boto3
+from moto import mock_aws
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives import hashes, serialization
+
+import core.kms_signing as kms_signing
+import core.pat_lifecycle as pat_life
+
+
+class _MockKms:
+    def __init__(self, priv, pub_der):
+        self._priv = priv
+        self._pub_der = pub_der
+
+    def sign(self, KeyId, Message, MessageType, SigningAlgorithm):
+        return {"Signature": self._priv.sign(Message, ec.ECDSA(hashes.SHA256()))}
+
+    def get_public_key(self, KeyId):
+        return {"PublicKey": self._pub_der}
+
+
+def _create_pats_table(ddb):
+    ddb.create_table(
+        TableName="nova-pats",
+        KeySchema=[{"AttributeName": "jti", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "jti", "AttributeType": "S"}],
+        BillingMode="PAY_PER_REQUEST",
+    )
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    tv._dynamodb = None
+    pat_life._dynamodb = None
+    yield
+    tv._dynamodb = None
+    pat_life._dynamodb = None
+    kms_signing.set_kms_client_for_testing(None)
+
+
+@mock_aws
+def test_pat_revocation_slo():
+    """Issue → vend (ok) → revoke → vend (403 pat_revoked) in <1s (REQ-351)."""
+    priv = ec.generate_private_key(ec.SECP256R1())
+    pub_der = priv.public_key().public_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    kms_signing.set_kms_client_for_testing(_MockKms(priv, pub_der))
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+
+    # 1. Issue a PAT.
+    pat = pat_life.issue_pat("user-1", ["developer"], "t1", ttl_seconds=3600)
+
+    # 2. Vend a token — succeeds (PAT active + ABAC allow developer+dev).
+    body = {"token": pat, "environment": "dev"}
+    resp1 = tv.lambda_handler({"body": json.dumps(body)}, None)
+    assert resp1["statusCode"] == 200, resp1["body"]
+    assert "token" in json.loads(resp1["body"])
+
+    # 3. Revoke the PAT.
+    import base64
+    payload = json.loads(base64.urlsafe_b64decode(pat.split(".")[1] + "=="))
+    jti = payload["jti"]
+    t0 = time.monotonic()
+    pat_life.revoke_pat(jti)
+
+    # 4. Vend again — 403 pat_revoked, immediately (<1s SLO, D-229 strong read).
+    resp2 = tv.lambda_handler({"body": json.dumps(body)}, None)
+    elapsed = time.monotonic() - t0
+    assert resp2["statusCode"] == 403
+    assert json.loads(resp2["body"])["reason"] == "pat_revoked"
+    assert elapsed < 1.0, f"revocation took {elapsed:.3f}s — expected <1s (D-229 strong read)"
+
+
+@mock_aws
+def test_pat_revocation_then_abac_still_denies():
+    """After revocation, the denial reason is pat_revoked (not abac)."""
+    priv = ec.generate_private_key(ec.SECP256R1())
+    pub_der = priv.public_key().public_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    kms_signing.set_kms_client_for_testing(_MockKms(priv, pub_der))
+    ddb = boto3.client("dynamodb", region_name="us-east-1")
+    _create_pats_table(ddb)
+    pat = pat_life.issue_pat("user-1", ["developer"], "t1", ttl_seconds=3600)
+    import base64
+    payload = json.loads(base64.urlsafe_b64decode(pat.split(".")[1] + "=="))
+    pat_life.revoke_pat(payload["jti"])
+    body = {"token": pat, "environment": "dev"}
+    resp = tv.lambda_handler({"body": json.dumps(body)}, None)
+    assert resp["statusCode"] == 403
+    assert json.loads(resp["body"])["reason"] == "pat_revoked"