fix(node): a transfer id names a lease, or it names nothing

`_do_file_request` read `tr` as a boolean. Present meant "this is a leased
transfer, skip the leaseless ceiling", and nothing asked whether this node had
ever granted such a lease — `slots.touch(tr)` was called beside it and its
answer, `False` if it is not granted, was discarded. So any non-empty string
bought the whole library with no ceiling of any kind: not the per-member cap,
not the node-wide one, not the leaseless bound that exists to bound a client
claiming to be browsing. The queue held only the clients that chose to wait.

`_lease_of` decides it now, and the three answers differ on purpose:

  - **granted**, and of *this* session — served, and touched so the sweeper
    does not reclaim a transfer that is plainly moving. The session is checked
    as well as the id, because touching another connection's lease refreshed
    its idle timer.
  - **queued** — refused with `lease_not_granted`, on the upload path too,
    before anything reaches the operator's disk. A member reading while queued
    is the cap not applying.
  - **unknown** — bounded by the leaseless ceiling rather than refused. That is
    also what a reconnect looks like from here, where the session's leases died
    with the old connection and the client is re-opening them, and it leaves
    the residual §5.5 already states: a client that lies gets that bound's
    worth of files at a time, not the group. Noted once per connection so the
    residual is visible rather than merely documented.

Nothing changes for the shipped client: the transfer store awaits
`lease.acquire()` before it reads a byte, so the refused case is one it never
enters. §5.5 gains a paragraph saying the node decides which of the two a
request is — the document described the accounting without ever saying it was
enforced, which is how it came not to be.

`test_lease_enforcement.py` drives the real handlers over a real index; six of
its nine cases fail against the previous source, each on the property.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UMxEQadpzPkYLFf5CYKhpW

1 files changed, 109 insertions, 10 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 8df88d8..d36b812 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -189,6 +189,11 @@ _LINK_PREVIEW_RATE_NODE = 60
 # recorded uploader, then delete it legitimately).
 MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file
 
+# What the `tr` on a chunk request turned out to be (see `_lease_of`).
+LEASE_GRANTED = "granted"
+LEASE_QUEUED = "queued"
+LEASE_NONE = "none"
+
 # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
 # nowhere near enough to be a memory-exhaustion primitive (H6).
 PRE_HANDSHAKE_MAX_MSG = 64 * 1024
@@ -441,6 +446,10 @@ class WebRTCPeerSession:
         # one session may do while claiming to be browsing, not a pool shared
         # between them. Three tabs open is browsing in three tabs.
         self._leaseless = transfers_mod.LeaselessReads()
+        # Whether this session has already been noted as transferring under a
+        # lease the node does not have (see `_note_unleased`). One line per
+        # connection, not per chunk.
+        self._unleased_noted = False
         # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message
         # arrived, so the heartbeat can report silence duration.
         self._last_msg_at: float = 0.0
@@ -3270,6 +3279,61 @@ class WebRTCPeerSession:
             slots.group_limits[self._group_id] = dict(limits)
         return slots
 
+    def _lease_of(self, tr) -> str:
+        """What the `tr` on a request actually is, from the node's own record.
+
+        `tr` is drawn by the client (§5.5) and arrives on every chunk request
+        and every upload chunk. It was read as a boolean: *present* meant "this
+        is a leased transfer", and nothing asked whether the node had ever
+        granted such a lease — so any non-empty string skipped the leaseless
+        ceiling and every cap the operator set. `touch()` has always answered
+        exactly this question (`False` if it is not granted) and its answer was
+        discarded.
+
+        Three outcomes, because they deserve different treatment:
+
+        - **`granted`** — a live lease of *this session*, and the transfer is
+          under the caps it was granted against. The session is checked as well
+          as the id: a lease belongs to a connection, and touching somebody
+          else's would refresh their idle timer.
+        - **`queued`** — the node has this lease and has not granted it. The
+          client is jumping its own queue; refused, and no shipped client does
+          it (the transfer store awaits the grant before it reads a byte).
+        - **`none`** — the node has no such lease. Deliberately *not* a refusal:
+          it is what a reconnect looks like from here, where the session's
+          leases died with the old connection and the client is re-opening
+          them, and it is what a client that never asked looks like. Both are
+          then bounded by the leaseless ceiling instead — which is the residual
+          §5.5 already states: a client that lies gets that bound's worth of
+          files at a time, not the whole library.
+        """
+        slots = self._ctx.get("_transfer_slots")
+        if slots is None:
+            return LEASE_NONE
+        lease = slots.leases.get(tr)
+        if lease is None or lease.session_key != self._registry_key:
+            return LEASE_NONE
+        if lease.state != "granted":
+            return LEASE_QUEUED
+        slots.touch(tr)
+        return LEASE_GRANTED
+
+    def _note_unleased(self, tr: str) -> None:
+        """Say once that this connection transferred outside its lease.
+
+        Once per session, not per chunk: the interesting fact is that it
+        happened, and a per-chunk line would bury it under itself. The bound is
+        the leaseless ceiling either way; this is what makes the residual
+        visible to the operator rather than merely stated in a document.
+        """
+        if self._unleased_noted:
+            return
+        self._unleased_noted = True
+        log.info("transfer: %s sent chunk requests under an unknown lease %s "
+                 "— bounded by the leaseless ceiling",
+                 (self._user_id or "?")[:8], str(tr)[:8])
+        self._audit("transfer_unleased", str(tr)[:16])
+
     def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict:
         slots = self._slots()
         out = {
@@ -3514,11 +3578,24 @@ class WebRTCPeerSession:
         # file was transferring at 20 MB/s. Found in the node's own log, which
         # repeated the same two reclaims every 30 s for as long as the daemon
         # ran.
-        tr = msg.get("tr")
+        #
+        # And it is also what makes the caps real: `tr` names a lease or it
+        # does not, and `_lease_of` is the only thing that decides which.
+        tr = str(msg.get("tr") or "")[:64]
+        leased = False
         if tr:
-            slots = self._ctx.get("_transfer_slots")
-            if slots is not None:
-                slots.touch(str(tr)[:64])
+            state = self._lease_of(tr)
+            if state == LEASE_QUEUED:
+                self._send({
+                    "type": "error",
+                    "detail": "This transfer is waiting for a slot.",
+                    "code": "lease_not_granted",
+                    "tr": tr,
+                })
+                return
+            leased = state == LEASE_GRANTED
+            if not leased:
+                self._note_unleased(tr)
         file_id = msg["file_id"]
         chunk_index = msg["chunk_index"]
         entry = ctx["index"].get_entry(file_id)
@@ -3545,7 +3622,12 @@ class WebRTCPeerSession:
         # posters and cover art never reach this line: they resolve through
         # `_try_serve_thumbnail` above, out of a cache the node built itself,
         # and are never leased, never counted, never queued.
-        if not tr:
+        #
+        # `not leased`, not `not tr`: a `tr` the node cannot match to a granted
+        # lease of this session is not a transfer, whatever the client calls it,
+        # and reading the field as a boolean is what let any string at all
+        # bypass this ceiling and the member cap behind it.
+        if not leased:
             if not self._leaseless.admit(str(file_id)):
                 self._send({
                     "type": "error",
@@ -3583,7 +3665,7 @@ class WebRTCPeerSession:
         # The last chunk is the only "close" a leaseless read has. Without this
         # the session carries the entry until it goes idle, and the person who
         # just looked at two photos cannot look at a third for a minute.
-        if not tr and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
+        if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
             self._leaseless.finish(str(file_id))
 
     @staticmethod
@@ -4683,11 +4765,28 @@ class WebRTCPeerSession:
         # `_do_file_request` marks a lease alive for exactly the same reason;
         # both sides of a transfer have to say they are still moving, or the
         # sweeper reclaims whichever one forgot.
-        tr = msg.get("tr")
+        #
+        # And a queued lease is refused here rather than written to disk. There
+        # is no leaseless fallback on this side — a write is never "browsing" —
+        # so the two cases part company: a lease this node has not granted is a
+        # member taking a slot they were told to wait for, and an unknown `tr`
+        # is the reconnect case, where the client is re-opening leases that died
+        # with the old connection and the four upload protections (§6.4) are
+        # what bound it meanwhile.
+        tr = str(msg.get("tr") or "")[:64]
         if tr:
-            slots = self._ctx.get("_transfer_slots")
-            if slots is not None:
-                slots.touch(str(tr)[:64])
+            state = self._lease_of(tr)
+            if state == LEASE_QUEUED:
+                self._send({
+                    "type": "error",
+                    "detail": "This upload is waiting for a slot.",
+                    "code": "lease_not_granted",
+                    "upload_id": upload_id,
+                    "tr": tr,
+                })
+                return
+            if state == LEASE_NONE:
+                self._note_unleased(tr)
 
         gek = ctx.get("gek")
         if not gek: