path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py

"""
MeshBay Node — WebRTC DataChannel server for browser clients.

Browsers cannot use QUIC for NAT traversal (WebTransport doesn't allow choosing
the UDP source port — Port-Restricted Cone NAT requires exact port matching).
WebRTC DataChannel with ICE/STUN handles this automatically.

The MNP protocol (handshake, file_request, file_chunk, chat, etc.) runs
identically over WebRTC DataChannel as over QUIC streams. Same E2E encryption,
same message types, same msgpack wire format.

Wire format on the DataChannel:
  - Each message is length-prefixed msgpack (4-byte big-endian + msgpack payload)
  - Same as QUIC streams and TCP+TLS
  - DataChannel is ordered and reliable (SCTP over DTLS)

Signaling flow (handled externally by the hub):
  Browser → Hub : POST /v1/nodes/{id}/webrtc/offer  {sdp, ice_candidates}
  Hub → Node    : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id}
  Node → Hub    : WS push {type: "webrtc_answer", sdp, ice_candidates, peer_id}
  Hub → Browser : SSE/response {sdp, ice_candidates}
  After signaling, DataChannel is P2P — hub is out of the loop.
"""

import asyncio
import logging
import time
from typing import Any

from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
)
from meshbay_common.protocol import (
    MNP,
)

from meshbay_node import transfers as transfers_mod
from meshbay_node.indexer import GroupIndex

# Re-imported under its original name: every call site and existing test in
# this module still refers to it as `_probe_video`. The implementation lives
# in media_probe.py so the indexer package (imported just above) can call it
# too, for index-time enrichment, without a circular import.
from meshbay_node.roots import (
    RootSet,
)
from meshbay_node.transport.webrtc.admin import AdminMixin
from meshbay_node.transport.webrtc.admission import AdmissionMixin
from meshbay_node.transport.webrtc.apps.music import MusicMixin
from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin
from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin
from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin
from meshbay_node.transport.webrtc.blobs import BlobsMixin
from meshbay_node.transport.webrtc.chat import ChatMixin
from meshbay_node.transport.webrtc.core import _WEBRTC_TRACE, SessionCore
from meshbay_node.transport.webrtc.files import FilesMixin
from meshbay_node.transport.webrtc.group_ops import GroupOpsMixin
from meshbay_node.transport.webrtc.handshake import HandshakeMixin
from meshbay_node.transport.webrtc.node_ops import NodeOpsMixin
from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin
from meshbay_node.transport.webrtc.upload_handlers import UploadMixin

log = logging.getLogger(__name__)


# 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 meters offers
# per account (signaling.py) — 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

# Bundle fetches are served in the pre-proof window (C4). Bounded and audited
# until the native client removes remote keypair bundles entirely.
MAX_PRE_PROOF_FETCHES = 4


class WebRTCPeerSession(
    AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin,
    HandshakeMixin, NodeOpsMixin, TransferMixin, UploadMixin,
    StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
    SessionCore,
):
    """One WebRTC peer connection, handling MNP over a DataChannel."""

    def _dispatch_message(self, msg: dict) -> None:
        mtype = msg.get("type")
        log.debug("WebRTC recv: %s", mtype)
        try:
            if mtype == MNP.HANDSHAKE:
                self._do_handshake(msg)
            elif mtype == MNP.HANDSHAKE_RESPONSE:
                self._do_handshake_response(msg)
            elif mtype in (MNP.GEK_BUNDLE_FETCH, MNP.KEYPAIR_BUNDLE_FETCH) \
                    and self._gek_challenge is not None:
                # Served before the GEK proof by necessity: the client needs its
                # wrapped bundle in order to compute the proof. That window is a
                # disclosure surface (C4) — a hub that forges a JWT reaches it — so
                # it is bounded and audited here, and closed properly when clients
                # stop storing keypair bundles on other people's nodes.
                self._pre_proof_fetches += 1
                if self._pre_proof_fetches > MAX_PRE_PROOF_FETCHES:
                    self._audit_auth_failed(
                        getattr(self, "_pending_group", ""), "pre-proof fetch flood")
                    self._send({"type": "error", "detail": "Too many requests"})
                    return
                self._audit_pre_proof_fetch(mtype)
                if mtype == MNP.GEK_BUNDLE_FETCH:
                    self._spawn(self._do_gek_bundle_fetch())
                else:
                    self._spawn(self._do_keypair_bundle_fetch())
            elif mtype == MNP.JOIN_REQUEST and self._nonce_node:
                # Valid both before the GEK proof (a new member has no GEK to prove
                # with) and after it (an operator pairing a browser is already
                # connected). Authority comes from the pairing code and the
                # signature, never from the session state.
                self._spawn(self._do_join_request(msg))
            elif self._user_id is None:
                self._send({"type": "error", "detail": "Handshake required"})
            elif mtype == MNP.INDEX_SYNC:
                self._do_index_sync()
            elif mtype == MNP.FILE_REQUEST:
                # Spawned rather than answered inline: the reply waits for room
                # on the channel, and blocking the message loop for that would
                # stop everything else this peer is doing — including the
                # uploads whose acks free the very buffer we are waiting on.
                # Chunks are matched by file and index on the client, so
                # answering out of order is safe.
                self._spawn(self._do_file_request(msg))
            elif mtype == MNP.CHAT_MESSAGE:
                self._do_chat_message(msg)
            elif mtype == MNP.CHAT_HISTORY:
                self._do_chat_history(msg)
            elif mtype == MNP.LINK_PREVIEW_REQ:
                self._spawn(self._do_link_preview_request(msg))
            elif mtype == MNP.PING:
                self._do_ping(msg)
            elif mtype == MNP.TRANSFER_OPEN:
                self._do_transfer_open(msg)
            elif mtype == MNP.TRANSFER_CLOSE:
                self._do_transfer_close(msg)
            elif mtype == MNP.FILE_UPLOAD:
                self._spawn(self._do_file_upload(msg))
            elif mtype == MNP.DIR_CREATE:
                self._spawn(self._do_dir_create(msg))
            elif mtype == MNP.DIR_DELETE:
                self._spawn(self._do_dir_delete(msg))
            elif mtype == MNP.FILE_DELETE:
                self._do_file_delete(msg)
            elif mtype == MNP.ADMIN_RESPONSE:
                self._do_admin_response(msg)
            elif mtype == MNP.INVITE_CREATE:
                self._do_invite_create(msg)
            elif mtype == MNP.INVITE_LINK_CREATE:
                self._do_invite_link_create(msg)
            elif mtype == MNP.INVITE_CANCEL:
                self._do_invite_cancel(msg)
            elif mtype == MNP.MEMBER_REVOKE:
                self._do_member_revoke(msg)
            elif mtype == MNP.DEVICE_REQUEST:
                self._spawn(self._do_device_request(msg))
            elif mtype == MNP.DEVICE_LOOKUP:
                self._spawn(self._do_device_lookup(msg))
            elif mtype == MNP.DEVICE_ADD:
                self._spawn(self._do_device_add(msg))
            elif mtype == MNP.DEVICE_LIST:
                self._spawn(self._do_device_list(msg))
            elif mtype == MNP.DEVICE_REVOKE:
                self._spawn(self._do_device_revoke(msg))
            elif mtype == MNP.DEVICE_HELLO:
                self._spawn(self._do_device_hello(msg))
            elif mtype == MNP.APPS_ENABLED:
                self._do_apps_enabled(msg)
            elif mtype == MNP.TRANSFER_LIMITS:
                self._do_transfer_limits(msg)
            elif mtype == MNP.SET_SCAN_SETTINGS:
                self._do_set_scan_settings(msg)
            elif mtype == MNP.TMDB_CONFIG:
                self._do_tmdb_config(msg)
            elif mtype == MNP.TMDB_ENABLED:
                self._do_tmdb_enabled(msg)
            elif mtype == MNP.APP_DIRECTORIES:
                self._do_app_directories(msg)
            elif mtype == MNP.CHAT_DIRECTORY:
                self._do_chat_directory(msg)
            elif mtype == MNP.CHAT_LINK_PREVIEW:
                self._do_chat_link_preview(msg)
            elif mtype == MNP.SEARCH_LISTED:
                self._do_search_listed(msg)
            elif mtype == MNP.CHAT_EPOCH:
                self._do_chat_epoch(msg)
            elif mtype == MNP.CHAT_KEYS_REQ:
                self._spawn(self._do_chat_keys_req(msg))
            elif mtype == MNP.GROUP_ROSTER_REQ:
                self._spawn(self._do_group_roster_req(msg))
            elif mtype == MNP.MEDIA_META_REQ:
                self._spawn(self._do_media_meta_request(msg))
            elif mtype == MNP.SEASON_META_REQ:
                self._spawn(self._do_season_meta_request(msg))
            elif mtype == MNP.TMDB_SEARCH_REQ:
                self._spawn(self._do_tmdb_search_request(msg))
            elif mtype == MNP.TMDB_OVERRIDE:
                self._do_tmdb_override(msg)
            elif mtype == MNP.TMDB_REMATCH:
                self._do_tmdb_rematch(msg)
            elif mtype == MNP.MUSICBRAINZ_ENABLED:
                self._do_musicbrainz_enabled(msg)
            elif mtype == MNP.MUSIC_META_REQ:
                self._spawn(self._do_music_meta_request(msg))
            elif mtype == MNP.AUDIO_TRANSCODE_REQ:
                self._spawn(self._do_audio_transcode_request(msg))
            elif mtype == MNP.SUBTITLE_REQ:
                self._spawn(self._do_subtitle_request(msg))
            elif mtype == MNP.MEMBER_UNPIN:
                self._do_member_unpin(msg)
            elif mtype == MNP.GEK_ROTATE:
                self._do_gek_rotate(msg)
            elif mtype == MNP.NODE_STATUS:
                self._spawn(self._do_node_status(msg))
            elif mtype == MNP.ROOT_ADD:
                self._do_root_add(msg)
            elif mtype == MNP.ROOT_REMOVE:
                self._do_root_remove(msg)
            elif mtype == MNP.ROOT_UPDATE:
                self._do_root_update(msg)
            elif mtype == MNP.ROOT_EJECT:
                self._do_root_eject(msg)
            elif mtype == MNP.ROOT_PLUG:
                self._do_root_plug(msg)
            elif mtype == MNP.ROSTER_READ:
                self._spawn(self._do_roster_read(msg))
            elif mtype == MNP.DENYLIST_READ:
                self._spawn(self._do_denylist_read(msg))
            elif mtype == MNP.DENYLIST_CLEAR:
                self._spawn(self._do_denylist_clear(msg))
            elif mtype == MNP.GROUP_ATTACH:
                self._do_group_attach(msg)
            elif mtype == MNP.GROUP_DETACH:
                self._do_group_detach(msg)
            elif mtype == MNP.NODE_SETTINGS_SET:
                self._spawn(self._do_node_settings_set(msg))
            elif mtype == MNP.NODE_RELOAD:
                self._spawn(self._do_node_reload(msg))
            elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
                self._spawn(self._do_keypair_bundle_store(msg))
            elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
                self._spawn(self._do_keypair_bundle_delete())
            elif mtype == MNP.USER_BLOB_STORE:
                self._spawn(self._do_user_blob_store(msg))
            elif mtype == MNP.USER_BLOB_FETCH:
                self._spawn(self._do_user_blob_fetch(msg))
            elif mtype == MNP.USER_BLOB_LIST:
                self._spawn(self._do_user_blob_list())
            elif mtype == MNP.USER_BLOB_DELETE:
                self._spawn(self._do_user_blob_delete(msg))
            elif mtype == MNP.STREAM_REQUEST:
                sem = self._ctx.get("_transcode_sem")
                log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
                         str(msg.get("file_id"))[:12], msg.get("credits"),
                         getattr(sem, "_value", "?"),
                         "alive" if (self._stream_task and
                                     not self._stream_task.done()) else "none")
                self._spawn(self._replace_stream(msg))
            elif mtype == MNP.STREAM_MORE:
                self._grant_stream_credit(msg)
            elif mtype == "client_diag":
                # Diagnostics only. The node acts on none of it — it writes it
                # next to its own view of the same stream, which is the only
                # place the two halves can be compared when the client is a
                # phone with no console.
                # Every field is peer-controlled, so each is stringified and
                # cut short: this is a log line, not a channel for writing
                # whatever one likes into the operator's file.
                def _f(key: str, n: int = 24) -> str:
                    return str(msg.get(key))[:n].replace("\n", " ")
                if msg.get("event"):
                    # Once per stream or per seek, not once per five seconds —
                    # and a seek nobody asked for looks exactly like a viewer
                    # dragging the scrubber from this side, so it has to be
                    # visible without turning DEBUG on.
                    log.info(
                        "stream: client %s target=%s t=%ss offset=%s ready=%s "
                        "duration=%s ranges=[%s]",
                        _f("event", 16), _f("target"), _f("t"), _f("offset"),
                        _f("ready"), _f("duration"), _f("ranges", 120))
                # Debug: one line every five seconds per viewer. Run the daemon
                # with --log-level debug to see inside a player that is
                # misbehaving — it is the only view of the browser there is
                # when the browser is a phone.
                else:
                    # `ahead` on its own cannot say whether a short buffer is
                    # the player's own gate holding or the network failing to
                    # keep up, and those two want opposite answers. `limit` is
                    # what the gate is set to for this film and `budget` the
                    # byte budget it was derived from, so the three read as one
                    # sentence.
                    log.debug(
                        "stream: client t=%ss ahead=%ss/%ss budget=%sMB "
                        "ready=%s paused=%s "
                        "stalled=%s q=%s inflight=%s appending=%s updating=%s "
                        "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)",
                        _f("t"), _f("ahead"), _f("limit"), _f("budgetMB"),
                        _f("ready"), _f("paused"),
                        _f("stalled"), _f("q"), _f("inflight"), _f("appending"),
                        _f("updating"), _f("quota"), _f("ms"), _f("err", 80),
                        _f("ranges", 120), self._stream_segments)
            elif mtype == MNP.STREAM_STOP:
                age = (time.monotonic() - self._stream_started_at
                       if self._stream_started_at else -1)
                log.info("stream: stop received %.1fs after start, %d segments sent",
                         age, self._stream_segments)
                self._stop_stream()
            else:
                log.warning("Unknown MNP message type on DataChannel: %s", mtype)
        except Exception as e:
            # Log the detail locally; send the peer a generic message. Exception
            # text here carries filesystem paths and internal state (finding L3).
            log.error("Error handling %s on DataChannel: %s", mtype, e, exc_info=True)
            self._send({"type": "error", "detail": "Request failed"})


class WebRTCTransport:
    """
    Manages WebRTC peer connections for browser clients.

    Usage:
        transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, index)
        answer_sdp = await transport.handle_offer(offer_sdp, peer_id)
        # Return answer_sdp to the browser via hub signaling
    """

    def __init__(
        self,
        sk_node: Ed25519PrivateKey,
        hub_pk_pem: bytes,
        gek: bytes,
        roots: RootSet,
        index: GroupIndex,
        groups: dict[str, dict] | None = None,
        denylist: Any | None = None,
        stun_servers: list[str] | None = None,
        max_concurrent_streams: int | None = None,
        max_concurrent_downloads: int | None = None,
        max_concurrent_uploads: int | None = None,
        max_upload_gb: float | None = None,
        transcode_incompatible_video: bool = True,
    ):
        self._ctx: dict[str, Any] = {
            "sk_node": sk_node,
            "hub_pk_pem": hub_pk_pem,
            "gek": gek,
            "roots": roots,
            "index": index,
            "_peers": {},
            # None means "the operator said nothing" — the default applies. It
            # is read once, when the first stream builds the semaphore.
            "max_concurrent_streams": max_concurrent_streams,
            # Read once, when the first transfer builds the pools. None means
            # the operator said nothing and transfers.py's defaults apply.
            "max_concurrent_downloads": max_concurrent_downloads,
            "max_concurrent_uploads": max_concurrent_uploads,
            # The per-file upload ceiling, in GB. None means the operator said
            # nothing and MAX_UPLOAD_BYTES stands.
            "max_upload_gb": max_upload_gb,
            # Operator opt-out (node.toml) for the HEVC-etc. transcode
            # fallback in _stream_video_inner — real CPU cost, unlike copy.
            "transcode_incompatible_video": transcode_incompatible_video,
        }
        if groups:
            self._ctx["groups"] = groups
        if denylist:
            self._ctx["denylist"] = denylist
        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,
                     max_concurrent_uploads: int | None = None,
                     max_upload_gb: float | None = None) -> dict:
        """Resize a live pool without restarting the daemon.

        `ops.set_node_settings` used to do this by assigning
        `webrtc._stream_sem`, an attribute that has never existed — the pool is
        `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always
        False. So the hot-swap was a no-op and **`max_concurrent_streams` has
        never taken effect from the Node page without a restart**, contrary to
        docs/MESHBAY_DESIGN.md §6.8. This is the one implementation, on the
        object that owns the state, so the next two caps do not each grow their
        own copy of the mistake.

        What resizing means, stated because it is a decision and not a
        detail: **the new cap governs new streams; the ones already running are
        never interrupted.** A slot is held for the length of a film, so
        lowering the cap below what is in flight cannot take a viewer's film
        away — it stops the next one starting. The replacement pool is therefore
        created with the permits that remain (`new - in_flight`, floored at
        zero), not with a full set, or lowering the cap would briefly allow more
        viewers than either the old value or the new one.
        """
        changed: dict = {}
        if max_concurrent_streams is not None:
            n = int(max_concurrent_streams)
            if n < 1:
                raise ValueError("max_concurrent_streams must be positive")
            before = self._ctx.get("max_concurrent_streams")
            self._ctx["max_concurrent_streams"] = n
            if self._ctx.get("_transcode_sem") is not None:
                in_flight = self._ctx.get("_streams_in_flight", 0)
                self._ctx["_transcode_sem"] = asyncio.Semaphore(
                    max(0, n - in_flight))
                log.info("stream: capacity %s -> %d (%d in flight, %d free now)",
                         before, n, in_flight, max(0, n - in_flight))
            else:
                # Nothing has streamed yet; the pool is built from this value on
                # first use, so there is nothing to resize.
                log.info("stream: capacity %s -> %d (no pool built yet)",
                         before, n)
            changed["max_concurrent_streams"] = n

        pools = {}
        if max_concurrent_downloads is not None:
            pools[transfers_mod.DOWNLOAD] = int(max_concurrent_downloads)
        if max_concurrent_uploads is not None:
            pools[transfers_mod.UPLOAD] = int(max_concurrent_uploads)
        for key, value in pools.items():
            if value < 1:
                raise ValueError(f"max_concurrent_{key}s must be positive")
        if pools:
            # Kept on the context whether or not a pool exists yet: the pools
            # are built on the first transfer, and would otherwise come up with
            # the defaults after an operator had already changed them.
            for key, value in pools.items():
                self._ctx[f"max_concurrent_{key}s"] = value
                changed[f"max_concurrent_{key}s"] = value
            slots = self._ctx.get("_transfer_slots")
            if slots is not None:
                granted = slots.set_caps(node=pools)
                log.info("transfer: capacity now %s (%d started at once)",
                         slots.summary(), len(granted))
                # Raising a cap can start queued transfers immediately, and the
                # peers waiting on them have to be told: a grant nobody hears
                # about is the "stuck at waiting" report this design exists to
                # prevent.
                for lease in granted:
                    self._notify_granted(lease)

        if max_upload_gb is not None:
            gb = float(max_upload_gb)
            if gb <= 0:
                raise ValueError("max_upload_gb must be greater than zero")
            self._ctx["max_upload_gb"] = gb
            changed["max_upload_gb"] = gb
            log.info("upload: per-file ceiling now %g GB", gb)
        return changed

    def _notify_granted(self, lease) -> None:
        """Tell the connection that owns `lease` it may start.

        On the transport rather than the session because a cap change has no
        session behind it — it arrives from the loopback API.
        """
        groups = self._ctx.get("groups")
        registries = ([g.get("_peers", {}) for g in groups.values()]
                      if groups else [self._ctx.get("_peers", {})])
        for reg in registries:
            session = reg.get(lease.session_key)
            if session is not None:
                try:
                    session._send(
                        session._transfer_state_msg(lease, "granted"))
                except Exception:
                    pass
                return

    async def handle_offer(
        self, offer_sdp: str, peer_id: str,
    ) -> tuple[str, list[dict]]:
        """
        Process a WebRTC SDP offer from a browser client.

        Returns (answer_sdp, ice_candidates) to relay back via hub signaling.
        ICE candidates are embedded in the SDP (aiortc gathers before returning).
        """
        from aiortc import RTCConfiguration, RTCIceServer

        # aiortc keeps only the first STUN entry it sees here; the actual
        # multi-server fan-out is done by transport/stun_multi, which patches
        # aioice. The full list is still passed so a one-server deploy and the
        # tests that read `_stun` stay coherent.
        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 meters offers *per account*
        # (signaling.py), 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):
            log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id)
            session._setup_channel(channel)

        if _WEBRTC_TRACE:
            @pc.on("iceconnectionstatechange")
            def on_ice_state_change():
                log.info("WebRTC ICE state: %s (peer=%s)", pc.iceConnectionState, peer_id)

        @pc.on("connectionstatechange")
        async def on_state_change():
            state = pc.connectionState
            log.info("WebRTC connection state: %s (peer=%s)", state, peer_id)
            if state in ("failed", "closed"):
                gone = self._sessions.pop(peer_id, None)
                if gone is not None:
                    # Popping only forgets the session. Its stream went on
                    # transcoding until the credit timeout — measured at 91s
                    # after the connection closed — holding one of the node's
                    # two slots the whole time. Closing the viewer, the tab or
                    # the browser all arrive here, so this is the one place
                    # that covers every way of walking away.
                    #
                    # And the group's peer set forgets it too, as close() does:
                    # otherwise every later broadcast to the group is written
                    # to a closed channel, and every reconnect leaves one more
                    # dead session held until the node restarts.
                    if gone._user_id:
                        gone._unregister_peer()
                    await gone.shutdown_tasks()

        offer = RTCSessionDescription(sdp=offer_sdp, type="offer")
        await pc.setRemoteDescription(offer)
        answer = await pc.createAnswer()
        gather_start = time.monotonic()
        await pc.setLocalDescription(answer)

        # ICE gathering runs inside setLocalDescription (non-trickle). A slow or
        # unreachable STUN server shows up here as seconds of wait and zero
        # srflx lines — the symptom the multi-server fan-out exists to prevent.
        answer_sdp = pc.localDescription.sdp
        srflx = answer_sdp.count(" typ srflx")
        # The host addresses this node put in the answer. When a peer reports
        # "DataChannel closed" the first question is whether the node offered
        # anything that peer could route to at all — on a NAT'd host or a VM the
        # only host candidate is an address no one else can reach, and the log
        # otherwise looks identical to a working connection.
        host_addrs: set[str] = set()
        for line in answer_sdp.splitlines():
            if line.startswith("a=candidate:") and " typ host " in line:
                parts = line.split()
                if len(parts) > 5:
                    host_addrs.add(parts[4])
        log.info(
            "WebRTC answer ready for peer=%s (ICE gather %.2fs, host: %s, %d srflx)",
            peer_id, time.monotonic() - gather_start,
            ", ".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()

    @property
    def active_peers(self) -> int:
        return len(self._sessions)