path: root/packages/meshbay-hub/tests/test_device_auth.py

"""
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"]