From f035859fd2e518d68aeadd4679e8dfd230deae5b Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Wed, 7 Oct 2026 17:59:13 +0000 Subject: [PATCH] Split hybrid public keys into separate wire fields The packed 1216-byte hybrid public had no consumer that wanted it whole -- everything unpacked immediately. Message layouts and the credential record now carry peer_m | peer_x as explicit fields; _PUB and the unpack-at-use idiom are gone. credential_id values change with the record layout. Step comments in the exchange methods explain what each solve/challenge proves. --- docs/design.md | 2 +- quantumpipe/handshake.py | 126 ++++++++++++++++++++++----------------- quantumpipe/sansio.py | 8 ++- tests/test_sansio.py | 27 ++++----- tests/test_ws.py | 2 +- 5 files changed, 92 insertions(+), 73 deletions(-) diff --git a/docs/design.md b/docs/design.md index 623f1fe..d7187a9 100644 --- a/docs/design.md +++ b/docs/design.md @@ -47,7 +47,7 @@ an application-chosen id: either the bare 32-byte registration secret unhashed; the kdf info strings provide all domain separation needed) or the composite written by the first successful registration: - idroot(32) | shared'(32) | peer_public_key(1216) | sas(4) | verified(1) + idroot(32) | shared'(32) | peer_m(1184) | peer_x(32) | sas(4) | verified(1) The rotation to shared' is derived when the exchange completes but only held in memory: the store swap waits for the key change -- the diff --git a/quantumpipe/handshake.py b/quantumpipe/handshake.py index 0486b7e..766c427 100644 --- a/quantumpipe/handshake.py +++ b/quantumpipe/handshake.py @@ -25,16 +25,18 @@ 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 _TAG = struct.Struct("32s") # bare kdf tag # handshake message bodies; box(N) is a chab3 box of N plaintext bytes (N+28) -_MSG1_REGISTER = struct.Struct("1216s32s") # e_alice | commitment -_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 +_MSG1_REGISTER = struct.Struct("1184s32s32s") # e_alice_m | e_alice_x | commitment +_MSG2_REGISTER = struct.Struct( + "32s1088s1244s" +) # e_bob | ee_ct | box(s_bob_m | s_bob_x) +_MSG2_REGISTER_INNER = struct.Struct("1184s32s") # s_bob_m | s_bob_x +_MSG3_REGISTER = struct.Struct("1088s1276s") # es_ct | box(s_alice_m | s_alice_x | r) +_MSG3_REGISTER_INNER = struct.Struct("1184s32s32s") # s_alice_m | s_alice_x | reveal _MSG4_REGISTER = struct.Struct("1088s32s") # se_ct | tb -_MSG1_RESUME = struct.Struct("1216s1088s") # e_alice | sb_ct +_MSG1_RESUME = struct.Struct("1184s32s1088s") # e_alice_m | e_alice_x | sb_ct _MSG2_RESUME = struct.Struct("32s1088s1088s32s") # e_bob | sa_ct | ee_ct | tb # msg5 = _TAG (ta); msg3 resumption = _TAG (ta) @@ -67,23 +69,23 @@ def _x25519_private_key(idroot: bytes) -> X25519PrivateKey: return X25519PrivateKey.from_private_bytes(kdf("idroot/x25519", idroot)) -def _public_from_idroot(idroot: bytes) -> bytes: - return _PUB.pack( +def _public_from_idroot(idroot: bytes) -> tuple[bytes, bytes]: + # (mlkem public, x25519 public) derived from the idroot + return ( _mlkem_private_key(idroot).public_key().public_bytes_raw(), _x25519_private_key(idroot).public_key().public_bytes_raw(), ) -def _fresh_ephemeral() -> tuple[MLKEM768PrivateKey, X25519PrivateKey, bytes]: - # (mlkem key, x25519 key, packed public) for a fresh connection ephemeral +def _fresh_ephemeral() -> tuple[MLKEM768PrivateKey, X25519PrivateKey, bytes, bytes]: + # (mlkem key, x25519 key, mlkem public, x25519 public), fresh per connection 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() - ), + m_sk.public_key().public_bytes_raw(), + x_sk.public_key().public_bytes_raw(), ) @@ -138,9 +140,9 @@ class _Chain: class Registration: """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).""" + keys (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 @@ -148,7 +150,7 @@ class Registration: # 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._static_m, self._static_x = _public_from_idroot(idroot) self._chain = _Chain(ad) def _root(self) -> bytes: @@ -157,35 +159,40 @@ class Registration: # -- alice ---------------------------------------------------------------- def startA(self) -> bytes: - # msg1: the ephemeral key, committed under a reveal hidden until msg3 - self._eph_m, self._ephemeral, e_alice = _fresh_ephemeral() + # msg1: our connection keys, with the reveal committed so it + # cannot be predicted or steered + self._eph_m, self._ephemeral, e_alice_m, e_alice_x = _fresh_ephemeral() self._reveal = random(32) - body = _MSG1_REGISTER.pack(e_alice, kdf("commit", e_alice, self._reveal)) + commitment = kdf("commit", e_alice_m, e_alice_x, self._reveal) + body = _MSG1_REGISTER.pack(e_alice_m, e_alice_x, commitment) return self._chain.encrypt(body, key=kdf("reg/msg1", self._shared)) def respondA(self, msg2: bytes) -> bytes: # open bob's reply -- an outer AEAD failure burns nothing, it - # proves no key knowledge -- then the static inside its inner - # box, and challenge it + # proves no key knowledge -- then challenge his static plain = self._chain.decrypt(msg2, key=kdf("reg/msg2", self._shared)) try: e_bob, ee_ct, inner = _MSG2_REGISTER.unpack(plain) + # ee is forward secrecy, made to our connection key only 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) + # solving ee lets us read bob's static from the inner box + s_bob_m, s_bob_x = _MSG2_REGISTER_INNER.unpack( + decrypt(inner, key=kdf("reg/msg2/inner", ee)) + ) + # we challenge his static: only the real bob can solve this 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._peer_pub = (s_bob_m, s_bob_x) self._e_bob = e_bob self._ee = ee self._es = es return self._chain.encrypt( es_ct + encrypt( - _MSG3_REGISTER_INNER.pack(self._static_pub, self._reveal), + _MSG3_REGISTER_INNER.pack(self._static_m, self._static_x, self._reveal), key=kdf("reg/msg3/inner", es), ), key=kdf("reg/msg3", self._shared, ee), @@ -197,6 +204,7 @@ class Registration: plain = self._chain.decrypt(msg4, key=kdf("reg/msg4", self._shared, self._ee)) try: se_ct, tb = _MSG4_REGISTER.unpack(plain) + # solving se proves to bob that we hold alice's static se = self._mlkem.decapsulate(se_ct) se += self._x25519.exchange(X25519PublicKey.from_public_bytes(self._e_bob)) except (struct.error, ValueError) as e: @@ -217,22 +225,25 @@ class Registration: def respondB(self, msg1: bytes, opened: bytes) -> bytes: # msg2 from an opened msg1 (open_hello did the caller scan); our - # connection x25519 ephemeral ships here, once + # connection key ships here once, alice answers every leg with it self._chain.advance(msg1) try: - e_alice, commitment = _MSG1_REGISTER.unpack(opened) + e_alice_m, e_alice_x, commitment = _MSG1_REGISTER.unpack(opened) self._ephemeral = X25519PrivateKey.generate() e_bob = self._ephemeral.public_key().public_bytes_raw() - e_alice_m, e_alice_x = _PUB.unpack(e_alice) + # forward secrecy, made to alice's connection key only 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) + self._commitment = (e_alice_m, e_alice_x, commitment) body = _MSG2_REGISTER.pack( e_bob, ee_ct, - encrypt(self._static_pub, key=kdf("reg/msg2/inner", ee)), + encrypt( + _MSG2_REGISTER_INNER.pack(self._static_m, self._static_x), + key=kdf("reg/msg2/inner", ee), + ), ) self._ee = ee return self._chain.encrypt(body, key=kdf("reg/msg2", self._shared)) @@ -243,24 +254,27 @@ class Registration: plain = self._chain.decrypt(msg3, key=kdf("reg/msg3", self._shared, self._ee)) try: 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 + e_alice_m, e_alice_x, commitment = self._commitment + # Solving es proves to Alice that the static key they encapsulated to was ours + es = self._mlkem.decapsulate(es_ct) es += self._x25519.exchange(X25519PublicKey.from_public_bytes(e_alice_x)) - s_alice, reveal = _MSG3_REGISTER_INNER.unpack( + # This enables us to get Alice's static keys and reveal + s_alice_m, s_alice_x, reveal = _MSG3_REGISTER_INNER.unpack( decrypt(inner, key=kdf("reg/msg3/inner", es)) ) - s_alice_m, s_alice_x = _PUB.unpack(s_alice) + # We send Alice the same test, verifying they have s_alice secrets 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 _check( - hmac.compare_digest(commitment, kdf("commit", e_alice, reveal)), + hmac.compare_digest( + commitment, kdf("commit", e_alice_m, e_alice_x, reveal) + ), "commitment reveal mismatch", ) self._reveal = reveal - self._peer_pub = s_alice + self._peer_pub = (s_alice_m, s_alice_x) self._es = es self._se = se sas = kdf("sas", self._shared, self._ee, reveal, length=4) @@ -271,7 +285,7 @@ class Registration: return msg4, sas def finishB(self, msg5: bytes) -> None: - # alice's root confirmation: proves she decapsulated se_ct + # alice's root confirmation: proves she solved our se challenge plain = self._chain.decrypt( msg5, key=kdf("reg/msg5", self._shared, self._ee, self._es) ) @@ -286,7 +300,7 @@ class Registration: # -- completion ------------------------------------------------------------- - def finish(self, initiator: bool) -> tuple[bytes, bytes, bytes, bytes, bytes]: + def finish(self, initiator: bool) -> tuple[bytes, bytes, bytes, bytes, tuple]: # -> (send, recv, rotated shared, sas, peer pub); the rotation is # held by the caller until the store swap root = self._root() @@ -300,34 +314,36 @@ class Resumption: the wire; all legs hybrid. Alice: startA/respondA; bob: respondB/completeB; finish() -> (send, recv).""" - def __init__(self, shared: bytes, idroot: bytes, peer_pub: bytes, ad: bytes): + def __init__(self, shared: bytes, idroot: bytes, peer_pub: tuple, 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._peer_pub = peer_pub # (mlkem public, x25519 public) self._chain = _Chain(ad) # -- alice ---------------------------------------------------------------- def startA(self) -> bytes: - # msg1: the ephemeral key plus an encapsulation to bob's static - self._eph_m, self._ephemeral, e_alice = _fresh_ephemeral() - peer_m, peer_x = _PUB.unpack(self._peer_pub) + # msg1: our connection keys plus a challenge to bob's static + self._eph_m, self._ephemeral, e_alice_m, e_alice_x = _fresh_ephemeral() + peer_m, peer_x = 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) + body = _MSG1_RESUME.pack(e_alice_m, e_alice_x, 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 + # open bob's reply and solve his three challenges plain = self._chain.decrypt(msg2, key=kdf("res/msg2", self._shared, self._sb)) try: e_bob, sa_ct, ee_ct, tb = _MSG2_RESUME.unpack(plain) - sa = self._mlkem.decapsulate(sa_ct) # proves our static + # solving sa proves to bob that we hold alice's static + sa = self._mlkem.decapsulate(sa_ct) sa += self._x25519.exchange(X25519PublicKey.from_public_bytes(e_bob)) - ee = self._eph_m.decapsulate(ee_ct) # forward secrecy + # ee is forward secrecy, made to our connection key + ee = self._eph_m.decapsulate(ee_ct) 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 @@ -345,18 +361,20 @@ class Resumption: # -- bob ------------------------------------------------------------------ def respondB(self, msg1: bytes, opened: bytes) -> bytes: - # prove our static, prove alice's, forward secrecy + # prove our static, challenge alice's, forward secrecy self._chain.advance(msg1) try: - 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 + e_alice_m, e_alice_x, sb_ct = _MSG1_RESUME.unpack(opened) + # solving sb proves to alice that we hold bob's static + sb = self._mlkem.decapsulate(sb_ct) 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) + peer_m, peer_x = self._peer_pub + # we challenge alice's static the same way sa, sa_ct = MLKEM768PublicKey.from_public_bytes(peer_m).encapsulate() sa += self._ephemeral.exchange(X25519PublicKey.from_public_bytes(peer_x)) + # and make forward secrecy to her connection key 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: diff --git a/quantumpipe/sansio.py b/quantumpipe/sansio.py index 0124cb1..af26680 100644 --- a/quantumpipe/sansio.py +++ b/quantumpipe/sansio.py @@ -38,7 +38,7 @@ def _describe(obj) -> str: # given to both parties out of band and stored as a bare 32-byte store # entry. The first successful registration replaces it with the composite. -_CREDENTIAL = struct.Struct("32s32s1216s4s?") # idroot | shared | peer pub | sas | vf +_CREDENTIAL = struct.Struct("32s32s1184s32s4s?") # idroot|shared|peer_m|peer_x|sas|vf def generate_registration_secret() -> bytes: @@ -240,10 +240,12 @@ class Session: ( self._idroot, self._shared, - self._peer_pub, + peer_m, + peer_x, self._sas, self._credential_verified, ) = _CREDENTIAL.unpack(cred) + self._peer_pub = (peer_m, peer_x) self._registering = False hs = Resumption(self._shared, self._idroot, self._peer_pub, self._ad) self._exchange = hs @@ -263,7 +265,7 @@ class Session: # persists the store itself if it wants durability self._credential_verified = verified self._stored = _CREDENTIAL.pack( - self._idroot, self._shared, self._peer_pub, self._sas, verified + self._idroot, self._shared, *self._peer_pub, self._sas, verified ) self._store[self._key_id] = self._stored diff --git a/tests/test_sansio.py b/tests/test_sansio.py index 7a6d7b4..9347db4 100644 --- a/tests/test_sansio.py +++ b/tests/test_sansio.py @@ -81,12 +81,12 @@ def test_registration_credential(psk): run_handshake(a, b) blob_a, blob_b = a.credential, b.credential assert len(blob_a) == len(blob_b) == _CREDENTIAL.size - id_a, shared_a, pub_b, sas_a, vf_a = _CREDENTIAL.unpack(blob_a) - id_b, shared_b, pub_a, sas_b, vf_b = _CREDENTIAL.unpack(blob_b) + id_a, shared_a, pub_b_m, pub_b_x, sas_a, vf_a = _CREDENTIAL.unpack(blob_a) + id_b, shared_b, pub_a_m, pub_a_x, sas_b, vf_b = _CREDENTIAL.unpack(blob_b) assert id_a != id_b # fresh random identity roots per party assert shared_a == shared_b # same rotated shared key - assert pub_a == _public_from_idroot(id_a) # peer pubs match peer idroots - assert pub_b == _public_from_idroot(id_b) + assert (pub_a_m, pub_a_x) == _public_from_idroot(id_a) # pubs match idroots + assert (pub_b_m, pub_b_x) == _public_from_idroot(id_b) assert sas_a == sas_b # the stored comparison digest assert vf_a and vf_b # confirmed by run_handshake's answers @@ -317,7 +317,7 @@ def test_verification_not_required(psk): assert not a.verified and not b.verified assert est_a.sas == est_b.sas # the SAS is still exposed assert a.state == b.state == "established" - assert _CREDENTIAL.unpack(a._store["x"])[4] is False # stored unverified + assert _CREDENTIAL.unpack(a._store["x"])[5] is False # stored unverified def test_wrong_calls_rejected(psk): @@ -478,7 +478,7 @@ def test_unverified_resumption_is_silent(psk): a, b = make_sessions(psk) run_handshake(a, b, answer=None) cred_a, cred_b = a.credential, b.credential - assert _CREDENTIAL.unpack(cred_a)[4] is False + assert _CREDENTIAL.unpack(cred_a)[5] is False a2, b2 = make_sessions(cred_a, cred_b) (m1,) = a2() @@ -492,7 +492,7 @@ def test_unverified_resumption_is_silent(psk): a2.confirm(True) # compared out of band: flag it verified b2.confirm(True) assert a2.verified and b2.verified - assert _CREDENTIAL.unpack(a2._store["x"])[4] is True # flag flipped + assert _CREDENTIAL.unpack(a2._store["x"])[5] is True # flag flipped a3, b3 = make_sessions(a2._store["x"], b2._store["x"]) run_handshake(a3, b3) # verified: silent as before @@ -639,7 +639,7 @@ def test_flag_flip_tolerates_parallel_confirm(psk): r1.confirm(True) # flips the slot to verified r2.confirm(True) # slot changed under it, but same credential assert r1.verified and r2.verified - assert _CREDENTIAL.unpack(cli["s"])[4] is True + assert _CREDENTIAL.unpack(cli["s"])[5] is True def test_commitment_reveal_checked(psk): @@ -695,8 +695,8 @@ 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._eph_m, hs._ephemeral, e_pub = _fresh_ephemeral() - body = _MSG1_REGISTER.pack(e_pub, kdf("commit", e_pub, r_a)) + hs._eph_m, hs._ephemeral, e_m, e_x = _fresh_ephemeral() + body = _MSG1_REGISTER.pack(e_m, e_x, kdf("commit", e_m, e_x, r_a)) msg1m = bytes(encrypt(body, key=kdf("reg/msg1", psk), ad=b"")) hs._chain.advance(msg1m) # startA was bypassed: advance the chain (m2b,) = b(msg1m) @@ -734,9 +734,8 @@ def test_foreign_mlkem_static_rejected_at_msg4(psk): # mlkem half, her own x25519 half): she cannot decapsulate bob's ct_se # challenge, so bob's confirm tag does not match her root and she burns a, b = make_sessions(psk) - real = a._exchange._static_pub - foreign = _public_from_idroot(secrets.token_bytes(32)) - a._exchange._static_pub = foreign[:1184] + real[1184:] + # alice presents a static whose ML-KEM half she does not hold + a._exchange._static_m = _public_from_idroot(secrets.token_bytes(32))[0] (m1,) = a() (m2,) = b(m1) (m3,) = a(m2) @@ -770,7 +769,7 @@ def test_credential_id(psk): run_handshake(a, b, answer=None) blob = a.credential # stable across the verified flag (metadata may change) - flagged = _CREDENTIAL.pack(*_CREDENTIAL.unpack(blob)[:4], True) + flagged = _CREDENTIAL.pack(*_CREDENTIAL.unpack(blob)[:5], True) assert credential_id(blob) == credential_id(flagged) # tied to the credential: a different record hashes differently assert credential_id(blob) != credential_id(psk) # rotation rotates it diff --git a/tests/test_ws.py b/tests/test_ws.py index 49b0fb9..d57c68e 100644 --- a/tests/test_ws.py +++ b/tests/test_ws.py @@ -186,7 +186,7 @@ async def test_verification_callback(): assert await srv.receive() == b"resumed" assert seen == [] # no comparison asked assert not srv.session.verified and not cli.session.verified - assert cli.session.verification.sas == _CREDENTIAL.unpack(cred_cli)[3] + assert cli.session.verification.sas == _CREDENTIAL.unpack(cred_cli)[4] cli.session.confirm(True) # compared out of band: mark verified srv.session.confirm(True) assert cli.session.verified and srv.session.verified