""" 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_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)