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

"""
A `tr` on a chunk request names a lease, or it names nothing.

The transfer system counts leases, queues them per member and per node, and
tells a client "waiting, 2 ahead" — and until this was written, none of it
decided whether a single byte was served. `_do_file_request` read `tr` as a
*boolean*: present meant "this is a leased transfer, skip the leaseless
ceiling", and nothing asked the node whether it had ever granted such a lease.
`slots.touch(tr)` was called beside it and its answer — `False` if it is not
granted — was thrown away.

So any non-empty string in `tr` bought the whole library with no ceiling of any
kind: not the per-member cap, not the node-wide cap, not the leaseless bound
that exists precisely to bound a client which claims to be browsing. The queue
was advisory even for an honest client, which waits only because its own code
waits (`transfers.js` awaits `lease.acquire()` before it reads a byte).

The rule now, and the reason each case is what it is:

  - **granted** — a live lease of *this* session. Served, and the lease is
    touched so the sweeper does not reclaim a transfer that is plainly moving.
  - **queued** — the node has the lease and has not granted it. Refused with a
    code, because a member reading while queued is the cap not applying.
  - **unknown** — bounded by the leaseless ceiling rather than refused. That is
    what a reconnect looks like from here (leases die with the connection and
    the client re-opens them), and it is the residual §5.5 already states: a
    client that lies gets that bound's worth of files at a time, not the group.

The `_do_transfer_close` handler next door had always checked that a `tr`
belonged to the session using it, with the comment that closing somebody
else's transfer would be a denial of service one random id away. The path that
*serves* the bytes never asked at all.
"""

import asyncio
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_common.protocol import MNP
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.transfers import DOWNLOAD, LeaselessReads, TransferSlots
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

pytestmark = pytest.mark.asyncio

GROUP = "g" * 32


class _Channel:
    """A DataChannel that is open and never full."""
    readyState = "open"
    bufferedAmount = 0


class _Session(WebRTCPeerSession):
    """A session with the channel replaced by a list, and a real everything else."""

    def __init__(self, ctx, *, key="s1", user="alice"):
        self._ctx = ctx
        self._registry_key = key
        self._user_id = user
        self._username = user
        self._group_id = GROUP
        self._channel = _Channel()
        self._leaseless = LeaselessReads()
        self._unleased_noted = False
        self.sent: list[dict] = []

    def _send(self, msg):
        self.sent.append(msg)

    def _audit(self, event, detail=""):
        pass

    def _spawn(self, coro):
        coro.close()
        return None

    def chunks(self):
        return [m for m in self.sent if m.get("type") == MNP.FILE_CHUNK]

    def errors(self):
        return [m for m in self.sent if m.get("type") == "error"]

    def last(self, mtype=MNP.TRANSFER_STATE):
        return next(m for m in reversed(self.sent) if m.get("type") == mtype)


@pytest.fixture
async def group(tmp_path):
    """A group with one real file in it, indexed, and a transport context."""
    shared = tmp_path / "shared"
    shared.mkdir()
    (shared / "one.bin").write_bytes(b"A" * 2_000_000)
    (shared / "two.bin").write_bytes(b"B" * 2_000_000)
    indexer = DirectoryIndexer(
        roots=one_root(shared), group_id=GROUP,
        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
    await indexer.initial_scan()

    ctx = {
        "_peers": {},
        "roots": one_root(shared),
        "index": indexer.index,
        "sk_node": indexer.sk_node,
        "gek": indexer.gek,
    }
    yield ctx
    task = ctx.get("_transfer_sweeper")
    if task is not None:
        task.cancel()


def _ids(ctx) -> list[str]:
    return [e.id for e in ctx["index"].entries]


async def _request(session, file_id, *, tr=""):
    msg = {"type": MNP.FILE_REQUEST, "file_id": file_id, "chunk_index": 0}
    if tr:
        msg["tr"] = tr
    await session._do_file_request(msg)


# ── The three cases ──────────────────────────────────────────────────────────

async def test_a_granted_lease_is_served(group):
    peer = _Session(group)
    group["_peers"]["s1"] = peer
    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
    assert peer.last()["state"] == "granted"

    await _request(peer, _ids(group)[0], tr="t1")
    assert peer.chunks(), peer.errors()


async def test_a_granted_lease_does_not_spend_the_leaseless_budget(group):
    """A download is not a preview, and must not consume what previews use."""
    peer = _Session(group)
    group["_peers"]["s1"] = peer
    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
    await _request(peer, _ids(group)[0], tr="t1")
    assert peer._leaseless.in_flight() == 0


async def test_a_queued_lease_is_refused(group):
    """The cap applies, or it is decoration.

    The node's own answer said "waiting"; the bytes went out anyway, and the
    only thing that made an honest client wait was the honest client.
    """
    slots = TransferSlots()
    slots.caps[DOWNLOAD] = 1
    group["_transfer_slots"] = slots

    first = _Session(group, key="s1", user="alice")
    second = _Session(group, key="s2", user="bob")
    group["_peers"].update({"s1": first, "s2": second})

    first._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
    second._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000})
    assert first.last()["state"] == "granted"
    assert second.last()["state"] == "queued"

    await _request(second, _ids(group)[0], tr="t2")
    assert not second.chunks(), "a queued transfer was served anyway"
    assert second.errors()[-1]["code"] == "lease_not_granted"


async def test_an_invented_transfer_id_is_bounded_like_a_preview(group):
    """Not refused — bounded. A `tr` the node has no record of is also what a
    reconnect looks like, so the answer is the leaseless ceiling rather than a
    failure an honest client cannot recover from."""
    peer = _Session(group)
    group["_peers"]["s1"] = peer

    await _request(peer, _ids(group)[0], tr="not-a-lease-at-all")
    assert peer.chunks(), peer.errors()
    assert peer._leaseless.in_flight() == 1, (
        "an unknown lease id bypassed the ceiling, which is the whole defect")


async def test_another_sessions_lease_is_not_this_sessions_lease(group):
    """A lease is scoped to a connection (§5.5). Borrowing one would be a slot
    nobody granted this session, and touching it would refresh somebody else's
    idle timer."""
    owner = _Session(group, key="s1", user="alice")
    borrower = _Session(group, key="s2", user="bob")
    group["_peers"].update({"s1": owner, "s2": borrower})
    owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
    assert owner.last()["state"] == "granted"

    await _request(borrower, _ids(group)[0], tr="t1")
    assert borrower._leaseless.in_flight() == 1, (
        "another session's lease was accepted as this one's")


async def test_a_bulk_read_under_invented_ids_hits_the_ceiling(group):
    """The residual, measured rather than asserted: a lying client gets the
    leaseless bound's worth of files at a time, and then it is refused."""
    peer = _Session(group)
    group["_peers"]["s1"] = peer
    # A limit this test sets itself — the shipped number moves with the client.
    peer._leaseless.limit = 1

    first, second = _ids(group)[:2]
    await _request(peer, first, tr="invented-1")
    await _request(peer, second, tr="invented-2")

    assert len(peer.chunks()) == 1
    assert peer.errors()[-1]["code"] == "transfer_required"


# ── The upload side of the same seam ─────────────────────────────────────────

async def test_a_queued_upload_is_refused_before_anything_is_written(group,
                                                                    tmp_path):
    """There is no leaseless fallback for a write: a write is never browsing."""
    from conftest import sealed_upload

    slots = TransferSlots()
    slots.caps["upload"] = 1
    group["_transfer_slots"] = slots

    first = _Session(group, key="s1", user="alice")
    second = _Session(group, key="s2", user="bob")
    group["_peers"].update({"s1": first, "s2": second})
    first._do_transfer_open({"tr": "u1", "kind": "upload", "bytes": 10})
    second._do_transfer_open({"tr": "u2", "kind": "upload", "bytes": 10})
    assert second.last()["state"] == "queued"

    msg = sealed_upload(second, filename="sent.bin", data=b"DATA",
                        dir=Path(group["roots"].roots[0].path).name)
    msg["tr"] = "u2"
    await second._do_file_upload(msg)

    assert second.errors()[-1]["code"] == "lease_not_granted"
    assert not list(Path(group["roots"].roots[0].path).glob("sent*")), (
        "a queued upload reached the operator's disk")


async def test_the_unleased_note_is_once_per_connection(group):
    """Visible to the operator, and not a line per chunk — a flood is exactly
    what would produce one."""
    peer = _Session(group)
    group["_peers"]["s1"] = peer
    noted = []
    peer._audit = lambda event, detail="": noted.append(event)

    for chunk in range(3):
        await peer._do_file_request({
            "type": MNP.FILE_REQUEST, "file_id": _ids(group)[0],
            "chunk_index": chunk, "tr": "invented"})

    assert noted.count("transfer_unleased") == 1, noted


# Kept last: it is about the pool, not the seam, and it would otherwise be the
# first thing to fail if `TransferSlots` moved.
async def test_the_sweeper_does_not_reclaim_a_transfer_that_is_moving(group):
    """`_lease_of` still touches a granted lease, which is what it was doing
    before any of this — losing that would reclaim live transfers every 30 s."""
    peer = _Session(group)
    group["_peers"]["s1"] = peer
    peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000})
    slots = peer._slots()
    assert slots.leases["t1"].used is False

    await _request(peer, _ids(group)[0], tr="t1")
    assert slots.leases["t1"].used is True
    await asyncio.sleep(0)