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.
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""Canonical ZAP router envelope codec — round-trips (matches zapd/frame.rs)."""
|
|
|
|
import socket
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from zap.frame import (
|
|
ROLE_PROVIDER,
|
|
ROUTE,
|
|
Frame,
|
|
_put_str,
|
|
decode_cmd,
|
|
encode_cmd,
|
|
encode_hello,
|
|
parse_providers,
|
|
)
|
|
|
|
|
|
def test_envelope_roundtrip():
|
|
f = Frame(ROUTE, "consumer:mcp/1", "browser:chrome/dbc/default", b"opaque\x00\x01")
|
|
raw = f.encode()
|
|
# frame body length prefix + parse back via the header struct
|
|
assert raw[:4] == struct.pack("<I", len(raw) - 4)
|
|
|
|
|
|
def test_cmd_roundtrip_preserves_str():
|
|
"""A str value round-trips back to str (the type tag fix)."""
|
|
payload = encode_cmd("Page.navigate", {"url": "https://hanzo.ai"})
|
|
method, params = decode_cmd(payload)
|
|
assert method == "Page.navigate"
|
|
assert params["url"] == "https://hanzo.ai" # str, not bytes
|
|
|
|
|
|
def test_cmd_roundtrip_preserves_bytes():
|
|
"""A bytes value round-trips back to bytes."""
|
|
payload = encode_cmd("Raw.send", {"blob": b"\x00\xff\x10"})
|
|
_method, params = decode_cmd(payload)
|
|
assert params["blob"] == b"\x00\xff\x10"
|
|
|
|
|
|
def test_cmd_roundtrip_mixed():
|
|
payload = encode_cmd("m", {"s": "text", "b": b"\x01\x02"})
|
|
_method, params = decode_cmd(payload)
|
|
assert params == {"s": "text", "b": b"\x01\x02"}
|
|
|
|
|
|
def test_providers_roundtrip():
|
|
body = (
|
|
struct.pack("<H", 1)
|
|
+ _put_str("browser:chrome/dbc/default")
|
|
+ bytes([ROLE_PROVIDER])
|
|
+ _put_str("hanzo")
|
|
+ struct.pack("<H", 0)
|
|
)
|
|
provs = parse_providers(body)
|
|
assert provs[0].id == "browser:chrome/dbc/default" and provs[0].brand == "hanzo"
|
|
|
|
|
|
def test_hello_encodes():
|
|
assert encode_hello(ROLE_PROVIDER, "hanzo", ["browser.tabs"])[0] == ROLE_PROVIDER
|
|
|
|
|
|
def test_encode_rejects_oversize_id():
|
|
"""An id past the u16 wire limit is a clean ValueError, not a struct crash."""
|
|
huge = "x" * 70000
|
|
with pytest.raises(ValueError, match="routing id too long"):
|
|
Frame(ROUTE, huge, "to", b"").encode()
|
|
|
|
|
|
def test_read_tolerates_non_utf8_id():
|
|
"""Frame.read decodes non-UTF8 routing ids leniently (no crash)."""
|
|
# Build a frame by hand with a 0xFF byte in the `from` id.
|
|
frm = b"\xff\xfe"
|
|
body = struct.pack("<BHHHI", ROUTE, 0, len(frm), 0, 0) + frm
|
|
raw = struct.pack("<I", len(body)) + body
|
|
a, b = socket.socketpair()
|
|
try:
|
|
a.sendall(raw)
|
|
f = Frame.read(b)
|
|
assert f.typ == ROUTE
|
|
assert f.frm # decoded leniently, did not raise
|
|
finally:
|
|
a.close()
|
|
b.close()
|