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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
|
"""
Passphrase reset by e-mail code — Flow B of docs/MESHBAY_DESIGN.md §3.6.
The hub's part re-opens sign-in only: it swaps the auth_key verifier, kills
every session, and drops every registered device key so a stored one cannot
sign back in past the reset. Restoring group access is the client's job with
the recovery key and is not exercised here.
"""
import base64
import time
from datetime import UTC, datetime, timedelta
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import pk_to_b64
from meshbay_hub.db.models import EmailVerification, IPLog, User
from sqlalchemy import select
def _email(username: str) -> str:
return f"{username}@example.com"
async def _register(client, username, auth_key="k" * 44):
r = await client.post("/v1/users/register", json={
"username": username, "email": _email(username),
"auth_key": auth_key})
assert r.status_code in (200, 201), r.text
async def _request_reset(client, username, email=None):
return await client.post("/v1/users/password/reset-request", json={
"username": username, "email": email or _email(username)})
async def _reset_code(db_session, username) -> str:
uid = (await db_session.execute(
select(User.id).where(User.username == username))).scalar_one()
row = (await db_session.execute(
select(EmailVerification).where(
EmailVerification.user_id == uid,
EmailVerification.purpose == "password_reset",
EmailVerification.verified_at.is_(None),
).order_by(EmailVerification.created_at.desc()))).scalars().first()
return row.code if row else None
@pytest.mark.asyncio
async def test_reset_lets_the_user_sign_in_with_a_new_passphrase(client, db_session):
await _register(client, "alice_test", "old" + "a" * 41)
r = await _request_reset(client, "alice_test")
assert r.status_code == 200 and r.json()["status"] == "sent_if_exists"
code = await _reset_code(db_session, "alice_test")
assert code
new = "new" + "b" * 41
r = await client.post("/v1/users/password/reset", json={
"username": "alice_test", "code": code, "new_auth_key": new})
assert r.status_code == 200, r.text
assert (await client.post("/v1/users/login", json={
"username": "alice_test", "auth_key": "old" + "a" * 41})).status_code == 401
assert (await client.post("/v1/users/login", json={
"username": "alice_test", "auth_key": new})).status_code == 200
@pytest.mark.asyncio
async def test_reset_request_never_reveals_whether_an_account_exists(
client, db_session):
r = await _request_reset(client, "ghost")
assert r.status_code == 200
assert r.json()["status"] == "sent_if_exists"
rows = (await db_session.execute(select(EmailVerification))).scalars().all()
assert rows == []
@pytest.mark.asyncio
async def test_reset_request_needs_the_username_and_email_to_match(client, db_session):
await _register(client, "hank_test")
# Right username, wrong e-mail — answered exactly like an unknown account,
# and no code is created.
r = await _request_reset(client, "hank_test", email="someone.else@example.com")
assert r.status_code == 200
assert r.json()["status"] == "sent_if_exists"
assert (await db_session.execute(
select(EmailVerification))).scalars().all() == []
# The real pair does create one.
await _request_reset(client, "hank_test")
assert (await db_session.execute(
select(EmailVerification))).scalars().first() is not None
@pytest.mark.asyncio
async def test_reset_request_rejects_a_malformed_email(client):
await _register(client, "iris_test")
r = await client.post("/v1/users/password/reset-request", json={
"username": "iris_test", "email": "not-an-email"})
assert r.status_code == 422
@pytest.mark.asyncio
async def test_a_wrong_code_is_rejected_and_counts_against_the_limit(
client, db_session):
await _register(client, "bob_test")
await _request_reset(client, "bob_test")
for _ in range(10):
r = await client.post("/v1/users/password/reset", json={
"username": "bob_test", "code": "000000", "new_auth_key": "x" * 44})
assert r.status_code == 400
r = await client.post("/v1/users/password/reset", json={
"username": "bob_test", "code": "000000", "new_auth_key": "x" * 44})
assert r.status_code == 429
@pytest.mark.asyncio
async def test_an_expired_code_is_refused(client, db_session):
await _register(client, "carol_test")
await _request_reset(client, "carol_test")
uid = (await db_session.execute(
select(User.id).where(User.username == "carol_test"))).scalar_one()
row = (await db_session.execute(select(EmailVerification).where(
EmailVerification.user_id == uid))).scalars().one()
row.expires_at = datetime.now(UTC) - timedelta(minutes=1)
await db_session.commit()
r = await client.post("/v1/users/password/reset", json={
"username": "carol_test", "code": row.code, "new_auth_key": "y" * 44})
assert r.status_code == 410
@pytest.mark.asyncio
async def test_a_reset_code_works_once(client, db_session):
await _register(client, "dave_test")
await _request_reset(client, "dave_test")
code = await _reset_code(db_session, "dave_test")
first = await client.post("/v1/users/password/reset", json={
"username": "dave_test", "code": code, "new_auth_key": "z" * 44})
assert first.status_code == 200
second = await client.post("/v1/users/password/reset", json={
"username": "dave_test", "code": code, "new_auth_key": "z" * 44})
assert second.status_code == 404
@pytest.mark.asyncio
async def test_reset_revokes_sessions_and_wipes_devices(client, db_session):
await _register(client, "erin_test", "erin_test" + "a" * 40)
login = await client.post("/v1/users/login", json={
"username": "erin_test", "auth_key": "erin_test" + "a" * 40})
refresh_token = login.json()["refresh_token"]
token = login.json()["access_token"]
sk = Ed25519PrivateKey.generate()
dev = await client.post(
"/v1/users/devices",
json={"pk_auth_ed25519": pk_to_b64(sk.public_key()), "label": "laptop"},
headers={"Authorization": f"Bearer {token}"})
assert dev.status_code == 201, dev.text
await _request_reset(client, "erin_test")
code = await _reset_code(db_session, "erin_test")
r = await client.post("/v1/users/password/reset", json={
"username": "erin_test", "code": code, "new_auth_key": "erin-new" + "b" * 36})
assert r.status_code == 200
# Old refresh token is dead.
assert (await client.post("/v1/users/token/refresh", json={
"refresh_token": refresh_token})).status_code == 401
# Every device key is gone; the stored one can no longer sign in.
uid = (await db_session.execute(
select(User.id).where(User.username == "erin_test"))).scalar_one()
from meshbay_hub.db.models import UserDevice
devices = (await db_session.execute(
select(UserDevice).where(UserDevice.user_id == uid))).scalars().all()
assert devices == []
ts = int(time.time())
msg = f"meshbay:user_auth:erin:{ts}".encode()
da = await client.post("/v1/users/auth", json={
"username": "erin_test", "timestamp": ts,
"signature": base64.b64encode(sk.sign(msg)).decode()})
assert da.status_code == 401
@pytest.mark.asyncio
async def test_the_request_and_the_reset_are_logged(client, db_session):
await _register(client, "frank_test")
await _request_reset(client, "frank_test")
code = await _reset_code(db_session, "frank_test")
await client.post("/v1/users/password/reset", json={
"username": "frank_test", "code": code, "new_auth_key": "f" * 44})
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_reset_request", "password_reset"} <= events
|