diff --git a/docs/protocol.md b/docs/protocol.md index 044d94d..f47e2a5 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -40,15 +40,16 @@ held. ML-KEM uses FIPS 203 implicit rejection: decapsulation never fails, and a wrong key yields garbage caught by the kdf tags. X25519 has no such -property, so low-order (non-contributory) public keys and capsules are -rejected explicitly. +property, so low-order (non-contributory) public keys are rejected +explicitly. -An encapsulation's output is a hybrid CAPSULE: an ML-KEM ciphertext || -a fresh X25519 ephemeral public key (the X25519 "encapsulation" is a -DH, and its public half rides inside the capsule). The only ephemeral -public key shipped separately is alice's e_alice in msg1 -- bob must -receive it before he can encapsulate to it; every other leg is an -encapsulation and brings its own ephemeral along. +Each party ships ONE X25519 ephemeral public per connection: alice's +inside e_alice in msg1, bob's as a bare public e_bob in msg2. Every +X25519 leg is then a plain DH against those shipped keys, and ML-KEM +legs ship as bare ML-KEM ciphertexts. Reusing one ephemeral per party +also binds related secrets to the same sender -- ee and se provably +come from the same bob, which per-capsule ephemerals cannot show +(ML-KEM alone cannot provide it; it rides on the AEAD transcript). ## Transcript chaining @@ -61,19 +62,20 @@ value), which is bound into msg1. ## Registration -First connection. Every leg is the same hybrid KEM operation: ee_cap -gives forward secrecy, es_cap and se_cap prove the two statics. The -statics pass late and hidden: s_bob inside msg2's inner box, s_alice -inside msg3's. Each box key lags one secret behind the proof inside it, -so invalid data inside an authentic box burns the credential, while an -AEAD failure proves no key knowledge and changes nothing: +First connection. Every leg is a bare ML-KEM ciphertext plus an X25519 +DH against the shipped connection ephemerals: ee for forward secrecy, +es and se proving the two statics. The statics pass late and hidden: +s_bob inside msg2's inner box, s_alice inside msg3's. Each box key lags +one secret behind the proof inside it, so invalid data inside an +authentic box burns the credential, while an AEAD failure proves no key +knowledge and changes nothing: msg1 A->B: box(e_alice | c) k1 = kdf(reg/msg1, shared) - msg2 B->A: box(ee_cap | box'(s_bob)) k2 = kdf(reg/msg2, shared) + msg2 B->A: box(e_bob | ee_ct | box'(s_bob)) k2 = kdf(reg/msg2, shared) k' = kdf(reg/msg2/inner, ee) - msg3 A->B: box(es_cap | box'(s_alice | r)) k3 = kdf(reg/msg3, shared, ee) + msg3 A->B: box(es_ct | box'(s_alice | r)) k3 = kdf(reg/msg3, shared, ee) k' = kdf(reg/msg3/inner, es) - msg4 B->A: box(se_cap | tb) k4 = kdf(reg/msg4, shared, ee) + msg4 B->A: box(se_ct | tb) k4 = kdf(reg/msg4, shared, ee) msg5 A->B: box(ta) k5 = kdf(reg/msg5, shared, ee, es) c = kdf(commit, e_alice, r) r = random reveal, chosen with e_alice @@ -82,18 +84,17 @@ AEAD failure proves no key knowledge and changes nothing: se = Encap(s_alice) challenges alice's static root = kdf(reg/root, shared, ee, es, se) - tb = kdf(reg/confirm-b, root) proves bob decapsulated es_cap - ta = kdf(reg/confirm-a, root) proves alice decapsulated se_cap + tb = kdf(reg/confirm-b, root) proves bob decapsulated es_ct + ta = kdf(reg/confirm-a, root) proves alice decapsulated se_ct sas = kdf(sas, shared, ee, r, length=4) -- compared out of band shared' = kdf(rotate, root), send/recv = kdf(a2b | b2a, root) -e_alice, s_alice, s_bob are hybrid public keys; the only ephemeral -public shipped is alice's e_alice (the responder needs none of his -own: every leg is an encapsulation, which already carries a fresh -ephemeral inside the capsule). The inner boxes key only on the secret -they actually protect: the outer box already binds shared and ee, so -there is no use in repeating them in the inner key. +e_alice, s_alice, s_bob are hybrid public keys; e_bob is a bare X25519 +ephemeral public. Every leg's shared secret is ML-KEM ss || DH ss, the +DH made against the shipped connection ephemerals. The inner boxes key +only on the secret they actually protect: the outer box already binds +shared and ee, so there is no use in repeating them in the inner key. The exchange's tail is gated on the out-of-band comparison: msg4 is released only after the responder's answer, msg5 only after the @@ -129,23 +130,21 @@ it exists, and every such attempt costs a fresh online connection. Statics PROVEN, never on the wire; all legs hybrid: - msg1 A->B: Box_k1(e_alice | sb_cap) k1 = kdf(res/msg1, shared) - msg2 B->A: Box_k2(sa_cap | ee_cap | tb) k2 = kdf(res/msg2, shared, sb) + msg1 A->B: Box_k1(e_alice | sb_ct) k1 = kdf(res/msg1, shared) + msg2 B->A: Box_k2(e_bob | sa_ct | ee_ct | tb) k2 = kdf(res/msg2, shared, sb) msg3 A->B: Box_k3(ta) k3 = kdf(res/msg3, shared, sb, sa, ee) tb = kdf(res/tag-b, sb, sa, ee) ta = kdf(res/tag-a, sb, sa, ee) key = kdf(res/session-key, shared, sb, sa, ee) -> send/recv = kdf(a2b | b2a, key) - sb_cap = Encap(s_bob) authenticates bob (to alice, via tb) - sa_cap = Encap(s_alice) authenticates alice (to bob, via ta) - ee_cap = Encap(e_alice) forward secrecy + sb_ct = Encap(s_bob) authenticates bob (to alice, via tb) + sa_ct = Encap(s_alice) authenticates alice (to bob, via ta) + ee_ct = Encap(e_alice) forward secrecy -Bob ships no ephemeral of his own here: no leg references it (ee is an -encapsulation to alice's ephemeral, whose fresh X25519 public half -already rides inside ee_cap). With the statics pre-shared, challenges -and answers fit into msg1..3, so possession is proven by the per-leg -tags tb/ta directly, without registration's extra tail messages. +With the statics pre-shared, challenges and answers fit into msg1..3, +so possession is proven by the per-leg tags tb/ta directly, without +registration's extra tail messages. ## Traffic diff --git a/quantumpipe/handshake.py b/quantumpipe/handshake.py index f15efb2..0486b7e 100644 --- a/quantumpipe/handshake.py +++ b/quantumpipe/handshake.py @@ -10,7 +10,14 @@ from __future__ import annotations import hmac import struct -from cryptography.hazmat.primitives.asymmetric import mlkem, x25519 +from cryptography.hazmat.primitives.asymmetric.mlkem import ( + MLKEM768PrivateKey, + MLKEM768PublicKey, +) +from cryptography.hazmat.primitives.asymmetric.x25519 import ( + X25519PrivateKey, + X25519PublicKey, +) from .chab3 import decrypt, encrypt, random from .exceptions import _Burn @@ -19,19 +26,17 @@ from .kdf import kdf # -- fixed-size record layouts (struct handles packing + length checks) ----- # hybrid scheme = ML-KEM-768 (pk 1184 / sk-seed 64 / ct 1088) || X25519 (32) _PUB = struct.Struct("1184s32s") # hybrid public key: 1216 -_CAP = struct.Struct("1088s32s") # hybrid capsule: mlkem ct | fresh x25519 pub -_PUB_CAP = struct.Struct("1216s1120s") # hybrid public key | hybrid capsule _TAG = struct.Struct("32s") # bare kdf tag -# handshake message bodies +# handshake message bodies; box(N) is a chab3 box of N plaintext bytes (N+28) _MSG1_REGISTER = struct.Struct("1216s32s") # e_alice | commitment -_STATIC = struct.Struct("1216s") # bare hybrid static public key +_MSG2_REGISTER = struct.Struct("32s1088s1244s") # e_bob | ee_ct | box(s_bob) +_MSG3_REGISTER = struct.Struct("1088s1276s") # es_ct | box(s_alice | reveal) _MSG3_REGISTER_INNER = struct.Struct("1216s32s") # s_alice | reveal -_MSG4_REGISTER = struct.Struct("1120s32s") # se_cap | tb -_MSG2_RESUME = struct.Struct("1120s1120s32s") # sa_cap | ee_cap | tb -# msg2 = ee_cap | inner box(s_bob); msg3 = es_cap | inner box(s_alice | reveal); -# msg5 = _TAG (ta); msg1 resumption = _PUB_CAP (e_alice | sb_cap); -# msg3 resumption = _TAG (ta) +_MSG4_REGISTER = struct.Struct("1088s32s") # se_ct | tb +_MSG1_RESUME = struct.Struct("1216s1088s") # e_alice | sb_ct +_MSG2_RESUME = struct.Struct("32s1088s1088s32s") # e_bob | sa_ct | ee_ct | tb +# msg5 = _TAG (ta); msg3 resumption = _TAG (ta) # kdf info strings: boxes are namespaced per flow and message (reg/msg1, # res/msg2, reg/msg2/inner, ...) so the two exchanges are disjoint; other @@ -40,21 +45,26 @@ _MSG2_RESUME = struct.Struct("1120s1120s32s") # sa_cap | ee_cap | tb # -------------------------------------------------------------------------- -# hybrid KEM: ML-KEM-768 || X25519, operating on raw byte buffers only +# hybrid KEM: ML-KEM-768 || X25519 # -# A private key is a single 32-byte idroot secret. Each scheme's key is -# derived from it with one domain-separated kdf call, so the two subkeys -# are independent PRF outputs and the idroot is the only thing held. +# A static private key is a single 32-byte idroot secret: each scheme's +# key is derived from it with one domain-separated kdf call, so the two +# subkeys are independent PRF outputs and the idroot is the only thing +# held (and the only thing the credential stores). Ephemeral keys are +# generated directly -- nothing of them persists. +# +# Each party ships one X25519 ephemeral public per connection (alice's +# inside e_alice in msg1, bob's as e_bob in msg2); every X25519 leg is a +# plain DH against those shipped keys. ML-KEM legs ship as bare ML-KEM +# ciphertexts. -def _mlkem_private_key(idroot: bytes) -> mlkem.MLKEM768PrivateKey: - return mlkem.MLKEM768PrivateKey.from_seed_bytes( - kdf("idroot/mlkem", idroot, length=64) - ) +def _mlkem_private_key(idroot: bytes) -> MLKEM768PrivateKey: + return MLKEM768PrivateKey.from_seed_bytes(kdf("idroot/mlkem", idroot, length=64)) -def _x25519_private_key(idroot: bytes) -> x25519.X25519PrivateKey: - return x25519.X25519PrivateKey.from_private_bytes(kdf("idroot/x25519", idroot)) +def _x25519_private_key(idroot: bytes) -> X25519PrivateKey: + return X25519PrivateKey.from_private_bytes(kdf("idroot/x25519", idroot)) def _public_from_idroot(idroot: bytes) -> bytes: @@ -64,37 +74,17 @@ def _public_from_idroot(idroot: bytes) -> bytes: ) -def _encapsulate(peer_public: bytes) -> tuple[bytes, bytes]: - # -> (shared_secret, capsule), each the concat of both schemes - mk_raw, xk_raw = _PUB.unpack(peer_public) - m_ss, m_ct = mlkem.MLKEM768PublicKey.from_public_bytes(mk_raw).encapsulate() - x_ek = x25519.X25519PrivateKey.generate() - try: - x_ss = x_ek.exchange(x25519.X25519PublicKey.from_public_bytes(xk_raw)) - except ValueError: - raise ValueError("non-contributory X25519 public key") from None - return m_ss + x_ss, _CAP.pack(m_ct, x_ek.public_key().public_bytes_raw()) - - -def _decapsulate(idroot: bytes, cap: bytes) -> bytes: - # ML-KEM has FIPS 203 implicit rejection (never fails; wrong key -> - # garbage caught by the kdf tags); X25519 does not, so a low-order - # result is rejected explicitly - m_ct, x_ct = _CAP.unpack(cap) - m_ss = _mlkem_private_key(idroot).decapsulate(m_ct) - try: - x_ss = _x25519_private_key(idroot).exchange( - x25519.X25519PublicKey.from_public_bytes(x_ct) - ) - except ValueError: - raise ValueError("non-contributory X25519 capsule") from None - return m_ss + x_ss - - -def _fresh_ephemeral() -> tuple[bytes, bytes]: - # (root, public) for a fresh per-connection ephemeral key - root = random(32) - return root, _public_from_idroot(root) +def _fresh_ephemeral() -> tuple[MLKEM768PrivateKey, X25519PrivateKey, bytes]: + # (mlkem key, x25519 key, packed public) for a fresh connection ephemeral + m_sk = MLKEM768PrivateKey.generate() + x_sk = X25519PrivateKey.generate() + return ( + m_sk, + x_sk, + _PUB.pack( + m_sk.public_key().public_bytes_raw(), x_sk.public_key().public_bytes_raw() + ), + ) def _traffic_keys(key: bytes, initiator: bool) -> tuple[bytes, bytes]: @@ -146,15 +136,18 @@ class _Chain: class Registration: - """Registration exchange (first connection): every leg is the same - hybrid KEM operation -- ee_cap for forward secrecy, es_cap/se_cap - challenging the statics passed late (s_bob inside msg2's inner box, - s_alice inside msg3's). finish() -> (send, recv, rotated shared, - sas, peer pub).""" + """Registration exchange (first connection): every leg is a bare + ML-KEM ciphertext plus an X25519 DH against the shipped connection + ephemerals (e_alice in msg1, e_bob in msg2), so related secrets + provably come from the same party. finish() -> (send, recv, rotated + shared, sas, peer pub).""" def __init__(self, shared: bytes, idroot: bytes, ad: bytes): self._shared = shared self._idroot = idroot + # static key objects, derived once from the persisted idroot + self._mlkem = _mlkem_private_key(idroot) + self._x25519 = _x25519_private_key(idroot) self._static_pub = _public_from_idroot(idroot) self._chain = _Chain(ad) @@ -165,7 +158,7 @@ class Registration: def startA(self) -> bytes: # msg1: the ephemeral key, committed under a reveal hidden until msg3 - self._ephemeral_root, e_alice = _fresh_ephemeral() + self._eph_m, self._ephemeral, e_alice = _fresh_ephemeral() self._reveal = random(32) body = _MSG1_REGISTER.pack(e_alice, kdf("commit", e_alice, self._reveal)) return self._chain.encrypt(body, key=kdf("reg/msg1", self._shared)) @@ -176,17 +169,21 @@ class Registration: # box, and challenge it plain = self._chain.decrypt(msg2, key=kdf("reg/msg2", self._shared)) try: - ee_cap, inner = plain[: _CAP.size], plain[_CAP.size :] - ee = _decapsulate(self._ephemeral_root, ee_cap) - (s_bob,) = _STATIC.unpack(decrypt(inner, key=kdf("reg/msg2/inner", ee))) - es, es_cap = _encapsulate(s_bob) # challenge: bob's static + e_bob, ee_ct, inner = _MSG2_REGISTER.unpack(plain) + ee = self._eph_m.decapsulate(ee_ct) + ee += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(e_bob)) + s_bob = decrypt(inner, key=kdf("reg/msg2/inner", ee)) + s_bob_m, s_bob_x = _PUB.unpack(s_bob) + es, es_ct = MLKEM768PublicKey.from_public_bytes(s_bob_m).encapsulate() + es += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(s_bob_x)) except (struct.error, ValueError) as e: raise _Burn(f"bad msg2 contents ({e})") from None self._peer_pub = s_bob + self._e_bob = e_bob self._ee = ee self._es = es return self._chain.encrypt( - es_cap + es_ct + encrypt( _MSG3_REGISTER_INNER.pack(self._static_pub, self._reveal), key=kdf("reg/msg3/inner", es), @@ -199,8 +196,9 @@ class Registration: # caller gates the answer (msg5) on the out-of-band comparison. plain = self._chain.decrypt(msg4, key=kdf("reg/msg4", self._shared, self._ee)) try: - se_cap, tb = _MSG4_REGISTER.unpack(plain) - se = _decapsulate(self._idroot, se_cap) + se_ct, tb = _MSG4_REGISTER.unpack(plain) + se = self._mlkem.decapsulate(se_ct) + se += self._x25519.exchange(X25519PublicKey.from_public_bytes(self._e_bob)) except (struct.error, ValueError) as e: raise _Burn(f"bad msg4 contents ({e})") from None self._se = se @@ -218,16 +216,23 @@ class Registration: # -- bob ------------------------------------------------------------------ def respondB(self, msg1: bytes, opened: bytes) -> bytes: - # msg2 from an opened msg1 (open_hello did the caller scan) + # msg2 from an opened msg1 (open_hello did the caller scan); our + # connection x25519 ephemeral ships here, once self._chain.advance(msg1) try: e_alice, commitment = _MSG1_REGISTER.unpack(opened) - ee, ee_cap = _encapsulate(e_alice) # forward secrecy + self._ephemeral = X25519PrivateKey.generate() + e_bob = self._ephemeral.public_key().public_bytes_raw() + e_alice_m, e_alice_x = _PUB.unpack(e_alice) + ee, ee_ct = MLKEM768PublicKey.from_public_bytes(e_alice_m).encapsulate() + ee += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(e_alice_x)) except (struct.error, ValueError) as e: raise _Burn(f"bad msg1 contents ({e})") from None self._commitment = (e_alice, commitment) - body = ee_cap + encrypt( - _STATIC.pack(self._static_pub), key=kdf("reg/msg2/inner", ee) + body = _MSG2_REGISTER.pack( + e_bob, + ee_ct, + encrypt(self._static_pub, key=kdf("reg/msg2/inner", ee)), ) self._ee = ee return self._chain.encrypt(body, key=kdf("reg/msg2", self._shared)) @@ -237,15 +242,19 @@ class Registration: # caller gates the answer (msg4) on the out-of-band comparison. plain = self._chain.decrypt(msg3, key=kdf("reg/msg3", self._shared, self._ee)) try: - es_cap, inner = plain[: _CAP.size], plain[_CAP.size :] - es = _decapsulate(self._idroot, es_cap) # proves our static + es_ct, inner = _MSG3_REGISTER.unpack(plain) + e_alice, commitment = self._commitment + e_alice_x = _PUB.unpack(e_alice)[1] + es = self._mlkem.decapsulate(es_ct) # proves our static + es += self._x25519.exchange(X25519PublicKey.from_public_bytes(e_alice_x)) s_alice, reveal = _MSG3_REGISTER_INNER.unpack( decrypt(inner, key=kdf("reg/msg3/inner", es)) ) - se, se_cap = _encapsulate(s_alice) + s_alice_m, s_alice_x = _PUB.unpack(s_alice) + se, se_ct = MLKEM768PublicKey.from_public_bytes(s_alice_m).encapsulate() + se += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(s_alice_x)) except (struct.error, ValueError) as e: raise _Burn(f"bad msg3 contents ({e})") from None - e_alice, commitment = self._commitment _check( hmac.compare_digest(commitment, kdf("commit", e_alice, reveal)), "commitment reveal mismatch", @@ -256,13 +265,13 @@ class Registration: self._se = se sas = kdf("sas", self._shared, self._ee, reveal, length=4) msg4 = self._chain.encrypt( - _MSG4_REGISTER.pack(se_cap, kdf("reg/confirm-b", self._root())), + _MSG4_REGISTER.pack(se_ct, kdf("reg/confirm-b", self._root())), key=kdf("reg/msg4", self._shared, self._ee), ) return msg4, sas def finishB(self, msg5: bytes) -> None: - # alice's root confirmation: proves she decapsulated se_cap + # alice's root confirmation: proves she decapsulated se_ct plain = self._chain.decrypt( msg5, key=kdf("reg/msg5", self._shared, self._ee, self._es) ) @@ -294,6 +303,8 @@ class Resumption: def __init__(self, shared: bytes, idroot: bytes, peer_pub: bytes, ad: bytes): self._shared = shared self._idroot = idroot + self._mlkem = _mlkem_private_key(idroot) + self._x25519 = _x25519_private_key(idroot) self._peer_pub = peer_pub # proven by decapsulation self._chain = _Chain(ad) @@ -301,19 +312,23 @@ class Resumption: def startA(self) -> bytes: # msg1: the ephemeral key plus an encapsulation to bob's static - # (the responder ships no ephemeral of his own: no leg uses it) - self._ephemeral_root, e_alice = _fresh_ephemeral() - self._sb, sb_cap = _encapsulate(self._peer_pub) - body = _PUB_CAP.pack(e_alice, sb_cap) + self._eph_m, self._ephemeral, e_alice = _fresh_ephemeral() + peer_m, peer_x = _PUB.unpack(self._peer_pub) + sb, sb_ct = MLKEM768PublicKey.from_public_bytes(peer_m).encapsulate() + sb += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(peer_x)) + self._sb = sb + body = _MSG1_RESUME.pack(e_alice, sb_ct) return self._chain.encrypt(body, key=kdf("res/msg1", self._shared)) def respondA(self, msg2: bytes) -> bytes: # open bob's reply, decapsulate all three legs plain = self._chain.decrypt(msg2, key=kdf("res/msg2", self._shared, self._sb)) try: - sa_cap, ee_cap, tb = _MSG2_RESUME.unpack(plain) - sa = _decapsulate(self._idroot, sa_cap) # proves our static - ee = _decapsulate(self._ephemeral_root, ee_cap) # forward secrecy + e_bob, sa_ct, ee_ct, tb = _MSG2_RESUME.unpack(plain) + sa = self._mlkem.decapsulate(sa_ct) # proves our static + sa += self._x25519.exchange(X25519PublicKey.from_public_bytes(e_bob)) + ee = self._eph_m.decapsulate(ee_ct) # forward secrecy + ee += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(e_bob)) except (struct.error, ValueError) as e: raise _Burn(f"bad msg2 contents ({e})") from None _check( @@ -333,16 +348,23 @@ class Resumption: # prove our static, prove alice's, forward secrecy self._chain.advance(msg1) try: - e_alice, sb_cap = _PUB_CAP.unpack(opened) - sb = _decapsulate(self._idroot, sb_cap) # proves our static - sa, sa_cap = _encapsulate(self._peer_pub) # proves peer's static - ee, ee_cap = _encapsulate(e_alice) # forward secrecy + e_alice, sb_ct = _MSG1_RESUME.unpack(opened) + e_alice_m, e_alice_x = _PUB.unpack(e_alice) + sb = self._mlkem.decapsulate(sb_ct) # proves our static + sb += self._x25519.exchange(X25519PublicKey.from_public_bytes(e_alice_x)) + self._ephemeral = X25519PrivateKey.generate() + e_bob = self._ephemeral.public_key().public_bytes_raw() + peer_m, peer_x = _PUB.unpack(self._peer_pub) + sa, sa_ct = MLKEM768PublicKey.from_public_bytes(peer_m).encapsulate() + sa += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(peer_x)) + ee, ee_ct = MLKEM768PublicKey.from_public_bytes(e_alice_m).encapsulate() + ee += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(e_alice_x)) except (struct.error, ValueError) as e: raise _Burn(f"bad msg1 contents ({e})") from None self._sb = sb self._sa = sa self._ee = ee - body = _MSG2_RESUME.pack(sa_cap, ee_cap, kdf("res/tag-b", sb, sa, ee)) + body = _MSG2_RESUME.pack(e_bob, sa_ct, ee_ct, kdf("res/tag-b", sb, sa, ee)) return self._chain.encrypt(body, key=kdf("res/msg2", self._shared, sb)) def completeB(self, msg3: bytes) -> None: diff --git a/tests/test_sansio.py b/tests/test_sansio.py index a463a2c..7a6d7b4 100644 --- a/tests/test_sansio.py +++ b/tests/test_sansio.py @@ -1,15 +1,16 @@ import secrets import pytest +from cryptography.hazmat.primitives.asymmetric.mlkem import MLKEM768PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PublicKey from quantumpipe.chab3 import encrypt from quantumpipe.exceptions import ProtocolError from quantumpipe.handshake import ( _MSG1_REGISTER, - _decapsulate, - _encapsulate, _fresh_ephemeral, _public_from_idroot, + _x25519_private_key, ) from quantumpipe.kdf import kdf from quantumpipe.sansio import ( @@ -694,7 +695,7 @@ def test_mitm_reused_reveal_detected(psk): # leg beta: the MITM recommits the stolen reveal toward bob hs = m_out._exchange hs._reveal = r_a - hs._ephemeral_root, e_pub = _fresh_ephemeral() + hs._eph_m, hs._ephemeral, e_pub = _fresh_ephemeral() body = _MSG1_REGISTER.pack(e_pub, kdf("commit", e_pub, r_a)) msg1m = bytes(encrypt(body, key=kdf("reg/msg1", psk), ad=b"")) hs._chain.advance(msg1m) # startA was bypassed: advance the chain @@ -722,13 +723,10 @@ def test_relayed_reveal_rejected(psk): def test_low_order_x25519_rejected(): - # non-contributory public keys are rejected outright in the hybrid - # encapsulation and decapsulation paths - idroot = secrets.token_bytes(32) + # non-contributory public keys are rejected at the DH exchange + sk = _x25519_private_key(secrets.token_bytes(32)) with pytest.raises(ValueError): - _encapsulate(b"\x00" * 1184 + b"\x00" * 32) - with pytest.raises(ValueError): - _decapsulate(idroot, b"\x00" * 1088 + b"\x00" * 32) + sk.exchange(X25519PublicKey.from_public_bytes(b"\x00" * 32)) def test_foreign_mlkem_static_rejected_at_msg4(psk): @@ -757,7 +755,7 @@ def test_foreign_mlkem_static_rejected_responder(psk): a, b = make_sessions(psk) (m1,) = a() (m2,) = b(m1) - b._exchange._idroot = secrets.token_bytes(32) # not the presented static's root + b._exchange._mlkem = MLKEM768PrivateKey.generate() # not the presented key (m3,) = a(m2) with pytest.raises(ProtocolError, match="burned"): b(m3)