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
|
"""
The `hub.toml` admin allow-list grants an account, not a username.
Deleting an account releases its name, so a list of names granted the admin role
to whoever registered a freed one next. Each listed name is now pinned to the
first active account seen holding it, and only that account is the admin.
"""
import hashlib
import pytest
from meshbay_hub.api.deps import set_admin_usernames
from meshbay_hub.app import _pin_config_admins
from meshbay_hub.db.models import AdminPin, User
from sqlalchemy import select
def _auth_key(password: str, username: str) -> str:
import base64
salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
return base64.b64encode(
hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
PASSWORD = "a-long-enough-passphrase"
async def _register(client, username, email=None):
r = await client.post("/v1/users/register", json={
"username": username, "email": email or f"{username}@example.com",
"auth_key": _auth_key(PASSWORD, username),
})
assert r.status_code in (200, 201), r.text
login = await client.post("/v1/users/login", json={
"username": username, "auth_key": _auth_key(PASSWORD, username)})
assert login.status_code == 200, login.text
return {"Authorization": f"Bearer {login.json()['access_token']}"}
async def _delete_own(client, headers, username):
r = await client.request("DELETE", "/v1/users/me", headers=headers,
json={"auth_key": _auth_key(PASSWORD, username)})
assert r.status_code == 200, r.text
async def _is_admin(client, headers) -> bool:
r = await client.get("/v1/admin/stats", headers=headers)
assert r.status_code in (200, 403), r.text
return r.status_code == 200
@pytest.mark.asyncio
async def test_a_released_name_registered_again_is_not_an_admin(client):
set_admin_usernames(["the_operator"])
first = await _register(client, "the_operator")
assert await _is_admin(client, first)
await _delete_own(client, first, "the_operator")
second = await _register(client, "the_operator", email="someone@example.com")
assert not await _is_admin(client, second)
@pytest.mark.asyncio
async def test_nor_after_a_restart(client, db_session):
"""Startup must not pin the name again to whoever holds it now."""
set_admin_usernames(["the_operator"])
first = await _register(client, "the_operator")
assert await _is_admin(client, first)
await _delete_own(client, first, "the_operator")
second = await _register(client, "the_operator", email="someone@example.com")
await _pin_config_admins(["the_operator"])
assert not await _is_admin(client, second)
impostor = (await db_session.execute(
select(User).where(User.username == "the_operator"))).scalar_one()
assert impostor.role == "user"
@pytest.mark.asyncio
async def test_startup_pins_an_existing_account_before_its_name_is_freed(client, db_session):
"""The upgrade path: the listed account exists before any pin does."""
first = await _register(client, "the_operator")
set_admin_usernames(["the_operator"])
await _pin_config_admins(["the_operator"])
pin = await db_session.get(AdminPin, "the_operator")
owner = (await db_session.execute(
select(User).where(User.username == "the_operator"))).scalar_one()
assert pin is not None and pin.user_id == owner.id
assert owner.role == "admin"
await _delete_own(client, first, "the_operator")
second = await _register(client, "the_operator", email="someone@example.com")
assert not await _is_admin(client, second)
@pytest.mark.asyncio
async def test_a_name_registered_while_the_hub_runs_is_admin_at_once(client):
"""No restart between listing a name and its first sign-in, as before pins."""
set_admin_usernames(["late_operator"])
await _pin_config_admins(["late_operator"])
headers = await _register(client, "late_operator")
assert await _is_admin(client, headers)
@pytest.mark.asyncio
async def test_unlisting_then_relisting_hands_the_name_to_its_new_holder(client, db_session):
set_admin_usernames(["the_operator"])
first = await _register(client, "the_operator")
assert await _is_admin(client, first)
await _delete_own(client, first, "the_operator")
second = await _register(client, "the_operator", email="successor@example.com")
set_admin_usernames([])
await _pin_config_admins([])
assert await db_session.get(AdminPin, "the_operator") is None
set_admin_usernames(["the_operator"])
await _pin_config_admins(["the_operator"])
assert await _is_admin(client, second)
|