feat(P02): JWS-from-PAT key derivation via HKDF-SHA256 (REQ-332, C-5.2, security-engineer)
---ci---
project: acdl
phase: 2
milestone: v1.28
status: execute
persona: security-engineer
---
C-5.2 grill fix: symmetric JWS (HS256) where the PAT is the shared secret.
derive_signing_key(pat) -> HKDF-SHA256(pat.encode(), salt=b'nova-local-
attestation', info=b'jws-signing-key', length=32) via cryptography (fallback
to hashlib HKDF). sign_attestation(payload, pat) -> compact JWS
b64url(header).b64url(payload).b64url(sig) with header {alg:HS256,typ:JWT}.
verify_attestation(jws, pat) -> payload (raises JWSValidationError on tamper
or wrong PAT; hmac.compare_digest constant-time). INV-14..17 enforced
(key derived from PAT, not cached, fixed salt/info, constant-time compare).
tests/test_jws_attestation.py: 20 tests (round-trip, tamper, wrong-PAT,
invariants, hashlib/crypto parity).
This commit is contained in:
Jon Chery
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user