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>

2 files changed, 130 insertions, 13 deletions

diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py
index a8232c1..7b75fd1 100644
--- a/packages/meshbay-hub/tests/test_hub_api.py
+++ b/packages/meshbay-hub/tests/test_hub_api.py
@@ -24,6 +24,35 @@ def _gen_user_keys():
     )
 
 
+
+async def _announce_signed(client, token: str) -> tuple[str, str]:
+    """
+    Announce a node with proof of possession (M8).
+
+    The node key is independent of the user's identity key, so this mints a fresh
+    one and signs the domain-separated announce message with it.
+    """
+    import base64 as _b64, time as _t
+
+    me = await client.get("/v1/users/me",
+                          headers={"Authorization": f"Bearer {token}"})
+    user_id = me.json()["user_id"]
+
+    sk_node = Ed25519PrivateKey.generate()
+    pk_node = pk_to_b64(sk_node.public_key())
+    ts = _t.time().__trunc__()
+    msg = f"meshbay:node_announce:{user_id}:{pk_node}:{ts}".encode()
+
+    r = await client.post("/v1/nodes/announce", json={
+        "pk_node": pk_node,
+        "endpoint_hint": "1.2.3.4:19000",
+        "timestamp": ts,
+        "signature": _b64.b64encode(sk_node.sign(msg)).decode(),
+    }, headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 201, r.text
+    return r.json()["node_id"], pk_node
+
+
 # ── Hub info ──────────────────────────────────────────────────────────────────
 
 @pytest.mark.asyncio
@@ -195,15 +224,12 @@ async def test_announce_and_get_node(client):
     token = login.json()["access_token"]
     hdrs = {"Authorization": f"Bearer {token}"}
 
-    r = await client.post("/v1/nodes/announce",
-                          json={"pk_node": pk_ed, "endpoint_hint": "1.2.3.4:19000"},
-                          headers=hdrs)
-    assert r.status_code == 201
-    node_id = r.json()["node_id"]
+    node_id, pk_node = await _announce_signed(client, token)
 
     r2 = await client.get(f"/v1/nodes/{node_id}", headers=hdrs)
     assert r2.status_code == 200
-    assert r2.json()["pk_node"] == pk_ed
+    # The node key is independent of the user identity key (M8).
+    assert r2.json()["pk_node"] == pk_node
     assert r2.json()["endpoint_hint"] == "1.2.3.4:19000"
 
 
@@ -409,10 +435,7 @@ async def test_group_online_nodes(client):
              json={"username": "gn_user", "password": "gnpass999"})).json()["access_token"]
 
     # Announce a node
-    r = await client.post("/v1/nodes/announce", json={
-        "pk_node": pk_ed, "endpoint_hint": "1.2.3.4:19000"},
-        headers={"Authorization": f"Bearer {token}"})
-    node_id = r.json()["node_id"]
+    node_id, pk_node = await _announce_signed(client, token)
 
     # No nodes online yet
     r = await client.get(f"/v1/groups/{group_id}/nodes",
@@ -433,7 +456,7 @@ async def test_group_online_nodes(client):
         nodes = r.json()["nodes"]
         assert len(nodes) == 1
         assert nodes[0]["node_id"] == node_id
-        assert nodes[0]["pk_node"] == pk_ed
+        assert nodes[0]["pk_node"] == pk_node
     finally:
         _connected_nodes.pop(node_id, None)
         _node_groups.pop(node_id, None)
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)")