diff options
Diffstat (limited to 'packages')
12 files changed, 715 insertions, 15 deletions
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, <tr> <th>${t('admin.col_username')}</th> <th>${t('members.group_role')}</th> + <th></th> </tr> </thead> <tbody> @@ -2079,10 +2125,24 @@ function MembersPanel({ groupId, group, token, transportRef, gekRef, : html`<span class="badge">${t('members.member')}</span>` } </td> + <td class="admin-actions"> + ${isAdmin && m.user_id !== adminId && html` + <button class="admin-btn danger" disabled=${removing === m.user_id} + onClick=${() => { + if (!confirm(t('members.remove_confirm', { user: m.username }))) return; + removeMember(m); + }}> + ${removing === m.user_id ? '...' : t('members.remove')} + </button> + `} + </td> </tr> `)} </tbody> </table> + ${isAdmin && members.length > 1 && html` + <p class="settings-hint">${t('members.remove_hint')}</p> + `} ${isNodeAdmin && !operatorPaired && html` <form class="invite-form" onSubmit=${doPair}> <h4>${t('members.pair_title')}</h4> @@ -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" |