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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
"""
What the administration panel counts and lists.
Deleting an account leaves a tombstone so the connection log stays readable
(see test_account_deletion). That row is not a user any more, and every place
that counts or lists people has to agree — a hub whose user count only ever
rises is measuring its own history rather than its population.
"""
import base64
import hashlib
import pytest
from meshbay_hub.db.models import User
from sqlalchemy import select
def _auth_key(password: str, username: str) -> str:
salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
return base64.b64encode(
hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
async def _user(client, username, password="a-long-enough-passphrase"):
await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@example.com",
"auth_key": _auth_key(password, username)})
r = await client.post("/v1/users/login", json={
"username": username, "auth_key": _auth_key(password, username)})
return {"Authorization": f"Bearer {r.json()['access_token']}"}
async def _admin(client, db_session, username="root_test"):
headers = await _user(client, username)
user = (await db_session.execute(
select(User).where(User.username == username))).scalar_one()
user.role = "admin"
await db_session.commit()
return headers
@pytest.mark.asyncio
async def test_a_deleted_account_stops_being_counted(client, db_session):
admin = await _admin(client, db_session)
leaver = await _user(client, "ghost_test")
before = (await client.get("/v1/admin/stats", headers=admin)).json()["users"]
await client.request("DELETE", "/v1/users/me", headers=leaver,
json={"auth_key": _auth_key(
"a-long-enough-passphrase", "ghost_test")})
after = (await client.get("/v1/admin/stats", headers=admin)).json()["users"]
assert after == before - 1, "a tombstone is still being counted as a user"
@pytest.mark.asyncio
async def test_a_deleted_account_is_not_listed(client, db_session):
admin = await _admin(client, db_session, "root2_test")
leaver = await _user(client, "vanishing")
await client.request("DELETE", "/v1/users/me", headers=leaver,
json={"auth_key": _auth_key(
"a-long-enough-passphrase", "vanishing")})
r = await client.get("/v1/admin/users", headers=admin)
names = [u["username"] for u in r.json()["users"]]
assert "vanishing" not in names
assert not any(n.startswith("deleted-") for n in names), \
"the tombstone is showing under its placeholder name"
assert r.json()["total"] == len(names)
@pytest.mark.asyncio
async def test_the_member_list_of_a_group_skips_them(client, db_session):
admin = await _admin(client, db_session, "root3_test")
owner = await _user(client, "host3_test")
leaver = await _user(client, "quitter_test")
g = await client.post("/v1/groups", json={"name": "party"}, headers=owner)
gid = g.json()["group_id"]
await client.post(f"/v1/groups/{gid}/members/quitter_test", json={}, headers=owner)
await client.request("DELETE", "/v1/users/me", headers=leaver,
json={"auth_key": _auth_key(
"a-long-enough-passphrase", "quitter_test")})
members = (await client.get(f"/v1/groups/{gid}/members",
headers=owner)).json()["members"]
assert [m["username"] for m in members] == ["host3_test"]
groups = (await client.get("/v1/admin/groups", headers=admin)).json()["groups"]
party = next(g for g in groups if g["name"] == "party")
assert party["member_count"] == 1, \
f"the group still counts its departed member ({party['member_count']})"
# ── Where a node actually is ────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_a_node_is_recorded_at_the_address_it_announced_from(
client, db_session):
"""
The hub sees the connection; the node only tells us what a STUN server told
it. The observed address is the evidence, and the announcement carries an
Ed25519 signature over a fresh timestamp, so it is the address of whoever
holds the node key.
"""
import base64
import time
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_hub.db.models import Node
admin = await _admin(client, db_session, "root4_test")
owner = await _user(client, "nodeowner")
uid = (await db_session.execute(
select(User.id).where(User.username == "nodeowner"))).scalar_one()
sk = Ed25519PrivateKey.generate()
pk_b64 = base64.b64encode(sk.public_key().public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw)).decode()
ts = int(time.time())
message = f"meshbay:node_announce:{uid}:{pk_b64}:{ts}".encode()
r = await client.post("/v1/nodes/announce", headers=owner, json={
"pk_node": pk_b64,
# A claim, and a deliberately implausible one.
"endpoint_hint": "203.0.113.42:19001",
"timestamp": ts,
"signature": base64.b64encode(sk.sign(message)).decode(),
})
assert r.status_code in (200, 201), r.text
node = (await db_session.execute(
select(Node).where(Node.pk_node == pk_b64))).scalar_one()
assert node.observed_ip, "no address was recorded for the announcement"
assert node.observed_ip != "203.0.113.42", \
"the hub recorded what the node claimed instead of what it saw"
assert node.last_seen is not None
listed = (await client.get("/v1/admin/nodes", headers=admin)).json()["nodes"]
row = next(n for n in listed if n["pk_node"] == pk_b64)
assert row["observed_ip"] == node.observed_ip
assert row["endpoint_hint"] == "203.0.113.42:19001", \
"the claim is still shown, just not as the answer"
assert row["username"] == "nodeowner"
|