path: root/packages/meshbay-common/src/meshbay_common/protocol.py

"""
MeshBay protocol constants and message type definitions.

MNP (Mesh Node Protocol) — v0.1
MHP (Mesh Bay Hub Protocol) — v0.1

All wire messages are length-prefixed msgpack (4-byte big-endian length header).
Every message carries a "v" field for protocol version.
"""

from dataclasses import dataclass, field
from typing import Any

# The wire versions live in meshbay_common/__init__.py — one source, because a
# second copy here said "0.1" while every message on the wire carried "0.2".
# Nothing imported it, which is the only reason it was harmless.
from meshbay_common import MNP_VERSION, MHP_VERSION   # noqa: F401  (re-export)
from meshbay_common.webcrypto import (
    chunk_key_aes,
    decrypt_chunk_aes,
    encrypt_chunk_aes,
)


# ── MNP message types ─────────────────────────────────────────────────────────

class MNP:
    HANDSHAKE       = "handshake"
    HANDSHAKE_ACK   = "handshake_ack"
    INDEX_SYNC      = "index_sync"       # full Mesh Group Index
    INDEX_DELTA     = "index_delta"      # incremental update
    # Node -> already-connected members: "the operator's node is scanning
    # right now, N/M bytes done". Never the entries themselves (that is
    # INDEX_SYNC/INDEX_DELTA's job) — just enough for a presence dot to
    # animate. Pushed periodically while scanning, and once more on the
    # transition back to idle, so the indicator is guaranteed to turn off.
    INDEX_PROGRESS  = "index_progress"
    FILE_REQUEST    = "file_req"         # request chunk(s)
    FILE_CHUNK      = "file_chunk"       # encrypted chunk response
    STREAM_SEGMENT  = "stream_seg"       # HLS/DASH segment
    # Not a Double Ratchet message, and never was — `first-review.md` C1
    # rejected exactly that for groups. Plaintext until a group turns
    # encryption on, then AES-256-GCM under a per-device subkey of the group's
    # chat epoch key, signed with the sending device's pinned Ed25519 key
    # (`chatbox.py`, docs/chat-sender-keys.md).
    CHAT_MESSAGE    = "chat_msg"         # one chat message, plain or sealed
    CHAT_ATTACHMENT = "chat_attach"      # attachment metadata
    CHAT_HISTORY    = "chat_hist"       # request message history (newest, or before a cursor)
    CHAT_HISTORY_RESPONSE = "chat_hist_resp"  # history response with messages
    # Link unfurl: the node fetches a URL a member pasted and returns an
    # OpenGraph card. Additive (0.12) — an older node just logs "unknown type"
    # and the client shows the bare link, exactly as before.
    LINK_PREVIEW_REQ  = "link_preview_req"    # client → node: unfurl this URL
    LINK_PREVIEW_RESP = "link_preview_resp"   # node → client: card fields, or ok:false
    # Liveness on an *already open* channel. A peer that goes away without
    # closing leaves a DataChannel that still reads as connected until the next
    # real request hangs, and there was no way to ask. This is not a discovery
    # mechanism: opening a connection in order to ping costs a full ICE/DTLS
    # handshake (measured at 0.6-7 s across two ISPs), so presence in the group
    # list comes from the hub's socket registry instead.
    PING            = "ping"
    PONG            = "pong"
    # GEK_REQUEST / GEK_RESPONSE were removed (NS3, and finding L1): the node must
    # never serve the GEK in plaintext. Members obtain it by unwrapping their own
    # ECIES bundle. The constants lingered after the handlers were deleted, leaving
    # the wire contract looking as though the endpoint still existed.
    FILE_UPLOAD      = "file_upload"       # client pushes file chunk to node
    FILE_UPLOAD_ACK  = "file_upload_ack"   # node acknowledges chunk receipt
    DIR_CREATE       = "dir_create"        # client → node: make a directory
    DIR_CREATE_ACK   = "dir_create_ack"    # node → client: created
    DIR_DELETE       = "dir_delete"        # client → node: remove an empty directory
    DIR_DELETE_ACK   = "dir_delete_ack"    # node → client: removed
    FILE_DELETE      = "file_delete"       # client requests file deletion
    FILE_DELETE_ACK  = "file_delete_ack"   # node confirms deletion
    STREAM_REQUEST   = "stream_req"        # client requests MSE video stream
    STREAM_INIT      = "stream_init"       # node sends codec info + signals stream start
    STREAM_DATA      = "stream_data"       # node sends encrypted fMP4 segment
    STREAM_END       = "stream_end"        # node signals end of stream
    STREAM_MORE      = "stream_more"       # client → node: room for N more segments
    STREAM_STOP      = "stream_stop"       # client → node: nobody is watching any more
    EPHEMERAL_STREAM = "ephemeral_stream"  # reserved — mobile live push
    HANDSHAKE_CHALLENGE  = "handshake_challenge"   # node → client: GEK proof nonce
    HANDSHAKE_RESPONSE   = "handshake_response"    # client → node: HMAC(GEK, nonce)
    ADMIN_CHALLENGE      = "admin_challenge"       # node → client: Ed25519 sign challenge
    ADMIN_RESPONSE       = "admin_response"        # client → node: Ed25519 signature
    # GEK_BUNDLE_STORE was removed with the invite redesign: the node wraps the GEK
    # itself, for a key the recipient proved possession of, so no member ever hands
    # the node key material (C5b, and the H3 substitution it enabled).
    GEK_BUNDLE_FETCH     = "gek_bundle_fetch"      # client → node: request own wrapped GEK
    GEK_BUNDLE_RESP      = "gek_bundle_resp"       # node → client: wrapped GEK bundle
    KEYPAIR_BUNDLE_STORE = "keypair_bundle_store"  # client → node: store encrypted keypair bundle
    #   optional `bundle_enc_recovery` (MNP 0.14): a second copy wrapped under the
    #   account's recovery key (docs/auth-confirm.md §4.3)
    KEYPAIR_BUNDLE_FETCH = "keypair_bundle_fetch"  # client → node: request own keypair bundle
    KEYPAIR_BUNDLE_RESP  = "keypair_bundle_resp"   # node → client: encrypted keypair bundle
    #   carries `bundle_enc_recovery` too when the node has one stored
    KEYPAIR_BUNDLE_DELETE = "keypair_bundle_delete"  # client → node: withdraw own backup
    JOIN_REQUEST         = "join_request"          # client → node: pair/recognise this identity
    JOIN_RESULT          = "join_result"           # node → client: outcome + wrapped GEK
    INVITE_CREATE        = "invite_create"         # operator → node: issue a pairing code
    MEMBER_REVOKE        = "member_revoke"         # operator → node: stop serving the key
    MEMBER_REVOKE_ACK    = "member_revoke_ack"
    MEMBER_UNPIN         = "member_unpin"          # operator → node: forget an identity
    MEMBER_UNPIN_ACK     = "member_unpin_ack"
    MEMBER_UPLOAD        = "member_upload"         # operator → node: may members upload?
    MEMBER_UPLOAD_ACK    = "member_upload_ack"
    APPS_ENABLED         = "apps_enabled"          # operator → node: which group apps to show
    APPS_ENABLED_ACK     = "apps_enabled_ack"
    SET_SCAN_SETTINGS     = "set_scan_settings"     # operator → node: reconcile/debounce timing
    SET_SCAN_SETTINGS_ACK = "set_scan_settings_ack"
    MEDIA_META_REQ        = "media_meta_req"        # client → node: TMDB metadata for a path
    MEDIA_META_RESP       = "media_meta_resp"       # node → client: TMDB metadata (or none)
    VIDEO_ROOT            = "video_root"             # operator → node: which folder is the Videos entry point
    VIDEO_ROOT_ACK        = "video_root_ack"
    TMDB_CONFIG           = "tmdb_config"            # operator → node: set custom TMDB token/language (node-wide)
    TMDB_CONFIG_ACK       = "tmdb_config_ack"        # node → everyone: new TMDB config (never the token)
    TMDB_ENABLED          = "tmdb_enabled"           # operator → node: enable/disable TMDB for this group
    TMDB_ENABLED_ACK      = "tmdb_enabled_ack"       # node → this group: new per-group TMDB enabled state
    SEASON_META_REQ       = "season_meta_req"        # client → node: TMDB overview/poster for one season
    SEASON_META_RESP      = "season_meta_resp"       # node → client: season-level TMDB fields (or none)
    TMDB_SEARCH_REQ       = "tmdb_search_req"        # client → node: candidate TMDB matches for a query
    TMDB_SEARCH_RESP      = "tmdb_search_resp"       # node → client: candidate list (id, title, year, poster)
    TMDB_OVERRIDE         = "tmdb_override"          # operator → node: replace a show/movie's TMDB match
    TMDB_OVERRIDE_ACK     = "tmdb_override_ack"
    TMDB_REMATCH          = "tmdb_rematch"           # operator → node: drop one file's match
    TMDB_REMATCH_ACK      = "tmdb_rematch_ack"
    # Music app (docs/musicbay.md). Contact is derived from the owner's hub
    # email at login — no config/ack pair needed. Only the per-group toggle
    # remains.
    MUSICBRAINZ_ENABLED = "musicbrainz_enabled"          # operator → node: enable/disable
    MUSICBRAINZ_ENABLED_ACK = "musicbrainz_enabled_ack"  # node → this group: new enabled state
    MUSIC_META_REQ = "music_meta_req"                    # client → node: metadata for a path
    MUSIC_META_RESP = "music_meta_resp"                  # node → client: metadata (or none)
    # A file whose format the browser's own <audio> element cannot decode at
    # all (WMA, Musepack) — resolved to a cached, browser-playable AAC/M4A
    # copy on request, the same "computed once, reused forever" shape as a
    # TMDB poster or a MusicBrainz cover.
    AUDIO_TRANSCODE_REQ = "audio_transcode_req"          # client → node: transcode this file id
    AUDIO_TRANSCODE_RESP = "audio_transcode_resp"        # node → client: cache hash/size/mime
    # Which folder is the Music app's entry point for this group — same
    # shape as VIDEO_ROOT above.
    AUDIO_ROOT            = "audio_root"                  # operator → node: which folder is the Music entry point
    AUDIO_ROOT_ACK        = "audio_root_ack"
    # Which folder(s) are the Photos app's entry points for this group — a
    # *set*, unlike VIDEO_ROOT/AUDIO_ROOT above, since a photo library is
    # routinely scattered across several unrelated folders (docs/photos.md §2.1).
    PHOTO_ROOTS           = "photo_roots"                 # operator → node: the whole root set, replaced
    PHOTO_ROOTS_ACK       = "photo_roots_ack"
    # Device linking. A new device files a request bound to a code it displays;
    # an already-pinned device of the same account approves it. Neither the hub
    # nor the node can produce the countersignature.
    DEVICE_REQUEST       = "device_add_request"    # new device → node
    DEVICE_REQUEST_ACK   = "device_add_request_ack"
    DEVICE_LOOKUP        = "device_lookup"         # approver → node: find by code
    DEVICE_LOOKUP_RESULT = "device_lookup_result"
    DEVICE_ADD           = "device_add"            # approver → node: countersigned
    DEVICE_ADD_ACK       = "device_add_ack"
    DEVICE_LIST          = "device_list"           # anyone → node: my devices
    DEVICE_LIST_RESULT   = "device_list_result"
    DEVICE_REVOKE        = "device_revoke"         # a device retires another
    # "Which of this account's devices am I?" — signed, on an
    # already-authenticated connection. The handshake proves the account and the
    # group; it never proved the device, so the node attributed uploads to the
    # account's oldest key and could not tell one device's chat from another's.
    # Additive (MNP 1.2): a client that stays silent leaves the node exactly
    # where it was.
    DEVICE_HELLO         = "device_hello"          # device → node: this is me
    DEVICE_HELLO_ACK     = "device_hello_ack"
    # Rotation is the half of revocation that revocation cannot do: the node
    # generates a fresh key itself, so no key material crosses the wire.
    GEK_ROTATE           = "gek_rotate"            # operator → node: new group key
    GEK_ROTATE_ACK       = "gek_rotate_ack"
    INVITE_RESULT        = "invite_result"         # node → operator: the code, once
    NODE_STATUS          = "node_status"           # operator → node: list all groups + roots
    NODE_STATUS_ACK      = "node_status_ack"       # node → operator: full status
    ROOT_ADD             = "root_add"              # operator → node: add a directory to a group
    ROOT_ADD_ACK         = "root_add_ack"          # node → operator: confirmed
    ROOT_REMOVE          = "root_remove"           # operator → node: remove a root from a group
    ROOT_REMOVE_ACK      = "root_remove_ack"       # node → operator: confirmed
    # One message for every application's directories, keyed by the app's own
    # name — adding an app adds no message type. VIDEO_ROOT / AUDIO_ROOT /
    # PHOTO_ROOTS above are the same instruction under three earlier names and
    # are still handled, for clients that predate this.
    APP_DIRECTORIES       = "app_directories"      # operator → node: an app's folder(s)
    APP_DIRECTORIES_ACK   = "app_directories_ack"
    # Chat's own two: where attachments are written (a destination, so it must
    # be a read-write root), and whether the node unfurls links members post.
    CHAT_DIRECTORY        = "chat_directory"
    CHAT_DIRECTORY_ACK    = "chat_directory_ack"
    CHAT_LINK_PREVIEW     = "chat_link_preview"
    CHAT_LINK_PREVIEW_ACK = "chat_link_preview_ack"
    # The keys a group's chat archive is encrypted under, on their way to a
    # member. Sealed under a group-derived subkey, so the payload carries an
    # authentication tag from a key the hub does not hold — and a member who has
    # not completed the handshake is served a ciphertext rather than the keys.
    # Requested rather than pushed on the ack: a group with no chat should not
    # pay for this on every connection.
    CHAT_KEYS_REQ         = "chat_keys_req"
    CHAT_KEYS_RESP        = "chat_keys_resp"
    # Node → this group: a new chat epoch was opened, because somebody was
    # removed. Not a setting — there is no switch; chat is always encrypted
    # (MNP 2.0). Pushed so a connected client stops sealing under the retired
    # key without having to reconnect.
    CHAT_EPOCH            = "chat_epoch"
    CHAT_EPOCH_ACK        = "chat_epoch_ack"
    ROOT_UPDATE          = "root_update"           # operator → node: change writable/removable on a root
    ROOT_UPDATE_ACK      = "root_update_ack"
    ROOT_EJECT           = "root_eject"            # operator → node: mark removable root as ejected
    ROOT_EJECT_ACK       = "root_eject_ack"
    ROOT_PLUG            = "root_plug"             # operator → node: re-enable an ejected root
    ROOT_PLUG_ACK        = "root_plug_ack"
    ROSTER_READ          = "roster_read"            # operator → node: list pinned identities + members
    ROSTER_READ_ACK      = "roster_read_ack"
    DENYLIST_READ        = "denylist_read"          # operator → node: show denylist entries
    DENYLIST_READ_ACK    = "denylist_read_ack"
    DENYLIST_CLEAR       = "denylist_clear"         # operator → node: remove denylist entry(ies)
    DENYLIST_CLEAR_ACK   = "denylist_clear_ack"
    GROUP_ATTACH         = "group_attach"           # operator → node: host a new group
    GROUP_ATTACH_ACK     = "group_attach_ack"
    GROUP_DETACH         = "group_detach"           # operator → node: stop hosting a group
    GROUP_DETACH_ACK     = "group_detach_ack"
    NODE_SETTINGS_SET    = "node_settings_set"      # operator → node: change daemon settings
    NODE_SETTINGS_SET_ACK = "node_settings_set_ack"
    NODE_RELOAD          = "node_reload"            # operator → node: re-read node.toml
    NODE_RELOAD_ACK      = "node_reload_ack"


# ── Index entry ───────────────────────────────────────────────────────────────

@dataclass
class IndexEntry:
    id:        str    # blake3 hash of file (hex)
    name:      str    # filename
    path:      str    # path relative to shared directory
    size:      int    # bytes
    type:      str    # video | audio | image | document | archive | other
    added_at:  int    # unix timestamp
    duration:  int | None = None   # seconds, for media
    thumb_hash: str | None = None  # blake3 of thumbnail
    uploader_id: str | None = None  # user_id of who uploaded (None = pre-existing on disk)
    uploader_pk: str | None = None  # Ed25519 public key of uploader (base64 raw 32 bytes)
    width:      int | None = None  # pixels, video/image
    height:     int | None = None  # pixels, video/image
    display_title: str | None = None  # parsed or cleaned-filename title, Videos app
    season:     int | None = None  # parsed season number, Videos app
    episode:    int | None = None  # parsed episode number, Videos app
    artist:     str | None = None  # tag or parsed, Music app
    album:      str | None = None  # tag or parsed, Music app
    track_no:   int | None = None  # tag or parsed, Music app
    taken_at:   int | None = None  # unix timestamp, EXIF DateTimeOriginal — Photos app
    camera:     str | None = None  # "Make Model", when both present — Photos app
    hash_version: int = 1          # 1 = full-file blake3, 2 = partial-read (45 MB sample)


def index_entry_wire(e: IndexEntry) -> dict:
    """
    The wire-dict shape used by INDEX_SYNC/INDEX_DELTA hand-built messages
    (as opposed to GroupIndex.serialize()'s asdict() encoding of the whole
    index). Centralized so the three call sites that build these
    (webrtc_server._do_index_sync, daemon._broadcast_index_change's two
    branches) can't drift from each other as fields are added.
    """
    return {
        "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,
        "duration": e.duration, "thumb_hash": e.thumb_hash,
        "width": e.width, "height": e.height,
        "display_title": e.display_title,
        "season": e.season, "episode": e.episode,
        "artist": e.artist, "album": e.album, "track_no": e.track_no,
        "taken_at": e.taken_at, "camera": e.camera,
        "hash_version": e.hash_version,
    }


@dataclass
class IndexDelta:
    base_version: int
    version:      int
    additions:    list[IndexEntry] = field(default_factory=list)
    deletions:    list[str] = field(default_factory=list)  # list of ids
    # Same id as before (same file, same content hash), different field
    # values — e.g. the Videos app's async enrichment filling in duration/
    # thumb_hash/etc. after the file was already indexed with hash+size only.
    # A distinct list from `additions`: `GroupIndex.diff()` only ever adds an
    # id here once it has already appeared unchanged in a prior snapshot.
    updates:      list[IndexEntry] = field(default_factory=list)


# ── File chunk ────────────────────────────────────────────────────────────────
#
# One encoder and one decoder, for every transport.
#
# There used to be two. WebRTC moved to a binary wire format in Phase 9.15 (base64
# costs 33%) and dropped the per-chunk Ed25519 signature with it; the QUIC encoder
# was not brought along, so `file_chunk` meant two different messages depending on
# which transport carried it — base64 fields, hashes and a signature on one, raw
# bytes and a `file_id` on the other. Nothing in the type name said which, and the
# dataclass that used to sit here described only the QUIC half while reading like
# the contract for both. That is finding C6 one size down: two implementations of
# one message, free to drift, with a test for neither.
#
# Why no per-chunk signature: the AES-GCM tag already authenticates the ciphertext
# under a key derived from the GEK, which only group members hold, and since C3 the
# node authenticates itself in the handshake and is pinned by the client. A
# signature per chunk re-proved, once per megabyte, what the session established
# once. (`sign_chunk` still exists in `crypto.py` — it signs the serialized index
# envelope, which is a different artifact; see `indexer/group_index.py`.)


def chunk_ciphertext(
    gek: bytes,
    plaintext: bytes,
    chunk_index: int,
    file_hash: bytes,
) -> tuple[bytes, bytes]:
    """
    `(nonce, ciphertext)` for one chunk.

    Split out because `stream_data` encrypts exactly like `file_chunk` — same key
    derivation, indexed by segment instead of by chunk — and differs only in the
    message it lands in. It used to do so through its own copy of these two lines.
    """
    ckey = chunk_key_aes(gek, file_hash, chunk_index)
    return encrypt_chunk_aes(ckey, plaintext)


def file_chunk_wire(
    gek: bytes,
    plaintext: bytes,
    chunk_index: int,
    file_hash: bytes,
    file_id: str = "",
) -> dict:
    """
    Encrypt one chunk and build the `file_chunk` message.

    `file_hash` is the raw content hash the chunk key is derived from; `file_id` is
    the same value hex-encoded, echoed so a client running several downloads at once
    can tell whose reply arrived. Serving a thumbnail or a cached transcode passes
    the cache blob's own hash for both.
    """
    nonce, ct = chunk_ciphertext(gek, plaintext, chunk_index, file_hash)
    return {
        "type": MNP.FILE_CHUNK,
        "v": MNP_VERSION,
        "file_id": file_id,
        "chunk_index": chunk_index,
        "plaintext_size": len(plaintext),
        "nonce": nonce,
        "ct": ct,
    }


def file_chunk_plaintext(
    gek: bytes,
    msg: dict,
    file_hash: bytes | None = None,
) -> bytes:
    """
    Decrypt a `file_chunk`. Raises `InvalidTag` if the ciphertext was tampered with.

    `file_hash` defaults to the message's own `file_id`, which is what a client that
    asked for a whole file already has. Pass it explicitly only when the caller knows
    better than the peer does.
    """
    if file_hash is None:
        file_hash = bytes.fromhex(msg["file_id"])
    ckey = chunk_key_aes(gek, file_hash, msg["chunk_index"])
    return decrypt_chunk_aes(ckey, msg["nonce"], msg["ct"])