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

"""The MNP handshake: the node's challenge, the client's proof of the group key
bound to this DTLS channel, and what a peer is sent once it is in."""

import base64
import logging
import os

from meshbay_common import MNP_VERSION
from meshbay_common.crypto import pk_to_b64
from meshbay_common.groupbox import PURPOSE_ACK, seal
from meshbay_common.handshake import (
    MNP_MIN_SUPPORTED,
    NONCE_LEN,
    ROLE_CLIENT,
    ROLE_NODE,
    HandshakeError,
    authorize_token,
    challenge_transcript,
    check_version,
    handshake_transcript,
    make_proof,
    verify_proof,
    webrtc_binding,
)
from meshbay_common.protocol import MNP

from meshbay_node import transfers as transfers_mod
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.transport.webrtc.channel import _extract_dtls_fingerprint, _get_remote_ip
from meshbay_node.transport.webrtc.limits import MAX_MSG

log = logging.getLogger("meshbay_node.transport.webrtc_server")


class HandshakeMixin:
    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", "")
        log.info("WebRTC handshake request: group=%s (peer=%s)",
                 group_id[:8] if group_id else "none", self._peer_id)
        # Before the token, and before anything is decided from it: a peer we
        # cannot speak to is refused with a code it can act on, rather than
        # served messages it will misread as missing fields (L2).
        try:
            check_version(msg.get("v", ""), msg.get("v_min", ""))
        except HandshakeError as refusal:
            self._send({"type": "error", "detail": str(refusal),
                        "code": refusal.code})
            return
        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"):
            log.warning("Handshake refused — no GEK for group=%s", peer.group_id[:8])
            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
        log.info("WebRTC handshake challenge sent (peer=%s)", self._peer_id)
        self._send({
            "type": MNP.HANDSHAKE_CHALLENGE,
            "v": MNP_VERSION,
            # Our half of the range. The client refuses us on this rather than
            # discovering the mismatch when a field it expected is not there.
            "v_min": MNP_MIN_SUPPORTED,
            "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(),
            # ...except that since 3.4 it is signed, so a client that already
            # knows which key to expect can check it before it sends a code.
            **self._challenge_sig(peer.group_id, self._channel_binding()),
        })

    def _challenge_sig(self, group_id: str, binding: bytes) -> dict:
        """
        `{"sig": ...}` over the challenge transcript, or nothing (MNP 3.4).

        What makes `node_pk` above more than an announcement: a client about to
        send an invitation code can check this node holds the key it was told
        to expect, before the code leaves. No binding means no signature rather
        than an unbound one — a signature that is not tied to the channel is one
        somebody can relay, and the handshake proof refuses that case anyway.
        """
        if not binding:
            return {}
        transcript = challenge_transcript(
            group_id, self._nonce_client, self._gek_challenge, binding)
        return {"sig": base64.b64encode(self._ctx["sk_node"].sign(transcript)).decode()}

    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
        self._spawn(self._load_pinned_pk())

        self._register_peer()

        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)

        # Everything the client needs in order to *authenticate* us stays in clear —
        # node_pk, proof and sig are what it checks before it would trust a
        # decryption, so they cannot themselves be behind one. The configuration
        # below is sealed under a GEK-derived subkey, which gives it an
        # authentication tag from a key the hub does not hold. Until MNP 1.0 the
        # signed transcript named no ack field at all, so is_node_admin,
        # enabled_apps, the app directories and the rest were authenticated by DTLS
        # channel and nothing else.
        config = {
            "is_node_admin": self._is_node_admin(),
            # Which group "applications" to show. Absent/empty falls back to
            # every registered one client-side, so a node that predates this
            # setting (or one whose context has not loaded it yet) hides
            # nothing.
            "enabled_apps": list(self._group_ctx().get("enabled_apps") or []),
            # Read once and kept current in place by the signed op, and
            # surfaced here rather than only via tmdb_enabled_ack, so a client
            # that connects after the operator configured it does not have to
            # wait for a live change to find out.
            "tmdb_enabled": bool(self._group_ctx().get("tmdb_enabled", True)),
            # Token/language stay node-wide (one shared credential/cache) —
            # via daemon_state, kept current by tmdb_config_ack.
            "tmdb_token_customized": bool(
                self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)),
            "tmdb_language": str(
                self._ctx.get("daemon_state", {}).get("tmdb_language") or ""),
            # Music app (docs/MESHBAY_DESIGN.md §9.8) — same shape as the TMDB
            # fields above. No language field: MusicBrainz search doesn't
            # take one the way TMDB does.
            "musicbrainz_enabled": bool(self._group_ctx().get("musicbrainz_enabled", True)),
            # Every app's configured folders — `<app>_directories`, keyed by
            # the app's own registry name, always a list. Built by daemon.py's
            # `_app_directories_ctx`, and the only form on the wire: the
            # `video_root` / `audio_root` / `photo_roots` scalars that used to
            # ride here are gone. One folder was never the general case, and two
            # spellings of one answer meant whichever the reader consulted first
            # decided it.
            #
            # `chat_directory` below is the one surviving second name, and it is
            # safe for the reason those were not: it is *derived* from this list
            # on every build rather than stored beside it, so the two cannot
            # drift apart.
            **self._app_directories_ack(),
            # Where chat attachments are written — the singular form, because
            # Chat genuinely has one destination. "" means the operator has not
            # chosen, and the paperclip says so.
            "chat_directory": self._group_ctx().get("chat_directory") or "",
            # Whether the node unfurls links members post here. Absent means
            # on, which is what it did before this existed.
            "chat_link_preview": bool(
                self._group_ctx().get("chat_link_preview", True)),
            # Whether the reader's cross-group Search should list this group.
            # Presentation only: the index below is served to Search and to the
            # group page alike, and this cannot tell them apart. Sealed like the
            # rest, so the hub cannot flip it. Absent means listed.
            "search_listed": bool(self._group_ctx().get("search_listed", True)),
            # Which chat epoch key a client should be sealing under. Inside
            # the sealed part of the ack like every other configuration field,
            # so it carries an authentication tag from a key the hub does not
            # hold — a forged epoch would have a client sealing under a key the
            # group has retired.
            #
            # No `chat_encrypted` beside it: there is no switch. A peer that
            # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat.
            "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0),
            # This member's own transfer caps in this group, so the interface
            # can say "2 of 2 of your slots are busy" rather than draw a bare
            # spinner. Absent reads as "no limit known" and the hint is simply
            # not drawn — never as "unlimited", which would have the interface
            # contradicting the node.
            "transfer_limits": {
                "download": self._slots().member_cap(
                    transfers_mod.DOWNLOAD,
                    (self._group_id or "", self._user_id or "")),
                "upload": self._slots().member_cap(
                    transfers_mod.UPLOAD,
                    (self._group_id or "", self._user_id or "")),
            },
            # So a client that connects mid-scan shows the indexing state
            # immediately, instead of waiting for the next periodic
            # INDEX_PROGRESS push. Never a path or filename — see
            # IndexProgress in indexer.py.
            "indexing": self._indexing_status(),
            # Current values only — not enforced from here, just shown to
            # the operator in Settings so the number on screen matches what
            # the indexer is actually doing (set_scan_settings, ops/settings.py).
            "scan_settings": {
                "reconcile_interval_secs": self._group_ctx().get(
                    "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS),
                "debounce_secs": self._group_ctx().get(
                    "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS),
            },
        }
        if node_user_id:
            config["node_user_id"] = node_user_id
        pk_x_b64 = self._ctx.get("pk_x25519_b64")
        if pk_x_b64:
            config["node_pk_x25519"] = pk_x_b64

        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(),
            **seal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, self._group_id or "", config),
        }
        self._send(ack)
        self._audit("handshake")

        # Someone is here now — reconcile's backstop should be prompt again
        # rather than however far its backoff had stretched while nobody
        # was connected (indexer.py DirectoryIndexer.note_activity).
        note_activity = self._group_ctx().get("note_activity")
        if note_activity:
            note_activity()

    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)
        self._spawn(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)
            self._spawn(audit.log_event(
                user_id="unknown",
                event="auth_failed",
                ip=self._remote_ip,
                group_id=group_id,
                detail=reason,
            ))

    def _indexing_status(self) -> dict:
        """
        The counters of `_push_index_progress` (daemon.py), for the handshake
        ack — never a path, a filename or a root name, which stay local to the
        operator's own admin UI. Absent "progress" (context not loaded, or a
        group with no indexer at all) reads as idle rather than erroring.
        """
        progress = self._group_ctx().get("progress")
        if progress is None:
            return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0,
                    "files_done": 0, "files_total": 0, "kind": "",
                    "root_pos": -1, "queued": 0}
        return {
            "scanning": progress.scanning,
            "scanned_bytes": progress.scanned_bytes,
            "total_bytes": progress.total_bytes,
            "files_done": progress.files_done,
            "files_total": progress.files_total,
            "kind": progress.kind,
            "root_pos": progress.root_pos,
            "queued": len(progress.queued),
        }