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 base64
import hashlib
import hmac
import logging
import os
import re
import struct
import time
from pathlib import Path
from typing import Any

import jwt
import msgpack
from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)

from meshbay_common import MNP_VERSION
from meshbay_common.handshake import (
    NONCE_LEN,
    ROLE_CLIENT,
    ROLE_NODE,
    HandshakeError,
    authorize_token,
    handshake_transcript,
    make_proof,
    verify_proof,
    webrtc_binding,
)
from meshbay_common.adminop import (
    ADMIN_CHALLENGE_TTL,
    OP_DIR_DELETE,
    OP_FILE_DELETE,
    OP_INVITE_CREATE,
    admin_transcript,
)
from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
from meshbay_common.join import (
    JOIN_TTL,
    ROLE_MEMBER,
    ROLE_OPERATOR,
    join_transcript,
)
from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
from meshbay_common.protocol import MNP
from meshbay_node.indexer import GroupIndex
from meshbay_node.roster import DEFAULT_INVITE_TTL

log = logging.getLogger(__name__)

CHUNK_SIZE = 1024 * 1024
MAX_MSG = 64 * 1024 * 1024

# Upload limits (finding C5a). Uploads used to land directly in the shared root under
# a name the client chose, overwriting whatever was already there — which both violated
# node sovereignty and defeated the delete authorization (overwrite a file, become its
# recorded uploader, then delete it legitimately).
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
# ffmpeg is spawned per stream request; without a cap any member can fork-bomb
# the node by requesting many streams at once (H6).
MAX_CONCURRENT_TRANSCODES = 2
# 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
# Pairing codes carry 40 bits and are single-use, but a connection must not be
# allowed to sit there guessing. Failures are audited, so a grind is visible.
MAX_JOIN_ATTEMPTS = 5
# Per-connection limits alone would not bind an attacker who can open connections
# at will — and the adversary who can mint tokens for any account is the hub. So
# failed pairings are also counted node-wide over a window.
MAX_JOIN_FAILURES_WINDOW = 20
JOIN_FAILURE_WINDOW = 600  # seconds
# Everything a member sends lands here: files from the Files panel and
# attachments from the chat alike. One visible directory the operator can look
# into, back up or empty — rather than a hidden tree of per-user uuids that
# nobody could read, or files scattered wherever someone happened to be looking.
UPLOAD_DIR_NAME = "uploads"
# Conservative allowlist: also what keeps markup out of filenames, which the node admin
# UI used to render unescaped (finding H2).
SAFE_UPLOAD_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._ -]{0,127}$")


def _free_name(directory: Path, filename: str) -> str:
    """
    `filename`, or the first "name (n).ext" that is not taken.

    Never returns the name of a file that exists, so an upload cannot replace
    one — the property the per-user quarantine used to provide (C5a).
    """
    if not (directory / filename).exists():
        return filename
    stem, dot, ext = filename.rpartition(".")
    if not dot:
        stem, ext = filename, ""
    for n in range(2, 1000):
        candidate = f"{stem} ({n}){dot}{ext}"
        if not (directory / candidate).exists():
            return candidate
    raise FileExistsError(filename)


def safe_subdir(shared_root: Path, rel: str) -> Path | None:
    """
    Resolve a client-supplied directory under the shared root, or refuse.

    Uploads land where the member is looking now rather than in a per-user
    quarantine, so the path arrives from the wire and every part of it has to be
    checked: each segment against the same allowlist as filenames, and the
    resolved result against the root. `..`, absolute paths, symlinks pointing
    out, and anything with a separator in a segment are all refused here rather
    than in the caller, so there is one place to get it right.

    The quarantine was the fix for C5a; what actually mattered in it — no
    overwrite, a name allowlist, and confinement — is kept by this plus the
    caller's existing checks.
    """
    rel = (rel or "").strip().strip("/")
    if not rel:
        return shared_root
    parts = [seg for seg in rel.split("/") if seg not in ("", ".")]
    if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts):
        return None
    try:
        target = (shared_root / Path(*parts)).resolve()
        root = shared_root.resolve()
    except OSError:
        return None
    if target != root and root not in target.parents:
        return None
    return target


def _extract_dtls_fingerprint(sdp: str) -> bytes:
    """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
    for line in sdp.splitlines():
        if line.startswith("a=fingerprint:sha-256 "):
            hex_str = line.split(" ", 1)[1].replace(":", "")
            return bytes.fromhex(hex_str)
    return b""


STREAM_SEGMENT_SIZE = 256 * 1024

_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}


async def _probe_video(path: str) -> tuple[str | None, float]:
    """Probe video file with ffprobe, return (MSE codec string, duration)."""
    import json as _json
    proc = await asyncio.create_subprocess_exec(
        "ffprobe", "-v", "error",
        "-show_entries", "stream=codec_name,profile,level,codec_type",
        "-show_entries", "format=duration",
        "-of", "json", path,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    stdout, _ = await proc.communicate()
    info = _json.loads(stdout)
    duration = float(info.get("format", {}).get("duration", 0))

    v_codec = a_codec = ""
    for s in info.get("streams", []):
        if s.get("codec_type") == "video" and not v_codec:
            cn = s.get("codec_name", "")
            if cn == "h264":
                p = _H264_PROFILES.get(s.get("profile", "High"), "64")
                lvl = int(s.get("level", 40))
                v_codec = f"avc1.{p}00{lvl:02x}"
            elif cn == "hevc":
                v_codec = "hev1.1.6.L93.B0"
            elif cn == "vp9":
                v_codec = "vp09.00.10.08"
            elif cn == "av1":
                v_codec = "av01.0.01M.08"
        elif s.get("codec_type") == "audio" and not a_codec:
            cn = s.get("codec_name", "")
            if cn == "aac":
                a_codec = "mp4a.40.2"
            elif cn in ("mp3", "mp2"):
                a_codec = "mp4a.6b"
            elif cn == "opus":
                a_codec = "opus"
            elif cn == "ac3":
                a_codec = "ac-3"
            elif cn == "flac":
                a_codec = "flac"

    if not v_codec:
        return None, duration
    codec = f"{v_codec},{a_codec}" if a_codec else v_codec
    return codec, duration


def _pack(obj: dict) -> bytes:
    data = msgpack.packb(obj, use_bin_type=True)
    return struct.pack(">I", len(data)) + data


class _DataChannelBuffer:
    """
    Accumulate DataChannel messages and extract length-prefixed msgpack.

    Finding H6: the limit was a flat 64 MB applied even before the handshake, so an
    unauthenticated peer could announce a 64 MB frame and dribble bytes into it,
    holding that much memory per connection. Until a peer has proved GEK
    possession it gets a small budget; the large one is for file uploads.
    """

    def __init__(self, max_message: int = MAX_MSG):
        self._buf = bytearray()
        self.max_message = max_message

    def feed(self, data: bytes):
        self._buf.extend(data)

    def messages(self):
        while len(self._buf) >= 4:
            length = struct.unpack(">I", self._buf[:4])[0]
            if length > self.max_message:
                raise ValueError(f"Message too large: {length}")
            if len(self._buf) < 4 + length:
                break
            msg_bytes = bytes(self._buf[4:4 + length])
            del self._buf[:4 + length]
            yield msgpack.unpackb(msg_bytes, raw=False)


def _get_remote_ip(pc: RTCPeerConnection) -> str:
    """Best-effort extraction of the remote peer IP from the ICE transport."""
    try:
        dtls = pc.sctp and pc.sctp.transport
        ice = dtls and dtls.transport
        conn = ice and ice._connection
        if conn and hasattr(conn, '_nominated') and conn._nominated:
            for pair in conn._nominated.values():
                return pair.remote_candidate.host
        if conn and conn.remote_candidates:
            return conn.remote_candidates[0].host
    except Exception:
        pass
    return ""


class WebRTCPeerSession:
    """One WebRTC peer connection, handling MNP over a DataChannel."""

    def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""):
        self._pc = pc
        self._ctx = node_ctx
        self._channel: RTCDataChannel | None = None
        self._buffer = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
        self._pre_proof_fetches = 0
        self._user_id: str | None = None
        self._group_id: str | None = None
        self._peer_id: str = peer_id
        self._remote_ip: str = ""
        self._username: str = ""
        # Set from the roster: the key this node pinned for this account. Never
        # from the JWT — the hub picks what goes in there.
        self._pinned_pk: str = ""
        self._gek_challenge: bytes | None = None
        # Same value as the GEK challenge, but kept for the life of the connection:
        # a join_request is signed over it, and it must stay verifiable after the
        # handshake clears the challenge (an operator pairs while already connected).
        self._nonce_node: bytes = b""
        self._join_attempts = 0
        self._nonce_client: bytes = b""
        self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
        self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}

    def _setup_channel(self, channel: RTCDataChannel) -> None:
        self._channel = channel

        @channel.on("message")
        def on_message(message):
            if isinstance(message, str):
                message = message.encode()
            self._buffer.feed(message)
            for msg in self._buffer.messages():
                self._handle_message(msg)

    def _handle_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:
                    asyncio.ensure_future(self._do_gek_bundle_fetch())
                else:
                    asyncio.ensure_future(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.
                asyncio.ensure_future(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:
                self._do_file_request(msg)
            elif mtype == MNP.STREAM_SEGMENT:
                self._do_stream_segment(msg)
            elif mtype == MNP.CHAT_MESSAGE:
                self._do_chat_message(msg)
            elif mtype == MNP.CHAT_HISTORY:
                self._do_chat_history(msg)
            elif mtype == MNP.FILE_UPLOAD:
                self._do_file_upload(msg)
            elif mtype == MNP.DIR_CREATE:
                self._do_dir_create(msg)
            elif mtype == MNP.DIR_DELETE:
                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.KEYPAIR_BUNDLE_STORE:
                asyncio.ensure_future(self._do_keypair_bundle_store(msg))
            elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
                asyncio.ensure_future(self._do_keypair_bundle_delete())
            elif mtype == MNP.STREAM_REQUEST:
                asyncio.ensure_future(self._stream_video(msg))
            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"})

    def _audit(self, event: str, detail: str = "") -> None:
        audit = self._ctx.get("audit_store")
        if audit and self._user_id:
            if not self._remote_ip:
                self._remote_ip = _get_remote_ip(self._pc)
            asyncio.ensure_future(audit.log_event(
                user_id=self._user_id,
                event=event,
                ip=self._remote_ip,
                username=self._username,
                group_id=self._group_id or "",
                detail=detail,
            ))

    def _channel_binding(self) -> bytes:
        """Both DTLS fingerprints, so a proof is valid on this connection only."""
        offer_fp = b""
        answer_fp = b""
        if self._pc.remoteDescription:
            offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp)
        if self._pc.localDescription:
            answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp)
        if not offer_fp or not answer_fp:
            return b""
        return webrtc_binding(offer_fp, answer_fp)

    def _do_handshake(self, msg: dict) -> None:
        group_id = msg.get("group_id", "")
        try:
            peer = authorize_token(
                msg.get("token", ""),
                self._ctx["hub_pk_pem"],
                group_id=group_id,
                hosted_groups=self._ctx.get("groups"),
                denylist=self._ctx.get("denylist"),
            )
        except HandshakeError as refusal:
            # HandshakeError messages are authored to be peer-safe, unlike arbitrary
            # exception text (L3) — the client needs to know *why* it was refused.
            self._send({"type": "error", "detail": str(refusal),
                        "code": getattr(refusal, "code", "")})
            self._audit_auth_failed(group_id, str(refusal))
            return

        try:
            self._nonce_client = base64.b64decode(msg.get("nonce", ""))
        except Exception:
            self._nonce_client = b""
        if len(self._nonce_client) < NONCE_LEN:
            # The client nonce is what makes the NODE's proof fresh (C3). Without
            # it a recorded ack could be replayed by an impersonating peer.
            self._send({"type": "error", "detail": "Client nonce required"})
            return

        # Decoded, but NOT authenticated: that happens on the GEK proof.
        self._pending_sub = peer.user_id
        self._pending_group = peer.group_id
        self._pending_username = peer.username

        gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx
        if not gctx.get("gek"):
            self._send({
                "type": "error",
                "detail": "Group encryption not initialized — contact node operator",
            })
            return

        self._gek_challenge = os.urandom(NONCE_LEN)
        self._nonce_node = self._gek_challenge
        self._send({
            "type": MNP.HANDSHAKE_CHALLENGE,
            "v": MNP_VERSION,
            "nonce": base64.b64encode(self._gek_challenge).decode(),
            # Announced here because a first-time joiner needs it *before* the
            # ack: join_request signs a transcript naming this node, and someone
            # who has never held the GEK cannot complete the handshake to learn
            # it. Unverified at this point — the ack proves it, the client checks
            # the two match, and a wrong value only makes our own verification
            # fail. It is never a substitute for the ack's proof and signature.
            "node_pk": self._node_pk_b64(),
        })

    def _do_handshake_response(self, msg: dict) -> None:
        if not self._gek_challenge or not hasattr(self, "_pending_sub"):
            self._send({"type": "error", "detail": "No pending handshake challenge"})
            return

        group_id = self._pending_group
        gctx = self._ctx["groups"][group_id] if "groups" in self._ctx else self._ctx
        gek = gctx.get("gek")
        if not gek:
            self._send({"type": "error", "detail": "Group encryption not initialized"})
            self._gek_challenge = None
            return

        try:
            proof_bytes = base64.b64decode(msg.get("proof", ""))
        except Exception:
            self._send({"type": "error", "detail": "Invalid proof encoding"})
            return

        binding = self._channel_binding()
        if not binding:
            # Refuse rather than fall back to an unbound proof (L4).
            self._send({"type": "error", "detail": "Channel binding unavailable"})
            self._gek_challenge = None
            self._audit_auth_failed(group_id, "no channel binding")
            return

        if not verify_proof(gek, proof_bytes, ROLE_CLIENT, group_id,
                            self._nonce_client, self._gek_challenge, binding):
            self._send({"type": "error", "detail": "GEK proof failed"})
            self._gek_challenge = None
            self._audit_auth_failed(group_id, "GEK HMAC mismatch")
            return

        self._complete_handshake(gek, binding)
        self._gek_challenge = None

    def _complete_handshake(self, gek: bytes, binding: bytes) -> None:
        # Authenticated peers may send large frames (file uploads); unauthenticated
        # ones may not (H6).
        self._buffer.max_message = MAX_MSG
        self._user_id = self._pending_sub
        self._group_id = self._pending_group
        self._username = self._pending_username
        asyncio.ensure_future(self._load_pinned_pk())

        self._peer_registry()[self._user_id] = self

        node_user_id = self._ctx.get("node_user_id")
        log.info("WebRTC handshake OK — user=%s group=%s",
                 self._user_id[:8],
                 self._group_id[:8] if self._group_id else "none")
        # The node proves itself too (C3): possession of the GEK over the client's
        # nonce, plus a signature over the same transcript with its long-term key.
        # Previously the client received an unverifiable node_pk and trusted
        # is_node_admin from whoever answered — so a peer that had hijacked
        # signaling could serve a forged index, chat history and permissions.
        node_transcript = handshake_transcript(
            ROLE_NODE, self._group_id or "", self._nonce_client,
            self._gek_challenge or b"", binding)
        node_proof = make_proof(
            gek, ROLE_NODE, self._group_id or "", self._nonce_client,
            self._gek_challenge or b"", binding)

        ack = {
            "type": MNP.HANDSHAKE_ACK,
            "v": MNP_VERSION,
            "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()),
            "proof": base64.b64encode(node_proof).decode(),
            "sig": base64.b64encode(
                self._ctx["sk_node"].sign(node_transcript)).decode(),
            "is_node_admin": bool(node_user_id and self._user_id == node_user_id),
        }
        if node_user_id:
            ack["node_user_id"] = node_user_id
        pk_x_b64 = self._ctx.get("pk_x25519_b64")
        if pk_x_b64:
            ack["node_pk_x25519"] = pk_x_b64
        self._send(ack)
        self._audit("handshake")

    async def _do_gek_bundle_fetch(self) -> None:
        """Serve the caller's wrapped GEK bundle during the handshake window."""
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
            return

        group_id = getattr(self, "_pending_group", "")
        user_id = getattr(self, "_pending_sub", "")
        if not group_id or not user_id:
            self._send({"type": "error", "detail": "No pending handshake"})
            return

        bundle = await bundle_store.fetch(group_id, user_id)
        if bundle:
            self._send({
                "type": MNP.GEK_BUNDLE_RESP,
                "v": MNP_VERSION,
                "found": True,
                "pk_eph_b64": bundle["pk_eph_b64"],
                "nonce_b64": bundle["nonce_b64"],
                "wrapped_b64": bundle["wrapped_b64"],
            })
        else:
            self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})

    def _do_invite_create(self, msg: dict) -> None:
        """
        Issue a one-time pairing code for someone the operator wants to admit.

        Replaces the old invite path, where the inviter fetched the invitee's
        public key from the hub and wrapped the group key for whatever came back
        (H3). The node now needs nothing but a name: it will wrap the key itself,
        later, for a key the invitee proves they hold.
        """
        roster = self._ctx.get("roster")
        if roster is None:
            self._send({"type": "error", "detail": "Roster not available"})
            return

        invitee_id = msg.get("user_id", "")
        group_id = msg.get("group_id") or self._group_id
        if not invitee_id or not group_id:
            self._send({"type": "error", "detail": "Missing user_id or group_id"})
            return
        if group_id != self._group_id:
            self._send({"type": "error", "detail": "Wrong group for this session"})
            return

        if not self._has_admin_authority():
            self._send({
                "type": "error",
                "detail": "No operator paired — run `meshbay-node operator pair`",
            })
            return

        self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, {
            "group_id": group_id,
            "user_id": invitee_id,
            "username": str(msg.get("username", ""))[:64],
        })

    async def _do_keypair_bundle_fetch(self) -> None:
        """Serve the caller's encrypted keypair bundle during the handshake window."""
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
            return

        user_id = getattr(self, "_pending_sub", "")
        if not user_id:
            self._send({"type": "error", "detail": "No pending handshake"})
            return

        bundle_enc = await bundle_store.fetch_keypair(user_id)
        if bundle_enc:
            self._send({
                "type": MNP.KEYPAIR_BUNDLE_RESP,
                "v": MNP_VERSION,
                "found": True,
                "bundle_enc": bundle_enc,
            })
        else:
            self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False})

    async def _do_keypair_bundle_store(self, msg: dict) -> None:
        """Store an encrypted keypair bundle (user backs up their own keys on node)."""
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": "error", "detail": "Bundle store not available"})
            return

        bundle_enc = msg.get("bundle_enc", "")
        if not bundle_enc:
            self._send({"type": "error", "detail": "Missing bundle_enc"})
            return

        await bundle_store.store_keypair(self._user_id, bundle_enc)
        log.info("Keypair bundle stored for user=%s", self._user_id[:8])
        self._audit("keypair_bundle_store")
        self._send({
            "type": "ack", "v": MNP_VERSION,
            "detail": "keypair_bundle_stored",
        })

    # ── Pairing and join (H3, M3) ────────────────────────────────────────────

    def _join_refuse(self, reason: str, audit_detail: str = "") -> None:
        self._join_attempts += 1
        # Node-wide window, shared across connections: reconnecting must not reset
        # the budget.
        now = time.time()
        failures = [t for t in self._ctx.get("join_failures", [])
                    if now - t < JOIN_FAILURE_WINDOW]
        failures.append(now)
        self._ctx["join_failures"] = failures
        self._audit_join("join_refused", audit_detail or reason)
        self._send({
            "type": MNP.JOIN_RESULT,
            "v": MNP_VERSION,
            "ok": False,
            "reason": reason,
        })

    def _audit_join(self, event: str, detail: str) -> None:
        audit = self._ctx.get("audit_store")
        if not audit:
            return
        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
        asyncio.ensure_future(audit.log_event(
            user_id=self._user_id or getattr(self, "_pending_sub", "unknown"),
            event=event,
            ip=self._remote_ip,
            username=self._username or getattr(self, "_pending_username", ""),
            group_id=self._group_id or getattr(self, "_pending_group", "") or "",
            detail=detail,
        ))

    async def _do_join_request(self, msg: dict) -> None:
        """
        Pin an identity, or recognise one already pinned.

        The client signs its own Ed25519 and X25519 keys together with the node's
        nonce, so the identity key vouches for the encryption key — that is what
        will make it safe for the node to wrap the GEK for a key that arrived over
        the wire instead of one fetched from the hub's directory (H3).

        A first pairing needs a one-time code, which the hub never sees. Afterwards
        the pin is the credential and a changed key is refused outright, the same
        rule the client applies to `pk_node` (11.5.8).
        """
        roster = self._ctx.get("roster")
        if roster is None:
            self._send({"type": "error", "detail": "Roster not available"})
            return

        if self._join_attempts >= MAX_JOIN_ATTEMPTS:
            self._send({"type": "error", "detail": "Too many attempts"})
            return

        now = time.time()
        recent = [t for t in self._ctx.get("join_failures", [])
                  if now - t < JOIN_FAILURE_WINDOW]
        if len(recent) >= MAX_JOIN_FAILURES_WINDOW:
            self._audit_join("join_throttled", f"{len(recent)} failures in window")
            self._send({"type": "error", "detail": "Pairing temporarily locked"})
            return

        user_id = self._user_id or getattr(self, "_pending_sub", "")
        username = self._username or getattr(self, "_pending_username", "")
        if not user_id:
            self._send({"type": "error", "detail": "Handshake required"})
            return

        pk_ed_b64 = msg.get("pk_ed25519", "")
        pk_x_b64 = msg.get("pk_x25519", "")
        code = msg.get("code", "")
        ts = msg.get("ts", 0)

        try:
            pk_ed_raw = base64.b64decode(pk_ed_b64)
            pk_x_raw = base64.b64decode(pk_x_b64)
            if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32:
                raise ValueError
            pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw)
        except Exception:
            self._join_refuse("invalid_keys")
            return

        if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL:
            self._join_refuse("stale_request")
            return

        # An empty group_id means operator pairing, which is node-wide. Anything
        # else must be the group this connection authenticated to — a signature
        # obtained for one group must not name another.
        group_id = msg.get("group_id", "") or ""
        session_group = self._group_id or getattr(self, "_pending_group", "") or ""
        if group_id and group_id != session_group:
            self._join_refuse("group_mismatch")
            return

        transcript = join_transcript(
            node_pk_b64=self._node_pk_b64(),
            group_id=group_id,
            user_id=user_id,
            pk_ed25519_b64=pk_ed_b64,
            pk_x25519_b64=pk_x_b64,
            nonce_node=self._nonce_node,
            ts=ts,
        )
        try:
            sig = base64.b64decode(msg.get("sig", ""))
        except Exception:
            self._join_refuse("invalid_signature_encoding")
            return
        if not self._verify_sig(pk_ed, transcript, sig):
            self._join_refuse("signature_invalid")
            return

        known = await roster.get_identity(user_id)
        if known:
            if known["pk_ed25519"] != pk_ed_b64 or known["pk_x25519"] != pk_x_b64:
                # The blocking warning, raised where it matters: whoever this is
                # holds a different key than the person the operator paired.
                self._join_refuse(
                    "key_changed",
                    f"pinned={known['pk_ed25519'][:16]} presented={pk_ed_b64[:16]}")
                return
            # An operator's row is node-wide (empty group), so a lookup for the
            # group they happen to be opening finds nothing. Fall back to it, or
            # the client is told it has no role on a node it administers.
            member = (await roster.get_member(group_id, user_id)
                      or await roster.get_member("", user_id))
            await self._join_ok(
                user_id, pk_x_raw, session_group,
                role=member["role"] if member else "",
                recognised=True,
            )
            return

        if not code:
            if self._group_join_policy(session_group) == "open":
                # An open-join group admits anyone the hub calls a member, so a
                # code would protect nothing — the hub can walk in through the
                # front door. Pin what turns up and say so in the audit log.
                await self._pin_and_admit(
                    roster, user_id, username, pk_ed_b64, pk_x_b64,
                    group_id=session_group, role=ROLE_MEMBER,
                    approved_by="open-join", via="tofu")
                await self._join_ok(user_id, pk_x_raw, session_group,
                                    role=ROLE_MEMBER, recognised=False)
                return
            self._join_refuse("code_required")
            return

        invite = await roster.consume_invite(code, user_id)
        if not invite:
            self._join_refuse("code_invalid")
            return

        await self._pin_and_admit(
            # The name comes from the invitation, not from the token: the hub does
            # not put a username claim in a JWT, so pinning from the session alone
            # left the roster nameless and `member revoke <name>` unable to match.
            roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64,
            group_id=invite["group_id"], role=invite["role"],
            approved_by=invite["created_by"], via="code")
        # The roster row comes from the invitation; the key comes from the
        # connection. An operator pairing is node-wide (empty group), but they
        # redeemed the code while opening a group and expect to read it — and
        # is_authorized() already grants an operator every group on this node.
        await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"],
                            role=invite["role"], recognised=False)

    def _group_join_policy(self, group_id: str) -> str:
        """
        Admission policy for a group, read from the node's own configuration.

        Never from the hub: a hub that could declare a group open would be handed
        the key to it (§3.4 of docs/invite-pairing-v1.md).
        """
        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
        return gctx.get("join_policy", "invite")

    async def _pin_and_admit(
        self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str,
        *, group_id: str, role: str, approved_by: str, via: str,
    ) -> None:
        await roster.pin_identity(
            user_id=user_id, username=username,
            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via,
        )
        await roster.set_member(
            group_id=group_id, user_id=user_id, role=role,
            status="active", approved_by=approved_by,
        )
        if role == ROLE_OPERATOR:
            self._ctx["has_admin_authority"] = True

        log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role)
        self._audit_join("join_pinned", f"role={role} via={via}")

    async def _join_ok(
        self, user_id: str, pk_x_raw: bytes, group_id: str,
        *, role: str, recognised: bool,
    ) -> None:
        """
        Answer a join, wrapping the group key for the key the caller just proved.

        This is the H3 fix. The inviter used to fetch the invitee's public key from
        the hub and wrap the GEK for whatever came back, so a hub that answered
        with its own key was handed the group key by an honest member following the
        protocol exactly. The node now wraps for a key that arrived from its owner
        over an authenticated channel, bound to a pinned identity.
        """
        reply = {
            "type": MNP.JOIN_RESULT,
            "v": MNP_VERSION,
            "ok": True,
            "recognised": recognised,
            "role": role,
        }

        roster = self._ctx["roster"]
        if group_id and not await roster.is_authorized(group_id, user_id):
            # Pinned on this node, but not admitted to this group. Hub membership
            # alone must not produce a key.
            reply["gek"] = False
            reply["reason"] = "not_authorized_for_group"
            self._send(reply)
            self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized")
            return

        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
        gek = gctx.get("gek")
        if not gek:
            reply["gek"] = False
            reply["reason"] = "no_gek"
            self._send(reply)
            return

        bundle = wrap_gek_aes(gek, pk_x_raw)
        reply["gek"] = True
        reply["pk_eph_b64"] = bundle["pk_eph_b64"]
        reply["nonce_b64"] = bundle["nonce_b64"]
        reply["wrapped_b64"] = bundle["wrapped_b64"]
        self._send(reply)
        self._audit_join("gek_wrapped", f"group={group_id[:8]}")

    def _do_dir_create(self, msg: dict) -> None:
        """
        Create a directory, for any member of the group.

        Same confinement as an upload: every segment passes the name allowlist and
        the result must resolve under the shared root. Making a directory is not a
        privileged act — a member who can add a file can organise where it goes —
        but it writes to the operator's disk, so it is audited like one.
        """
        ctx = self._group_ctx()
        shared_root = ctx.get("shared_root")
        if not shared_root:
            self._send({"type": "error", "detail": "No shared directory"})
            return

        name = str(msg.get("name", "")).strip()
        if not SAFE_UPLOAD_NAME.match(name):
            self._send({"type": "error", "detail": "Invalid directory name"})
            return

        parent = safe_subdir(shared_root, msg.get("dir") or "")
        if parent is None or not parent.is_dir():
            self._send({"type": "error", "detail": "Invalid directory"})
            return

        target = safe_subdir(shared_root, f"{(msg.get('dir') or '').strip('/')}/{name}")
        if target is None:
            self._send({"type": "error", "detail": "Invalid directory"})
            return
        if target.exists():
            self._send({"type": "error", "detail": "Already exists"})
            return

        target.mkdir(parents=False)
        log.info("Directory created by %s: %s", self._user_id[:8],
                 target.relative_to(shared_root))
        self._audit("dir_create", str(target.relative_to(shared_root)))
        self._send({
            "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION,
            "dir": str(target.relative_to(shared_root)),
        })

    def _do_dir_delete(self, msg: dict) -> None:
        """
        Remove an empty directory, for the node operator.

        Creating one is not privileged — a member who can add a file may organise
        where it goes — but removing one is: it acts on a name other members are
        using, and on the operator's disk. Empty is the whole safety property
        here. Nothing recursive: refusing a directory with anything in it means
        this can never destroy content, whatever the caller intended, so the
        operator deletes the files first and sees what they are losing.
        """
        ctx = self._group_ctx()
        shared_root = ctx.get("shared_root")
        if not shared_root:
            self._send({"type": "error", "detail": "No shared directory"})
            return

        target = safe_subdir(shared_root, msg.get("dir") or "")
        if target is None or target == shared_root:
            self._send({"type": "error", "detail": "Invalid directory"})
            return
        if not target.is_dir():
            self._send({"type": "error", "detail": "Not a directory"})
            return
        if any(target.iterdir()):
            self._send({"type": "error", "detail": "Directory is not empty"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for deletion"})
            return

        self._issue_admin_challenge(
            OP_DIR_DELETE, str(target.relative_to(shared_root)))

    async def _admin_exec_dir_delete(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        rel = pending["subject"]
        ctx = self._group_ctx()
        shared_root = ctx.get("shared_root")
        target = safe_subdir(shared_root, rel) if shared_root else None
        if target is None or target == shared_root or not target.is_dir():
            self._send({"type": "error", "detail": "Not a directory"})
            return

        # Operator only. A file has an uploader who may remove their own; a
        # directory has none, so there is no second key to accept here.
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"dir_delete:{rel}")
            return

        # Checked again after the signature: the emptiness test that let this
        # through happened before a round trip to the operator's browser, and a
        # file could have landed in the meantime.
        if any(target.iterdir()):
            self._send({"type": "error", "detail": "Directory is not empty"})
            return

        target.rmdir()
        log.info("Directory removed by %s: %s", self._user_id[:8], rel)
        self._audit("dir_delete", rel)
        self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel})

    async def _do_keypair_bundle_delete(self) -> None:
        """
        Withdraw our own key backup from this node.

        Only ever our own: the user_id comes from the authenticated session, never
        from the message. Someone who does not want a second browser should not be
        leaving a PBKDF2-protected blob on every node they have ever joined (C4),
        and turning the setting off has to remove what is already there — not just
        stop adding to it.
        """
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": "error", "detail": "Bundle store not available"})
            return

        removed = await bundle_store.delete_keypair(self._user_id)
        if removed:
            log.info("Keypair bundle withdrawn by user=%s", self._user_id[:8])
            self._audit("keypair_bundle_delete")
        self._send({"type": "ack", "v": MNP_VERSION,
                    "detail": "keypair_bundle_deleted", "removed": removed})

    def _audit_pre_proof_fetch(self, mtype: str) -> None:
        """Record bundle access made before the GEK proof (C4)."""
        audit = self._ctx.get("audit_store")
        if not audit:
            return
        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
        asyncio.ensure_future(audit.log_event(
            user_id=getattr(self, "_pending_sub", "unknown"),
            event="pre_proof_fetch",
            ip=self._remote_ip,
            username=self._username or getattr(self, "_pending_username", ""),
            group_id=getattr(self, "_pending_group", "") or "",
            detail=mtype,
        ))

    def _audit_auth_failed(self, group_id: str, reason: str) -> None:
        audit = self._ctx.get("audit_store")
        if audit:
            self._remote_ip = _get_remote_ip(self._pc)
            asyncio.ensure_future(audit.log_event(
                user_id="unknown",
                event="auth_failed",
                ip=self._remote_ip,
                group_id=group_id,
                detail=reason,
            ))

    def _group_ctx(self) -> dict:
        if "groups" in self._ctx and self._group_id:
            return self._ctx["groups"][self._group_id]
        return self._ctx

    def _peer_registry(self) -> dict:
        """
        Connected peers for THIS group only.

        Finding H1: this used to live on the shared transport context, so a chat
        message was broadcast to every peer on the node regardless of which group
        they had authenticated to.
        """
        return self._group_ctx().setdefault("_peers", {})

    def _user_names(self) -> dict:
        """Display-name cache, per group — same leak as _peer_registry (H1)."""
        return self._group_ctx().setdefault("_user_names", {})

    def _do_index_sync(self) -> None:
        ctx = self._group_ctx()
        idx = ctx["index"]
        entries = [
            {
                "id": e.id, "name": e.name, "path": e.path,
                "size": e.size, "type": e.type, "added_at": e.added_at,
                "uploader_id": e.uploader_id,
            }
            for e in idx.entries
        ]
        self._send({
            "type": MNP.INDEX_SYNC,
            "v": MNP_VERSION,
            "group_id": idx.group_id,
            "version": idx.version,
            "entries": entries,
            # Directories are not index entries, so the client used to infer them
            # from file paths — which means a folder someone just created, or one
            # they emptied, simply did not exist as far as the UI was concerned.
            "dirs": self._list_dirs(ctx.get("shared_root")),
        })

    @staticmethod
    def _list_dirs(shared_root: Path | None) -> list[str]:
        """Directories under the shared root, relative and sorted."""
        if not shared_root:
            return []
        out = []
        try:
            for path in sorted(shared_root.rglob("*")):
                if path.is_dir() and not path.name.startswith("."):
                    rel = path.relative_to(shared_root)
                    if not any(part.startswith(".") for part in rel.parts):
                        out.append(str(rel))
        except OSError:
            return []
        return out[:2000]

    def _do_file_request(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg["file_id"]
        chunk_index = msg["chunk_index"]
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            log.warning("File not found: %s", file_id[:16])
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path = ctx["shared_root"] / entry.path / entry.name
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        file_hash = bytes.fromhex(entry.id)
        chunk_data = _read_and_encrypt(
            self._ctx["sk_node"],
            ctx["gek"],
            file_path,
            chunk_index,
            file_hash,
        )
        self._send(chunk_data)
        if chunk_index == 0:
            self._audit("file_download", entry.name)

    def _do_stream_segment(self, msg: dict) -> None:
        asyncio.ensure_future(self._do_stream_segment_async(msg))

    async def _do_stream_segment_async(self, msg: dict) -> None:
        """
        Legacy HLS segment extraction (superseded by stream_req/MSE).

        Finding H6: this ran subprocess.run(..., timeout=30) directly inside the
        event loop, so a single request stalled the whole daemon — every peer,
        every group — for up to thirty seconds. Now async and under the same
        transcode semaphore as _stream_video.
        """
        ctx = self._group_ctx()
        file_id = msg["file_id"]
        segment_index = msg["segment_index"]
        segment_duration = msg.get("segment_duration", 4)

        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path = ctx["shared_root"] / entry.path / entry.name
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        sem = self._ctx.get("_transcode_sem")
        if sem is None:
            sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES)
            self._ctx["_transcode_sem"] = sem

        try:
            async with sem:
                proc = await asyncio.create_subprocess_exec(
                    "ffmpeg", "-hide_banner", "-loglevel", "error",
                    "-ss", str(segment_index * segment_duration),
                    "-i", str(file_path),
                    "-t", str(segment_duration),
                    "-c:v", "copy", "-c:a", "copy",
                    "-f", "mpegts", "pipe:1",
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                try:
                    stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30)
                except asyncio.TimeoutError:
                    proc.kill()
                    await proc.wait()
                    self._send({"type": "error", "detail": "Segment extraction timed out"})
                    return
            if proc.returncode != 0 or not stdout:
                self._send({"type": "error", "detail": "Segment extraction failed"})
                return
            segment_data = stdout
        except Exception:
            self._send({"type": "error", "detail": "Segment extraction failed"})
            return

        self._send({
            "type": MNP.STREAM_SEGMENT,
            "v": MNP_VERSION,
            "file_id": file_id,
            "segment_index": segment_index,
            "data_b64": base64.b64encode(segment_data).decode(),
            "size": len(segment_data),
        })

    def _do_chat_message(self, msg: dict) -> None:
        # Per-group store — see _peer_registry() and finding H1. Reading chat_store
        # off the shared transport context sent every group's messages to the first
        # group's database, and served them back to anyone on the node.
        chat_store = self._group_ctx().get("chat_store")
        payload = msg.get("payload", "")
        sender_name = msg.get("sender_name", "")
        if sender_name:
            self._user_names()[self._user_id] = sender_name
        if chat_store:
            raw = payload.encode() if isinstance(payload, str) else payload
            asyncio.ensure_future(chat_store.save_message(
                sender_id=self._user_id,
                iteration=msg.get("iteration", 0),
                payload=raw,
                thread_id=msg.get("thread_id"),
                sender_name=sender_name,
            ))

        peers = self._peer_registry()
        broadcast = {
            "type": MNP.CHAT_MESSAGE,
            "v": MNP_VERSION,
            "sender_id": self._user_id,
            "sender_name": sender_name,
            "payload": payload,
            "thread_id": msg.get("thread_id"),
            "timestamp": __import__("time").time(),
        }
        for uid, session in list(peers.items()):
            if uid != self._user_id and session is not self:
                try:
                    session._send(broadcast)
                except Exception:
                    pass

        hub_ws = self._ctx.get("hub_ws")
        if hub_ws and self._group_id:
            try:
                import json as _json
                asyncio.ensure_future(hub_ws.send(_json.dumps({
                    "type": "chat_notify",
                    "group_id": self._group_id,
                    "sender_name": sender_name,
                    # Who actually wrote it, from the authenticated session. The
                    # hub used to fall back to this node's own token subject —
                    # the operator — so everyone was notified of their own
                    # messages and the operator was notified of nobody's.
                    "sender_user_id": self._user_id,
                })))
            except Exception:
                pass

        self._send({"type": "ack", "v": MNP_VERSION})
        self._audit("chat_message")

    def _do_chat_history(self, msg: dict) -> None:
        chat_store = self._group_ctx().get("chat_store")
        if not chat_store:
            self._send({
                "type": MNP.CHAT_HISTORY_RESPONSE,
                "v": MNP_VERSION,
                "messages": [],
            })
            return

        since = msg.get("since", 0)
        limit = msg.get("limit", 100)
        asyncio.ensure_future(self._send_chat_history(chat_store, since, limit))

    async def _send_chat_history(self, chat_store, since: float, limit: int) -> None:
        msgs = await chat_store.get_messages(since=since, limit=limit)
        names = self._user_names()
        self._send({
            "type": MNP.CHAT_HISTORY_RESPONSE,
            "v": MNP_VERSION,
            "messages": [
                {
                    "id": m.id,
                    "sender_id": m.sender_id,
                    "sender_name": m.sender_name or names.get(m.sender_id, ""),
                    "payload": m.payload.decode("utf-8", errors="replace")
                    if isinstance(m.payload, bytes) else m.payload,
                    "timestamp": m.timestamp,
                    "thread_id": m.thread_id,
                }
                for m in msgs
            ],
        })

    def _do_file_upload(self, msg: dict) -> None:
        ctx = self._group_ctx()
        filename = msg.get("filename", "")
        chunk_index = msg.get("chunk_index", 0)
        total_chunks = msg.get("total_chunks", 1)
        data = msg.get("data")

        if not filename or data is None:
            self._send({"type": "error", "detail": "Missing filename or data"})
            return

        if not SAFE_UPLOAD_NAME.match(filename):
            self._send({"type": "error", "detail": "Invalid filename"})
            return

        shared_root = ctx.get("shared_root")
        if not shared_root:
            self._send({"type": "error", "detail": "No shared directory"})
            return

        # One destination, chosen here and not by the client: uploads/ at the root
        # of the shared directory. C5a is still honoured — the name passed the
        # allowlist above, and an existing file is never replaced, which was the
        # real defect (overwriting a file also made the attacker its recorded
        # uploader, and therefore able to delete it).
        rel_dir = UPLOAD_DIR_NAME
        target_dir = shared_root / UPLOAD_DIR_NAME
        target_dir.mkdir(parents=True, exist_ok=True)

        upload_key = f"{rel_dir}/{filename}"
        state = self._uploads.get(upload_key)
        # A shared directory means two people can send the same name. Refusing the
        # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
        # a free name is found instead. Never a replacement.
        stored_name = state["stored_name"] if state else _free_name(target_dir, filename)
        tmp_path = target_dir / f"{stored_name}.part"
        final_path = target_dir / stored_name

        if chunk_index == 0:
            # Backstop: _free_name already guarantees this, and it stays because
            # it asserts the invariant where the write happens.
            if final_path.exists():
                self._send({"type": "error", "detail": "File already exists"})
                return
            state = {"next_index": 0, "bytes": 0, "stored_name": stored_name}
            self._uploads[upload_key] = state
        elif state is None:
            self._send({"type": "error", "detail": "Upload not started"})
            return

        # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
        # blindly to whatever .part file is already on disk.
        if chunk_index != state["next_index"]:
            self._send({"type": "error", "detail": "Unexpected chunk index"})
            return

        if isinstance(data, str):
            chunk_bytes = base64.b64decode(data)
        else:
            chunk_bytes = bytes(data)

        if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES:
            self._uploads.pop(upload_key, None)
            tmp_path.unlink(missing_ok=True)
            self._send({"type": "error", "detail": "Upload exceeds size limit"})
            return

        with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f:
            f.write(chunk_bytes)
        state["next_index"] = chunk_index + 1
        state["bytes"] += len(chunk_bytes)

        self._send({
            "type": MNP.FILE_UPLOAD_ACK,
            "v": MNP_VERSION,
            "chunk_index": chunk_index,
            "filename": filename,
            # What it is actually called on disk, which a chat attachment has to
            # reference and the uploader deserves to be told.
            "stored_as": stored_name,
            "dir": rel_dir,
        })

        if chunk_index + 1 >= total_chunks:
            self._uploads.pop(upload_key, None)
            tmp_path.rename(final_path)
            log.info("Upload complete: %s (%d chunks, %d bytes)",
                     stored_name, total_chunks, state["bytes"])
            self._audit("file_upload", f"{rel_dir}/{stored_name}")
            self._register_uploader(ctx, rel_dir, stored_name)

    def _register_uploader(self, ctx: dict, rel_dir: str, filename: str) -> None:
        """
        Tag the index entry with the uploader's identity after upload completes.

        The key recorded here is the one this node pinned, not the one the token
        carried. `pk_user` was a hub-chosen claim, and it decided who could later
        delete the file: a hub issuing a token naming its own key could delete
        anyone's uploads on any node. Deletion is supposed to be authorized by the
        node, and this closes the last place where it was not.
        """
        idx = ctx.get("index")
        if not idx:
            return
        for entry in idx.entries:
            if entry.name == filename and entry.path == rel_dir:
                entry.uploader_id = self._user_id
                entry.uploader_pk = self._pinned_pk
                return

    def _do_file_delete(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        if not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return

        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        has_uploader_pk = bool(entry.uploader_pk)
        if not self._has_admin_authority() and not has_uploader_pk:
            self._send({"type": "error", "detail": "No authorized key for deletion"})
            return

        self._issue_admin_challenge(OP_FILE_DELETE, file_id)

    # ── Admin operation challenge/response (finding H5) ──────────────────────

    def _node_pk_b64(self) -> str:
        return pk_to_b64(self._ctx["sk_node"].public_key())

    def _issue_admin_challenge(
        self, op: str, subject: str, payload: dict | None = None,
    ) -> None:
        """
        Ask the client to authorize `op` on `subject` with its Ed25519 identity key.

        The client is sent the transcript *fields*, not opaque bytes, so it can
        rebuild and inspect what it signs. The node keeps the authoritative copy and
        rebuilds the transcript itself at verification time — nothing signed is ever
        taken from the response message.
        """
        nonce = os.urandom(32)
        ts = int(time.time())
        op_id = base64.b64encode(os.urandom(16)).decode()
        self._admin_ops[op_id] = {
            "op": op, "subject": subject, "nonce": nonce, "ts": ts,
            "payload": payload or {},
        }
        self._send({
            "type": MNP.ADMIN_CHALLENGE,
            "v": MNP_VERSION,
            "op_id": op_id,
            "op": op,
            "subject": subject,
            "nonce": base64.b64encode(nonce).decode(),
            "ts": ts,
            "node_pk": self._node_pk_b64(),
            "group_id": self._group_id or "",
        })

    @staticmethod
    def _verify_sig(pk: Ed25519PublicKey | None, transcript: bytes, sig: bytes) -> bool:
        if pk is None:
            return False
        try:
            pk.verify(sig, transcript)
            return True
        except Exception:
            return False

    async def _load_pinned_pk(self) -> None:
        """Remember which key this node pinned for the peer we just authenticated."""
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id:
            return
        ident = await roster.get_identity(self._user_id)
        if ident:
            self._pinned_pk = ident["pk_ed25519"]

    def _has_admin_authority(self) -> bool:
        """
        Cheap synchronous pre-check: is there anyone who could authorize this?

        Only decides whether to issue a challenge at all — the gate is
        `_verify_admin_sig`. The flag is set at startup and refreshed in-process
        when an operator pairs.
        """
        return bool(self._ctx.get("has_admin_authority"))

    async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool:
        """
        Check a signature against every key holding node-operator authority.

        Read from the roster on each call rather than cached: revoking a paired
        browser must take effect immediately, and admin operations are rare enough
        that a SQLite read costs nothing.

        There is one source of operator authority and this is it. `admin_pk_ed25519`
        in node.toml used to be honoured alongside the roster; it is gone, and a
        config that still names it is warned about at startup rather than obeyed.
        """
        roster = self._ctx.get("roster")
        if roster is None:
            return False
        for pk_b64 in await roster.operator_pks():
            try:
                pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64))
            except Exception:
                continue
            if self._verify_sig(pk, transcript, sig):
                return True
        return False

    def _do_admin_response(self, msg: dict) -> None:
        op_id = msg.get("op_id", "")
        sig_b64 = msg.get("signature", "")

        pending = self._admin_ops.pop(op_id, None)
        if not pending:
            self._send({"type": "error", "detail": "No pending admin operation"})
            return

        if time.time() - pending["ts"] > ADMIN_CHALLENGE_TTL:
            self._send({"type": "error", "detail": "Admin challenge expired"})
            return

        try:
            sig_bytes = base64.b64decode(sig_b64)
        except Exception:
            self._send({"type": "error", "detail": "Invalid signature encoding"})
            return

        transcript = admin_transcript(
            op=pending["op"],
            node_pk_b64=self._node_pk_b64(),
            group_id=self._group_id or "",
            subject=pending["subject"],
            nonce=pending["nonce"],
            ts=pending["ts"],
        )

        if pending["op"] == OP_FILE_DELETE:
            asyncio.ensure_future(
                self._admin_exec_file_delete(pending, transcript, sig_bytes))
        elif pending["op"] == OP_DIR_DELETE:
            asyncio.ensure_future(
                self._admin_exec_dir_delete(pending, transcript, sig_bytes))
        elif pending["op"] == OP_INVITE_CREATE:
            asyncio.ensure_future(
                self._admin_exec_invite_create(pending, transcript, sig_bytes))
        else:
            self._send({"type": "error", "detail": "Unknown admin operation"})

    async def _admin_exec_file_delete(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        file_id = pending["subject"]
        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        uploader_pk = None
        if entry.uploader_pk:
            try:
                uploader_pk = Ed25519PublicKey.from_public_bytes(
                    base64.b64decode(entry.uploader_pk))
            except Exception:
                uploader_pk = None

        # Node operator, or the user who uploaded this file — verified by the key
        # recorded at upload time, never by a JWT claim (the hub controls those).
        if not (await self._verify_admin_sig(transcript, sig)
                or self._verify_sig(uploader_pk, transcript, sig)):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
            return

        self._exec_file_delete(ctx, file_id, entry)

    async def _admin_exec_invite_create(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        # Node operator only. A group admin who does not run the node has no
        # authority over who this node admits (deny by default). Delegation is
        # designed but deferred — see §6.2 of docs/invite-pairing-v1.md.
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}")
            return

        roster = self._ctx.get("roster")
        if roster is None:
            self._send({"type": "error", "detail": "Roster not available"})
            return

        payload = pending["payload"]
        code = await roster.create_invite(
            group_id=payload["group_id"],
            user_id=payload["user_id"],
            role=ROLE_MEMBER,
            created_by=self._user_id or "",
            ttl=self._ctx.get("invite_ttl", DEFAULT_INVITE_TTL),
            username=payload.get("username", ""),
        )
        invites = await roster.list_invites()
        expires = next(
            (i["expires_at"] for i in invites
             if i["user_id"] == payload["user_id"]
             and i["group_id"] == payload["group_id"]), "")

        log.info("Invite created: group=%s user=%s",
                 payload["group_id"][:8], payload["user_id"][:8])
        self._audit("invite_create", f"target={payload['user_id'][:8]}")
        # The code exists in the clear exactly here and in the operator's hands.
        self._send({
            "type": MNP.INVITE_RESULT,
            "v": MNP_VERSION,
            "code": code,
            "expires_at": expires,
            "user_id": payload["user_id"],
            "username": payload.get("username", ""),
        })

    def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
        file_path = ctx["shared_root"] / entry.path / entry.name
        if file_path.exists():
            file_path.unlink()
            log.info("File deleted: %s", entry.name)
            self._audit("file_delete", entry.name)

        ctx["index"].remove_entry(file_id)
        self._send({
            "type": MNP.FILE_DELETE_ACK,
            "v": MNP_VERSION,
            "file_id": file_id,
        })

    async def _stream_video(self, msg: dict) -> None:
        """Stream a video file as fMP4 segments via MSE-compatible output."""
        # One ffmpeg per request with no cap lets any member exhaust the node's
        # CPU and process table (H6). The semaphore lives on the transport context
        # so it is shared across all peers, not per-session.
        sem = self._ctx.get("_transcode_sem")
        if sem is None:
            sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES)
            self._ctx["_transcode_sem"] = sem
        if sem.locked() and sem._value <= 0:
            self._send({"type": "error", "detail": "Server busy, retry shortly"})
            return
        async with sem:
            await self._stream_video_inner(msg)

    async def _stream_video_inner(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path = ctx["shared_root"] / entry.path / entry.name
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        gek = ctx.get("gek")
        file_hash = bytes.fromhex(entry.id)

        try:
            codec_str, duration = await _probe_video(str(file_path))
        except Exception as e:
            self._send({"type": "error", "detail": f"Probe failed: {e}"})
            return

        if not codec_str:
            self._send({"type": "error", "detail": "Unsupported video codec"})
            return

        proc = await asyncio.create_subprocess_exec(
            "ffmpeg", "-hide_banner", "-loglevel", "error",
            "-i", str(file_path),
            "-c", "copy",
            "-movflags", "frag_keyframe+empty_moov+default_base_moof",
            "-f", "mp4", "pipe:1",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )

        self._send({
            "type": MNP.STREAM_INIT,
            "v": MNP_VERSION,
            "file_id": file_id,
            "codec": codec_str,
            "duration": duration,
        })

        index = 0
        try:
            while True:
                data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
                if not data:
                    break
                ckey = chunk_key_aes(gek, file_hash, index)
                nonce, ct = encrypt_chunk_aes(ckey, data)
                self._send({
                    "type": MNP.STREAM_DATA,
                    "v": MNP_VERSION,
                    "file_id": file_id,
                    "segment_index": index,
                    "nonce": nonce,
                    "ct": ct,
                    "plaintext_size": len(data),
                })
                index += 1
                await asyncio.sleep(0)
        except Exception as e:
            log.error("Stream error: %s", e)
        finally:
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            await proc.wait()

        self._send({
            "type": MNP.STREAM_END,
            "v": MNP_VERSION,
            "file_id": file_id,
        })
        log.info("Streamed %s: %d segments", entry.name, index)
        self._audit("stream_video", entry.name)

    def _send(self, obj: dict) -> None:
        if self._channel and self._channel.readyState == "open":
            self._channel.send(_pack(obj))
        else:
            log.warning("WebRTC send skipped: channel=%s",
                        self._channel.readyState if self._channel else "none")

    async def close(self) -> None:
        self._audit("disconnect")
        if self._user_id:
            self._peer_registry().pop(self._user_id, None)
        await self._pc.close()


def _read_and_encrypt(
    sk_node: Ed25519PrivateKey,
    gek: bytes,
    file_path: Path,
    chunk_index: int,
    file_hash: bytes,
) -> dict:
    with open(file_path, "rb") as f:
        f.seek(chunk_index * CHUNK_SIZE)
        plaintext = f.read(CHUNK_SIZE)

    ckey = chunk_key_aes(gek, file_hash, chunk_index)
    nonce, ct = encrypt_chunk_aes(ckey, plaintext)

    return {
        "type": MNP.FILE_CHUNK,
        "v": MNP_VERSION,
        "chunk_index": chunk_index,
        "plaintext_size": len(plaintext),
        "nonce": nonce,
        "ct": ct,
    }


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

    Usage:
        transport = WebRTCTransport(sk_node, hub_pk_pem, gek, shared_root, 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,
        shared_root: Path,
        index: GroupIndex,
        groups: dict[str, dict] | None = None,
        denylist: Any | None = None,
        stun_servers: list[str] | None = None,
    ):
        self._ctx: dict[str, Any] = {
            "sk_node": sk_node,
            "hub_pk_pem": hub_pk_pem,
            "gek": gek,
            "shared_root": shared_root,
            "index": index,
            "_peers": {},
        }
        if groups:
            self._ctx["groups"] = groups
        if denylist:
            self._ctx["denylist"] = denylist
        self._stun = stun_servers or ["stun:stun.l.google.com:19302"]
        self._sessions: dict[str, WebRTCPeerSession] = {}

    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 RTCIceServer, RTCConfiguration

        config = RTCConfiguration(
            iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else []
        )
        pc = RTCPeerConnection(configuration=config)
        session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id)
        self._sessions[peer_id] = session

        @pc.on("datachannel")
        def on_datachannel(channel: RTCDataChannel):
            log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id)
            session._setup_channel(channel)

        @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"):
                self._sessions.pop(peer_id, None)

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

        log.info("WebRTC answer ready for peer=%s", peer_id)
        return pc.localDescription.sdp, []

    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 session in list(self._sessions.values()):
            await session.close()
        self._sessions.clear()

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