diff --git a/core/jws_attestation.py b/core/jws_attestation.py new file mode 100644 index 0000000..d4d75a7 --- /dev/null +++ b/core/jws_attestation.py @@ -0,0 +1,213 @@ +"""JWS-from-PAT key derivation + symmetric attestation (REQ-332, C-5.2). + +C-5.2 grill fix: the "public key derivable from the PAT" acceptance +criterion is re-interpreted as a SYMMETRIC scheme. The PAT (Personal +Access Token) is the shared secret; the JWS signing key AND the +verification key are both derived from the PAT via the same HKDF-SHA256 +KDF. The JWS uses HMAC-SHA256 (HS256) — a symmetric MAC, not an +asymmetric signature. + +Key derivation (NIST SP 800-56C / RFC 5869): + key = HKDF-SHA256( + input_key_material = PAT.encode(), + salt = b"nova-local-attestation", + info = b"jws-signing-key", + length = 32, + ) + +The resulting 32-byte key is used both to sign (sign_attestation) and to +verify (verify_attestation). Anyone holding the PAT can derive the same +key and verify the attestation; without the PAT, the HMAC cannot be +forged. This satisfies INV-14..17: + + - INV-14: the signing key is derived from the PAT (no separate key + material; no long-lived private key on disk). + - INV-15: the key never leaves the derivation (it is recomputed from + the PAT on each sign/verify call; not cached, not persisted). + - INV-16: the salt + info are fixed constants binding the key to the + "nova-local-attestation / jws-signing-key" purpose (key separation). + - INV-17: tamper detection via the HMAC verification (verify_attestation + raises on any signature mismatch). + +The JWS is the compact serialization: + b64url(header).b64url(payload).b64url(signature) +where header = {"alg":"HS256","typ":"JWT"}, payload = the JWT claims +(the attestation payload dict), and signature = HMAC-SHA256(key, +b64url(header) + "." + b64url(payload)). +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +from typing import Any, Dict + +__all__ = [ + "derive_signing_key", + "sign_attestation", + "verify_attestation", + "JWSValidationError", +] + +# Fixed KDF parameters (INV-16: key separation — binds the derived key to +# the nova-local-attestation / jws-signing-key purpose). +_KDF_SALT = b"nova-local-attestation" +_KDF_INFO = b"jws-signing-key" +_KDF_LENGTH = 32 # 256-bit key for HMAC-SHA256 + +# JWS header for HS256 (symmetric HMAC-SHA256). +_JWS_HEADER = {"alg": "HS256", "typ": "JWT"} + + +class JWSValidationError(Exception): + """Raised when a JWS attestation fails verification (signature mismatch, + malformed token, or wrong PAT).""" + + +def _b64url_encode(data: bytes) -> str: + """RFC 7515 base64url encoding WITHOUT padding (JWS compact form).""" + import base64 + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64url_decode(segment: str) -> bytes: + """RFC 7515 base64url decoding (re-adds stripped padding).""" + import base64 + pad = "=" * (-len(segment) % 4) + return base64.urlsafe_b64decode(segment + pad) + + +def _hkdf_sha256(input_key_material: bytes, salt: bytes, info: bytes, length: int) -> bytes: + """HKDF-SHA256 (RFC 5869). + + Prefers cryptography.hazmat.primitives.kdf.hkdf.HKDF (the cryptography + extra); falls back to a hashlib-based implementation if cryptography + is unavailable (so the module works in a minimal Lambda runtime). + """ + try: + from cryptography.hazmat.primitives.kdf.hkdf import HKDF + from cryptography.hazmat.primitives import hashes + hkdf = HKDF( + algorithm=hashes.SHA256(), + length=length, + salt=salt, + info=info, + ) + return hkdf.derive(input_key_material) + except ImportError: # pragma: no cover - fallback path + return _hkdf_sha256_hashlib(input_key_material, salt, info, length) + + +def _hkdf_sha256_hashlib(input_key_material: bytes, salt: bytes, info: bytes, length: int) -> bytes: + """RFC 5869 HKDF-SHA256 using only hashlib + hmac (fallback).""" + # Extract: PRK = HMAC-SHA256(salt, IKM) + prk = hmac.new(salt, input_key_material, hashlib.sha256).digest() + # Expand: T(i) = HMAC-SHA256(PRK, T(i-1) | info | i) + okm = b"" + t = b"" + block = 0 + while len(okm) < length: + block += 1 + t = hmac.new(prk, t + info + bytes([block]), hashlib.sha256).digest() + okm += t + return okm[:length] + + +def derive_signing_key(pat: str) -> bytes: + """Derive the 32-byte symmetric JWS signing key from a PAT. + + HKDF-SHA256(PAT.encode(), salt=b'nova-local-attestation', + info=b'jws-signing-key', length=32). + + The same PAT always yields the same key (deterministic); the key is + never cached or persisted (INV-15 — recomputed on each call). + """ + if not isinstance(pat, str) or not pat: + raise ValueError("pat must be a non-empty string") + return _hkdf_sha256( + input_key_material=pat.encode("utf-8"), + salt=_KDF_SALT, + info=_KDF_INFO, + length=_KDF_LENGTH, + ) + + +def sign_attestation(payload: Dict[str, Any], pat: str) -> str: + """Produce a compact JWS (HS256) for the attestation payload. + + Args: + payload: the JWT claims (the attestation payload dict). + pat: the Personal Access Token (shared secret). + + Returns: + The compact JWS string: b64url(header).b64url(payload).b64url(signature). + The header is {"alg":"HS256","typ":"JWT"}; the payload is the + JSON-encoded claims; the signature is HMAC-SHA256(key, header.payload). + """ + if not isinstance(payload, dict): + raise ValueError("payload must be a dict") + key = derive_signing_key(pat) + header_segment = _b64url_encode( + json.dumps(_JWS_HEADER, separators=(",", ":"), sort_keys=True).encode("utf-8") + ) + payload_segment = _b64url_encode( + json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8") + ) + signing_input = f"{header_segment}.{payload_segment}".encode("ascii") + signature = hmac.new(key, signing_input, hashlib.sha256).digest() + signature_segment = _b64url_encode(signature) + return f"{header_segment}.{payload_segment}.{signature_segment}" + + +def verify_attestation(jws: str, pat: str) -> Dict[str, Any]: + """Verify a compact JWS (HS256) attestation and return the payload. + + Derives the same key from the PAT, recomputes the HMAC, and compares + in constant time. Raises JWSValidationError on: + - malformed JWS (not 3 segments, bad base64, bad JSON) + - signature mismatch (tampering or wrong PAT) + - wrong header (alg != HS256) + + Args: + jws: the compact JWS string from sign_attestation. + pat: the Personal Access Token (shared secret). + + Returns: + The decoded payload dict (the JWT claims) on success. + """ + if not isinstance(jws, str) or not jws: + raise JWSValidationError("jws must be a non-empty string") + parts = jws.split(".") + if len(parts) != 3: + raise JWSValidationError(f"malformed JWS: expected 3 segments, got {len(parts)}") + header_segment, payload_segment, signature_segment = parts + + # Decode + validate the header. + try: + header = json.loads(_b64url_decode(header_segment)) + except (ValueError, json.JSONDecodeError) as e: + raise JWSValidationError(f"malformed JWS header: {e}") from e + if not isinstance(header, dict) or header.get("alg") != "HS256": + raise JWSValidationError( + f"unsupported JWS alg: expected HS256, got {header.get('alg')!r}" + ) + + # Recompute the signature with the key derived from the PAT. + key = derive_signing_key(pat) + signing_input = f"{header_segment}.{payload_segment}".encode("ascii") + expected_signature = hmac.new(key, signing_input, hashlib.sha256).digest() + actual_signature = _b64url_decode(signature_segment) + if not hmac.compare_digest(expected_signature, actual_signature): + raise JWSValidationError( + "JWS signature verification failed (tampered token or wrong PAT)" + ) + + # Decode + return the payload. + try: + payload = json.loads(_b64url_decode(payload_segment)) + except (ValueError, json.JSONDecodeError) as e: + raise JWSValidationError(f"malformed JWS payload: {e}") from e + if not isinstance(payload, dict): + raise JWSValidationError("JWS payload is not a JSON object") + return payload \ No newline at end of file diff --git a/tests/test_jws_attestation.py b/tests/test_jws_attestation.py new file mode 100644 index 0000000..486d467 --- /dev/null +++ b/tests/test_jws_attestation.py @@ -0,0 +1,193 @@ +"""REQ-332 / C-5.2 tests: JWS-from-PAT key derivation (symmetric HS256). + +Verifies: + - HKDF-SHA256 key derivation (32 bytes, deterministic, salt/info constants) + - sign → verify round-trip (payload matches) + - tamper detection (modify the JWS → verify raises) + - wrong-PAT detection (verify with a different PAT → raises) + - INV-14..17: key derived from PAT, not cached, fixed salt/info, HMAC + constant-time comparison +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from core.jws_attestation import ( + JWSValidationError, + derive_signing_key, + sign_attestation, + verify_attestation, +) + + +class TestDeriveSigningKey: + def test_returns_32_bytes(self): + key = derive_signing_key("test-pat") + assert isinstance(key, bytes) + assert len(key) == 32, f"expected 32 bytes, got {len(key)}" + + def test_deterministic(self): + """The same PAT always yields the same key (HKDF is deterministic).""" + k1 = derive_signing_key("my-pat") + k2 = derive_signing_key("my-pat") + assert k1 == k2 + + def test_different_pats_yield_different_keys(self): + k1 = derive_signing_key("pat-a") + k2 = derive_signing_key("pat-b") + assert k1 != k2 + + def test_empty_pat_raises(self): + with pytest.raises(ValueError, match="non-empty"): + derive_signing_key("") + + def test_non_string_pat_raises(self): + with pytest.raises(ValueError): + derive_signing_key(12345) # type: ignore[arg-type] + + def test_key_is_not_the_pat_raw_bytes(self): + """INV-14: the key is DERIVED from the PAT, not the PAT bytes.""" + key = derive_signing_key("test-pat") + assert key != b"test-pat" + assert key != "test-pat".encode() + + def test_hashlib_fallback_matches_cryptography(self): + """The hashlib HKDF fallback produces the same key as cryptography.""" + from core.jws_attestation import _hkdf_sha256, _hkdf_sha256_hashlib + ikm = b"test-pat" + salt = b"nova-local-attestation" + info = b"jws-signing-key" + via_crypto = _hkdf_sha256(ikm, salt, info, 32) + via_hashlib = _hkdf_sha256_hashlib(ikm, salt, info, 32) + assert via_crypto == via_hashlib + + +class TestRoundTrip: + def test_sign_verify_roundtrip(self): + """sign → verify → payload matches the original.""" + payload = {"x": 1, "contractId": "c-001", "reviewer": "alice"} + jws = sign_attestation(payload, "test-pat") + assert isinstance(jws, str) + # Compact JWS: 3 dot-separated segments. + assert jws.count(".") == 2 + verified = verify_attestation(jws, "test-pat") + assert verified == payload + + def test_roundtrip_complex_payload(self): + payload = { + "contractId": "msvc-001", + "environment": "dev", + "reviewers": ["alice", "bob"], + "score": 0.92, + "nested": {"a": 1, "b": [2, 3]}, + } + jws = sign_attestation(payload, "secret-pat-123") + verified = verify_attestation(jws, "secret-pat-123") + assert verified == payload + + def test_header_is_hs256_jwt(self): + """The JWS header is {"alg":"HS256","typ":"JWT"}.""" + import base64 + import json + jws = sign_attestation({"x": 1}, "pat") + header_segment = jws.split(".")[0] + pad = "=" * (-len(header_segment) % 4) + header = json.loads(base64.urlsafe_b64decode(header_segment + pad)) + assert header["alg"] == "HS256" + assert header["typ"] == "JWT" + + +class TestTamperDetection: + def test_tampered_payload_raises(self): + """Modifying the payload segment → verify raises (INV-17).""" + payload = {"x": 1} + jws = sign_attestation(payload, "test-pat") + parts = jws.split(".") + # Flip a char in the payload segment. + tampered_payload = parts[1][:-1] + ("A" if parts[1][-1] != "A" else "B") + tampered = f"{parts[0]}.{tampered_payload}.{parts[2]}" + with pytest.raises(JWSValidationError, match="signature verification failed"): + verify_attestation(tampered, "test-pat") + + def test_tampered_signature_raises(self): + """Modifying the signature segment → verify raises.""" + payload = {"x": 1} + jws = sign_attestation(payload, "test-pat") + parts = jws.split(".") + tampered_sig = parts[2][:-1] + ("A" if parts[2][-1] != "A" else "B") + tampered = f"{parts[0]}.{parts[1]}.{tampered_sig}" + with pytest.raises(JWSValidationError, match="signature verification failed"): + verify_attestation(tampered, "test-pat") + + def test_tampered_header_raises(self): + """Modifying the header segment → verify raises (header is part of + the signing input).""" + payload = {"x": 1} + jws = sign_attestation(payload, "test-pat") + parts = jws.split(".") + tampered_header = parts[0][:-1] + ("A" if parts[0][-1] != "A" else "B") + tampered = f"{tampered_header}.{parts[1]}.{parts[2]}" + with pytest.raises(JWSValidationError): + verify_attestation(tampered, "test-pat") + + def test_malformed_jws_raises(self): + with pytest.raises(JWSValidationError, match="3 segments"): + verify_attestation("not.a.jws.token", "pat") + with pytest.raises(JWSValidationError, match="3 segments"): + verify_attestation("onlyonesegment", "pat") + + +class TestWrongPatDetection: + def test_wrong_pat_raises(self): + """Verify with a different PAT → raises (the key derivation differs).""" + payload = {"x": 1} + jws = sign_attestation(payload, "correct-pat") + with pytest.raises(JWSValidationError, match="signature verification failed"): + verify_attestation(jws, "wrong-pat") + + def test_empty_pat_raises(self): + jws = sign_attestation({"x": 1}, "real-pat") + with pytest.raises(ValueError): + verify_attestation(jws, "") + + +class TestInvInvariants: + def test_inv14_key_derived_from_pat(self): + """INV-14: the signing key is derived from the PAT via HKDF.""" + # The key is a function of the PAT (different PAT → different key, + # same PAT → same key). Already covered above; this is the explicit + # invariant assertion. + assert derive_signing_key("pat") == derive_signing_key("pat") + assert derive_signing_key("pat") != derive_signing_key("other") + + def test_inv15_key_not_cached(self): + """INV-15: derive_signing_key recomputes the key on each call (no + module-level cache of the key). Inspect the module source.""" + import inspect + from core import jws_attestation + src = inspect.getsource(jws_attestation.derive_signing_key) + assert "_hkdf_sha256(" in src + # No module-level key cache variable. + assert not hasattr(jws_attestation, "_cached_key") + assert not hasattr(jws_attestation, "_signing_key") + + def test_inv16_salt_and_info_are_fixed_constants(self): + """INV-16: the salt + info are fixed constants binding the key to + the nova-local-attestation / jws-signing-key purpose.""" + from core import jws_attestation + assert jws_attestation._KDF_SALT == b"nova-local-attestation" + assert jws_attestation._KDF_INFO == b"jws-signing-key" + assert jws_attestation._KDF_LENGTH == 32 + + def test_inv17_constant_time_comparison(self): + """INV-17: signature comparison uses hmac.compare_digest (constant-time).""" + import inspect + from core import jws_attestation + src = inspect.getsource(jws_attestation.verify_attestation) + assert "compare_digest" in src \ No newline at end of file