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
|
"""
Passphrase change — Flow A of docs/auth-confirm.md.
The hub's part is small: re-prove the current passphrase, swap the auth_key
verifier, invalidate every other session, keep the caller's. The re-wrapping of
per-node identity bundles is the client's job and does not touch the hub, so it
is not exercised here.
"""
import base64
import hashlib
import pytest
from sqlalchemy import select
from meshbay_hub.db.models import IPLog, RefreshToken, User
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 _register(client, username, password="the-first-passphrase"):
r = await client.post("/v1/users/register", json={
"username": username, "email": 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 login.json()
@pytest.mark.asyncio
async def test_change_then_sign_in_with_the_new_passphrase(client):
old, new = "the-first-passphrase", "a-second-passphrase-entirely"
session = await _register(client, "alice_test", old)
r = await client.post("/v1/users/password", json={
"old_auth_key": _auth_key(old, "alice_test"),
"new_auth_key": _auth_key(new, "alice_test"),
}, headers={"Authorization": f"Bearer {session['access_token']}"})
assert r.status_code == 200, r.text
assert r.json()["status"] == "changed"
assert (await client.post("/v1/users/login", json={
"username": "alice_test", "auth_key": _auth_key(old, "alice_test")})).status_code == 401
assert (await client.post("/v1/users/login", json={
"username": "alice_test", "auth_key": _auth_key(new, "alice_test")})).status_code == 200
@pytest.mark.asyncio
async def test_wrong_current_passphrase_is_refused_and_changes_nothing(client):
old = "the-first-passphrase"
session = await _register(client, "bob_test", old)
r = await client.post("/v1/users/password", json={
"old_auth_key": _auth_key("not the passphrase", "bob_test"),
"new_auth_key": _auth_key("some-new-passphrase", "bob_test"),
}, headers={"Authorization": f"Bearer {session['access_token']}"})
assert r.status_code == 403
assert (await client.post("/v1/users/login", json={
"username": "bob_test", "auth_key": _auth_key(old, "bob_test")})).status_code == 200
@pytest.mark.asyncio
async def test_new_must_differ_from_old(client):
old = "the-first-passphrase"
session = await _register(client, "carol_test", old)
r = await client.post("/v1/users/password", json={
"old_auth_key": _auth_key(old, "carol_test"),
"new_auth_key": _auth_key(old, "carol_test"),
}, headers={"Authorization": f"Bearer {session['access_token']}"})
assert r.status_code == 400
@pytest.mark.asyncio
async def test_unauthenticated_call_is_rejected(client):
"""A session is required — the current passphrase alone is not a credential."""
await _register(client, "dave_test", "the-first-passphrase")
r = await client.post("/v1/users/password", json={
"old_auth_key": _auth_key("the-first-passphrase", "dave_test"),
"new_auth_key": _auth_key("a-new-one", "dave_test"),
})
assert r.status_code in (401, 403, 422)
@pytest.mark.asyncio
async def test_other_sessions_are_invalidated_and_the_caller_keeps_one(
client, db_session):
old, new = "the-first-passphrase", "a-second-passphrase-entirely"
first = await _register(client, "erin_test", old)
# A second browser signs in before the change.
second = (await client.post("/v1/users/login", json={
"username": "erin_test", "auth_key": _auth_key(old, "erin_test")})).json()
r = await client.post("/v1/users/password", json={
"old_auth_key": _auth_key(old, "erin_test"),
"new_auth_key": _auth_key(new, "erin_test"),
}, headers={"Authorization": f"Bearer {first['access_token']}"})
assert r.status_code == 200, r.text
# The other browser's refresh token is dead.
stale = await client.post("/v1/users/token/refresh", json={
"refresh_token": second["refresh_token"]})
assert stale.status_code == 401
# The caller was handed a fresh pair that still works.
fresh = await client.post("/v1/users/token/refresh", json={
"refresh_token": r.json()["refresh_token"]})
assert fresh.status_code == 200, fresh.text
uid = (await db_session.execute(
select(User.id).where(User.username == "erin_test"))).scalar_one()
live = (await db_session.execute(
select(RefreshToken).where(RefreshToken.user_id == uid,
RefreshToken.revoked.is_(False)))).scalars().all()
# Only the family issued to the caller (the refresh above rotated it once).
assert len(live) == 1
@pytest.mark.asyncio
async def test_the_change_is_logged(client, db_session):
old, new = "the-first-passphrase", "a-second-passphrase-entirely"
session = await _register(client, "frank_test", old)
await client.post("/v1/users/password", json={
"old_auth_key": _auth_key(old, "frank_test"),
"new_auth_key": _auth_key(new, "frank_test"),
}, headers={"Authorization": f"Bearer {session['access_token']}"})
uid = (await db_session.execute(
select(User.id).where(User.username == "frank_test"))).scalar_one()
events = {e.event for e in (await db_session.execute(
select(IPLog).where(IPLog.user_id == uid))).scalars().all()}
assert "password_change" in events
|