mirror of
https://github.com/zap-proto/py.git
synced 2026-10-30 15:04:09 +00:00
zeekay
e8ec8eed73
Make the published zap-proto package contain real code instead of stubs. Wire (the product, was entirely absent): - src/zap/wire.py — pure-stdlib port of the canonical Go runtime (zap.go + builder.py + accessors + list cursors). Message/Object/List zero-copy reads, Builder/ObjectBuilder/ListBuilder writes, 16-byte ZAP\x00 header, Version1+2 accept, reader hardening (reject backward pointers into the header, clamp list length). Byte-faithful to Go — pinned by the shared cross-wire golden vector. RPC (was a sleep-loop server + all-mock client): - src/zap/rpc.py — real TCP request/response transport, Server dispatches the Zap interface method ordinals (@0..@8 from zap.zap), Client calls them. - app.run() now serves real RPC; client.Client is a real RPC client. Crypto (was secrets.token_bytes() fakes whose two sides never agreed and whose verify() accepted any blob): - src/zap/crypto.py — real ML-KEM-768 / ML-DSA-65 (pqcrypto) + X25519 (cryptography), hybrid HKDF-SHA256 KEX. Hard-fails with CryptoError when a backend is missing — never fabricates bytes. Package wiring: - __init__.py exposes the wire API with ZERO third-party dep on the import path; pydantic/httpx are lazy (app.ZAP, Client). All __all__ names resolve on a default install (was 18 of 22 unresolved, swallowed ImportError). - pyproject: drop nonexistent pyzap>=2.0.0; pydantic -> [app]; add [crypto]; fix repo URL to zap-proto/py; target py312. Bug fixes (audit PoCs): - frame.py: reject oversize routing ids cleanly (was u16 struct crash); lenient decode of non-UTF8 ids (was strict-decode crash); encode/decode_cmd type tag so str round-trips to str (was silently bytes). - protocol.py: decode_stream reports bytes consumed (decode dropped 2nd frame). - app.py: _match_uri_template escapes literal template text (was regex over-match). - consensus.py: fix votes int/dict type collision. Quality: ruff clean, ruff format clean, mypy --strict 0 issues, 52 tests pass. Add .github/workflows/ci.yml (ruff + format + mypy + pytest, py3.10-3.13). Rebrand: zap.capnp -> zap.zap, $Go.import -> github.com/zap-proto/go.
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
"""Real post-quantum crypto — the two sides agree, verify rejects forgeries.
|
|
|
|
These tests require the ``[crypto]`` extra. They are skipped (not faked) when the
|
|
backend is unavailable; the module itself raises ``CryptoError`` rather than
|
|
returning fabricated bytes.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from zap import crypto
|
|
from zap.crypto import (
|
|
CryptoError,
|
|
HybridKeyExchange,
|
|
MLDSAKeyPair,
|
|
MLKEMKeyPair,
|
|
X25519KeyPair,
|
|
)
|
|
|
|
pq = pytest.mark.skipif(not crypto.PQ_AVAILABLE, reason="pqcrypto not installed")
|
|
x = pytest.mark.skipif(not crypto.X25519_AVAILABLE, reason="cryptography not installed")
|
|
|
|
|
|
@pq
|
|
def test_mlkem_encapsulate_decapsulate_agree():
|
|
kp = MLKEMKeyPair.generate()
|
|
ciphertext, secret_a = kp.encapsulate()
|
|
secret_b = kp.decapsulate(ciphertext)
|
|
assert secret_a == secret_b
|
|
assert len(secret_a) == crypto.SHARED_SECRET_SIZE
|
|
|
|
|
|
@pq
|
|
def test_mldsa_sign_verify():
|
|
kp = MLDSAKeyPair.generate()
|
|
sig = kp.sign(b"zap message")
|
|
assert kp.verify(b"zap message", sig) is True
|
|
|
|
|
|
@pq
|
|
def test_mldsa_rejects_tampered_message():
|
|
kp = MLDSAKeyPair.generate()
|
|
sig = kp.sign(b"zap message")
|
|
assert kp.verify(b"tampered", sig) is False
|
|
|
|
|
|
@pq
|
|
def test_mldsa_rejects_garbage_signature():
|
|
"""A random blob of the right length must NOT verify (no universal forgery)."""
|
|
kp = MLDSAKeyPair.generate()
|
|
forged = bytes(crypto.MLDSA_SIGNATURE_SIZE)
|
|
assert kp.verify(b"zap message", forged) is False
|
|
|
|
|
|
@x
|
|
def test_x25519_exchange_agrees():
|
|
a = X25519KeyPair.generate()
|
|
b = X25519KeyPair.generate()
|
|
assert a.exchange(b.public_key) == b.exchange(a.public_key)
|
|
|
|
|
|
@pq
|
|
@x
|
|
def test_hybrid_handshake_both_sides_agree():
|
|
"""The headline guarantee: initiator and responder derive the same secret."""
|
|
a = HybridKeyExchange.generate()
|
|
b = HybridKeyExchange.generate()
|
|
a_x_pub, a_k_pub = a.initiate()
|
|
b_x_pub, ciphertext, secret_responder = b.respond(a_x_pub, a_k_pub)
|
|
secret_initiator = a.finalize(b_x_pub, ciphertext)
|
|
assert secret_initiator == secret_responder
|
|
assert len(secret_initiator) == crypto.SHARED_SECRET_SIZE
|
|
|
|
|
|
def test_hard_fail_when_unavailable(monkeypatch):
|
|
"""With the backend disabled, generate() raises CryptoError — never fakes."""
|
|
monkeypatch.setattr(crypto, "PQ_AVAILABLE", False)
|
|
with pytest.raises(CryptoError):
|
|
MLKEMKeyPair.generate()
|
|
with pytest.raises(CryptoError):
|
|
MLDSAKeyPair.generate()
|