path: root/packages/meshbay-node/tests/test_user_blob_mnp.py

"""
`user_blob_*` over MNP (MNP 3.1, docs/playlists.md §8.1).

The store itself is covered by `test_user_blob_store.py`. What is here is
everything the handlers add on top, and each one is a refusal:

  - `user_id` comes from the authenticated session and **never** from the
    message. A `user_id` in the body would let any member of this group read or
    overwrite any other member's blob — finding C5, one size down.
  - `kind` is validated against a pattern, so the table does not become an
    arbitrary key/value store for whatever a client feels like writing.
  - every cap **refuses, with a stated reason**, and never truncates. A
    truncating cap silently loses tracks; a bare `error` reaches the reader as
    a dead button.

The session is built the way `test_admin_ops_mnp.py` builds one — the real
handler methods against a real store, with `_send` captured.
"""

import pytest
from meshbay_common.protocol import MNP
from meshbay_node.bundle_store import BundleStore
from meshbay_node.transport.webrtc_server import (
    USER_BLOB_ACCOUNT_MAX,
    USER_BLOB_BODY_MAX,
    USER_BLOB_MANIFEST_MAX,
    WebRTCPeerSession,
)


@pytest.fixture
async def store(tmp_path):
    s = BundleStore(db_path=tmp_path / "bundles.db")
    await s.open()
    yield s
    await s.close()


def _session(store, user_id="alice"):
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {"bundle_store": store}
    session._user_id = user_id
    session._username = user_id
    session._group_id = "g" * 32
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session.audited = []
    session._audit = lambda event, detail="": session.audited.append((event, detail))
    return session


def _last(session):
    return session.sent[-1] if session.sent else {}


# ── the ordinary path ────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_store_then_fetch_returns_the_same_bytes(store):
    s = _session(store)
    sealed = bytes(range(256))
    await s._do_user_blob_store(
        {"kind": "playlists", "rev": 4, "blob_enc": sealed})
    assert _last(s)["detail"] == "user_blob_stored"

    await s._do_user_blob_fetch({"kind": "playlists"})
    reply = _last(s)
    assert reply["type"] == MNP.USER_BLOB_RESP
    assert reply["kind"] == "playlists"
    assert reply["rev"] == 4
    assert reply["blob_enc"] == sealed


@pytest.mark.asyncio
async def test_fetching_something_never_written_is_null_not_an_error(store):
    """A fresh node has no playlists, and that is not a failure."""
    s = _session(store)
    await s._do_user_blob_fetch({"kind": "playlist:abc"})
    reply = _last(s)
    assert reply["type"] == MNP.USER_BLOB_RESP
    assert reply["rev"] is None and reply["blob_enc"] is None


@pytest.mark.asyncio
async def test_listing_carries_kinds_and_revisions_only(store):
    s = _session(store)
    await s._do_user_blob_store({"kind": "playlists", "rev": 1, "blob_enc": b"m"})
    await s._do_user_blob_store(
        {"kind": "playlist:abc", "rev": 9, "blob_enc": b"body-bytes"})

    await s._do_user_blob_list()
    reply = _last(s)
    assert reply["type"] == MNP.USER_BLOB_LIST_RESP
    assert reply["blobs"] == [{"kind": "playlist:abc", "rev": 9},
                              {"kind": "playlists", "rev": 1}]
    assert "blob_enc" not in str(reply["blobs"])


@pytest.mark.asyncio
async def test_delete_removes_it_and_says_so_when_there_was_nothing(store):
    s = _session(store)
    await s._do_user_blob_store({"kind": "playlist:x", "rev": 1, "blob_enc": b"b"})
    await s._do_user_blob_delete({"kind": "playlist:x"})
    assert _last(s)["detail"] == "user_blob_deleted"
    await s._do_user_blob_delete({"kind": "playlist:x"})
    assert _last(s)["detail"] == "user_blob_absent"


# ── the refusals ─────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_a_user_id_in_the_message_is_ignored(store):
    """The one that matters. Alice writes; Mallory asks for Alice's blob by
    naming her in the body and gets her own empty slot, not Alice's playlist."""
    alice = _session(store, "alice")
    await alice._do_user_blob_store(
        {"kind": "playlists", "rev": 1, "blob_enc": b"alice-private"})

    mallory = _session(store, "mallory")
    await mallory._do_user_blob_fetch({"kind": "playlists", "user_id": "alice"})
    assert _last(mallory)["blob_enc"] is None

    await mallory._do_user_blob_store(
        {"kind": "playlists", "rev": 99, "blob_enc": b"mallory", "user_id": "alice"})
    await alice._do_user_blob_fetch({"kind": "playlists"})
    assert _last(alice)["blob_enc"] == b"alice-private"


@pytest.mark.asyncio
@pytest.mark.parametrize("kind", [
    "", "playlist", "playlists2", "../../etc/passwd", "resume_positions",
    "playlist:", "playlist:../x", "playlist:" + "a" * 65, 42, None,
])
async def test_an_unknown_kind_is_refused(store, kind):
    s = _session(store)
    await s._do_user_blob_store({"kind": kind, "rev": 1, "blob_enc": b"x"})
    assert _last(s)["type"] == "error"
    assert await store.list_user_blobs("alice") == []


@pytest.mark.asyncio
@pytest.mark.parametrize("kind", [
    "playlists", "playlist:favorites", "playlist:abc-123",
    "playlist:0f8fad5b-d9cb-469f-a165-70867728950e",
])
async def test_the_kinds_the_client_actually_writes_are_accepted(store, kind):
    s = _session(store)
    await s._do_user_blob_store({"kind": kind, "rev": 1, "blob_enc": b"x"})
    assert _last(s).get("detail") == "user_blob_stored"


@pytest.mark.asyncio
async def test_an_oversized_body_is_refused_and_nothing_is_written(store):
    """Refused, not truncated: a truncating cap loses tracks silently, which is
    the failure playlists exist to prevent."""
    s = _session(store)
    await s._do_user_blob_store(
        {"kind": "playlist:big", "rev": 1, "blob_enc": b"x" * (USER_BLOB_BODY_MAX + 1)})
    reply = _last(s)
    assert reply["type"] == "error"
    assert "too large" in reply["detail"], "the refusal must say why"
    assert str(USER_BLOB_BODY_MAX) in reply["detail"]
    assert await store.fetch_user_blob("alice", "playlist:big") is None


@pytest.mark.asyncio
async def test_the_manifest_has_a_tighter_cap_than_a_body(store):
    """It holds names and revisions; a manifest the size of a playlist means
    something is writing tracks into the wrong blob."""
    s = _session(store)
    assert USER_BLOB_MANIFEST_MAX < USER_BLOB_BODY_MAX
    await s._do_user_blob_store(
        {"kind": "playlists", "rev": 1,
         "blob_enc": b"x" * (USER_BLOB_MANIFEST_MAX + 1)})
    assert _last(s)["type"] == "error"

    await s._do_user_blob_store(
        {"kind": "playlists", "rev": 1, "blob_enc": b"x" * USER_BLOB_MANIFEST_MAX})
    assert _last(s).get("detail") == "user_blob_stored"


@pytest.mark.asyncio
async def test_the_account_quota_bounds_the_whole_collection(store):
    """Per-blob caps bound one playlist; only this bounds what one account can
    put on somebody else's disk."""
    s = _session(store)
    body = b"x" * USER_BLOB_BODY_MAX
    fits = USER_BLOB_ACCOUNT_MAX // USER_BLOB_BODY_MAX
    for i in range(fits):
        await s._do_user_blob_store(
            {"kind": f"playlist:p{i}", "rev": 1, "blob_enc": body})
        assert _last(s).get("detail") == "user_blob_stored", f"blob {i} refused early"

    await s._do_user_blob_store(
        {"kind": "playlist:over", "rev": 1, "blob_enc": body})
    assert _last(s)["type"] == "error"
    assert "quota" in _last(s)["detail"]


@pytest.mark.asyncio
async def test_replacing_a_blob_at_the_quota_is_not_refused(store):
    """The check subtracts what this write replaces. Without that, an account
    at its limit could never edit a playlist again — only delete one."""
    s = _session(store)
    body = b"x" * USER_BLOB_BODY_MAX
    for i in range(USER_BLOB_ACCOUNT_MAX // USER_BLOB_BODY_MAX):
        await s._do_user_blob_store(
            {"kind": f"playlist:p{i}", "rev": 1, "blob_enc": body})

    await s._do_user_blob_store({"kind": "playlist:p0", "rev": 2, "blob_enc": body})
    assert _last(s).get("detail") == "user_blob_stored"
    assert (await store.fetch_user_blob("alice", "playlist:p0"))["rev"] == 2


@pytest.mark.asyncio
@pytest.mark.parametrize("msg", [
    {"kind": "playlists", "rev": 1},                       # no blob
    {"kind": "playlists", "rev": 1, "blob_enc": b""},      # empty blob
    {"kind": "playlists", "rev": 1, "blob_enc": "a string"},
    {"kind": "playlists", "blob_enc": b"x"},               # no rev
    {"kind": "playlists", "rev": "4", "blob_enc": b"x"},   # rev not a number
    {"kind": "playlists", "rev": -1, "blob_enc": b"x"},
])
async def test_a_malformed_store_is_refused(store, msg):
    s = _session(store)
    await s._do_user_blob_store(msg)
    assert _last(s)["type"] == "error"
    assert await store.list_user_blobs("alice") == []


@pytest.mark.asyncio
async def test_an_unauthenticated_session_reaches_nothing(store):
    """`_user_id` is set by the handshake; before it, there is no account to
    read or write and the handler must not invent one."""
    s = _session(store, user_id=None)
    await s._do_user_blob_fetch({"kind": "playlists"})
    assert _last(s)["type"] == "error"
    await s._do_user_blob_store({"kind": "playlists", "rev": 1, "blob_enc": b"x"})
    assert _last(s)["type"] == "error"
    assert await store.list_user_blobs(None) == []


@pytest.mark.asyncio
@pytest.mark.parametrize("msg, why", [
    ({"kind": "nonsense", "rev": 1, "blob_enc": b"x"}, "Unknown blob kind"),
    ({"kind": "playlists", "rev": 1}, "Missing blob_enc"),
    ({"kind": "playlists", "blob_enc": b"x"}, "Missing rev"),
    ({"kind": "playlists", "rev": 1,
      "blob_enc": b"x" * (USER_BLOB_MANIFEST_MAX + 1)}, "too large"),
])
async def test_a_refusal_is_audited_and_not_merely_sent(store, msg, why):
    """A refusal used to leave no trace at all: the audit line was written only
    after a store *succeeded*. So a client whose every write was being turned
    away looked exactly like a client that never wrote — which is how a wedged
    sync went unnoticed, with the operator's own log saying nothing."""
    s = _session(store)
    await s._do_user_blob_store(msg)
    assert _last(s)["type"] == "error"
    events = [(e, d) for e, d in s.audited if e == "user_blob_refused"]
    assert events, "the refusal is invisible in the audit log"
    assert why in events[0][1]


@pytest.mark.asyncio
async def test_a_listing_is_audited(store):
    """Half the traffic was invisible: `user_blob_list` is what every sync does
    first, and it wrote no audit line, so the log could not distinguish a client
    that was syncing from one that was not."""
    s = _session(store)
    await s._do_user_blob_list()
    assert ("user_blob_list", "0 blobs") in s.audited


@pytest.mark.asyncio
async def test_reads_and_writes_are_audited(store):
    """The node logs that a blob moved, never what was in it — the same line
    `keypair_bundle_store` already writes."""
    s = _session(store)
    await s._do_user_blob_store(
        {"kind": "playlist:abc", "rev": 2, "blob_enc": b"sealed"})
    await s._do_user_blob_fetch({"kind": "playlist:abc"})
    await s._do_user_blob_delete({"kind": "playlist:abc"})

    events = [e for e, _ in s.audited]
    assert events == ["user_blob_store", "user_blob_fetch", "user_blob_delete"]
    assert "sealed" not in " ".join(d for _, d in s.audited)