/** * MeshBay Browser Transport — WebRTC DataChannel client. * * Connects to a MeshBay node via WebRTC DataChannel (P2P, E2E). * The hub is only used for signaling (SDP/ICE relay) — after connection, * all data flows directly between browser and node. * * Wire format: length-prefixed msgpack (4-byte big-endian + msgpack payload). * Same format as QUIC and TCP+TLS transports on the node side. * * Usage: * const transport = new MeshBayTransport(hubUrl, accessToken); * await transport.connect(nodeId, jwtToken, groupId); * const index = await transport.fetchIndex(); * const chunk = await transport.fetchChunk(fileId, 0); * transport.close(); */ async function _pkFromSk(skPkcs8B64) { const raw = Uint8Array.from(atob(skPkcs8B64), c => c.charCodeAt(0)); const sk = await crypto.subtle.importKey('pkcs8', raw, { name: 'X25519' }, true, ['deriveBits']); const jwk = await crypto.subtle.exportKey('jwk', sk); const b64url = jwk.x; const b64 = b64url.replace(/-/g, '+').replace(/_/g, '/'); const pad = b64.length % 4; return pad ? b64 + '='.repeat(4 - pad) : b64; } async function _pkEdFromSk(skPkcs8B64) { const raw = Uint8Array.from(atob(skPkcs8B64), c => c.charCodeAt(0)); const sk = await crypto.subtle.importKey('pkcs8', raw, { name: 'Ed25519' }, true, ['sign']); const jwk = await crypto.subtle.exportKey('jwk', sk); const b64 = jwk.x.replace(/-/g, '+').replace(/_/g, '/'); const pad = b64.length % 4; return pad ? b64 + '='.repeat(4 - pad) : b64; } // 48 KB is what fits comfortably in one SCTP message across stacks; the window is // what makes the rate independent of the round trip. 32 × 48 KB = 1.5 MB in // flight, which saturates any path up to roughly 100 Mb/s at 100 ms. const UPLOAD_CHUNK_SIZE = 48 * 1024; const UPLOAD_WINDOW = 32; const UPLOAD_BUFFER_HIGH = 1024 * 1024; // Segments of 256 KB: 24 in flight is 6 MB, enough to keep playback fed over a // slow link and small enough that nothing accumulates. // How long to collect ICE candidates before sending the offer anyway. Long // enough for a STUN round trip on a slow link, short enough that a STUN server // that never answers costs a pause rather than the whole attempt. const ICE_GATHER_TIMEOUT_MS = 4000; const STREAM_CREDITS = 24; function _aborted() { const err = new Error('Cancelled'); err.name = 'AbortError'; return err; } const JOIN_REFUSALS = { code_required: 'This node does not know this browser yet. Ask the node operator ' + 'for a pairing code (meshbay-node operator pair).', code_invalid: 'That pairing code is not valid — it may be mistyped, expired, ' + 'already used, or issued for a different account.', key_changed: 'This account is already paired with a different key on this node. ' + 'If you reset your keys, the operator must unpin you before pairing again.', not_authorized_for_group: 'The node does not list you as a member of this group. ' + 'Being a member on the hub is not enough — ask the operator for an invite.', no_gek: 'This group has no key yet. The node operator must run ' + '`meshbay-node gek-init` for it.', signature_invalid: 'The node rejected the signature over your keys.', stale_request: 'Your clock is too far from the node\'s — check the system time.', group_mismatch: 'The node refused a request naming a different group.', }; class MeshBayTransport { constructor(hubUrl, accessToken) { this._hubUrl = hubUrl; this._accessToken = accessToken; this._pc = null; this._channel = null; this._pending = new Map(); this._seqId = 0; this._recvBuf = new Uint8Array(0); this._connected = false; this._onChat = null; this._onStreamInit = null; this._onStreamData = null; this._onStreamEnd = null; this._onStreamError = null; this._onIndexSync = null; // filename → the uploader waiting on it. Keyed rather than FIFO because // several uploads may be in flight at once and their acks interleave; the // node names the file in every one. this._uploaders = new Map(); } get connected() { return this._connected; } set onChat(fn) { this._onChat = fn; } set onStreamInit(fn) { this._onStreamInit = fn; } set onStreamData(fn) { this._onStreamData = fn; } set onStreamEnd(fn) { this._onStreamEnd = fn; } set onStreamError(fn) { this._onStreamError = fn; } set onIndexSync(fn) { this._onIndexSync = fn; } set onIndexDelta(fn) { this._onIndexDelta = fn; } set onUploadPolicy(fn) { this._onUploadPolicy = fn; } set onAppsEnabled(fn) { this._onAppsEnabled = fn; } set onTmdbConfig(fn) { this._onTmdbConfig = fn; } set onTmdbEnabled(fn) { this._onTmdbEnabled = fn; } set onVideoRoot(fn) { this._onVideoRoot = fn; } set onIndexProgress(fn) { this._onIndexProgress = fn; } get sessionKeys() { return this._sessionKeys; } /** Set on a first join: the identity created for this node, still to be left with it. */ get newNodeBundle() { return this._newNodeBundle || null; } set newNodeBundle(v) { this._newNodeBundle = v; } async connect(nodeId, jwtToken, groupId, gekRaw, sessionKeys, bundleKey, username, userId, joinCode) { this._gekRaw = gekRaw || null; this._sessionKeys = sessionKeys || null; this._bundleKey = bundleKey || null; this._username = username || null; this._userId = userId || null; this._newNodeBundle = null; this._joinError = null; this._pc = new RTCPeerConnection({ iceServers: [{ urls: 'stun:stun.l.google.com:19302' }], }); this._channel = this._pc.createDataChannel('mnp', { ordered: true }); this._channel.binaryType = 'arraybuffer'; let channelReject = null; const channelReady = new Promise((resolve, reject) => { channelReject = reject; const timeout = setTimeout(() => reject(new Error('DataChannel open timeout')), 30000); this._channel.onopen = () => { clearTimeout(timeout); this._connected = true; resolve(); }; }); this._channel.onmessage = (event) => this._onMessage(event.data); this._channel.onclose = (ev) => { console.warn('[MeshBay] DataChannel closed', this._channel?.readyState, ev); this._connected = false; if (channelReject) channelReject(new Error('DataChannel closed')); for (const [, p] of this._pending) p.reject(new Error('DataChannel closed')); this._pending.clear(); }; this._channel.onerror = (ev) => { console.error('[MeshBay] DataChannel error', ev); if (channelReject) channelReject(new Error('DataChannel error')); }; this._pc.onconnectionstatechange = () => { console.log('[MeshBay] PC state:', this._pc.connectionState); }; this._pc.oniceconnectionstatechange = () => { console.log('[MeshBay] ICE state:', this._pc.iceConnectionState); }; const offer = await this._pc.createOffer(); await this._pc.setLocalDescription(offer); // Wait for candidates, but not indefinitely. // // This is non-trickle signaling: the offer carries its candidates, so the // SDP is only sent once gathering is done. When gathering *never* finishes // — a STUN server that is slow, filtered, or being resolved through a DNS // that is not answering — this promise never settles, and joining a group // hangs with no error and nothing on screen. Reported after exactly that, // and it succeeded on a later attempt, which is the shape of a network // wait rather than a refusal. // // Past the deadline the offer goes out with whatever has been gathered. // Host candidates are already there, which is enough on a LAN — the case // this project cares most about — and the reflexive ones normally arrive // in well under a second when STUN is reachable at all. A partial offer // that usually connects beats a promise that never returns. await new Promise((resolve) => { if (this._pc.iceGatheringState === 'complete') return resolve(); const done = () => { clearTimeout(timer); resolve(); }; const timer = setTimeout(() => { console.warn('[MeshBay] ICE gathering did not finish in', ICE_GATHER_TIMEOUT_MS, 'ms — offering what we have'); done(); }, ICE_GATHER_TIMEOUT_MS); this._pc.onicegatheringstatechange = () => { if (this._pc.iceGatheringState === 'complete') done(); }; }); // Signaling is a hub call like any other, so it goes the same way — in the // application that means through the main process, because the renderer's // app:// origin is refused by CORS. const call = (window.MeshBayPlatform && window.MeshBayPlatform.apiFetch) || fetch; const resp = await call( `${this._hubUrl}/v1/nodes/${nodeId}/webrtc/offer`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${this._accessToken}`, }, body: JSON.stringify({ sdp: this._pc.localDescription.sdp, ice_candidates: [], }), }); if (!resp.ok) { const detail = await resp.json().catch(() => ({})); throw new Error(`Signaling failed: ${resp.status} ${detail.detail || ''}`); } const answer = await resp.json(); this._rawAnswerSdp = answer.sdp; await this._pc.setRemoteDescription({ type: 'answer', sdp: answer.sdp }); await channelReady; console.log('[MeshBay] DataChannel open, sending handshake for group', groupId, 'channel=', this._channel?.readyState, 'crypto=', !!window.MeshBayCrypto); // The client nonce is what makes the NODE's proof fresh (C3) — without it a // recorded handshake_ack could be replayed by an impersonating peer. this._nonceClient = crypto.getRandomValues(new Uint8Array(32)); const reply = await this._sendAndWait({ type: 'handshake', v: '0.1', token: jwtToken, group_id: groupId || '', nonce: window.MeshBayCrypto.b64encode(this._nonceClient), }); console.log('[MeshBay] Handshake reply:', reply.type); if (reply.type === 'handshake_challenge') { if (!window.MeshBayCrypto) { throw new Error('Node requires GEK proof but no crypto available'); } // Recorded the moment the challenge arrives, because everything below may // need them — joining, in particular, happens before the proof and signs a // transcript over both. Reading them further down, next to the proof that // also uses them, meant join_request ran with neither. // // nonce_node ties a join to this connection, so one cannot be lifted onto // another. node_pk is announced here because a first-time member has no // GEK and so cannot complete the handshake that would prove it; it is // unverified at this point and checked against the ack below. this._nonceNode = window.MeshBayCrypto.b64decode(reply.nonce); this.nodePk = reply.node_pk || null; // Our identity for THIS node: fetched from it, or created if this is a // first join. Keys are per node, so there is nothing to carry between // them — and an operator who cracks the copy on their own disk gets a key // that opens nothing anywhere else. let fresh = false; if (!this._sessionKeys && this._bundleKey && window.MeshBayKeys) { const kpResp = await this._sendAndWait({ type: 'keypair_bundle_fetch', v: '0.1', }); if (kpResp.type === 'keypair_bundle_resp' && kpResp.found) { const keys = await window.MeshBayKeys.decryptBundleWithKey( kpResp.bundle_enc, this._bundleKey); const pkXB64 = await _pkFromSk(keys.skX); this._sessionKeys = { skXB64: keys.skX, skEdB64: keys.skEd, pkXB64 }; } else { // This node has never seen us. Generate the identity we will use here // and nowhere else; it is stored on this node once the join succeeds, // which is what lets another browser become the same person here. const id = await window.MeshBayKeys.generateNodeIdentity(this._bundleKey); this._sessionKeys = { skEdB64: id.skEdB64, skXB64: id.skXB64, pkXB64: id.pkXB64, }; this._newNodeBundle = id.bundleEnc; fresh = true; } } // An identity this node already knows still needs its group key, which the // node wraps on every connection. if (!gekRaw && this._sessionKeys && !fresh) { const bundleResp = await this._sendAndWait({ type: 'gek_bundle_fetch', v: '0.1', }); if (bundleResp.type === 'gek_bundle_resp' && bundleResp.found) { const skXRaw = Uint8Array.from(atob(this._sessionKeys.skXB64), c => c.charCodeAt(0)); const myPkX = Uint8Array.from(atob(this._sessionKeys.pkXB64), c => c.charCodeAt(0)); try { gekRaw = await window.MeshBayCrypto.unwrapGEK(bundleResp, skXRaw, myPkX); this._gekRaw = gekRaw; } catch (e) { console.warn('[MeshBay] stored GEK bundle did not open; joining instead'); } } } // No stored bundle: ask the node to recognise us and wrap the key itself. // This is the normal path for anyone who joined after the invite redesign — // no bundle is pre-stored for members any more. A code is needed only the // first time this node sees this account. if (!gekRaw && this._sessionKeys && userId) { try { gekRaw = await this.joinGroup(userId, groupId, joinCode); } catch (e) { // The UI turns this into "ask the operator for an invite code". this._joinError = e; } } if (!gekRaw && !this._sessionKeys) { // No identity keys in this browser and none recoverable from the node: // the keypair bundle is created where you register and only reaches a // node after a first successful connection, so a brand-new member opening // a second browser has nothing to sign or unwrap with. Say that, rather // than blaming the GEK — a code prompt here would be useless, since a // code proves who you are and we have no key to bind to. const err = new Error( 'This browser does not hold your keys. Open the group once from the ' + 'browser where you registered — after that this one can recover them.'); err.reason = 'no_keys'; throw err; } if (!gekRaw) { throw this._joinError || new Error('Node requires GEK proof but no GEK available'); } const C = window.MeshBayCrypto; // Node's answer SDP carries ITS fingerprint; our offer carries ours. Throws // if either is missing rather than proceeding with an unbound proof (L4). const binding = C.webrtcBinding( _extractDtlsFingerprint(this._pc.localDescription.sdp), _extractDtlsFingerprint(this._rawAnswerSdp), ); const nonceNode = this._nonceNode; // captured when the challenge arrived const gid = groupId || ''; const proof = await C.handshakeProof( gekRaw, 'client', gid, this._nonceClient, nonceNode, binding); const ack = await this._sendAndWait({ type: 'handshake_response', v: '0.1', proof: C.b64encode(proof), }); if (ack.type !== 'handshake_ack') { throw new Error('GEK proof rejected: ' + (ack.detail || JSON.stringify(ack))); } // Authenticate the NODE before trusting anything it says (C3). Until this // ran, node_pk was decorative: a peer that had hijacked signaling could // accept our proof, ignore it, and serve a forged index, chat history and // is_node_admin flag. const expected = await C.handshakeProof( gekRaw, 'node', gid, this._nonceClient, nonceNode, binding); if (!ack.proof || !C.constantTimeEqual(C.b64decode(ack.proof), expected)) { throw new Error('Node failed to prove GEK possession — refusing connection'); } const transcript = C.handshakeTranscript( 'node', gid, this._nonceClient, nonceNode, binding); if (!ack.node_pk || !ack.sig || !await C.verifyNodeSignature(ack.node_pk, ack.sig, transcript)) { throw new Error('Node signature invalid — refusing connection'); } // Trust On First Use (11.5.8). With C6 closed, a substituted node already // fails the GEK proof — this covers the case where an attacker HAS the GEK // (an ex-member, or a leaked key) and swaps the node underneath. // Strict refusal: a warning users can click through is decorative. // The key announced in the challenge must be the one that just proved // itself. A peer that changed identity mid-handshake is not one to trust // with anything, including a join we may already have signed for it. if (this.nodePk && this.nodePk !== ack.node_pk) { throw new Error('Node identity changed during the handshake — refusing'); } _checkNodePin(nodeId, ack.node_pk); this.nodePk = ack.node_pk; return ack; } // A node that answers a handshake with anything other than a challenge is not // running the mutual protocol. Accepting a bare handshake_ack here would let a // peer skip proving GEK possession entirely (C3/C6). console.warn('[MeshBay] Handshake rejected:', reply.detail, 'code:', reply.code); const rejected = new Error( 'MNP handshake rejected: ' + (reply.detail || `unexpected ${reply.type}`)); rejected.reason = reply.code || ''; throw rejected; } /** * Pair this browser with the node using a one-time code (M3, and the same * substitution as H3). * * The node has no way to know which key belongs to its operator unless someone * tells it locally — asking the hub would let the hub name itself node * administrator. The code comes from `meshbay-node operator pair`, over SSH, and * the hub never sees it. */ async pairOperator(userId, code) { if (!this._connected) throw new Error('Not connected to the node'); if (!userId) throw new Error('Missing user id'); if (!this._sessionKeys || !this._sessionKeys.skEdB64 || !this._sessionKeys.skXB64) { throw new Error('Identity keys unavailable in this browser — sign in again'); } if (!this._nonceNode || !this.nodePk) { throw new Error('Handshake incomplete — reconnect and retry'); } const C = window.MeshBayCrypto; // Both public keys are derived from OUR OWN secret keys, never read back from // the hub: signing a public key the directory handed us would reintroduce the // substitution this whole mechanism exists to close. const pkEdB64 = await _pkEdFromSk(this._sessionKeys.skEdB64); const pkXB64 = await _pkFromSk(this._sessionKeys.skXB64); const ts = Math.floor(Date.now() / 1000); // group_id is empty: operator authority is node-wide, not per group. const transcript = C.joinTranscript( this.nodePk, '', userId, pkEdB64, pkXB64, this._nonceNode, ts); const sig = await window.MeshBayKeys.signBytes(this._sessionKeys.skEdB64, transcript); const resp = await this._sendAndWait({ type: 'join_request', v: '0.1', group_id: '', pk_ed25519: pkEdB64, pk_x25519: pkXB64, code: code || '', ts, sig, }); if (resp.type === 'error') throw new Error(resp.detail || 'Pairing refused'); if (resp.type !== 'join_result' || !resp.ok) { const reason = resp.reason || 'unknown'; const err = new Error(JOIN_REFUSALS[reason] || `Pairing refused: ${reason}`); err.reason = reason; throw err; } this.memberRole = 'operator'; return resp; } async fetchIndex() { const msg = await this._sendAndWait({ type: 'index_sync', v: '0.1' }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } async fetchChunk(fileId, chunkIndex) { const msg = await this._sendAndWait({ type: 'file_req', v: '0.1', file_id: fileId, chunk_index: chunkIndex, }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } /** * TMDB metadata for one file (Videos app, docs/mediacenter.md §5.4). * `path` is root+relpath, exactly what index_sync/index_delta already * gave this browser — never a raw filesystem path constructed here. * `confidence: 0` (no tmdb_id, no fields) means no confident match — * the caller falls back to a thumbnail-only card (§4.1), not an error. */ async fetchMediaMeta(path) { const msg = await this._sendAndWait({ type: 'media_meta_req', v: '0.5', path }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } /** * One season's own overview/air_date/poster (docs/mediacenter.md §5.4's * per-season view) — a show's own tmdb_meta is one static field that does * not necessarily describe every season alike, found live: a 3-season * show whose overview read as season-3-specific for every season. * Keyed like media_meta_req: a season-tab bar can fire a request per tab * before the previous one lands, and matching by arrival order would hand * one season's data to a different season's tab whenever two responses * reordered. */ async fetchSeasonMeta(tmdbId, season) { const msg = await this._sendAndWait({ type: 'season_meta_req', v: '0.6', tmdb_id: tmdbId, season, }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } /** * Raw TMDB search candidates for an operator correcting a wrong automatic * match — unlike fetchMediaMeta, this never collapses to one best guess: * a human picks from several, so several is the point. Read-only, not an * admin op: it looks nothing up in this node's own state and changes * nothing, so it needs no signature (mirrors why media_meta_req isn't * signed either). */ async searchTmdb(mediaType, query) { const msg = await this._sendAndWait({ type: 'tmdb_search_req', v: '0.6', media_type: mediaType, query, }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } /** * Correct a wrong automatic TMDB match. Signed like setVideoRoot/ * setTmdbConfig: it replaces what every member sees for a show/movie, * node-wide (media_cache is shared, not per-viewer) — an unsigned * override would let any member vandalize another show's metadata. * Applies to every file sharing the representative one's display_title, * not just the file the operator happened to be looking at (webrtc_ * server.py's _admin_exec_tmdb_override). */ async overrideTmdbMatch(path, tmdbId, mediaType, signFn) { const msg = await this._sendAndWait({ type: 'tmdb_override', v: '0.6', path, tmdb_id: tmdbId, media_type: mediaType, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { const subject = `path=${path},tmdb_id=${tmdbId},media_type=${mediaType}`; return this._authorizeAdminOp(msg, 'tmdb_override', subject, signFn); } return msg; } /** * Set/clear a custom TMDB API token, and/or set the language TMDB is * queried in (e.g. "fr-FR") — one for the whole node, since both are one * operator's shared credential/cache, not a per-group concern (see * setTmdbEnabled below for the per-group on/off switch). Signed like * setAppsEnabled/setMemberUpload — an unsigned change would let any * member alter outbound third-party network traffic the operator never * agreed to (docs/mediacenter.md §5.5, §8). `token: ''` explicitly clears * a previously-set custom token; omit it (undefined/null), like * `language`, to leave whatever is stored unchanged. */ async setTmdbConfig(token, language, signFn) { const msg = await this._sendAndWait({ type: 'tmdb_config', v: '0.7', token: token === undefined ? null : token, language: language === undefined ? null : language, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { // Must match the node's subject byte-for-byte (webrtc_server.py // _do_tmdb_config) — the token itself is never part of the subject // (it would end up in the audit log in plaintext), only whether one // was supplied. The language is not a secret, so it appears as-is. const subject = `custom_token=${token ? 'yes' : 'no'},language=${language || 'default'}`; return this._authorizeAdminOp(msg, 'tmdb_config', subject, signFn); } return msg; } /** * Whether TMDB lookups run for this group at all — per-group (2026-08-24, * used to be node-wide): a real media-library group and a test/demo group * on the same node need not share the decision to spend TMDB quota and * make outbound requests. Signed like setVideoRoot — it decides whether * this group's members' Videos tab ever makes outbound TMDB traffic. */ async setTmdbEnabled(enabled, signFn) { const msg = await this._sendAndWait({ type: 'tmdb_enabled', v: '0.7', enabled: Boolean(enabled), }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { // Must match the node's subject byte-for-byte (webrtc_server.py // _do_tmdb_enabled): Python's f"{bool}" is "True"/"False", not JS's // lowercase. const subject = enabled ? 'True' : 'False'; return this._authorizeAdminOp(msg, 'tmdb_enabled', subject, signFn); } return msg; } /** * Which folder (possibly a subfolder of a shared root) the Videos app * treats as its entry point for this group. `path: ''` means the whole * group index. Signed like setAppsEnabled — it decides what every * member's Videos tab shows. */ async setVideoRoot(path, signFn) { const clean = (path || '').replace(/^\/+|\/+$/g, ''); const msg = await this._sendAndWait({ type: 'video_root', v: '0.5', path: clean }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'video_root', clean, signFn); } return msg; } async fetchStreamSegment(fileId, segmentIndex, segmentDuration) { const msg = await this._sendAndWait({ type: 'stream_seg', v: '0.1', file_id: fileId, segment_index: segmentIndex, segment_duration: segmentDuration || 4, }); if (msg.type === 'error') throw new Error(msg.detail); return _b64decode(msg.data_b64); } /** * A page of chat history, newest first by default. * * `before` is a message id, not a timestamp: it pages backwards from the * newest, which is the direction a conversation is read. Asking without it * used to mean `since: 0`, which paged *forwards* from the very first message * — so a busy group opened on its oldest page and never showed the recent * exchange. * * Returns { messages, hasMore } — hasMore says whether anything older exists, * so the "load older" control knows when to stop offering. */ async fetchChatHistory({ before = null, limit = 100 } = {}) { const msg = await this._sendAndWait({ type: 'chat_hist', v: '0.2', before: before, limit: limit, }); if (msg.type === 'error') throw new Error(msg.detail); return { messages: msg.messages || [], hasMore: !!msg.has_more }; } /** * Liveness on this already-open channel. Resolves with the round trip in ms, * rejects on timeout — a DataChannel whose peer vanished without closing * still reads as connected, and nothing else here notices until a real * request hangs. */ async ping(timeoutMs = 5000) { const token = Math.random().toString(36).slice(2); const started = performance.now(); const msg = await this._sendAndWait({ type: 'ping', v: '0.2', token }, timeoutMs); if (msg.type === 'error') throw new Error(msg.detail); return Math.round(performance.now() - started); } async sendChat(payload, iteration, threadId, senderName) { const msg = await this._sendAndWait({ type: 'chat_msg', v: '0.1', payload: payload, iteration: iteration || 0, thread_id: threadId || null, sender_name: senderName || null, }); return msg; } /** * Authorize a privileged node operation with the user's Ed25519 identity key. * * The client rebuilds the signed transcript from the challenge fields and refuses * to sign unless the operation and subject match what the user actually asked for. * Previously the node sent 32 opaque random bytes and the client signed them * blind, which let any peer obtain a signature over content of its choosing * (finding H5). */ async _authorizeAdminOp(challenge, expectedOp, expectedSubject, signFn) { if (challenge.op !== expectedOp || challenge.subject !== expectedSubject) { throw new Error( `Refusing to sign: node asked to authorize "${challenge.op}" on ` + `"${challenge.subject}", but the requested action was "${expectedOp}" ` + `on "${expectedSubject}"`); } if (!signFn) throw new Error('Admin challenge received but no signing key available'); const transcript = window.MeshBayCrypto.adminTranscript( challenge.op, challenge.node_pk, challenge.group_id, challenge.subject, challenge.nonce, challenge.ts); const signature = await signFn(transcript); const ack = await this._sendAndWait({ type: 'admin_response', v: '0.1', op_id: challenge.op_id, signature, }); if (ack.type === 'error') throw new Error(ack.detail); return ack; } async deleteFile(fileId, signFn) { const msg = await this._sendAndWait({ type: 'file_delete', v: '0.1', file_id: fileId, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'file_delete', fileId, signFn); } return msg; } /** * Remove an empty directory. Operator only, and the node checks that — this * signs with the identity it pinned for us, exactly like deleting a file. */ async deleteDirectory(dir, signFn) { const msg = await this._sendAndWait({ type: 'dir_delete', v: '0.1', dir, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { // `dir`, not msg.subject: comparing the node's answer against itself is // no check at all, and the point of this one is that we know what we // asked for without being told. return this._authorizeAdminOp(msg, 'dir_delete', dir, signFn); } return msg; } /** * Stop this node serving the group key to someone. Operator only. * * Only the node can do this: its roster decides who it serves. Removing them * on the hub is the other half, and neither implies the other. */ /** * Turn uploading by ordinary members on or off. * * Signed by the operator like any other privileged operation — the node * refuses an unsigned one, which is what stops a member turning it back on. */ async setMemberUpload(allowed, signFn) { const msg = await this._sendAndWait({ type: 'member_upload', v: '0.1', allowed: Boolean(allowed), }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp( msg, 'member_upload', allowed ? 'on' : 'off', signFn); } return msg; } /** * Turn a group "application" (Chat, Files, ...) on or off for everyone. * * Takes the whole set in one signed message rather than one op per app, so * ticking several boxes in Settings costs one signature. `apps` is sorted * and joined the same way on the node before it is shown for signing — * `_authorizeAdminOp` below checks the two match. */ async setAppsEnabled(apps, signFn) { const msg = await this._sendAndWait({ type: 'apps_enabled', v: '0.1', apps, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp( msg, 'apps_enabled', [...apps].sort().join(','), signFn); } return msg; } /** * How often the node's reconciliation backstop runs, and how long it * waits after a file's last write before hashing it (indexer.py * DirectoryIndexer). Whole seconds only: the node builds the signing * subject with Python's `%g` (drops a trailing ".0"), and the simplest * way to always match it byte-for-byte from JS is to never send a * fractional value in the first place. */ async setScanSettings(reconcileIntervalSecs, debounceSecs, signFn) { const reconcile = Math.round(reconcileIntervalSecs); const debounce = Math.round(debounceSecs); const msg = await this._sendAndWait({ type: 'set_scan_settings', v: '0.1', reconcile_interval_secs: reconcile, debounce_secs: debounce, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp( msg, 'set_scan_settings', `${reconcile},${debounce}`, signFn); } return msg; } async revokeMember(userId, signFn) { const msg = await this._sendAndWait({ type: 'member_revoke', v: '0.1', user_id: userId, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'member_revoke', userId, signFn); } return msg; } // ── Node management (D5) ─────────────────────────────────────────────── async fetchNodeStatus() { const msg = await this._sendAndWait({ type: 'node_status', v: '0.1' }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } async addRoot(groupId, path, { name, kind, upload } = {}, signFn) { const msg = await this._sendAndWait({ type: 'root_add', v: '0.1', group_id: groupId, path, name: name || '', kind: kind || 'generic', upload: !!upload, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'root_add', path, signFn); } return msg; } async removeRoot(groupId, rootName, signFn) { const msg = await this._sendAndWait({ type: 'root_remove', v: '0.1', group_id: groupId, root_name: rootName, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'root_remove', rootName, signFn); } return msg; } async unpinMember(userId, signFn) { const msg = await this._sendAndWait({ type: 'member_unpin', v: '0.1', user_id: userId, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'member_unpin', userId, signFn); } return msg; } async rotateGek(groupId, signFn) { const msg = await this._sendAndWait({ type: 'gek_rotate', v: '0.1', group_id: groupId, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'gek_rotate', groupId, signFn); } return msg; } async fetchRoster(groupId) { const msg = await this._sendAndWait({ type: 'roster_read', v: '0.1', group_id: groupId || '', }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } async fetchDenylist() { const msg = await this._sendAndWait({ type: 'denylist_read', v: '0.1' }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } async clearDenylist(subject) { const msg = await this._sendAndWait({ type: 'denylist_clear', v: '0.1', subject: subject || '', }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } async attachGroup(name, sharedDir, uploadDir, signFn) { const msg = await this._sendAndWait({ type: 'group_attach', v: '0.1', name, shared_dir: sharedDir, upload_dir: uploadDir || '', }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'group_attach', name, signFn); } return msg; } async detachGroup(name, signFn) { const msg = await this._sendAndWait({ type: 'group_detach', v: '0.1', name, }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'group_detach', name, signFn); } return msg; } async reloadConfig() { const msg = await this._sendAndWait({ type: 'node_reload', v: '0.1' }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } /** * Ask for a video stream, and say how much we can take. * * `credits` bounds what is in flight. Without it the node pushes the whole * film as fast as ffmpeg produces it and the browser holds all of it while * MediaSource consumes a segment at a time — which is fine for a clip and * fatal for anything worth streaming. */ requestStream(fileId, credits = STREAM_CREDITS, start = 0) { // `start` is a seek: the node retires whatever this session was streaming // and spawns ffmpeg again from there. Omitted or zero is the film's // beginning, which is what an 0.1 node understands. console.log('[stream] sending stream_req start:', start, 'credits:', credits); this._send({ type: 'stream_req', v: '0.1', file_id: fileId, credits, start }); } /** Room for `n` more segments. */ grantStreamCredit(n = 1) { if (!this._connected) return; console.log('[stream] grant credit:', n); this._send({ type: 'stream_more', v: '0.1', n }); } /** * Tell the node what the player sees. * * A hang on a phone is unreadable from here: there is no console to open and * the node's own log shows a stream it is feeding perfectly well. This puts * the two halves in one file. The node only logs it. */ sendStreamDiag(diag) { if (!this._connected) return; try { this._send({ type: 'client_diag', v: '0.1', ...diag }); } catch { /* gone */ } } /** * Nobody is watching any more. * * Closing the viewer used to say nothing to the node, which went on * transcoding and holding one of its two slots until the credit timeout — so * the next video answered "server busy". */ stopStream() { if (!this._connected) return; try { this._send({ type: 'stream_stop', v: '0.1' }); } catch { /* gone */ } } /** * Push a whole file, several chunks in flight at once. * * One chunk per round trip is 48 KB of throughput per RTT no matter how much * bandwidth there is: 4.8 MB/s on a 10 ms path, 480 KB/s on a 100 ms one, and * the sender is idle for almost all of it — which also keeps SCTP's congestion * window shut, so the transport never gets a chance to speed up either. A * window of chunks makes the rate depend on bandwidth rather than distance. * * Order is not at risk: a DataChannel is ordered and reliable by default, and * the node refuses any chunk that is not the one it expects next. * * The node decides where this lands (uploads/) and under what name — it finds a * free one rather than replacing anything. The ack says which, and that is what * this returns. */ async uploadFile(file, { chunkSize, onProgress, signal } = {}) { // The same file twice at once would confuse the node, which keys its own // upload state by name — and would race for the same destination. if (this._uploaders.has(file.name)) { throw new Error(`${file.name} is already being uploaded`); } const size = chunkSize || UPLOAD_CHUNK_SIZE; const total = Math.max(1, Math.ceil(file.size / size)); let acked = 0; let stored = null; let failure = null; const acks = []; this._uploaders.set(file.name, (msg) => { if (msg.type === 'error') { failure = new Error(msg.detail || 'Upload refused'); } else if (msg.stored_as) { stored = msg; } acked += 1; if (onProgress) onProgress(Math.min(file.size, acked * size), file.size); const waiter = acks.shift(); if (waiter) waiter(); }); const nextAck = () => new Promise(r => acks.push(r)); try { for (let i = 0; i < total; i++) { if (signal && signal.aborted) throw _aborted(); // Backpressure: without it the whole file lands in the browser's send // buffer in seconds and the progress bar becomes a work of fiction. while (this._channel && this._channel.bufferedAmount > UPLOAD_BUFFER_HIGH) { if (signal && signal.aborted) throw _aborted(); await new Promise(r => setTimeout(r, 20)); } while (i - acked >= UPLOAD_WINDOW) { await nextAck(); if (failure) throw failure; } if (failure) throw failure; const buf = new Uint8Array( await file.slice(i * size, (i + 1) * size).arrayBuffer()); this._send({ type: 'file_upload', v: '0.1', filename: file.name, chunk_index: i, total_chunks: total, data: buf, }); } while (acked < total) { await nextAck(); if (failure) throw failure; } } finally { this._uploaders.delete(file.name); } return stored || {}; } /** Create a directory under the current one. Any member may. */ async createDirectory(dir, name) { const msg = await this._sendAndWait({ type: 'dir_create', v: '0.1', dir: dir || '', name, }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } /** * Ask the node for a one-time pairing code admitting `userId` to this group. * * This replaces wrapping the group key in the browser. We no longer fetch the * invitee's public key from the hub, so the hub can no longer answer with its own * and be handed the group key (H3). The node wraps the key later, itself, for a * key the invitee proves possession of. * * Returns {code, expires_at} — the code is displayed once and passed to the * invitee out of band. */ async createInvite(userId, groupId, username, signFn) { const msg = await this._sendAndWait({ type: 'invite_create', v: '0.1', user_id: userId, group_id: groupId, username: username || '', }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { return this._authorizeAdminOp(msg, 'invite_create', userId, signFn); } return msg; } /** * Ask the node to recognise us and hand over the group key. * * Sent when we hold no GEK for a group. `code` is needed only the first time * this node sees this account (and not at all in an open-join group). */ async joinGroup(userId, groupId, code) { if (!this._sessionKeys || !this._sessionKeys.skEdB64 || !this._sessionKeys.skXB64) { throw new Error('Identity keys unavailable in this browser — sign in again'); } if (!this._nonceNode || !this.nodePk) { throw new Error('Handshake incomplete — reconnect and retry'); } const C = window.MeshBayCrypto; const pkEdB64 = await _pkEdFromSk(this._sessionKeys.skEdB64); const pkXB64 = await _pkFromSk(this._sessionKeys.skXB64); const ts = Math.floor(Date.now() / 1000); const transcript = C.joinTranscript( this.nodePk, groupId || '', userId, pkEdB64, pkXB64, this._nonceNode, ts); const sig = await window.MeshBayKeys.signBytes(this._sessionKeys.skEdB64, transcript); const resp = await this._sendAndWait({ type: 'join_request', v: '0.1', group_id: groupId || '', pk_ed25519: pkEdB64, pk_x25519: pkXB64, code: code || '', ts, sig, }); if (resp.type === 'error') throw new Error(resp.detail || 'Join refused'); if ((resp.type !== 'join_result' || !resp.ok) || !resp.gek) { const reason = resp.reason || 'unknown'; const err = new Error(JOIN_REFUSALS[reason] || `Join refused: ${reason}`); // The UI reacts to `code_required` by asking for one; everything else is // shown as-is. err.reason = reason; throw err; } // Unwrap with our own secret key — the node wrapped for the public key we // just proved we hold, so nobody else can open this. const skXRaw = Uint8Array.from(atob(this._sessionKeys.skXB64), c => c.charCodeAt(0)); const myPkX = Uint8Array.from(atob(pkXB64), c => c.charCodeAt(0)); const gekRaw = await C.unwrapGEK(resp, skXRaw, myPkX); this._gekRaw = gekRaw; // What the node's roster says this identity is, which is not what the hub // says: `operator` here means this browser's key was paired with the node, // not merely that the account owns it. this.memberRole = resp.role || ''; return gekRaw; } // ── Device linking ───────────────────────────────────────────────────── // // Identity keys are per node, so a browser and a desktop client are two keys // on one account here. A new one is admitted by a key this node already // pinned — never by the hub, which holds no user keys and so cannot // countersign anything. See docs/desktop-client-v1.md §4. /** * Ask to be added, and return the code to show the person. * * They read it off this screen and type it into a device already paired with * this node. The code is hashed together with our own keys, so that other * device cannot be handed a substituted key and sign for it by mistake. */ async requestDeviceAdd(userId) { if (!this._sessionKeys || !this._sessionKeys.skEdB64) { throw new Error('Identity keys unavailable in this browser — sign in again'); } if (!this._nonceNode || !this.nodePk) { throw new Error('Handshake incomplete — reconnect and retry'); } const C = window.MeshBayCrypto; const pkEdB64 = await _pkEdFromSk(this._sessionKeys.skEdB64); const pkXB64 = await _pkFromSk(this._sessionKeys.skXB64); // 40 bits from the platform CSPRNG, in the same alphabet as a pairing code // so it reads and types the same way. const alphabet = '0123456789ABCDEFGHJKMNPQRSTVWXYZ'; const bytes = crypto.getRandomValues(new Uint8Array(8)); const raw = Array.from(bytes, b => alphabet[b % alphabet.length]).join(''); const code = `${raw.slice(0, 4)}-${raw.slice(4)}`; const codeHash = await C.deviceCodeHash( C.normalizeCode(code), pkEdB64, pkXB64); const ts = Math.floor(Date.now() / 1000); const transcript = C.deviceRequestTranscript( this.nodePk, userId, pkEdB64, pkXB64, codeHash, this._nonceNode, ts); const sig = await window.MeshBayKeys.signBytes( this._sessionKeys.skEdB64, transcript); const resp = await this._sendAndWait({ type: 'device_add_request', v: '0.1', pk_ed25519: pkEdB64, pk_x25519: pkXB64, code_hash: codeHash, ts, sig, }); if (resp.type === 'error') throw new Error(resp.detail || 'Refused'); return { code, expiresAt: resp.expires_at }; } /** * Approve a device waiting with this code. * * The node is a mailbox: it is asked for a request matching * sha256(code ‖ keys), and the keys in that hash came from the device that * filed it. A node returning something else produces no match, so there is * nothing to sign and nothing for a person to misread. */ async approveDevice(userId, code) { if (!this._sessionKeys || !this._sessionKeys.skEdB64) { throw new Error('Identity keys unavailable in this browser — sign in again'); } if (!this._nonceNode || !this.nodePk) { throw new Error('Handshake incomplete — reconnect and retry'); } const C = window.MeshBayCrypto; const normalized = C.normalizeCode(code); // The code never leaves this browser. The node lists what is pending, each // with the hash the requesting device computed over the code and its own // keys; we recompute and keep the one that matches. A node offering // fabricated keys would have to produce a hash matching sha256(code ‖ // fabricated) — and it does not know the code. const listed = await this._sendAndWait({ type: 'device_lookup', v: '0.1' }); if (listed.type === 'error') throw new Error(listed.detail || 'Not found'); let match = null; for (const req of listed.requests || []) { const expect = await C.deviceCodeHash( normalized, req.pk_ed25519, req.pk_x25519); if (expect === req.code_hash) { match = req; break; } } if (!match) { throw new Error('No device is waiting with that code'); } return this._countersign(userId, match.code_hash, match.pk_ed25519, match.pk_x25519); } async _countersign(userId, codeHash, pkEdB64, pkXB64) { const C = window.MeshBayCrypto; const ts = Math.floor(Date.now() / 1000); const transcript = C.deviceAddTranscript( this.nodePk, userId, pkEdB64, pkXB64, this._nonceNode, ts); const sig = await window.MeshBayKeys.signBytes( this._sessionKeys.skEdB64, transcript); const resp = await this._sendAndWait({ type: 'device_add', v: '0.1', pk_ed25519: pkEdB64, pk_x25519: pkXB64, code_hash: codeHash, ts, sig, }); if (resp.type === 'error') throw new Error(resp.detail || 'Refused'); return resp; } async listDevices() { const resp = await this._sendAndWait({ type: 'device_list', v: '0.1' }); if (resp.type === 'error') throw new Error(resp.detail || 'Refused'); return { devices: resp.devices || [], pending: resp.pending || 0 }; } /** Retire a device — a lost laptop. Countersigned like an addition. */ async revokeDevice(userId, pkEdB64, pkXB64) { const C = window.MeshBayCrypto; const ts = Math.floor(Date.now() / 1000); const transcript = C.deviceAddTranscript( this.nodePk, userId, pkEdB64, pkXB64, this._nonceNode, ts); const sig = await window.MeshBayKeys.signBytes( this._sessionKeys.skEdB64, transcript); const resp = await this._sendAndWait({ type: 'device_revoke', v: '0.1', pk_ed25519: pkEdB64, ts, sig, }); if (resp.type === 'error') throw new Error(resp.detail || 'Refused'); return resp; } /** * Withdraw our key backup from this node. * * The counterpart of storeKeypairBundle: turning the setting off has to remove * what is already stored, not merely stop adding to it — otherwise the blob * stays on every node the account has ever joined (C4). */ async deleteKeypairBundle() { const msg = await this._sendAndWait({ type: 'keypair_bundle_delete', v: '0.1', }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } async storeKeypairBundle(bundleEnc) { const msg = await this._sendAndWait({ type: 'keypair_bundle_store', v: '0.1', bundle_enc: bundleEnc, }); if (msg.type === 'error') throw new Error(msg.detail); return msg; } get gekRaw() { return this._gekRaw; } close() { if (this._channel) this._channel.close(); if (this._pc) this._pc.close(); this._connected = false; for (const [, p] of this._pending) p.reject(new Error('Transport closed')); this._pending.clear(); } // ── Internal ────────────────────────────────────────────────────────────── _sendAndWait(obj, timeoutMs = 30000) { return new Promise((resolve, reject) => { const id = this._seqId++; const timeout = setTimeout(() => { this._pending.delete(id); console.error('[MeshBay] Response timeout for', obj.type, 'after', timeoutMs, 'ms, channel=', this._channel?.readyState); reject(new Error('Response timeout')); }, timeoutMs); this._pending.set(id, { _reqType: obj.type, // Chunks are the one request that runs several at a time and can be // interleaved with anything else on the channel. Matching them by // arrival order was only ever true by luck; this makes it true. // // A ping is keyed for the same reason and a sharper one: it is sent // *while* other traffic is in flight, so the fallback below would hand // a pong to whatever was waiting — resolving a history request with a // message that has no messages in it, and emptying the conversation. // media_meta_req is the same shape as file_req: video-app.js fires // one per visible poster-grid tile, several at a time — matching by // arrival order handed one tile's TMDB result to a different tile // whenever two responses reordered (reproduced live: which of two // shows got the confident match flipped across reloads). _key: obj.type === 'file_req' ? `chunk:${obj.file_id}:${obj.chunk_index}` : obj.type === 'ping' ? `ping:${obj.token}` : obj.type === 'media_meta_req' ? `media_meta:${obj.path}` // Same reordering hazard as media_meta_req: a season-tab bar or a // search box can have more than one of these in flight at once. : obj.type === 'season_meta_req' ? `season_meta:${obj.tmdb_id}:${obj.season}` : obj.type === 'tmdb_search_req' ? `tmdb_search:${obj.media_type}:${obj.query}` : null, resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); }, reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); }, }); this._send(obj); }); } _send(obj) { if (!this._channel || this._channel.readyState !== 'open') { throw new Error(`DataChannel not open (state: ${this._channel?.readyState})`); } const encoded = msgpack_encode(obj); const header = new Uint8Array(4); new DataView(header.buffer).setUint32(0, encoded.byteLength, false); const frame = new Uint8Array(4 + encoded.byteLength); frame.set(header); frame.set(encoded, 4); this._channel.send(frame); } _onMessage(data) { const incoming = new Uint8Array(data); this._msgCount = (this._msgCount || 0) + 1; if (this._msgCount <= 3) { console.log('[MeshBay] recv', incoming.length, 'bytes, msg #' + this._msgCount); } const combined = new Uint8Array(this._recvBuf.length + incoming.length); combined.set(this._recvBuf); combined.set(incoming, this._recvBuf.length); this._recvBuf = combined; while (this._recvBuf.length >= 4) { const len = new DataView(this._recvBuf.buffer, this._recvBuf.byteOffset).getUint32(0, false); if (this._recvBuf.length < 4 + len) break; const msgBytes = this._recvBuf.slice(4, 4 + len); this._recvBuf = this._recvBuf.slice(4 + len); const msg = msgpack_decode(msgBytes); this._dispatch(msg); } } _dispatch(msg) { // While an upload is in flight the acks are its own, and there are many of // them: they must not be handed to whatever request happens to be oldest in // the pending map. if (msg.type === 'file_upload_ack' && this._uploaders.has(msg.filename)) { this._uploaders.get(msg.filename)(msg); return; } // An upload refusal names the file it is about, so only that upload fails. // It did not use to, and there was no way to tell whose error it was, so // every upload in flight was failed together — send a second file whose // name the node dislikes and both died. The broadcast is kept for a node // that does not name it, where guessing wrong is worse than stopping. if (msg.type === 'error' && this._uploaders.size) { if (msg.filename && this._uploaders.has(msg.filename)) { this._uploaders.get(msg.filename)(msg); return; } if (!msg.filename) { for (const handler of [...this._uploaders.values()]) handler(msg); return; } // Named, but for an upload that is no longer running — not ours to act on. return; } if (msg.type === 'chat_msg' && this._onChat) { this._onChat(msg); return; } if (msg.type === 'stream_init') { console.log('[stream] recv stream_init, start:', msg.start, 'codec:', msg.codec, 'handler:', !!this._onStreamInit); if (this._onStreamInit) this._onStreamInit(msg); return; } if (msg.type === 'stream_data') { if (this._onStreamData) this._onStreamData(msg); return; } if (msg.type === 'stream_end') { console.log('[stream] recv stream_end'); if (this._onStreamEnd) this._onStreamEnd(msg); return; } // The operator changed who may upload. Unsolicited: it arrives at everyone // connected, not only at whoever asked. It still has to reach a pending // caller — the operator's own request resolves on this reply — so it falls // through to the matching below rather than returning here. if (msg.type === 'member_upload_ack' && this._onUploadPolicy) { this._onUploadPolicy(Boolean(msg.allowed)); } // Same shape: the operator changed which apps are shown, and everyone // connected hears about it without reconnecting. if (msg.type === 'apps_enabled_ack' && this._onAppsEnabled) { this._onAppsEnabled(msg.apps || []); } // Node-wide (not per-group) — the operator supplied/cleared a custom // token, or changed the query language. `token_customized` only says // whether one is set, never the token itself. if (msg.type === 'tmdb_config_ack' && this._onTmdbConfig) { this._onTmdbConfig({ tokenCustomized: Boolean(msg.token_customized), language: msg.language || '', }); } // Per-group (2026-08-24, used to be folded into tmdb_config_ack above) — // the operator turned TMDB on/off for this group specifically. if (msg.type === 'tmdb_enabled_ack' && this._onTmdbEnabled) { this._onTmdbEnabled(Boolean(msg.enabled)); } // Same shape: an operator corrected a wrong automatic TMDB match, and // everyone connected needs to know their poster grid/detail modal for // this show is now stale — falls through so the operator's own // admin_response promise resolves on this same message, exactly like // member_upload_ack/apps_enabled_ack above. if (msg.type === 'tmdb_override_ack' && this._onTmdbOverride) { this._onTmdbOverride({ path: msg.path || '', tmdbId: msg.tmdb_id || '', mediaType: msg.media_type || '', }); } // Same shape: the operator changed which folder is the Videos app's // entry point for this group. if (msg.type === 'video_root_ack' && this._onVideoRoot) { this._onVideoRoot(msg.path || ''); } // The operator's node is scanning — never the entries themselves, just // enough to animate a presence dot. Pushed periodically while it runs, // plus once more on the transition back to idle (daemon.py // _progress_pusher). UNLIKE member_upload_ack/apps_enabled_ack above, // this is never a reply to anything this browser asked for — nobody // calls _sendAndWait for it — so it MUST return here. Falling through // to the "oldest pending" guess below hands it to whatever unrelated // request happens to be waiting (a handshake, a chat history fetch), // which then waits forever for its real answer while this one already // "arrived" — and every message after that is one slot off too. Found // live: a group mid-scan corrupted its own handshake and chat history // this way, arriving roughly every 2s for as long as scanning ran. if (msg.type === 'index_progress') { if (this._onIndexProgress) { this._onIndexProgress({ scanning: Boolean(msg.scanning), scanned_bytes: msg.scanned_bytes || 0, total_bytes: msg.total_bytes || 0, }); } return; } // Same reasoning as index_progress: nobody awaits this one either, it // is purely informational (group-settings.js does not currently act on // it), so it must not be left to fall through to the oldest pending // request. if (msg.type === 'set_scan_settings_ack') { return; } if (msg.type === 'index_sync' && msg.entries) { if (this._onIndexSync) this._onIndexSync(msg); const oldest = this._pending.entries().next(); if (!oldest.done && oldest.value[1]._reqType === 'index_sync') { oldest.value[1].resolve(msg); } return; } // Incremental update — additions/deletions/updates, never the whole // index. Only ever arrives after the full index this browser already // has (the node's first push to a newly connected peer is always // index_sync, see daemon.py _broadcast_index_change), so there is // always a base to apply it to. if (msg.type === 'index_delta') { if (this._onIndexDelta) this._onIndexDelta(msg); return; } if (msg.type === 'file_chunk') { const key = `chunk:${msg.file_id}:${msg.chunk_index}`; for (const [, handler] of this._pending) { // A node from before the reply carried a file_id: fall back to the // index, which is still better than the oldest pending request. const match = msg.file_id ? handler._key === key : handler._key && handler._key.endsWith(`:${msg.chunk_index}`); if (match) { handler.resolve(msg); return; } } // Nobody asked for it any more — a cancelled download, most likely. It // must not be handed to whatever request happens to be waiting. console.warn('[MeshBay] file_chunk for nobody', msg.file_id, msg.chunk_index); return; } // "Server busy, retry shortly" and friends arrive as a bare error while a // stream is being set up, with no request waiting for them. They used to // fall through to the oldest pending handler — usually nobody — so the // player sat on "buffering" with the answer already in hand. if (msg.type === 'error' && this._onStreamError) { this._onStreamError(msg); return; } if (msg.type === 'pong') { const key = `ping:${msg.token}`; for (const [, handler] of this._pending) { if (handler._key === key) { handler.resolve(msg); return; } } // A pong for a probe that already timed out. It must not fall through to // the oldest pending request. return; } if (msg.type === 'media_meta_resp') { const key = `media_meta:${msg.path}`; for (const [, handler] of this._pending) { if (handler._key === key) { handler.resolve(msg); return; } } // Nobody asked for this path any more (tile scrolled out and a fresh // request superseded it, most likely) — must not fall through to the // oldest pending request, which would hand a different tile's promise // a TMDB result for a path it never asked about. return; } // Same reasoning as media_meta_resp: keyed, not arrival-order, and // "nobody's waiting any more" must not fall through either. if (msg.type === 'season_meta_resp') { const key = `season_meta:${msg.tmdb_id}:${msg.season}`; for (const [, handler] of this._pending) { if (handler._key === key) { handler.resolve(msg); return; } } return; } if (msg.type === 'tmdb_search_resp') { const key = `tmdb_search:${msg.media_type}:${msg.query}`; for (const [, handler] of this._pending) { if (handler._key === key) { handler.resolve(msg); return; } } return; } // chat_hist_resp answers a `chat_hist` request, but under a different // type string — unlike index_sync, which is asked for and answered under // the same name, so the generic fallback below happens to work for it by // accident. Without this check, whenever a chat_hist_resp arrives while // something else this browser asked for (fetchIndex, even the handshake // itself) is still the oldest pending entry, it gets handed to that // instead: the request chat_hist_resp actually belongs to then hangs // until _sendAndWait's own 30s timeout, and whatever it stole from // resolves with the wrong shape entirely — reproduced live as a // consistent ~30s hang immediately after a successful handshake, for one // specific group and not others connected the same way, which is exactly // what depending on response arrival order rather than on request type // predicts: it fires only when the two responses happen to reorder. if (msg.type === 'chat_hist_resp') { const oldest = this._pending.entries().next(); if (!oldest.done && oldest.value[1]._reqType === 'chat_hist') { oldest.value[1].resolve(msg); } else { console.warn('[MeshBay] chat_hist_resp with no matching chat_hist pending'); } return; } // Everything above is routed by something in the message. What is left is // matched by arrival order, which is only ever a guess — and a wrong guess // here hands one request's answer to another, which then waits for a reply // that already came. Logged so that guess is visible. const oldest = this._pending.entries().next(); if (!oldest.done) { const [, handler] = oldest.value; if (msg.type !== handler._reqType + '_resp' && handler._reqType !== 'index_sync') { console.warn('[MeshBay] unrouted', msg.type, '-> oldest pending', handler._reqType, '(pending:', this._pending.size, ')'); } handler.resolve(msg); } else { console.warn('[MeshBay] unrouted', msg.type, 'with nothing waiting'); } } } // ── Minimal msgpack encode/decode ──────────────────────────────────────────── // Covers the subset used by MNP: maps, strings, integers, binary, arrays, null. function msgpack_encode(obj) { const parts = []; _encodeValue(obj, parts); const total = parts.reduce((s, p) => s + p.length, 0); const result = new Uint8Array(total); let off = 0; for (const p of parts) { result.set(p, off); off += p.length; } return result; } function _encodeValue(val, parts) { if (val === null || val === undefined) { parts.push(new Uint8Array([0xc0])); } else if (typeof val === 'boolean') { parts.push(new Uint8Array([val ? 0xc3 : 0xc2])); } else if (typeof val === 'number') { if (Number.isInteger(val)) { if (val >= 0 && val <= 127) { parts.push(new Uint8Array([val])); } else if (val >= 0 && val <= 0xff) { parts.push(new Uint8Array([0xcc, val])); } else if (val >= 0 && val <= 0xffff) { const b = new Uint8Array(3); b[0] = 0xcd; new DataView(b.buffer).setUint16(1, val, false); parts.push(b); } else if (val >= 0 && val <= 0xffffffff) { const b = new Uint8Array(5); b[0] = 0xce; new DataView(b.buffer).setUint32(1, val, false); parts.push(b); } else if (val >= 0 && val <= Number.MAX_SAFE_INTEGER) { // Same split as the 0xcf decoder case above, in reverse — without // this, a value over 0xffffffff fell to the plain int32 branch // below and silently wrapped to a wrong, unrelated number instead // of failing loudly. const b = new Uint8Array(9); b[0] = 0xcf; const dv = new DataView(b.buffer); dv.setUint32(1, Math.floor(val / 4294967296), false); dv.setUint32(5, val % 4294967296, false); parts.push(b); } else if (val >= -32 && val < 0) { parts.push(new Uint8Array([val & 0xff])); } else if (val >= -128 && val < 0) { const b = new Uint8Array(2); b[0] = 0xd0; b[1] = val & 0xff; parts.push(b); } else { const b = new Uint8Array(5); b[0] = 0xd2; new DataView(b.buffer).setInt32(1, val, false); parts.push(b); } } else { const b = new Uint8Array(9); b[0] = 0xcb; new DataView(b.buffer).setFloat64(1, val, false); parts.push(b); } } else if (typeof val === 'string') { const encoded = new TextEncoder().encode(val); if (encoded.length <= 31) { parts.push(new Uint8Array([0xa0 | encoded.length])); } else if (encoded.length <= 0xff) { parts.push(new Uint8Array([0xd9, encoded.length])); } else if (encoded.length <= 0xffff) { const b = new Uint8Array(3); b[0] = 0xda; new DataView(b.buffer).setUint16(1, encoded.length, false); parts.push(b); } else { const b = new Uint8Array(5); b[0] = 0xdb; new DataView(b.buffer).setUint32(1, encoded.length, false); parts.push(b); } parts.push(encoded); } else if (val instanceof Uint8Array) { if (val.length <= 0xff) { parts.push(new Uint8Array([0xc4, val.length])); } else if (val.length <= 0xffff) { const b = new Uint8Array(3); b[0] = 0xc5; new DataView(b.buffer).setUint16(1, val.length, false); parts.push(b); } else { const b = new Uint8Array(5); b[0] = 0xc6; new DataView(b.buffer).setUint32(1, val.length, false); parts.push(b); } parts.push(val); } else if (Array.isArray(val)) { if (val.length <= 15) { parts.push(new Uint8Array([0x90 | val.length])); } else if (val.length <= 0xffff) { const b = new Uint8Array(3); b[0] = 0xdc; new DataView(b.buffer).setUint16(1, val.length, false); parts.push(b); } else { const b = new Uint8Array(5); b[0] = 0xdd; new DataView(b.buffer).setUint32(1, val.length, false); parts.push(b); } for (const item of val) _encodeValue(item, parts); } else if (typeof val === 'object') { const keys = Object.keys(val); if (keys.length <= 15) { parts.push(new Uint8Array([0x80 | keys.length])); } else if (keys.length <= 0xffff) { const b = new Uint8Array(3); b[0] = 0xde; new DataView(b.buffer).setUint16(1, keys.length, false); parts.push(b); } else { const b = new Uint8Array(5); b[0] = 0xdf; new DataView(b.buffer).setUint32(1, keys.length, false); parts.push(b); } for (const k of keys) { _encodeValue(k, parts); _encodeValue(val[k], parts); } } } function msgpack_decode(buf) { const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength); const [val] = _decodeValue(buf, view, 0); return val; } function _decodeValue(buf, view, offset) { const byte = buf[offset]; if (byte <= 0x7f) return [byte, offset + 1]; if ((byte & 0xe0) === 0xe0) return [byte - 256, offset + 1]; if ((byte & 0xa0) === 0xa0) { const len = byte & 0x1f; return [new TextDecoder().decode(buf.slice(offset + 1, offset + 1 + len)), offset + 1 + len]; } if ((byte & 0xf0) === 0x90) { const len = byte & 0x0f; return _decodeArray(buf, view, offset + 1, len); } if ((byte & 0xf0) === 0x80) { const len = byte & 0x0f; return _decodeMap(buf, view, offset + 1, len); } switch (byte) { case 0xc0: return [null, offset + 1]; case 0xc2: return [false, offset + 1]; case 0xc3: return [true, offset + 1]; case 0xc4: { const len = buf[offset + 1]; return [buf.slice(offset + 2, offset + 2 + len), offset + 2 + len]; } case 0xc5: { const len = view.getUint16(offset + 1, false); return [buf.slice(offset + 3, offset + 3 + len), offset + 3 + len]; } case 0xc6: { const len = view.getUint32(offset + 1, false); return [buf.slice(offset + 5, offset + 5 + len), offset + 5 + len]; } case 0xcc: return [buf[offset + 1], offset + 2]; case 0xcd: return [view.getUint16(offset + 1, false), offset + 3]; case 0xce: return [view.getUint32(offset + 1, false), offset + 5]; // uint64/int64 — never emitted by this file's own encoder (a JS number // above 0xffffffff falls to float64 there), but the node's real msgpack // library sends a plain uint64 for any Python int over ~4.3 billion, and // a raw byte count crosses that easily (found live: IndexProgress. // scanned_bytes/total_bytes in the handshake ack, indexer.py, for a // group whose total library size exceeds ~4 GB). Split into two 32-bit // halves rather than DataView's getBigUint64/getBigInt64 — a BigInt // would silently poison every arithmetic use of these fields elsewhere // (percentage math, comparisons) — and every real byte count fits in a // plain JS number well under Number.MAX_SAFE_INTEGER (2^53). case 0xcf: { const hi = view.getUint32(offset + 1, false); const lo = view.getUint32(offset + 5, false); return [hi * 4294967296 + lo, offset + 9]; } case 0xd3: { const hi = view.getInt32(offset + 1, false); const lo = view.getUint32(offset + 5, false); return [hi * 4294967296 + lo, offset + 9]; } case 0xcb: return [view.getFloat64(offset + 1, false), offset + 9]; case 0xd0: return [view.getInt8(offset + 1), offset + 2]; case 0xd1: return [view.getInt16(offset + 1, false), offset + 3]; case 0xd2: return [view.getInt32(offset + 1, false), offset + 5]; case 0xd9: { const len = buf[offset + 1]; return [new TextDecoder().decode(buf.slice(offset + 2, offset + 2 + len)), offset + 2 + len]; } case 0xda: { const len = view.getUint16(offset + 1, false); return [new TextDecoder().decode(buf.slice(offset + 3, offset + 3 + len)), offset + 3 + len]; } case 0xdb: { const len = view.getUint32(offset + 1, false); return [new TextDecoder().decode(buf.slice(offset + 5, offset + 5 + len)), offset + 5 + len]; } case 0xdc: { const len = view.getUint16(offset + 1, false); return _decodeArray(buf, view, offset + 3, len); } case 0xdd: { const len = view.getUint32(offset + 1, false); return _decodeArray(buf, view, offset + 5, len); } case 0xde: { const len = view.getUint16(offset + 1, false); return _decodeMap(buf, view, offset + 3, len); } case 0xdf: { const len = view.getUint32(offset + 1, false); return _decodeMap(buf, view, offset + 5, len); } default: throw new Error(`Unknown msgpack type: 0x${byte.toString(16)}`); } } function _decodeArray(buf, view, offset, count) { const arr = []; for (let i = 0; i < count; i++) { const [val, newOff] = _decodeValue(buf, view, offset); arr.push(val); offset = newOff; } return [arr, offset]; } function _decodeMap(buf, view, offset, count) { const obj = {}; for (let i = 0; i < count; i++) { const [key, off1] = _decodeValue(buf, view, offset); const [val, off2] = _decodeValue(buf, view, off1); obj[key] = val; offset = off2; } return [obj, offset]; } function _b64decode(b64) { const binary = atob(b64); const bytes = new Uint8Array(binary.length); for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i); return bytes; } function _extractDtlsFingerprint(sdp) { const match = sdp.match(/a=fingerprint:sha-256 ([0-9A-Fa-f:]+)/); if (!match) return new Uint8Array(0); const hex = match[1].replace(/:/g, ''); const bytes = new Uint8Array(hex.length / 2); for (let i = 0; i < hex.length; i += 2) bytes[i / 2] = parseInt(hex.substring(i, i + 2), 16); return bytes; } // ── Node identity pinning (11.5.8) ─────────────────────────────────────────── const NODE_PIN_PREFIX = 'mb_nodepin_'; function _checkNodePin(nodeId, nodePk) { if (!nodeId || !nodePk) return; const key = NODE_PIN_PREFIX + nodeId; let pinned = null; try { pinned = localStorage.getItem(key); } catch { return; } if (pinned === null) { try { localStorage.setItem(key, nodePk); } catch {} return; } if (pinned !== nodePk) { throw new Error( 'This node\'s identity key has changed. That is expected only if its ' + 'operator reinstalled the node — otherwise someone may be impersonating ' + 'it. Verify with the operator out of band, then clear the pin in ' + 'Settings to accept the new key.'); } } /** Forget a pinned node identity — the deliberate escape hatch for a legitimate rotation. */ function clearNodePin(nodeId) { try { if (nodeId) localStorage.removeItem(NODE_PIN_PREFIX + nodeId); else { for (const k of Object.keys(localStorage)) if (k.startsWith(NODE_PIN_PREFIX)) localStorage.removeItem(k); } } catch {} } function pinnedNodeCount() { try { return Object.keys(localStorage).filter(k => k.startsWith(NODE_PIN_PREFIX)).length; } catch { return 0; } } // Export MeshBayTransport.clearNodePin = clearNodePin; MeshBayTransport.pinnedNodeCount = pinnedNodeCount; window.MeshBayTransport = MeshBayTransport;