diff options
Diffstat (limited to 'packages')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/api/users.py | 24 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/auth.py | 37 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_argon2_off_loop.py | 91 |
3 files changed, 140 insertions, 12 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py index 2a6baf0..05f58cd 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/users.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py @@ -24,11 +24,11 @@ from meshbay_hub.auth import ( encrypt_email, generate_refresh_token, hash_email_blind, - hash_password, + hash_password_off_loop, hash_refresh_token, issue_access_token, pw_needs_rehash, - verify_password, + verify_password_off_loop, ) from meshbay_hub.config import HubConfig from meshbay_hub.db.engine import get_db @@ -192,7 +192,7 @@ async def register( if not credential: raise HTTPException(status_code=400, detail="auth_key or password required") - pw_hash, pw_salt = hash_password(credential) + pw_hash, pw_salt = await hash_password_off_loop(credential) pw_ver = current_pw_version() if body.auth_key else 2 hub_id = _cfg.identity.id if _cfg else "meshbay.org" user = User( @@ -346,7 +346,7 @@ async def login( if user.pw_version >= 3: # New scheme: verify auth_key - if not body.auth_key or not verify_password( + if not body.auth_key or not await verify_password_off_loop( body.auth_key, user.pw_hash, user.pw_salt, version=user.pw_version ): await _login_failed(db, body.username, ip, user.id) @@ -357,19 +357,19 @@ async def login( await login_throttle.release(db, body.username) await db.commit() raise HTTPException(status_code=401, detail="auth_upgrade_required") - if not verify_password( + if not await verify_password_off_loop( body.password, user.pw_hash, user.pw_salt, version=user.pw_version ): await _login_failed(db, body.username, ip, user.id) # Migrate to new scheme if auth_key provided alongside password if body.auth_key: - new_hash, new_salt = hash_password(body.auth_key) + new_hash, new_salt = await hash_password_off_loop(body.auth_key) user.pw_hash = new_hash user.pw_salt = new_salt user.pw_version = current_pw_version() elif user.pw_version < 2: # Legacy rehash: upgrade Argon2 params within the password scheme (v1 -> v2) - new_hash, new_salt = hash_password(body.password) + new_hash, new_salt = await hash_password_off_loop(body.password) user.pw_hash = new_hash user.pw_salt = new_salt user.pw_version = 2 @@ -386,7 +386,7 @@ async def login( # Rehash within the auth_key scheme if Argon2 params upgraded beyond v3 if user.pw_version >= 3 and pw_needs_rehash(user.pw_version): - new_hash, new_salt = hash_password(body.auth_key) + new_hash, new_salt = await hash_password_off_loop(body.auth_key) user.pw_hash = new_hash user.pw_salt = new_salt user.pw_version = current_pw_version() @@ -885,7 +885,7 @@ async def change_password( db: AsyncSession = Depends(get_db), ): await _take_login_attempt(db, current_user.username) - if not verify_password(body.old_auth_key, current_user.pw_hash, + if not await verify_password_off_loop(body.old_auth_key, current_user.pw_hash, current_user.pw_salt, current_user.pw_version): raise HTTPException(status_code=403, detail="Current passphrase does not match") @@ -894,7 +894,7 @@ async def change_password( raise HTTPException(status_code=400, detail="New passphrase must differ from the current one") - new_hash, new_salt = hash_password(body.new_auth_key) + new_hash, new_salt = await hash_password_off_loop(body.new_auth_key) current_user.pw_hash = new_hash current_user.pw_salt = new_salt current_user.pw_version = current_pw_version() @@ -1077,7 +1077,7 @@ async def password_reset( raise HTTPException(status_code=400, detail="Invalid code") verif.verified_at = now - new_hash, new_salt = hash_password(body.new_auth_key) + new_hash, new_salt = await hash_password_off_loop(body.new_auth_key) user.pw_hash = new_hash user.pw_salt = new_salt user.pw_version = current_pw_version() @@ -1330,7 +1330,7 @@ async def delete_own_account( still never sees the passphrase itself. """ await _take_login_attempt(db, current_user.username) - if not verify_password(body.auth_key, current_user.pw_hash, current_user.pw_salt, + if not await verify_password_off_loop(body.auth_key, current_user.pw_hash, current_user.pw_salt, current_user.pw_version): raise HTTPException(status_code=403, detail="Passphrase does not match") await login_throttle.clear(db, current_user.username) diff --git a/packages/meshbay-hub/src/meshbay_hub/auth.py b/packages/meshbay-hub/src/meshbay_hub/auth.py index c5ea34d..58e2310 100644 --- a/packages/meshbay-hub/src/meshbay_hub/auth.py +++ b/packages/meshbay-hub/src/meshbay_hub/auth.py @@ -7,11 +7,13 @@ MeshBay Hub — authentication helpers. - Hub keypair: loaded from PEM file on startup """ +import asyncio import base64 import hashlib import os import time import uuid +from concurrent.futures import ThreadPoolExecutor from pathlib import Path import blake3 @@ -139,6 +141,41 @@ def verify_password(password: str, pw_hash: bytes, salt: bytes, version: int = 2 return False +# ── Argon2 off the event loop, one at a time ───────────────────────────────── +# +# One derivation is 256 MB and a quarter to half a second of CPU. Called from an +# async handler it stops the whole hub for that long — every request, every node +# socket, every offer relayed — once per sign-in, passphrase change, reset and +# registration. +# +# It cannot simply go to a thread pool: **two concurrent derivations with +# `lanes=4` deadlock inside OpenSSL** and never return, at no CPU (measured +# 2026-09-14 on cryptography 50.0.x / OpenSSL 4.0.x, here and on meshbay.org; +# `lanes=1` does not, and `lanes` is part of every stored hash, so it is not +# ours to change). Inline on the loop they could never overlap, which is the only +# reason this never hung in production. +# +# So: a dedicated executor with exactly one worker. The loop is free while a +# derivation runs, and derivations still never overlap — including when the +# request that queued one is cancelled mid-way, which a semaphore around +# `to_thread` would get wrong (the permit is released while the thread is still +# deriving, and the next one starts beside it). It also bounds Argon2's memory +# to one derivation, whatever the number of callers. +_argon2_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="argon2") + + +async def hash_password_off_loop(password: str) -> tuple[bytes, bytes]: + loop = asyncio.get_running_loop() + return await loop.run_in_executor(_argon2_executor, hash_password, password) + + +async def verify_password_off_loop(password: str, pw_hash: bytes, salt: bytes, + version: int = 2) -> bool: + loop = asyncio.get_running_loop() + return await loop.run_in_executor( + _argon2_executor, verify_password, password, pw_hash, salt, version) + + def pw_needs_rehash(version: int) -> bool: return version < _ARGON2_CURRENT_VERSION diff --git a/packages/meshbay-hub/tests/test_argon2_off_loop.py b/packages/meshbay-hub/tests/test_argon2_off_loop.py new file mode 100644 index 0000000..b156ebc --- /dev/null +++ b/packages/meshbay-hub/tests/test_argon2_off_loop.py @@ -0,0 +1,91 @@ +""" +Argon2 runs off the event loop, and never two at a time. + +One derivation is 256 MB and a quarter to half a second of CPU. 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 (`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(): + pw_hash, salt = auth.hash_password("k" * 44) + started = time.perf_counter() + auth.verify_password("k" * 44, pw_hash, salt, auth.current_pw_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, auth.current_pw_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 |