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