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
|
"""
Argon2 runs off the event loop, and never two at a time.
A derivation is 64 MiB and ~0.1 s at the current version, 256 MB and ~0.45 s
for a v3 hash not yet rewritten. On the loop it stopped the whole hub for that
long at every sign-in. In a thread pool it would have been worse: two concurrent
`lanes=4` derivations deadlock inside OpenSSL and never return, at 64 MiB as at
256 MB (`auth._argon2_executor`). These pin both halves.
"""
import asyncio
import base64
import pathlib
import re
import time
import pytest
from meshbay_hub import auth
SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "meshbay_hub"
def test_nothing_derives_argon2_on_the_event_loop():
"""Every line of every module, not the first match: a call added above an
existing one must not hide behind it."""
direct = re.compile(r"(?<![\w.])(hash_password|verify_password)\s*\(")
offenders = []
for path in SRC.rglob("*.py"):
if path.name == "auth.py":
continue
for n, line in enumerate(path.read_text().splitlines(), 1):
if direct.search(line):
offenders.append(f"{path.relative_to(SRC)}:{n}: {line.strip()}")
assert not offenders, (
"these derive Argon2 on the calling thread; use the *_off_loop versions:\n"
+ "\n".join(offenders))
def test_the_executor_has_exactly_one_worker():
"""More than one lets two `lanes=4` derivations overlap, and they deadlock."""
assert auth._argon2_executor._max_workers == 1
@pytest.mark.asyncio
async def test_concurrent_derivations_all_return():
pw_hash, salt = auth.hash_password("k" * 44)
results = await asyncio.wait_for(asyncio.gather(*[
auth.verify_password_off_loop("k" * 44, pw_hash, salt, auth.current_pw_version())
for _ in range(4)]), timeout=30)
assert results == [True] * 4
@pytest.mark.asyncio
async def test_the_loop_keeps_turning_while_argon2_runs():
# A v3 (256 MB) hash, the heaviest the hub still verifies: at 64 MiB an
# inline derivation is ~45 ms and half of it is too close to scheduling
# noise to tell a stalled loop from a busy machine.
version = 3
pw_hash, salt = auth.hash_password("k" * 44, version)
started = time.perf_counter()
auth.verify_password("k" * 44, pw_hash, salt, version)
inline = time.perf_counter() - started
gaps, done = [], asyncio.Event()
async def ticker():
last = time.perf_counter()
while not done.is_set():
await asyncio.sleep(0.005)
now = time.perf_counter()
gaps.append(now - last)
last = now
task = asyncio.create_task(ticker())
await asyncio.sleep(0.02)
assert await auth.verify_password_off_loop("k" * 44, pw_hash, salt, version)
done.set()
await task
# Inline, the loop stalls for the whole derivation; off it, for scheduling noise.
assert max(gaps) < inline / 2, (max(gaps), inline)
@pytest.mark.asyncio
async def test_concurrent_sign_ins_all_complete(client):
key = base64.b64encode(b"k" * 32).decode()
names = [f"concurrent{i}" for i in range(4)]
for name in names:
r = await client.post("/v1/users/register", json={
"username": name, "email": f"{name}@example.test", "auth_key": key})
assert r.status_code == 201, r.text
responses = await asyncio.wait_for(asyncio.gather(*[
client.post("/v1/users/login", json={"username": n, "auth_key": key})
for n in names]), timeout=60)
assert [r.status_code for r in responses] == [200] * 4
|