diff options
16 files changed, 2293 insertions, 58 deletions
diff --git a/packages/meshbay-client/src/main.js b/packages/meshbay-client/src/main.js index 0ed06d9..e7e7b15 100644 --- a/packages/meshbay-client/src/main.js +++ b/packages/meshbay-client/src/main.js @@ -559,6 +559,15 @@ function registerBridge() { return true; }); + ipcMain.handle('root:choose', async () => { + const result = await dialog.showOpenDialog(mainWindow, { + properties: ['openDirectory', 'createDirectory'], + }); + if (result.canceled || !result.filePaths.length) return null; + const chosen = result.filePaths[0]; + return { path: chosen, name: path.basename(chosen) }; + }); + ipcMain.handle('save:begin', async (_e, suggestedName, opts) => { const wanted = path.basename(String(suggestedName || 'download')); const chosen = chosenDownloadDir(); diff --git a/packages/meshbay-client/src/preload.js b/packages/meshbay-client/src/preload.js index 3804ad3..e3af68f 100644 --- a/packages/meshbay-client/src/preload.js +++ b/packages/meshbay-client/src/preload.js @@ -74,6 +74,12 @@ contextBridge.exposeInMainWorld('meshbay', { forget: () => ipcRenderer.invoke('folder:forget'), }, + // Pick a directory to share as a group root. Returns { path, name } — the + // path is forwarded over MNP to the node, which is the local machine for D5. + rootPicker: { + choose: () => ipcRenderer.invoke('root:choose'), + }, + // A sink that writes to disk as chunks arrive, never a buffer handed over at // the end. `auto` uses the remembered folder without a dialog, which is what // "save automatically" means; without one, or when the person asked to be diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py index a793471..73f07d9 100644 --- a/packages/meshbay-common/src/meshbay_common/adminop.py +++ b/packages/meshbay-common/src/meshbay_common/adminop.py @@ -49,6 +49,10 @@ OP_MEMBER_UNPIN = "member_unpin" # setting decides who may write to the operator's disk, so a node that took it # from an unsigned message would let any member re-enable it for everyone. OP_MEMBER_UPLOAD = "member_upload" +OP_ROOT_ADD = "root_add" +OP_ROOT_REMOVE = "root_remove" +OP_GROUP_ATTACH = "group_attach" +OP_GROUP_DETACH = "group_detach" # 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 1c6dd3e..06acb8b 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -97,6 +97,24 @@ class MNP: GEK_ROTATE = "gek_rotate" # operator → node: new group key GEK_ROTATE_ACK = "gek_rotate_ack" INVITE_RESULT = "invite_result" # node → operator: the code, once + NODE_STATUS = "node_status" # operator → node: list all groups + roots + NODE_STATUS_ACK = "node_status_ack" # node → operator: full status + ROOT_ADD = "root_add" # operator → node: add a directory to a group + ROOT_ADD_ACK = "root_add_ack" # node → operator: confirmed + ROOT_REMOVE = "root_remove" # operator → node: remove a root from a group + ROOT_REMOVE_ACK = "root_remove_ack" # node → operator: confirmed + ROSTER_READ = "roster_read" # operator → node: list pinned identities + members + ROSTER_READ_ACK = "roster_read_ack" + DENYLIST_READ = "denylist_read" # operator → node: show denylist entries + DENYLIST_READ_ACK = "denylist_read_ack" + DENYLIST_CLEAR = "denylist_clear" # operator → node: remove denylist entry(ies) + DENYLIST_CLEAR_ACK = "denylist_clear_ack" + GROUP_ATTACH = "group_attach" # operator → node: host a new group + GROUP_ATTACH_ACK = "group_attach_ack" + GROUP_DETACH = "group_detach" # operator → node: stop hosting a group + GROUP_DETACH_ACK = "group_detach_ack" + NODE_RELOAD = "node_reload" # operator → node: re-read node.toml + NODE_RELOAD_ACK = "node_reload_ack" # ── Index entry ─────────────────────────────────────────────────────────────── diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index 816c0ab..3fdba07 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -403,6 +403,9 @@ const ICON_PATHS = { 'bell-off': ['M18 9a6 6 0 0 0-12 0c0 6-2.5 7.5-2.5 7.5h17S18 15 18 9', 'M10.3 20a2 2 0 0 0 3.4 0', 'M4 4l16 16'], + server: ['M4 6.5a2 2 0 0 1 2-2h12a2 2 0 0 1 2 2v2a2 2 0 0 1-2 2H6a2 2 0 0 1-2-2v-2z', + 'M4 15.5a2 2 0 0 1 2-2h12a2 2 0 0 1 2 2v2a2 2 0 0 1-2 2H6a2 2 0 0 1-2-2v-2z', + 'M8 7.5h.01', 'M8 16.5h.01'], }; // The M of the wordmark is a picture; the rest is text. Resolved from this @@ -618,7 +621,7 @@ function Nav({ user, theme, onThemeChange, onLogout, onMenuToggle, unreadCount, // ── Sidebar ────────────────────────────────────────────────────────────────── -function Sidebar({ groups, presence, route, menuOpen, role }) { +function Sidebar({ groups, presence, route, menuOpen, role, hasNodeKey }) { const isStaff = role === 'moderator' || role === 'admin'; return html` <aside class="sidebar ${menuOpen ? 'open' : ''}"> @@ -638,6 +641,13 @@ function Sidebar({ groups, presence, route, menuOpen, role }) { <a class="sidebar-item ${route === '/create-group' ? 'active' : ''}" href="#/create-group"><${Icon} name="plus" /> ${t('sidebar.create_group')}</a> </div> + ${platform.capabilities.nodeAdmin && hasNodeKey && html` + <div class="sidebar-section"> + <div class="sidebar-heading">${t('sidebar.node')}</div> + <a class="sidebar-item ${route === '/node' ? 'active' : ''}" + href="#/node"><${Icon} name="server" /> ${t('node.title')}</a> + </div> + `} <div class="sidebar-section"> <div class="sidebar-heading">${t('sidebar.my_groups')}</div> ${groups.length === 0 @@ -1479,7 +1489,10 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, if (err.reason === 'not_a_member' && !refreshedRef.current && onRefreshAuth) { refreshedRef.current = true; try { - if (await onRefreshAuth()) return; // new token → effect re-runs + if (await onRefreshAuth()) { + setRetryKey(k => k + 1); + return; + } } catch { /* fall through to the message below */ } } @@ -1494,6 +1507,10 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, if (err.reason === 'unknown_device') setNeedsDevice(true); setError(err.message); setStatus('error'); + if (transportRef.current) { + try { transportRef.current.close(); } catch { /* already gone */ } + transportRef.current = null; + } // A refusal means the node answered, so it is up; only a failure to // reach it at all is evidence of absence. if (onPresence) { @@ -1966,7 +1983,10 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, </button> `} </div> - ${error && html`<div class="error-msg" style="margin-bottom:12px">${error}</div>`} + ${error && html`<div class="error-msg" style="margin-bottom:12px">${error}${' '} + <button class="admin-btn" style="margin-left:8px;font-size:0.9em" + onClick=${() => setRetryKey(k => k + 1)}>${t('group.retry')}</button> + </div>`} ${needsDevice && html` <div class="invite-form" style="margin-bottom:12px"> <h4>${t('device.add_title')}</h4> @@ -2016,9 +2036,12 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, <${Icon} name="gear" cls="tab-icon" /></button> </div> - ${tab === 'files' && status !== 'connected' && html` + ${tab === 'files' && (status === 'discovering' || status === 'connecting' || status === 'fetching') && html` <p class="page-message"><span class="spinner"></span>${' '}${t('status.connecting_short')}</p> `} + ${tab === 'files' && status === 'offline' && html` + <p class="page-message">${t('group.offline_title')} ${t('group.offline_hint')}</p> + `} ${tab === 'files' && status === 'connected' && html` <div class="file-toolbar"> @@ -2153,9 +2176,12 @@ function GroupPage({ groupId, group, token, username, userId, userPrefs, else setPreviewEntry(entry); }} /> `} - ${tab === 'chat' && status !== 'connected' && html` + ${tab === 'chat' && (status === 'discovering' || status === 'connecting' || status === 'fetching') && html` <p class="page-message"><span class="spinner"></span>${' '}${t('status.connecting_short')}</p> `} + ${tab === 'chat' && status === 'offline' && html` + <p class="page-message">${t('group.offline_title')} ${t('group.offline_hint')}</p> + `} ${tab === 'settings' && html` <${GroupSettingsPanel} groupId=${groupId} group=${group} token=${token} @@ -4935,6 +4961,529 @@ function BlocklistForm({ onAdd }) { `; } +// ── Node management (D5) ──────────────────────────────────────────────────── + +function NodePage({ token, username, userId, groups }) { + const [status, setStatus] = useState('idle'); + const [error, setError] = useState(''); + const [nodeGroups, setNodeGroups] = useState([]); + const [busy, setBusy] = useState(false); + const [actionMsg, setActionMsg] = useState(''); + const [roster, setRoster] = useState(null); + const [rosterGroup, setRosterGroup] = useState(''); + const [denylist, setDenylist] = useState(null); + const [showDenylist, setShowDenylist] = useState(false); + const [attachGroup_, setAttachGroup_] = useState(''); + const [attachDir, setAttachDir] = useState(''); + const [attachUpload, setAttachUpload] = useState(''); + const transportRef = useRef(null); + + const connectAndFetch = useCallback(async () => { + if (!groups.length) { + setStatus('no_groups'); + return; + } + setStatus('connecting'); + setError(''); + try { + if (!_bundleKey) _bundleKey = await _loadBundleKey(); + const live = (await ensureFreshToken()) || token; + + const nodeCandidates = (await Promise.all( + groups.map(g => + hubFetch(`/v1/groups/${g.id}/nodes`, { token: live }) + .then(d => (d.nodes || []).map(n => ({ groupId: g.id, nodeId: n.node_id }))) + .catch(() => [])) + )).flat(); + + if (!nodeCandidates.length) { + setError(t('node.offline')); + setStatus('error'); + return; + } + + // Group candidates by node, preserving all groupIds. The same node + // is reachable through any group it serves; a group that was just + // attached but not yet loaded will fail, so we try every groupId + // before giving up on a node. + const byNode = new Map(); + for (const c of nodeCandidates) { + if (!byNode.has(c.nodeId)) byNode.set(c.nodeId, []); + const ids = byNode.get(c.nodeId); + if (!ids.includes(c.groupId)) ids.push(c.groupId); + } + + console.log('[NodePage] candidates:', byNode.size, 'node(s)', + [...byNode.entries()].map(([n, gs]) => n.slice(0, 8) + '×' + gs.length)); + let lastErr = null; + for (const [nodeId, groupIds] of byNode) { + for (const groupId of groupIds) { + const transport = new window.MeshBayTransport(HUB, live); + try { + console.log('[NodePage] connecting via group', groupId.slice(0, 8), + 'node', nodeId.slice(0, 8)); + await transport.connect(nodeId, live, groupId, + null, null, _bundleKey, username, userId); + console.log('[NodePage] connected, fetching status'); + const result = await transport.fetchNodeStatus(); + console.log('[NodePage] got status:', result.groups?.length, 'groups'); + transportRef.current = transport; + setNodeGroups(result.groups || []); + setStatus('connected'); + return; + } catch (err) { + console.warn('[NodePage] candidate failed:', nodeId.slice(0, 8), + 'group', groupId.slice(0, 8), err.message); + try { transport.close(); } catch {} + lastErr = err; + } + } + } + setError(lastErr ? lastErr.message : t('node.not_operator')); + setStatus('error'); + } catch (err) { + setError(err.message); + setStatus('error'); + } + }, [token, username, userId, groups]); + + useEffect(() => { + connectAndFetch(); + return () => { + const tr = transportRef.current; + if (tr && tr.connected) try { tr.close(); } catch {} + }; + }, [connectAndFetch]); + + const refresh = useCallback(async () => { + const transport = transportRef.current; + if (!transport || !transport.connected) return; + try { + const result = await transport.fetchNodeStatus(); + setNodeGroups(result.groups || []); + } catch {} + }, []); + + const signFn = useCallback(async (transcript) => { + const transport = transportRef.current; + if (!transport || !transport.sessionKeys || !transport.sessionKeys.skEdB64) + throw new Error('No signing key'); + return window.MeshBayKeys.signBytes(transport.sessionKeys.skEdB64, transcript); + }, []); + + const addRoot = useCallback(async (groupId) => { + const chosen = await platform.rootPicker.choose(); + if (!chosen) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.addRoot(groupId, chosen.path, {}, signFn); + await refresh(); + setActionMsg(t('node.root_added')); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [signFn, refresh]); + + const removeRoot = useCallback(async (groupId, rootName) => { + if (!confirm(t('node.root_remove_confirm', { name: rootName }))) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.removeRoot(groupId, rootName, signFn); + await refresh(); + setActionMsg(t('node.root_removed')); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [signFn, refresh]); + + const loadRoster = useCallback(async (groupId) => { + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + try { + const result = await transport.fetchRoster(groupId); + setRoster(result); + setRosterGroup(groupId); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, []); + + const unpinMember = useCallback(async (userId) => { + if (!confirm(t('node.unpin_confirm', { name: userId }))) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.unpinMember(userId, signFn); + setActionMsg(t('node.unpin_done')); + if (rosterGroup) await loadRoster(rosterGroup); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [signFn, rosterGroup, loadRoster]); + + const rotateGek = useCallback(async (groupId) => { + if (!confirm(t('node.gek_rotate_confirm'))) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.rotateGek(groupId, signFn); + setActionMsg(t('node.gek_rotated')); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [signFn]); + + const loadDenylist = useCallback(async () => { + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + try { + const result = await transport.fetchDenylist(); + setDenylist(result); + setShowDenylist(true); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, []); + + const clearDenylist = useCallback(async (subject) => { + const label = subject || t('node.denylist_clear_all'); + if (!confirm(t('node.denylist_clear_confirm', { subject: label }))) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.clearDenylist(subject); + setActionMsg(t('node.denylist_cleared')); + await loadDenylist(); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [loadDenylist]); + + const pickAttachDir = useCallback(async (setter) => { + const chosen = await platform.rootPicker.choose(); + if (chosen) setter(chosen.path); + }, []); + + const detachGroup = useCallback(async (name) => { + if (!confirm(t('node.detach_confirm', { name }))) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.detachGroup(name, signFn); + setActionMsg(t('node.detached')); + await refresh(); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [signFn, refresh]); + + const attachGroup = useCallback(async () => { + if (!attachGroup_ || !attachDir.trim()) return; + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + const ack = await transport.attachGroup(attachGroup_, attachDir.trim(), + attachUpload.trim(), signFn); + setActionMsg(t('node.attached', { name: ack.name || attachGroup_ })); + setAttachGroup_(''); + setAttachDir(''); + setAttachUpload(''); + await refresh(); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [attachGroup_, attachDir, attachUpload, signFn]); + + const reloadConfig = useCallback(async () => { + const transport = transportRef.current; + if (!transport || !transport.connected) return; + setBusy(true); + setActionMsg(''); + try { + await transport.reloadConfig(); + setActionMsg(t('node.reloaded')); + await refresh(); + } catch (err) { + setActionMsg(err.message); + } finally { + setBusy(false); + } + }, [refresh]); + + if (status === 'idle' || status === 'connecting') { + return html`<div class="page-content"> + <p class="page-message">${t('status.connecting')}</p> + </div>`; + } + if (status === 'no_groups') { + return html`<div class="page-content"> + <p class="page-message">${t('node.no_groups')}</p> + </div>`; + } + if (status === 'error') { + return html`<div class="page-content"> + <h2>${t('node.title')}</h2> + <p class="error-msg">${error}</p> + <button class="btn btn-primary" onClick=${connectAndFetch}> + ${t('node.retry')}</button> + </div>`; + } + + return html` + <div class="page-content node-page"> + <div class="node-header"> + <h2>${t('node.title')}</h2> + <button class="btn btn-small btn-secondary" disabled=${busy} + onClick=${reloadConfig}> + ${t('node.reload')}</button> + </div> + ${actionMsg && html`<div class="node-message">${actionMsg}</div>`} + ${(() => { + const hubIds = new Set((groups || []).map(g => g.id)); + return nodeGroups.map(g => { + const stale = !hubIds.has(g.id); + return html` + <div class="node-group ${stale ? 'node-group-stale' : ''}" key=${g.id}> + <div class="node-group-header"> + <h3>${g.name}${stale ? html` <span class="node-warn">${t('node.stale')}</span>` : ''}</h3> + <div class="node-group-stats"> + <span>${t('node.files', { n: g.file_count })}</span> + <span>${t('node.peers', { n: g.peers })}</span> + ${!g.has_gek && html` + <span class="node-warn">${t('node.no_gek')}</span>`} + </div> + </div> + <div class="node-roots"> + <div class="node-roots-header"> + <span class="settings-heading">${t('node.roots')}</span> + </div> + ${(g.roots || []).map(r => html` + <div class="node-root ${!r.available ? 'node-root-unavailable' : ''}" + key=${r.name}> + <div class="node-root-info"> + <span class="node-root-name"> + <${Icon} name="folder" /> + ${r.name} + </span> + ${r.upload && html` + <span class="node-root-badge">${t('node.upload_root')}</span>`} + ${!r.available && html` + <span class="node-root-badge node-root-badge-warn"> + ${t('node.unavailable')}</span>`} + <span class="node-root-kind">${r.kind}</span> + </div> + ${(g.roots || []).length > 1 && html` + <button class="btn btn-small btn-danger" + disabled=${busy} onClick=${() => removeRoot(g.id, r.name)}> + ${t('node.remove_root')}</button>`} + </div> + `)} + <button class="btn btn-small btn-secondary" + disabled=${busy} onClick=${() => addRoot(g.id)}> + <${Icon} name="folder-plus" /> ${t('node.add_root')} + </button> + </div> + + <div class="node-section"> + <span class="settings-heading">${t('node.gek')}</span> + ${g.has_gek ? html` + <button class="btn btn-small btn-secondary" disabled=${busy} + onClick=${() => rotateGek(g.id)}> + ${t('node.gek_rotate')}</button> + ` : html` + <p class="node-hint">${t('node.gek_init_hint')}</p> + `} + </div> + + <div class="node-section"> + <div class="node-section-header"> + <span class="settings-heading">${t('node.roster')}</span> + <button class="btn btn-small btn-secondary" disabled=${busy} + onClick=${() => loadRoster(g.id)}> + ${t('node.roster_load')}</button> + </div> + ${roster && rosterGroup === g.id && html` + <div class="node-roster"> + ${(roster.members || []).length === 0 ? html` + <p class="node-hint">${t('node.roster_empty')}</p> + ` : html` + <table class="node-table"> + <thead><tr> + <th>${t('node.roster_user')}</th> + <th>${t('node.roster_role')}</th> + <th>${t('node.roster_status')}</th> + <th>${t('node.roster_via')}</th> + <th></th> + </tr></thead> + <tbody> + ${(roster.members || []).map(m => html` + <tr key=${m.user_id}> + <td>${m.username || m.user_id.slice(0, 12)}</td> + <td>${m.role}</td> + <td>${m.status}</td> + <td>${m.pinned_via || ''}</td> + <td>${m.pk_ed25519 && m.user_id !== userId && html` + <button class="btn btn-small btn-danger" + disabled=${busy} + onClick=${() => unpinMember(m.user_id)}> + ${t('node.unpin')}</button> + `}</td> + </tr> + `)} + </tbody> + </table> + `} + ${(roster.invites || []).length > 0 && html` + <p class="node-hint">${t('node.roster_invites', + { n: roster.invites.length })}</p> + `} + </div> + `} + </div> + + <div class="node-section"> + <button class="btn btn-small btn-danger" disabled=${busy} + onClick=${() => detachGroup(g.name)}> + ${t('node.detach_group')}</button> + </div> + </div> + `; + }); + })()} + + <div class="node-group"> + <div class="node-section-header"> + <span class="settings-heading">${t('node.denylist')}</span> + <button class="btn btn-small btn-secondary" disabled=${busy} + onClick=${loadDenylist}> + ${t('node.denylist_load')}</button> + </div> + ${showDenylist && denylist && html` + <div class="node-denylist"> + ${denylist.count === 0 ? html` + <p class="node-hint">${t('node.denylist_empty')}</p> + ` : html` + <p class="node-hint">${t('node.denylist_count', + { n: denylist.count })}</p> + ${(denylist.users || []).map(u => html` + <div class="node-deny-entry" key=${'u:' + u}> + <span>user: ${u}</span> + <button class="btn btn-small btn-danger" disabled=${busy} + onClick=${() => clearDenylist(u)}> + ${t('node.denylist_remove')}</button> + </div> + `)} + ${(denylist.groups || []).map(g => html` + <div class="node-deny-entry" key=${'g:' + g}> + <span>group: ${g}</span> + <button class="btn btn-small btn-danger" disabled=${busy} + onClick=${() => clearDenylist(g)}> + ${t('node.denylist_remove')}</button> + </div> + `)} + ${(denylist.jtis || []).map(j => html` + <div class="node-deny-entry" key=${'j:' + j}> + <span>token: ${j.slice(0, 16)}</span> + <button class="btn btn-small btn-danger" disabled=${busy} + onClick=${() => clearDenylist(j)}> + ${t('node.denylist_remove')}</button> + </div> + `)} + <button class="btn btn-small btn-danger" disabled=${busy} + onClick=${() => clearDenylist('')}> + ${t('node.denylist_clear_all')}</button> + `} + </div> + `} + </div> + + ${(() => { + const hostedIds = new Set(nodeGroups.map(g => g.id)); + const unhosted = (groups || []).filter(g => !hostedIds.has(g.id)); + if (!unhosted.length) return null; + return html` + <div class="node-group"> + <span class="settings-heading">${t('node.attach_group')}</span> + <div class="node-attach-form"> + <label>${t('node.attach_pick')}</label> + <select value=${attachGroup_} + onChange=${e => setAttachGroup_(e.target.value)}> + <option value="">---</option> + ${unhosted.map(g => html` + <option key=${g.id} value=${g.name}>${g.name}</option> + `)} + </select> + <label>${t('node.attach_dir')}</label> + <div class="node-attach-dir-row"> + <input type="text" placeholder=${t('node.attach_dir')} + value=${attachDir} readOnly + onInput=${e => setAttachDir(e.target.value)} /> + ${platform.rootPicker && html` + <button class="btn btn-small btn-secondary" + onClick=${() => pickAttachDir(setAttachDir)}> + <${Icon} name="folder" /></button> + `} + </div> + <label>${t('node.attach_upload_label')}</label> + <div class="node-attach-dir-row"> + <input type="text" placeholder=${t('node.attach_upload_dir')} + value=${attachUpload} readOnly + onInput=${e => setAttachUpload(e.target.value)} /> + ${platform.rootPicker && html` + <button class="btn btn-small btn-secondary" + onClick=${() => pickAttachDir(setAttachUpload)}> + <${Icon} name="folder" /></button> + `} + </div> + <button class="btn btn-primary" + disabled=${busy || !attachGroup_ || !attachDir.trim()} + onClick=${attachGroup}> + ${t('node.attach_group')}</button> + </div> + <p class="node-hint">${t('node.restart_needed')}</p> + </div> + `; + })()} + </div> + `; +} + // ── App ────────────────────────────────────────────────────────────────────── function App() { @@ -4954,6 +5503,7 @@ function App() { const [unreadCount, setUnreadCount] = useState(0); const [notifDisabled, setNotifDisabled] = useState(false); const [userPrefs, setUserPrefs] = useState({}); + const [hasNodeKey, setHasNodeKey] = useState(false); const resolved = resolveTheme(theme); @@ -5002,7 +5552,7 @@ function App() { }, [user, notifDisabled]); useEffect(() => { - if (!user) { setGroups([]); setNotifications([]); setUnreadCount(0); return; } + if (!user) { setGroups([]); setNotifications([]); setUnreadCount(0); setHasNodeKey(false); return; } hubFetch('/v1/groups/mine', { token: user.token }) .then(data => setGroups(data.groups || [])) .catch(() => setGroups([])); @@ -5012,6 +5562,11 @@ function App() { if (prefs.notifications_disabled === 'true') setNotifDisabled(true); }) .catch(() => {}); + if (platform.capabilities.nodeAdmin) { + hubFetch(`/v1/users/${user.username}/pubkeys`, { token: user.token }) + .then(data => setHasNodeKey(Boolean(data.pk_node_ed25519))) + .catch(() => {}); + } fetchNotifications(); }, [user]); @@ -5226,6 +5781,9 @@ function App() { .then(data => setGroups(data.groups || [])) .catch(() => {}); }} />`; + } else if (route === '/node' && platform.capabilities.nodeAdmin && hasNodeKey) { + page = html`<${NodePage} token=${user.token} username=${user.username} + userId=${user.userId} groups=${groups} />`; } else if (route.startsWith('/group/')) { const groupId = route.slice(7); const group = groups.find(g => g.id === groupId); @@ -5274,7 +5832,8 @@ function App() { presence=${presence} route=${route} menuOpen=${menuOpen} - role=${user.role} />`} + role=${user.role} + hasNodeKey=${hasNodeKey} />`} ${menuOpen && html`<div class="overlay visible" onClick=${() => setMenuOpen(false)} />`} <main class="main"> diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index cb4b349..4193f00 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -442,6 +442,63 @@ export default { 'group.leave_confirm': 'Leave “{name}”? You will lose access to its files and chat. Files you uploaded stay on the node, and the node keeps the identity it pinned for you until its operator removes it.', 'group.mute': 'Mute notifications', 'group.unmute': 'Unmute notifications', + 'group.retry': 'Retry', 'profile.title': 'Profile', 'usermenu.profile': 'Profile', + + // Node management (D5) + 'sidebar.node': 'Node', + 'node.title': 'Node', + 'node.offline': 'Node is offline', + 'node.no_groups': 'No groups configured on this node.', + 'node.retry': 'Retry', + 'node.files': { one: '1 file', other: '{n} files' }, + 'node.peers': { one: '1 peer', other: '{n} peers' }, + 'node.no_gek': 'No group key', + 'node.roots': 'Directories', + 'node.upload_root': 'uploads', + 'node.unavailable': 'unavailable', + 'node.add_root': 'Add directory', + 'node.remove_root': 'Remove', + 'node.root_added': 'Directory added.', + 'node.root_removed': 'Directory removed. Restart recommended to update the index.', + 'node.root_remove_confirm': 'Remove "{name}" from this group?', + 'node.not_operator': 'Could not reach your node. Make sure it is running.', + 'node.gek': 'Group key', + 'node.gek_rotate': 'Rotate group key', + 'node.gek_rotate_confirm': 'Rotate the group key? Connected members will receive the new key automatically. Content already downloaded is unaffected.', + 'node.gek_rotated': 'Group key rotated.', + 'node.gek_init_hint': 'No group key yet. Run GEK init from the CLI to set one up.', + 'node.roster': 'Roster', + 'node.roster_load': 'Load roster', + 'node.roster_empty': 'No members pinned yet.', + 'node.roster_user': 'User', + 'node.roster_role': 'Role', + 'node.roster_status': 'Status', + 'node.roster_via': 'Pinned via', + 'node.roster_invites': { one: '1 pending invite', other: '{n} pending invites' }, + 'node.unpin': 'Unpin', + 'node.unpin_confirm': 'Unpin "{name}"? They will need to pair again.', + 'node.unpin_done': 'Identity unpinned.', + 'node.denylist': 'Denylist', + 'node.denylist_load': 'Load denylist', + 'node.denylist_empty': 'Denylist is empty.', + 'node.denylist_count': { one: '1 entry', other: '{n} entries' }, + 'node.denylist_remove': 'Remove', + 'node.denylist_clear_all': 'Clear all', + 'node.denylist_clear_confirm': 'Remove "{subject}" from the denylist?', + 'node.denylist_cleared': 'Denylist entry removed.', + 'node.reload': 'Reload config', + 'node.reloaded': 'Config reloaded. Root changes on existing groups are now active.', + 'node.attach_group': 'Add group', + 'node.attach_pick': 'Group to host', + 'node.attach_dir': 'Shared directory', + 'node.attach_upload_label': 'Upload directory (for file uploads and chat attachments)', + 'node.attach_upload_dir': 'Upload directory', + 'node.attached': 'Group added and loaded. Run meshbay-node gek-init --group {name} if not done yet.', + 'node.restart_needed': 'After adding: initialise the group key (meshbay-node gek-init --group <name>).', + 'node.detach_group': 'Remove group', + 'node.detach_confirm': 'Remove "{name}" from this node?', + 'node.detached': 'Group removed.', + 'node.stale': 'not on hub', }; diff --git a/packages/meshbay-hub/src/meshbay_hub/static/platform.js b/packages/meshbay-hub/src/meshbay_hub/static/platform.js index 6f8f654..d379e9d 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/platform.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/platform.js @@ -201,8 +201,14 @@ export const folder = { async forget() { return bridge && bridge.folder ? bridge.folder.forget() : false; }, }; +/** Pick a directory to add as a group root. Returns { path, name } or null. */ +export const rootPicker = { + available: Boolean(bridge && bridge.rootPicker), + async choose() { return bridge && bridge.rootPicker ? bridge.rootPicker.choose() : null; }, +}; + export default { isNative, hubBase, capabilities, secrets, nativeSave, - apiFetch, device, bridgeMessage, folder }; + apiFetch, device, bridgeMessage, folder, rootPicker }; // Also a global, because `transport.js` is loaded as a classic script — it // predates the module graph and exposes `MeshBayTransport` the same way. The @@ -211,5 +217,5 @@ export default { isNative, hubBase, capabilities, secrets, nativeSave, if (typeof window !== 'undefined') { window.MeshBayPlatform = { isNative, hubBase, capabilities, secrets, nativeSave, apiFetch, device, - bridgeMessage, folder }; + bridgeMessage, folder, rootPicker }; } diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css index 9ff5923..ac3d08f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/style.css +++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css @@ -2048,3 +2048,177 @@ a.transfer-name { margin-left: 6px; } .dir-row .root-offline { white-space: nowrap; } + +/* ── Node management (D5) ────────────────────────────────────────────────── */ + +.node-page { max-width: 700px; } + +.node-group { + background: var(--bg-surface); + border: 1px solid var(--border); + border-radius: 8px; + padding: 16px 20px; + margin-bottom: 16px; +} +.node-group-stale { + opacity: 0.65; + border-color: var(--warn, #d97706); +} +.node-group-header { + display: flex; + align-items: baseline; + justify-content: space-between; + gap: 12px; + flex-wrap: wrap; + margin-bottom: 12px; +} +.node-group-header h3 { font-size: 1.05em; margin: 0; } +.node-group-stats { + display: flex; + gap: 12px; + font-size: 0.85em; + color: var(--text-secondary); +} +.node-warn { color: var(--warn, #d97706); font-weight: 500; } + +.node-roots-header { margin-bottom: 8px; } + +.node-root { + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; + padding: 6px 0; +} +.node-root + .node-root { border-top: 1px solid var(--border); } +.node-root-info { + display: flex; + align-items: center; + gap: 8px; + flex-wrap: wrap; + min-width: 0; +} +.node-root-name { + display: flex; + align-items: center; + gap: 4px; + font-weight: 500; +} +.node-root-name .icon { width: 16px; height: 16px; flex-shrink: 0; } +.node-root-badge { + font-size: 0.75em; + padding: 1px 6px; + border-radius: 4px; + background: var(--accent); + color: var(--accent-text); +} +.node-root-badge-warn { + background: var(--warn, #d97706); + color: #fff; +} +.node-root-kind { + font-size: 0.8em; + color: var(--text-dim); +} +.node-root-unavailable { opacity: 0.6; } + +.node-message { + font-size: 0.9em; + color: var(--text-secondary); + margin-bottom: 12px; + padding: 8px 12px; + background: var(--bg-raised); + border-radius: 6px; +} + +.btn-small { + padding: 4px 10px; + font-size: 0.82em; + border-radius: 5px; +} + +.node-roots > .btn-small { margin-top: 8px; } + +.node-header { + display: flex; + align-items: baseline; + justify-content: space-between; + gap: 12px; + margin-bottom: 4px; +} +.node-header h2 { margin-bottom: 0; } + +.node-section { margin-top: 16px; } +.node-section-header { + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; + margin-bottom: 8px; +} +.node-hint { + font-size: 0.85em; + color: var(--text-secondary); + margin: 4px 0; +} + +.node-table { + width: 100%; + font-size: 0.85em; + border-collapse: collapse; + margin-top: 4px; +} +.node-table th { + text-align: left; + padding: 4px 8px; + border-bottom: 1px solid var(--border); + color: var(--text-secondary); + font-weight: 500; +} +.node-table td { + padding: 4px 8px; + border-bottom: 1px solid var(--border); +} + +.node-deny-entry { + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; + padding: 4px 0; + font-size: 0.85em; +} +.node-deny-entry + .node-deny-entry { border-top: 1px solid var(--border); } + +.node-attach-form { + display: flex; + flex-direction: column; + gap: 8px; + margin-top: 8px; + max-width: 400px; +} +.node-attach-form label { + font-size: 0.82em; + color: var(--text-secondary); + margin-top: 4px; +} +.node-attach-form input, +.node-attach-form select { + padding: 6px 10px; + border: 1px solid var(--border); + border-radius: 5px; + background: var(--bg-base); + color: var(--text); + font-size: 0.9em; + cursor: pointer; +} +.node-attach-form select option { + background: var(--bg-base); + color: var(--text); +} +.node-attach-dir-row { + display: flex; + gap: 6px; + align-items: center; +} +.node-attach-dir-row input { flex: 1; } diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index a0509a0..17886f4 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -217,6 +217,9 @@ class MeshBayTransport { await this._pc.setRemoteDescription({ type: 'answer', sdp: answer.sdp }); await channelReady; + console.log('[MeshBay] DataChannel open, sending handshake for group', groupId, + 'channel=', this._channel?.readyState, + 'crypto=', !!window.MeshBayCrypto); // The client nonce is what makes the NODE's proof fresh (C3) — without it a // recorded handshake_ack could be replayed by an impersonating peer. @@ -229,6 +232,7 @@ class MeshBayTransport { group_id: groupId || '', nonce: window.MeshBayCrypto.b64encode(this._nonceClient), }); + console.log('[MeshBay] Handshake reply:', reply.type); if (reply.type === 'handshake_challenge') { if (!window.MeshBayCrypto) { @@ -380,11 +384,9 @@ class MeshBayTransport { // A node that answers a handshake with anything other than a challenge is not // running the mutual protocol. Accepting a bare handshake_ack here would let a // peer skip proving GEK possession entirely (C3/C6). + console.warn('[MeshBay] Handshake rejected:', reply.detail, 'code:', reply.code); const rejected = new Error( 'MNP handshake rejected: ' + (reply.detail || `unexpected ${reply.type}`)); - // `not_a_member` usually means our token predates being added to the group; - // the caller refreshes it and tries again rather than showing that to someone - // who was invited thirty seconds ago. rejected.reason = reply.code || ''; throw rejected; } @@ -622,6 +624,112 @@ class MeshBayTransport { return msg; } + // ── Node management (D5) ─────────────────────────────────────────────── + + async fetchNodeStatus() { + const msg = await this._sendAndWait({ type: 'node_status', v: '0.1' }); + if (msg.type === 'error') throw new Error(msg.detail); + return msg; + } + + async addRoot(groupId, path, { name, kind, upload } = {}, signFn) { + const msg = await this._sendAndWait({ + type: 'root_add', v: '0.1', + group_id: groupId, path, + name: name || '', kind: kind || 'generic', upload: !!upload, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, 'root_add', path, signFn); + } + return msg; + } + + async removeRoot(groupId, rootName, signFn) { + const msg = await this._sendAndWait({ + type: 'root_remove', v: '0.1', + group_id: groupId, root_name: rootName, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, 'root_remove', rootName, signFn); + } + return msg; + } + + async unpinMember(userId, signFn) { + const msg = await this._sendAndWait({ + type: 'member_unpin', v: '0.1', user_id: userId, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, 'member_unpin', userId, signFn); + } + return msg; + } + + async rotateGek(groupId, signFn) { + const msg = await this._sendAndWait({ + type: 'gek_rotate', v: '0.1', group_id: groupId, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, 'gek_rotate', groupId, signFn); + } + return msg; + } + + async fetchRoster(groupId) { + const msg = await this._sendAndWait({ + type: 'roster_read', v: '0.1', group_id: groupId || '', + }); + if (msg.type === 'error') throw new Error(msg.detail); + return msg; + } + + async fetchDenylist() { + const msg = await this._sendAndWait({ type: 'denylist_read', v: '0.1' }); + if (msg.type === 'error') throw new Error(msg.detail); + return msg; + } + + async clearDenylist(subject) { + const msg = await this._sendAndWait({ + type: 'denylist_clear', v: '0.1', subject: subject || '', + }); + if (msg.type === 'error') throw new Error(msg.detail); + return msg; + } + + async attachGroup(name, sharedDir, uploadDir, signFn) { + const msg = await this._sendAndWait({ + type: 'group_attach', v: '0.1', + name, shared_dir: sharedDir, upload_dir: uploadDir || '', + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, 'group_attach', name, signFn); + } + return msg; + } + + async detachGroup(name, signFn) { + const msg = await this._sendAndWait({ + type: 'group_detach', v: '0.1', name, + }); + if (msg.type === 'error') throw new Error(msg.detail); + if (msg.type === 'admin_challenge') { + return this._authorizeAdminOp(msg, 'group_detach', name, signFn); + } + return msg; + } + + async reloadConfig() { + const msg = await this._sendAndWait({ type: 'node_reload', v: '0.1' }); + if (msg.type === 'error') throw new Error(msg.detail); + return msg; + } + /** * Ask for a video stream, and say how much we can take. * @@ -1003,6 +1111,8 @@ class MeshBayTransport { const id = this._seqId++; const timeout = setTimeout(() => { this._pending.delete(id); + console.error('[MeshBay] Response timeout for', obj.type, + 'after', timeoutMs, 'ms, channel=', this._channel?.readyState); reject(new Error('Response timeout')); }, timeoutMs); this._pending.set(id, { @@ -1040,6 +1150,10 @@ class MeshBayTransport { _onMessage(data) { const incoming = new Uint8Array(data); + this._msgCount = (this._msgCount || 0) + 1; + if (this._msgCount <= 3) { + console.log('[MeshBay] recv', incoming.length, 'bytes, msg #' + this._msgCount); + } const combined = new Uint8Array(this._recvBuf.length + incoming.length); combined.set(this._recvBuf); combined.set(incoming, this._recvBuf.length); diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 6d4f172..7e3ebc1 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -111,6 +111,7 @@ class NodeDaemon: "quic_port": config.node.quic_port, "endpoint_hint": None, "indexes": {}, + "indexers": {}, } self._quic_server = None self._webrtc = None @@ -243,6 +244,7 @@ class NodeDaemon: await indexer.start() self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index + self._state["indexers"][group_cfg.id] = indexer log.info("Indexing group %s: %s (%d files)", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots), @@ -416,12 +418,15 @@ class NodeDaemon: self._state["webrtc"] = self._webrtc self._state["quic_server"] = self._quic_server self._state["hub"] = hub + self._state["reload_fn"] = self._reload_config # Rotating a key has to reach every transport holding a copy of it, # and clearing the denylist has to reach the one the handshake # consults — so both are published rather than reachable only # through the object that happens to own them. self._state["denylist"] = self._denylist self._state["pk_x25519_raw"] = pk_x_raw + self._state["sk_x25519_raw"] = sk_x_raw + self._state["sk_ed25519"] = keys.sk_ed25519 self._state["status"] = "running" log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", @@ -460,19 +465,13 @@ class NodeDaemon: async def _reload_config(self) -> None: """ - Re-read node.toml on SIGHUP. + Re-read node.toml and reconcile groups. - Deliberately narrow: it picks up **root changes on groups already - hosted**, which is what an operator adjusts day to day, and reports - anything else as needing a restart. Adding or removing a whole group - means new indexers, chat stores, GEK loads and transport contexts, and - doing that under a live daemon is how a half-built group ends up serving - content. Saying "restart for that" is honest and costs one restart. - - Nothing here touches connections: a member watching a film keeps - watching it. + Handles root changes on existing groups, hot-loads new groups, and + tears down removed groups. Existing connections are untouched: a member + watching a film keeps watching it. """ - log.info("SIGHUP — re-reading %s", self._config_path) + log.info("Reloading config from %s", self._config_path) try: fresh = load_config(self._config_path) except Exception as e: @@ -482,13 +481,8 @@ class NodeDaemon: groups_ctx = self._state.get("groups_ctx") or {} hosted = set(groups_ctx) incoming = {g.id for g in fresh.groups if g.id} - if incoming != hosted: - added = ", ".join(sorted(incoming - hosted)) or "none" - removed = ", ".join(sorted(hosted - incoming)) or "none" - log.warning("Group set changed (added: %s, removed: %s) — restart the " - "daemon for that; roots of existing groups reloaded anyway", - added, removed) + # ── Root changes on existing groups ────────────────────────────── changed = 0 for group_cfg in fresh.groups: ctx = groups_ctx.get(group_cfg.id) @@ -515,9 +509,112 @@ class NodeDaemon: ctx["roots"] = roots changed += 1 + # ── Hot-load new groups ────────────────────────────────────────── + added_names = [] + sk_ed = self._state.get("sk_ed25519") + sk_x_raw = self._state.get("sk_x25519_raw") + pk_x_raw = self._state.get("pk_x25519_raw") + node_user_id = self._state.get("node_user_id") + data_dir = fresh.data_dir + + for group_cfg in fresh.groups: + if group_cfg.id in hosted: + continue + if not group_cfg.id or not group_cfg.roots: + log.warning("New group %r has no id or roots — skipping", + group_cfg.name) + continue + if not sk_ed: + log.warning("Cannot hot-load %r — signing key not available", + group_cfg.name) + continue + + try: + roots = RootSet.build([asdict(r) for r in group_cfg.roots]) + except RootError as e: + log.error("New group %r: %s — skipping", group_cfg.name, e) + continue + roots.refresh_availability() + + gek = None + if group_cfg.visibility == "private" and sk_x_raw and pk_x_raw: + gek = await self._load_gek( + group_cfg.id, node_user_id, sk_x_raw, pk_x_raw) + if gek: + log.info("GEK loaded for new group %s", group_cfg.id[:8]) + + indexer = DirectoryIndexer( + roots=roots, + group_id=group_cfg.id, + sk_node=sk_ed, + gek=gek, + on_change=self._on_index_change, + ) + await indexer.start() + self._indexers.append(indexer) + self._state["indexes"][group_cfg.id] = indexer.index + self._state["indexers"][group_cfg.id] = indexer + + data_dir.mkdir(parents=True, exist_ok=True) + chat_db = data_dir / group_cfg.id[:16] / "chat.db" + store = ChatStore(db_path=chat_db) + await store.open() + self._chat_stores[group_cfg.id] = store + + new_ctx = { + "gek": gek, + "roots": roots, + "index": indexer.index, + "visibility": group_cfg.visibility, + "join_policy": group_cfg.join_policy, + "member_upload": ( + await self._roster.member_upload_allowed(group_cfg.id) + if self._roster else True), + "chat_store": store, + } + groups_ctx[group_cfg.id] = new_ctx + + if self._webrtc: + self._webrtc._ctx["groups"][group_cfg.id] = new_ctx + log.info("Hot-loaded group %s (%s, %d roots)", + group_cfg.name, group_cfg.id[:8], len(roots)) + added_names.append(group_cfg.name) + + # ── Tear down removed groups ───────────────────────────────────── + removed_names = [] + for gid in hosted - incoming: + indexer = next((i for i in self._indexers + if i.group_id == gid), None) + if indexer: + try: + await indexer.stop() + except Exception: + pass + self._indexers.remove(indexer) + store = self._chat_stores.pop(gid, None) + if store: + try: + await store.close() + except Exception: + pass + self._state["indexes"].pop(gid, None) + self._state["indexers"].pop(gid, None) + old_name = gid[:8] + for g_cfg in self._config.groups: + if g_cfg.id == gid: + old_name = g_cfg.name + break + groups_ctx.pop(gid, None) + if self._webrtc and self._webrtc._ctx.get("groups") is not groups_ctx: + self._webrtc._ctx["groups"].pop(gid, None) + log.info("Unloaded group %s (%s)", old_name, gid[:8]) + removed_names.append(old_name) + self._config = fresh self._state["config"] = fresh - log.info("Reload complete — %d group(s) re-rooted", changed) + self._state["groups"] = [g.name for g in fresh.groups] + log.info("Reload complete — %d re-rooted, %d added, %d removed", + changed, len(added_names), len(removed_names)) async def _login_with_retry(self, hub: HubClient): """Login to hub, retrying if the node key hasn't been linked yet.""" @@ -782,16 +879,18 @@ def main() -> None: parser.add_argument("command", nargs="?", choices=["init", "status", "ui", "gek-init", "gek", "operator", "member", "group", "file", - "denylist", "reload", "calibrate-argon2"], + "denylist", "reload", "restart-daemon", + "calibrate-argon2"], help="init: write example config | status: node state and keys " "| ui: print the admin UI URL | operator pair: pair a " "browser with this node | member list|invite|revoke|unpin " - "| group list|add | gek init|rotate | file list|rm " + "| group list|add|remove | gek init|rotate | file list|rm " "| denylist show|clear | reload: re-read node.toml " + "| restart-daemon: full stop + start " "| calibrate-argon2: benchmark") parser.add_argument("subcommand", nargs="?", help="'pair' for operator; list|invite|revoke|unpin for " - "member; list|add for group; init|rotate for gek; " + "member; list|add|remove for group; init|rotate for gek; " "list|rm for file; show|clear for denylist") parser.add_argument("target", nargs="?", help="username for member invite|revoke|unpin; group name " @@ -813,7 +912,8 @@ def main() -> None: # Query commands print a report; library logging would interleave with it. quiet = args.command in ("status", "ui", "gek-init", "gek", "operator", - "member", "group", "file", "denylist", "reload") + "member", "group", "file", "denylist", "reload", + "restart-daemon") logging.basicConfig( level=logging.ERROR if quiet else getattr(logging, args.log_level), format="%(asctime)s %(levelname)-8s %(name)s: %(message)s", @@ -1051,6 +1151,49 @@ def main() -> None: print("watch the result: tail -f /tmp/meshbay-node.log") return + if args.command == "restart-daemon": + import os as _os + import signal as _signal + import subprocess as _subprocess + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + pid_out = _subprocess.run( + ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"], + capture_output=True, text=True) + pids = [int(x) for x in pid_out.stdout.split()] + if pids: + for pid in pids: + _os.kill(pid, _signal.SIGTERM) + print(f"stopped {len(pids)} daemon process(es)") + for pid in pids: + try: + _os.waitpid(pid, 0) + except ChildProcessError: + import time as _time + _time.sleep(2) + else: + print("no running daemon found — starting fresh") + log_path = "/tmp/meshbay-node.log" + config_flag = ["--config", str(args.config)] if args.config else [] + _subprocess.Popen( + [sys.executable, "-m", "meshbay_node.daemon"] + config_flag, + stdout=open(log_path, "a"), + stderr=_subprocess.STDOUT, + start_new_session=True, + ) + import time as _time + _time.sleep(3) + pid_out2 = _subprocess.run( + ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"], + capture_output=True, text=True) + new_pids = [int(x) for x in pid_out2.stdout.split()] + if new_pids: + print(f"daemon started (PID {new_pids[0]})") + print(f"log: tail -f {log_path}") + else: + print(f"daemon may have failed to start — check {log_path}") + sys.exit(1) + return + if args.command == "denylist": cfg = load_config(args.config or DEFAULT_CONFIG_PATH) sub = args.subcommand or "show" @@ -1159,8 +1302,26 @@ def main() -> None: f"--group {g['name']}") return + if args.subcommand == "remove": + if not args.target: + print("usage: meshbay-node group remove <name>") + sys.exit(1) + cfg = load_config(args.config or DEFAULT_CONFIG_PATH) + if not args.yes: + answer = input(f"Remove group '{args.target}' from this node? [y/N] ") + if answer.lower() not in ("y", "yes"): + print("cancelled") + return + out = _daemon_api(cfg, "/api/groups/detach", method="POST", + body={"name": args.target}) + print(f"{out['name']} ({out['group_id'][:8]}) removed from {out['config']}") + print() + print("Restart the daemon to stop hosting it:") + print(" meshbay-node restart-daemon") + return + if args.subcommand != "add": - print("usage: meshbay-node group list|add <name> --dir <path> [--upload-dir <path>]") + print("usage: meshbay-node group list|add|remove <name>") sys.exit(1) if not args.target or not args.dir: print("usage: meshbay-node group add <name> --dir <path> [--upload-dir <path>]") diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 20345bb..2a76290 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -23,6 +23,7 @@ in the adapter. from __future__ import annotations import logging +import re from dataclasses import asdict from pathlib import Path from typing import Any @@ -93,9 +94,11 @@ async def read_roster(state: dict, group_id: str = "") -> dict: roster = state.get("roster") if not roster: return {"identities": [], "members": [], "invites": []} + members = await roster.list_members(group_id or None) + members = [m for m in members if m.get("pk_ed25519") is not None] return { "identities": await roster.list_identities(), - "members": await roster.list_members(group_id or None), + "members": members, "invites": await roster.list_invites(), } @@ -387,6 +390,125 @@ async def attach_group(state: dict, name: str, shared_dir: str, return result +async def detach_group(state: dict, name: str) -> dict: + """ + Remove a [[groups]] block from node.toml. + + Does not touch the hub — only stops this node from hosting the group + after the next reload or restart. + """ + if not name: + raise OpError("group name or id is required") + config = _config(state) + + match = [g for g in config.groups if g.id == name or g.name == name] + if not match: + raise OpError(f"No hosted group matches {name!r}", status=404, + extra={"available": [{"name": g.name, "id": g.id} + for g in config.groups]}) + group = match[0] + + conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) + text = conf_path.read_text() + lines = text.split("\n") + + rng = _find_group_range(lines, group.id) + if rng is None: + raise OpError(f"Group {group.id[:8]} not found in {conf_path}") + + start, end = rng + while end < len(lines) and lines[end].strip() == "": + end += 1 + + new_lines = lines[:start] + lines[end:] + conf_path.write_text("\n".join(new_lines)) + log.info("Group detached: %s (%s) removed from %s", group.name, group.id[:8], conf_path) + + return {"group_id": group.id, "name": group.name, "config": str(conf_path), + "note": "restart the node to stop hosting it"} + + +def _find_group_range(lines: list[str], group_id: str) -> tuple[int, int] | None: + """Line range of a [[groups]] block by id: (start, end_exclusive).""" + id_re = re.compile(r'^\s*id\s*=\s*"([^"]*)"') + block_starts: list[int] = [] + for i, line in enumerate(lines): + if line.strip() == "[[groups]]": + block_starts.append(i) + + for j, start in enumerate(block_starts): + boundary = block_starts[j + 1] if j + 1 < len(block_starts) else len(lines) + for k in range(start + 1, boundary): + s = lines[k].strip() + if s.startswith("[") and s != "[[groups.roots]]": + boundary = k + break + for k in range(start + 1, boundary): + m = id_re.match(lines[k]) + if m and m.group(1) == group_id: + return (start, boundary) + return None + + +def _insert_roots_block(conf_path: Path, group_id: str, + root_block: str) -> None: + """Append a [[groups.roots]] block inside the matching [[groups]] section.""" + text = conf_path.read_text() + lines = text.split("\n") + + rng = _find_group_range(lines, group_id) + if rng is None: + raise OpError(f"Group {group_id[:8]} not found in {conf_path}") + + _start, end = rng + insert_at = end + while insert_at > _start + 1 and lines[insert_at - 1].strip() == "": + insert_at -= 1 + + new_lines = (lines[:insert_at] + + [""] + + root_block.rstrip("\n").split("\n") + + lines[insert_at:]) + conf_path.write_text("\n".join(new_lines)) + + +def _remove_roots_block(conf_path: Path, group_id: str, + resolved_path: str) -> None: + """Remove a [[groups.roots]] block whose resolved path matches.""" + text = conf_path.read_text() + lines = text.split("\n") + + rng = _find_group_range(lines, group_id) + if rng is None: + raise OpError(f"Group {group_id[:8]} not found in {conf_path}") + + start, end = rng + path_re = re.compile(r'^\s*path\s*=\s*"([^"]*)"') + roots_starts: list[int] = [] + for i in range(start + 1, end): + if lines[i].strip() == "[[groups.roots]]": + roots_starts.append(i) + + for j, rs in enumerate(roots_starts): + rs_end = roots_starts[j + 1] if j + 1 < len(roots_starts) else end + for k in range(rs, rs_end): + m = path_re.match(lines[k]) + if m: + try: + p = str(Path(m.group(1)).expanduser().resolve()) + except OSError: + continue + if p == resolved_path: + rm_start = rs + if rm_start > 0 and lines[rm_start - 1].strip() == "": + rm_start -= 1 + new_lines = lines[:rm_start] + lines[rs_end:] + conf_path.write_text("\n".join(new_lines)) + return + + raise OpError(f"Root path not found in config", status=404) + + async def add_root(state: dict, group_id: str, path: str, *, name: str = "", kind: str = "generic", upload: bool = False) -> dict: @@ -410,22 +532,85 @@ async def add_root(state: dict, group_id: str, path: str, *, raise OpError(str(e)) from e added = built.roots[-1] + + try: + added.path.mkdir(parents=True, exist_ok=True) + except OSError as e: + raise OpError(f"Cannot create {added.path}: {e}") from e + conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) - raise OpError( - # Writing into the middle of a hand-written TOML file means finding the - # right [[groups]] block and appending inside it, which a text append - # cannot do. Until that is written, say so plainly rather than appending - # to the wrong group. - f"Add this to {conf_path} under the [[groups]] block for " - f"{cfg.name!r}, then restart the node:\n\n" - f' [[groups.roots]]\n' - f' path = "{added.path}"\n' - + (f' name = "{added.name}"\n' if name else "") - + (f' kind = "{added.kind}"\n' if kind != "generic" else "") - + (f' upload = true\n' if upload else ""), - status=501, - extra={"validated": True, "name": added.name, "path": str(added.path)}, - ) + root_block = f' [[groups.roots]]\n path = "{added.path}"' + if name: + root_block += f'\n name = "{added.name}"' + if kind != "generic": + root_block += f'\n kind = "{added.kind}"' + if upload: + root_block += f'\n upload = true' + _insert_roots_block(conf_path, group_id, root_block) + + from meshbay_node.config import RootSpec + cfg.roots.append(RootSpec( + path=str(added.path), name=added.name, kind=added.kind, + upload=added.upload, direct=added.direct)) + groups_ctx = state.get("groups_ctx", {}) + if group_id in groups_ctx: + groups_ctx[group_id]["roots"] = built + + log.info("Root added: %s → group %s", added.name, group_id[:8]) + return {"status": "added", "name": added.name, "path": str(added.path), + "group_id": group_id, "roots": built.describe()} + + +async def remove_root(state: dict, group_id: str, root_name: str) -> dict: + """Remove a named root from a group. At least one root must remain.""" + config = _config(state) + cfg = next((g for g in config.groups if g.id == group_id), None) + if cfg is None: + raise OpError("Group not configured on this node", status=404) + + from meshbay_common.paths import fold + from meshbay_node.roots import derive_name + target = fold(root_name) + match_idx = None + for i, r in enumerate(cfg.roots): + try: + rname = r.name or derive_name(Path(r.path).expanduser().resolve()) + except Exception: + continue + if fold(rname) == target: + match_idx = i + break + + if match_idx is None: + raise OpError(f"No root named {root_name!r} in this group", status=404) + if len(cfg.roots) < 2: + raise OpError("Cannot remove the only root", status=400) + + removed = cfg.roots[match_idx] + if removed.upload: + raise OpError( + "Cannot remove the upload root — file uploads and chat " + "attachments are stored there", status=400) + resolved = str(Path(removed.path).expanduser().resolve()) + + conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH) + _remove_roots_block(conf_path, group_id, resolved) + + cfg.roots.pop(match_idx) + remaining = [asdict(r) for r in cfg.roots] + try: + built = RootSet.build(remaining) + except RootError: + built = None + if built is not None: + groups_ctx = state.get("groups_ctx", {}) + if group_id in groups_ctx: + groups_ctx[group_id]["roots"] = built + + log.info("Root removed: %s from group %s", root_name, group_id[:8]) + return {"status": "removed", "name": root_name, "group_id": group_id, + "roots": built.describe() if built else [], + "note": "restart recommended to update the file index"} # ── Files ──────────────────────────────────────────────────────────────────── diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index c811016..9a52b53 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -359,6 +359,8 @@ class Roster: assert self._db cur = await self._db.execute( "DELETE FROM identities WHERE user_id = ?", (user_id,)) + await self._db.execute( + "DELETE FROM members WHERE user_id = ?", (user_id,)) await self._db.commit() return cur.rowcount > 0 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 f182ab2..1f1f2d2 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -63,6 +63,10 @@ from meshbay_common.adminop import ( OP_GEK_ROTATE, OP_MEMBER_UNPIN, OP_MEMBER_UPLOAD, + OP_ROOT_ADD, + OP_ROOT_REMOVE, + OP_GROUP_ATTACH, + OP_GROUP_DETACH, admin_transcript, ) from meshbay_common.crypto import pk_to_b64, wrap_gek_aes @@ -385,11 +389,16 @@ class WebRTCPeerSession: def _setup_channel(self, channel: RTCDataChannel) -> None: self._channel = channel + self._msg_count = 0 @channel.on("message") def on_message(message): if isinstance(message, str): message = message.encode() + self._msg_count += 1 + if self._msg_count <= 3: + log.info("WebRTC data received: %d bytes, msg #%d (peer=%s)", + len(message), self._msg_count, self._peer_id) self._buffer.feed(message) for msg in self._buffer.messages(): self._handle_message(msg) @@ -476,6 +485,24 @@ class WebRTCPeerSession: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: self._do_gek_rotate(msg) + elif mtype == MNP.NODE_STATUS: + self._spawn(self._do_node_status(msg)) + elif mtype == MNP.ROOT_ADD: + self._do_root_add(msg) + elif mtype == MNP.ROOT_REMOVE: + self._do_root_remove(msg) + elif mtype == MNP.ROSTER_READ: + self._spawn(self._do_roster_read(msg)) + elif mtype == MNP.DENYLIST_READ: + self._spawn(self._do_denylist_read(msg)) + elif mtype == MNP.DENYLIST_CLEAR: + self._spawn(self._do_denylist_clear(msg)) + elif mtype == MNP.GROUP_ATTACH: + self._do_group_attach(msg) + elif mtype == MNP.GROUP_DETACH: + self._do_group_detach(msg) + elif mtype == MNP.NODE_RELOAD: + self._spawn(self._do_node_reload(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: self._spawn(self._do_keypair_bundle_store(msg)) elif mtype == MNP.KEYPAIR_BUNDLE_DELETE: @@ -565,6 +592,8 @@ class WebRTCPeerSession: def _do_handshake(self, msg: dict) -> None: group_id = msg.get("group_id", "") + log.info("WebRTC handshake request: group=%s (peer=%s)", + group_id[:8] if group_id else "none", self._peer_id) try: peer = authorize_token( msg.get("token", ""), @@ -598,6 +627,7 @@ class WebRTCPeerSession: gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx if not gctx.get("gek"): + log.warning("Handshake refused — no GEK for group=%s", peer.group_id[:8]) self._send({ "type": "error", "detail": "Group encryption not initialized — contact node operator", @@ -606,6 +636,7 @@ class WebRTCPeerSession: self._gek_challenge = os.urandom(NONCE_LEN) self._nonce_node = self._gek_challenge + log.info("WebRTC handshake challenge sent (peer=%s)", self._peer_id) self._send({ "type": MNP.HANDSHAKE_CHALLENGE, "v": MNP_VERSION, @@ -1510,13 +1541,14 @@ class WebRTCPeerSession: one. The node generates the replacement itself — nothing arriving here contributes key material, which is what the C5b rule is about. """ - if not self._group_id: + group_id = str(msg.get("group_id", "")).strip() or self._group_id + if not group_id: self._send({"type": "error", "detail": "No group on this connection"}) return if not self._has_admin_authority(): self._send({"type": "error", "detail": "No authorized key for this"}) return - self._issue_admin_challenge(OP_GEK_ROTATE, self._group_id) + self._issue_admin_challenge(OP_GEK_ROTATE, group_id, group_id=group_id) async def _admin_exec_gek_rotate( self, pending: dict, transcript: bytes, sig: bytes, @@ -1627,6 +1659,241 @@ class WebRTCPeerSession: except Exception: pass + # ── Node management (D5) ───────────────────────────────────────────────── + + async def _do_node_status(self, msg: dict) -> None: + """All groups, roots, peers — the operator's overview.""" + node_uid = self._ctx.get("node_user_id") + log.info("node_status: user=%s node_user=%s admin=%s", + self._user_id, node_uid, self._is_node_admin()) + if not self._is_node_admin(): + self._send({"type": "error", "detail": "Not the node operator"}) + return + try: + result = await self._run_op(ops.list_groups) + self._send({"type": MNP.NODE_STATUS_ACK, "v": MNP_VERSION, **result}) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + except Exception as e: + log.error("node_status failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + + async def _do_roster_read(self, msg: dict) -> None: + if not self._is_node_admin(): + self._send({"type": "error", "detail": "Not the node operator"}) + return + group_id = str(msg.get("group_id", "")).strip() + try: + result = await self._run_op(ops.read_roster, group_id) + self._send({"type": MNP.ROSTER_READ_ACK, "v": MNP_VERSION, **result}) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + except Exception as e: + log.error("roster_read failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + + async def _do_denylist_read(self, msg: dict) -> None: + if not self._is_node_admin(): + self._send({"type": "error", "detail": "Not the node operator"}) + return + try: + result = await self._run_op(ops.read_denylist) + self._send({"type": MNP.DENYLIST_READ_ACK, "v": MNP_VERSION, **result}) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + except Exception as e: + log.error("denylist_read failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + + async def _do_denylist_clear(self, msg: dict) -> None: + if not self._is_node_admin(): + self._send({"type": "error", "detail": "Not the node operator"}) + return + subject = str(msg.get("subject", "")).strip() + try: + result = await self._run_op(ops.clear_denylist, subject=subject) + self._audit("denylist_clear", subject or "all") + self._send({"type": MNP.DENYLIST_CLEAR_ACK, "v": MNP_VERSION, **result}) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + except Exception as e: + log.error("denylist_clear failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + + def _do_group_attach(self, msg: dict) -> None: + name = str(msg.get("name", "")).strip() + shared_dir = str(msg.get("shared_dir", "")).strip() + if not name or not shared_dir: + self._send({"type": "error", "detail": "Missing name or shared_dir"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + upload_dir = str(msg.get("upload_dir", "")).strip() + self._issue_admin_challenge( + OP_GROUP_ATTACH, name, + payload={"name": name, "shared_dir": shared_dir, + "upload_dir": upload_dir}, + group_id="") + + async def _admin_exec_group_attach( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", + f"group_attach:{pending['subject'][:16]}") + return + p = pending.get("payload") or {} + try: + result = await self._run_op( + ops.attach_group, p["name"], p["shared_dir"], p.get("upload_dir", "")) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("group_attach", pending["subject"]) + self._send({"type": MNP.GROUP_ATTACH_ACK, "v": MNP_VERSION, **result}) + state = self._ctx.get("daemon_state") + reload_fn = state.get("reload_fn") if state else None + if reload_fn: + try: + await reload_fn() + except Exception as e: + log.error("Reload after group_attach failed: %s", e) + + def _do_group_detach(self, msg: dict) -> None: + name = str(msg.get("name", "")).strip() + if not name: + self._send({"type": "error", "detail": "Missing group name or id"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + OP_GROUP_DETACH, name, + payload={"name": name}, + group_id="") + + async def _admin_exec_group_detach( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", + f"group_detach:{pending['subject'][:16]}") + return + p = pending.get("payload") or {} + try: + result = await self._run_op(ops.detach_group, p["name"]) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("group_detach", pending["subject"]) + self._send({"type": MNP.GROUP_DETACH_ACK, "v": MNP_VERSION, **result}) + state = self._ctx.get("daemon_state") + reload_fn = state.get("reload_fn") if state else None + if reload_fn: + try: + await reload_fn() + except Exception as e: + log.error("Reload after group_detach failed: %s", e) + + async def _do_node_reload(self, msg: dict) -> None: + if not self._is_node_admin(): + self._send({"type": "error", "detail": "Not the node operator"}) + return + state = self._ctx.get("daemon_state") + reload_fn = state.get("reload_fn") if state else None + if not reload_fn: + self._send({"type": "error", "detail": "Reload not available"}) + return + try: + await reload_fn() + self._send({"type": MNP.NODE_RELOAD_ACK, "v": MNP_VERSION, + "status": "reloaded"}) + except Exception as e: + log.error("node_reload failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Reload failed"}) + + def _do_root_add(self, msg: dict) -> None: + target_group = str(msg.get("group_id", "")).strip() + path = str(msg.get("path", "")).strip() + if not target_group or not path: + self._send({"type": "error", "detail": "Missing group_id or path"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + OP_ROOT_ADD, path, + payload={ + "group_id": target_group, "path": path, + "name": str(msg.get("name", ""))[:128], + "kind": str(msg.get("kind", "generic"))[:16], + "upload": bool(msg.get("upload", False)), + }, + group_id=target_group) + + async def _admin_exec_root_add( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"root_add:{pending['subject'][:24]}") + return + p = pending["payload"] + try: + result = await self._run_op( + ops.add_root, p["group_id"], p["path"], + name=p.get("name", ""), kind=p.get("kind", "generic"), + upload=p.get("upload", False)) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + except Exception as e: + log.error("root_add failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + return + self._audit("root_add", f"{p['path']}→{p['group_id'][:8]}") + await self._retarget_indexer(p["group_id"]) + self._send({"type": MNP.ROOT_ADD_ACK, "v": MNP_VERSION, **result}) + + def _do_root_remove(self, msg: dict) -> None: + target_group = str(msg.get("group_id", "")).strip() + root_name = str(msg.get("root_name", "")).strip() + if not target_group or not root_name: + self._send({"type": "error", "detail": "Missing group_id or root_name"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + OP_ROOT_REMOVE, root_name, + payload={"group_id": target_group, "root_name": root_name}, + group_id=target_group) + + async def _admin_exec_root_remove( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"root_remove:{pending['subject'][:24]}") + return + p = pending["payload"] + try: + result = await self._run_op( + ops.remove_root, p["group_id"], p["root_name"]) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + except Exception as e: + log.error("root_remove failed: %s", e, exc_info=True) + self._send({"type": "error", "detail": "Internal error"}) + return + self._audit("root_remove", f"{p['root_name']}←{p['group_id'][:8]}") + await self._retarget_indexer(p["group_id"]) + self._send({"type": MNP.ROOT_REMOVE_ACK, "v": MNP_VERSION, **result}) + async def _run_op(self, fn, *args, **kwargs): """ Call an operation from `meshbay_node.ops` with the daemon's own view. @@ -1642,6 +1909,16 @@ class WebRTCPeerSession: raise ops.OpError("Node state not available", status=503) return await fn(state, *args, **kwargs) + async def _retarget_indexer(self, group_id: str) -> None: + """Tell the indexer to rescan after roots changed.""" + state = self._ctx.get("daemon_state") + if not state: + return + indexer = state.get("indexers", {}).get(group_id) + roots = state.get("groups_ctx", {}).get(group_id, {}).get("roots") + if indexer and roots: + await indexer.retarget(roots) + async def _admin_exec_member_revoke( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: @@ -1740,7 +2017,11 @@ class WebRTCPeerSession: """ task = asyncio.ensure_future(coro) self._tasks.add(task) - task.add_done_callback(self._tasks.discard) + def _on_done(t): + self._tasks.discard(t) + if not t.cancelled() and t.exception(): + log.error("Spawned task failed: %s", t.exception(), exc_info=t.exception()) + task.add_done_callback(_on_done) return task def _group_ctx(self) -> dict: @@ -2213,6 +2494,7 @@ class WebRTCPeerSession: def _issue_admin_challenge( self, op: str, subject: str, payload: dict | None = None, + group_id: str | None = None, ) -> None: """ Ask the client to authorize `op` on `subject` with its Ed25519 identity key. @@ -2221,13 +2503,17 @@ class WebRTCPeerSession: rebuild and inspect what it signs. The node keeps the authoritative copy and rebuilds the transcript itself at verification time — nothing signed is ever taken from the response message. + + `group_id` overrides the connection's group for cross-group operations + (e.g. root management from a NodePage connection). """ + gid = group_id if group_id is not None else (self._group_id or "") nonce = os.urandom(32) ts = int(time.time()) op_id = base64.b64encode(os.urandom(16)).decode() self._admin_ops[op_id] = { "op": op, "subject": subject, "nonce": nonce, "ts": ts, - "payload": payload or {}, + "payload": payload or {}, "group_id": gid, } self._send({ "type": MNP.ADMIN_CHALLENGE, @@ -2238,7 +2524,7 @@ class WebRTCPeerSession: "nonce": base64.b64encode(nonce).decode(), "ts": ts, "node_pk": self._node_pk_b64(), - "group_id": self._group_id or "", + "group_id": gid, }) @staticmethod @@ -2327,7 +2613,7 @@ class WebRTCPeerSession: transcript = admin_transcript( op=pending["op"], node_pk_b64=self._node_pk_b64(), - group_id=self._group_id or "", + group_id=pending["group_id"] if pending.get("group_id") is not None else (self._group_id or ""), subject=pending["subject"], nonce=pending["nonce"], ts=pending["ts"], @@ -2354,6 +2640,18 @@ class WebRTCPeerSession: elif pending["op"] == OP_MEMBER_UPLOAD: self._spawn( self._admin_exec_member_upload(pending, transcript, sig_bytes)) + elif pending["op"] == OP_ROOT_ADD: + self._spawn( + self._admin_exec_root_add(pending, transcript, sig_bytes)) + elif pending["op"] == OP_ROOT_REMOVE: + self._spawn( + self._admin_exec_root_remove(pending, transcript, sig_bytes)) + elif pending["op"] == OP_GROUP_ATTACH: + self._spawn( + self._admin_exec_group_attach(pending, transcript, sig_bytes)) + elif pending["op"] == OP_GROUP_DETACH: + self._spawn( + self._admin_exec_group_detach(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index 74a7c8a..a068a8d 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -12,6 +12,7 @@ Served only on 127.0.0.1 — not exposed to the network. Gated by a per-run session token (11.5.3) — printed at daemon startup. """ +import asyncio import base64 import json import logging @@ -132,12 +133,27 @@ def create_ui_app(state: dict) -> FastAPI: return await _op(lambda: ops.list_groups(state)) @app.post("/api/groups/attach") async def attach_group(payload: dict): - return await _op(lambda: ops.attach_group( + result = await _op(lambda: ops.attach_group( state, (payload.get("name") or "").strip(), (payload.get("shared_dir") or "").strip(), upload_dir=(payload.get("upload_dir") or "").strip(), )) + reload_fn = state.get("reload_fn") + if reload_fn: + asyncio.ensure_future(reload_fn()) + return result + + @app.post("/api/groups/detach") + async def detach_group(payload: dict): + result = await _op(lambda: ops.detach_group( + state, + (payload.get("name") or payload.get("group_id") or "").strip(), + )) + reload_fn = state.get("reload_fn") + if reload_fn: + asyncio.ensure_future(reload_fn()) + return result @app.delete("/api/groups/{group_id}/files/{file_id}") async def delete_file(group_id: str, file_id: str): diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py index faca0ce..62dfe13 100644 --- a/packages/meshbay-node/tests/test_cli_dispatch.py +++ b/packages/meshbay-node/tests/test_cli_dispatch.py @@ -40,6 +40,7 @@ VERBS = [ ["denylist", "show"], ["denylist", "clear", "--yes"], ["reload"], + ["restart-daemon"], ] diff --git a/packages/meshbay-node/tests/test_node_status.py b/packages/meshbay-node/tests/test_node_status.py new file mode 100644 index 0000000..b56eb6e --- /dev/null +++ b/packages/meshbay-node/tests/test_node_status.py @@ -0,0 +1,625 @@ +""" +node_status, root_add, root_remove over MNP. + +These test the D5 node management panel's server-side behaviour: the admin +identity check on node_status, the list_groups operation, and the root +add/remove flows through the MNP handlers. +""" + +import base64 +from dataclasses import asdict +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from conftest import one_root +from meshbay_common.adminop import ( + OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, OP_MEMBER_UNPIN, + admin_transcript, +) +from meshbay_node.transport.quic_server import Denylist +from meshbay_common.crypto import pk_to_b64 +from meshbay_common.join import ROLE_OPERATOR +from meshbay_common.protocol import MNP +from meshbay_node import ops +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import open_roster +from meshbay_node.roots import RootSet +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +GROUP = "g" * 32 + + +@pytest.fixture +async def roster(tmp_path): + r = await open_roster(tmp_path) + yield r + await r.close() + + +def _keypair(): + sk = Ed25519PrivateKey.generate() + return sk, pk_to_b64(sk.public_key()) + + +class _FakeBundleStore: + def __init__(self): + self.stored = [] + + async def store(self, *args): + self.stored.append(args) + + +class _FakeHub: + class _S: + user_id = "node-user" + _session = _S() + + +def _last(session): + return session.sent[-1] if session.sent else {} + + +async def _drain(session): + for coro in session.spawned: + await coro + session.spawned.clear() + + +async def _session( + tmp_path: Path, roster, *, operator: bool, node_user_id: str = "node-user", +) -> WebRTCPeerSession: + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + roots = one_root(shared) + + sk_op, pk_op = _keypair() + if operator: + await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code") + await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli") + + group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index, + "join_policy": "invite"} + denylist = Denylist() + reload_called = [] + async def _reload(): + reload_called.append(True) + + state = { + "groups_ctx": {GROUP: group_ctx}, + "roster": roster, + "indexes": {GROUP: index}, + "bundle_store": _FakeBundleStore(), + "pk_x25519_raw": b"\x02" * 32, + "hub": _FakeHub(), + "node_user_id": node_user_id, + "denylist": denylist, + "reload_fn": _reload, + } + + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": roots, "index": index, "sk_node": index.sk_node, + "roster": roster, "groups": {GROUP: group_ctx}, + "has_admin_authority": operator, + "daemon_state": state, + "node_user_id": node_user_id, + } + session._group_id = GROUP + session._user_id = "grenet" if operator else "mallory" + session._username = session._user_id + session._pk_user = "" + session._uploads = {} + session._admin_ops = {} + session._remote_ip = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + session.state = state + session.sk_op = sk_op + session.denylist = denylist + session.reload_called = reload_called + session.spawned = [] + session._spawn = session.spawned.append + return session + + +async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn, + group_id: str = GROUP): + challenge = _last(session) + assert challenge["type"] == "admin_challenge", challenge + transcript = admin_transcript( + op=op, node_pk_b64=session._node_pk_b64(), group_id=group_id, + subject=subject, nonce=base64.b64decode(challenge["nonce"]), + ts=challenge["ts"]) + pending = session._admin_ops.get(challenge["op_id"]) or { + "op": op, "subject": subject} + await exec_fn(pending, transcript, sk.sign(transcript)) + + +# ── _is_node_admin ────────────────────────────────────────────────────────── + +async def test_is_node_admin_matches_user_id(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + assert session._is_node_admin() + + +async def test_is_node_admin_rejects_different_user(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="someone-else") + assert not session._is_node_admin() + + +async def test_is_node_admin_rejects_missing_node_user_id(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + del session._ctx["node_user_id"] + assert not session._is_node_admin() + + +# ── node_status ───────────────────────────────────────────────────────────── + +async def test_node_status_returns_groups_for_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._spawn(session._do_node_status({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == MNP.NODE_STATUS_ACK + assert len(msg["groups"]) == 1 + assert msg["groups"][0]["id"] == GROUP + + +async def test_node_status_refused_for_non_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False, + node_user_id="grenet") + session._spawn(session._do_node_status({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "error" + assert "operator" in msg["detail"].lower() + + +async def test_node_status_refused_when_user_is_operator_but_ids_mismatch( + tmp_path, roster, +): + """A paired operator who is not the node owner cannot see node_status.""" + session = await _session(tmp_path, roster, operator=True, + node_user_id="someone-else") + session._spawn(session._do_node_status({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "error" + + +async def test_node_status_catches_send_failure(tmp_path, roster): + """If _send itself throws (e.g. msgpack encoding fails), the error must + not silently vanish — it used to, because _send was outside the try block.""" + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + original_send = session._send + sent = [] + call_count = [0] + + def _exploding_send(msg): + call_count[0] += 1 + if msg.get("type") == "node_status_ack": + raise TypeError("msgpack cannot encode this") + sent.append(msg) + + session._send = _exploding_send + session._spawn(session._do_node_status({})) + await _drain(session) + # The try/except around _send should catch the error and send an error reply + assert any(m.get("type") == "error" for m in sent) + + +# ── ops.list_groups ───────────────────────────────────────────────────────── + +async def test_list_groups_returns_group_metadata(tmp_path): + shared = tmp_path / "shared" + shared.mkdir() + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + roots = one_root(shared) + state = { + "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}}, + "peers": {"p1": {"group_id": GROUP}, "p2": {"group_id": GROUP}, + "p3": {"group_id": "other"}}, + "config": None, + } + result = await ops.list_groups(state) + groups = result["groups"] + assert len(groups) == 1 + g = groups[0] + assert g["id"] == GROUP + assert g["has_gek"] is True + assert g["peers"] == 2 + assert isinstance(g["roots"], list) + + +async def test_list_groups_empty(tmp_path): + state = {"groups_ctx": {}, "peers": {}, "config": None} + result = await ops.list_groups(state) + assert result["groups"] == [] + + +# ── ops.add_root ──────────────────────────────────────────────────────────── + +async def test_add_root_creates_directory_and_returns_info(tmp_path): + shared = tmp_path / "shared" + shared.mkdir() + new_dir = tmp_path / "new_root" + + from meshbay_node.config import NodeConfig, GroupConfig, RootSpec + cfg = GroupConfig(id=GROUP, name="test", roots=[ + RootSpec(path=str(shared), name="shared", kind="generic", upload=True), + ]) + + conf = tmp_path / "node.toml" + conf.write_text(f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n' + f' [[groups.roots]]\n path = "{shared}"\n') + node_cfg = NodeConfig.__new__(NodeConfig) + node_cfg.groups = [cfg] + + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + roots = one_root(shared) + state = { + "config": node_cfg, + "config_path": str(conf), + "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}}, + } + + result = await ops.add_root(state, GROUP, str(new_dir)) + assert result["status"] == "added" + assert new_dir.is_dir() + assert len(result["roots"]) == 2 + + +async def test_add_root_rejects_unknown_group(tmp_path): + from meshbay_node.config import NodeConfig + node_cfg = NodeConfig.__new__(NodeConfig) + node_cfg.groups = [] + state = {"config": node_cfg, "groups_ctx": {}} + with pytest.raises(ops.OpError, match="not configured"): + await ops.add_root(state, "nonexistent", "/tmp/nope") + + +# ── ops.remove_root ───────────────────────────────────────────────────────── + +async def test_remove_root_requires_at_least_one_remaining(tmp_path): + shared = tmp_path / "shared" + shared.mkdir() + + from meshbay_node.config import GroupConfig, RootSpec, NodeConfig + cfg = GroupConfig(id=GROUP, name="test", roots=[ + RootSpec(path=str(shared), name="shared", kind="generic", upload=True), + ]) + node_cfg = NodeConfig.__new__(NodeConfig) + node_cfg.groups = [cfg] + + conf = tmp_path / "node.toml" + conf.write_text(f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n' + f' [[groups.roots]]\n path = "{shared}"\n') + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + roots = one_root(shared) + state = { + "config": node_cfg, + "config_path": str(conf), + "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}}, + } + with pytest.raises(ops.OpError, match="only root"): + await ops.remove_root(state, GROUP, "shared") + + +async def test_remove_root_refuses_upload_root(tmp_path): + d1 = tmp_path / "uploads" + d2 = tmp_path / "shared" + d1.mkdir() + d2.mkdir() + + from meshbay_node.config import GroupConfig, RootSpec, NodeConfig + cfg = GroupConfig(id=GROUP, name="test", roots=[ + RootSpec(path=str(d1), name="uploads", kind="generic", upload=True), + RootSpec(path=str(d2), name="shared", kind="generic", upload=False), + ]) + node_cfg = NodeConfig.__new__(NodeConfig) + node_cfg.groups = [cfg] + + conf = tmp_path / "node.toml" + conf.write_text( + f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n' + f' [[groups.roots]]\n path = "{d1}"\n name = "uploads"\n upload = true\n\n' + f' [[groups.roots]]\n path = "{d2}"\n name = "shared"\n') + roots = RootSet.build([asdict(r) for r in cfg.roots]) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + state = { + "config": node_cfg, + "config_path": str(conf), + "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}}, + } + with pytest.raises(ops.OpError, match="upload root"): + await ops.remove_root(state, GROUP, "uploads") + + +async def test_remove_root_succeeds_with_two_roots(tmp_path): + d1 = tmp_path / "dir1" + d2 = tmp_path / "dir2" + d1.mkdir() + d2.mkdir() + + from meshbay_node.config import GroupConfig, RootSpec, NodeConfig + cfg = GroupConfig(id=GROUP, name="test", roots=[ + RootSpec(path=str(d1), name="dir1", kind="generic", upload=True), + RootSpec(path=str(d2), name="dir2", kind="generic", upload=False), + ]) + node_cfg = NodeConfig.__new__(NodeConfig) + node_cfg.groups = [cfg] + + conf = tmp_path / "node.toml" + conf.write_text( + f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n' + f' [[groups.roots]]\n path = "{d1}"\n name = "dir1"\n\n' + f' [[groups.roots]]\n path = "{d2}"\n name = "dir2"\n') + roots = RootSet.build([asdict(r) for r in cfg.roots]) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + state = { + "config": node_cfg, + "config_path": str(conf), + "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}}, + } + + result = await ops.remove_root(state, GROUP, "dir2") + assert result["status"] == "removed" + assert len(cfg.roots) == 1 + assert cfg.roots[0].name == "dir1" + + +# ── root_add MNP handler ─────────────────────────────────────────────────── + +async def test_root_add_issues_challenge(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._do_root_add({"group_id": GROUP, "path": "/tmp/test"}) + msg = _last(session) + assert msg["type"] == "admin_challenge" + + +async def test_root_add_refuses_without_authority(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False) + session._do_root_add({"group_id": GROUP, "path": "/tmp/test"}) + msg = _last(session) + assert msg["type"] == "error" + assert "authorized" in msg["detail"].lower() + + +async def test_root_add_refuses_missing_fields(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._do_root_add({"group_id": GROUP}) + assert _last(session)["type"] == "error" + assert "Missing" in _last(session)["detail"] + + +# ── root_remove MNP handler ──────────────────────────────────────────────── + +async def test_root_remove_issues_challenge(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._do_root_remove({"group_id": GROUP, "root_name": "shared"}) + msg = _last(session) + assert msg["type"] == "admin_challenge" + + +async def test_root_remove_refuses_without_authority(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False) + session._do_root_remove({"group_id": GROUP, "root_name": "shared"}) + msg = _last(session) + assert msg["type"] == "error" + + +async def test_root_remove_refuses_missing_fields(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._do_root_remove({"group_id": GROUP}) + assert _last(session)["type"] == "error" + assert "Missing" in _last(session)["detail"] + + +# ── roster_read MNP handler ────────────────────────────────────────────── + +async def test_roster_read_returns_members_for_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._spawn(session._do_roster_read({"group_id": GROUP})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "roster_read_ack" + assert "members" in msg + assert "identities" in msg + + +async def test_roster_read_refused_for_non_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False, + node_user_id="grenet") + session._spawn(session._do_roster_read({"group_id": GROUP})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "error" + + +async def test_roster_read_filters_ghost_members(tmp_path, roster): + """Members whose identity was deleted (revoked then unpinned) are filtered + out by read_roster — the LEFT JOIN returns them with pk_ed25519 = NULL but + they should never reach the UI.""" + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + sk2, pk2 = _keypair() + await roster.pin_identity("ghost", "ghost", pk2, pk2, "code") + await roster.set_member(GROUP, "ghost", "member", "revoked", "local-cli") + await roster._db.execute("DELETE FROM identities WHERE user_id = 'ghost'") + await roster._db.commit() + # Also add a real member so the roster isn't empty + sk3, pk3 = _keypair() + await roster.pin_identity("real", "real", pk3, pk3, "code") + await roster.set_member(GROUP, "real", "member", "active", "local-cli") + + session._spawn(session._do_roster_read({"group_id": GROUP})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "roster_read_ack" + ghost = [m for m in msg["members"] if m["user_id"] == "ghost"] + assert len(ghost) == 0, "ghost members must be filtered out" + real = [m for m in msg["members"] if m["user_id"] == "real"] + assert len(real) == 1 + + +async def test_unpin_fails_for_ghost_member(tmp_path, roster): + """Unpinning a member with no identity gives a clear error, not a crash.""" + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + await roster.set_member(GROUP, "ghost", "member", "revoked", "local-cli") + # ghost has no identity row + session._do_member_unpin({"user_id": "ghost"}) + challenge = _last(session) + assert challenge["type"] == "admin_challenge" + + await _sign_and_exec(session, OP_MEMBER_UNPIN, "ghost", + session.sk_op, session._admin_exec_member_unpin) + msg = _last(session) + assert msg["type"] == "error" + assert "No such pinned identity" in msg["detail"] + + +async def test_unpin_succeeds_for_real_identity(tmp_path, roster): + """Full unpin flow: challenge → sign → exec → identity deleted.""" + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + sk2, pk2 = _keypair() + await roster.pin_identity("target", "target", pk2, pk2, "code") + await roster.set_member(GROUP, "target", "member", "active", "local-cli") + + session._do_member_unpin({"user_id": "target"}) + challenge = _last(session) + assert challenge["type"] == "admin_challenge" + + await _sign_and_exec(session, OP_MEMBER_UNPIN, "target", + session.sk_op, session._admin_exec_member_unpin) + msg = _last(session) + assert msg["type"] == "member_unpin_ack" + assert msg["user_id"] == "target" + + idents = await roster.list_identities() + assert not any(i["user_id"] == "target" for i in idents) + + +# ── denylist_read MNP handler ──────────────────────────────────────────── + +async def test_denylist_read_returns_entries_for_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session.denylist.deny_user("bad-user") + session._spawn(session._do_denylist_read({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "denylist_read_ack" + assert msg["count"] == 1 + assert "bad-user" in msg["users"] + + +async def test_denylist_read_refused_for_non_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False, + node_user_id="grenet") + session._spawn(session._do_denylist_read({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "error" + + +# ── denylist_clear MNP handler ─────────────────────────────────────────── + +async def test_denylist_clear_removes_entry_for_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session.denylist.deny_user("bad-user") + session.denylist.deny_user("other-user") + session._spawn(session._do_denylist_clear({"subject": "bad-user"})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "denylist_clear_ack" + assert msg["removed"] == 1 + assert "bad-user" not in session.denylist.entries()["users"] + assert "other-user" in session.denylist.entries()["users"] + + +async def test_denylist_clear_all_for_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session.denylist.deny_user("a") + session.denylist.deny_user("b") + session.denylist.deny_jti("j") + session._spawn(session._do_denylist_clear({"subject": ""})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "denylist_clear_ack" + assert msg["removed"] == 3 + + +async def test_denylist_clear_refused_for_non_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False, + node_user_id="grenet") + session._spawn(session._do_denylist_clear({"subject": "bad-user"})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "error" + + +# ── group_attach MNP handler ──────────────────────────────────────────── + +async def test_group_attach_issues_challenge(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._do_group_attach({"name": "test-group", "shared_dir": "/tmp/share"}) + msg = _last(session) + assert msg["type"] == "admin_challenge" + + +async def test_group_attach_refused_without_authority(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False) + session._do_group_attach({"name": "test-group", "shared_dir": "/tmp/share"}) + msg = _last(session) + assert msg["type"] == "error" + assert "authorized" in msg["detail"].lower() + + +async def test_group_attach_refuses_missing_fields(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._do_group_attach({"name": "test-group"}) + msg = _last(session) + assert msg["type"] == "error" + assert "Missing" in msg["detail"] + + +# ── node_reload MNP handler ───────────────────────────────────────────── + +async def test_node_reload_runs_for_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=True, + node_user_id="grenet") + session._spawn(session._do_node_reload({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "node_reload_ack" + assert msg["status"] == "reloaded" + assert len(session.reload_called) == 1 + + +async def test_node_reload_refused_for_non_admin(tmp_path, roster): + session = await _session(tmp_path, roster, operator=False, + node_user_id="grenet") + session._spawn(session._do_node_reload({})) + await _drain(session) + msg = _last(session) + assert msg["type"] == "error" |