Add explicit `verify_wg_peer()` inspection

Validate a declared tunnel key against one `pyroute2` snapshot
containing the iface's own key and configured peers.

Deats,
- share worker offload across all WireGuard key readers
- forward `WGTunnelSpec.iface` and `.netns` to the read
- reject malformed declarations before netlink I/O
- export the async helper and cover local, peer and absent keys
- replace multihost's `wg show` subprocess probe

Prompt-IO: ai/prompt-io/opencode/20260822T023226Z_59a8ecfd_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

ai/prompt-io/opencode/20260822T023226Z_59a8ecfd_prompt_io.md

@@ -0,0 +1,46 @@
+---
+model: gpt-5.6-sol
+service: opencode
+session: tractor-addr-unpacking
+timestamp: 2026-08-22T02:32:26Z
+git_ref: 59a8ecfd
+scope: code
+substantive: true
+raw_file: 20260822T023226Z_59a8ecfd_prompt_io.raw.md
+---
+
+## Prompt
+
+After committing the read-only pyroute2 helpers and `wgman` design
+update, the human authorized the next isolated Layer B change:
+explicit `verify_wg_peer()` composition over WireGuard inspection.
+
+## Response summary
+
+Added and exported async `verify_wg_peer()` using one validated
+WireGuard key snapshot. It recognizes local-interface and configured
+peer identities without coupling kernel inspection to address
+parsing. Updated the multihost examples to use the production helper
+and removed their subprocess-based probe.
+
+## Files changed
+
+- `tractor/discovery/_tunnel.py` - shared async snapshot reader and
+  explicit verification helper.
+- `tractor/discovery/__init__.py` - public verification export.
+- `tests/discovery/test_wg_inspection.py` - local, peer, absent and
+  malformed-key verification coverage.
+- `examples/multihost/wg_lan/host_a_srv.py` - async local-key check.
+- `examples/multihost/wg_lan/host_b_client.py` - async peer-key check.
+- `examples/multihost/wg_lan/wg_maddr.py` - removed obsolete
+  subprocess probe.
+- `examples/multihost/wg_lan/README.md` - pyroute2 requirements and
+  verification workflow.
+- `ai/tpt-backends/03_wg_tunnel_bindspace.md` - async API contract.
+
+## Human edits
+
+The human selected this pre-agreed verification layer as the next
+atomic change after reviewing and committing the preceding read and
+architecture changes. The agent implemented the source changes; no
+direct manual edits or follow-up corrections were observed.

ai/prompt-io/opencode/20260822T023226Z_59a8ecfd_prompt_io.raw.md

@@ -0,0 +1,31 @@
+---
+model: gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-22T02:32:26Z
+git_ref: 59a8ecfd
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+# Raw output - verify declared WireGuard identities
+
+The human authorized the next incremental Layer B change after
+committing the read-only pyroute2 helpers and first-child `wgman`
+design update.
+
+> `git diff HEAD~1..HEAD -- tractor/discovery/_tunnel.py tractor/discovery/__init__.py tests/discovery/test_wg_inspection.py examples/multihost/wg_lan ai/tpt-backends/03_wg_tunnel_bindspace.md`
+
+Added async `verify_wg_peer()` over one WireGuard key snapshot. It
+validates the declared `WGTunnelSpec.peer_pubkey` before I/O, forwards
+the spec's iface/netns, and accepts either the local interface key for
+a source/listen declaration or a configured peer key for a
+destination/dial declaration.
+
+Refactored worker offload behind one shared async reader so
+verification cannot compare two different netlink snapshots. Exported
+the helper, added local/peer/absent/malformed-key coverage, and moved
+the multihost examples from their local `wg show` subprocess probe to
+the production API.
+
+Ruff and lock checks passed. Focused WireGuard/tunnel/multiaddr
+coverage passed 51 tests; the complete discovery suite passed 92
+tests with 2 xpasses.

ai/tpt-backends/03_wg_tunnel_bindspace.md

@@ -32,8 +32,10 @@ onto `trio` as the library's sans-io layer allows.
   by multiformats/py-multiaddr#107 and gh #483.
 - **today's deployable story remains declarative**: run `wg-quick`
   out-of-band, parse the maddr, strip its wrapper to the overlay
-  `(host, port)`, verify the pubkey in its host-specific role,
+  `(host, port)`, explicitly verify the declared pubkey against the
-  hand the overlay addr to `registry_addrs=`/`tpt_bind_addrs=`.
+  local interface key or configured peers with async
+  `verify_wg_peer()`, then hand the overlay addr to
+  `registry_addrs=`/`tpt_bind_addrs=`.
   The repaired `examples/multihost/wg_lan/` implementation derives
   from and supersedes #482's original example.
 - `Address.namespace` exists in the Protocol
@@ -200,31 +202,25 @@ Observed protocol-name lists, for writing the `match`:
 
 ### 3.3 pure parser helpers + explicit verification
 
-The parser/key-codec helpers live in
+The parser/key-codec helpers and async production verifier live in
-`tractor/discovery/_tunnel.py`; the impure verifier remains
+`tractor/discovery/_tunnel.py`; parsing remains pure while verification
-example-local until layer B:
+is an explicit, impure caller step:
 
 ```python
 def parse_wg_maddr(maddr: str|Multiaddr) -> TunnelledAddress: ...
 def mb_pubkey(wg8_key: str) -> str: ...
 def wg8_pubkey(multibase_key: str) -> str: ...
-async def verify_wg_key(
+async def verify_wg_peer(spec: WGTunnelSpec) -> bool: ... # layer B
-    addr: TunnelledAddress,
-    role: Literal['local', 'peer'],
-    iface: str|None = None,
-    timeout: float = 5,
-    inspection: str|None = None,
-) -> bool: ...  # example-local impure probe
 ```
 
-In layer A `verify_wg_key()` may shell out to role-specific
+Layer A's example-local `verify_wg_key()` used role-specific
-`wg show <if> public-key|peers` queries, but it must be a *single*
+`wg show <if> public-key|peers` queries. Layer B replaces it with
-async, time-bounded function so it never blocks trio's run thread
+`verify_wg_peer()`, backed by one pyroute2 key snapshot selected by
-and layer B swaps only its body. It verifies key presence only,
+`spec.iface` and `spec.netns`. It validates the declared key before
-not `Endpoint`, `AllowedIPs`, handshake state, or routing. Never
+I/O and accepts either the interface's own public key or a configured
-run `tractor` as root: privileged inspection stays a separate
+peer key. It does not enforce a host-specific role and verifies key
-step whose public-key output can be passed as `inspection`. Never
+presence only, not `Endpoint`, `AllowedIPs`, handshake state, or
-call it implicitly from
+routing. Never call it implicitly from
 `wrap_address()`/`parse_maddr()` — parsing must stay pure and
 side-effect-free; verification is the *caller's* explicit step
 (and later, the bindspace `@acm`'s).
@@ -285,13 +281,13 @@ Three integration options, in increasing trio-nativeness:
 - (3) reimplement the codecs. Never.
 
 **Recommended split**: ship (1) first so layer B is a small,
-reviewable, behaviour-preserving swap of `verify_wg_key()`'s
+reviewable replacement of the example-local verification probe with
-body; then land (2) as a follow-up commit for the read path
+production `verify_wg_peer()`; then land (2) as a follow-up commit for
-(`wg get`, `link get`) where the sans-io surface is smallest,
+the read path (`wg get`, `link get`) where the sans-io surface is
-and keep (1) for the privileged mutating ops. Measure before
+smallest, and keep (1) for the privileged mutating ops. Measure before
-converting anything else — there is no perf argument here, only
+converting anything else — there is no perf argument here, only a "no
-a "no foreign event loop in a trio actor" argument, which (1)
+foreign event loop in a trio actor" argument, which (1) already
-already satisfies (a thread is not an event loop).
+satisfies (a thread is not an event loop).
 
 Explicitly **do not** pull in `trio-asyncio` for pyroute2 or infect
 every wg-using actor merely to service one-shot netlink calls. A
@@ -313,7 +309,8 @@ async def read_wg_peers(
 async def read_wg_pubkey(iface: str = 'wg0', ...) -> str: ...
 ```
 
-and `verify_wg_key()` becomes a thin composition over the two.
+and `verify_wg_peer()` becomes a thin composition over one shared key
+snapshot.
 Note the pure-getter rule: no `read_wg_peers(..., create=True)`.
 
 ---

examples/multihost/wg_lan/README.md

@@ -45,12 +45,13 @@ no `wg` codec. So `pyproject.toml` temporarily pins the merge commit
 in its PEP 621 dependency metadata, and a plain
 
 ```bash
-uv sync
+uv sync --extra wg
 ```
 
-gets you a `wg`-aware `multiaddr`. That pin goes away once a
+gets you a `wg`-aware `multiaddr` plus pyroute2's Linux netlink API.
-release carries the codec. `py-multibase` is a direct project
+The multiaddr pin goes away once a release carries the codec.
-dependency, so no separate install command is needed.
+`py-multibase` is a direct project dependency, so no separate install
+command is needed.
 
 Without the codec `parse_wg_maddr()` raises immediately with an
 actionable message — there is deliberately **no** degraded
@@ -130,27 +131,21 @@ the same string — A's bearer, A's key, A's overlay ep).
 
 ## 2. verify the keys
 
-Interface inspection commonly needs `CAP_NET_ADMIN`. Keep that
+Both scripts explicitly call `await verify_wg_peer(addr.tunnel)`
-privileged operation separate from the `tractor` processes:
+before starting `tractor`. The helper validates the maddr's declared
+key, reads one `wg0` key snapshot through pyroute2's Linux
+generic-netlink API, and accepts the key when it is either the
+interface's own public key or one of its configured peers.
 
-```bash
+This establishes key presence only. It does not enforce a
-# host A: output must equal the maddr's A_pub key
+host-specific local/peer role and does not verify `Endpoint`,
-export WG_KEY_INSPECTION="$(sudo wg show wg0 public-key)"
+`AllowedIPs`, a recent handshake, or routing.
 
-# host B: output must contain the maddr's A_pub key
+Interface inspection commonly requires `CAP_NET_ADMIN` in the user
-export WG_KEY_INSPECTION="$(sudo wg show wg0 peers)"
+namespace that owns the target network namespace. Run each program in
-```
+a security context that already has the required inspection authority.
-
+The helper never invokes `sudo` or `wg(8)`, escalates privileges, or
-These checks establish only that host A uses the declared local
+creates a namespace.
-key and host B has that key as a configured peer. They do not
-verify `Endpoint`, `AllowedIPs`, a recent handshake, or routing.
-The exported text contains public keys only. Each script passes it
-to `verify_wg_key()` with its host-specific role before starting
-`tractor`. Callers that already have permission to inspect the
-interface may omit that argument; the helper's direct query is
-async and requests cancellation after five seconds. Trio's
-subprocess termination escalation can make final process cleanup
-take longer than that cancellation deadline.
 
 ## 3. run
 
@@ -162,11 +157,10 @@ python host_a_srv.py
 python host_b_client.py
 ```
 
-Run both `tractor` programs as the normal application account,
+Run both `tractor` programs as the normal application account in a
-not as root. Privilege is needed only for tunnel setup and the
+security context with the inspection authority described above. No
-separate inspection above. If using that preflight, keep the
+`WG_KEY_INSPECTION` export or subprocess preflight is used; tunnel
-host-specific `WG_KEY_INSPECTION` value exported in each
+setup remains out-of-band. Do not run the applications as root.
-program's shell.
 
 The client binds its own actor listener to `10.0.11.2:0`, while
 the service actor binds to host A's `10.0.11.1` overlay host with
@@ -189,12 +183,13 @@ Four corrections, all from
    all. `parse_wg_maddr()` now rejects it with an actionable
    error.
 2. **parsing is pure.** #482's helper had the key-check adjacent
-   to the parse; `verify_wg_key()` is now a separate, explicitly
+   to the parse; async `verify_wg_peer()` is now a separate,
-   composed step for inspection-capable callers. A parser that
+   explicitly composed step that the caller invokes. Implicit
-   shells out is a nasty surprise.
+   kernel inspection from a parser is a nasty surprise.
-3. **no `sudo`.** #482 ran `sudo wg show`; a library/example must
+3. **no `sudo` or subprocess.** #482 ran `sudo wg show`; tractor's
-   never escalate or run `tractor` as root. Privileged tunnel
+   helper reads generic netlink through pyroute2 and never attempts
-   setup and key inspection are separate shell steps.
+   privilege escalation or namespace creation. The caller must
+   already have the required inspection authority.
 4. **no new `Address` proto-type.** The tunnel rides *beside* the
    overlay addr in a frozen `TunnelledAddress`, and only `.overlay`
    crosses into `open_nursery()`. #482 §6 floated a `WGAddress`
@@ -205,7 +200,7 @@ Four corrections, all from
 
 ## next
 
-Layer A's `TunnelledAddress` and native maddr parser now live in
+Layer A's `TunnelledAddress` and native maddr parser plus Layer B's
-`tractor.discovery`. Next, replace this example's `wg(8)` verification
+explicit pyroute2 verification now live in `tractor.discovery`. Next,
-probe with `pyroute2`, then add `open_bindspace()` `@acm`s which
+add `open_bindspace()` `@acm`s which create/tear down the iface and
-create/tear down the iface and netns.
+netns.

examples/multihost/wg_lan/host_a_srv.py

@@ -8,18 +8,15 @@ tunnel's *overlay* addr, declared as a single `wg` maddr.
 '''
 from __future__ import annotations
 
-import os
-
 import tractor
 import trio
 from tractor.discovery import (
     TunnelledAddress,
     mk_maddr,
     parse_wg_maddr,
+    verify_wg_peer,
 )
 
-from wg_maddr import verify_wg_key
-
 # bearer = host A's underlay `(ip, wg ListenPort)`
 # key    = host A's OWN tunnel pubkey
 # overlay = the ep `tractor` binds, on the wg iface's addr
@@ -37,12 +34,7 @@ async def echo(msg: str) -> str:
 
 async def main():
     addr: TunnelledAddress = parse_wg_maddr(WG_MADDR)
-    inspection: str | None = os.environ.get('WG_KEY_INSPECTION')
+    if not await verify_wg_peer(addr.tunnel):
-    if not await verify_wg_key(
-        addr,
-        role='local',
-        inspection=inspection,
-    ):
         raise RuntimeError(
             f'Maddr key is not wg0 local public key!\n'
             f'maddr: {WG_MADDR}\n'

examples/multihost/wg_lan/host_b_client.py

@@ -6,17 +6,15 @@ Host B: workstation dialing host A's actor tree through the
 '''
 from __future__ import annotations
 
-import os
-
 import tractor
 import trio
 from tractor.discovery import (
     TunnelledAddress,
     parse_wg_maddr,
+    verify_wg_peer,
 )
 
 from host_a_srv import echo  # noqa: F401  (RPC refs it by mod path)
-from wg_maddr import verify_wg_key
 
 # same maddr as host A: A's bearer, A's key, A's overlay ep
 WG_MADDR: str = (
@@ -29,12 +27,7 @@ LOCAL_OVERLAY_BIND: tuple[str, int] = ('10.0.11.2', 0)
 
 async def main():
     addr: TunnelledAddress = parse_wg_maddr(WG_MADDR)
-    inspection: str | None = os.environ.get('WG_KEY_INSPECTION')
+    if not await verify_wg_peer(addr.tunnel):
-    if not await verify_wg_key(
-        addr,
-        role='peer',
-        inspection=inspection,
-    ):
         raise RuntimeError(
             f'Maddr key is not a configured wg0 peer!\n'
             f'maddr: {WG_MADDR}\n'

examples/multihost/wg_lan/wg_maddr.py

@@ -1,85 +0,0 @@
-# tractor: distributed structured concurrency.
-r'''
-Verify `wg` keys declared by tractor's multiaddr parser.
-
-`tractor.discovery.parse_wg_maddr()` owns pure parsing and delegates
-all tunnel peeling to `py-multiaddr`. This example keeps only the
-explicit impure probe used by the two-host demo; parsing never shells
-out or verifies local interface state implicitly.
-
-The canonical maddr form is:
-
-    /ip4/10.0.0.1/udp/51820/wg/u<key>/ip4/10.0.11.1/tcp/1616
-    \_______ wg bearer ______/\_ key _/\____ tractor ep _____/
-
-The kernel owns the bearer socket. A future tractor bindspace may
-provision it through netlink, but only the overlay is an application
-`MsgTransport` endpoint.
-
-'''
-from __future__ import annotations
-from typing import Literal
-
-import trio
-
-from tractor.discovery import (
-    TunnelledAddress,
-    WGTunnelSpec,
-)
-
-
-async def verify_wg_key(
-    addr: TunnelledAddress,
-    role: Literal['local', 'peer'],
-    iface: str | None = None,
-    timeout: float = 5,
-    inspection: str | None = None,
-) -> bool:
-    '''
-    Verify the declared key in the role required on this host.
-
-    A bearer host uses `role='local'`; a dialer uses `role='peer'`.
-    This verifies only key presence. It does not inspect the peer's
-    endpoint, AllowedIPs, handshake state, or iface addresses.
-
-    `inspection` accepts output captured by a separate privileged
-    `wg show` step. Without it, query asynchronously for callers
-    which already have interface-inspection permission.
-
-    IMPURE + explicit by design: neither `parse_wg_maddr()` nor
-    `tractor.discovery.parse_maddr()` calls this probe.
-
-    ?TODO, per plan-03 layer B, swap this body for `pyroute2`
-    while retaining the explicit verification boundary.
-
-    '''
-    spec = addr.tunnel
-    if not isinstance(spec, WGTunnelSpec):
-        raise TypeError(
-            f'Unsupported tunnel spec: {type(spec)!r}'
-        )
-
-    iface = iface or spec.iface
-
-    match role:
-        case 'local':
-            field = 'public-key'
-        case 'peer':
-            field = 'peers'
-        case _:
-            raise ValueError(
-                f'Unknown WireGuard key role: {role!r}'
-            )
-
-    if inspection is None:
-        with trio.fail_after(timeout):
-            proc = await trio.run_process(
-                ['wg', 'show', iface, field],
-                capture_stdout=True,
-                check=True,
-            )
-        inspection = proc.stdout.decode()
-
-    if role == 'local':
-        return spec.peer_pubkey == inspection.strip()
-    return spec.peer_pubkey in inspection.split()

tests/discovery/test_wg_inspection.py

@@ -16,7 +16,10 @@ import trio
 from tractor.discovery import (
     read_wg_peers,
     read_wg_pubkey,
+    verify_wg_peer,
+    WGTunnelSpec,
 )
+from tractor.discovery import _tunnel
 
 pyroute2: Any = pytest.importorskip('pyroute2')
 
@@ -24,6 +27,7 @@ pyroute2: Any = pytest.importorskip('pyroute2')
 _PUBKEY: str = 'g3x7z0AdV1rM6UQU22CC7IL3/ivn4DzrE7ikDhCZ/Dc='
 _PEER_1: str = '7PClzcj8o1yAjyPJb0zL2Gt0s2J7yZ6c0JXYqNBGr0E='
 _PEER_2: str = 'H7bJbl1bpY7VzDlB5wI3KjA7JsiYoMWGDJd8dYgc5iw='
+_MISSING_KEY: str = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA='
 
 
 class Attrs:
@@ -215,3 +219,87 @@ def test_wg_client_closes_when_read_fails(
 
     assert instance is not None
     assert instance.closed
+
+
+@pytest.mark.parametrize(
+    ('declared_key', 'expected'),
+    (
+        (_PUBKEY, True),
+        (_PEER_2, True),
+        (_MISSING_KEY, False),
+    ),
+)
+def test_verify_wg_peer(
+    monkeypatch: pytest.MonkeyPatch,
+    declared_key: str,
+    expected: bool,
+) -> None:
+    '''
+    A tunnel declaration can identify either side of one local iface.
+
+    Return one stable key snapshot from the async reader, then prove
+    a local interface key and configured peer both verify while an
+    absent key does not. Also prove the spec selects the iface/netns
+    supplied to the read instead of silently using process defaults.
+
+    '''
+    reads: list[tuple[str, str|None]] = []
+
+    async def read_keys(
+        iface: str,
+        netns: str|None,
+    ) -> tuple[str, tuple[str, ...]]:
+        '''
+        Return one deterministic WireGuard key snapshot.
+
+        '''
+        reads.append((iface, netns))
+        return _PUBKEY, (_PEER_1, _PEER_2)
+
+    monkeypatch.setattr(
+        _tunnel,
+        '_read_wg_keys',
+        read_keys,
+    )
+    spec: WGTunnelSpec = WGTunnelSpec(
+        peer_pubkey=declared_key,
+        iface='wg-test',
+        netns='actor-net',
+    )
+
+    assert trio.run(verify_wg_peer, spec) is expected
+    assert reads == [('wg-test', 'actor-net')]
+
+
+def test_verify_wg_peer_validates_before_read(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A directly constructed tunnel spec can contain a malformed key.
+
+    Install a reader which would fail if called, pass malformed
+    base64, and prove validation rejects the declaration before any
+    kernel-state inspection occurs.
+
+    '''
+    async def unexpected_read(
+        iface: str,
+        netns: str|None,
+    ) -> NoReturn:
+        '''
+        Fail if malformed-key validation reaches the read boundary.
+
+        '''
+        raise AssertionError('WireGuard read must not run')
+
+    monkeypatch.setattr(
+        _tunnel,
+        '_read_wg_keys',
+        unexpected_read,
+    )
+    spec: WGTunnelSpec = WGTunnelSpec(
+        peer_pubkey='not-base64',
+    )
+
+    with pytest.raises(ValueError):
+        trio.run(verify_wg_peer, spec)

tractor/discovery/__init__.py

@@ -40,5 +40,6 @@ from ._tunnel import (
     read_wg_pubkey as read_wg_pubkey,
     strip_tunnels as strip_tunnels,
     tunnels_of as tunnels_of,
+    verify_wg_peer as verify_wg_peer,
     wg8_pubkey as wg8_pubkey,
 )

tractor/discovery/_tunnel.py

@@ -198,7 +198,7 @@ def _wg8_key_str(
     return key
 
 
-def _read_wg_keys(
+def _sync_read_wg_keys(
     iface: str,
     netns: str|None,
 ) -> tuple[str, tuple[str, ...]]:
@@ -275,6 +275,22 @@ def _read_wg_keys(
     )
 
 
+async def _read_wg_keys(
+    iface: str,
+    netns: str|None,
+) -> tuple[str, tuple[str, ...]]:
+    '''
+    Read one WireGuard key snapshot without blocking Trio.
+
+    '''
+    return await trio.to_thread.run_sync(
+        _sync_read_wg_keys,
+        iface,
+        netns,
+        abandon_on_cancel=False,
+    )
+
+
 async def read_wg_pubkey(
     iface: str = 'wg0',
     netns: str|None = None,
@@ -286,11 +302,9 @@ async def read_wg_pubkey(
     keys: tuple[
         str,
         tuple[str, ...],
-    ] = await trio.to_thread.run_sync(
+    ] = await _read_wg_keys(
-        _read_wg_keys,
         iface,
         netns,
-        abandon_on_cancel=False,
     )
     return keys[0]
 
@@ -306,15 +320,39 @@ async def read_wg_peers(
     keys: tuple[
         str,
         tuple[str, ...],
-    ] = await trio.to_thread.run_sync(
+    ] = await _read_wg_keys(
-        _read_wg_keys,
         iface,
         netns,
-        abandon_on_cancel=False,
     )
     return keys[1]
 
 
+async def verify_wg_peer(
+    spec: WGTunnelSpec,
+) -> bool:
+    '''
+    Verify a declared WireGuard identity against local kernel state.
+
+    A source/listen maddr names the local interface key, while a
+    destination/dial maddr names one configured peer. Accept either
+    match without making verification an implicit part of parsing.
+
+    '''
+    declared_key: str = _wg8_key_str(spec.peer_pubkey)
+    keys: tuple[
+        str,
+        tuple[str, ...],
+    ] = await _read_wg_keys(
+        spec.iface,
+        spec.netns,
+    )
+    return (
+        declared_key == keys[0]
+        or
+        declared_key in keys[1]
+    )
+
+
 def _wg_proto_code() -> int:
     '''
     Deliver the installed `py-multiaddr` `/wg/` protocol code.