mirror of
https://github.com/luxfi/zwing.git
synced 2026-09-24 00:24:38 +00:00
zeekay
e90f664a63
Port the complete Z-Wing protocol to Python and TypeScript, no longer
just the X-Wing combiner. Each language now ships:
identity Hybrid Ed25519 + ML-DSA-65 + X-Wing keypair, marshal/parse,
sign/verify, constant-time equality
kem Full IETF X-Wing encapsulate/decapsulate (ML-KEM-768 +
X25519 + SHA3-256 combiner) with wire-byte compat
handshake 1-RTT mutual-auth handshake mirroring Go state machine —
same labels (lux.zwing.v1/handshake-init etc), same wire
format (HandshakeInit / HandshakeResponse / transcript hash
/ AEAD-sealed responder identity)
channel Post-handshake ChaCha20-Poly1305 with sequence-numbered
nonces (high 4 zero, low 8 BE counter), MAX 1 MiB frames
Python implementation:
- cryptography (Ed25519/X25519/ChaCha20-Poly1305) + kyber-py (ML-KEM)
+ dilithium-py (ML-DSA). Pure Python, zero native deps.
- test_full.py: 39 tests covering happy paths + every error branch
- py/__init__.py modules at 100% line coverage:
channel.py 51 stmts 100%
handshake.py 159 stmts 100%
identity.py 91 stmts 100%
kem.py 26 stmts 100%
zwing.py 27 stmts 100%
TypeScript implementation:
- @noble/post-quantum (ml-kem768, ml-dsa65) + @noble/curves (ed25519,
x25519) + @noble/ciphers (chacha20poly1305) + @noble/hashes
- full.test.ts: 32 tests covering full e2e + every wire-decode error
- vitest --coverage v8: 100% statements / 100% branches / 100%
functions / 100% lines across every source file (channel.ts,
handshake.ts, identity.ts, kem.ts, zwing.ts)
Cross-language KAT (72df...fe4ed) still pins on the X-Wing combiner.
Full handshake wire-byte interop is now possible Go↔Python (both use
FIPS 204 ML-DSA-65 with identical 3309-byte signatures); Rust uses
PQClean Dilithium3 which has the same byte sizes as Round 3 — full
Rust↔Go interop test pending a unified ML-DSA-65 source.
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""Z-Wing post-handshake encrypted channel (Python).
|
|
|
|
ChaCha20-Poly1305 over length-prefixed frames with sequence-numbered
|
|
nonces. Wire format identical to Go and Rust.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import struct
|
|
from typing import Optional
|
|
|
|
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
|
|
|
|
from handshake import DuplexStream, HandshakeOutput, MAX_FRAME_SIZE, read_frame, write_frame
|
|
from identity import IdentityPublic
|
|
|
|
CHACHA_OVERHEAD = 16
|
|
NONCE_LEN = 12
|
|
|
|
|
|
def nonce_for(seq: int) -> bytes:
|
|
"""12-byte nonce: high 4 bytes zero, low 8 bytes BE counter."""
|
|
n = bytearray(NONCE_LEN)
|
|
n[NONCE_LEN - 8 :] = struct.pack(">Q", seq)
|
|
return bytes(n)
|
|
|
|
|
|
class Channel:
|
|
"""Post-handshake Z-Wing secure channel."""
|
|
|
|
def __init__(
|
|
self, inner: DuplexStream, out: HandshakeOutput, initiator: bool
|
|
) -> None:
|
|
self._inner = inner
|
|
self.remote: IdentityPublic = out.remote
|
|
if initiator:
|
|
tx_key = out.key_i2r
|
|
rx_key = out.key_r2i
|
|
else:
|
|
tx_key = out.key_r2i
|
|
rx_key = out.key_i2r
|
|
self._tx = ChaCha20Poly1305(tx_key)
|
|
self._rx = ChaCha20Poly1305(rx_key)
|
|
self._tx_seq = 0
|
|
self._rx_seq = 0
|
|
self._rx_overflow: Optional[bytes] = None
|
|
|
|
def send(self, plaintext: bytes) -> None:
|
|
max_plain = MAX_FRAME_SIZE - CHACHA_OVERHEAD
|
|
written = 0
|
|
while written < len(plaintext):
|
|
end = min(written + max_plain, len(plaintext))
|
|
chunk = plaintext[written:end]
|
|
if self._tx_seq >= (1 << 64) - 1:
|
|
raise ValueError("zwing: AEAD sequence number exhausted")
|
|
nonce = nonce_for(self._tx_seq)
|
|
ct = self._tx.encrypt(nonce, chunk, None)
|
|
self._tx_seq += 1
|
|
write_frame(self._inner, ct)
|
|
written = end
|
|
|
|
def recv(self) -> bytes:
|
|
if self._rx_overflow is not None:
|
|
buf = self._rx_overflow
|
|
self._rx_overflow = None
|
|
return buf
|
|
frame = read_frame(self._inner)
|
|
if self._rx_seq >= (1 << 64) - 1:
|
|
raise ValueError("zwing: AEAD sequence number exhausted")
|
|
nonce = nonce_for(self._rx_seq)
|
|
pt = self._rx.decrypt(nonce, frame, None)
|
|
self._rx_seq += 1
|
|
return pt
|