1 files changed, 181 insertions, 0 deletions
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)