LeoVasanko/quantumtunnel
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

One x25519 ephemeral per party per connection

Browse Source 
Each exchange party ships a single x25519 ephemeral public (e_alice in
msg1, e_bob in msg2) and every x25519 leg is a plain DH against those
shipped keys; ML-KEM legs ship as bare ciphertexts. Encapsulation is
split into explicit ML-KEM + DH steps at the call sites, key objects
are kept as members (statics derived once from the persisted idroot,
ephemerals generated directly), and all fixed-width wire layouts are
full-message structs. Also: role-suffixed key naming throughout, dead
struct/layout cleanup.
This commit is contained in:
LeoVasanko committed 2026-10-07 04:35:31 +00:00
1 parent 03a7758fa6
commit 168287ca98
3 files changed
+150 -131
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/protocol.md
+34 -35
View File
@@ -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
quantumpipe/handshake.py
+108 -86
View File
@@ -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:
tests/test_sansio.py
+8 -10
View File
@@ -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)