From dd3927a661273734493f65a593755b95aecf5f09 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sat, 15 Aug 2026 17:23:10 +0200 Subject: feat(groups): remove a member, and keep gigabytes out of the tab MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Removing a member.** The owner can do it from the Members tab, and it is two halves in the order that fails safe: the node stops serving the group key first (an operator-signed request, so a paired browser only), then the hub drops the membership row. The other order would leave someone able to reach a node that still serves them. It is a membership, not an account. The user row is never written: their other groups, their files and their pinned identity survive, because one group's owner must not be able to erase someone from the hub. It is also per group — a node hosting two loses them from one — and it does not take back the key they already unwrapped, which is what rotating the GEK is for. The confirmation and the panel both say so. **Downloads and streaming through the disk, in both browsers.** The audit this started as found two ways to put gigabytes in a tab. Firefox and Safari have no File System Access API, so every download there was collected in memory. A service worker fixes it: the page keeps the writable half of a transferred stream, the worker answers a made-up URL with the readable half and a Content-Disposition header, and the browser writes it to disk as it arrives, with real backpressure. The worker caches nothing and falls through on every request that is not one of these downloads. A zip announces no Content-Length, since the archive is larger than the files in it and a length we miss truncates the file. Video was worse and affected both browsers. The node pushed ffmpeg's whole output as fast as it was produced while the player consumed a segment at a time, so the queue held the film — and appending all of it hit the SourceBuffer's cap, where the handler logged the error and dropped the segment, leaving a hole in the middle of the film with nothing to show for it. Streaming is credit-based now, 24 segments of 256 KB in flight, verified against the live node: three credits, three segments, then silence until more are granted. The player evicts what is more than a minute behind the playhead and retries a refused segment rather than dropping it. Co-Authored-By: Claude Opus 5 --- docs/USERGUIDE.md | 49 ++++++++- .../meshbay-common/src/meshbay_common/adminop.py | 1 + .../meshbay-common/src/meshbay_common/protocol.py | 3 + packages/meshbay-hub/src/meshbay_hub/api/groups.py | 48 +++++++++ packages/meshbay-hub/src/meshbay_hub/static/app.js | 119 ++++++++++++++++++-- .../src/meshbay_hub/static/downloads.js | 81 +++++++++++++- .../meshbay-hub/src/meshbay_hub/static/i18n.js | 7 ++ packages/meshbay-hub/src/meshbay_hub/static/sw.js | 60 +++++++++++ .../src/meshbay_hub/static/transport.js | 39 ++++++- packages/meshbay-hub/tests/test_downloads.py | 56 ++++++++++ .../meshbay-hub/tests/test_group_membership.py | 114 ++++++++++++++++++++ .../src/meshbay_node/transport/webrtc_server.py | 120 +++++++++++++++++++++ packages/meshbay-node/tests/test_roster_pairing.py | 82 ++++++++++++++ 13 files changed, 760 insertions(+), 19 deletions(-) create mode 100644 packages/meshbay-hub/src/meshbay_hub/static/sw.js create mode 100644 packages/meshbay-hub/tests/test_group_membership.py diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md index fea02fd..7242dc2 100644 --- a/docs/USERGUIDE.md +++ b/docs/USERGUIDE.md @@ -548,10 +548,13 @@ one and something else in another before choosing an action. - **Ask every time** opens a Save As dialog per file, which is right for one file and wrong for a selection of twenty. -With no folder chosen, automatic still does not put a dialog in your way: the file -goes to the browser's own download folder. The exception is a download over 512 MB, -which cannot be held in memory to get there — that one asks where to put it, and -choosing a folder in Settings stops it asking again. +With no folder chosen, automatic still does not put a dialog in your way, and it +still does not hold the file in memory: a service worker hands the browser a +stream, which it writes to its own download folder as the bytes arrive. That is +how this works in Firefox and Safari, which have no way to open a file for writing +from a page. If even that is unavailable, a download under 512 MB is collected in +memory and handed over; a larger one asks where to put it, because a tab does not +survive a multi-gigabyte blob. A finished download offers **Open** in the transfers widget when it went into a folder you granted: the file is handed to a new tab and the browser decides what @@ -599,6 +602,30 @@ disk, so a 40 GB folder costs 40 GB of disk and a few megabytes of memory. browser says so, with the size, before starting. Use Chrome or Edge for a large one. +### Removing a member + +The group's owner can remove someone from the Members tab. It does two things, in +the order that fails safe: + +1. **The node stops serving them the group key** — an operator-signed request, so + it works only from a paired browser (§3). This is the half that matters. +2. **The hub drops their membership**, which is what stops them reaching the node + through signaling at all. + +What it does **not** do: + +- It does not delete their account. Their other groups, their files and their + identity are untouched — one group's owner cannot erase someone from the hub. +- It does not make the node forget them. The pinned key stays, so they can be + admitted again without a new pairing code; `meshbay-node member unpin` forgets. +- It does not take back the key they already hold. Anyone who has connected has + unwrapped the current GEK, and no protocol reaches into their browser to remove + it. Rotate it with `meshbay-node gek-init --group ` if that matters — + members still in the group pick the new one up on their next connection. + +Removal is per group: on a node hosting several, someone removed from one keeps +the others. + ### Deleting a directory **⋮ → Delete folder**, for the node operator, from a paired browser (§3). The @@ -707,6 +734,20 @@ Each `stream_data` segment is decrypted with the chunk key for its `segment_inde appended to a `SourceBuffer`. The web client does this in `static/app.js` (`VideoPlayer`); ffmpeg must be installed on the node for transcoding. +**Flow control.** The client says how many segments it can take — `stream_req` +carries a credit count — and the node sends no more than that until `stream_more` +grants more. Without it the node hands ffmpeg's entire output to the channel as +fast as it is produced, and the browser holds a whole film in memory while the +player consumes it a segment at a time. Segments are 256 KB and the web client +keeps 24 outstanding, so roughly 6 MB is in flight whatever the film's length. A +client that sends no credit count gets the old unpaced behaviour. + +**The player drops what has been watched.** A SourceBuffer is not a file: browsers +cap it and refuse the append that goes past, so anything more than a minute behind +the playhead is evicted. A segment refused for want of room is retried rather than +dropped — dropping it leaves a hole in the middle of the film and no error +anywhere. + --- ## 8. Security Model diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index 8b301db..fe4be83 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -36,6 +36,7 @@ ADMIN_TRANSCRIPT_PREFIX = b"meshbay:admin:v1" OP_FILE_DELETE = "file_delete" OP_DIR_DELETE = "dir_delete" OP_INVITE_CREATE = "invite_create" +OP_MEMBER_REVOKE = "member_revoke" # OP_GEK_BUNDLE_STORE is gone. Members no longer hand the node key material at # all: the node holds the GEK and wraps it itself, for a key the recipient proved # they hold (see `join.py` and docs/invite-pairing-v1.md). The operation existed diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index 62d8847..2972d58 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -45,6 +45,7 @@ class MNP: 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 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) @@ -62,6 +63,8 @@ class MNP: 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" INVITE_RESULT = "invite_result" # node → operator: the code, once diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py index 83feeb4..74c6c9e 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py @@ -276,6 +276,54 @@ async def create_group( return {"group_id": group.id, "name": group.name} +@router.delete("/{group_id}/members/{username}") +async def remove_group_member( + group_id: str, + username: str, + request: Request, + current_user: User = Depends(require_user_scope), + db: AsyncSession = Depends(get_db), +): + """ + Remove someone from a group. The group's owner only. + + This is half of removing a member, and the half the hub can do: without a + membership row they cannot reach the node through signaling, and their next + token will not name this group. What it does not do is make the node forget + them — the node's roster decides who it serves, and only a paired operator + can change that (`member_revoke` over MNP, or `meshbay-node member revoke`). + The browser does both; a caller using this endpoint alone should know it did + one. + """ + group = await db.get(Group, group_id) + if not group: + raise HTTPException(status_code=404, detail="Group not found") + if group.admin_id != current_user.id: + raise HTTPException(status_code=403, + detail="Only the group owner can remove members") + + target = (await db.execute( + select(User).where(User.username == username))).scalar_one_or_none() + if not target: + raise HTTPException(status_code=404, detail="User not found") + if target.id == group.admin_id: + raise HTTPException( + status_code=409, + detail="The owner cannot be removed from their own group. Hand the " + "group over or delete it.") + + membership = await db.get(GroupMember, (group_id, target.id)) + if not membership: + raise HTTPException(status_code=404, detail="Not a member of this group") + + await db.delete(membership) + db.add(IPLog(user_id=current_user.id, event="group_leave", + ip_address=client_ip(request), + detail=f"{username} removed from {group.name}")) + await db.commit() + return {"status": "removed", "group_id": group_id, "username": username} + + class GroupUpdateRequest(BaseModel): description: str | None = None diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index c8fb694..18ce83a 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -941,6 +941,10 @@ function canPreview(e) { } const CHUNK_SIZE = 1024 * 1024; +// Seconds of already-watched video kept in the SourceBuffer, and the queue depth +// past which we start making room before being forced to. +const BUFFER_BEHIND_S = 60; +const QUEUE_HIGH_WATER = 12; const PIPELINE_WINDOW = 8; /** @@ -950,7 +954,8 @@ const PIPELINE_WINDOW = 8; * it and hand the browser a blob", or false for "the person dismissed the * dialog", which is not an error and must not start a transfer. */ -async function _openDownloadTarget(filename, size = 0, pickerOpts = {}) { +async function _openDownloadTarget(filename, size = 0, pickerOpts = {}, + swSize = size) { try { const target = await downloads.openTarget(filename); if (target) return target; @@ -958,12 +963,16 @@ async function _openDownloadTarget(filename, size = 0, pickerOpts = {}) { console.warn('[MeshBay] download folder unusable:', err.message); } - // No granted folder. Saving automatically means not putting a dialog in the - // way, so anything that fits in memory goes to the browser's own download - // folder — which is what "automatic" meant to whoever chose the setting. - // Past that a blob would take the tab down with it, and one dialog is the - // lesser evil; Settings is where to stop it happening again. - if (downloads.getMode() === 'auto' && size < downloads.BLOB_LIMIT) return null; + // No granted folder. A service worker can still hand the browser a stream to + // write, which is how this works at all in Firefox: the alternative there is + // to collect gigabytes in a tab. It goes to the browser's own download + // folder, without a dialog, which is what "save automatically" meant. + if (downloads.getMode() === 'auto') { + const streamed = await downloads.openStreamedDownload(filename, swSize); + if (streamed) return streamed; + // Nothing to stream to: small enough for memory, and no dialog. + if (size < downloads.BLOB_LIMIT) return null; + } if (!window.showSaveFilePicker) return null; try { @@ -1331,10 +1340,13 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth, const totalBytes = files.reduce((n, f) => n + (f.entry.size || 0), 0); const suggested = (dir.split('/').pop() || 'files') + '.zip'; + // totalBytes decides how this is delivered, but it is not the archive's + // size — headers and the central directory come on top — so it is not + // announced as a Content-Length that the download would then miss. const target = await _openDownloadTarget(suggested, totalBytes, { types: [{ description: 'ZIP archive', accept: { 'application/zip': ['.zip'] } }], - }); + }, 0); if (target === false) return; if (!target && !confirm(t('group.zip_no_stream', { size: formatSize(totalBytes), name: suggested, @@ -1970,6 +1982,39 @@ function MembersPanel({ groupId, group, token, transportRef, gekRef, } }, [pairCode, transportRef, userId]); + const [removing, setRemoving] = useState(''); + + /** + * Take someone out of this group: both halves, in the order that fails safe. + * + * The node first, because that is the half that stops the group key being + * wrapped for them; if the hub removal then fails, they are a member on paper + * with no key. The other order would leave them able to reach a node that + * still serves them. + */ + const removeMember = useCallback(async (member) => { + const transport = transportRef && transportRef.current; + setError(''); + setRemoving(member.user_id); + try { + if (transport && transport.connected && operatorPaired) { + const sk = transport.sessionKeys && transport.sessionKeys.skEdB64; + const signFn = (sk && window.MeshBayKeys) + ? (transcript) => window.MeshBayKeys.signBytes(sk, transcript) + : null; + await transport.revokeMember(member.user_id, signFn); + } + await hubFetch(`/v1/groups/${groupId}/members/${member.username}`, { + method: 'DELETE', token, + }); + loadMembers(); + } catch (err) { + setError(err.message); + } finally { + setRemoving(''); + } + }, [groupId, token, transportRef, operatorPaired]); + const loadMembers = useCallback(() => { setLoading(true); hubFetch(`/v1/groups/${groupId}/members`, { token }) @@ -2067,6 +2112,7 @@ function MembersPanel({ groupId, group, token, transportRef, gekRef, ${t('admin.col_username')} ${t('members.group_role')} + @@ -2079,10 +2125,24 @@ function MembersPanel({ groupId, group, token, transportRef, gekRef, : html`${t('members.member')}` } + + ${isAdmin && m.user_id !== adminId && html` + + `} + `)} + ${isAdmin && members.length > 1 && html` +

${t('members.remove_hint')}

+ `} ${isNodeAdmin && !operatorPaired && html`

${t('members.pair_title')}

@@ -2379,6 +2439,28 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) { const endedRef = useRef(false); const durationRef = useRef(0); + /** + * Drop what has already been watched. + * + * A SourceBuffer is not a file: browsers cap it at a few hundred megabytes + * and refuse the append that goes past. Keeping a minute behind the playhead + * is enough for a small seek backwards and bounded for a three-hour film. + */ + const evictBehind = useCallback(() => { + const sb = sbRef.current; + const v = videoRef.current; + if (!sb || !v || sb.updating || !sb.buffered.length) return false; + const keepFrom = Math.max(0, v.currentTime - BUFFER_BEHIND_S); + const start = sb.buffered.start(0); + if (keepFrom - start < 10) return false; + try { + sb.remove(start, keepFrom); + return true; + } catch { + return false; + } + }, []); + const flushQueue = useCallback(() => { const sb = sbRef.current; if (!sb || appendingRef.current || sb.updating) return; @@ -2389,14 +2471,25 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) { return; } appendingRef.current = true; - const chunk = queueRef.current.shift(); + const chunk = queueRef.current[0]; try { sb.appendBuffer(chunk); + queueRef.current.shift(); } catch (e) { appendingRef.current = false; + if (e.name === 'QuotaExceededError') { + // The segment stays at the head of the queue and is tried again once + // there is room. Dropping it — which is what this used to do — leaves a + // hole in the middle of the film and no error anywhere. + if (!evictBehind()) { + console.warn('[MSE] buffer full and nothing to evict yet'); + } + return; + } + queueRef.current.shift(); console.error('[MSE] appendBuffer error:', e); } - }, []); + }, [evictBehind]); useEffect(() => { let cancelled = false; @@ -2445,6 +2538,12 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) { sb.mode = 'sequence'; sb.addEventListener('updateend', () => { appendingRef.current = false; + // One segment consumed, so the node may send one more. The credit + // is granted here, after the append, because this is the point at + // which the memory is genuinely free again. + const transport = transportRef.current; + if (transport && transport.connected) transport.grantStreamCredit(1); + if (queueRef.current.length > QUEUE_HIGH_WATER) evictBehind(); flushQueue(); }); setPhase('streaming'); diff --git a/packages/meshbay-hub/src/meshbay_hub/static/downloads.js b/packages/meshbay-hub/src/meshbay_hub/static/downloads.js index a71f289..7f48fed 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/downloads.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/downloads.js @@ -171,8 +171,83 @@ export async function openTarget(filename) { } /** - * Below this, a download with no granted folder is collected in memory and - * handed to the browser, which saves it without asking. Above it that would - * mean holding gigabytes in a tab, so it is worth one Save As dialog instead. + * Below this, a download with no granted folder and no service worker is + * collected in memory and handed to the browser. Above it that would mean + * holding gigabytes in a tab, so it is worth one Save As dialog instead. */ export const BLOB_LIMIT = 512 * 1024 * 1024; + +// ── Streaming to disk without the File System Access API ──────────────────── + +const SW_PATH = '/sw.js'; +let _swReady = null; + +export const STREAMS_VIA_SW = typeof window !== 'undefined' + && 'serviceWorker' in navigator + && typeof TransformStream === 'function' + && window.isSecureContext; + +async function serviceWorker() { + if (!STREAMS_VIA_SW) return null; + if (!_swReady) { + _swReady = navigator.serviceWorker.register(SW_PATH, { scope: '/' }) + .then(() => navigator.serviceWorker.ready) + .then(reg => reg.active || navigator.serviceWorker.controller) + .catch(err => { + console.warn('[MeshBay] service worker unavailable:', err.message); + return null; + }); + } + return _swReady; +} + +/** + * A sink the browser writes to disk, for Firefox and anything else without the + * File System Access API. + * + * The page keeps the writable half of a stream and gives the readable half to + * the service worker, which answers a made-up URL with it. Navigating a hidden + * iframe there turns it into an ordinary download: written as it arrives, with + * the browser's own progress, and nothing held in the tab. Backpressure is + * real — `writer.write()` waits when the browser is behind. + * + * Returns {writable, name} shaped like the File System Access one, or null if + * this browser cannot do it either. + */ +export async function openStreamedDownload(filename, size = 0) { + const worker = await serviceWorker(); + if (!worker) return null; + + const id = `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}`; + const { readable, writable } = new TransformStream(); + + try { + worker.postMessage({ type: 'mbdl', id, filename, size, readable }, [readable]); + } catch (err) { + // Transferable streams are what makes the backpressure work; without them + // this would be a memory buffer wearing a stream's clothes. + console.warn('[MeshBay] streams cannot be transferred here:', err.message); + return null; + } + + const frame = document.createElement('iframe'); + frame.hidden = true; + frame.src = `/_mbdl/${id}`; + document.body.appendChild(frame); + + const writer = writable.getWriter(); + return { + name: filename, + writable: { + write: (bytes) => writer.write(bytes), + close: async () => { + await writer.close(); + setTimeout(() => frame.remove(), 2000); + }, + abort: async (reason) => { + try { await writer.abort(reason); } catch { /* already gone */ } + frame.remove(); + }, + }, + }; +} diff --git a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js index a2d13c5..71f6cce 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js @@ -284,6 +284,13 @@ const en = { + 'pinned for your browser, so pair it below before inviting anyone.', 'members.invite_ask_operator': 'Only the operator of the node hosting this group can ' + 'invite, from a browser paired with it.', + 'members.remove': 'Remove', + 'members.remove_confirm': 'Remove {user} from this group? Their account, their ' + + 'other groups and the files they uploaded are not affected.', + 'members.remove_hint': 'Removing someone takes away their membership and stops ' + + 'the node serving them the group key. It does not delete their account, and ' + + 'it does not take back the key they already hold — rotate it on the node ' + + '(meshbay-node gek-init) if that matters.', 'members.pair_title': 'Pair this browser with your node', 'members.pair_hint': 'Your node only accepts operator actions — invites, file ' + 'deletion — from a browser it has been paired with. Run ' diff --git a/packages/meshbay-hub/src/meshbay_hub/static/sw.js b/packages/meshbay-hub/src/meshbay_hub/static/sw.js new file mode 100644 index 0000000..d0a8805 --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/static/sw.js @@ -0,0 +1,60 @@ +/** + * Service worker: the only way to stream a download to disk in Firefox. + * + * Chrome and Edge have the File System Access API — the page opens a file and + * writes to it. Firefox and Safari do not, and the alternative there was to + * collect the whole download in memory and hand the browser a blob, which is + * not an option for a file measured in gigabytes. + * + * So the page makes up a URL, tells this worker what stream answers it, and + * navigates a hidden iframe there. The worker replies with the stream and a + * Content-Disposition header, and the browser does what it does with any + * download: writes it to disk as it arrives, showing its own progress, with + * nothing buffered in the tab. + * + * It caches nothing and intercepts nothing else. Every request that is not one + * of these downloads falls through untouched. + */ + +const PREFIX = '/_mbdl/'; +const pending = new Map(); + +self.addEventListener('install', () => self.skipWaiting()); +self.addEventListener('activate', (event) => event.waitUntil(self.clients.claim())); + +self.addEventListener('message', (event) => { + const data = event.data || {}; + if (data.type !== 'mbdl' || !data.id || !data.readable) return; + pending.set(data.id, { + readable: data.readable, + filename: data.filename || 'download', + size: Number(data.size) || 0, + }); + // A tab that is closed before it navigates would leave a stream here for the + // life of the worker. + setTimeout(() => pending.delete(data.id), 60000); +}); + +self.addEventListener('fetch', (event) => { + const url = new URL(event.request.url); + if (url.origin !== self.location.origin || !url.pathname.startsWith(PREFIX)) { + return; // not ours — the network handles it + } + + const entry = pending.get(url.pathname.slice(PREFIX.length)); + if (!entry) return; + pending.delete(url.pathname.slice(PREFIX.length)); + + const headers = { + 'Content-Type': 'application/octet-stream', + // filename* so a name with accents or spaces survives the trip. + 'Content-Disposition': + `attachment; filename*=UTF-8''${encodeURIComponent(entry.filename)}`, + 'Cache-Control': 'no-store', + }; + // Only when it is known. A zip is assembled as it goes and announcing a + // length we then miss would truncate the file. + if (entry.size > 0) headers['Content-Length'] = String(entry.size); + + event.respondWith(new Response(entry.readable, { headers })); +}); diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 24c08b9..844a201 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -42,6 +42,10 @@ 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. +const STREAM_CREDITS = 24; + function _aborted() { const err = new Error('Cancelled'); err.name = 'AbortError'; @@ -522,8 +526,39 @@ class MeshBayTransport { return msg; } - requestStream(fileId) { - this._send({ type: 'stream_req', v: '0.1', file_id: fileId }); + /** + * 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. + */ + 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; + } + + /** + * 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) { + this._send({ type: 'stream_req', v: '0.1', file_id: fileId, credits }); + } + + /** Room for `n` more segments. */ + grantStreamCredit(n = 1) { + if (!this._connected) return; + this._send({ type: 'stream_more', v: '0.1', n }); } /** diff --git a/packages/meshbay-hub/tests/test_downloads.py b/packages/meshbay-hub/tests/test_downloads.py index cd0cded..41d61bf 100644 --- a/packages/meshbay-hub/tests/test_downloads.py +++ b/packages/meshbay-hub/tests/test_downloads.py @@ -108,3 +108,59 @@ def test_the_open_action_reads_the_file_back(tmp_path): target = src[src.index("export async function openTarget"):] assert "getFile()" in target and "window.open(" in target assert "revokeObjectURL" in target, "the blob URL must not be leaked" + + +# ── Streaming to disk without the File System Access API ──────────────────── + +SW = STATIC / "sw.js" + + +def test_the_worker_only_answers_its_own_urls(): + """ + It is registered at the root scope, so it sees every request the page makes. + Anything that is not a download of ours has to fall through untouched — a + service worker that answers more than it should is a cache bug waiting to + happen. + """ + src = SW.read_text() + assert "startsWith(PREFIX)" in src + assert "self.location.origin" in src, "cross-origin requests must fall through" + # The API, not the word: the file explains in prose that it caches nothing. + for api in ("caches.open", "caches.match", "cache.put"): + assert api not in src, f"this worker must not cache anything ({api})" + + +def test_the_download_is_announced_as_an_attachment(): + src = SW.read_text() + assert "Content-Disposition" in src and "attachment" in src + assert "filename*=UTF-8''" in src, "a name with accents would be mangled" + assert "Content-Length" in src + + +def test_a_length_is_only_promised_when_it_is_known(tmp_path): + """ + An archive is assembled as it goes and is larger than the files in it. + Announcing the sum of their sizes would truncate the download at that mark. + """ + src = SW.read_text() + assert "if (entry.size > 0)" in src + + app = (STATIC / "app.js").read_text() + zip_call = app[app.index("const target = await _openDownloadTarget(suggested"):] + zip_call = zip_call[:zip_call.index(");") + 2] + assert zip_call.rstrip().endswith(", 0);"), ( + "the zip download announces a Content-Length it will not match") + + +def test_backpressure_is_real(tmp_path): + """ + The point of the service worker path is not holding the file. A stream that + is transferred gives `writer.write()` something to wait on; posting chunks + to a port would queue them in memory and look identical from here. + """ + src = DOWNLOADS.read_text() + fn = src[src.index("export async function openStreamedDownload"):] + assert "new TransformStream()" in fn + assert "[readable]" in fn, "the readable half must be transferred, not copied" + assert "writer.write(bytes)" in fn + assert "return null" in fn, "a browser that cannot transfer streams must say so" diff --git a/packages/meshbay-hub/tests/test_group_membership.py b/packages/meshbay-hub/tests/test_group_membership.py new file mode 100644 index 0000000..7eeaa2f --- /dev/null +++ b/packages/meshbay-hub/tests/test_group_membership.py @@ -0,0 +1,114 @@ +""" +Removing someone from a group. + +The distinction these pin is the one that matters: removing a member from a +group is not deleting their account. It takes away one membership row, and +leaves the person, their other groups and everything they have uploaded exactly +where they were. +""" + +import base64 +import hashlib + +import pytest +from sqlalchemy import select + +from meshbay_hub.db.models import Group, GroupMember, User + + +def _auth_key(password: str, username: str) -> str: + salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest() + return base64.b64encode( + hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode() + + +async def _user(client, username, password="a-long-enough-passphrase"): + await client.post("/v1/users/register", json={ + "username": username, "email": f"{username}@example.com", + "auth_key": _auth_key(password, username)}) + r = await client.post("/v1/users/login", json={ + "username": username, "auth_key": _auth_key(password, username)}) + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + +async def _group_with_member(client, owner, member_name, name="crew"): + g = await client.post("/v1/groups", json={"name": name}, headers=owner) + gid = g.json()["group_id"] + await client.post(f"/v1/groups/{gid}/members/{member_name}", json={}, + headers=owner) + return gid + + +@pytest.mark.asyncio +async def test_the_owner_removes_a_member(client, db_session): + owner = await _user(client, "chief") + await _user(client, "hanger_on") + gid = await _group_with_member(client, owner, "hanger_on") + + r = await client.delete(f"/v1/groups/{gid}/members/hanger_on", headers=owner) + assert r.status_code == 200, r.text + + rows = (await db_session.execute( + select(GroupMember).where(GroupMember.group_id == gid))).scalars().all() + assert [m.user_id for m in rows] != [], "the owner lost their own membership" + names = {(await db_session.get(User, m.user_id)).username for m in rows} + assert names == {"chief"} + + +@pytest.mark.asyncio +async def test_removing_a_member_is_not_deleting_an_account(client, db_session): + """ + The account survives untouched, with its other groups. Anything else would + make one group's owner able to erase someone from the whole hub. + """ + owner = await _user(client, "boss") + member = await _user(client, "member_x") + elsewhere = await _user(client, "other_owner") + + gid = await _group_with_member(client, owner, "member_x") + other = await _group_with_member(client, elsewhere, "member_x", name="elsewhere") + + await client.delete(f"/v1/groups/{gid}/members/member_x", headers=owner) + + user = (await db_session.execute( + select(User).where(User.username == "member_x"))).scalar_one() + assert user.status == "active", "the account was touched" + + me = await client.get("/v1/users/me", headers=member) + assert me.status_code == 200, "they can no longer sign in" + + still = await db_session.get(GroupMember, (other, user.id)) + assert still is not None, "removing them from one group emptied another" + + +@pytest.mark.asyncio +async def test_a_member_cannot_remove_anyone(client): + owner = await _user(client, "owner_y") + member = await _user(client, "member_y") + await _user(client, "victim_y") + gid = await _group_with_member(client, owner, "member_y") + await client.post(f"/v1/groups/{gid}/members/victim_y", json={}, headers=owner) + + r = await client.delete(f"/v1/groups/{gid}/members/victim_y", headers=member) + assert r.status_code == 403 + + +@pytest.mark.asyncio +async def test_the_owner_cannot_be_removed_from_their_own_group(client): + """Otherwise the group is left with nobody who can invite or remove.""" + owner = await _user(client, "owner_z") + gid = await _group_with_member(client, owner, "owner_z") + + r = await client.delete(f"/v1/groups/{gid}/members/owner_z", headers=owner) + assert r.status_code == 409 + + +@pytest.mark.asyncio +async def test_removing_someone_who_is_not_a_member_says_so(client): + owner = await _user(client, "owner_w") + await _user(client, "stranger") + g = await client.post("/v1/groups", json={"name": "closed"}, headers=owner) + gid = g.json()["group_id"] + + r = await client.delete(f"/v1/groups/{gid}/members/stranger", headers=owner) + assert r.status_code == 404 diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 81db0e9..eab1cac 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -59,6 +59,7 @@ from meshbay_common.adminop import ( OP_DIR_DELETE, OP_FILE_DELETE, OP_INVITE_CREATE, + OP_MEMBER_REVOKE, admin_transcript, ) from meshbay_common.crypto import pk_to_b64, wrap_gek_aes @@ -171,6 +172,10 @@ def _extract_dtls_fingerprint(sdp: str) -> bytes: STREAM_SEGMENT_SIZE = 256 * 1024 +# What a client may ask for in one go, and how long the node waits for it to ask +# again before deciding nobody is watching any more. +STREAM_MAX_CREDIT = 256 +STREAM_CREDIT_TIMEOUT = 120 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} @@ -289,6 +294,9 @@ class WebRTCPeerSession: # Set from the roster: the key this node pinned for this account. Never # from the JWT — the hub picks what goes in there. self._pinned_pk: str = "" + # Flow control for video: how many segments the client says it can take. + self._stream_credit = 0 + self._stream_credit_evt = asyncio.Event() self._gek_challenge: bytes | None = None # Same value as the GEK challenge, but kept for the life of the connection: # a join_request is signed over it, and it must stay verifiable after the @@ -366,12 +374,16 @@ class WebRTCPeerSession: self._do_admin_response(msg) elif mtype == MNP.INVITE_CREATE: self._do_invite_create(msg) + elif mtype == MNP.MEMBER_REVOKE: + self._do_member_revoke(msg) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: asyncio.ensure_future(self._do_keypair_bundle_store(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: asyncio.ensure_future(self._do_keypair_bundle_delete()) elif mtype == MNP.STREAM_REQUEST: asyncio.ensure_future(self._stream_video(msg)) + elif mtype == MNP.STREAM_MORE: + self._grant_stream_credit(msg) else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -999,6 +1011,69 @@ class WebRTCPeerSession: self._audit("dir_delete", rel) self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel}) + def _do_member_revoke(self, msg: dict) -> None: + """ + Stop serving the group key to someone, at the operator's request. + + The same authority as an invite, and the same reason: the roster decides + who this node serves, so only a key the node pinned as an operator may + change it. Membership on the hub is not consulted — the hub can remove + someone from a group, and that stops them reaching the node at all, but + it cannot make the node forget them. + """ + user_id = str(msg.get("user_id", "")).strip() + if not user_id: + self._send({"type": "error", "detail": "Missing user_id"}) + return + if user_id == self._user_id: + # Removing yourself from your own node is not a member operation; + # it would leave the group with nobody able to invite. + self._send({"type": "error", "detail": "Cannot revoke yourself"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id) + + async def _admin_exec_member_revoke( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + user_id = pending["subject"] + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"member_revoke:{user_id[:8]}") + return + + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) + return + + group_id = self._group_id or "" + if not await roster.set_status(group_id, user_id, "revoked"): + self._send({"type": "error", "detail": "Not a member of this group"}) + return + + # Anyone connected right now keeps the key they already unwrapped; what + # they lose is the next one. Rotating it is the operator's call, and the + # ack says so rather than implying this undid anything already read. + peer = self._peer_registry().get(user_id) + if peer is not None: + try: + await peer.close() + except Exception: + pass + + log.info("Member revoked by %s: user=%s group=%s", + self._user_id[:8], user_id[:8], group_id[:8] or "-") + self._audit("member_revoke", user_id) + self._send({ + "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION, + "user_id": user_id, + "reminder": "they still hold the current group key — rotate it with " + "meshbay-node gek-init", + }) + async def _do_keypair_bundle_delete(self) -> None: """ Withdraw our own key backup from this node. @@ -1542,6 +1617,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_DIR_DELETE: asyncio.ensure_future( self._admin_exec_dir_delete(pending, transcript, sig_bytes)) + elif pending["op"] == OP_MEMBER_REVOKE: + asyncio.ensure_future( + self._admin_exec_member_revoke(pending, transcript, sig_bytes)) elif pending["op"] == OP_INVITE_CREATE: asyncio.ensure_future( self._admin_exec_invite_create(pending, transcript, sig_bytes)) @@ -1634,6 +1712,37 @@ class WebRTCPeerSession: "file_id": file_id, }) + def _grant_stream_credit(self, msg: dict) -> None: + """The client has room for more segments.""" + try: + n = int(msg.get("n", 1)) + except (TypeError, ValueError): + n = 1 + self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT)) + self._stream_credit_evt.set() + + async def _await_stream_credit(self) -> bool: + """ + Block until the client has room. False if it stopped asking. + + Without this the node hands ffmpeg's entire output to the channel as + fast as it is produced, and the browser holds a four gigabyte film in a + JavaScript array while MediaSource consumes it a segment at a time. + """ + while self._stream_credit <= 0: + self._stream_credit_evt.clear() + try: + await asyncio.wait_for(self._stream_credit_evt.wait(), + timeout=STREAM_CREDIT_TIMEOUT) + except asyncio.TimeoutError: + log.info("Stream stalled: no credit from peer=%s", + (self._user_id or "?")[:8]) + return False + if self._channel is None or self._channel.readyState != "open": + return False + self._stream_credit -= 1 + return True + async def _stream_video(self, msg: dict) -> None: """Stream a video file as fMP4 segments via MSE-compatible output.""" # One ffmpeg per request with no cap lets any member exhaust the node's @@ -1692,9 +1801,20 @@ class WebRTCPeerSession: "duration": duration, }) + # A client that says nothing gets the old behaviour, which is why this + # defaults to unlimited rather than to zero: a stream that waits for + # credit from a peer that will never send any is a stream that hangs. + try: + self._stream_credit = int(msg.get("credits", 0) or 0) + except (TypeError, ValueError): + self._stream_credit = 0 + paced = self._stream_credit > 0 + index = 0 try: while True: + if paced and not await self._await_stream_credit(): + break data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py index 88e426d..435cc76 100644 --- a/packages/meshbay-node/tests/test_roster_pairing.py +++ b/packages/meshbay-node/tests/test_roster_pairing.py @@ -914,3 +914,85 @@ async def test_someone_elses_signature_does_not_remove_it(tmp_path, roster): assert _last(session).get("detail") == "Signature verification failed" assert (tmp_path / "shared" / "theirs").is_dir(), ( "a member's signature removed a directory — only the operator may") + + +# ── Removing a member ──────────────────────────────────────────────────────── + +async def test_revoking_needs_an_operator_signature(tmp_path, roster): + from meshbay_common.adminop import admin_transcript + + sk_op, pk_op, pk_x_op = _keypair() + await roster.pin_identity("grenet", "grenet", pk_op, pk_x_op, "code") + await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") + + sk_m, pk_m, pk_x_m = _keypair() + await roster.pin_identity("victim", "victim", pk_m, pk_x_m, "code") + await roster.set_member("g1", "victim", ROLE_MEMBER, "active", "grenet") + + session = _session(tmp_path, roster, group_id="g1", gek=generate_gek()) + session._admin_ops = {} + session._ctx["has_admin_authority"] = True + session._ctx["peers"] = {} + + transcript = admin_transcript( + op="member_revoke", node_pk_b64=session._node_pk_b64(), group_id="g1", + subject="victim", nonce=b"\x33" * 32, ts=int(time.time())) + + # A member's own signature is not enough. + await session._admin_exec_member_revoke( + {"op": "member_revoke", "subject": "victim"}, transcript, + sk_m.sign(transcript)) + assert _last(session).get("detail") == "Signature verification failed" + assert (await roster.get_member("g1", "victim"))["status"] == "active" + + # The operator's is. + await session._admin_exec_member_revoke( + {"op": "member_revoke", "subject": "victim"}, transcript, + sk_op.sign(transcript)) + assert _last(session)["type"] == "member_revoke_ack" + assert (await roster.get_member("g1", "victim"))["status"] == "revoked" + + +async def test_revoking_is_confined_to_the_group_it_was_asked_for(tmp_path, roster): + """ + A node hosting two groups must not lose someone from both. Their pinned + identity survives as well — forgetting a key is `member unpin`, and saying + "remove them" should not silently do it. + """ + from meshbay_common.adminop import admin_transcript + + sk_op, pk_op, pk_x_op = _keypair() + await roster.pin_identity("grenet", "grenet", pk_op, pk_x_op, "code") + await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") + + _, pk_m, pk_x_m = _keypair() + await roster.pin_identity("both", "both", pk_m, pk_x_m, "code") + await roster.set_member("g1", "both", ROLE_MEMBER, "active", "grenet") + await roster.set_member("g2", "both", ROLE_MEMBER, "active", "grenet") + + session = _session(tmp_path, roster, group_id="g1", gek=generate_gek()) + session._admin_ops = {} + session._ctx["has_admin_authority"] = True + session._ctx["peers"] = {} + + transcript = admin_transcript( + op="member_revoke", node_pk_b64=session._node_pk_b64(), group_id="g1", + subject="both", nonce=b"\x44" * 32, ts=int(time.time())) + await session._admin_exec_member_revoke( + {"op": "member_revoke", "subject": "both"}, transcript, + sk_op.sign(transcript)) + + assert (await roster.get_member("g1", "both"))["status"] == "revoked" + assert (await roster.get_member("g2", "both"))["status"] == "active" + assert await roster.get_identity("both") is not None, ( + "the pinned identity was dropped; that is `member unpin`, not this") + + +async def test_an_operator_cannot_revoke_themselves(tmp_path, roster): + """It would leave the group with nobody able to invite or remove.""" + session = _session(tmp_path, roster, group_id="g1", gek=generate_gek()) + session._admin_ops = {} + session._ctx["has_admin_authority"] = True + + session._do_member_revoke({"user_id": session._user_id}) + assert _last(session).get("detail") == "Cannot revoke yourself" -- cgit v1.2.3