fix(hub): require proof of possession on node announce — closes M8

Phase 11.5.10.

POST /v1/nodes/announce accepted any pk_node with no proof the announcer held
the matching private key, so a user could register a node record carrying
someone else's node key, and records accumulated without limit.

The announcer now signs a domain-separated message binding the key to their
account — meshbay:node_announce:{user_id}:{pk_node}:{timestamp} — reusing the
shape already proven by /v1/nodes/auth, so a signature for one can never
satisfy the other. Same 60-second window.

Re-announcing the same key now updates the existing record in place instead of
creating a new row.

Three test helpers had to be taught to sign, which is the useful part: nothing
in the suite had ever exercised announce with an attacker's key. The new tests
cover the missing proof, a foreign key, a stale timestamp, and idempotence.

Note for the record: the node key is independent of the user's identity key.
Two hub tests asserted the announced pk_node equalled the user's pk_ed, which
happened to be true only because the daemon announces its keystore key. They
now assert against the announced key itself.

Tests: 157 hub+common, node suite green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

1 files changed, 96 insertions, 2 deletions

diff --git a/packages/meshbay-hub/tests/test_node_ws_auth.py b/packages/meshbay-hub/tests/test_node_ws_auth.py
index 9ec251d..def5e66 100644
--- a/packages/meshbay-hub/tests/test_node_ws_auth.py
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -40,13 +40,22 @@ async def _make_user(client, username: str) -> dict:
         "auth_key": base64.b64encode(b"k" * 32).decode(),
     })
     assert r.status_code == 200, r.text
-    return {"user_id": user_id, "token": r.json()["access_token"], "pk_ed": pk_ed}
+    return {"user_id": user_id, "token": r.json()["access_token"],
+            "pk_ed": pk_ed, "sk_ed": sk_ed}
 
 
 async def _announce_node(client, user: dict) -> str:
+    # Announce now requires proof of possession of the node key (M8).
+    import time as _t
+    ts = int(_t.time())
+    msg = f"meshbay:node_announce:{user['user_id']}:{user['pk_ed']}:{ts}".encode()
     r = await client.post(
         "/v1/nodes/announce",
-        json={"pk_node": user["pk_ed"], "endpoint_hint": "test"},
+        json={
+            "pk_node": user["pk_ed"], "endpoint_hint": "test",
+            "timestamp": ts,
+            "signature": base64.b64encode(user["sk_ed"].sign(msg)).decode(),
+        },
         headers={"Authorization": f"Bearer {user['token']}"},
     )
     assert r.status_code == 201, r.text
@@ -250,3 +259,88 @@ async def test_ws_node_may_narrow_its_group_set(client):
         _node_token(user), node_id, [created[0]])
     assert resolved == node_id
     assert groups == [created[0]]
+
+
+# ── M8: announce proof of possession ─────────────────────────────────────────
+
+def _announce_payload(user_id: str, sk, pk_b64: str, ts: int | None = None):
+    import time as _t
+    ts = ts if ts is not None else int(_t.time())
+    msg = f"meshbay:node_announce:{user_id}:{pk_b64}:{ts}".encode()
+    return {
+        "pk_node": pk_b64,
+        "endpoint_hint": "test",
+        "timestamp": ts,
+        "signature": base64.b64encode(sk.sign(msg)).decode(),
+    }
+
+
+@pytest.mark.asyncio
+async def test_announce_requires_proof_of_possession(client):
+    """
+    M8: /v1/nodes/announce accepted any pk_node with no proof the announcer held
+    the private key, so a user could announce a record carrying someone else's
+    node key.
+    """
+    user = await _make_user(client, "ann1")
+    r = await client.post(
+        "/v1/nodes/announce",
+        json={"pk_node": user["pk_ed"], "endpoint_hint": "test"},
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert r.status_code == 400, r.text
+
+
+@pytest.mark.asyncio
+async def test_announce_rejects_foreign_key(client):
+    """M8: announcing someone else's public key must fail — no matching private key."""
+    user = await _make_user(client, "ann2")
+    victim_sk = Ed25519PrivateKey.generate()
+    victim_pk = pk_to_b64(victim_sk.public_key())
+
+    attacker_sk = Ed25519PrivateKey.generate()
+    payload = _announce_payload(user["user_id"], attacker_sk, victim_pk)
+
+    r = await client.post(
+        "/v1/nodes/announce", json=payload,
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert r.status_code == 401, r.text
+
+
+@pytest.mark.asyncio
+async def test_announce_rejects_stale_timestamp(client):
+    """M8: a captured announce must not be replayable later."""
+    import time as _t
+    user = await _make_user(client, "ann3")
+    sk = Ed25519PrivateKey.generate()
+    payload = _announce_payload(
+        user["user_id"], sk, pk_to_b64(sk.public_key()), ts=int(_t.time()) - 3600)
+
+    r = await client.post(
+        "/v1/nodes/announce", json=payload,
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert r.status_code == 401, r.text
+
+
+@pytest.mark.asyncio
+async def test_announce_with_valid_proof_succeeds_and_is_idempotent(client):
+    """The legitimate path works, and re-announcing updates rather than piling up rows."""
+    user = await _make_user(client, "ann4")
+    sk = Ed25519PrivateKey.generate()
+    pk_b64 = pk_to_b64(sk.public_key())
+
+    first = await client.post(
+        "/v1/nodes/announce", json=_announce_payload(user["user_id"], sk, pk_b64),
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert first.status_code == 201, first.text
+
+    second = await client.post(
+        "/v1/nodes/announce", json=_announce_payload(user["user_id"], sk, pk_b64),
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert second.status_code == 201, second.text
+    assert second.json()["node_id"] == first.json()["node_id"], (
+        "re-announcing the same key must not create a second node record (M8)")