""" Signing in to the hub with a device key. The passphrase stays the account's credential and its only recovery path; this is the day-to-day path once a device is registered, so a client does not derive a key from the passphrase on every sign-in. The thing to be careful about, and the reason these tests are written as refusals: this looks like the key directory that was **H3**, and must not become one. What keeps it apart — * nothing reads these keys but the hub itself, and no endpoint publishes them; * no group key is ever wrapped for one; * they are **not** the per-node identity keys, which are generated per node, pinned there, and never leave that relationship. `test_the_hub_publishes_no_device_keys` is the one that would notice if that stopped being true. """ import base64 import time import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import pk_to_b64 def _device(): sk = Ed25519PrivateKey.generate() return sk, pk_to_b64(sk.public_key()) def _sign(sk, username: str, ts: int | None = None) -> dict: ts = int(time.time()) if ts is None else ts message = f"meshbay:user_auth:{username}:{ts}".encode() return {"username": username, "timestamp": ts, "signature": base64.b64encode(sk.sign(message)).decode()} async def _account(client, username="alice_test") -> str: await client.post("/v1/users/register", json={ "username": username, "auth_key": "k" * 44, "email": f"{username}@example.invalid"}) resp = await client.post("/v1/users/login", json={ "username": username, "auth_key": "k" * 44}) return resp.json()["access_token"] async def _register_device(client, token: str, pk: str, label: str = ""): return await client.post( "/v1/users/devices", json={"pk_auth_ed25519": pk, "label": label}, headers={"Authorization": f"Bearer {token}"}) # ── The path that must work ────────────────────────────────────────────────── async def test_a_registered_device_signs_in(client): token = await _account(client) sk, pk = _device() assert (await _register_device(client, token, pk, "laptop")).status_code == 201 resp = await client.post("/v1/users/auth", json=_sign(sk, "alice_test")) assert resp.status_code == 200, resp.text body = resp.json() assert body["access_token"] and body["refresh_token"] assert body["token_type"] == "bearer" async def test_the_session_it_returns_is_a_real_one(client): """A device sign-in must produce a token that works, not a special case.""" token = await _account(client) sk, pk = _device() await _register_device(client, token, pk) device_token = (await client.post( "/v1/users/auth", json=_sign(sk, "alice_test"))).json()["access_token"] me = await client.get("/v1/users/me", headers={"Authorization": f"Bearer {device_token}"}) assert me.status_code == 200 assert me.json()["username"] == "alice_test" async def test_several_devices_on_one_account(client): """The whole point: a browser and a desktop client are both this person.""" token = await _account(client) sk_a, pk_a = _device() sk_b, pk_b = _device() await _register_device(client, token, pk_a, "browser") await _register_device(client, token, pk_b, "desktop") for sk in (sk_a, sk_b): assert (await client.post("/v1/users/auth", json=_sign(sk, "alice_test"))).status_code == 200 listed = await client.get("/v1/users/devices", headers={"Authorization": f"Bearer {token}"}) assert {d["label"] for d in listed.json()["devices"]} == {"browser", "desktop"} # ── What must not work ─────────────────────────────────────────────────────── async def test_an_unregistered_key_is_refused(client): await _account(client) sk, _ = _device() resp = await client.post("/v1/users/auth", json=_sign(sk, "alice_test")) assert resp.status_code == 401 async def test_another_accounts_device_cannot_sign_in_as_you(client): token_a = await _account(client, "alice_test") await _account(client, "bob_test") sk, pk = _device() await _register_device(client, token_a, pk) # Alice's device, Bob's name. The signature covers the username, so it does # not verify — and even if it did, the key is not on Bob's account. resp = await client.post("/v1/users/auth", json=_sign(sk, "bob_test")) assert resp.status_code == 401 async def test_a_stale_signature_is_refused(client): """The window is what stops a captured signature being replayed later.""" token = await _account(client) sk, pk = _device() await _register_device(client, token, pk) old = int(time.time()) - 3600 resp = await client.post("/v1/users/auth", json=_sign(sk, "alice_test", ts=old)) assert resp.status_code == 401 assert "timestamp" in resp.json()["detail"].lower() async def test_a_signature_for_a_different_timestamp_does_not_verify(client): token = await _account(client) sk, pk = _device() await _register_device(client, token, pk) signed = _sign(sk, "alice_test") signed["timestamp"] = signed["timestamp"] + 1 # inside the window, wrong assert (await client.post("/v1/users/auth", json=signed)).status_code == 401 async def test_a_device_cannot_enrol_itself(client): """Registration needs an existing session, which means the passphrase was entered a moment ago. Otherwise anyone could add a key to any account.""" await _account(client) _, pk = _device() resp = await client.post("/v1/users/devices", json={"pk_auth_ed25519": pk, "label": "sneaky"}) # 422 rather than 401: with no Authorization header at all, FastAPI refuses # at dependency resolution before the handler runs. A refusal either way — # what matters is that nothing was created. assert resp.status_code in (401, 403, 422) signed_in = await client.post("/v1/users/auth", json=_sign( Ed25519PrivateKey.generate(), "alice_test")) assert signed_in.status_code == 401 async def test_one_key_belongs_to_one_account(client): """Sharing it would make "who signed in" a question with two answers.""" token_a = await _account(client, "alice_test") token_b = await _account(client, "bob_test") _, pk = _device() await _register_device(client, token_a, pk) resp = await _register_device(client, token_b, pk) assert resp.status_code == 409 async def test_a_suspended_account_cannot_sign_in_with_a_device(client): token = await _account(client) sk, pk = _device() await _register_device(client, token, pk) from meshbay_hub.db.engine import get_session_factory from meshbay_hub.db.models import User from sqlalchemy import update async with get_session_factory()() as s: await s.execute(update(User).where(User.username == "alice_test") .values(status="suspended")) await s.commit() resp = await client.post("/v1/users/auth", json=_sign(sk, "alice_test")) assert resp.status_code == 403 async def test_garbage_is_not_a_key(client): token = await _account(client) resp = await _register_device(client, token, "not-base64-at-all!!") assert resp.status_code == 400 # ── Not a key directory ────────────────────────────────────────────────────── async def test_the_hub_publishes_no_device_keys(client): """ **H3 is what this is guarding.** The hub used to publish user public keys and the invite path wrapped the group key for whatever came back. Device auth keys must stay invisible to everyone but the hub: no endpoint returns another account's, and `/pubkeys` must not grow one. """ token = await _account(client, "alice_test") _, pk = _device() await _register_device(client, token, pk) # `/pubkeys` is itself behind a session — it is an account lookup for # invitations, not a public directory — so ask it as a signed-in member. public = await client.get("/v1/users/alice_test/pubkeys", headers={"Authorization": f"Bearer {token}"}) assert public.status_code == 200 body = public.text assert pk not in body, "a device key is reachable through the public lookup" assert "pk_auth" not in body async def test_you_cannot_read_another_accounts_devices(client): token_a = await _account(client, "alice_test") token_b = await _account(client, "bob_test") _, pk = _device() await _register_device(client, token_a, pk, "alice-laptop") listed = await client.get("/v1/users/devices", headers={"Authorization": f"Bearer {token_b}"}) assert listed.status_code == 200 assert listed.json()["devices"] == [] async def test_removing_a_device_stops_it_signing_in(client): token = await _account(client) sk, pk = _device() created = await _register_device(client, token, pk) device_id = created.json()["id"] gone = await client.delete(f"/v1/users/devices/{device_id}", headers={"Authorization": f"Bearer {token}"}) assert gone.status_code == 200 assert (await client.post("/v1/users/auth", json=_sign(sk, "alice_test"))).status_code == 401 async def test_you_cannot_remove_someone_elses_device(client): token_a = await _account(client, "alice_test") token_b = await _account(client, "bob_test") _, pk = _device() device_id = (await _register_device(client, token_a, pk)).json()["id"] resp = await client.delete(f"/v1/users/devices/{device_id}", headers={"Authorization": f"Bearer {token_b}"}) assert resp.status_code == 404 # ── Version floor (C3) ─────────────────────────────────────────────────────── async def test_the_hub_states_a_minimum_client_version(client): """ An installed client meets a newer hub for the first time once the interface ships in a package. Cheap to add now, awkward to retrofit. """ resp = await client.get("/v1/hub/version") assert resp.status_code == 200 client_floor = resp.json()["client"] assert client_floor["minimum"] and client_floor["recommended"]