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.adminop import (
    ADMIN_CHALLENGE_TTL,
    OP_FILE_DELETE,
    OP_GEK_BUNDLE_STORE,
    admin_transcript,
)
from meshbay_common.crypto import pk_to_b64
from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
from meshbay_common.protocol import MNP
from meshbay_node.indexer import GroupIndex

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
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 _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."""

    def __init__(self):
        self._buf = bytearray()

    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 > MAX_MSG:
                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()
        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 = ""
        self._pk_user: str = ""
        self._gek_challenge: bytes | None = None
        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 == MNP.GEK_BUNDLE_FETCH and self._gek_challenge is not None:
                asyncio.ensure_future(self._do_gek_bundle_fetch())
            elif mtype == MNP.KEYPAIR_BUNDLE_FETCH and self._gek_challenge is not None:
                asyncio.ensure_future(self._do_keypair_bundle_fetch())
            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.FILE_DELETE:
                self._do_file_delete(msg)
            elif mtype == MNP.ADMIN_RESPONSE:
                self._do_admin_response(msg)
            elif mtype == MNP.GEK_BUNDLE_STORE:
                self._do_gek_bundle_store(msg)
            elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
                asyncio.ensure_future(self._do_keypair_bundle_store(msg))
            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 _do_handshake(self, msg: dict) -> None:
        token = msg.get("token", "")
        group_id = msg.get("group_id", "")
        try:
            decoded = jwt.decode(token, self._ctx["hub_pk_pem"], algorithms=["EdDSA"])
        except Exception as e:
            self._send({"type": "error", "detail": f"Invalid JWT: {e}"})
            self._audit_auth_failed(group_id, str(e))
            return

        denylist = self._ctx.get("denylist")
        if denylist and denylist.is_denied(
                decoded.get("sub", ""), decoded.get("jti", ""), group_id):
            self._send({"type": "error", "detail": "Token revoked"})
            return

        if group_id and group_id not in decoded.get("groups", []):
            self._send({"type": "error", "detail": "Not a member of this group"})
            return

        if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]:
            self._send({"type": "error", "detail": "Group not hosted on this node"})
            return

        # Store decoded JWT data but DO NOT set self._user_id yet —
        # the user is not authenticated until they prove GEK possession.
        self._pending_sub = decoded["sub"]
        self._pending_group = group_id
        self._pending_username = decoded.get("username", "")
        self._pending_pk_user = decoded.get("pk_user", "")

        ctx = self._ctx
        if "groups" in ctx and group_id:
            gctx = ctx["groups"].get(group_id, ctx)
        else:
            gctx = ctx
        gek = gctx.get("gek")

        nonce = os.urandom(32)
        self._gek_challenge = nonce
        challenge = {
            "type": MNP.HANDSHAKE_CHALLENGE,
            "v": MNP_VERSION,
            "nonce": base64.b64encode(nonce).decode(),
        }
        if not gek:
            self._send({
                "type": "error",
                "detail": "Group encryption not initialized — contact node operator",
            })
            return
        self._send(challenge)

    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
        ctx = self._ctx
        if "groups" in ctx and group_id:
            gctx = ctx["groups"].get(group_id, ctx)
        else:
            gctx = ctx
        gek = gctx.get("gek")

        if not gek:
            self._send({"type": "error", "detail": "Group encryption not initialized"})
            self._gek_challenge = None
            return

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

        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)

        data = self._gek_challenge + offer_fp + answer_fp
        expected = hmac.new(gek, data, hashlib.sha256).digest()
        if not hmac.compare_digest(proof_bytes, expected):
            self._send({"type": "error", "detail": "GEK proof failed"})
            self._gek_challenge = None
            self._audit_auth_failed(group_id, "GEK HMAC mismatch")
            return

        self._gek_challenge = None
        self._complete_handshake()

    def _complete_handshake(self) -> None:
        self._user_id = self._pending_sub
        self._group_id = self._pending_group
        self._username = self._pending_username
        self._pk_user = self._pending_pk_user

        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")
        ack = {
            "type": MNP.HANDSHAKE_ACK,
            "v": MNP_VERSION,
            "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()),
            "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_gek_bundle_store(self, msg: dict) -> None:
        """
        Request to store a wrapped GEK bundle for a target user.

        Finding C5b: this used to write whatever any authenticated member sent, with
        INSERT OR REPLACE semantics, and then auto-activate the bundle if it was
        addressed to the node operator. Since the operator's X25519 public key is
        public — the node even hands it out in handshake_ack — any member could wrap
        a GEK of their own choosing for the operator and make the node adopt it,
        locking every legitimate member out of the group and taking over the key.

        Storing a bundle is now a node-operator operation gated by an Ed25519
        challenge, and nothing arriving over MNP can activate a GEK: activation
        happens only through the local admin UI or the CLI.
        """
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": "error", "detail": "Bundle store not available"})
            return

        target_user_id = msg.get("user_id", "")
        group_id = msg.get("group_id") or self._group_id
        pk_eph = msg.get("pk_eph_b64", "")
        nonce = msg.get("nonce_b64", "")
        wrapped = msg.get("wrapped_b64", "")

        if not target_user_id or not pk_eph or not nonce or not wrapped or not group_id:
            self._send({"type": "error", "detail": "Missing bundle fields"})
            return

        if not self._ctx.get("admin_pk_ed25519"):
            self._send({
                "type": "error",
                "detail": "No admin key pinned — bundle storage refused",
            })
            return

        self._issue_admin_challenge(OP_GEK_BUNDLE_STORE, target_user_id, {
            "group_id": group_id,
            "user_id": target_user_id,
            "pk_eph_b64": pk_eph,
            "nonce_b64": nonce,
            "wrapped_b64": wrapped,
        })

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

    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,
        })

    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:
        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

        import subprocess
        try:
            result = subprocess.run(
                ["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"],
                capture_output=True, timeout=30,
            )
            if result.returncode != 0 or not result.stdout:
                self._send({"type": "error", "detail": "Segment extraction failed"})
                return
            segment_data = result.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,
                })))
            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

        # Per-user quarantine: a member can only ever write inside their own directory,
        # so they cannot overwrite the operator's files or another member's (C5a).
        rel_dir = f"{UPLOAD_DIR_NAME}/{self._user_id}"
        user_dir = shared_root / UPLOAD_DIR_NAME / self._user_id
        user_dir.mkdir(parents=True, exist_ok=True)
        tmp_path = user_dir / f"{filename}.part"
        final_path = user_dir / filename

        state = self._uploads.get(filename)
        if chunk_index == 0:
            if final_path.exists():
                self._send({"type": "error", "detail": "File already exists"})
                return
            state = {"next_index": 0, "bytes": 0}
            self._uploads[filename] = 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(filename, 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,
        })

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

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

        admin_pk = self._ctx.get("admin_pk_ed25519")
        has_uploader_pk = bool(entry.uploader_pk)
        if not admin_pk 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

    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:
            self._admin_exec_file_delete(pending, transcript, sig_bytes)
        elif pending["op"] == OP_GEK_BUNDLE_STORE:
            asyncio.ensure_future(
                self._admin_exec_bundle_store(pending, transcript, sig_bytes))
        else:
            self._send({"type": "error", "detail": "Unknown admin operation"})

    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 (self._verify_sig(self._ctx.get("admin_pk_ed25519"), 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_bundle_store(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        # Node operator only. A group admin who does not run the node has no
        # authority over what this node stores (draft-v4 §4.2.x, deny by default).
        if not self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"gek_bundle_store:{pending['subject'][:16]}")
            return

        payload = pending["payload"]
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": "error", "detail": "Bundle store not available"})
            return

        await bundle_store.store(
            payload["group_id"], payload["user_id"],
            payload["pk_eph_b64"], payload["nonce_b64"], payload["wrapped_b64"],
        )
        log.info("GEK bundle stored: group=%s user=%s",
                 payload["group_id"][:8], payload["user_id"][:8])
        self._audit("gek_bundle_store", f"target={payload['user_id'][:8]}")
        self._send({
            "type": "ack", "v": MNP_VERSION,
            "detail": "gek_bundle_stored",
            "user_id": payload["user_id"],
        })

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