diff options
| -rw-r--r-- | CLAUDE.md | 10 | ||||
| -rw-r--r-- | docs/MESHBAY_DESIGN.md | 2 | ||||
| -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 |
5 files changed, 150 insertions, 14 deletions
@@ -209,7 +209,15 @@ These are about working on the tree rather than about the design: catches — everything works, slowly, for everyone, whenever the mail server is having a bad day. `mail.send_off_loop` is the door, and `test_no_mail_is_sent_from_the_event_loop` reads the source for direct calls, - because there is nothing else to read + because there is nothing else to read. **Argon2 is the same, with a trap + under it**: one derivation is 256 MB and ~0.25–0.5 s, so it goes through + `auth.hash_password_off_loop` / `verify_password_off_loop` — but **two + concurrent `lanes=4` derivations deadlock inside OpenSSL** and never return, + at no CPU (cryptography 50.0.x, OpenSSL 4.0.x, reproduced locally and on + meshbay.org). Inline on the loop they could never overlap, which hid it. The + executor has **exactly one worker**, not a semaphore around `to_thread`: a + cancelled request releases a permit while its thread is still deriving. + `test_argon2_off_loop.py` holds both halves - **Before adding an endpoint or a message, ask who pays.** A participant supplies input; if anyone other than the sender bears the cost, there is a diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md index 6b1fd05..6615f66 100644 --- a/docs/MESHBAY_DESIGN.md +++ b/docs/MESHBAY_DESIGN.md @@ -2666,7 +2666,7 @@ had already been asked. | **AV6** | **A relay proves possession of its approved key.** A public key is not a password, and the register call is unauthenticated by design — it is not a user — so the proof is the only thing standing between a stranger and where nodes send relayed traffic | | **AV7** | **A node bounds how many peers it holds and how long an unproven one lasts.** The hub's cap is per calling account, which is a limit on each member and not on the machine, so without this an operator's exposure grew with the size of their groups | | **AV8** | **One account cannot make the hub mail another at will.** The invitation email's subject comes from the group row, never from the request, and the endpoint is metered | -| **AV9** | **No mail is sent from the event loop.** `smtplib` is synchronous and waits up to ten seconds; called from an async handler that wait is the whole instance's, not one request's. Every send goes through `mail.send_off_loop` | +| **AV9** | **No mail is sent from the event loop.** `smtplib` is synchronous and waits up to ten seconds; called from an async handler that wait is the whole instance's, not one request's. Every send goes through `mail.send_off_loop`. **Argon2 is held to the same rule**: every derivation runs on one dedicated worker thread (`auth.*_off_loop`), never on the loop and never two at a time, because two concurrent `lanes=4` derivations deadlock in OpenSSL | | **AV10** | **Every path that makes the hub send mail is metered, per account.** A rate limit that counts by IP bounds a caller, not an inbox. Changing one's address mails an arbitrary stranger, so it carries a cooldown *and* a daily ceiling; a reset request and a registration resend carry cooldowns | | **AV13** | **The mail server is not a relay, and `mail.py` is where that is decided.** Every message passes one function; `purpose` is keyword-required and checked against a closed list, so a helper that names anything else does not send and one that names nothing is a TypeError. Under it sit a bound per **recipient** — the thing a person being mail-bombed actually experiences, unmoved by which account, address or endpoint asks — and an instance-wide hourly ceiling, because registration is open and "per account" is a bound an attacker buys more of | | **AV11** | **A namespace a client writes into is closed, and its rows are capped.** The preference key space is an allow-list plus `default_tab:<group_id>` checked as a group id, the value is length-bounded, and the row count per account is bounded | 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 |