""" 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": "", "auth_key": KEY}) 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) async def _ask(client, token, auth_key=None): body = {} if auth_key is None else {"auth_key": auth_key} return await client.post("/v1/users/me/bundle-pepper", headers=_bearer(token), json=body) @pytest.mark.asyncio async def test_a_session_asks_with_the_passphrase_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 _ask(client, login["access_token"], KEY) 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 _ask(client, node, KEY) assert r.status_code == 403 assert login["bundle_pepper"] not in r.text @pytest.mark.asyncio async def test_a_token_alone_never_gets_it(client): """ Not a refreshed one, not the one sign-in handed out: a token is what a page holds, and a pepper beside the bundles an operator keeps is the offline passphrase oracle again. Two calls — refresh, then ask — must not add up to what a refresh is refused. """ await _register(client, "pepper_bare") login = await _login(client, "pepper_bare") refreshed = (await client.post("/v1/users/token/refresh", json={"refresh_token": login["refresh_token"]})).json() for token in (login["access_token"], refreshed["access_token"]): bare = await _ask(client, token) wrong = await _ask(client, token, "w" * 44) assert bare.status_code == 422 assert wrong.status_code == 403 assert login["bundle_pepper"] not in bare.text + wrong.text assert (await client.get("/v1/users/me/bundle-pepper", headers=_bearer(login["access_token"]))).status_code in (404, 405) @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 _ask(client, login["access_token"], KEY) 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) @pytest.mark.asyncio async def test_the_browser_access_mirror_is_a_preference_like_any_other(client): """The desktop application writes what it holds, and a browser reads it to say why it cannot open a group. Nothing on the hub or a node acts on it.""" await _register(client, "pepper_mirror") login = await _login(client, "pepper_mirror") headers = _bearer(login["access_token"]) r = await client.put("/v1/users/me/preferences/browser_access", headers=headers, json={"value": "off"}) assert r.status_code == 200, r.text prefs = (await client.get("/v1/users/me/preferences", headers=headers)).json() assert prefs["browser_access"] == "off" # Still handed the pepper: the mirror decides nothing. assert "bundle_pepper" in await _login(client, "pepper_mirror")