feat: Phase 12 — P2P crypto material, password split, node Ed25519 auth

Baseline commit capturing in-progress Phase 12 work that was already present
in the working tree (uncommitted) before the Phase 11.5 security remediation
begins. Committed as-is, without review or modification, so that remediation
changes arrive as a separable diff.

Contents: BundleStore (P2P GEK + keypair bundles), password split
(auth_key / bundle_key), node Ed25519 auth (POST /v1/nodes/auth, node-scoped
JWT), GEK-HMAC handshake proof with DTLS channel binding, Ed25519 admin
challenge-response, node local admin UI rewrite, browser key persistence.

Not authored in this session — captured to establish a baseline.

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

2 files changed, 507 insertions, 53 deletions

diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py
index f7c499e..a8232c1 100644
--- a/packages/meshbay-hub/tests/test_hub_api.py
+++ b/packages/meshbay-hub/tests/test_hub_api.py
@@ -10,7 +10,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from cryptography.hazmat.primitives import serialization
 
-from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek
+from meshbay_common.crypto import pk_to_b64
 from meshbay_hub.api.deps import set_admin_usernames
 
 
@@ -210,12 +210,8 @@ async def test_announce_and_get_node(client):
 # ── Groups + GEK bundles ──────────────────────────────────────────────────────
 
 @pytest.mark.asyncio
-async def test_group_gek_roundtrip(client):
-    """Admin creates group, wraps GEK for member, member retrieves and can unwrap."""
-    import jwt as pyjwt
-    from meshbay_common.crypto import unwrap_gek
-
-    # Register admin (alice2) and member (bob2)
+async def test_group_member_add(client):
+    """Admin creates group and adds member (GEK exchange happens P2P on node)."""
     pk_ed_a, pk_x_a, _ = _gen_user_keys()
     pk_ed_b, pk_x_b, sk_x_b = _gen_user_keys()
 
@@ -227,46 +223,29 @@ async def test_group_gek_roundtrip(client):
             "username": uname, "email": email, "password": pwd,
             "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
 
-    def _token(uname, pwd):
-        async def _inner():
-            r = await client.post("/v1/users/login",
-                                  json={"username": uname, "password": pwd})
-            return r.json()["access_token"]
-        return _inner
-
     alice_token = (await client.post("/v1/users/login",
                    json={"username": "alice2", "password": "alicepass99"})).json()["access_token"]
-    bob_token   = (await client.post("/v1/users/login",
-                   json={"username": "bob2",   "password": "bobpass99"})).json()["access_token"]
 
     a_hdrs = {"Authorization": f"Bearer {alice_token}"}
-    b_hdrs = {"Authorization": f"Bearer {bob_token}"}
 
-    # Alice creates group
     r = await client.post("/v1/groups", json={"name": "mygroup"}, headers=a_hdrs)
     assert r.status_code == 201
     group_id = r.json()["group_id"]
 
-    # Alice generates GEK and wraps it for bob
-    gek = generate_gek()
-    pk_bob_raw = base64.b64decode(pk_x_b)
-    bundle = wrap_gek(gek, pk_bob_raw)
-
-    r = await client.post(f"/v1/groups/{group_id}/members/bob2/gek",
-                          json=bundle, headers=a_hdrs)
+    # Add bob as member (hub handles membership only, GEK exchange is P2P)
+    r = await client.post(f"/v1/groups/{group_id}/members/bob2",
+                          json={}, headers=a_hdrs)
     assert r.status_code == 201
 
-    # Bob retrieves his bundle
-    r = await client.get(f"/v1/groups/{group_id}/gek", headers=b_hdrs)
+    # Verify bob is in the group
+    bob_token = (await client.post("/v1/users/login",
+                 json={"username": "bob2", "password": "bobpass99"})).json()["access_token"]
+    b_hdrs = {"Authorization": f"Bearer {bob_token}"}
+    r = await client.get(f"/v1/groups/{group_id}/members", headers=b_hdrs)
     assert r.status_code == 200
-    retrieved = r.json()
-
-    # Bob unwraps — must recover original GEK
-    sk_b_raw = sk_x_b.private_bytes(
-        serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
-        serialization.NoEncryption())
-    recovered = unwrap_gek(retrieved, sk_b_raw, pk_bob_raw)
-    assert recovered == gek
+    members = [m["username"] for m in r.json()["members"]]
+    assert "alice2" in members
+    assert "bob2" in members
 
 
 @pytest.mark.asyncio
@@ -291,12 +270,9 @@ async def test_non_admin_cannot_add_member(client):
                           headers={"Authorization": f"Bearer {charlie_token}"})
     group_id = r.json()["group_id"]
 
-    gek = generate_gek()
-    bundle = wrap_gek(gek, base64.b64decode(pk_x_b))
-
     # Dan (non-admin) tries to add a member → 403
-    r = await client.post(f"/v1/groups/{group_id}/members/charlie/gek",
-                          json=bundle,
+    r = await client.post(f"/v1/groups/{group_id}/members/charlie",
+                          json={},
                           headers={"Authorization": f"Bearer {dan_token}"})
     assert r.status_code == 403
 
@@ -331,10 +307,8 @@ async def test_jwt_contains_groups_claim(client):
                           headers={"Authorization": f"Bearer {alice_token}"})
     group_id = r.json()["group_id"]
 
-    gek = generate_gek()
-    bundle = wrap_gek(gek, base64.b64decode(pk_x_b))
-    await client.post(f"/v1/groups/{group_id}/members/grp_bob/gek",
-                      json=bundle,
+    await client.post(f"/v1/groups/{group_id}/members/grp_bob",
+                      json={},
                       headers={"Authorization": f"Bearer {alice_token}"})
 
     # Login again — groups should contain the new group
@@ -381,10 +355,8 @@ async def test_my_groups(client):
     r = await client.post("/v1/groups", json={"name": "mg-group"},
                           headers={"Authorization": f"Bearer {alice_token}"})
     group_id = r.json()["group_id"]
-    gek = generate_gek()
-    bundle = wrap_gek(gek, base64.b64decode(pk_x_b))
-    await client.post(f"/v1/groups/{group_id}/members/mg_bob/gek",
-                      json=bundle,
+    await client.post(f"/v1/groups/{group_id}/members/mg_bob",
+                      json={},
                       headers={"Authorization": f"Bearer {alice_token}"})
 
     # Re-login to get fresh token with group claims
@@ -559,8 +531,7 @@ async def test_registered_email_not_plaintext(client, app):
 
 @pytest.mark.asyncio
 async def test_password_rehash_on_login(client, app):
-    """Users with pw_version=1 get rehashed to current version on login."""
-    from meshbay_hub.auth import current_pw_version
+    """Users with pw_version=1 get rehashed to v2 on legacy password login."""
     from meshbay_hub.db.engine import get_db
     from meshbay_hub.db.models import User
     from sqlalchemy import select
@@ -590,16 +561,16 @@ async def test_password_rehash_on_login(client, app):
         await db.commit()
         break
 
-    # Login should succeed and trigger rehash
+    # Login should succeed and trigger legacy rehash (v1 -> v2)
     r = await client.post("/v1/users/login", json={
         "username": "rehash_user", "password": "rehashpass9"})
     assert r.status_code == 200
 
-    # Verify pw_version is now current
+    # Verify pw_version is now 2 (legacy rehash stays within password scheme)
     async for db in get_db():
         result = await db.execute(select(User).where(User.username == "rehash_user"))
         user = result.scalar_one()
-        assert user.pw_version == current_pw_version()
+        assert user.pw_version == 2
         break
 
     # Login still works after rehash
@@ -722,3 +693,205 @@ async def test_webapp_html_includes_scripts(client):
     assert 'type="module"' in html
     assert 'rel="stylesheet"' in html
     assert 'href="/style.css"' in html
+
+
+# ── Password split (T1 fix) ─────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_register_with_auth_key(client, app):
+    """Registration with auth_key sets pw_version 3."""
+    from meshbay_hub.db.engine import get_db
+    from meshbay_hub.db.models import User
+    from sqlalchemy import select
+
+    pk_ed, pk_x, _ = _gen_user_keys()
+    r = await client.post("/v1/users/register", json={
+        "username": "authuser",
+        "email": "auth@test.com",
+        "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+    assert r.status_code == 201
+    assert "user_id" in r.json()
+
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "authuser"))
+        user = result.scalar_one()
+        assert user.pw_version == 3
+        break
+
+
+@pytest.mark.asyncio
+async def test_register_with_password_sets_v2(client, app):
+    """Registration with raw password (legacy) sets pw_version 2."""
+    from meshbay_hub.db.engine import get_db
+    from meshbay_hub.db.models import User
+    from sqlalchemy import select
+
+    pk_ed, pk_x, _ = _gen_user_keys()
+    r = await client.post("/v1/users/register", json={
+        "username": "legacyreg",
+        "email": "legacy@test.com",
+        "password": "legacypass99",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+    assert r.status_code == 201
+
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "legacyreg"))
+        user = result.scalar_one()
+        assert user.pw_version == 2
+        break
+
+
+@pytest.mark.asyncio
+async def test_register_no_credentials_rejected(client):
+    """Registration without auth_key or password returns 400."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    r = await client.post("/v1/users/register", json={
+        "username": "nocred",
+        "email": "nocred@test.com",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+    assert r.status_code == 400
+    assert "auth_key or password required" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_login_with_auth_key(client, app):
+    """Login with auth_key for pw_version 3 account succeeds."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    auth_key = "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo"
+    await client.post("/v1/users/register", json={
+        "username": "authlogin",
+        "email": "authlogin@test.com",
+        "auth_key": auth_key,
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+
+    r = await client.post("/v1/users/login", json={
+        "username": "authlogin", "auth_key": auth_key})
+    assert r.status_code == 200
+    data = r.json()
+    assert "access_token" in data
+    assert "refresh_token" in data
+
+
+@pytest.mark.asyncio
+async def test_login_auth_key_wrong_rejected(client):
+    """Login with wrong auth_key returns 401."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "authwrong",
+        "email": "authwrong@test.com",
+        "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+
+    r = await client.post("/v1/users/login", json={
+        "username": "authwrong", "auth_key": "d3JvbmdrZXl3cm9uZ2tleXdyb25na2V5d3Jvbmc="})
+    assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_login_v3_account_password_only_rejected(client):
+    """Login with raw password to a v3 (auth_key) account returns 401."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "v3nopw",
+        "email": "v3nopw@test.com",
+        "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+
+    r = await client.post("/v1/users/login", json={
+        "username": "v3nopw", "password": "somepassword"})
+    assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_login_legacy_upgrade_required(client):
+    """Legacy account (pw_version 2) with auth_key only returns auth_upgrade_required."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "legacyupg",
+        "email": "legacyupg@test.com",
+        "password": "legacypass99",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+
+    r = await client.post("/v1/users/login", json={
+        "username": "legacyupg", "auth_key": "dGVzdGF1dGhrZXl0ZXN0YXV0aGtleXRlc3RhdXRo"})
+    assert r.status_code == 401
+    assert r.json()["detail"] == "auth_upgrade_required"
+
+
+@pytest.mark.asyncio
+async def test_login_legacy_migration(client, app):
+    """Legacy account migrates to auth_key on login with both fields."""
+    from meshbay_hub.auth import current_pw_version
+    from meshbay_hub.db.engine import get_db
+    from meshbay_hub.db.models import User
+    from sqlalchemy import select
+
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "migrateuser",
+        "email": "migrate@test.com",
+        "password": "migratepass9",
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+
+    # Verify starts at pw_version 2
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "migrateuser"))
+        user = result.scalar_one()
+        assert user.pw_version == 2
+        break
+
+    auth_key = "bWlncmF0ZWF1dGhrZXltaWdyYXRlYXV0aGtleW1p"
+
+    # Login with password + auth_key → should succeed and migrate
+    r = await client.post("/v1/users/login", json={
+        "username": "migrateuser",
+        "password": "migratepass9",
+        "auth_key": auth_key,
+    })
+    assert r.status_code == 200
+
+    # Verify pw_version is now 3 (migrated)
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "migrateuser"))
+        user = result.scalar_one()
+        assert user.pw_version == current_pw_version()
+        assert user.pw_version == 3
+        break
+
+    # Login again with auth_key only → should succeed (migrated account)
+    r = await client.post("/v1/users/login", json={
+        "username": "migrateuser", "auth_key": auth_key})
+    assert r.status_code == 200
+    assert "access_token" in r.json()
+
+    # Old password no longer works (hash was replaced with auth_key hash)
+    r = await client.post("/v1/users/login", json={
+        "username": "migrateuser", "password": "migratepass9"})
+    assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_login_no_credentials_rejected(client):
+    """Login without auth_key or password returns 401."""
+    r = await client.post("/v1/users/login", json={"username": "nobody"})
+    assert r.status_code == 401
+    assert "No credentials" in r.json()["detail"]
+
+
diff --git a/packages/meshbay-hub/tests/test_node_auth.py b/packages/meshbay-hub/tests/test_node_auth.py
new file mode 100644
index 0000000..e629d20
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_node_auth.py
@@ -0,0 +1,281 @@
+"""Tests for node Ed25519 authentication and JWT scope enforcement.
+
+Node auth flow: register user (browser keys) → login → link node key → node auth.
+The node's Ed25519 key is separate from the user's browser key.
+"""
+
+import base64
+import time
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+from cryptography.hazmat.primitives import serialization
+
+
+def _gen_ed25519():
+    sk = Ed25519PrivateKey.generate()
+    pk_raw = sk.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    return sk, base64.b64encode(pk_raw).decode()
+
+
+def _gen_x25519():
+    sk = X25519PrivateKey.generate()
+    pk_raw = sk.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    return sk, base64.b64encode(pk_raw).decode()
+
+
+async def _register(client, username, pk_ed_b64, pk_x_b64):
+    r = await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@test.local",
+        "password": "testpass99",
+        "pk_user_ed25519": pk_ed_b64, "pk_user_x25519": pk_x_b64,
+    })
+    assert r.status_code == 201
+    return r.json()["user_id"]
+
+
+async def _user_login(client, username, password="testpass99"):
+    r = await client.post("/v1/users/login", json={
+        "username": username, "password": password,
+    })
+    assert r.status_code == 200
+    return r.json()["access_token"]
+
+
+async def _link_node_key(client, user_token, pk_node_ed_b64):
+    r = await client.put("/v1/users/me/node_key", json={
+        "pk_node_ed25519": pk_node_ed_b64,
+    }, headers={"Authorization": f"Bearer {user_token}"})
+    assert r.status_code == 200
+    return r
+
+
+async def _node_auth(client, username, sk_node_ed):
+    timestamp = int(time.time())
+    message = f"meshbay:node_auth:{username}:{timestamp}".encode()
+    signature = sk_node_ed.sign(message)
+    r = await client.post("/v1/nodes/auth", json={
+        "username": username,
+        "timestamp": timestamp,
+        "signature": base64.b64encode(signature).decode(),
+    })
+    return r
+
+
+async def _setup_node_user(client, username):
+    """Register user with browser keys, login, link node key. Returns (node_sk, node_token)."""
+    _, pk_ed = _gen_ed25519()
+    _, pk_x = _gen_x25519()
+    await _register(client, username, pk_ed, pk_x)
+
+    user_token = await _user_login(client, username)
+
+    sk_node, pk_node = _gen_ed25519()
+    await _link_node_key(client, user_token, pk_node)
+
+    return sk_node, user_token
+
+
+# ── Core auth tests ──────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_node_auth_success(client):
+    sk_node, _ = await _setup_node_user(client, "nodeuser")
+
+    r = await _node_auth(client, "nodeuser", sk_node)
+    assert r.status_code == 200
+    data = r.json()
+    assert "access_token" in data
+    assert data["token_type"] == "bearer"
+
+
+@pytest.mark.asyncio
+async def test_node_auth_wrong_key_rejected(client):
+    await _setup_node_user(client, "nodeuser2")
+
+    wrong_sk, _ = _gen_ed25519()
+    r = await _node_auth(client, "nodeuser2", wrong_sk)
+    assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_node_auth_stale_timestamp(client):
+    sk_node, _ = await _setup_node_user(client, "nodeuser3")
+
+    timestamp = int(time.time()) - 120
+    message = f"meshbay:node_auth:nodeuser3:{timestamp}".encode()
+    signature = sk_node.sign(message)
+    r = await client.post("/v1/nodes/auth", json={
+        "username": "nodeuser3",
+        "timestamp": timestamp,
+        "signature": base64.b64encode(signature).decode(),
+    })
+    assert r.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_node_auth_fails_without_node_key(client):
+    """Node auth must fail if no node key has been registered."""
+    _, pk_ed = _gen_ed25519()
+    _, pk_x = _gen_x25519()
+    await _register(client, "nokey_user", pk_ed, pk_x)
+
+    sk_any, _ = _gen_ed25519()
+    r = await _node_auth(client, "nokey_user", sk_any)
+    assert r.status_code == 401
+    assert "No node key registered" in r.json()["detail"]
+
+
+# ── Scope enforcement tests ──────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_node_scope_blocks_group_create(client):
+    sk_node, _ = await _setup_node_user(client, "operator")
+
+    r = await _node_auth(client, "operator", sk_node)
+    node_token = r.json()["access_token"]
+
+    r = await client.post("/v1/groups", json={
+        "name": "forbidden-group", "visibility": "private",
+    }, headers={"Authorization": f"Bearer {node_token}"})
+    assert r.status_code == 403
+    assert "Node-scoped" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_node_scope_blocks_add_member(client):
+    sk_node, user_token = await _setup_node_user(client, "op1")
+
+    r = await client.post("/v1/groups", json={
+        "name": "mygroup", "visibility": "private", "join_policy": "invite",
+    }, headers={"Authorization": f"Bearer {user_token}"})
+    assert r.status_code == 201
+    gid = r.json()["group_id"]
+
+    _, pk2 = _gen_ed25519()
+    _, px2 = _gen_x25519()
+    await _register(client, "member1", pk2, px2)
+
+    r = await _node_auth(client, "op1", sk_node)
+    node_token = r.json()["access_token"]
+
+    r = await client.post(f"/v1/groups/{gid}/members/member1",
+                          headers={"Authorization": f"Bearer {node_token}"})
+    assert r.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_node_scope_blocks_delete_group(client):
+    sk_node, user_token = await _setup_node_user(client, "op2")
+
+    r = await client.post("/v1/groups", json={
+        "name": "deleteme", "visibility": "public", "join_policy": "open",
+    }, headers={"Authorization": f"Bearer {user_token}"})
+    gid = r.json()["group_id"]
+
+    r = await _node_auth(client, "op2", sk_node)
+    node_token = r.json()["access_token"]
+
+    r = await client.delete(f"/v1/groups/{gid}",
+                            headers={"Authorization": f"Bearer {node_token}"})
+    assert r.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_node_scope_allows_read_members(client):
+    sk_node, user_token = await _setup_node_user(client, "op3")
+
+    r = await client.post("/v1/groups", json={
+        "name": "readgroup", "visibility": "private", "join_policy": "invite",
+    }, headers={"Authorization": f"Bearer {user_token}"})
+    gid = r.json()["group_id"]
+
+    r = await _node_auth(client, "op3", sk_node)
+    node_token = r.json()["access_token"]
+
+    r = await client.get(f"/v1/groups/{gid}/members",
+                         headers={"Authorization": f"Bearer {node_token}"})
+    assert r.status_code == 200
+    assert len(r.json()["members"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_node_scope_allows_pubkey_lookup(client):
+    sk_node, _ = await _setup_node_user(client, "op4")
+
+    r = await _node_auth(client, "op4", sk_node)
+    node_token = r.json()["access_token"]
+
+    r = await client.get("/v1/users/op4/pubkeys",
+                         headers={"Authorization": f"Bearer {node_token}"})
+    assert r.status_code == 200
+    assert "pk_ed25519" in r.json()
+    assert r.json()["pk_node_ed25519"] is not None
+
+
+@pytest.mark.asyncio
+async def test_user_scope_still_works(client):
+    """Verify that user-scoped tokens (browser login) still have full access."""
+    _, pk_ed = _gen_ed25519()
+    _, pk_x = _gen_x25519()
+    await _register(client, "webuser", pk_ed, pk_x)
+
+    token = await _user_login(client, "webuser")
+
+    r = await client.post("/v1/groups", json={
+        "name": "browser-group", "visibility": "public", "join_policy": "open",
+    }, headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 201
+
+
+# ── Node key registration tests ──────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_link_node_key(client):
+    _, pk_ed = _gen_ed25519()
+    _, pk_x = _gen_x25519()
+    await _register(client, "linkuser", pk_ed, pk_x)
+
+    token = await _user_login(client, "linkuser")
+    _, pk_node = _gen_ed25519()
+
+    r = await client.put("/v1/users/me/node_key", json={
+        "pk_node_ed25519": pk_node,
+    }, headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 200
+    assert r.json()["pk_node_ed25519"] == pk_node
+
+
+@pytest.mark.asyncio
+async def test_link_node_key_invalid_format(client):
+    _, pk_ed = _gen_ed25519()
+    _, pk_x = _gen_x25519()
+    await _register(client, "badkey", pk_ed, pk_x)
+
+    token = await _user_login(client, "badkey")
+
+    r = await client.put("/v1/users/me/node_key", json={
+        "pk_node_ed25519": "not-valid-base64!!!",
+    }, headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_link_node_key_blocked_for_node_scope(client):
+    """Node-scoped tokens must not be able to change the node key."""
+    sk_node, _ = await _setup_node_user(client, "sneaky")
+
+    r = await _node_auth(client, "sneaky", sk_node)
+    node_token = r.json()["access_token"]
+
+    _, pk_evil = _gen_ed25519()
+    r = await client.put("/v1/users/me/node_key", json={
+        "pk_node_ed25519": pk_evil,
+    }, headers={"Authorization": f"Bearer {node_token}"})
+    assert r.status_code == 403