diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5b2430c..ba132fa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,9 +25,9 @@ jobs: go-version: "1.27" - name: Install the package and the test tools run: python -m pip install -e ".[dev]" - - name: Build awp, for the conformance suite - run: go install github.com/agentwireprotocol/awp/cmd/awp@v0.6.0 + - name: Build awp, for the tunnel and the conformance suite + run: go install github.com/agentwireprotocol/awp/cmd/awp@1bd8112984bf9a323034d2fc031e019c1bbbac4a - name: Tests, including the conformance suite run: AWP_BIN="$(go env GOPATH)/bin/awp" python -m pytest -v - name: Tests with the pure-Python Ed25519 - run: AWP_PURE_ED25519=1 python -m pytest -q -k "conversation" + run: AWP_BIN="$(go env GOPATH)/bin/awp" AWP_PURE_ED25519=1 python -m pytest -q -k "conversation or pure_python" diff --git a/README.md b/README.md index a1091da..f23cb6b 100644 --- a/README.md +++ b/README.md @@ -2,9 +2,12 @@ The Python SDK for the [Agent Wire Protocol](https://agentwireprotocol.com) (AWP), the peer-to-peer messaging protocol for coding agents. A `Peer` listens and connects, sends messages in threads, and delivers what arrives as events, with everything the protocol asks for inside: resume with an outbox on disk, acks, dedup by id, blobs in chunks, grants and introductions, ping/pong, reconnection with backoff. -Python 3.11 or later, asyncio, no required dependencies. Ed25519 uses the [`cryptography`](https://pypi.org/project/cryptography/) package when it is installed and a pure-Python implementation otherwise. +Python 3.11 or later, asyncio, no required Python dependencies. Ed25519 uses the [`cryptography`](https://pypi.org/project/cryptography/) package when it is installed and a pure-Python implementation otherwise. + +Every connection is a WireGuard tunnel between the two peers' keys (SPEC.md draft 2). Python has no WireGuard of its own, so a Peer runs the reference implementation's `awp tunnel` as a helper process, which needs the `awp` binary: ```sh +curl -fsSL https://agentwireprotocol.com/install.sh | sh # awp, for the tunnel pip install awp # or: pip install "awp[crypto]" ``` @@ -16,10 +19,10 @@ import awp async def main(): async with awp.Peer("~/.mybot", name="mybot@builder") as peer: - address = await peer.listen("tailcat") # or "tcp:127.0.0.1:7000", "unix:/tmp/awp.sock" - print("share this:", address) + address = await peer.listen("tailcat") # or "udp:0.0.0.0:41641", "cloudflare", "unix:/tmp/awp.sock" + print("share this:", address) # awp1... - key = await peer.connect("tc...") # an address shared out of band + key = await peer.connect("awp1...") # an address shared out of band sent = peer.send(key, "Please run make test at a1b2c3.", subject="Run the suite") async for event in peer.events(): @@ -36,26 +39,26 @@ asyncio.run(main()) ## What a Peer does -- **`Peer(state_dir, name=...)`** loads or creates the Ed25519 identity under `state_dir`, along with the outbox, received ids, threads, grants and blobs. `None` uses a temporary directory with a fresh identity, removed on close. Use it as an async context manager, or call `await peer.start()` and `await peer.close()`. -- **`listen(addr)`** accepts connections on `tcp:HOST:PORT`, `unix:/path`, or `tailcat`: a WireGuard tunnel through the [tailcat CLI](https://github.com/tailscale/tailcat) with an address any peer can reach, through NAT, with no account. It returns the address to share. -- **`connect(addr)`** dials an address and returns the peer's key once the handshake is done. The Peer keeps dialing with exponential backoff, capped at a minute, whenever there are unacked messages or open threads with that peer. Sleeping sandboxes wake on connect. +- **`Peer(state_dir, name=..., awp=None)`** loads or creates the Ed25519 identity under `state_dir`, along with the outbox, received ids, threads, grants and blobs. `None` uses a temporary directory with a fresh identity, removed on close. `awp` names the binary (default `$AWP_BIN` or `awp` on PATH). Use it as an async context manager, or call `await peer.start()` and `await peer.close()`. +- **`listen(carrier)`** accepts connections on a carrier and returns the address to share (`awp1...`), which lists every carrier the Peer listens on: `tailcat` (reachable from anywhere, through NAT, no account), `udp:HOST:PORT`, `ws:HOST:PORT` or `ws:HOST:PORT=URL`, `cloudflare` (a WebSocket through a Cloudflare quick tunnel; needs `cloudflared`), `unix:/path`. **`address`** is the current one. **`rotate_psk()`** replaces its admission secret: strangers holding old copies are shut out, peers already met are not. +- **`connect(address)`** dials an address and returns the peer's key once the handshake is done. An explicit `connect` also lifts a `bye`. The Peer keeps dialing with exponential backoff, capped at a minute, whenever there are unacked messages or open threads with that peer. Sleeping sandboxes wake on connect. - **`send(to, text, thread=None, subject=None, reply_to=None, parts=None, files=())`** queues a message. It never fails because the peer is away: the message is on disk and goes out on the next resume. Files travel as blobs in 256 KiB chunks ahead of the message. **`wait_ack(id)`** waits for the peer's ack; `Acked` events carry it too. - **`set_state(to, thread, state, note=None)`** sets this side's state on a thread: `working`, `waiting`, `done`, `failed`, `closed`, or any word the two agents agree on. - **`events()`** and **`next_event(timeout)`**: `Connected`, `Disconnected`, `Message`, `State`, `Acked`, `Blob`, `GrantReceived`, `Introduced`, `PeerError`, `Bye`, `Error`. Messages are stored and acked before they are reported. -- **`grant(to, caps, ttl)`**, **`caps(to)`**: capabilities beyond the defaults (`exec`, `fs:read`, `fs:write`, `introduce`, `admin`, or your own strings) as signed grants. Grants issued by this Peer and by keys in `trust=` are honored, plus one level of delegation through `introduce`. A request in a message (a data part of a `vnd.awp` type) arrives in `Message.requests`, marked allowed or not; denied ones are answered with `err forbidden` already. +- **`grant(to, caps, ttl)`**, **`caps(to)`**, **`grants()`**, **`revoke(hash)`**, **`introduce(to, peer, caps)`**: capabilities beyond the defaults (`exec`, `fs:read`, `fs:write`, `introduce`, `admin`, or your own strings) as signed grants. Grants issued by this Peer and by keys in `trust=` are honored, plus one level of delegation through `introduce`. Grants this Peer holds are presented in its hello to the peer they concern. A request in a message (a data part of a `vnd.awp` type) arrives in `Message.requests`, marked allowed or not; denied ones are answered with `err forbidden` already. - **`bye(to)`** closes gracefully and parks the peer. **`list_peers()`**, **`threads()`**, **`connected(to)`**: what the Peer knows. ## A peer driven over stdin and stdout -`python -m awp --state DIR listen tcp:127.0.0.1:7000` runs a peer that reads JSON commands on stdin and writes JSON events on stdout, for tests and for other languages. The [reference implementation's interop tests](https://github.com/agentwireprotocol/awp/tree/main/python) drive it. +`python -m awp --state DIR listen udp:127.0.0.1:0` (or `connect awp1...`) runs a peer that reads JSON commands on stdin and writes JSON events on stdout, for tests and for other languages. `awp conform --listen udp:127.0.0.1:0 --run 'python -m awp --state DIR connect {addr}'` checks it. ## Conformance -`pytest` runs a two-peer conversation with states, a reply, a multi-chunk blob, grants and bye; queued delivery across a `kill -9`; and, when `AWP_BIN` points at an `awp` binary, the protocol's conformance suite ([`awp conform`](https://docs.agentwireprotocol.com/reference/conformance)) against a `Peer`, both with the Peer listening and with the Peer dialing. CI does all of it. +`pytest` needs an `awp` binary (`AWP_BIN`, or on PATH). It runs a two-peer conversation with states, a reply, a multi-chunk blob, grants and bye; queued delivery across a `kill -9`; introductions with a grant presented in hello; pre-shared key rotation; a raw client for the protocol's edges (a hello naming another key, bad versions, bad frames, oversized lines); and the protocol's conformance suite ([`awp conform`](https://docs.agentwireprotocol.com/reference/conformance)) against a `Peer`, both with the Peer listening and with the Peer dialing. CI does all of it. ## Status -v0.1. The API may change before v1; the wire protocol is v0 and stable. +v0.2. The API may change before v1. It speaks protocol v1 (SPEC.md draft 2). ## License diff --git a/pyproject.toml b/pyproject.toml index 0627d2a..e66b2ed 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "awp" -version = "0.1.0" +version = "0.2.0" description = "The Python SDK for the Agent Wire Protocol: peer-to-peer messaging between coding agents." readme = "README.md" license = "Apache-2.0" diff --git a/src/awp/__init__.py b/src/awp/__init__.py index aa1e0a8..61679d8 100644 --- a/src/awp/__init__.py +++ b/src/awp/__init__.py @@ -6,7 +6,7 @@ async with awp.Peer("~/.mybot", name="mybot@host") as peer: address = await peer.listen("tailcat") - key = await peer.connect("tc...") + key = await peer.connect("awp1...") peer.send(key, "Please run make test.", subject="Run the suite") async for event in peer.events(): if isinstance(event, awp.Message): @@ -19,7 +19,7 @@ from .wire import (AwpError, CommandError, StartupError, PROTOCOL_VERSION, mint_grant, verify_grant) -__version__ = "0.1.0" +__version__ = "0.2.0" __all__ = [ "Peer", "PeerInfo", "Sent", "Thread", diff --git a/src/awp/cli.py b/src/awp/cli.py index 81268ba..5d25820 100644 --- a/src/awp/cli.py +++ b/src/awp/cli.py @@ -2,8 +2,10 @@ :: - python -m awp --state DIR [--name NAME] listen tcp:127.0.0.1:7000 - python -m awp --state DIR [--name NAME] connect tc... + python -m awp --state DIR [--name NAME] listen tailcat (or udp:HOST:PORT, ws:HOST:PORT, cloudflare, unix:/path) + python -m awp --state DIR [--name NAME] connect awp1... + +The tunnel is `awp tunnel` (--awp, $AWP_BIN, or awp on PATH). One JSON command per line on stdin, one JSON event per line on stdout: @@ -166,7 +168,7 @@ def reader() -> None: async def amain(args: argparse.Namespace) -> int: try: peer = Peer(args.state, name=args.name, about=args.about, trust=args.trust or (), - ping_interval=args.ping_interval, blob_limit=args.max_blob) + ping_interval=args.ping_interval, blob_limit=args.max_blob, awp=args.awp) except AwpError as e: emit({"event": "error", "detail": str(e)}) return 1 @@ -210,11 +212,13 @@ def main(argv: list[str] | None = None) -> int: help="largest blob accepted, in bytes (default 50 MiB)") ap.add_argument("--trust", action="append", default=[], help="trust grants issued by this key (repeatable)") ap.add_argument("--about", default=None, help="free text for hello.about") + ap.add_argument("--awp", default=None, help="the awp binary, whose `awp tunnel` carries the connections") ap.add_argument("-v", "--verbose", action="store_true", help="debug logging on stderr") sub = ap.add_subparsers(dest="mode", required=True, metavar="{listen,connect}") - for mode in ("listen", "connect"): - sp = sub.add_parser(mode, help=f"{mode} on ADDR (tcp:HOST:PORT, unix:/path, or tailcat)") - sp.add_argument("addr") + sp = sub.add_parser("listen", help="listen on a carrier: tailcat, udp:HOST:PORT, ws:HOST:PORT, cloudflare or unix:/path") + sp.add_argument("addr", metavar="carrier") + sp = sub.add_parser("connect", help="connect to an address (awp1...)") + sp.add_argument("addr", metavar="address") args = ap.parse_args(argv) logging.basicConfig(stream=sys.stderr, level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s [awp] %(message)s", datefmt="%H:%M:%S") diff --git a/src/awp/connection.py b/src/awp/connection.py index 4712694..8ca2515 100644 --- a/src/awp/connection.py +++ b/src/awp/connection.py @@ -1,5 +1,7 @@ -"""One connection to a peer: the handshake, the reader, the FIFO writer and -the timers (SPEC.md sections 7 and 9).""" +"""One connection to a peer: the hello exchange, the reader, the FIFO +writer and the timers (SPEC.md sections 10 and 12). The stream comes from +the tunnel helper, which has already checked that the peer's hello names +the key on the other end of the tunnel.""" from __future__ import annotations @@ -9,15 +11,14 @@ import traceback from ._log import log, debug -from .wire import (ACKED_TYPES, AUTH_CONTEXT, BYE_TIMEOUT, CLOSING_ERR_CODES, HANDSHAKE_TIMEOUT, - MAX_LINE, PRE_AUTH_FORBIDDEN, PROTOCOL_VERSION, READ_SIZE, RESUME_TIMEOUT, - b64decode_any, b64url, dumps_line, parse_key) -from .wire import ed25519_verify +from .wire import (ACKED_TYPES, BYE_TIMEOUT, CLOSING_ERR_CODES, HANDSHAKE_TIMEOUT, + MAX_LINE, PRE_HELLO_FORBIDDEN, PROTOCOL_VERSION, READ_SIZE, RESUME_TIMEOUT, + dumps_line, parse_key, x25519_public) class Connection: """One transport connection: handshake, reader, FIFO writer, timers.""" - def __init__(self, node: "Node", reader, writer, label: str): + def __init__(self, node: "Node", reader, writer, label: str, remote: bytes | None = None): self.node = node self.reader = reader self.writer = writer @@ -29,7 +30,10 @@ def __init__(self, node: "Node", reader, writer, label: str): self.closing = False self.close_reason = None self.linger = False - self.hello = node.make_hello() + # remote is the peer's tunnel (X25519) key, from the helper: it + # picks the grants worth presenting before the peer's hello arrives. + self.remote = remote + self.hello = node.make_hello(remote) self.my_hello_line = dumps_line(self.hello) self.peer_hello_line = None self.peer_hello = None @@ -358,29 +362,14 @@ def _handshake(self, line: bytes, obj: dict, t) -> None: if raw == node.identity.pub or line == self.my_hello_line: self.fail("auth", "hello carries our own key (reflected handshake)", re=oid) return + if self.remote is not None and x25519_public(raw) != self.remote: + self.fail("auth", "hello key is not the key on the other end of the tunnel", re=oid) + return self.peer_hello_line = line self.peer_hello = obj self.peer_key_raw = raw self.peer_key_str = obj.get("key") log(f"{self.label}: hello from {self.peer_key_str} name={obj.get('name')!r}") - sig = node.identity.sign(AUTH_CONTEXT + b"\x00" + self.my_hello_line + b"\x00" + line) - held = node.held.valid() - auth = node.envelope("auth", sig=b64url(sig), grants=held if held else None) - self._write(dumps_line(auth)) - return - if t == "auth": - if self.peer_hello is None: - self.fail("auth", "auth received before hello", re=oid) - return - node.emit_recv(self, obj) - try: - sig = b64decode_any(obj.get("sig")) - except ValueError: - sig = b"" - transcript = AUTH_CONTEXT + b"\x00" + self.peer_hello_line + b"\x00" + self.my_hello_line - if not ed25519_verify(self.peer_key_raw, sig, transcript): - self.fail("auth", "auth signature does not verify", re=oid) - return self._on_established(obj) return if t == "err": @@ -396,22 +385,22 @@ def _handshake(self, line: bytes, obj: dict, t) -> None: node.emit_recv(self, obj) self.close("bye before handshake completed") return - if t in PRE_AUTH_FORBIDDEN: - self.fail("auth", f"{t!r} received before the handshake completed", re=oid) + if t in PRE_HELLO_FORBIDDEN: + self.fail("bad_frame", f"{t!r} received before hello", re=oid) return node.emit_recv(self, obj) log(f"{self.label}: ignoring unknown message type {t!r} during handshake") - def _on_established(self, auth: dict) -> None: + def _on_established(self, hello: dict) -> None: node = self.node self.established = True self.established_at = self.loop.time() self.ps = node.bind_peer(self.peer_key_raw, self.peer_key_str, self.peer_hello) node.register_connection(self) - grants = auth.get("grants") + grants = hello.get("grants") if isinstance(grants, list): for g in grants: - node.receive_grant(self, g, "auth") + node.receive_grant(self, g, "hello") h = self.peer_hello name = h.get("name") if isinstance(h.get("name"), str) else "" caps = h.get("caps") if isinstance(h.get("caps"), list) else [] @@ -462,8 +451,8 @@ def _dispatch(self, obj: dict, t) -> None: self._on_bye(obj) elif t == "err": self._on_err(obj) - elif t in ("hello", "auth"): - log(f"{self.label}: ignoring {t} after the handshake") + elif t == "hello": + log(f"{self.label}: ignoring hello after the handshake") else: debug(f"{self.label}: ignoring unknown message type {t!r}") diff --git a/src/awp/peer.py b/src/awp/peer.py index b9eb096..1c0ede9 100644 --- a/src/awp/peer.py +++ b/src/awp/peer.py @@ -22,13 +22,13 @@ from .connection import Connection from .store import (PENDING, GrantList, Identity, PeerState, _rec_bytes, atomic_write_json, load_json) -from .transport import (TailcatListener, fmt_host, is_loopback, make_tcp_listener, open_stream, - parse_addr, prepare_unix_path) +from .transport import Tunnel, TunnelError, find_awp, remote_from_preamble from .wire import (ACKED_TYPES, BACKOFF_CAP, BACKOFF_INITIAL, CHUNK_SIZE, DEFAULT_BLOB_LIMIT, DEFAULT_GRANT_TTL, DEFAULT_PING_INTERVAL, FSYNC, MAX_LINE, MY_CAPS, PROTOCOL_VERSION, REQUEST_MIMES, CommandError, StartupError, UlidGen, - b64std, b64url, blob_refs, dumps_line, format_key, grant_hash, key_fp, key_matches, - mint_grant, now_ts, parse_key, safe_name, verify_grant) + b64decode_any, b64std, blob_refs, dumps_line, format_key, grant_hash, + is_address, key_fp, key_matches, mint_grant, now_ts, parse_key, safe_name, + verify_grant, x25519_public) @dataclass(frozen=True) @@ -69,8 +69,8 @@ class Peer: :: async with awp.Peer("~/.mybot", name="mybot@host") as peer: - await peer.listen("tcp:127.0.0.1:7000") - key = await peer.connect("tcp:10.0.0.2:7000") + address = await peer.listen("tailcat") # awp1...: share it + key = await peer.connect("awp1...") # an address shared with you peer.send(key, "Please run make test.", subject="Run the suite") async for event in peer.events(): ... @@ -80,12 +80,17 @@ class Peer: blobs. ``None`` uses a temporary directory with a fresh identity, removed on close. ``send`` never fails because the peer is away: the message is on disk and goes out on the next resume. + + Every connection is a WireGuard tunnel between the two peers' keys. The + tunnel is ``awp tunnel``, the reference implementation's, run as a + helper process: ``awp`` names the binary (default ``$AWP_BIN`` or + ``awp`` on PATH). """ def __init__(self, state_dir: str | os.PathLike[str] | None = None, *, name: str | None = None, about: str | None = None, trust: Iterable[str] = (), ping_interval: float = DEFAULT_PING_INTERVAL, blob_limit: int = DEFAULT_BLOB_LIMIT, - caps: Iterable[str] | None = None, tailcat: str = "tailcat") -> None: + caps: Iterable[str] | None = None, awp: str | None = None) -> None: self._ephemeral = state_dir is None self.state_dir = (tempfile.mkdtemp(prefix="awp-") if state_dir is None else os.path.abspath(os.path.expanduser(os.fspath(state_dir)))) @@ -103,7 +108,7 @@ def __init__(self, state_dir: str | os.PathLike[str] | None = None, *, name: str self.ping_interval = max(0.05, float(ping_interval)) if ping_interval > 0 else 0.0 self.blob_limit = int(blob_limit) self.hello_caps = list(caps) if caps is not None else list(MY_CAPS) - self.tailcat_bin = tailcat + self.awp_bin = awp self.ids = UlidGen() self.identity = Identity.load_or_create(os.path.join(self.state_dir, "identity.json")) self.key = self.identity.key @@ -135,10 +140,11 @@ def __init__(self, state_dir: str | os.PathLike[str] | None = None, *, name: str if isinstance(dp, dict) and dp.get("fp") in self.peers: self.default_fp = dp["fp"] self._parked: set[str] = set() - self._servers: list[Any] = [] - self._tailcat: TailcatListener | None = None + self._tunnel: Tunnel | None = None + self._tunnel_lock: asyncio.Lock | None = None self._dialers: dict[str, asyncio.Task[None]] = {} - self._conn_procs: dict[Connection, Any] = {} + self._dial_waiters: dict[str, asyncio.Future[str]] = {} + self._want_dial: set[str] = set() self._queue: asyncio.Queue[ev.Event] | None = None self._pending_events: list[ev.Event] = [] self.wake: asyncio.Event | None = None @@ -186,8 +192,6 @@ async def close(self) -> None: except (asyncio.CancelledError, Exception): pass self._dialers.clear() - for s in self._servers: - s.close() for c in list(self.all_conns): c.close("closing") for c in list(self.all_conns): @@ -195,20 +199,9 @@ async def close(self) -> None: await asyncio.wait_for(c.done.wait(), 2) except asyncio.TimeoutError: pass - for s in self._servers: - try: - await asyncio.wait_for(s.wait_closed(), 2) - except (asyncio.TimeoutError, Exception): - pass - self._servers.clear() - if self._tailcat is not None: - await self._tailcat.close() - self._tailcat = None - for proc in list(self._conn_procs.values()): - try: - proc.kill() - except ProcessLookupError: - pass + if self._tunnel is not None: + await self._tunnel.close() + self._tunnel = None try: os.close(self._lock_fd) except OSError: @@ -218,52 +211,68 @@ async def close(self) -> None: # -- listening and connecting -------------------------------------------------- - @property - def addresses(self) -> list[str]: - """The addresses this Peer listens on, the tailcat one first.""" - out = [f"tailcat:{self._tailcat.address}"] if self._tailcat and self._tailcat.address else [] - return out + [s.awp_addr for s in self._servers if getattr(s, "awp_addr", None)] - - async def listen(self, addr: str = "tailcat") -> str: - """Accept connections on ``tcp:HOST:PORT``, ``unix:/path``, or - ``tailcat`` (a WireGuard tunnel through the tailcat CLI, reachable - from anywhere). Returns the address to share, with the port that - was bound.""" + async def _ensure_tunnel(self) -> Tunnel: + """Start ``awp tunnel`` for this Peer, once.""" await self.start() - if addr in ("tailcat", "tailcat:"): - sock = make_tcp_listener("127.0.0.1", 0) - server = await asyncio.start_server(self._accept, sock=sock, limit=2 * MAX_LINE) - port = sock.getsockname()[1] - server.awp_addr = None # type: ignore[attr-defined] - self._servers.append(server) - self._tailcat = await TailcatListener.start(self.tailcat_bin, port) - shown = f"tailcat:{self._tailcat.address}" - log(f"listening on {shown} (tailcat serve 1:127.0.0.1:{port})") - return shown - target = parse_addr(addr) - if target[0] == "tcp": - _, host, port = target - if not is_loopback(host): - log("warning: plaintext TCP on a non-loopback address has no transport encryption " - "(spec 4.2); use it on trusted networks only") - sock = make_tcp_listener(host, port) - server = await asyncio.start_server(self._accept, sock=sock, limit=2 * MAX_LINE) - shown = f"tcp:{fmt_host(host)}:{sock.getsockname()[1]}" - else: - path = target[1] - prepare_unix_path(path) - server = await asyncio.start_unix_server(self._accept, path=path, limit=2 * MAX_LINE) - shown = f"unix:{path}" - server.awp_addr = shown # type: ignore[attr-defined] - self._servers.append(server) - log(f"listening on {shown}") - return shown + if self._tunnel_lock is None: + self._tunnel_lock = asyncio.Lock() + async with self._tunnel_lock: + if self._tunnel is None: + t = Tunnel(find_awp(self.awp_bin), os.path.join(self.state_dir, "identity.json"), + os.path.join(self.state_dir, "tunnel")) + await t.start(self._accept) + self._tunnel = t + log(f"tunnel up (awp tunnel, pid {t.proc.pid if t.proc else '?'})") + return self._tunnel + + @property + def address(self) -> str | None: + """The address to share (``awp1...``): this Peer's key, the + pre-shared key that admits peers it has not met, and every carrier + it listens on. None until it listens on something.""" + return self._tunnel.address if self._tunnel else None + + async def listen(self, carrier: str = "tailcat") -> str: + """Accept connections on a carrier, and return the address to share, + which lists every carrier this Peer listens on: + + ``tailcat`` + reachable from anywhere, through NAT, with no account + ``udp:HOST:PORT`` + plain UDP: a LAN, Fly's 6PN, a public address + ``ws:HOST:PORT`` or ``ws:HOST:PORT=URL`` + WebSockets, or behind a proxy or HTTP tunnel at URL + ``cloudflare`` + WebSockets through a Cloudflare quick tunnel (needs cloudflared) + ``unix:/path`` + peers on the same machine + + tailcat and cloudflare take a few seconds to come up.""" + t = await self._ensure_tunnel() + try: + address = await t.listen(carrier) + except TunnelError as e: + raise StartupError(f"cannot listen on {carrier}: {e}") from None + log(f"listening on {carrier}") + return address + + async def rotate_psk(self) -> str | None: + """Replace the pre-shared key in this Peer's address. Copies of the + address shared before stop admitting peers not met yet; peers + already met keep working. Returns the new address.""" + t = await self._ensure_tunnel() + return await t.rotate() async def _accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + """A stream a peer opened, forwarded by the tunnel helper.""" + try: + pre = await asyncio.wait_for(reader.readline(), 10) + except (asyncio.TimeoutError, OSError): + writer.close() + return + remote = remote_from_preamble(pre) self.conn_counter += 1 - peer = writer.get_extra_info("peername") - where = f"{peer[0]}:{peer[1]}" if isinstance(peer, tuple) and len(peer) >= 2 else "unix peer" - conn = Connection(self, reader, writer, f"conn#{self.conn_counter} from {where}") + conn = Connection(self, reader, writer, f"conn#{self.conn_counter} in", remote=remote) conn.outbound = False # type: ignore[attr-defined] try: await conn.run() @@ -275,20 +284,26 @@ async def connect(self, addr: str, *, timeout: float | None = 30.0) -> str: peer that answered, once the handshake is done. The Peer keeps dialing with backoff, capped at a minute, whenever there are unacked messages or open threads with that peer.""" - await self.start() - target = parse_addr(addr) - if target[0] == "tcp" and not is_loopback(target[1]): - log("warning: plaintext TCP to a non-loopback address; use it on trusted networks only") + if not is_address(addr): + raise StartupError(f"{addr!r} is not an address; addresses start with awp1") + addr = addr.strip() + await self._ensure_tunnel() loop = asyncio.get_running_loop() fut: asyncio.Future[str] = loop.create_future() + self._want_dial.add(addr) # dial now, work or not if addr in self._dialers and not self._dialers[addr].done(): for ps in self.peers.values(): - if ps.meta.get("dialed") == addr and ps.fp in self.connections: - return ps.key_str or format_key(ps.key_raw) + if ps.meta.get("dialed") == addr: + if ps.fp in self.connections: + self._want_dial.discard(addr) + return ps.key_str or format_key(ps.key_raw) + self._parked.discard(ps.fp) # an explicit connect lifts a bye fut = self._dial_waiters.setdefault(addr, fut) + assert self.wake is not None + self.wake.set() else: self._dial_waiters[addr] = fut - self._dialers[addr] = asyncio.create_task(self._dial_loop(addr, target)) + self._dialers[addr] = asyncio.create_task(self._dial_loop(addr)) try: return await asyncio.wait_for(asyncio.shield(fut), timeout) except asyncio.TimeoutError: @@ -298,35 +313,43 @@ async def connect(self, addr: str, *, timeout: float | None = 30.0) -> str: self._dial_waiters.pop(addr, None) raise ConnectionError(f"no handshake with {addr} within {timeout}s") from None - _dial_waiters: dict[str, asyncio.Future[str]] = {} - async def _dial_loop(self, addr: str, target: tuple[Any, ...]) -> None: + async def _dial_loop(self, addr: str) -> None: delay = BACKOFF_INITIAL ps: PeerState | None = None try: while not self._closed: - if ps is not None and (ps.fp in self._parked or not (ps.has_work() or self.pending.has_work())): + idle = lambda: (addr not in self._want_dial and ps is not None and # noqa: E731 + (ps.fp in self._parked or not (ps.has_work() or self.pending.has_work()))) + if idle(): assert self.wake is not None self.wake.clear() - if ps.fp in self._parked or not (ps.has_work() or self.pending.has_work()): + if idle(): log(f"{addr}: idle; waiting for new work") await self.wake.wait() continue + addrs = [addr] + if ps is not None and is_address(ps.meta.get("addr")) and ps.meta["addr"] != addr: + addrs.append(ps.meta["addr"]) # where the peer said it is, in hello try: - reader, writer, proc = await asyncio.wait_for(open_stream(target, self.tailcat_bin), timeout=15) - except (OSError, asyncio.TimeoutError, StartupError) as e: + assert self._tunnel is not None + reader, writer, rep = await self._tunnel.dial(addrs, timeout=60) + except (OSError, asyncio.TimeoutError, ValueError, AssertionError) as e: d = delay * random.uniform(0.9, 1.1) log(f"connect {addr} failed: {e or type(e).__name__}; retrying in {d:.2f}s") await asyncio.sleep(d) delay = min(delay * 2, BACKOFF_CAP) continue self.conn_counter += 1 - conn = Connection(self, reader, writer, f"conn#{self.conn_counter} to {addr}") + try: + remote = b64decode_any(rep.get("remote")) + except ValueError: + remote = None + conn = Connection(self, reader, writer, f"conn#{self.conn_counter} to {addr[:16]}…", remote=remote) conn.outbound = True # type: ignore[attr-defined] conn.dialed = addr # type: ignore[attr-defined] - if proc is not None: - self._conn_procs[conn] = proc established = await conn.run() + self._want_dial.discard(addr) if established: ps = conn.ps delay = BACKOFF_INITIAL @@ -340,9 +363,20 @@ async def _dial_loop(self, addr: str, target: tuple[Any, ...]) -> None: # -- what the connections call --------------------------------------------------- - def make_hello(self) -> dict[str, Any]: + def make_hello(self, remote: bytes | None = None) -> dict[str, Any]: + """Our hello: our address without its pre-shared key, so the peer can + reconnect to us, and the grants we hold that concern the peer on the + other end of the tunnel (ones it issued, and introductions meant for + it).""" + grants = [] + if remote is not None: + for g in self.held.valid(): + if any(isinstance(g.get(f), str) and _tunnel_key(g[f]) == remote for f in ("iss", "aud")): + grants.append(g) + public = self._tunnel.public if self._tunnel else None return self.envelope("hello", v=PROTOCOL_VERSION, key=self.key, name=self.name, - nonce=b64url(os.urandom(32)), caps=list(self.hello_caps), about=self.about) + caps=list(self.hello_caps), about=self.about, addr=public, + grants=grants or None) def make_err(self, code: str, detail: str, re: str | None = None, ref: str | None = None) -> dict[str, Any]: return self.envelope("err", re=re, code=code, detail=detail, ref=ref) @@ -473,12 +507,6 @@ def on_conn_closed(self, conn: Connection) -> None: self.all_conns.discard(conn) if conn.ps is not None and self.connections.get(conn.ps.fp) is conn: del self.connections[conn.ps.fp] - proc = self._conn_procs.pop(conn, None) - if proc is not None: - try: - proc.kill() - except ProcessLookupError: - pass reason = conn.close_reason or "closed" log(f"{conn.label}: closed ({reason})") if conn.established: @@ -821,7 +849,7 @@ def on_introduce(self, conn: Connection, obj: dict[str, Any]) -> None: log(f"{conn.label}: introduced to {key}") self._emit(ev.Introduced(peer=conn.peer_key_str or "", key=key, name=peer.get("name") if isinstance(peer.get("name"), str) else None, - address=peer.get("address") if isinstance(peer.get("address"), str) else None, + address=peer.get("address") if is_address(peer.get("address")) else None, thread=obj.get("th") if isinstance(obj.get("th"), str) else None, grant=dict(g) if isinstance(g, dict) else None)) @@ -952,7 +980,9 @@ def introduce(self, to: str, peer: str, caps: Iterable[str] = (), ttl: float = D known = self.peers.get(key_fp(parse_key(peer))) if known is None or known.key_raw is None: raise CommandError(f"unknown peer {peer}") - address = known.meta.get("addr") or known.meta.get("dialed") + # The address we dialed carries the peer's pre-shared key; the one it + # sent in hello does not. + address = known.meta.get("dialed") or known.meta.get("addr") if not address: raise CommandError(f"no known address for {peer}") g = mint_grant(self.identity, to, list(caps), ttl, aud=peer) @@ -1045,6 +1075,13 @@ def __repr__(self) -> str: return f"" +def _tunnel_key(key: str) -> bytes | None: + try: + return x25519_public(parse_key(key)) + except ValueError: + return None + + def _derive_subject(parts: list[dict[str, Any]]) -> str | None: for p in parts: if p.get("k") == "text" and isinstance(p.get("text"), str) and p["text"].strip(): diff --git a/src/awp/transport.py b/src/awp/transport.py index 46e2bf8..f8f46d0 100644 --- a/src/awp/transport.py +++ b/src/awp/transport.py @@ -1,176 +1,175 @@ -"""Addresses and byte streams: tcp:HOST:PORT, unix:/path, and tailcat -addresses through the tailcat CLI.""" +"""The tunnel: `awp tunnel`, the reference implementation's WireGuard +tunnel as a helper process (SPEC.md section 18.2). + +Python has no WireGuard of its own, so a Peer runs one helper for its +whole life. The helper holds the Peer's key, listens on the carriers it is +asked to (tailcat, udp, ws, cloudflare, unix), opens streams to other peers +on request, and forwards the streams they open to a Unix socket of the +Peer's. Everything on those streams is the protocol from section 8 on, +which the Peer speaks itself.""" from __future__ import annotations import asyncio +import json import os import shutil -import socket -import stat +import tempfile +from typing import Any, Awaitable, Callable from ._log import log -from .wire import MAX_LINE, TAILCAT_PORT, StartupError - -def looks_tailcat(s: str) -> bool: - return len(s) >= 40 and s.startswith("tc") and all(c.isalnum() or c in "-_" for c in s) - - -def parse_addr(s: str): - """tcp:HOST:PORT, unix:/path, tailcat:tc..., or a bare tailcat address.""" - s = s.strip() - if s.startswith("tailcat:"): - t = s[len("tailcat:"):].lstrip("/") - if not looks_tailcat(t): - raise StartupError(f"bad tailcat address {s!r}") - return ("tailcat", t) - if looks_tailcat(s): - return ("tailcat", s) - if s.startswith("tcp:"): - host, sep, port = s[4:].lstrip("/").rpartition(":") - if not sep or not port.isdigit(): - raise StartupError(f"bad tcp address {s!r}; expected tcp:HOST:PORT") - return ("tcp", host.strip("[]"), int(port)) - if s.startswith("unix:"): - path = s[5:] - if path.startswith("//"): - path = path[2:] - if not path: - raise StartupError(f"bad unix address {s!r}; expected unix:/path") - return ("unix", path) - if "/" in s: - return ("unix", s) - host, sep, port = s.rpartition(":") - if sep and port.isdigit(): - return ("tcp", host.strip("[]"), int(port)) - raise StartupError(f"unrecognised address {s!r}; use tc..., tailcat:tc..., tcp:HOST:PORT or unix:/path") - - -def fmt_host(host: str) -> str: - return f"[{host}]" if ":" in host else host - - -def is_loopback(host: str) -> bool: - return host in ("localhost", "::1") or host.startswith("127.") - - -async def open_stream(target, tailcat_bin: str = "tailcat"): - """Open a byte stream to a parsed address: (reader, writer, process). - The process is the tailcat client for tailcat addresses, else None.""" - if target[0] == "tcp": - r, w = await asyncio.open_connection(target[1], target[2], limit=2 * MAX_LINE) - return r, w, None - if target[0] == "unix": - r, w = await asyncio.open_unix_connection(target[1], limit=2 * MAX_LINE) - return r, w, None - binary = shutil.which(tailcat_bin) - if binary is None: - raise StartupError(f"{tailcat_bin} is not installed; tailcat addresses need the tailcat CLI " - "(https://github.com/tailscale/tailcat)") - proc = await asyncio.create_subprocess_exec( - binary, target[1], str(TAILCAT_PORT), - stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, - limit=2 * MAX_LINE) - assert proc.stdout is not None and proc.stdin is not None - return proc.stdout, proc.stdin, proc - - -class TailcatListener: - """`tailcat serve 1:127.0.0.1:PORT`: the tunnel's port 1, which awp peers - dial, proxied to a local TCP listener.""" - - def __init__(self, proc, address: str) -> None: - self.proc = proc - self.address = address - - @classmethod - async def start(cls, tailcat_bin: str, port: int, timeout: float = 60.0) -> "TailcatListener": - binary = shutil.which(tailcat_bin) - if binary is None: - raise StartupError(f"{tailcat_bin} is not installed; listening on tailcat needs the tailcat CLI " - "(https://github.com/tailscale/tailcat)") - proc = await asyncio.create_subprocess_exec( - binary, "serve", f"{TAILCAT_PORT}:127.0.0.1:{port}", - stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE) - assert proc.stderr is not None - - async def find_address() -> str: - while True: - line = await proc.stderr.readline() - if not line: - raise StartupError("tailcat exited before printing an address") - text = line.decode("utf-8", "replace").strip() - log(f"tailcat: {text}") - for word in text.split(): - if looks_tailcat(word): - return word - +from .wire import MAX_LINE, StartupError, b64decode_any + +INSTALL_HINT = "install it with: curl -fsSL https://agentwireprotocol.com/install.sh | sh" + + +def find_awp(awp: str | None) -> str: + """The awp binary: the one given, $AWP_BIN, or awp on PATH.""" + cand = awp or os.environ.get("AWP_BIN") or "awp" + path = cand if os.path.sep in cand and os.access(cand, os.X_OK) else shutil.which(cand) + if not path: + raise StartupError(f"{cand} not found: the tunnel runs as `awp tunnel`; {INSTALL_HINT}") + return path + + +class TunnelError(ConnectionError): + """The helper could not do what was asked: dial, listen or rotate.""" + + +class Tunnel: + """One `awp tunnel` process.""" + + def __init__(self, awp: str, identity_file: str, state_dir: str) -> None: + self.awp = awp + self.identity_file = identity_file + self.state_dir = state_dir + self.address: str | None = None # with the pre-shared key: the one to share + self.public: str | None = None # without: for hello.addr + self.proc: asyncio.subprocess.Process | None = None + self._dir: str | None = None + self._server: asyncio.base_events.Server | None = None + self._events: asyncio.Task[None] | None = None + self._stderr: asyncio.Task[None] | None = None + self.on_address: Callable[[], None] | None = None + + async def start(self, accept: Callable[[asyncio.StreamReader, asyncio.StreamWriter], Awaitable[None]]) -> None: + # Socket paths are limited to about 100 bytes: keep them short. + self._dir = tempfile.mkdtemp(prefix="awp-", dir="/tmp" if os.path.isdir("/tmp") else None) + ctl = os.path.join(self._dir, "ctl.sock") + fwd = os.path.join(self._dir, "in.sock") + self.ctl = ctl + self._server = await asyncio.start_unix_server(accept, path=fwd, limit=2 * MAX_LINE) + os.makedirs(self.state_dir, exist_ok=True) + self.proc = await asyncio.create_subprocess_exec( + self.awp, "tunnel", "--identity", self.identity_file, "--state", self.state_dir, + "--socket", ctl, "--forward", fwd, + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + assert self.proc.stdout is not None try: - address = await asyncio.wait_for(find_address(), timeout) - except (asyncio.TimeoutError, StartupError): - proc.kill() - raise - listener = cls(proc, address) - asyncio.get_running_loop().create_task(listener._drain()) - return listener + line = await asyncio.wait_for(self.proc.stdout.readline(), 30) + except asyncio.TimeoutError: + await self.close() + raise StartupError("awp tunnel did not start within 30s") from None + if not line: + err = b"" + if self.proc.stderr is not None: + err = await self.proc.stderr.read() + await self.close() + raise StartupError(f"awp tunnel exited: {err.decode('utf-8', 'replace').strip() or 'no output'}") + self._apply(json.loads(line)) + self._events = asyncio.get_running_loop().create_task(self._read_events()) + self._stderr = asyncio.get_running_loop().create_task(self._drain_stderr()) + + def _apply(self, ev: dict[str, Any]) -> None: + if ev.get("address"): + self.address, self.public = ev["address"], ev.get("public") + if self.on_address is not None: + self.on_address() + + async def _read_events(self) -> None: + assert self.proc is not None and self.proc.stdout is not None + while True: + line = await self.proc.stdout.readline() + if not line: + return + try: + self._apply(json.loads(line)) + except ValueError: + pass - async def _drain(self) -> None: - assert self.proc.stderr is not None + async def _drain_stderr(self) -> None: + assert self.proc is not None and self.proc.stderr is not None while True: line = await self.proc.stderr.readline() if not line: return - log(f"tailcat: {line.decode('utf-8', 'replace').rstrip()}") + log(f"awp tunnel: {line.decode('utf-8', 'replace').rstrip()}") - async def close(self) -> None: + async def _request(self, req: dict[str, Any], timeout: float | None): + r, w = await asyncio.open_unix_connection(self.ctl, limit=2 * MAX_LINE) try: - self.proc.kill() - except ProcessLookupError: - return - try: - await asyncio.wait_for(self.proc.wait(), 5) - except asyncio.TimeoutError: - pass - - -def make_tcp_listener(host: str, port: int) -> socket.socket: - infos = socket.getaddrinfo(host or None, port, type=socket.SOCK_STREAM, - flags=socket.AI_PASSIVE) - if not infos: - raise StartupError(f"cannot resolve {host!r}") - fam, typ, proto, _, sa = infos[0] - s = socket.socket(fam, typ, proto) - s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - try: - s.bind(sa) - except OSError as e: - s.close() - raise StartupError(f"cannot listen on tcp:{fmt_host(host)}:{port}: {e}") from None - s.listen(128) - s.setblocking(False) - return s - + w.write(json.dumps(req).encode() + b"\n") + await w.drain() + line = await asyncio.wait_for(r.readline(), timeout) + except BaseException: + w.close() + raise + if not line: + w.close() + raise TunnelError("awp tunnel closed the request") + rep = json.loads(line) + if not rep.get("ok"): + w.close() + raise TunnelError(rep.get("error") or "awp tunnel refused the request") + return rep, r, w + + async def dial(self, addresses: list[str], key: str | None = None, timeout: float | None = 60.0): + """A stream to the peer the addresses (all of one key) describe: + (reader, writer, reply), reply carrying "key" and "remote".""" + req: dict[str, Any] = {"dial": addresses} + if key: + req["key"] = key + rep, r, w = await self._request(req, timeout) + return r, w, rep + + async def listen(self, carrier: str) -> str: + rep, _, w = await self._request({"listen": carrier}, 120) + w.close() + self._apply(rep) + return rep["address"] + + async def rotate(self) -> str | None: + rep, _, w = await self._request({"rotate": True}, 30) + w.close() + self._apply(rep) + return rep.get("address") -def prepare_unix_path(path: str) -> None: - try: - st = os.lstat(path) - except FileNotFoundError: - d = os.path.dirname(path) - if d: - os.makedirs(d, exist_ok=True) - return - if not stat.S_ISSOCK(st.st_mode): - raise StartupError(f"{path} exists and is not a socket") - probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + async def close(self) -> None: + if self.proc is not None and self.proc.returncode is None: + try: + assert self.proc.stdin is not None + self.proc.stdin.close() # the helper exits when stdin closes + await asyncio.wait_for(self.proc.wait(), 5) + except (asyncio.TimeoutError, OSError, AssertionError): + try: + self.proc.kill() + except ProcessLookupError: + pass + for t in (self._events, self._stderr): + if t is not None: + t.cancel() + if self._server is not None: + self._server.close() + if self._dir: + shutil.rmtree(self._dir, ignore_errors=True) + self._dir = None + + +def remote_from_preamble(line: bytes) -> bytes | None: + """The peer's tunnel key from the helper's first line on a forwarded + stream: {"tunnel": {"remote": "..."}}.""" try: - probe.connect(path) - except OSError: - os.unlink(path) # stale socket from a dead process - return - finally: - probe.close() - raise StartupError(f"{path}: another process is already listening there") - - -# ---------------------------------------------------------------- connection - + obj = json.loads(line) + return b64decode_any(obj["tunnel"]["remote"]) + except (ValueError, KeyError, TypeError): + return None diff --git a/src/awp/wire.py b/src/awp/wire.py index c85b819..c202760 100644 --- a/src/awp/wire.py +++ b/src/awp/wire.py @@ -21,8 +21,7 @@ if TYPE_CHECKING: from .store import Identity -PROTOCOL_VERSION = 0 -AUTH_CONTEXT = b"awp-auth-v0" +PROTOCOL_VERSION = 1 MAX_LINE = 1 << 20 # bytes, excluding the "\n" CHUNK_SIZE = 256 * 1024 # payload bytes per chunk, before base64 DEFAULT_BLOB_LIMIT = 50 * 1024 * 1024 @@ -41,12 +40,12 @@ "application/vnd.awp.fs-write+json": "fs:write", "application/vnd.awp.admin+json": "admin", } -CLOSING_ERR_CODES = frozenset({"bad_frame", "version", "auth", "too_large"}) +CLOSING_ERR_CODES = frozenset({"bad_frame", "too_large", "version", "auth", "refused"}) ACKED_TYPES = frozenset({"msg", "state"}) SEEN_TYPES = frozenset({"msg", "state", "chunk"}) -PRE_AUTH_FORBIDDEN = frozenset({"msg", "state", "ack", "chunk", "resume", "grant", "introduce"}) +PRE_HELLO_FORBIDDEN = frozenset({"msg", "state", "ack", "chunk", "resume", "grant", "introduce"}) READ_SIZE = 256 * 1024 -TAILCAT_PORT = 1 +ADDRESS_PREFIX = "awp1" FSYNC = os.environ.get("AWP_FSYNC", "1") != "0" @@ -338,3 +337,18 @@ def grant_hash(g: dict) -> str: # ---------------------------------------------------------------- persistence + + +def x25519_public(ed_pub: bytes) -> bytes: + """The WireGuard (X25519) key of an Ed25519 public key: the Montgomery + u coordinate (1 + y) / (1 - y) of the same point (SPEC.md section 4.2). + The tunnel knows peers by this key.""" + if len(ed_pub) != 32: + raise ValueError("Ed25519 public key must be 32 bytes") + y = int.from_bytes(ed_pub, "little") & ((1 << 255) - 1) + u = (1 + y) * pow((1 - y) % _P, _P - 2, _P) % _P + return u.to_bytes(32, "little") + + +def is_address(s) -> bool: + return isinstance(s, str) and s.strip().startswith(ADDRESS_PREFIX) diff --git a/tests/test_driver.py b/tests/test_driver.py index 5779744..fd26d8b 100644 --- a/tests/test_driver.py +++ b/tests/test_driver.py @@ -1,12 +1,11 @@ #!/usr/bin/env python3 -"""The stdin/stdout driver (python -m awp), tested the way the original -single-file peer was: as a subprocess, plus raw sockets against it. +"""The stdin/stdout driver (python -m awp), tested as a subprocess, plus a +raw client against it. -Runs real peer processes against each other over TCP and Unix sockets, and uses -raw sockets (with an independent handshake written against the spec) for the -negative cases. - - cd python && python3 -m unittest -v test_awp_peer +Runs real peer processes against each other over the udp and unix carriers, +and uses a raw client (its own `awp tunnel`, and lines written by hand +against the spec) for the negative cases. Needs an awp binary (AWP_BIN, or +awp on PATH). """ import base64 @@ -54,14 +53,25 @@ def ts() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") -def free_port() -> int: - s = socket.socket() +def free_udp_port() -> int: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.bind(("127.0.0.1", 0)) port = s.getsockname()[1] s.close() return port +def awp_bin(): + b = os.environ.get("AWP_BIN") or shutil.which("awp") + if not b: + return None + out = subprocess.run([b, "tunnel", "--help"], capture_output=True, text=True) + return b if "identity" in out.stderr + out.stdout else None + + +AWP = awp_bin() + + def is_recv(t, **match): def pred(e): if e.get("event") != "recv" or e["msg"].get("t") != t: @@ -188,22 +198,32 @@ def stop(self): class RawPeer: - """A minimal hand-rolled awp client used to poke at the peer's edges.""" + """A minimal hand-rolled awp client used to poke at the peer's edges. It + reaches the peer through an `awp tunnel` of its own, holding its own + key, and writes every line itself.""" def __init__(self, addr, sk=None, timeout=10.0): - kind, rest = addr.split(":", 1) - if kind == "tcp": - host, port = rest.rsplit(":", 1) - self.sock = socket.create_connection((host, int(port)), timeout=timeout) - else: - self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - self.sock.settimeout(timeout) - self.sock.connect(rest) - self.buf = b"" self.sk = sk or Ed25519PrivateKey.generate() self.pub = self.sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) self.key = "ed25519:" + b64url(self.pub) + seed = self.sk.private_bytes(serialization.Encoding.Raw, serialization.PrivateFormat.Raw, + serialization.NoEncryption()) + self.dir = tempfile.mkdtemp(prefix="awpr-", dir="/tmp") + ctl = os.path.join(self.dir, "ctl.sock") + self.helper = subprocess.Popen([AWP, "tunnel", "--socket", ctl], stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + env=dict(os.environ, AWP_IDENTITY_SEED=b64url(seed))) + ready = json.loads(self.helper.stdout.readline()) + assert ready.get("event") == "ready", ready + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(30) + self.sock.connect(ctl) + self.sock.sendall(json.dumps({"dial": [addr]}).encode() + b"\n") + self.buf = b"" + reply = json.loads(self.read_line(30)) + assert reply.get("ok"), reply + self.sock.settimeout(timeout) self.n = 0 def next_id(self) -> str: @@ -253,36 +273,24 @@ def read_to_eof(self, timeout=10.0): return out out.append(obj) - def hello_obj(self, v=0): - return {"t": "hello", "id": self.next_id(), "ts": ts(), "v": v, "key": self.key, - "name": "raw-test-client", "nonce": b64url(os.urandom(32)), "caps": ["chat"], - "about": "unit test"} + def hello_obj(self, v=1, key=None, grants=None): + h = {"t": "hello", "id": self.next_id(), "ts": ts(), "v": v, "key": key or self.key, + "name": "raw-test-client", "caps": ["chat"], "about": "unit test"} + if grants: + h["grants"] = grants + return h - def handshake(self, seen=None, tamper=False, grants=None): - """Full hello/auth/resume exchange. Returns (peer_hello, peer_auth, peer_resume).""" + def handshake(self, seen=None, grants=None): + """Hello both ways, then resume both ways. Returns (peer_hello, peer_resume).""" peer_hello_line = self.read_line() assert peer_hello_line is not None, "no hello from peer" peer_hello = json.loads(peer_hello_line) - assert peer_hello["t"] == "hello" - my_line = self.send(self.hello_obj()) - peer_auth = self.read() - assert peer_auth and peer_auth["t"] == "auth", peer_auth - # Independently verify the peer's signature (section 7.2). - Ed25519PublicKey.from_public_bytes(raw_pub(peer_hello["key"])).verify( - b64url_dec(peer_auth["sig"]), - b"awp-auth-v0\x00" + peer_hello_line + b"\x00" + my_line) - sig = bytearray(self.sk.sign(b"awp-auth-v0\x00" + my_line + b"\x00" + peer_hello_line)) - if tamper: - sig[5] ^= 0x01 - auth = {"t": "auth", "id": self.next_id(), "ts": ts(), "sig": b64url(bytes(sig))} - if grants: - auth["grants"] = grants - self.send(auth) - if tamper: - return peer_hello, peer_auth, None + assert peer_hello["t"] == "hello" and peer_hello["v"] == 1, peer_hello + assert "nonce" not in peer_hello + self.send(self.hello_obj(grants=grants)) peer_resume = self.read_until(lambda o: o.get("t") == "resume") self.send({"t": "resume", "id": self.next_id(), "ts": ts(), "seen": seen or {}}) - return peer_hello, peer_auth, peer_resume + return peer_hello, peer_resume def ping_roundtrip(self, timeout=10.0, seen=None): pid = self.next_id() @@ -295,10 +303,17 @@ def close(self): self.sock.close() except OSError: pass + try: + self.helper.stdin.close() # the helper exits when stdin closes + self.helper.wait(5) + except (OSError, subprocess.TimeoutExpired): + self.helper.kill() + shutil.rmtree(self.dir, ignore_errors=True) # ---------------------------------------------------------------- base class +@unittest.skipIf(AWP is None, "needs an awp binary with `awp tunnel` (set AWP_BIN)") class PeerTestBase(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="hp-") @@ -318,7 +333,7 @@ def tearDown(self): def state(self, name) -> str: return os.path.join(self.tmp, f"state-{name}") - def unix_addr(self, name="listen") -> str: + def unix_carrier(self, name="listen") -> str: return "unix:" + os.path.join(self.tmp, f"{name}.sock") def spawn(self, state, mode, addr, name, **kw) -> PeerProc: @@ -336,9 +351,9 @@ def raw(self, addr, **kw) -> RawPeer: self.raws.append(r) return r - def pair(self, addr, **kw): - """alice listens on addr, bob connects; both report connected.""" - a, real = self.listener(addr, "alice", **kw) + def pair(self, carrier, **kw): + """alice listens on the carrier, bob connects; both report connected.""" + a, real = self.listener(carrier, "alice", **kw) b = self.spawn(self.state("bob"), "connect", real, "bob", **kw) ida = a.wait_event("identity") idb = b.wait_event("identity") @@ -353,9 +368,9 @@ def pair(self, addr, **kw): class TwoPeerTests(PeerTestBase): - def _roundtrip(self, addr): - a, b, real = self.pair(addr) - self.assertTrue(a.wait_event("listening")["addr"].startswith(addr.split(":")[0])) + def _roundtrip(self, carrier): + a, b, real = self.pair(carrier) + self.assertTrue(a.wait_event("listening")["addr"].startswith("awp1")) # First connection: both resumes carry an empty seen map. self.assertEqual(a.wait_for(is_recv("resume"))["msg"]["seen"], {}) self.assertEqual(b.wait_for(is_recv("resume"))["msg"]["seen"], {}) @@ -387,15 +402,15 @@ def _roundtrip(self, addr): ids = [e["id"] for e in a.find(lambda e: e.get("event") == "sent")] self.assertEqual(ids, sorted(ids)) - def test_tcp_handshake_and_message_roundtrip(self): - self._roundtrip("tcp:127.0.0.1:0") + def test_udp_handshake_and_message_roundtrip(self): + self._roundtrip("udp:127.0.0.1:0") def test_unix_handshake_and_message_roundtrip(self): - self._roundtrip(self.unix_addr()) + self._roundtrip(self.unix_carrier()) def test_queued_while_disconnected_delivered_in_order(self): - addr = self.unix_addr() - a, b, _ = self.pair(addr) + carrier = self.unix_carrier() + a, b, _ = self.pair(carrier) b.cmd(cmd="send", th="t1", subject="queue test", text="first") first = b.wait_event("sent")["id"] a.wait_for(is_recv("msg", id=first)) @@ -411,7 +426,7 @@ def test_queued_while_disconnected_delivered_in_order(self): queued_ids = [e["id"] for e in sent[1:]] time.sleep(0.3) self.assertEqual(b.find(lambda e: e.get("event") == "acked", start=mark), []) - a2 = self.spawn(self.state("alice"), "listen", addr, "alice") # same state + address + a2 = self.spawn(self.state("alice"), "listen", carrier, "alice") # same state + address a2.wait_event("connected") # alice's resume after restart remembers what it saw before the kill resume = b.wait_for(is_recv("resume"), start=mark)["msg"] @@ -424,22 +439,25 @@ def test_queued_while_disconnected_delivered_in_order(self): b.wait_event("acked", id=q) def test_messages_queued_before_first_connection(self): - addr = self.unix_addr() - b = self.spawn(self.state("bob"), "connect", addr, "bob") # nobody listening yet + carrier = self.unix_carrier() + a0, addr = self.listener(carrier) # learn alice's address, then take her down + a0.stop() + b = self.spawn(self.state("bob"), "connect", addr, "bob") # nobody listening now b.wait_event("identity") for i in range(3): b.cmd(cmd="send", th="early", text=f"early-{i}") ids = [e["id"] for e in b.wait_count(lambda e: e.get("event") == "sent", 3)] time.sleep(0.7) # a few failed attempts - a, _ = self.listener(addr) + a, again = self.listener(carrier) + self.assertEqual(again, addr) # the same address after a restart got = a.wait_count(is_recv("msg"), 3) self.assertEqual([e["msg"]["id"] for e in got], ids) for i in ids: b.wait_event("acked", id=i) def test_kill9_receiver_resumes_without_loss_or_duplicates(self): - addr = f"tcp:127.0.0.1:{free_port()}" - a1, b, _ = self.pair(addr) + carrier = f"udp:127.0.0.1:{free_udp_port()}" + a1, b, _ = self.pair(carrier) n1, n2 = 10, 30 # A paced conversation first. The kill lands between messages: the peer # emits `recv` a few microseconds before it persists the id, so a kill @@ -457,7 +475,7 @@ def test_kill9_receiver_resumes_without_loss_or_duplicates(self): sent = b.wait_count(lambda e: e.get("event") == "sent", n1 + n2) sent_ids = [e["id"] for e in sent] b.wait_event("disconnected") - a2 = self.spawn(self.state("alice"), "listen", addr, "alice") + a2 = self.spawn(self.state("alice"), "listen", carrier, "alice") a2.wait_event("connected") a2.wait_for(lambda e: is_recv("msg")(e) and text_of(e["msg"]) == f"m{n1 + n2 - 1}", what="last message") @@ -474,8 +492,7 @@ def test_kill9_receiver_resumes_without_loss_or_duplicates(self): self.assertEqual([m["id"] for m in received if m["th"] == th], expect) def test_kill9_sender_replays_from_durable_outbox(self): - addr = self.unix_addr() - a, b1, _ = self.pair(addr) + a, b1, addr = self.pair(self.unix_carrier()) n = 30 for i in range(n): b1.cmd(cmd="send", th="t1", text=f"s{i}") @@ -496,7 +513,7 @@ def test_kill9_sender_replays_from_durable_outbox(self): self.assertTrue(acked >= set(ids), f"unacked: {set(ids) - acked}") def test_multichunk_blob_arrives_byte_identical(self): - a, b, _ = self.pair("tcp:127.0.0.1:0") + a, b, _ = self.pair("udp:127.0.0.1:0") data = os.urandom(700 * 1024 + 123) path = os.path.join(self.tmp, "integration.log") with open(path, "wb") as f: @@ -526,8 +543,8 @@ def test_multichunk_blob_arrives_byte_identical(self): self.assertEqual(os.listdir(outgoing), []) def test_kill9_receiver_mid_blob_completes_after_restart(self): - addr = f"tcp:127.0.0.1:{free_port()}" - a1, b, _ = self.pair(addr) + carrier = f"udp:127.0.0.1:{free_udp_port()}" + a1, b, _ = self.pair(carrier) data = os.urandom(11 * 256 * 1024 + 999) # 12 chunks path = os.path.join(self.tmp, "big.tar") with open(path, "wb") as f: @@ -538,7 +555,7 @@ def test_kill9_receiver_mid_blob_completes_after_restart(self): a1.kill9() self.assertEqual(a1.find(lambda e: e.get("event") == "blob"), []) b.wait_event("disconnected") - a2 = self.spawn(self.state("alice"), "listen", addr, "alice") + a2 = self.spawn(self.state("alice"), "listen", carrier, "alice") ev = a2.wait_event("blob", timeout=30) with open(ev["path"], "rb") as f: self.assertEqual(hashlib.sha256(f.read()).hexdigest(), hashlib.sha256(data).hexdigest()) @@ -550,7 +567,7 @@ def test_kill9_receiver_mid_blob_completes_after_restart(self): self.assertEqual(ns, list(range(ns[0], 12))) def test_blob_over_limit_is_refused(self): - a, _ = self.listener("tcp:127.0.0.1:0", extra=["--max-blob", "100000"]) + a, _ = self.listener("udp:127.0.0.1:0", extra=["--max-blob", "100000"]) addr = a.wait_event("listening")["addr"] b = self.spawn(self.state("bob"), "connect", addr, "bob") b.wait_event("connected") @@ -568,7 +585,7 @@ def test_blob_over_limit_is_refused(self): self.assertEqual(os.listdir(os.path.join(self.state("bob"), "outgoing")), []) def test_grant_gates_exec_requests(self): - a, b, _ = self.pair(self.unix_addr()) + a, b, addr = self.pair(self.unix_carrier()) exec_part = {"k": "data", "mime": EXEC_MIME, "data": {"argv": ["make", "test"]}} b.cmd(cmd="send", th="x1", text="please run", parts=[exec_part]) m1 = b.wait_event("sent")["id"] @@ -586,7 +603,7 @@ def test_grant_gates_exec_requests(self): Ed25519PublicKey.from_public_bytes(raw_pub(akey)).verify(b64url_dec(g["sig"]), canon) exp = datetime.strptime(g["exp"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) self.assertTrue(500 < (exp - datetime.now(timezone.utc)).total_seconds() <= 601) - # bob keeps the grant (it will present it in future auth messages) + # bob keeps the grant (it will present it in future hellos) with open(os.path.join(self.state("bob"), "grants_held.json")) as f: self.assertEqual(json.load(f)[0]["sig"], g["sig"]) # the same request is now allowed: acked with no forbidden err before the ack @@ -595,17 +612,17 @@ def test_grant_gates_exec_requests(self): m2 = b.wait_event("sent", start=mark, t="msg")["id"] b.wait_event("acked", id=m2) self.assertEqual(b.find(is_recv("err", re=m2)), []) - # after a restart bob presents the held grant in its auth message + # after a restart bob presents the held grant in its hello b.kill9() a.wait_event("disconnected") mark = a.mark() - b2 = self.spawn(self.state("bob"), "connect", self.unix_addr(), "bob") + b2 = self.spawn(self.state("bob"), "connect", addr, "bob") b2.wait_event("connected") - auth = a.wait_for(is_recv("auth"), start=mark)["msg"] - self.assertEqual([x["sig"] for x in auth.get("grants", [])], [g["sig"]]) + hello = a.wait_for(is_recv("hello"), start=mark)["msg"] + self.assertEqual([x["sig"] for x in hello.get("grants", [])], [g["sig"]]) def test_bye_closes_both_sides_and_suppresses_reconnect(self): - a, b, _ = self.pair(self.unix_addr()) + a, b, _ = self.pair(self.unix_carrier()) b.cmd(cmd="send", th="t1", text="one thing") # t1 is now an open thread mid = b.wait_event("sent")["id"] b.wait_event("acked", id=mid) @@ -627,28 +644,30 @@ def test_bye_closes_both_sides_and_suppresses_reconnect(self): class RawSocketTests(PeerTestBase): - def test_tampered_auth_signature_rejected(self): - a, addr = self.listener(self.unix_addr()) + def test_hello_naming_another_key_is_refused(self): + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) - r.handshake(tamper=True) + self.assertEqual(json.loads(r.read_line())["t"], "hello") + other = Ed25519PrivateKey.generate().public_key().public_bytes(serialization.Encoding.Raw, + serialization.PublicFormat.Raw) + r.send(r.hello_obj(key="ed25519:" + b64url(other))) rest = r.read_to_eof() - errs = [o for o in rest if o.get("t") == "err"] - self.assertEqual([e["code"] for e in errs], ["auth"]) - self.assertEqual([o for o in rest if o.get("t") == "resume"], []) - a.wait_event("error") + self.assertEqual([(o["t"], o.get("code")) for o in rest], [("err", "auth")]) + time.sleep(0.3) self.assertEqual(a.find(lambda e: e.get("event") == "connected"), []) + self.assertEqual(a.find(is_recv("hello")), []) # the lie never reached the peer - def test_version_1_rejected(self): - a, addr = self.listener("tcp:127.0.0.1:0") + def test_version_2_rejected(self): + a, addr = self.listener("udp:127.0.0.1:0") r = self.raw(addr) self.assertEqual(json.loads(r.read_line())["t"], "hello") - r.send(r.hello_obj(v=1)) + r.send(r.hello_obj(v=2)) rest = r.read_to_eof() - self.assertEqual([o["t"] for o in rest], ["err"]) # no auth after a bad version + self.assertEqual([o["t"] for o in rest], ["err"]) # no resume after a bad version self.assertEqual(rest[0]["code"], "version") def test_garbage_line_gets_bad_frame(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) for garbage in (b"this is not json\n", b"[1,2,3]\n"): r = self.raw(addr) r.read_line() @@ -663,7 +682,7 @@ def test_garbage_line_gets_bad_frame(self): self.assertEqual([(o["t"], o.get("code")) for o in rest if o["t"] == "err"], [("err", "bad_frame")]) def test_line_over_1mib_gets_too_large(self): - a, addr = self.listener("tcp:127.0.0.1:0") + a, addr = self.listener("udp:127.0.0.1:0") r = self.raw(addr) r.read_line() r.send_raw(b"x" * ((1 << 20) + 100)) @@ -671,7 +690,7 @@ def test_line_over_1mib_gets_too_large(self): self.assertEqual([(o["t"], o["code"]) for o in rest], [("err", "too_large")]) def test_unknown_type_and_unknown_fields_ignored(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) r.handshake() a.wait_event("connected") @@ -697,7 +716,7 @@ def test_unknown_type_and_unknown_fields_ignored(self): self.assertEqual(a.find(lambda e: e.get("event") in ("error", "disconnected")), []) def test_duplicate_id_is_delivered_once_and_reacked(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) r.handshake() mid = r.next_id() @@ -712,7 +731,7 @@ def test_duplicate_id_is_delivered_once_and_reacked(self): self.assertEqual(len(a.find(is_recv("msg", id=mid))), 1) def test_resume_replays_unacked_in_original_order_with_original_ids(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) for i in range(3): # queued before anyone connects a.cmd(cmd="send", th="t1", text=f"r{i}", subject="replay" if i == 0 else None) ids = [e["id"] for e in a.wait_count(lambda e: e.get("event") == "sent", 3)] @@ -729,7 +748,7 @@ def test_resume_replays_unacked_in_original_order_with_original_ids(self): a.wait_event("disconnected") # same key again, claiming to have durably seen up to ids[1] r2 = self.raw(addr, sk=r.sk) - _, _, resume = r2.handshake(seen={"t1": ids[1]}) + _, resume = r2.handshake(seen={"t1": ids[1]}) self.assertEqual(resume["seen"], {"t7": mine}) # alice's own seen map a.wait_event("acked", id=ids[1]) # covered by our seen: implicit ack seen = [] @@ -741,7 +760,7 @@ def test_resume_replays_unacked_in_original_order_with_original_ids(self): self.assertEqual([o["id"] for o in seen if o.get("t") == "msg"], [ids[2]]) def test_silent_peer_is_dropped_after_two_missed_pongs(self): - a, addr = self.listener(self.unix_addr(), ping=0.3) + a, addr = self.listener(self.unix_carrier(), ping=0.3) r = self.raw(addr) r.handshake() t0 = time.time() @@ -754,17 +773,14 @@ def test_silent_peer_is_dropped_after_two_missed_pongs(self): self.assertIn("missed pongs", ev["reason"]) def test_go_style_encodings_and_omitted_fields_accepted(self): - """Padded std-alphabet key/sig, seen:null (Go nil map), msg-before-chunks, + """Padded std-alphabet key, seen:null (Go nil map), msg-before-chunks, chunks without th and with omitempty-dropped n/last, unpadded/url data.""" - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) - peer_hello_line = r.read_line() + r.read_line() # their hello hello = r.hello_obj() hello["key"] = "ed25519:" + base64.b64encode(r.pub).decode() # padded, std - my_line = r.send(hello) - r.read_until(lambda o: o.get("t") == "auth") - sig = r.sk.sign(b"awp-auth-v0\x00" + my_line + b"\x00" + peer_hello_line) - r.send({"t": "auth", "id": r.next_id(), "ts": ts(), "sig": base64.b64encode(sig).decode()}) + r.send(hello) r.read_until(lambda o: o.get("t") == "resume") r.send({"t": "resume", "id": r.next_id(), "ts": ts(), "seen": None}) self.assertEqual(a.wait_event("connected")["key"], hello["key"]) # echoed as sent @@ -785,7 +801,7 @@ def test_go_style_encodings_and_omitted_fields_accepted(self): self.assertEqual(ev["ref"], "L1") def test_bye_is_answered_with_bye(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) r.handshake() r.send({"t": "bye", "id": r.next_id(), "ts": ts(), "reason": "done"}) @@ -794,7 +810,7 @@ def test_bye_is_answered_with_bye(self): a.wait_event("disconnected", reason="bye") def test_err_close_semantics(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) r.handshake() for code in ("unsupported", "forbidden", "internal", "blob_refused", "some_future_code"): @@ -805,7 +821,7 @@ def test_err_close_semantics(self): a.wait_event("disconnected") def test_one_level_grant_delegation_via_introduce(self): - a, addr = self.listener(self.unix_addr()) + a, addr = self.listener(self.unix_carrier()) akey = a.wait_event("identity")["key"] def mint(sk, sub, caps): @@ -836,7 +852,7 @@ def exec_forbidden(r): # X introduces itself with a grant for alice (stored as held by alice) g_xa = mint(x.sk, akey, ["chat-extra"]) x.send({"t": "introduce", "id": x.next_id(), "ts": ts(), "th": "i1", - "peer": {"key": x.key, "name": "x", "address": "unix:/nowhere"}, "grant": g_xa}) + "peer": {"key": x.key, "name": "x"}, "grant": g_xa}) x.ping_roundtrip() with open(os.path.join(self.state("alice"), "grants_held.json")) as f: self.assertIn(g_xa["sig"], [g["sig"] for g in json.load(f)]) @@ -854,15 +870,14 @@ def exec_forbidden(r): z.handshake(grants=[mint(wsk, z.key, ["exec"])]) self.assertTrue(exec_forbidden(z)) - def test_messages_before_auth_are_rejected(self): - a, addr = self.listener(self.unix_addr()) + def test_messages_before_hello_are_rejected(self): + a, addr = self.listener(self.unix_carrier()) r = self.raw(addr) r.read_line() - r.send(r.hello_obj()) r.send({"t": "msg", "id": r.next_id(), "ts": ts(), "th": "t1", "parts": [{"k": "text", "text": "sneaky"}]}) rest = r.read_to_eof() - self.assertEqual([o.get("code") for o in rest if o["t"] == "err"], ["auth"]) + self.assertEqual([o.get("code") for o in rest if o["t"] == "err"], ["bad_frame"]) self.assertEqual(a.find(is_recv("msg")), []) @@ -886,27 +901,34 @@ def test_pure_python_ed25519_matches_cryptography(self): self.assertFalse(hp.pure_ed25519_verify(pub, bytes(bad), msg)) self.assertFalse(hp.pure_ed25519_verify(pub, sig, msg + b"x")) + @unittest.skipIf(AWP is None, "needs an awp binary with `awp tunnel` (set AWP_BIN)") def test_pure_python_peer_interoperates(self): - """A peer forced onto the fallback crypto still handshakes with a raw client.""" - tmp = tempfile.mkdtemp(prefix="hp-") + """A peer forced onto the fallback crypto mints grants that verify with + the cryptography package, over a real connection.""" + tmp = tempfile.mkdtemp(prefix="hp-", dir="/tmp") try: env = dict(os.environ, AWP_PURE_ED25519="1") - sock = os.path.join(tmp, "p.sock") - if len(sock) > 100: - shutil.rmtree(tmp) - tmp = tempfile.mkdtemp(prefix="hp-", dir="/tmp") - sock = os.path.join(tmp, "p.sock") p = subprocess.Popen([*PEER_CMD, "--state", os.path.join(tmp, "s"), - "listen", "unix:" + sock], + "listen", "unix:" + os.path.join(tmp, "p.sock")], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=env) try: + addr = key = None for line in p.stdout: - if json.loads(line).get("event") == "listening": + ev = json.loads(line) + if ev.get("event") == "identity": + key = ev["key"] + if ev.get("event") == "listening": + addr = ev["addr"] break - r = RawPeer("unix:" + sock) - r.handshake() # verifies the peer's signature too - r.ping_roundtrip() + r = RawPeer(addr) + r.handshake() + p.stdin.write((json.dumps({"cmd": "grant", "sub": r.key, "caps": ["exec"], "ttl": 60}) + "\n").encode()) + p.stdin.flush() + g = r.read_until(lambda o: o.get("t") == "grant")["grant"] + body = {k: v for k, v in g.items() if k != "sig"} + canon = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + Ed25519PublicKey.from_public_bytes(raw_pub(key)).verify(b64url_dec(g["sig"]), canon) r.close() finally: p.kill() diff --git a/tests/test_peer.py b/tests/test_peer.py index 6d9c87a..cbddcd8 100644 --- a/tests/test_peer.py +++ b/tests/test_peer.py @@ -16,9 +16,6 @@ import awp -pytestmark = pytest.mark.asyncio - - async def next_of(peer: awp.Peer, kind: type, timeout: float = 15.0): """The next event of a kind, skipping others.""" loop = asyncio.get_running_loop() @@ -43,7 +40,7 @@ async def test_conversation(tmp): async with awp.Peer(os.path.join(tmp, "a"), name="a@test", ping_interval=2) as a, \ awp.Peer(os.path.join(tmp, "b"), name="b@test", ping_interval=2) as b: addr = await b.listen("unix:" + os.path.join(tmp, "b.sock")) - assert b.addresses == [addr] + assert b.address == addr and addr.startswith("awp1") key = await a.connect(addr, timeout=10) assert key == b.key c = await next_of(b, awp.Connected) @@ -109,8 +106,8 @@ async def test_queued_while_away_and_restart(tmp): b_sock = "unix:" + os.path.join(tmp, "b.sock") async with awp.Peer(os.path.join(tmp, "a"), name="a@test", ping_interval=1) as a: b = await awp.Peer(b_dir, name="b@test").start() - await b.listen(b_sock) - key = await a.connect(b_sock, timeout=10) + b_addr = await b.listen(b_sock) + key = await a.connect(b_addr, timeout=10) await next_of(a, awp.Connected) await b.close() await next_of(a, awp.Disconnected) @@ -135,7 +132,7 @@ async def test_queued_while_away_and_restart(tmp): b = await awp.Peer(b_dir, name="b@test").start() await b.listen(b_sock) try: - await a.connect(b_sock, timeout=10) + await a.connect(b_addr, timeout=10) m = await next_of(b, awp.Message, 30) assert m.id == sent.id and m.text == "after the restart" finally: @@ -155,14 +152,19 @@ def awp_bin() -> str | None: b = os.environ.get("AWP_BIN") or shutil.which("awp") if not b: return None - out = subprocess.run([b, "conform", "--list"], capture_output=True, text=True) - return b if out.returncode == 0 else None + out = subprocess.run([b, "tunnel", "--help"], capture_output=True, text=True) + return b if "identity" in out.stderr + out.stdout else None + + +# Every test needs `awp tunnel`: it is the transport. +pytestmark = [pytest.mark.asyncio, + pytest.mark.skipif(awp_bin() is None, reason="needs an awp binary with `awp tunnel` (set AWP_BIN)")] + -@pytest.mark.skipif(awp_bin() is None, reason="needs an awp binary with `conform` (set AWP_BIN)") async def test_conformance_peer_listening(tmp): async with awp.Peer(os.path.join(tmp, "p"), name="py@test") as p: - addr = await p.listen("tcp:127.0.0.1:0") + addr = await p.listen("udp:127.0.0.1:0") proc = await asyncio.create_subprocess_exec(awp_bin(), "conform", "--json", "--timeout", "10s", addr, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) out, err = await asyncio.wait_for(proc.communicate(), 120) @@ -172,12 +174,11 @@ async def test_conformance_peer_listening(tmp): assert rep["passed"] == len(rep["results"]) -@pytest.mark.skipif(awp_bin() is None, reason="needs an awp binary with `conform` (set AWP_BIN)") async def test_conformance_peer_dialing(tmp): cmd = f"{sys.executable} -m awp --state {os.path.join(tmp, 'p')} --name py@test connect {{addr}}" - env = dict(os.environ, PYTHONPATH=os.path.join(os.path.dirname(__file__), "..", "src")) + env = dict(os.environ, PYTHONPATH=os.path.join(os.path.dirname(__file__), "..", "src"), AWP_BIN=awp_bin()) proc = await asyncio.create_subprocess_exec(awp_bin(), "conform", "--json", "--timeout", "10s", - "--listen", "tcp:127.0.0.1:0", "--run", cmd, + "--listen", "udp:127.0.0.1:0", "--run", cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env) out, err = await asyncio.wait_for(proc.communicate(), 120) rep = json.loads(out) @@ -216,3 +217,36 @@ async def test_revoke_and_introduce(tmp): # b meets c through the address it was handed. assert await b.connect(intro.address, timeout=10) == c.key await next_of(c, awp.Connected) + + +async def test_rotation_and_presented_grants(tmp): + """c trusts a with introduce; a introduces b to c with a grant bound to + c, which b presents in its hello and c honors. Rotating c's pre-shared + key keeps peers it met, and shuts out strangers with the old address.""" + async with awp.Peer(os.path.join(tmp, "a"), name="a@test") as a, \ + awp.Peer(os.path.join(tmp, "b"), name="b@test") as b, \ + awp.Peer(os.path.join(tmp, "c"), name="c@test", trust=[]) as c: + c_addr = await c.listen("udp:127.0.0.1:0") + c.trusted.add(awp.wire.parse_key(a.key)) + b_addr = await b.listen("unix:" + os.path.join(tmp, "b.sock")) + await a.connect(b_addr, timeout=15) + await a.connect(c_addr, timeout=15) + await next_of(b, awp.Connected) + a.introduce(b.key, c.key, ["fs:read"], 3600) + intro = await next_of(b, awp.Introduced) + assert intro.address == c_addr and intro.grant["aud"] == c.key + assert await b.connect(intro.address, timeout=15) == c.key + conn = await next_of(c, awp.Connected) + while conn.peer != b.key: + conn = await next_of(c, awp.Connected) + assert c.caps(b.key) == {"fs:read"} # presented in b's hello, honored by c + + new_addr = await c.rotate_psk() + assert new_addr and new_addr != c_addr + async with awp.Peer(name="stranger@test") as d: + with pytest.raises(ConnectionError): + await d.connect(c_addr, timeout=6) + assert await d.connect(new_addr, timeout=15) == c.key + await b.bye(c.key) + await next_of(b, awp.Disconnected) + assert await b.connect(intro.address, timeout=15) == c.key # met before: still admitted