From a55d40b74bda77dff6ec565abdd551607fc665d6 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 30 Sep 2026 13:55:59 +0200 Subject: feat(hub): a bundle pepper per account, handed only to a proven session Sealed at rest and bound to the account; returned by sign-in, device sign-in, a passphrase change and GET /me/bundle-pepper, never by a refresh, to a node token, in a token or in a log. Erasure clears it. Co-Authored-By: Claude Opus 5.5 --- packages/meshbay-hub/tests/test_bundle_pepper.py | 181 +++++++++++++++++++++++ 1 file changed, 181 insertions(+) create mode 100644 packages/meshbay-hub/tests/test_bundle_pepper.py (limited to 'packages/meshbay-hub/tests') diff --git a/packages/meshbay-hub/tests/test_bundle_pepper.py b/packages/meshbay-hub/tests/test_bundle_pepper.py new file mode 100644 index 0000000..e7b126f --- /dev/null +++ b/packages/meshbay-hub/tests/test_bundle_pepper.py @@ -0,0 +1,181 @@ +""" +The bundle pepper: half of what opens an account's keypair bundles on nodes. + +A bundle sealed under the passphrase alone lets the operator holding it test +guesses offline, and a right guess is the whole account. Sealed under the +passphrase and a pepper only the hub holds, it does not. So what matters here is +who gets the pepper: a session that proved the passphrase or a device key, and +nobody else — not a refresh, not a node, not a token, not a log. +""" + +import base64 +import logging +import time + +import jwt as _jwt +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import pk_to_b64 +from meshbay_hub.auth import open_pepper +from meshbay_hub.db.models import User +from sqlalchemy import select + +KEY = "k" * 44 + + +async def _register(client, username): + r = await client.post("/v1/users/register", json={ + "username": username, "auth_key": KEY, "email": f"{username}@example.invalid"}) + assert r.status_code == 201, r.text + + +async def _login(client, username, key=KEY): + r = await client.post("/v1/users/login", json={"username": username, "auth_key": key}) + assert r.status_code == 200, r.text + return r.json() + + +def _bearer(token): + return {"Authorization": f"Bearer {token}"} + + +@pytest.mark.asyncio +async def test_sign_in_hands_over_one_stable_pepper(client): + await _register(client, "pepper_user") + first = await _login(client, "pepper_user") + second = await _login(client, "pepper_user") + + assert len(base64.b64decode(first["bundle_pepper"])) == 32 + assert first["bundle_pepper_version"] == 1 + assert second["bundle_pepper"] == first["bundle_pepper"], \ + "a pepper that changed between sign-ins would lock every bundle out" + + +@pytest.mark.asyncio +async def test_a_device_sign_in_gets_it_too(client): + await _register(client, "pepper_dev") + login = await _login(client, "pepper_dev") + sk = Ed25519PrivateKey.generate() + r = await client.post("/v1/users/devices", headers=_bearer(login["access_token"]), + json={"pk_auth_ed25519": pk_to_b64(sk.public_key()), "label": ""}) + assert r.status_code == 201 + ts = int(time.time()) + sig = sk.sign(f"meshbay:user_auth:pepper_dev:{ts}".encode()) + r = await client.post("/v1/users/auth", json={ + "username": "pepper_dev", "timestamp": ts, + "signature": base64.b64encode(sig).decode()}) + assert r.status_code == 200, r.text + assert r.json()["bundle_pepper"] == login["bundle_pepper"] + + +@pytest.mark.asyncio +async def test_a_refresh_does_not(client): + """A refresh token proves only that it was not spent yet.""" + await _register(client, "pepper_ref") + login = await _login(client, "pepper_ref") + r = await client.post("/v1/users/token/refresh", + json={"refresh_token": login["refresh_token"]}) + assert r.status_code == 200, r.text + assert "bundle_pepper" not in r.json() + + +@pytest.mark.asyncio +async def test_it_is_in_no_token(client): + await _register(client, "pepper_jwt") + login = await _login(client, "pepper_jwt") + claims = _jwt.decode(login["access_token"], options={"verify_signature": False}) + assert login["bundle_pepper"] not in repr(claims) + assert not any("pepper" in k for k in claims) + + +@pytest.mark.asyncio +async def test_an_open_session_may_ask_and_a_node_may_not(client): + from test_node_scope_not_admin import _node_token + await _register(client, "pepper_node") + login = await _login(client, "pepper_node") + + r = await client.get("/v1/users/me/bundle-pepper", headers=_bearer(login["access_token"])) + assert r.status_code == 200 + assert r.json()["bundle_pepper"] == login["bundle_pepper"] + + node = await _node_token(client, "pepper_node", login["access_token"]) + r = await client.get("/v1/users/me/bundle-pepper", headers=_bearer(node)) + assert r.status_code == 403 + assert login["bundle_pepper"] not in r.text + + +@pytest.mark.asyncio +async def test_a_passphrase_change_carries_it(client): + """The new passphrase makes a new bundle key, and the client does not keep + the pepper to re-seal under it.""" + await _register(client, "pepper_chg") + login = await _login(client, "pepper_chg") + r = await client.post("/v1/users/password", headers=_bearer(login["access_token"]), + json={"old_auth_key": KEY, "new_auth_key": "n" * 44}) + assert r.status_code == 200, r.text + assert r.json()["bundle_pepper"] == login["bundle_pepper"] + + +@pytest.mark.asyncio +async def test_it_is_sealed_at_rest_and_bound_to_its_account(client, db_session): + await _register(client, "pepper_rest") + login = await _login(client, "pepper_rest") + user = (await db_session.execute( + select(User).where(User.username == "pepper_rest"))).scalar_one() + raw = base64.b64decode(login["bundle_pepper"]) + assert raw not in base64.b64decode(user.bundle_pepper) + assert open_pepper(user.bundle_pepper, user.id) == raw + with pytest.raises(Exception): + open_pepper(user.bundle_pepper, "another-account-id") + + +@pytest.mark.asyncio +async def test_erasure_takes_it_and_the_name_gets_a_new_one(client, db_session): + await _register(client, "pepper_gone") + login = await _login(client, "pepper_gone") + r = await client.request("DELETE", "/v1/users/me", headers=_bearer(login["access_token"]), + json={"auth_key": KEY}) + assert r.status_code == 200, r.text + gone = (await db_session.execute( + select(User).where(User.status == "deleted"))).scalar_one() + await db_session.refresh(gone) + assert gone.bundle_pepper is None + + await _register(client, "pepper_gone") + assert (await _login(client, "pepper_gone"))["bundle_pepper"] != login["bundle_pepper"] + + +@pytest.mark.asyncio +async def test_it_is_never_logged(client): + """Levels and `disabled` pinned here: an earlier test may have changed + either, and a handler that sees nothing proves nothing.""" + seen: list[str] = [] + + class Grab(logging.Handler): + def emit(self, record): + seen.append(record.getMessage()) + + root = logging.getLogger() + grab = Grab(level=logging.DEBUG) + saved = root.level, logging.root.manager.disable + root.addHandler(grab) + root.setLevel(logging.DEBUG) + logging.disable(logging.NOTSET) + loggers = [logging.getLogger(n) for n in ("meshbay_hub", "meshbay_hub.api.users")] + was = [(lg.level, lg.disabled) for lg in loggers] + for lg in loggers: + lg.setLevel(logging.DEBUG) + lg.disabled = False + try: + await _register(client, "pepper_log") + login = await _login(client, "pepper_log") + await client.get("/v1/users/me/bundle-pepper", headers=_bearer(login["access_token"])) + finally: + root.removeHandler(grab) + root.setLevel(saved[0]) + logging.disable(saved[1]) + for lg, (level, disabled) in zip(loggers, was): + lg.setLevel(level) + lg.disabled = disabled + assert seen, "the handler saw nothing, so it proves nothing" + assert not any(login["bundle_pepper"] in m for m in seen) -- cgit v1.2.3