diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub/static/transport.js')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/transport.js | 253 |
1 files changed, 185 insertions, 68 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 927956d..1f6dd8f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -251,7 +251,7 @@ window.addEventListener('hashchange', () => { // The `v: '0.1'` on every other message in this file is the historical value // and is read by nothing; it is left alone deliberately. The range is // negotiated once, at the start, not restated per message. -const MNP_V = '1.1'; +const MNP_V = '2.0'; const MNP_V_MIN = '1.0'; // Codes a NODE sends us, in its own vocabulary (meshbay_common/handshake.py's @@ -290,6 +290,11 @@ class MeshBayTransport { this._pc = null; this._channel = null; this._pending = new Map(); + // Set the first time this connection sees a reply that names the request + // it answers (see _dispatch). A node either stamps every reply or none, + // so one is proof for the connection — and once there is proof, the + // arrival-order fallback at the bottom of _dispatch is never right again. + this._correlates = false; this._seqId = 0; this._recvBuf = new Uint8Array(0); this._connected = false; @@ -299,10 +304,17 @@ class MeshBayTransport { 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. + // upload_id → the uploader waiting on it. Keyed rather than FIFO because + // several uploads may be in flight at once and their acks interleave. + // + // It was keyed by filename until MNP 2.0, which is no longer possible: the + // name is sealed under the group key, and echoing it in clear so the two + // sides could match on it would give back precisely what the seal is for. + // `upload_id` is drawn per upload here and is opaque to the node. this._uploaders = new Map(); + // Names, not ids: the "already being uploaded" guard is about the file the + // caller passed, and two `uploadFile` calls for one file draw two ids. + this._inFlightUploads = new Set(); // Set once close() runs — stops the automatic reconnect from firing on a // connection the caller tore down on purpose (leaving the group, page // unload), which would otherwise race back in right as everything else @@ -379,6 +391,19 @@ class MeshBayTransport { if (!m) return false; return (Number(m[1]) > 1) || (Number(m[1]) === 1 && Number(m[2]) >= 1); } + /** + * Whether the node opens a sealed upload (MNP 2.0). + * + * A 1.x node reads `filename` and `data` off the message itself, finds + * neither — they are inside the seal — and answers "Missing filename or + * data", an error about the wrong thing that names no upload_id and so fails + * every upload in flight. Asked before sending rather than discovered after, + * for the same reason `supportsAppOps` is. + */ + get supportsSealedUpload() { + const m = /^(\d+)\.(\d+)$/.exec(this._nodeVersion || ''); + return !!m && Number(m[1]) >= 2; + } set onAppsEnabled(fn) { this._onAppsEnabled = fn; } set onAppDirectories(fn) { this._onAppDirectories = fn; } set onChatDirectory(fn) { this._onChatDirectory = fn; } @@ -1456,18 +1481,6 @@ class MeshBayTransport { 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. * @@ -1583,7 +1596,7 @@ class MeshBayTransport { this._sessionKeys.skEdB64, C.chatSigningTranscript(gid, epoch, device, nonce, ct))); - return this._sendAndWait({ + const msg = await this._sendAndWait({ type: 'chat_msg', v: '2.0', format: 1, @@ -1596,6 +1609,15 @@ class MeshBayTransport { // Deliberately absent: the display name is inside the envelope now. sender_name: null, }); + // Every other request in this file refuses an `error` reply; this one + // returned it as though the node had accepted the message. It never + // mattered while a refusal reached the wrong caller anyway — now that a + // reply finds the request that made it, a message the node rejected would + // otherwise appear in the conversation as sent. It rejects more of them + // than it used to: a stale epoch, an envelope the node dislikes, or a + // device claim that is not this connection's all come back as `error`. + if (msg.type === 'error') throw new Error(msg.detail || 'chat send refused'); + return msg; } /** @@ -2080,10 +2102,21 @@ class MeshBayTransport { */ async uploadFile(file, { chunkSize, onProgress, signal, root, dir } = {}) { // 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)) { + // upload state by name — and would race for the same destination. The guard + // is by name for that reason, even though the map below is keyed by id. + if (this._inFlightUploads.has(file.name)) { throw new Error(`${file.name} is already being uploaded`); } + if (!this._gekRaw) throw new Error('This group has no key on this device'); + if (!this.supportsSealedUpload) { + throw new Error( + 'This node is running an older MeshBay and cannot accept an upload ' + + 'from this page. Its operator has to update it.'); + } + const C = window.MeshBayCrypto; + const groupId = (this._connectArgs && this._connectArgs.groupId) || ''; + this._inFlightUploads.add(file.name); + const uploadId = _hex(crypto.getRandomValues(new Uint8Array(16))); const size = chunkSize || UPLOAD_CHUNK_SIZE; const total = Math.max(1, Math.ceil(file.size / size)); let acked = 0; @@ -2091,16 +2124,32 @@ class MeshBayTransport { 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; - } + const wake = () => { acked += 1; if (onProgress) onProgress(Math.min(file.size, acked * size), file.size); const waiter = acks.shift(); if (waiter) waiter(); + }; + this._uploaders.set(uploadId, (msg) => { + if (msg.type === 'error') { + failure = new Error(msg.detail || 'Upload refused'); + wake(); + return; + } + // The ack is sealed too — `stored_as` and the folder it landed in name + // the operator's content. Opening it is what makes the result usable, so + // a failure here fails the upload rather than being swallowed: a chat + // attachment that cannot learn its stored name would point at nothing. + C.openGroup(this._gekRaw, 'upload', 'file_upload_ack', groupId, msg) + .then((plain) => { + const payload = msgpack_decode(plain); + if (payload.stored_as) stored = payload; + }) + .catch((e) => { + failure = new Error( + `The node's upload reply did not open under the group key (${e.message})`); + }) + .finally(wake); }); const nextAck = () => new Promise(r => acks.push(r)); @@ -2122,15 +2171,20 @@ class MeshBayTransport { const buf = new Uint8Array( await file.slice(i * size, (i + 1) * size).arrayBuffer()); + // The name, the destination and the bytes go inside the seal together. + // Mirrors `file_upload_wire` in meshbay_common/protocol.py; only the + // fields the node routes on stay outside it. + const sealed = await C.sealGroup( + this._gekRaw, 'upload', 'file_upload', groupId, + msgpack_encode({ filename: file.name, data: buf, + dir: dir || '', root: root || '' })); this._send({ type: 'file_upload', v: '0.1', - filename: file.name, + upload_id: uploadId, chunk_index: i, total_chunks: total, - data: buf, - ...(root ? { root } : {}), - ...(dir ? { dir } : {}), + ...sealed, }); } while (acked < total) { @@ -2138,7 +2192,8 @@ class MeshBayTransport { if (failure) throw failure; } } finally { - this._uploaders.delete(file.name); + this._uploaders.delete(uploadId); + this._inFlightUploads.delete(file.name); } return stored || {}; } @@ -2458,12 +2513,15 @@ class MeshBayTransport { if (msg.type === 'index_sync') { if (this._onIndexSync) this._onIndexSync(opened); - for (const [, handler] of this._pending) { - if (handler._reqType === 'index_sync') { - handler.resolve(opened); - break; - } - } + // The node's first push to a newly connected peer is an index_sync + // nobody asked for, so there is not always a request to resolve. When + // there is, `req_id` says which one — the type match below is what a + // node too old to stamp one leaves us, and it is why two fetches in + // flight at once used to resolve the wrong one. + const handler = opened.req_id !== undefined && opened.req_id !== null + ? this._pending.get(opened.req_id) + : [...this._pending.values()].find(h => h._reqType === 'index_sync'); + if (handler) handler.resolve(opened); return; } if (this._onIndexDelta) this._onIndexDelta(opened); @@ -2550,7 +2608,13 @@ class MeshBayTransport { resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); }, reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); }, }); - this._send(obj); + // The id goes on the wire (MNP 1.1+): a node that understands it stamps + // the reply with it, and _dispatch matches on that alone. It used to be + // local to this map, which is why every reply had to be recognised by + // some field of its own — and why the ones that carry no such field + // reached their caller by luck. An older node ignores the extra key and + // is routed by the per-type fallbacks below, exactly as before. + this._send({ ...obj, req_id: id }); }); } @@ -2590,6 +2654,45 @@ class MeshBayTransport { } _dispatch(msg) { + // A reply that names the request it answers. Nothing below this needs to + // recognise it, and nothing below this may see it: every remaining branch + // exists to identify a reply by some field of its own, which is the job + // this makes unnecessary. + // + // What is left underneath is genuinely unsolicited — a broadcast to every + // connected client, a push, a challenge — or a reply from a node too old + // to stamp one, which is what the per-type keys are for now. + if (msg.req_id !== undefined && msg.req_id !== null) { + this._correlates = true; + // The one exception, and the only one: an index message is sealed under + // the GEK and cannot be handed to its caller until it is opened, which + // is not something this synchronous function can do. Resolving it here + // would give `fetchIndex` the envelope — nonce and ciphertext, no + // entries — and skip `_onIndexSync` entirely. `_queueIndexMessage` + // opens it and then resolves, by this same id. + const sealed = msg.type === 'index_sync' || msg.type === 'index_delta'; + if (!sealed) { + const handler = this._pending.get(msg.req_id); + if (handler) { + handler.resolve(msg); + // The acks whose *broadcast* half their own requester also needs: + // every other client learns the change from the broadcast, and the + // one that asked for it is the only one that would not, because its + // own request swallowed its copy. Same call the keyed `_ack` branch + // below makes, for the same reason. + if (BROADCAST_ACK_TYPES.has(msg.type)) _replayBroadcast(this, msg); + return; + } + // Answers a request that is no longer waiting: it gave up at its own + // timeout, or a reconnect rejected everything in flight. It belongs to + // nobody, and the whole point of this change is that it is not offered + // to somebody else instead. + console.warn('[MeshBay] late reply to req', msg.req_id, '(', msg.type, + ') — nothing waiting'); + return; + } + } + // Two-step admin-op flow (_authorizeAdminOp, ADMIN_OP_TYPES) — resolve // by (op) key before anything below gets a chance to steal it via the // generic "oldest pending" fallback further down. Returns as soon as a @@ -2647,21 +2750,21 @@ class MeshBayTransport { // 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); + if (msg.type === 'file_upload_ack' && this._uploaders.has(msg.upload_id)) { + this._uploaders.get(msg.upload_id)(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. + // An upload refusal names the upload 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 + // refusal that names none, 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); + if (msg.upload_id && this._uploaders.has(msg.upload_id)) { + this._uploaders.get(msg.upload_id)(msg); return; } - if (!msg.filename) { + if (!msg.upload_id) { for (const handler of [...this._uploaders.values()]) handler(msg); return; } @@ -2955,10 +3058,10 @@ class MeshBayTransport { return; } - // device_hello_ack ends in `_ack` but is not an admin op, so the branch - // above looks it up under `admin:device_hello`, finds nothing, and drops it - // through to the arrival-order guess. Routed by request type instead: a - // request type deserves a key, and a reply deserves something to key it by. + // device_hello_ack ends in `_ack` but is not an admin op, so the admin + // branch looks it up under `admin:device_hello` and finds nothing. A 2.0 + // node stamps `req_id` and this is never reached; it is the per-type key + // for a node that does not, alongside chat_hist_resp above. if (msg.type === 'device_hello_ack') { for (const [, handler] of this._pending) { if (handler._reqType === 'device_hello') { handler.resolve(msg); return; } @@ -2968,14 +3071,13 @@ class MeshBayTransport { } // Same shape as chat_hist_resp above, and found the same way — by driving - // the panel rather than by reading this file. `chat_keys_resp` answers a - // `chat_keys_req` under a different type string, so without this it fell - // to the arrival-order guess at the end and was handed to whatever was - // oldest in `_pending`. `chat_send_probe.py` caught it on its first run: - // the Videos tab's unanswered `media_meta_req` swallowed the chat keys, - // and the send then waited out its own 30s timeout with the composer - // disabled — which is a frozen Chat tab, the exact defect that harness - // exists for, reappearing one feature later. + // the panel rather than by reading this file. Before `req_id` existed, + // `chat_keys_resp` fell to the arrival-order guess and was handed to + // whatever was oldest in `_pending`; `chat_send_probe.py` caught it on its + // first run, with the Videos tab's unanswered `media_meta_req` swallowing + // the chat keys and the send then waiting out its own 30s timeout with the + // composer disabled. `req_id` is what closes that class now, and this is + // the per-type key for a node that does not stamp one. if (msg.type === 'chat_keys_resp') { for (const [, handler] of this._pending) { if (handler._reqType === 'chat_keys_req') { handler.resolve(msg); return; } @@ -3020,10 +3122,28 @@ class MeshBayTransport { 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. + // Everything above is routed by something in the message. What is left + // used to be matched by arrival order — a guess, and a wrong guess hands + // one request's answer to another, which then waits out its own 30s + // timeout for a reply that already came and went. That is how the Chat + // composer, disabled while a send is in flight, could stay disabled for + // thirty seconds on a message the node had already stored. + // + // A node that stamps its replies (`req_id`, handled at the top) has taken + // every one of its answers out of this path, so anything arriving here is + // unsolicited and the guess can only ever be wrong. Dropping it loses + // nothing and stops the theft. + if (this._correlates) { + console.warn('[MeshBay] unsolicited', msg.type, '— dropped (pending:', + this._pending.size, ')'); + return; + } + + // Only a node too old to stamp anything reaches here, where arrival order + // is still the only thing there is. Kept deliberately, and no wider than + // it was: the alternative for such a node is that half the protocol + // (device_list_result, join_result, the handshake's own replies) reaches + // nobody at all. const oldest = this._pending.entries().next(); if (!oldest.done) { const [, handler] = oldest.value; @@ -3276,11 +3396,8 @@ function _decodeMap(buf, view, offset, count) { 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 _hex(bytes) { + return [...bytes].map(b => b.toString(16).padStart(2, '0')).join(''); } function _extractDtlsFingerprint(sdp) { |