1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
"""The MNP token: a member's credential to a node, useless at the hub API.
A member hands whatever token it presents to every node it connects to (the MNP
handshake). That must not be the hub session token, which opens the hub API —
otherwise a node operator holds a live credential for the member. `POST
/v1/nodes/mnp-token` mints a short-lived, node-audience token for that purpose;
these tests pin that it authorises to a node and is refused by the hub API.
"""
import pytest
from meshbay_common.handshake import HandshakeError, authorize_token
from meshbay_common.tokens import MNP_AUD
async def _session_token(client, username="mnp_user_test"):
await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@test.local", "auth_key": "k" * 44})
r = await client.post("/v1/users/login", json={
"username": username, "auth_key": "k" * 44})
return r.json()["access_token"]
@pytest.mark.asyncio
async def test_mnp_token_endpoint_needs_a_session(client):
# No Authorization header at all — FastAPI rejects the required header (422),
# like every other authenticated route; the point is it is not minted anonymously.
r = await client.post("/v1/nodes/mnp-token")
assert r.status_code in (401, 403, 422)
@pytest.mark.asyncio
async def test_mnp_token_is_minted_for_a_member(client):
tok = await _session_token(client)
r = await client.post("/v1/nodes/mnp-token",
headers={"Authorization": f"Bearer {tok}"})
assert r.status_code == 200
assert r.json().get("mnp_token")
@pytest.mark.asyncio
async def test_mnp_token_is_refused_at_the_hub_api(client):
"""The whole point: the credential a node receives opens nothing at the hub."""
tok = await _session_token(client, "mnp_api_test")
mnp = (await client.post("/v1/nodes/mnp-token",
headers={"Authorization": f"Bearer {tok}"})).json()["mnp_token"]
# Presenting it to a hub endpoint fails.
r = await client.get("/v1/users/me", headers={"Authorization": f"Bearer {mnp}"})
assert r.status_code == 401
@pytest.mark.asyncio
async def test_a_session_token_is_refused_by_a_node_but_the_mnp_token_is_not(client):
"""The mirror image, at the node's decode: the session token (what the API
accepts) is refused by `authorize_token`, and the MNP token is accepted."""
from meshbay_hub.auth import hub_public_key_pem
# Make the member a member of a group so the MNP token carries it.
tok = await _session_token(client, "mnp_node_test")
H = {"Authorization": f"Bearer {tok}"}
gid = (await client.post("/v1/groups", headers=H, json={
"name": "g", "visibility": "private", "join_policy": "invite"})).json()["group_id"]
mnp = (await client.post("/v1/nodes/mnp-token", headers=H)).json()["mnp_token"]
pk = hub_public_key_pem()
# The session token is refused by the node handshake (wrong audience).
with pytest.raises(HandshakeError):
authorize_token(tok, pk, group_id=gid)
# The MNP token authorises the member to the node.
peer = authorize_token(mnp, pk, group_id=gid)
assert peer.group_id == gid
|