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

"""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_token_may_add_a_member_to_its_own_operators_group(client):
    """The node calls this after a CLI `member invite` so the group shows up in
    the invitee's SPA (commit 0443cf8). A node-scoped token is accepted here —
    the `group.admin_id == caller` check is the guard — but only for a group the
    node's operator owns."""
    sk_op, op_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 {op_token}"})
    gid = r.json()["group_id"]

    _, pk2 = _gen_ed25519()
    _, px2 = _gen_x25519()
    await _register(client, "member1", pk2, px2)

    node_token = (await _node_auth(client, "op1", sk_op)).json()["access_token"]

    r = await client.post(f"/v1/groups/{gid}/members/member1",
                          headers={"Authorization": f"Bearer {node_token}"})
    assert r.status_code == 201

    # …but not to a group it does not own.
    sk_other, other_token = await _setup_node_user(client, "op2")
    r = await client.post("/v1/groups", json={"name": "theirs", "visibility": "private"},
                          headers={"Authorization": f"Bearer {other_token}"})
    other_gid = r.json()["group_id"]
    r = await client.post(f"/v1/groups/{other_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
    # An account id and the node's linking key — no user identity keys (H3).
    assert "pk_ed25519" not 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