fix: bound what one member can cost the others

An availability review, prompted by the group claim above: a participant
supplies input — who else bears the cost? Six answers where the cost fell on
someone other than the sender, and none of them needs an attacker.

  AV3  `chat_notify` carried a `group_id` the hub believed, so any connected
       node could write a notification to every member of any group on the
       hub, carrying a display string of its choosing, with its account
       having no relation to that group. This is the group claim again, two
       hundred lines further down the same socket. Gated on what the node is
       registered for, and metered: the fan-out is one write per member. The
       budget expires by time rather than on disconnect, or reconnecting
       would refill it and a node token is good for an hour.

  AV4  A swarm source named its own `endpoint` as free text documented as
       "ip:port", so an account could publish a third party's address — H6's
       `peer_ip` defect, never applied here. Nothing dials a swarm source
       today, which is the only reason it was not already a reflection
       primitive. It is a transport and a port now, never a host, and the
       number of hashes one account may claim is bounded: rows were keyed
       (hash, account) with no cap at all.

  AV5  `handle_webrtc_answer` resolved any pending `peer_id` from any node's
       socket. The answer is the SDP a browser then connects to. That this
       had not happened rested on a uuid4 being unguessable.

  AV6  `relay_register` had no authentication of any kind: it compared
       `pk_relay` against the approved value, which is a *public* key, so
       anyone who could read it could rewrite where the hub tells nodes to
       send relayed traffic. The module docstring promised signed JWTs and
       `jwt` was imported and never used.

  AV7  The node held unlimited peer connections and kept one that never
       completed a handshake for the life of the daemon. H6 bounded what one
       unauthenticated peer costs; the hub's cap is three offers in flight
       per *account*, a limit on each caller and not on the machine, so an
       operator's exposure grew with the size of their groups.

  AV8  `invite-notify` put a request-supplied `group_name` into the subject
       of an email the hub sends under its own domain, to any account, with
       no rate limit. The name comes from the group row now.

The tests are two accounts each, in one file that says why: a one-member test
proves a one-member property, and every finding here needed a second person
to exist at all. Each was checked against the unfixed code. Two did not
survive that check and were rewritten — one re-enacted the disconnect path
instead of running it (hence `forget_node`), the other called the reaper
itself and would have passed with the call removed from `handle_offer`.

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

11 files changed, 809 insertions, 28 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 88125c0..b1a22f7 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -1,5 +1,7 @@
 """Group endpoints — /v1/groups/*"""
 
+import re
+
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 from datetime import datetime, timezone
@@ -10,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from meshbay_hub import hub_settings, mail
 from meshbay_hub.auth import decrypt_email
 from meshbay_hub.api.deps import get_current_user, require_user_scope
+from meshbay_hub.api.middleware import limiter
 from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import (
@@ -192,12 +195,31 @@ async def list_public_groups(
 
 class SwarmRegisterRequest(BaseModel):
     content_hash: str    # blake3 hex
-    endpoint:     str    # "ip:port"
+    endpoint:     str    # "<scheme>:<port>" — a port on the caller, never a host
+
+
+# A transport and a port, and deliberately no host. The field used to be free
+# text documented as "ip:port", so a caller could name *someone else's*
+# address as a source; nothing dials a swarm source today, which is the only
+# reason that was not already a reflection primitive. A reader learns where a
+# node is from the node record, which is stamped with the address the announce
+# actually came from — so a host here would be a second, weaker, answer to a
+# question already settled elsewhere.
+_SWARM_ENDPOINT = re.compile(r"^(webrtc|quic):([0-9]{1,5})$")
+
+# One account, this many public hashes. Rows are keyed (hash, account) with no
+# cap, so a loop of invented hashes was unbounded storage growth on a hub
+# shared with everyone else. A public library far larger than this is a real
+# thing — but it is one a hub operator should be asked about, not something a
+# client establishes by writing rows.
+MAX_SWARM_HASHES_PER_ACCOUNT = 10_000
 
 
 @swarm_router.post("/register", status_code=201)
+@limiter.limit("120/minute")
 async def swarm_register(
     body: SwarmRegisterRequest,
+    request: Request,
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
@@ -209,11 +231,21 @@ async def swarm_register(
     node's calls 404'd and the leak was masked by a routing bug rather than
     prevented. Nodes now filter by group visibility before calling, and the path is
     correct, so the filter has to be right.
+
+    Availability: the endpoint is a port, not an address, and the number of
+    hashes one account may claim is bounded. See the two constants above.
     """
     from meshbay_hub.csam import check_content_hash
     if check_content_hash(body.content_hash):
         raise HTTPException(status_code=451, detail="Content blocked")
 
+    m = _SWARM_ENDPOINT.match(body.endpoint or "")
+    if not m or not (0 < int(m.group(2)) < 65536):
+        raise HTTPException(
+            status_code=422,
+            detail="endpoint must be '<webrtc|quic>:<port>' — a port on the "
+                   "registering node, not an address")
+
     from datetime import datetime, timezone
     existing = await db.get(SwarmSource, (body.content_hash, current_user.id))
     now = datetime.now(timezone.utc)
@@ -221,6 +253,14 @@ async def swarm_register(
         existing.endpoint  = body.endpoint
         existing.last_seen = now
     else:
+        held = (await db.execute(
+            select(func.count()).select_from(SwarmSource)
+            .where(SwarmSource.node_id == current_user.id))).scalar() or 0
+        if held >= MAX_SWARM_HASHES_PER_ACCOUNT:
+            raise HTTPException(
+                status_code=429,
+                detail="This account already claims the maximum number of "
+                       "public content hashes")
         db.add(SwarmSource(
             content_hash=body.content_hash,
             node_id=current_user.id,
@@ -672,16 +712,27 @@ async def delete_group(
     return {"status": "deleted", "group_id": group_id}
 
 
+# `XXXX-XXXX`, as `roster.generate_code` produces. Checked because this string
+# is placed in an email the hub sends under its own domain, and the endpoint
+# used to accept any text at all.
+_INVITE_CODE = re.compile(r"^[0-9A-Za-z]{4}-[0-9A-Za-z]{4}$")
+
+
 class InviteNotifyRequest(BaseModel):
     username:   str
     code:       str
-    group_name: str
+    # `group_name` used to be here and went straight into the email's subject
+    # line. The hub knows the group's name — it is reading the row two lines
+    # into the handler — so accepting a second answer only let the sender
+    # choose the subject of a message the hub signs with its own domain.
 
 
 @router.post("/{group_id}/invite-notify")
+@limiter.limit("20/hour")
 async def invite_notify(
     group_id: str,
     body: InviteNotifyRequest,
+    request: Request,
     current_user: User = Depends(require_user_scope),
     db: AsyncSession = Depends(get_db),
 ):
@@ -691,6 +742,15 @@ async def invite_notify(
     because the inviter's browser sends it here. The hub looks up the
     invitee's encrypted email, decrypts it, and sends the notification.
     The inviter never sees the email address.
+
+    Availability: the target is any account on the hub — it has to be, since
+    an invitee is by definition not yet a member — so this is the one endpoint
+    where one user causes mail to be sent to another. It was unmetered and the
+    subject line came from the request. Anyone who created a group, which is to
+    say anyone, could send any registered account arbitrary text from the hub's
+    own domain, as fast as they liked. The rate limit and the two checks below
+    are what keep that from being a phishing kit with the hub's reputation
+    attached.
     """
     group = await db.get(Group, group_id)
     if not group:
@@ -699,6 +759,9 @@ async def invite_notify(
         raise HTTPException(status_code=403,
                             detail="Only the group owner can send invitations")
 
+    if not _INVITE_CODE.match(body.code or ""):
+        raise HTTPException(status_code=422, detail="Not an invite code")
+
     target = (await db.execute(
         select(User).where(User.username == body.username))).scalar_one_or_none()
     if not target:
@@ -715,7 +778,7 @@ async def invite_notify(
 
     try:
         mail.send_invite_notification(
-            email, body.code, current_user.username, body.group_name)
+            email, body.code, current_user.username, group.name)
     except Exception:
         return {"status": "send_failed"}
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/relay.py b/packages/meshbay-hub/src/meshbay_hub/api/relay.py
index f82495b..c6ef26e 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/relay.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/relay.py
@@ -8,24 +8,25 @@ Relay registration:
   POST /v1/relays/register   — relay announces itself (signed JWT)
   GET  /v1/relays            — list active relays (for nodes)
 
-Relay authentication: relay generates an Ed25519 keypair at install time.
-It registers its public key with the hub admin, then signs keepalive JWTs.
+Relay authentication: relay generates an Ed25519 keypair at install time. An
+admin approves the public key, and every register call carries an Ed25519
+signature over "meshbay:relay_register:<relay_id>:<endpoint>:<timestamp>" —
+the same proof-of-possession shape as /v1/nodes/announce.
 
 Relay is responsible for E2E encrypted QUIC traffic only (it cannot
 read the application-layer content, only forward UDP packets).
 """
 
+import base64
 import logging
 import time
-import uuid
 
-import jwt
-from fastapi import APIRouter, Depends, Header, HTTPException
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from meshbay_hub.api.deps import get_current_user, require_admin
-from meshbay_hub.auth import _hub_id, hub_public_key_pem
+from meshbay_hub.api.deps import require_admin
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import User
 
@@ -40,11 +41,13 @@ _relays: dict[str, dict] = {}   # relay_id → {endpoint, pk, last_seen, capacit
 # ── Models ────────────────────────────────────────────────────────────────────
 
 class RelayRegisterRequest(BaseModel):
-    """Relay self-registers with a signed JWT."""
+    """Relay self-registers, proving possession of its approved key."""
     relay_id:   str
     endpoint:   str          # "ip:port" (UDP)
     pk_relay:   str          # base64 Ed25519 public key
     capacity:   int = 100    # max concurrent connections
+    timestamp:  int | None = None    # unix seconds
+    signature:  str | None = None    # base64 Ed25519 over the register message
 
 
 class RelayAdminApproveRequest(BaseModel):
@@ -54,20 +57,48 @@ class RelayAdminApproveRequest(BaseModel):
 
 # ── Relay endpoints ───────────────────────────────────────────────────────────
 
+REGISTER_TIMESTAMP_WINDOW = 300   # seconds either side, as /v1/nodes/announce
+
+
 @router.post("/register", status_code=201)
 async def relay_register(
     body: RelayRegisterRequest,
     db: AsyncSession = Depends(get_db),
 ):
     """
-    Relay announces itself. Must be pre-approved by a hub admin.
-    The relay's public key must already be in the approved list.
+    Relay announces itself. Must be pre-approved by a hub admin, and must prove
+    it holds the private key that approval registered.
+
+    This endpoint has no `Depends` on an account on purpose — a relay is not a
+    user — but it had no proof of anything either: it compared `pk_relay`
+    against the approved value, which is a **public** key, so anyone who could
+    read it could rewrite where the hub tells nodes to send relayed traffic.
+    The module docstring said "signs keepalive JWTs" and nothing verified a
+    signature; `jwt` was imported and never used. A key is not a password, and
+    the fix is the proof-of-possession pattern already used by
+    /v1/nodes/announce and /v1/nodes/auth.
     """
     approved = _relays.get(body.relay_id)
     if not approved or approved.get("pk") != body.pk_relay:
         raise HTTPException(status_code=403,
                             detail="Relay not approved — ask hub admin to run POST /v1/relays/approve")
 
+    if body.timestamp is None or not body.signature:
+        raise HTTPException(
+            status_code=400,
+            detail="register requires timestamp and signature (proof of possession)")
+    if abs(int(time.time()) - body.timestamp) > REGISTER_TIMESTAMP_WINDOW:
+        raise HTTPException(status_code=401, detail="Timestamp too old or too far ahead")
+
+    message = (f"meshbay:relay_register:{body.relay_id}:"
+               f"{body.endpoint}:{body.timestamp}").encode()
+    try:
+        pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(body.pk_relay))
+        pk.verify(base64.b64decode(body.signature), message)
+    except Exception:
+        log.warning("Relay %s failed proof of possession", body.relay_id[:8])
+        raise HTTPException(status_code=401, detail="Invalid relay key proof of possession")
+
     _relays[body.relay_id].update({
         "endpoint":  body.endpoint,
         "capacity":  body.capacity,
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 7c37a1e..0e274f9 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -68,6 +68,53 @@ def get_online_nodes_for_group(group_id: str) -> list[str]:
     return [nid for nid, gids in _node_groups.items() if group_id in gids]
 
 
+# A chat_notify costs one query per member of the group plus a write for each,
+# and nothing on the node's side paces it. Without a budget, one node can keep
+# the hub's database busy on behalf of a group it belongs to — a cost borne by
+# every other group on the instance. Generous enough that a lively conversation
+# never meets it: notifications aggregate to one row per person per group, so
+# the useful rate is far below this.
+NOTIFY_BURST = 30            # messages
+NOTIFY_WINDOW_SECONDS = 60
+
+_notify_window: dict[str, tuple[float, int]] = {}   # node_id → (window start, count)
+
+
+def forget_node(node_id: str) -> None:
+    """Drop everything a disconnected node's socket owned.
+
+    One function rather than three lines in a `finally`, so that what a
+    disconnect does can be asserted by running it instead of by re-enacting it
+    — a test that re-enacts cleanup tests its own re-enactment, and would not
+    have noticed `_notify_window` being added here.
+
+    Which is the point: `_notify_window` is **not** cleared. Dropping it would
+    make reconnecting the way to refill the budget, and a node's token stays
+    valid for an hour. It expires by time, in `_notify_budget`.
+    """
+    _connected_nodes.pop(node_id, None)
+    _node_groups.pop(node_id, None)
+
+
+def _notify_budget(node_id: str) -> bool:
+    """True if this node may send one more chat_notify now."""
+    now = time.monotonic()
+    if len(_notify_window) > 1000:
+        # Swept here rather than on disconnect, which would let a node refill
+        # its budget by reconnecting — the same token stays valid for an hour.
+        for nid, (started, _) in list(_notify_window.items()):
+            if now - started >= NOTIFY_WINDOW_SECONDS:
+                _notify_window.pop(nid, None)
+    start, count = _notify_window.get(node_id, (now, 0))
+    if now - start >= NOTIFY_WINDOW_SECONDS:
+        start, count = now, 0
+    if count >= NOTIFY_BURST:
+        _notify_window[node_id] = (start, count)
+        return False
+    _notify_window[node_id] = (start, count + 1)
+    return True
+
+
 
 async def _mark_hosted(group_ids: list[str]) -> None:
     """Stamp the first time a node announced it hosts each of these groups.
@@ -133,10 +180,27 @@ def _sign_revocation(target: str, target_id: str, reason: str) -> str:
 
 # ── WebSocket endpoint ────────────────────────────────────────────────────────
 
-async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str) -> None:
-    """Node informs hub that a chat message was posted — create notifications for offline members."""
+async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str,
+                              *, node_id: str) -> None:
+    """Node informs hub that a chat message was posted — create notifications for offline members.
+
+    `group_id` arrives in the node's own message and is checked against what
+    that node is registered for. Without the check, any connected node could
+    write a notification to every member of **any** group on the hub, carrying
+    a display string of its choosing, with its account having no relation to
+    that group at all. Same shape as the empty group claim: something believed
+    about a group the sender has nothing to do with.
+
+    `node_id` is keyword-**required** rather than defaulted. A default here
+    would mean "unchecked when the caller forgets", which is the failure this
+    whole review is about.
+    """
     if not group_id:
         return
+    if group_id not in _node_groups.get(node_id, ()):
+        log.warning("Node %s sent chat_notify for a group it does not host",
+                    (node_id or "?")[:8])
+        return
     try:
         from meshbay_hub.db.engine import get_session_factory
         from meshbay_hub.db.models import GroupMember, Group
@@ -318,7 +382,7 @@ async def node_websocket(ws: WebSocket):
                     event.set()
             elif msg.get("type") == "webrtc_answer":
                 from meshbay_hub.api.signaling import handle_webrtc_answer
-                handle_webrtc_answer(msg)
+                handle_webrtc_answer(msg, node_id)
             elif msg.get("type") == "update_groups":
                 # Through the same gate as the registration above. This used to
                 # assign the message's list verbatim, so the ceiling that makes
@@ -330,6 +394,9 @@ async def node_websocket(ws: WebSocket):
                 await _mark_hosted(new_gids)
                 log.info("Node %s updated groups: %d", node_id[:8], len(new_gids))
             elif msg.get("type") == "chat_notify":
+                if not _notify_budget(node_id):
+                    log.warning("Node %s exceeded its chat_notify rate", node_id[:8])
+                    continue
                 asyncio.ensure_future(_handle_chat_notify(
                     msg.get("group_id", ""),
                     msg.get("sender_name", ""),
@@ -339,6 +406,7 @@ async def node_websocket(ws: WebSocket):
                     # that lied here could only suppress one notification, which
                     # is the same power it has by not sending the message at all.
                     msg.get("sender_user_id", ""),
+                    node_id=node_id,
                 ))
 
     except WebSocketDisconnect:
@@ -347,8 +415,7 @@ async def node_websocket(ws: WebSocket):
         log.error("Node WS error: %s", e)
     finally:
         if node_id:
-            _connected_nodes.pop(node_id, None)
-            _node_groups.pop(node_id, None)
+            forget_node(node_id)
 
 
 # ── Admin revocation endpoint ─────────────────────────────────────────────────
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
index a8feae8..b4be3f2 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
@@ -34,6 +34,9 @@ log = logging.getLogger(__name__)
 router = APIRouter(prefix="/v1/nodes", tags=["signaling"])
 
 _webrtc_answers: dict[str, asyncio.Future] = {}
+# peer_id → the node the offer was relayed to. An answer is only accepted
+# from that node (see handle_webrtc_answer).
+_answer_owner: dict[str, str] = {}
 
 
 class WebRTCOfferRequest(BaseModel):
@@ -133,6 +136,7 @@ async def webrtc_offer(
     peer_id = str(uuid.uuid4())
     answer_future: asyncio.Future = asyncio.get_event_loop().create_future()
     _webrtc_answers[peer_id] = answer_future
+    _answer_owner[peer_id] = node_id
     _pending_per_user[current_user.id] = _pending_per_user.get(current_user.id, 0) + 1
 
     try:
@@ -157,6 +161,7 @@ async def webrtc_offer(
         )
     finally:
         _webrtc_answers.pop(peer_id, None)
+        _answer_owner.pop(peer_id, None)
         remaining = _pending_per_user.get(current_user.id, 1) - 1
         if remaining > 0:
             _pending_per_user[current_user.id] = remaining
@@ -164,13 +169,26 @@ async def webrtc_offer(
             _pending_per_user.pop(current_user.id, None)
 
 
-def handle_webrtc_answer(msg: dict) -> None:
-    """Called from the node WebSocket message loop when a webrtc_answer arrives."""
+def handle_webrtc_answer(msg: dict, node_id: str) -> None:
+    """Called from the node WebSocket message loop when a webrtc_answer arrives.
+
+    `node_id` is the socket this arrived on, and the answer is accepted only for
+    a `peer_id` the hub issued to **that** node. The answer carries the SDP the
+    browser then connects to, so without the check any connected node could
+    resolve any pending offer and stand in for the node the client asked for.
+    That it had not happened rested on a uuid4 being unguessable, which is a
+    reason it was hard, not a reason it was refused.
+    """
     peer_id = msg.get("peer_id")
     if not peer_id:
         log.warning("webrtc_answer without peer_id")
         return
 
+    if _answer_owner.get(peer_id) != node_id:
+        log.warning("Node %s answered an offer it was never sent (peer=%s)",
+                    (node_id or "?")[:8], str(peer_id)[:8])
+        return
+
     future = _webrtc_answers.get(peer_id)
     if future and not future.done():
         future.set_result({
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js
index 8e11088..0f8ccdd 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js
@@ -794,7 +794,10 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef,
       try {
         const notif = await hubFetch(`/v1/groups/${groupId}/invite-notify`, {
           method: 'POST', token,
-          body: { username, code: result.code, group_name: group?.name || '' },
+          // No group_name: the hub reads it from the group row it has already
+          // loaded. Sending one offered a second answer to a settled question,
+          // and that answer was the subject line of an email the hub signs.
+          body: { username, code: result.code },
         });
         emailStatus = notif.status;
       } catch { /* best effort */ }
diff --git a/packages/meshbay-hub/tests/test_account_deletion.py b/packages/meshbay-hub/tests/test_account_deletion.py
index b97417d..4cbca8b 100644
--- a/packages/meshbay-hub/tests/test_account_deletion.py
+++ b/packages/meshbay-hub/tests/test_account_deletion.py
@@ -135,7 +135,10 @@ async def test_deletion_clears_device_keys_and_swarm_sources(client, db_session)
     """
     The privacy statement says every account row goes but the IP log. Swarm
     sources are keyed by the *user* id despite the column's name, and carry the
-    node's ip:port.
+    node's transport and port — `webrtc:<port>`, which is what `daemon.py`
+    actually sends. This asked with `192.0.2.7:4433`, from the days when the
+    field was free text documented as "ip:port": a shape no node has ever
+    produced, and one that let a caller name a third party's address.
     """
     from meshbay_hub.db.models import SwarmSource, UserDevice
 
@@ -148,7 +151,7 @@ async def test_deletion_clears_device_keys_and_swarm_sources(client, db_session)
                           json={"pk_auth_ed25519": _device_pk(), "label": "desktop"})
     assert r.status_code == 201, r.text
     r = await client.post("/v1/swarm/register", headers=headers,
-                          json={"content_hash": "ab" * 32, "endpoint": "192.0.2.7:4433"})
+                          json={"content_hash": "ab" * 32, "endpoint": "webrtc:4433"})
     assert r.status_code == 201, r.text
 
     # Present before, or the emptiness asserted below proves nothing.
diff --git a/packages/meshbay-hub/tests/test_availability_between_members.py b/packages/meshbay-hub/tests/test_availability_between_members.py
new file mode 100644
index 0000000..f2c9a4f
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_availability_between_members.py
@@ -0,0 +1,388 @@
+"""
+Availability: what one participant can do to the others.
+
+The three security reviews asked who can *read* what, who can impersonate whom,
+and what a hostile node can forge. None of them asked what a legitimate but
+misconfigured or careless member costs everyone else — and that is the question
+a group platform lives or dies on, because every member is invited by someone
+who trusted them and none of them is an attacker.
+
+Written after 2026-09-11, where a member's unconfigured node was registered by
+the hub as a host for a group it could not serve, and — being answered first —
+made that group unopenable for everyone in it. Nothing was compromised and
+nothing was forged. The group was simply gone.
+
+Each test below is two accounts, because that is the shape the single-node
+suite could not express: `_make_user` twice, a group owned by one, the other
+holding whatever the defect needs. A one-member test proves a one-member
+property, and every finding here needed a second person to exist at all.
+"""
+
+import base64
+import time
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+
+from meshbay_common.crypto import pk_to_b64
+
+
+async def _make_user(client, username: str) -> dict:
+    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
+    pk_ed, pk_x = pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key())
+    r = await client.post("/v1/users/register", json={
+        "username": username,
+        "email": f"{username}@example.test",
+        "auth_key": base64.b64encode(b"k" * 32).decode(),
+        "pk_user_ed25519": pk_ed,
+        "pk_user_x25519": pk_x,
+    })
+    assert r.status_code == 201, r.text
+    user_id = r.json()["user_id"]
+    r = await client.post("/v1/users/login", json={
+        "username": username,
+        "auth_key": base64.b64encode(b"k" * 32).decode(),
+    })
+    assert r.status_code == 200, r.text
+    return {"user_id": user_id, "username": username,
+            "token": r.json()["access_token"], "pk_ed": pk_ed, "sk_ed": sk_ed}
+
+
+async def _announce_node(client, user: dict) -> str:
+    ts = int(time.time())
+    msg = f"meshbay:node_announce:{user['user_id']}:{user['pk_ed']}:{ts}".encode()
+    r = await client.post("/v1/nodes/announce", json={
+        "pk_node": user["pk_ed"], "endpoint_hint": "test", "timestamp": ts,
+        "signature": base64.b64encode(user["sk_ed"].sign(msg)).decode(),
+    }, headers={"Authorization": f"Bearer {user['token']}"})
+    assert r.status_code == 201, r.text
+    return r.json()["node_id"]
+
+
+async def _make_group(client, owner: dict, name: str) -> str:
+    r = await client.post(
+        "/v1/groups",
+        json={"name": name, "visibility": "private", "join_policy": "invite"},
+        headers={"Authorization": f"Bearer {owner['token']}"})
+    assert r.status_code == 201, r.text
+    return r.json()["group_id"]
+
+
+async def _add_member(client, owner: dict, group_id: str, member: dict) -> None:
+    r = await client.post(
+        f"/v1/groups/{group_id}/members/{member['username']}",
+        headers={"Authorization": f"Bearer {owner['token']}"})
+    assert r.status_code == 201, r.text
+
+
+# ── A member's node must not speak for a group it does not host ──────────────
+
+@pytest.mark.asyncio
+async def test_a_members_node_cannot_notify_a_group_it_does_not_host(client):
+    """
+    `chat_notify` carried a `group_id` the hub believed, so any connected node
+    could write a notification to every member of any group on the hub, with a
+    display string of its own choosing. The node's account needed no relation
+    to the group whatsoever — this is the same defect as the group claim, one
+    message further along the same socket.
+    """
+    from meshbay_hub.api import revocation as rev
+
+    owner = await _make_user(client, "av_owner1")
+    outsider = await _make_user(client, "av_outsider1")
+    owner_node = await _announce_node(client, owner)
+    outsider_node = await _announce_node(client, outsider)
+    group_id = await _make_group(client, owner, "not-yours")
+
+    rev._node_groups[owner_node] = [group_id]
+    rev._node_groups[outsider_node] = []          # hosts nothing, as in the incident
+    try:
+        await rev._handle_chat_notify(
+            group_id, "Someone you do not know", outsider["user_id"],
+            node_id=outsider_node)
+
+        r = await client.get("/v1/notifications",
+                             headers={"Authorization": f"Bearer {owner['token']}"})
+        assert r.status_code == 200, r.text
+        assert r.json()["notifications"] == [], (
+            "an unrelated node wrote into this account's notifications")
+
+        # And the node that does host it is still able to.
+        await rev._handle_chat_notify(
+            group_id, "A real member", outsider["user_id"], node_id=owner_node)
+        r = await client.get("/v1/notifications",
+                             headers={"Authorization": f"Bearer {owner['token']}"})
+        assert len(r.json()["notifications"]) == 1
+    finally:
+        rev._node_groups.pop(owner_node, None)
+        rev._node_groups.pop(outsider_node, None)
+
+
+@pytest.mark.asyncio
+async def test_a_member_whose_node_hosts_nothing_still_cannot_notify(client):
+    """
+    The incident's own shape, and the one a check on *membership* would have
+    missed: the second account is a real member of the group, invited by its
+    owner. Their node is simply not the one holding the files. Being entitled
+    to be in a group is not being entitled to speak for it.
+    """
+    from meshbay_hub.api import revocation as rev
+
+    owner = await _make_user(client, "av_owner2")
+    member = await _make_user(client, "av_member2")
+    owner_node = await _announce_node(client, owner)
+    member_node = await _announce_node(client, member)
+    group_id = await _make_group(client, owner, "shared-room")
+    await _add_member(client, owner, group_id, member)
+
+    rev._node_groups[owner_node] = [group_id]
+    rev._node_groups[member_node] = []
+    try:
+        await rev._handle_chat_notify(
+            group_id, "Sounds like a member", member["user_id"],
+            node_id=member_node)
+        r = await client.get("/v1/notifications",
+                             headers={"Authorization": f"Bearer {owner['token']}"})
+        assert r.json()["notifications"] == [], (
+            "a member's empty node wrote into the group owner's notifications")
+    finally:
+        rev._node_groups.pop(owner_node, None)
+        rev._node_groups.pop(member_node, None)
+
+
+@pytest.mark.asyncio
+async def test_one_node_cannot_spend_the_hubs_database_on_notifications():
+    """
+    Each chat_notify is a query and a write per member of the group, spawned
+    without backpressure. The budget is what stops one group's node from
+    costing every other group on the instance.
+    """
+    from meshbay_hub.api.revocation import (
+        NOTIFY_BURST, _notify_budget, _notify_window)
+
+    node = "budget-node"
+    _notify_window.pop(node, None)
+    try:
+        assert all(_notify_budget(node) for _ in range(NOTIFY_BURST))
+        assert not _notify_budget(node), "the burst was not bounded"
+
+        # A second node is unaffected — the budget is per node, not global.
+        assert _notify_budget("another-node")
+    finally:
+        _notify_window.pop(node, None)
+        _notify_window.pop("another-node", None)
+
+
+@pytest.mark.asyncio
+async def test_the_notify_budget_is_not_refilled_by_reconnecting(client):
+    """
+    The obvious place to clear this is the socket's `finally`, beside the other
+    two registries — and that would make reconnecting the way around it. The
+    same node token is valid for an hour.
+    """
+    from meshbay_hub.api import revocation as rev
+
+    user = await _make_user(client, "av_reconnect")
+    node_id = await _announce_node(client, user)
+    rev._notify_window.pop(node_id, None)
+    try:
+        for _ in range(rev.NOTIFY_BURST):
+            rev._notify_budget(node_id)
+        assert not rev._notify_budget(node_id)
+
+        # The disconnect path itself, not a re-enactment of it: `forget_node`
+        # is what the socket's `finally` calls, so adding a line there is
+        # caught here.
+        rev.forget_node(node_id)
+        assert node_id not in rev._connected_nodes
+        assert node_id not in rev._node_groups
+
+        assert not rev._notify_budget(node_id), (
+            "disconnecting refilled the budget, so reconnecting defeats it")
+    finally:
+        rev._notify_window.pop(node_id, None)
+
+
+# ── A member must not aim other people's traffic ─────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_swarm_source_cannot_name_someone_elses_address(client):
+    """
+    `endpoint` was free text documented as "ip:port", so an account could
+    publish a third party's address as a source for any content. Nothing dials
+    a swarm source today, which is the only reason this was not already the
+    reflection primitive that `notify_incoming` was fixed for (H6). A port is
+    all a reader needs: where the node is comes from the node record, which is
+    stamped with the address its announce arrived from.
+    """
+    user = await _make_user(client, "av_swarm1")
+    headers = {"Authorization": f"Bearer {user['token']}"}
+
+    for bad in ("192.0.2.7:4433", "evil.example:53", "webrtc:0", "webrtc:70000",
+                "webrtc:4433 ", "http://example.test"):
+        r = await client.post("/v1/swarm/register", headers=headers,
+                              json={"content_hash": "ab" * 32, "endpoint": bad})
+        assert r.status_code == 422, f"{bad!r} was accepted: {r.text}"
+
+    r = await client.post("/v1/swarm/register", headers=headers,
+                          json={"content_hash": "ab" * 32, "endpoint": "webrtc:19010"})
+    assert r.status_code == 201, r.text
+
+
+@pytest.mark.asyncio
+async def test_one_account_cannot_fill_the_swarm_table(client, monkeypatch):
+    """Rows are keyed (hash, account) with no cap — an invented hash each time."""
+    import meshbay_hub.api.groups as groups_api
+    monkeypatch.setattr(groups_api, "MAX_SWARM_HASHES_PER_ACCOUNT", 3)
+
+    user = await _make_user(client, "av_swarm2")
+    headers = {"Authorization": f"Bearer {user['token']}"}
+    for i in range(3):
+        r = await client.post("/v1/swarm/register", headers=headers,
+                              json={"content_hash": f"{i:064x}",
+                                    "endpoint": "webrtc:19010"})
+        assert r.status_code == 201, r.text
+
+    r = await client.post("/v1/swarm/register", headers=headers,
+                          json={"content_hash": f"{99:064x}", "endpoint": "webrtc:19010"})
+    assert r.status_code == 429, r.text
+
+    # Refreshing one already held is not a new claim and must still work.
+    r = await client.post("/v1/swarm/register", headers=headers,
+                          json={"content_hash": f"{0:064x}", "endpoint": "webrtc:19011"})
+    assert r.status_code == 201, r.text
+
+
+# ── A member's node must not answer for another's ────────────────────────────
+
+def test_a_node_cannot_answer_an_offer_it_was_never_sent():
+    """
+    `handle_webrtc_answer` resolved any pending `peer_id` from any node's
+    socket. The answer is the SDP the browser then connects to, so the check is
+    what keeps one node from standing in for the node a client asked for. That
+    it had not happened rested on a uuid4 being unguessable.
+    """
+    import asyncio
+
+    from meshbay_hub.api.signaling import (
+        _answer_owner, _webrtc_answers, handle_webrtc_answer)
+
+    loop = asyncio.new_event_loop()
+    try:
+        future = loop.create_future()
+        _webrtc_answers["peer-1"] = future
+        _answer_owner["peer-1"] = "the-node-asked-for"
+
+        handle_webrtc_answer({"peer_id": "peer-1", "sdp": "v=0 impostor"},
+                             "a-different-node")
+        assert not future.done(), "another node resolved this offer"
+
+        handle_webrtc_answer({"peer_id": "peer-1", "sdp": "v=0 genuine"},
+                             "the-node-asked-for")
+        assert future.done() and future.result()["sdp"] == "v=0 genuine"
+    finally:
+        _webrtc_answers.pop("peer-1", None)
+        _answer_owner.pop("peer-1", None)
+        loop.close()
+
+
+# ── One account must not be able to mail another at will ─────────────────────
+
+@pytest.mark.asyncio
+async def test_an_invite_email_says_what_the_hub_knows_not_what_it_is_told(
+        client, monkeypatch):
+    """
+    `group_name` went from the request body into the subject line of an email
+    the hub sends under its own domain, and the invitee need not be a member of
+    anything — as they cannot be, being invited. So any account, having made a
+    group, could mail any other account arbitrary text. The name now comes from
+    the group row, and the code has to look like one.
+    """
+    sent: list = []
+    import meshbay_hub.mail as mail_mod
+    monkeypatch.setattr(mail_mod, "send_invite_notification",
+                        lambda *a, **kw: sent.append(a))
+
+    owner = await _make_user(client, "av_inviter")
+    await _make_user(client, "av_invitee")
+    group_id = await _make_group(client, owner, "real-group-name")
+    headers = {"Authorization": f"Bearer {owner['token']}"}
+
+    r = await client.post(f"/v1/groups/{group_id}/invite-notify", headers=headers,
+                          json={"username": "av_invitee", "code": "not a code",
+                                "group_name": "Your account is suspended"})
+    assert r.status_code == 422, r.text
+    assert not sent
+
+    r = await client.post(f"/v1/groups/{group_id}/invite-notify", headers=headers,
+                          json={"username": "av_invitee", "code": "AB12-CD34",
+                                "group_name": "Your account is suspended"})
+    assert r.status_code == 200, r.text
+    assert sent, "the legitimate path stopped working"
+    assert sent[0][3] == "real-group-name", (
+        "the sender chose the subject line of a message the hub signs")
+
+
+# ── A relay is not authenticated by the key it publishes ─────────────────────
+
+@pytest.mark.asyncio
+async def test_a_relay_must_prove_it_holds_the_approved_key(client, monkeypatch):
+    """
+    `relay_register` had no `Depends` and verified nothing: it compared
+    `pk_relay` against the approved value, which is a **public** key. Anyone
+    who could read it could rewrite where the hub tells nodes to send relayed
+    traffic — an unauthenticated write to state other people's machines act
+    on. The module docstring said the relay "signs keepalive JWTs"; `jwt` was
+    imported and never used.
+    """
+    from meshbay_hub.api import relay as relay_mod
+
+    sk = Ed25519PrivateKey.generate()
+    pk = pk_to_b64(sk.public_key())
+    relay_mod._relays["r1"] = {"pk": pk, "active": False}
+    try:
+        # The public key alone, which used to be enough.
+        r = await client.post("/v1/relays/register", json={
+            "relay_id": "r1", "endpoint": "198.51.100.9:9999",
+            "pk_relay": pk, "capacity": 100})
+        assert r.status_code == 400, r.text
+        assert relay_mod._relays["r1"].get("endpoint") is None
+
+        # A signature over someone else's endpoint does not carry either: the
+        # endpoint is inside the signed message.
+        ts = int(time.time())
+        sig = sk.sign(f"meshbay:relay_register:r1:10.0.0.1:4433:{ts}".encode())
+        r = await client.post("/v1/relays/register", json={
+            "relay_id": "r1", "endpoint": "198.51.100.9:9999", "pk_relay": pk,
+            "timestamp": ts, "signature": base64.b64encode(sig).decode()})
+        assert r.status_code == 401, r.text
+
+        endpoint = "203.0.113.4:4433"
+        sig = sk.sign(f"meshbay:relay_register:r1:{endpoint}:{ts}".encode())
+        r = await client.post("/v1/relays/register", json={
+            "relay_id": "r1", "endpoint": endpoint, "pk_relay": pk,
+            "timestamp": ts, "signature": base64.b64encode(sig).decode()})
+        assert r.status_code == 201, r.text
+        assert relay_mod._relays["r1"]["endpoint"] == endpoint
+    finally:
+        relay_mod._relays.pop("r1", None)
+
+
+@pytest.mark.asyncio
+async def test_a_captured_relay_registration_is_not_replayable(client):
+    """Same reason /v1/nodes/announce bounds its timestamp."""
+    from meshbay_hub.api import relay as relay_mod
+
+    sk = Ed25519PrivateKey.generate()
+    pk = pk_to_b64(sk.public_key())
+    relay_mod._relays["r2"] = {"pk": pk, "active": False}
+    try:
+        ts = int(time.time()) - 3600
+        sig = sk.sign(f"meshbay:relay_register:r2:203.0.113.5:4433:{ts}".encode())
+        r = await client.post("/v1/relays/register", json={
+            "relay_id": "r2", "endpoint": "203.0.113.5:4433", "pk_relay": pk,
+            "timestamp": ts, "signature": base64.b64encode(sig).decode()})
+        assert r.status_code == 401, r.text
+    finally:
+        relay_mod._relays.pop("r2", None)
diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py
index 724c8ec..5bb6af9 100644
--- a/packages/meshbay-hub/tests/test_hub_api.py
+++ b/packages/meshbay-hub/tests/test_hub_api.py
@@ -658,9 +658,14 @@ async def test_webrtc_signaling_roundtrip(client, app):
     from meshbay_hub.api.revocation import _connected_nodes
     from meshbay_hub.api.signaling import handle_webrtc_answer
 
+    node_id = "test-node-sig"
+
     class FakeWS:
-        def __init__(self):
+        def __init__(self, answering_as):
             self.sent = []
+            # An answer is accepted only from the node the offer was relayed
+            # to, so the stand-in has to say which node it is.
+            self._node_id = answering_as
 
         async def send_text(self, text):
             self.sent.append(json.loads(text))
@@ -672,10 +677,9 @@ async def test_webrtc_signaling_roundtrip(client, app):
                     "peer_id": msg["peer_id"],
                     "sdp": "v=0\r\nanswer-sdp",
                     "ice_candidates": [{"candidate": "test"}],
-                })
+                }, self._node_id)
 
-    fake_ws = FakeWS()
-    node_id = "test-node-sig"
+    fake_ws = FakeWS(node_id)
     _connected_nodes[node_id] = fake_ws
 
     try:
diff --git a/packages/meshbay-hub/tests/test_notifications_behaviour.py b/packages/meshbay-hub/tests/test_notifications_behaviour.py
index 376b9b0..5fe9170 100644
--- a/packages/meshbay-hub/tests/test_notifications_behaviour.py
+++ b/packages/meshbay-hub/tests/test_notifications_behaviour.py
@@ -161,6 +161,7 @@ async def test_you_are_not_notified_of_your_own_message(client, db_session):
     the operator and nobody else: everyone was told about their own messages,
     and the operator was told about no one's.
     """
+    from meshbay_hub.api import revocation as rev
     from meshbay_hub.api.revocation import _handle_chat_notify
 
     owner = await _user(client, "operator")
@@ -177,7 +178,14 @@ async def test_you_are_not_notified_of_your_own_message(client, db_session):
         select(User).where(User.username.in_(["operator", "chatty", "quiet"]))
     )).scalars().all()}
 
-    await _handle_chat_notify(gid, "chatty", ids["chatty"])
+    # A node registered for this group, because a notification is now written
+    # only for a group the sending node actually hosts (AV3).
+    node_id = "notify-test-node"
+    rev._node_groups[node_id] = [gid]
+    try:
+        await _handle_chat_notify(gid, "chatty", ids["chatty"], node_id=node_id)
+    finally:
+        rev._node_groups.pop(node_id, None)
 
     async def chat_rows(uid):
         return (await db_session.execute(
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 a0776d2..29d6e7f 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -192,6 +192,18 @@ MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file
 # 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
+
+# How many peer connections this node holds at once, and how long one may stay
+# without completing the MNP handshake. The budget above bounds what *one*
+# unauthenticated peer costs; these bound how many there may be and how long
+# each lasts, which is the other half and was missing. The hub caps three
+# offers in flight per account — a limit on each caller, not on this machine —
+# so the cost to an operator grew with the number of people in their groups.
+# Sized to be unreachable in ordinary use: a browser holds one connection per
+# open group, and a handshake unfinished after a minute is not going to finish.
+MAX_PEER_SESSIONS = 64
+UNAUTHENTICATED_SESSION_TIMEOUT = 60   # seconds
+
 # ffmpeg is spawned per stream request; without a cap any member can fork-bomb
 # the node by requesting many streams at once (H6).
 #
@@ -5882,6 +5894,7 @@ class WebRTCTransport:
         from meshbay_node.config import DEFAULT_STUN_SERVERS
         self._stun = stun_servers or list(DEFAULT_STUN_SERVERS)
         self._sessions: dict[str, WebRTCPeerSession] = {}
+        self._reapers: set[asyncio.Task] = set()
 
     def set_capacity(self, *, max_concurrent_streams: int | None = None,
                      max_concurrent_downloads: int | None = None,
@@ -5991,9 +6004,21 @@ class WebRTCTransport:
         config = RTCConfiguration(
             iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else []
         )
+        # Before anything is allocated. Every offer costs an RTCPeerConnection
+        # with its own DTLS and SCTP stacks, and nothing here used to bound how
+        # many a node would hold: the hub caps three in flight *per account*,
+        # which is a limit on each caller and not on this machine, so the cost
+        # grew with the number of members in the group. An operator's node must
+        # not be exhaustible by the people they invited.
+        if len(self._sessions) >= MAX_PEER_SESSIONS:
+            log.warning("Refusing WebRTC offer: %d peer sessions already open",
+                        len(self._sessions))
+            raise RuntimeError("Node is at its peer-connection limit")
+
         pc = RTCPeerConnection(configuration=config)
         session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id)
         self._sessions[peer_id] = session
+        self._reap_if_unauthenticated(peer_id)
 
         @pc.on("datachannel")
         def on_datachannel(channel: RTCDataChannel):
@@ -6048,12 +6073,47 @@ class WebRTCTransport:
             ", ".join(sorted(host_addrs)) or "none", srflx)
         return answer_sdp, []
 
+    def _reap_if_unauthenticated(self, peer_id: str) -> None:
+        """Close a session that never completes the handshake.
+
+        A peer that connects and then says nothing is indistinguishable from a
+        working one until it is asked to prove something, and it was never
+        asked: `connectionstatechange` reaps a connection that *fails*, and one
+        that succeeds and stays silent was held for the node's lifetime. That
+        is the cheapest way to spend someone else's memory — no GEK, no token,
+        no group, just an open connection. `_user_id` is set by the GEK proof
+        (`_do_handshake_response`), so it is the one honest test of whether
+        this peer ever became anybody.
+        """
+        async def reap() -> None:
+            try:
+                await asyncio.sleep(UNAUTHENTICATED_SESSION_TIMEOUT)
+                session = self._sessions.get(peer_id)
+                if session is not None and not session._user_id:
+                    log.warning("Closing peer %s: no handshake within %ds",
+                                peer_id[:8], UNAUTHENTICATED_SESSION_TIMEOUT)
+                    await self.close_peer(peer_id)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                log.warning("Reaping peer %s failed: %s", peer_id[:8], e)
+
+        # Held in a set for the same reason every other task here is: asyncio
+        # keeps only a weak reference, and a reaper collected mid-sleep reaps
+        # nothing (see WebRTCPeerSession.__init__).
+        task = asyncio.ensure_future(reap())
+        self._reapers.add(task)
+        task.add_done_callback(self._reapers.discard)
+
     async def close_peer(self, peer_id: str) -> None:
         session = self._sessions.pop(peer_id, None)
         if session:
             await session.close()
 
     async def close_all(self) -> None:
+        for task in list(self._reapers):
+            task.cancel()
+        self._reapers.clear()
         for session in list(self._sessions.values()):
             await session.close()
         self._sessions.clear()
diff --git a/packages/meshbay-node/tests/test_peer_session_limits.py b/packages/meshbay-node/tests/test_peer_session_limits.py
new file mode 100644
index 0000000..64961ad
--- /dev/null
+++ b/packages/meshbay-node/tests/test_peer_session_limits.py
@@ -0,0 +1,136 @@
+"""
+Availability, on the node: what the people you invited can cost you.
+
+The message budget for an unauthenticated peer (`PRE_HANDSHAKE_MAX_MSG`, H6)
+bounds what *one* of them spends. Nothing bounded how many there could be, or
+how long one could stay without ever proving anything — and the hub's cap is
+three offers in flight **per account**, which is a limit on each caller rather
+than on this machine, so an operator's exposure grew with the number of people
+in their groups. A member who never meant any harm — a tab left open through a
+sleep, a client retrying a connection it cannot complete — arrives here the
+same way as one who does.
+"""
+
+import asyncio
+from unittest.mock import MagicMock
+
+import pytest
+
+from meshbay_node.transport import webrtc_server as ws_mod
+from meshbay_node.transport.webrtc_server import (
+    MAX_PEER_SESSIONS, UNAUTHENTICATED_SESSION_TIMEOUT, WebRTCTransport)
+
+
+def _transport() -> WebRTCTransport:
+    return WebRTCTransport(sk_node=MagicMock(), hub_pk_pem=b"", gek=None,
+                           roots=None, index=None)
+
+
+class _FakeSession:
+    """Stands in for a peer that connected and then said nothing."""
+
+    def __init__(self, user_id=None):
+        self._user_id = user_id
+        self.closed = False
+
+    async def close(self):
+        self.closed = True
+
+
+@pytest.mark.asyncio
+async def test_the_node_refuses_more_peers_than_it_will_hold():
+    """The refusal comes before an RTCPeerConnection is allocated, or the cap
+    would be counting the thing it is meant to prevent."""
+    tp = _transport()
+    for i in range(MAX_PEER_SESSIONS):
+        tp._sessions[f"peer-{i}"] = _FakeSession(user_id="someone")
+
+    with pytest.raises(RuntimeError, match="peer-connection limit"):
+        await tp.handle_offer("v=0", "one-too-many")
+
+    assert "one-too-many" not in tp._sessions
+
+
+@pytest.mark.asyncio
+async def test_a_peer_that_never_handshakes_is_closed(monkeypatch):
+    """`connectionstatechange` reaps a connection that *fails*. One that
+    succeeds and stays silent was held for the life of the daemon."""
+    monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
+    tp = _transport()
+    session = _FakeSession(user_id=None)
+    tp._sessions["quiet"] = session
+
+    tp._reap_if_unauthenticated("quiet")
+    await asyncio.sleep(0.2)
+
+    assert session.closed, "a peer that never proved anything was kept"
+    assert "quiet" not in tp._sessions
+
+
+@pytest.mark.asyncio
+async def test_a_peer_that_handshaked_is_left_alone(monkeypatch):
+    """`_user_id` is set by the GEK proof, and is the one honest test of
+    whether this peer ever became anybody."""
+    monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
+    tp = _transport()
+    session = _FakeSession(user_id=None)
+    tp._sessions["real"] = session
+
+    tp._reap_if_unauthenticated("real")
+    session._user_id = "a-real-member"        # the handshake completes
+    await asyncio.sleep(0.2)
+
+    assert not session.closed, "a member who completed the handshake was cut off"
+    assert tp._sessions["real"] is session
+
+
+@pytest.mark.asyncio
+async def test_handle_offer_arms_the_reaper(monkeypatch):
+    """
+    The seam, driven rather than described. The three tests above call
+    `_reap_if_unauthenticated` themselves, so every one of them would still
+    pass with the call removed from `handle_offer` and no peer reaped at all —
+    which is the defect, not the helper. aiortc is stubbed because a real
+    RTCPeerConnection wants a real SDP; everything else here is the shipped
+    code path.
+    """
+    class _FakePC:
+        def __init__(self, *a, **kw):
+            self.localDescription = MagicMock(sdp="v=0\r\n")
+            self.remoteDescription = None
+
+        def on(self, _event):
+            return lambda fn: fn
+
+        async def setRemoteDescription(self, _d): pass
+        async def createAnswer(self): return MagicMock()
+        async def setLocalDescription(self, _d): pass
+        async def close(self): pass
+
+    monkeypatch.setattr(ws_mod, "RTCPeerConnection", _FakePC)
+    monkeypatch.setattr(ws_mod, "RTCSessionDescription",
+                        lambda **kw: MagicMock(**kw))
+
+    tp = _transport()
+    await tp.handle_offer("v=0\r\n", "fresh-peer")
+    try:
+        assert tp._reapers, (
+            "handle_offer allocated a peer session and armed nothing to "
+            "close it if the handshake never comes")
+    finally:
+        await tp.close_all()
+
+
+@pytest.mark.asyncio
+async def test_the_reaper_is_held_and_cancelled_with_the_transport():
+    """asyncio keeps only a weak reference to a task, and a reaper collected
+    mid-sleep reaps nothing — the same trap as every other task in this file."""
+    tp = _transport()
+    tp._sessions["held"] = _FakeSession()
+    tp._reap_if_unauthenticated("held")
+    assert tp._reapers, "the reaper was fired and forgotten"
+
+    await tp.close_all()
+    assert not tp._reapers
+    # The timeout is a real duration, not something a test has to wait out.
+    assert UNAUTHENTICATED_SESSION_TIMEOUT >= 30