From 8872e73e7668bbd39896b19ade4e26c2160c98e6 Mon Sep 17 00:00:00 2001 From: lyuksovannyy Date: Sun, 11 Oct 2026 08:23:19 +0000 Subject: [PATCH] dsh-remote 0.1.0: master/slave remote control for DSH - authenticated WebSocket link (dependency-free RFC6455 + stream mux), pairing codes, per-device keys (hashed), revoke/remove - /r// proxy of the full DSH UI (HTTP, Remote WebSocket, SSE) with per-device cookie minting; slave->slave relay through the master - Devices sidebar panel (all devices' sessions, new session on device), Remote devices settings page, finished-session toasts - plugin + config sync master->slave: registry bundles via pluginManager, link: plugins as tarballs, cordis.patch.yml into a managed bundle layer (!!js preserved), allow/deny, opt-in per-secret credential sync - deploy helpers + e2e probes, unit tests --- .gitignore | 4 + README.md | 129 +++++++ client.js | 396 ++++++++++++++++++++ cordis.patch.yml | 25 ++ deploy/dsh-web.user.service | 15 + deploy/dshr-cli.mjs | 38 ++ deploy/e2e-events.mjs | 20 + deploy/e2e-exc.mjs | 43 +++ deploy/e2e-ui.mjs | 73 ++++ deploy/e2e-ws.mjs | 28 ++ deploy/install-local.mjs | 61 +++ deploy/push-slave.sh | 13 + index.js | 727 ++++++++++++++++++++++++++++++++++++ lib/auth-cookie.mjs | 58 +++ lib/mux.mjs | 245 ++++++++++++ lib/proxy.mjs | 199 ++++++++++ lib/store.mjs | 150 ++++++++ lib/sync.mjs | 308 +++++++++++++++ lib/ws.mjs | 215 +++++++++++ package.json | 27 ++ test/link.test.mjs | 140 +++++++ 21 files changed, 2914 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 client.js create mode 100644 cordis.patch.yml create mode 100644 deploy/dsh-web.user.service create mode 100644 deploy/dshr-cli.mjs create mode 100644 deploy/e2e-events.mjs create mode 100644 deploy/e2e-exc.mjs create mode 100644 deploy/e2e-ui.mjs create mode 100644 deploy/e2e-ws.mjs create mode 100644 deploy/install-local.mjs create mode 100755 deploy/push-slave.sh create mode 100644 index.js create mode 100644 lib/auth-cookie.mjs create mode 100644 lib/mux.mjs create mode 100644 lib/proxy.mjs create mode 100644 lib/store.mjs create mode 100644 lib/sync.mjs create mode 100644 lib/ws.mjs create mode 100644 package.json create mode 100644 test/link.test.mjs diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f9d077c --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +node_modules/ +*.log +*.tgz +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..b3e6974 --- /dev/null +++ b/README.md @@ -0,0 +1,129 @@ +# dsh-remote + +Remote control for [DeepSeek Harness](https://github.com/deepseek-ai/deepseek-harness), in the style of Claude Desktop: one **master** and any number of **slaves**. + +- **Every session on every device.** Sessions run on the machine that created them. From any device you can browse all of them, open them, and chat in them; the work still executes on the owning machine. +- **The full remote UI at `/r//`.** The whole DSH UI of another device is proxied over the link: HTTP, the Remote WebSocket, SSE, plugin bundles and terminals. Slave→slave goes through the master's relay. +- **Plugin and config sync, master → slaves.** Slaves get the master's npm/registry bundles, its local `link:` plugins (packed and shipped over the link) and its `cordis.patch.yml`. The config lands in a *managed* bundle layer, so each slave's own `cordis.patch.yml` still wins and slave-only plugins stay untouched. +- **Opt-in secrets.** A credential (`OMNI_API_KEY`, …) reaches a slave only if three places agree: the master offers it, the master enables it for that device, and the slave accepts it. +- **Devices UI.** A sidebar panel lists every device and its sessions (live running state, filter, "New session on "). A "Remote devices" settings page handles pairing, revocation and per-device sync policy. You get a toast when a session on any device finishes. + +No dependencies: the WebSocket, the multiplexer and the proxy are plain Node. + +``` + slave (laptop) ──outbound WebSocket /dsh-remote/link──► master ◄── slave (ai-pc) + own sessions + own local plugins plugins, config, relay hub +``` + +## Install + +On every machine, register the plugin in the DSH web profile and restart DSH: + +```bash +# master (this repo as a link: plugin, e.g. in Beast-Workspace) +# profile package.json → "@local/dsh-remote": "link:/path/to/dsh-remote" + bundle entry +# or use the helper (also creates the node_modules link and a config block): +node deploy/install-local.mjs --role master --name "Pelican-Node" +node deploy/install-local.mjs --role auto --name "ai-pc" # on a slave +systemctl --user restart dsh # or restart `dsh web` however you run it +``` + +`deploy/dsh-web.user.service` is a ready-made systemd user unit for a rootless `dsh web` (127.0.0.1:3081). + +## Pair a device + +1. **Master:** Settings → *Remote devices* → **Create pairing code** (single use, valid 15 min). +2. **Slave:** Settings → *Remote devices* → enter the master URL (e.g. `http://192.168.0.111:3080` or `https://dsh.rizzy.dev`) and the code → **Pair**. Then restart the slave's DSH once: with `role: auto` it switches from master to slave. + +The slave gets its own device key; the master stores only its hash. **Revoke** cuts the device off immediately: open links close with code 4401 and later connections are refused with 401. + +## Use it + +- **Sidebar → Devices:** every device, its sessions grouped by workspace, a running indicator, search. Click a session to open it. A local session opens in place. A remote one opens `/r//#/session/`, the full UI of that device, with a "Remote: · back to local" pill in the corner. +- **+ New session** on a device card creates a session in one of that device's workspaces and opens it. +- **Open DSH ↗** opens a device's whole UI. +- **Toasts:** ": '' finished", with an *Open* action. + +## Configuration (`cordis.patch.yml`) + +```yaml +- id: dsh-remote + config: + role: master # master | slave | auto (auto = slave once paired) + deviceName: Pelican-Node + lanUrl: http://192.168.0.111:3080 # shown on the pairing page + publicUrl: https://dsh.rizzy.dev + sync: + auto: true # slave: sync on link-up and when the master pushes + allow: [] # slave: only these bundles (exact or prefix*) + deny: [] # slave: never these bundles (wins over allow) + secrets: [] # slave: credential refs it accepts from the master + offerSecrets: [OMNI_API_KEY] # master: refs it may hand out (per-device opt-in on the Devices page) + excludeBundles: [] # master: never sync these bundles + excludePatchIds: [] # master: never sync these cordis.patch.yml entry ids +``` + +Never synced: the master's `dsh-remote` and `dsh-lanmode` entries, and `connection`/`webserver` settings, since these are machine-local. + +## How sync works + +| Master has | Slave gets | +| --- | --- | +| registry bundle `name@ver` | `pluginManager.installBundle(name@ver)` (pnpm in the slave profile) | +| `link:` local plugin | a tarball over the link, unpacked to `<profile>/dsh-remote/sync/plugins/<name>/` and linked into the profile. Its `@deepseek-ai/*` imports are re-linked to the slave's DSH install | +| `cordis.patch.yml` entries | `<profile>/dsh-remote/managed-bundle/cordis.patch.yml`, added as the **last bundle**. The slave's own `cordis.patch.yml` is applied after it and wins. `!!js` expressions are kept | +| offered secret (opted in on both sides) | `ctx.credentials.set(name, value)` on the slave | + +Local plugins are compared by content hash and re-shipped only when they change. When the bundle list changes, DSH needs a restart to load host code; the sync result reports `restartRequired`. *Push sync to all devices* on the master triggers an immediate pull everywhere. + +## Security model + +- Browsers authenticate with DSH's own cookie. Every `/dsh-remote/*` and `/r/*` route checks `ctx.connection.requestRejection`. +- The link is authenticated with a per-device key, sent in the upgrade headers and stored as a SHA-256 hash on the master. +- The proxy strips the browser's DSH cookies. The **owning** device mints a short-lived cookie for its own loopback authority, signed with its own browser-session secret, and rewrites `Host`/`Origin`. A remote DSH therefore only trusts its own minted cookie, and credentials never leave the machine they belong to, except secrets you explicitly opted into. +- `/dsh-remote/enroll` is the only unauthenticated route. It accepts a single-use pairing code. +- Use TLS (`https://` master URL) whenever the link crosses an untrusted network. + +## Public access via rizzy.dev + +`dsh.rizzy.dev` is served by the edge Traefik behind **Authentik forward-auth**. Browsers log in through Authentik as usual. A slave's DSH can't do an interactive login, so the edge must let these two paths through without forward-auth (they carry their own device-key / pairing-code authentication): + +``` +/dsh-remote/link (WebSocket upgrade) +/dsh-remote/enroll (POST, one-time pairing code) +``` + +For example, add a second Traefik router with a higher priority, `Host(\`dsh.rizzy.dev\`) && (Path(\`/dsh-remote/link\`) || Path(\`/dsh-remote/enroll\`))`, that uses the same service but no Authentik middleware. Until that exists, pair and link slaves over the LAN (`http://192.168.0.111:3080`). + +## HTTP API (used by the UI) + +| Route | | +| --- | --- | +| `GET /dsh-remote/state` | role, self, link status, devices | +| `GET /dsh-remote/sessions` | sessions of every reachable device | +| `GET /dsh-remote/stream` | SSE: `session-finished`, `devices`, `sync` events | +| `POST /dsh-remote/pairing` | master: create a pairing code | +| `POST /dsh-remote/pair` | slave: `{masterUrl, code}` | +| `POST /dsh-remote/unpair` | slave: forget the master | +| `POST /dsh-remote/devices/<id>/{revoke,remove,sync}` | master: device management / sync policy | +| `POST /dsh-remote/sync` | slave: pull now · master: push to all | +| `GET /dsh-remote/sync/status` | manifest (master) / last result (slave) | +| `POST /dsh-remote/new-session` | `{device, workspaceId? , cwd?}` | +| `* /r/<device>/…` | that device's DSH | + +## Development + +```bash +npm test # link, mux, proxy (HTTP + WS), cookie, allow/deny +node deploy/dshr-cli.mjs GET state # call the local API as the DSH owner (loopback) +node deploy/e2e-ws.mjs ai-pc # Remote WebSocket through /r/ai-pc/ +node deploy/e2e-events.mjs 30 # watch the SSE event stream +node deploy/e2e-ui.mjs <chrome> /tmp/out # headless UI check + screenshots +deploy/push-slave.sh agent@192.168.0.110 # copy to a slave + restart its dsh +``` + +## Known limitations + +- **`/r/<device>/` needs relative URLs.** Plugins that call absolute URLs (`fetch('/x')`) miss the mount there. Currently that is `@liustack/modlens` (paste/config) and parts of `@goodandready/dsh-lanmode`'s own UI. Core DSH is fully relative. +- **Restart after plugin changes.** A sync that adds or removes plugins needs a DSH restart on the slave to load their host code, as the result reports. Config-only changes apply after a restart too: DSH's HMR doesn't watch bundle patch files. +- **Multiple devices on one origin.** Viewing several devices in one browser happens under one origin, so `localStorage`-based UI state (e.g. layout choices) is shared between them. diff --git a/client.js b/client.js new file mode 100644 index 0000000..0d9c5d3 --- /dev/null +++ b/client.js @@ -0,0 +1,396 @@ +/** + * @local/dsh-remote — browser side. + * + * - Sidebar panel "Devices" (sidebar.panellist + keyed main slot): every linked device + * with its sessions grouped by workspace, live running state, "open" (local session → + * in place, remote → /r/<device>/#/session/<id>), "new session on device". + * - Settings section "Remote devices": role, link status, pairing (master: create code; + * slave: enter master URL + code), device list with revoke/remove and per-device sync + * policy (enabled, allow/deny, opt-in secrets), "sync now". + * - Toast when a session on any device finishes (server-sent events). + * - When the page itself is served under /r/<device>/, a small banner shows which + * device you are looking at, with a link back to the local DSH. + */ +window.__ModuleLoader__.load({ + id: '@local/dsh-remote', + factory(require) { + const React = require('react'); + const ReactDOM = require('react-dom'); + const h = React.createElement; + const { useState, useEffect, useCallback, useMemo, useRef } = React; + let P = {}; + try { P = require('@deepseek-ai/dsh-client-ui-primitives') || {}; } catch { P = {}; } + + const PANEL_ID = 'dsh-remote-devices'; + + // When this UI is itself proxied (/r/<dev>/), API calls must go to *this* page's + // host DSH, i.e. the remote device. Relative URLs already do that. For the device + // overview we always want the browser's own DSH, which is the URL root before /r/. + const mount = (() => { + const m = location.pathname.match(/^(.*?)\/r\/([a-z0-9-]+)\//); + return m ? { root: m[1] + '/', device: m[2] } : { root: location.pathname.replace(/[^/]*$/, ''), device: null }; + })(); + const rootUrl = (p) => mount.root + p; + + async function api(path, init) { + const res = await fetch(rootUrl('dsh-remote/' + path), { + ...init, + headers: { accept: 'application/json', ...(init && init.body ? { 'content-type': 'application/json' } : {}) }, + credentials: 'same-origin', + }); + let json; + try { json = await res.json(); } catch { json = {}; } + if (!res.ok) throw new Error(json.error || res.statusText || String(res.status)); + return json; + } + const post = (path, body) => api(path, { method: 'POST', body: JSON.stringify(body || {}) }); + + function sessionUrl(device, selfId, sessionId) { + const base = device === selfId ? mount.root : `${mount.root}r/${device}/`; + return `${base}#/session/${encodeURIComponent(sessionId)}`; + } + + const ago = (t) => { + if (!t) return ''; + const s = Math.max(0, (Date.now() - t) / 1000); + if (s < 60) return 'just now'; + if (s < 3600) return `${Math.floor(s / 60)}m ago`; + if (s < 86400) return `${Math.floor(s / 3600)}h ago`; + return `${Math.floor(s / 86400)}d ago`; + }; + const base = (p) => String(p || '').replace(/\/+$/, '').split('/').pop() || p || ''; + + // ------------------------------------------------------------------ styles + const S = { + page: { padding: '20px 28px', overflow: 'auto', height: '100%', boxSizing: 'border-box', color: 'var(--color-text, inherit)' }, + h1: { fontSize: 20, fontWeight: 600, margin: '0 0 4px' }, + sub: { opacity: 0.65, fontSize: 13, margin: '0 0 18px' }, + card: { border: '1px solid var(--color-border, rgba(127,127,127,.25))', borderRadius: 10, padding: '12px 14px', margin: '0 0 14px', background: 'var(--color-surface, transparent)' }, + devHead: { display: 'flex', alignItems: 'center', gap: 10, marginBottom: 8 }, + dot: (on) => ({ width: 9, height: 9, borderRadius: 9, background: on ? '#3fb950' : '#8b949e', flex: 'none' }), + badge: (tone) => ({ fontSize: 11, padding: '1px 7px', borderRadius: 9, border: '1px solid currentColor', opacity: 0.8, color: tone === 'accent' ? '#58a6ff' : tone === 'warn' ? '#d29922' : tone === 'ok' ? '#3fb950' : 'inherit' }), + ws: { fontSize: 12, textTransform: 'uppercase', letterSpacing: '.04em', opacity: 0.6, margin: '10px 0 4px' }, + row: { display: 'flex', alignItems: 'center', gap: 8, padding: '5px 8px', borderRadius: 6, cursor: 'pointer', fontSize: 14 }, + title: { flex: 1, minWidth: 0, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }, + meta: { fontSize: 12, opacity: 0.55, flex: 'none' }, + btn: { font: 'inherit', fontSize: 13, padding: '4px 10px', borderRadius: 6, border: '1px solid var(--color-border, rgba(127,127,127,.35))', background: 'transparent', color: 'inherit', cursor: 'pointer' }, + btnPrimary: { font: 'inherit', fontSize: 13, padding: '5px 12px', borderRadius: 6, border: '1px solid #2f81f7', background: '#2f81f7', color: '#fff', cursor: 'pointer' }, + input: { font: 'inherit', fontSize: 13, padding: '5px 8px', borderRadius: 6, border: '1px solid var(--color-border, rgba(127,127,127,.35))', background: 'transparent', color: 'inherit', minWidth: 0 }, + err: { color: '#f85149', fontSize: 13 }, + code: { fontFamily: 'ui-monospace, monospace', fontSize: 18, letterSpacing: '.08em', padding: '6px 10px', border: '1px dashed currentColor', borderRadius: 6, display: 'inline-block' }, + spin: { display: 'inline-block', width: 8, height: 8, borderRadius: 8, background: '#d29922', animation: 'dshr-pulse 1s infinite' }, + }; + if (!document.getElementById('dshr-style')) { + const st = document.createElement('style'); + st.id = 'dshr-style'; + st.textContent = '@keyframes dshr-pulse{0%,100%{opacity:.3}50%{opacity:1}} .dshr-row:hover{background:var(--color-hover,rgba(127,127,127,.12))}'; + document.head.appendChild(st); + } + + // ------------------------------------------------------------------ live store + const store = { + state: null, sessions: null, error: null, listeners: new Set(), events: new Set(), + emit() { for (const l of this.listeners) l(); }, + async refresh() { + try { + const [state, sessions] = await Promise.all([api('state'), api('sessions')]); + this.state = state; this.sessions = sessions.devices; this.error = null; + } catch (e) { this.error = e.message; } + this.emit(); + }, + }; + let es = null; + let refreshTimer = null; + function startStream() { + if (es) return; + try { + es = new EventSource(rootUrl('dsh-remote/stream')); + es.onmessage = (m) => { + let ev; try { ev = JSON.parse(m.data); } catch { return; } + for (const l of store.events) l(ev); + clearTimeout(refreshTimer); + refreshTimer = setTimeout(() => store.refresh(), 300); + }; + es.onerror = () => { /* EventSource reconnects on its own */ }; + } catch { es = null; } + } + + function useStore() { + const [, force] = useState(0); + useEffect(() => { + const l = () => force((x) => x + 1); + store.listeners.add(l); + if (!store.state) store.refresh(); + const t = setInterval(() => store.refresh(), 10000); + return () => { store.listeners.delete(l); clearInterval(t); }; + }, []); + return store; + } + + // ------------------------------------------------------------------ Devices panel + function DeviceIcon({ size }) { + return h('svg', { 'data-dsh-panel-entry': PANEL_ID, viewBox: '0 0 16 16', width: size, height: size, fill: 'none', stroke: 'currentColor', strokeWidth: 1.3, strokeLinecap: 'round', strokeLinejoin: 'round', 'aria-hidden': true }, + h('rect', { x: 1.5, y: 2.5, width: 9, height: 6.5, rx: 1 }), + h('path', { d: 'M4 12h4M6 9v3' }), + h('rect', { x: 11.5, y: 5.5, width: 3, height: 8, rx: .8 })); + } + + function SessionRow({ dev, s, selfId }) { + return h('a', { + className: 'dshr-row', style: { ...S.row, color: 'inherit', textDecoration: 'none' }, + href: sessionUrl(dev.device, selfId, s.sessionId), + title: `${s.title || s.sessionId}\n${s.cwd || ''}`, + }, + s.running ? h('span', { style: S.spin, title: 'running' }) : h('span', { style: { width: 8, flex: 'none' } }), + h('span', { style: S.title }, s.title || h('i', { style: { opacity: .6 } }, 'untitled')), + h('span', { style: S.meta }, ago(s.updatedAt))); + } + + function NewSession({ dev, selfId }) { + const [open, setOpen] = useState(false); + const [busy, setBusy] = useState(false); + const [err, setErr] = useState(''); + const options = dev.workspaces || []; + const [ws, setWs] = useState(options[0]?.id || ''); + const go = async () => { + setBusy(true); setErr(''); + try { + const r = await post('new-session', { device: dev.device, workspaceId: ws || undefined }); + location.href = sessionUrl(dev.device, selfId, r.sessionId); + if (dev.device === selfId) setTimeout(() => location.reload(), 50); + } catch (e) { setErr(e.message); setBusy(false); } + }; + if (!open) return h('button', { style: S.btn, onClick: () => setOpen(true) }, '+ New session'); + return h('span', { style: { display: 'inline-flex', gap: 6, alignItems: 'center', flexWrap: 'wrap' } }, + h('select', { style: S.input, value: ws, onChange: (e) => setWs(e.target.value) }, + options.length ? options.map((w) => h('option', { key: w.id, value: w.id }, w.title || base(w.path))) : h('option', { value: '' }, '(no workspace)')), + h('button', { style: S.btnPrimary, disabled: busy || !options.length, onClick: go }, busy ? 'Creating…' : 'Create'), + h('button', { style: S.btn, onClick: () => setOpen(false) }, 'Cancel'), + err && h('span', { style: S.err }, err)); + } + + function DevicesPage() { + const st = useStore(); + const [filter, setFilter] = useState(''); + const self = st.state?.self; + if (!st.state && !st.error) return h('div', { style: S.page }, 'Loading devices…'); + const devices = st.sessions || []; + const q = filter.trim().toLowerCase(); + const meta = Object.fromEntries((st.state?.devices || []).map((d) => [d.id, d])); + return h('div', { style: S.page, 'data-dsh-plugin': 'dsh-remote' }, + h('h1', { style: S.h1 }, 'Devices'), + h('p', { style: S.sub }, + st.state ? `This device: ${self?.name} (${st.state.role})` : '', + st.state?.role === 'slave' && st.state.master + ? (st.state.master.linked ? ` · linked to ${st.state.master.info?.name || st.state.master.url}` : ` · master unreachable${st.state.master.error ? ': ' + st.state.master.error : ''}`) + : ''), + st.error && h('p', { style: S.err }, st.error), + h('div', { style: { display: 'flex', gap: 8, margin: '0 0 14px' } }, + h('input', { style: { ...S.input, flex: 1 }, placeholder: 'Filter sessions…', value: filter, onChange: (e) => setFilter(e.target.value) }), + h('button', { style: S.btn, onClick: () => store.refresh() }, 'Refresh')), + devices.map((dev) => { + const m = meta[dev.device] || {}; + const items = (dev.items || []).filter((s) => !q || `${s.title || ''} ${s.cwd || ''}`.toLowerCase().includes(q)); + const byWs = new Map(); + for (const s of items) { + const key = s.cwd || ''; + if (!byWs.has(key)) byWs.set(key, []); + byWs.get(key).push(s); + } + const wsTitle = (cwd) => (dev.workspaces || []).find((w) => w.path === cwd)?.title || base(cwd) || 'No workspace'; + return h('div', { key: dev.device, style: S.card }, + h('div', { style: S.devHead }, + h('span', { style: S.dot(!dev.error) }), + h('strong', null, dev.name || dev.device), + dev.self && h('span', { style: S.badge('accent') }, 'this device'), + m.role === 'master' && h('span', { style: S.badge() }, 'master'), + m.via && h('span', { style: S.badge() }, 'via master'), + h('span', { style: { flex: 1 } }), + !dev.self && h('a', { style: { ...S.btn, textDecoration: 'none' }, href: `${mount.root}r/${dev.device}/` }, 'Open DSH ↗'), + h(NewSession, { dev, selfId: self?.id })), + dev.error && h('div', { style: S.err }, dev.error), + !dev.error && !items.length && h('div', { style: { opacity: .6, fontSize: 13 } }, q ? 'No matching sessions.' : 'No sessions yet.'), + [...byWs.entries()].map(([cwd, list]) => h('div', { key: cwd }, + h('div', { style: S.ws, title: cwd }, wsTitle(cwd)), + list.map((s) => h(SessionRow, { key: s.sessionId, dev, s, selfId: self?.id }))))); + })); + } + + // ------------------------------------------------------------------ Settings section + function Field({ label, children, hint }) { + return h('label', { style: { display: 'grid', gap: 4, margin: '0 0 10px', fontSize: 13 } }, + h('span', { style: { opacity: .8 } }, label), children, hint && h('span', { style: { opacity: .55, fontSize: 12 } }, hint)); + } + const listToText = (l) => (l || []).join(', '); + const textToList = (t) => String(t || '').split(/[\s,]+/).map((x) => x.trim()).filter(Boolean); + + function DeviceSync({ d, offer, onSaved }) { + const [enabled, setEnabled] = useState(d.sync?.enabled !== false); + const [allow, setAllow] = useState(listToText(d.sync?.allow)); + const [deny, setDeny] = useState(listToText(d.sync?.deny)); + const [secrets, setSecrets] = useState(new Set(d.sync?.secrets || [])); + const [msg, setMsg] = useState(''); + const save = async () => { + try { + await post(`devices/${d.id}/sync`, { enabled, allow: textToList(allow), deny: textToList(deny), secrets: [...secrets] }); + setMsg('Saved; device notified.'); onSaved(); + } catch (e) { setMsg(e.message); } + }; + return h('div', { style: { margin: '8px 0 0 18px', paddingLeft: 10, borderLeft: '2px solid rgba(127,127,127,.25)' } }, + h('label', { style: { display: 'flex', gap: 6, alignItems: 'center', fontSize: 13, marginBottom: 8 } }, + h('input', { type: 'checkbox', checked: enabled, onChange: (e) => setEnabled(e.target.checked) }), 'Sync plugins + config to this device'), + h(Field, { label: 'Only these bundles (empty = all)', hint: 'Exact names or prefix*, comma separated' }, + h('input', { style: S.input, value: allow, onChange: (e) => setAllow(e.target.value) })), + h(Field, { label: 'Never these bundles' }, h('input', { style: S.input, value: deny, onChange: (e) => setDeny(e.target.value) })), + h('div', { style: { fontSize: 13, margin: '0 0 8px' } }, + h('div', { style: { opacity: .8, marginBottom: 4 } }, 'Secrets this device may pull (it must also opt in locally)'), + offer.length ? offer.map((n) => h('label', { key: n, style: { display: 'inline-flex', gap: 4, marginRight: 12 } }, + h('input', { type: 'checkbox', checked: secrets.has(n), onChange: (e) => { const s = new Set(secrets); e.target.checked ? s.add(n) : s.delete(n); setSecrets(s); } }), + h('code', null, n))) + : h('span', { style: { opacity: .55 } }, 'None offered. Add names to sync.offerSecrets in the dsh-remote config.')), + h('button', { style: S.btn, onClick: save }, 'Save sync policy'), ' ', msg && h('span', { style: { fontSize: 12, opacity: .75 } }, msg)); + } + + function SettingsPanel() { + const st = useStore(); + const [code, setCode] = useState(null); + const [masterUrl, setMasterUrl] = useState(''); + const [pairCode, setPairCode] = useState(''); + const [msg, setMsg] = useState(''); + const [busy, setBusy] = useState(false); + const [open, setOpen] = useState({}); + const [syncRes, setSyncRes] = useState(null); + const s = st.state; + if (!s) return h('div', { style: { padding: 4 } }, st.error ? h('span', { style: S.err }, st.error) : 'Loading…'); + const act = async (fn) => { setBusy(true); setMsg(''); try { await fn(); } catch (e) { setMsg(e.message); } setBusy(false); store.refresh(); }; + + const header = h('div', { style: S.card }, + h('div', { style: S.devHead }, h('strong', null, s.self.name), h('span', { style: S.badge('accent') }, s.role), h('code', { style: { opacity: .6, fontSize: 12 } }, s.self.id)), + s.role === 'master' && h('div', { style: { fontSize: 13, opacity: .8 } }, + 'Slaves pair with this URL: ', h('code', null, s.publicUrl || s.lanUrl || location.origin + mount.root.replace(/\/$/, '')), + s.publicUrl && s.lanUrl ? h('span', null, ' (LAN: ', h('code', null, s.lanUrl), ')') : null), + s.role === 'slave' && h('div', { style: { fontSize: 13 } }, + h('span', { style: S.dot(s.master?.linked) }), ' ', + s.master?.linked ? `Linked to ${s.master.info?.name} · ${s.master.url}` : `Not linked (${s.master?.url || 'no master'})${s.master?.error ? ': ' + s.master.error : ''}`)); + + const pairing = s.role === 'master' + ? h('div', { style: S.card }, + h('strong', null, 'Pair a new device'), + h('p', { style: { fontSize: 13, opacity: .75, margin: '4px 0 10px' } }, 'Create a one-time code (valid 15 minutes), then enter it in Settings → Remote devices on the new machine.'), + code ? h('div', null, h('span', { style: S.code }, code.code), h('div', { style: { fontSize: 12, opacity: .6, marginTop: 4 } }, `expires ${new Date(code.expiresAt).toLocaleTimeString()}`)) + : h('button', { style: S.btnPrimary, disabled: busy, onClick: () => act(async () => setCode(await post('pairing', {}))) }, 'Create pairing code')) + : h('div', { style: S.card }, + h('strong', null, s.master?.url ? 'Re-pair with a master' : 'Pair with a master'), + h(Field, { label: 'Master URL', hint: 'e.g. https://dsh.rizzy.dev or http://192.168.0.111:3080' }, + h('input', { style: S.input, value: masterUrl, placeholder: 'https://…', onChange: (e) => setMasterUrl(e.target.value) })), + h(Field, { label: 'Pairing code' }, h('input', { style: { ...S.input, fontFamily: 'ui-monospace,monospace' }, value: pairCode, placeholder: 'ABCD-EFGH-…', onChange: (e) => setPairCode(e.target.value.toUpperCase()) })), + h('button', { style: S.btnPrimary, disabled: busy || !masterUrl || !pairCode, onClick: () => act(async () => { + const r = await post('pair', { masterUrl, code: pairCode }); + setMsg(r.restartRequired ? 'Paired. Restart DSH on this machine to switch it to slave mode.' : `Paired with ${r.master?.name}.`); + }) }, 'Pair'), + s.master?.url && h('span', null, ' ', h('button', { style: S.btn, disabled: busy, onClick: () => act(() => post('unpair', {})) }, 'Unpair'))); + + const devices = s.role === 'master' && h('div', { style: S.card }, + h('strong', null, 'Paired devices'), + (s.devices || []).filter((d) => !d.self).length === 0 && h('div', { style: { fontSize: 13, opacity: .6, marginTop: 6 } }, 'No devices yet.'), + (s.devices || []).filter((d) => !d.self).map((d) => h('div', { key: d.id, style: { padding: '8px 0', borderTop: '1px solid rgba(127,127,127,.15)' } }, + h('div', { style: { display: 'flex', alignItems: 'center', gap: 8, fontSize: 14 } }, + h('span', { style: S.dot(d.online) }), h('strong', null, d.name), h('code', { style: { fontSize: 12, opacity: .6 } }, d.id), + d.revoked && h('span', { style: S.badge('warn') }, 'revoked'), + h('span', { style: { fontSize: 12, opacity: .6 } }, d.online ? `online since ${ago(d.connectedAt)}` : d.lastSeenAt ? `seen ${ago(d.lastSeenAt)}` : 'never connected'), + h('span', { style: { flex: 1 } }), + h('button', { style: S.btn, onClick: () => setOpen({ ...open, [d.id]: !open[d.id] }) }, open[d.id] ? 'Hide sync' : 'Sync policy'), + !d.revoked && h('button', { style: S.btn, disabled: busy, onClick: () => confirm(`Revoke ${d.name}? It will be disconnected and must re-pair.`) && act(() => post(`devices/${d.id}/revoke`, {})) }, 'Revoke'), + h('button', { style: S.btn, disabled: busy, onClick: () => confirm(`Remove ${d.name} completely?`) && act(() => post(`devices/${d.id}/remove`, {})) }, 'Remove')), + open[d.id] && h(DeviceSync, { d, offer: s.sync?.offerSecrets || [], onSaved: () => store.refresh() })))); + + const syncCard = h('div', { style: S.card }, + h('strong', null, 'Plugin + config sync'), + s.role === 'master' + ? h('p', { style: { fontSize: 13, opacity: .75, margin: '4px 0 8px' } }, 'Slaves pull this machine\'s bundles (registry + local link: plugins) and its cordis.patch.yml (minus machine-local entries) into a managed layer that their own config overrides.') + : h('div', { style: { fontSize: 13, margin: '4px 0 8px' } }, + h('div', null, `Auto sync: ${s.sync?.auto ? 'on' : 'off'} · allow: ${listToText(s.sync?.allow) || 'all'} · deny: ${listToText(s.sync?.deny) || 'none'}`), + h('div', null, 'Accepted secrets: ', listToText(s.sync?.secrets) || h('span', { style: { opacity: .6 } }, 'none (set sync.secrets in this device\'s dsh-remote config)')), + s.lastSync && h('div', { style: { marginTop: 6, opacity: .8 } }, + `Last sync ${ago(s.lastSync.at)}: ${s.lastSync.installed?.length || 0} installed, ${s.lastSync.updated?.length || 0} updated, ${s.lastSync.failed?.length || 0} failed, ${s.lastSync.patchEntries || 0} config entries`, + (s.lastSync.failed || []).map((f, i) => h('div', { key: i, style: S.err }, f)))), + h('button', { style: S.btn, disabled: busy, onClick: () => act(async () => { + const r = await post('sync', {}); + setSyncRes(r); + if (r.restartRequired) setMsg('Synced. Some plugins changed: restart DSH on this device to load them.'); + }) }, s.role === 'master' ? 'Push sync to all devices' : 'Sync now'), + syncRes && s.role === 'master' && h('span', { style: { fontSize: 12, opacity: .7 } }, ` notified ${syncRes.pushed} device(s)`)); + + return h('div', { 'data-dsh-plugin': 'dsh-remote' }, header, pairing, devices, syncCard, msg && h('div', { style: { ...S.card, fontSize: 13 } }, msg)); + } + + // ------------------------------------------------------------------ toasts + function ToastHost() { + const [items, setItems] = useState([]); + useEffect(() => { + const l = (ev) => { + if (ev.type !== 'session-finished') return; + const id = `${ev.device}:${ev.sessionId}:${ev.at}`; + setItems((xs) => [...xs.slice(-2), { id, ev }]); + }; + store.events.add(l); + return () => store.events.delete(l); + }, []); + return h(React.Fragment, null, items.map(({ id, ev }) => { + const text = `${ev.deviceName}: “${ev.title || 'session'}” finished`; + const done = () => setItems((xs) => xs.filter((x) => x.id !== id)); + const actions = [{ label: 'Open', prefix: ' · ', onClick: () => { location.href = sessionUrl(ev.device, store.state?.self?.id, ev.sessionId); done(); } }]; + if (P.Toast) return h(P.Toast, { key: id, text, tone: 'success', holdMs: 6000, actions, onDone: done }); + return h('div', { key: id, onClick: actions[0].onClick, style: { position: 'fixed', top: 16, left: '50%', transform: 'translateX(-50%)', zIndex: 9999, background: '#1f6feb', color: '#fff', padding: '8px 14px', borderRadius: 8, cursor: 'pointer', fontSize: 14 } }, text); + })); + } + + function mountToasts() { + const el = document.createElement('div'); + el.id = 'dshr-toasts'; + document.body.appendChild(el); + const root = ReactDOM.createRoot ? ReactDOM.createRoot(el) : null; + if (root) root.render(h(ToastHost)); + else ReactDOM.render(h(ToastHost), el); + return () => { try { root ? root.unmount() : ReactDOM.unmountComponentAtNode(el); } catch { /* ignore */ } el.remove(); }; + } + + // Banner when this whole UI is a remote device seen through /r/<device>/. + function mountRemoteBanner() { + if (!mount.device) return () => {}; + const el = document.createElement('a'); + el.href = mount.root; + el.textContent = `● Remote: ${mount.device} · back to local`; + el.title = 'You are using the DSH of another device through dsh-remote. Sessions run there.'; + Object.assign(el.style, { position: 'fixed', bottom: '10px', right: '12px', zIndex: 9998, background: 'rgba(31,111,235,.92)', color: '#fff', padding: '4px 10px', borderRadius: '12px', fontSize: '12px', textDecoration: 'none', fontFamily: 'system-ui, sans-serif' }); + document.body.appendChild(el); + return () => el.remove(); + } + + // ------------------------------------------------------------------ plugin + return { + inject: ['slots'], + apply(ctx) { + // Toasts / banner only belong to the outermost (local) UI, not proxied copies. + startStream(); + const disposers = []; + if (!mount.device) disposers.push(mountToasts()); + disposers.push(mountRemoteBanner()); + + disposers.push(ctx.slots.inject('sidebar.panellist', () => ctx.slots.register({ + name: 'sidebar.panellist', id: PANEL_ID, order: 50, label: 'Devices', + }, DeviceIcon))); + disposers.push(ctx.slots.inject('main', () => ctx.slots.register({ name: 'main', key: PANEL_ID }, DevicesPage))); + disposers.push(ctx.slots.inject('settings.section', () => ctx.slots.register({ + name: 'settings.section', id: 'dsh-remote', order: 160, label: 'Remote devices', + }, SettingsPanel))); + + ctx.effect(() => () => { + for (const d of disposers.splice(0)) { try { d && d(); } catch { /* ignore */ } } + try { es && es.close(); } catch { /* ignore */ } + es = null; + }, 'dsh-remote: ui'); + }, + }; + }, +}); diff --git a/cordis.patch.yml b/cordis.patch.yml new file mode 100644 index 0000000..b198248 --- /dev/null +++ b/cordis.patch.yml @@ -0,0 +1,25 @@ +- insert: + - id: dsh-remote + name: '@local/dsh-remote' + config: + # master | slave | auto (auto = slave once paired with a master, else master) + role: auto + # Display name of this device (defaults to the hostname); its id is the slug of it. + deviceName: '' + # Shown in the Devices page so other machines know where to pair. + lanUrl: '' + publicUrl: '' + sync: + # slave: pull plugins/config automatically on link-up and when the master pushes. + auto: true + # slave: bundle allow/deny lists (exact names or prefix*). deny wins. + allow: [] + deny: [] + # slave: names of credential refs to accept from the master (opt-in, per secret). + secrets: [] + # master: credential refs offered to devices (each device still opts in on the master + # Devices page AND locally via `secrets` above). + offerSecrets: [] + # master: never sync these bundles / cordis.patch.yml entry ids. + excludeBundles: [] + excludePatchIds: [] diff --git a/deploy/dsh-web.user.service b/deploy/dsh-web.user.service new file mode 100644 index 0000000..423dc4b --- /dev/null +++ b/deploy/dsh-web.user.service @@ -0,0 +1,15 @@ +[Unit] +Description=DeepSeek Harness (dsh web) for this user on 127.0.0.1:3081 +After=network-online.target + +[Service] +Type=simple +WorkingDirectory=%h +Environment=NODE_ENV=production +Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +ExecStart=%h/.local/bin/dsh web --host 127.0.0.1 --port 3081 --no-open +Restart=always +RestartSec=5 + +[Install] +WantedBy=default.target diff --git a/deploy/dshr-cli.mjs b/deploy/dshr-cli.mjs new file mode 100644 index 0000000..98a3169 --- /dev/null +++ b/deploy/dshr-cli.mjs @@ -0,0 +1,38 @@ +#!/usr/bin/env node +// Tiny admin CLI for dsh-remote that talks to the LOCAL DSH over loopback, authenticating +// with DSH's own browser-session secret from $DSH_HOME/.credentials.yaml (owner-only file). +// +// node dshr-cli.mjs GET state +// node dshr-cli.mjs POST pairing '{}' +// node dshr-cli.mjs POST pair '{"masterUrl":"http://192.168.0.111:3080","code":"ABCD-..."}' +// node dshr-cli.mjs GET /r/<device>/ (raw path, prints status + first bytes) +// +// Env: DSH_PORT (default 3081), DSH_HOME (default ~/.dsh) +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { mintCookie } from '../lib/auth-cookie.mjs' + +const [method = 'GET', target = 'state', body] = process.argv.slice(2) +const port = process.env.DSH_PORT || '3081' +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const creds = fs.readFileSync(path.join(home, '.credentials.yaml'), 'utf8') +const m = creds.match(/client-connection\/browser-session:[\s\S]*?secret:\s*([A-Za-z0-9_-]{43})/) +if (!m) { console.error('browser-session secret not found'); process.exit(2) } +const authority = `127.0.0.1:${port}` +const cookie = mintCookie(authority, m[1], 1) +const url = target.startsWith('/') ? `http://${authority}${target}` : `http://${authority}/dsh-remote/${target}` +const res = await fetch(url, { + method, + headers: { cookie, ...(body ? { 'content-type': 'application/json' } : {}) }, + body: body || undefined, + redirect: 'manual', +}) +const text = await res.text() +if (target.startsWith('/')) { + console.log(res.status, res.headers.get('content-type') || '', res.headers.get('location') || '') + console.log(text.slice(0, Number(process.env.BYTES || 400))) +} else { + console.log(res.status) + try { console.log(JSON.stringify(JSON.parse(text), null, 2)) } catch { console.log(text) } +} diff --git a/deploy/e2e-events.mjs b/deploy/e2e-events.mjs new file mode 100644 index 0000000..0de1474 --- /dev/null +++ b/deploy/e2e-events.mjs @@ -0,0 +1,20 @@ +#!/usr/bin/env node +// E2E: listen to the local /dsh-remote/stream (SSE) for N seconds and print events. +// node deploy/e2e-events.mjs [seconds] [port] +import fs from 'node:fs' +import http from 'node:http' +import os from 'node:os' +import path from 'node:path' +import { mintCookie } from '../lib/auth-cookie.mjs' + +const secs = Number(process.argv[2] || 30) +const port = process.argv[3] || process.env.DSH_PORT || '3081' +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const secret = fs.readFileSync(path.join(home, '.credentials.yaml'), 'utf8').match(/browser-session:[\s\S]*?secret:\s*([A-Za-z0-9_-]{43})/)[1] +const authority = `127.0.0.1:${port}` +const req = http.get({ host: '127.0.0.1', port, path: '/dsh-remote/stream', headers: { cookie: mintCookie(authority, secret, 1), host: authority } }, (res) => { + console.log('stream', res.statusCode) + res.setEncoding('utf8') + res.on('data', (d) => { for (const line of d.split('\n')) if (line.startsWith('data:')) console.log(new Date().toISOString().slice(11, 19), line.slice(5).trim()) }) +}) +setTimeout(() => { req.destroy(); process.exit(0) }, secs * 1000) diff --git a/deploy/e2e-exc.mjs b/deploy/e2e-exc.mjs new file mode 100644 index 0000000..8b9802e --- /dev/null +++ b/deploy/e2e-exc.mjs @@ -0,0 +1,43 @@ +#!/usr/bin/env node +// Load one URL N times in headless Chromium and count uncaught exceptions per load. +// node deploy/e2e-exc.mjs <chrome> <url> <cookieAuthority> [loads] [settleMs] +import { spawn } from 'node:child_process' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { cookieName, mintCookie } from '../lib/auth-cookie.mjs' + +const [chrome, url, authority, loads = '3', settle = '8000'] = process.argv.slice(2) +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const secret = fs.readFileSync(path.join(home, '.credentials.yaml'), 'utf8').match(/browser-session:[\s\S]*?secret:\s*([A-Za-z0-9_-]{43})/)[1] +const prof = fs.mkdtempSync(path.join(os.tmpdir(), 'dshr-exc-')) +const proc = spawn(chrome, ['--headless=new', '--no-sandbox', '--disable-gpu', `--user-data-dir=${prof}`, '--remote-debugging-port=0', 'about:blank'], { stdio: ['ignore', 'ignore', 'pipe'] }) +const wsUrl = await new Promise((resolve, reject) => { + let buf = '' + proc.stderr.on('data', (d) => { buf += d; const m = buf.match(/DevTools listening on (ws:\/\/\S+)/); if (m) resolve(m[1]) }) + setTimeout(() => reject(new Error('chrome did not start')), 15000) +}) +const page = (await (await fetch(wsUrl.replace(/^ws/, 'http').replace(/\/devtools\/browser\/.*/, '/json/list'))).json()).find((t) => t.type === 'page') +const ws = new WebSocket(page.webSocketDebuggerUrl) +await new Promise((r) => ws.addEventListener('open', r)) +let id = 0 +const pending = new Map() +let exc = [] +ws.addEventListener('message', (ev) => { + const m = JSON.parse(ev.data) + if (m.id && pending.has(m.id)) { pending.get(m.id)(m); pending.delete(m.id) } + if (m.method === 'Runtime.exceptionThrown') exc.push((m.params.exceptionDetails?.exception?.description || '').split('\n')[0].slice(0, 120)) +}) +const send = (method, params = {}) => new Promise((resolve) => { const i = ++id; pending.set(i, resolve); ws.send(JSON.stringify({ id: i, method, params })) }) +await send('Page.enable'); await send('Runtime.enable'); await send('Network.enable') +const host = authority.split(':')[0] +await send('Network.setCookie', { name: cookieName(authority), value: mintCookie(authority, secret, 1).split('=').slice(1).join('='), domain: host, path: '/', httpOnly: true }) +for (let i = 0; i < Number(loads); i++) { + exc = [] + await send('Page.navigate', { url: `${url}${url.includes('?') ? '&' : '?'}n=${i}` }) + await new Promise((r) => setTimeout(r, Number(settle))) + console.log(`load ${i}: ${exc.length} exceptions ${exc.join(' || ')}`) +} +ws.close(); proc.kill('SIGKILL') +try { fs.rmSync(prof, { recursive: true, force: true }) } catch { /* chrome may still hold files */ } +process.exit(0) diff --git a/deploy/e2e-ui.mjs b/deploy/e2e-ui.mjs new file mode 100644 index 0000000..a170efd --- /dev/null +++ b/deploy/e2e-ui.mjs @@ -0,0 +1,73 @@ +#!/usr/bin/env node +// E2E UI check with headless Chromium over the DevTools protocol (no playwright dep). +// node deploy/e2e-ui.mjs <chrome-binary> <out-dir> [port] +// Loads the local DSH (cookie minted from the browser-session secret), opens the Devices +// panel, screenshots it, then opens /r/<first remote device>/ and screenshots that. +import { spawn } from 'node:child_process' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { cookieName, mintCookie } from '../lib/auth-cookie.mjs' + +const [chrome, outDir, port = '3081'] = process.argv.slice(2) +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const secret = fs.readFileSync(path.join(home, '.credentials.yaml'), 'utf8').match(/browser-session:[\s\S]*?secret:\s*([A-Za-z0-9_-]{43})/)[1] +const authority = `127.0.0.1:${port}` +const cookieVal = mintCookie(authority, secret, 1).split('=').slice(1).join('=') +fs.mkdirSync(outDir, { recursive: true }) +const prof = fs.mkdtempSync(path.join(os.tmpdir(), 'dshr-chrome-')) +const proc = spawn(chrome, ['--headless=new', '--no-sandbox', '--disable-gpu', `--user-data-dir=${prof}`, '--remote-debugging-port=0', '--window-size=1500,950', 'about:blank'], { stdio: ['ignore', 'ignore', 'pipe'] }) +const wsUrl = await new Promise((resolve, reject) => { + let buf = '' + proc.stderr.on('data', (d) => { buf += d; const m = buf.match(/DevTools listening on (ws:\/\/\S+)/); if (m) resolve(m[1]) }) + setTimeout(() => reject(new Error('chrome did not start')), 15000) +}) +const list = await (await fetch(wsUrl.replace(/^ws/, 'http').replace(/\/devtools\/browser\/.*/, '/json/list'))).json() +const page = list.find((t) => t.type === 'page') +const ws = new WebSocket(page.webSocketDebuggerUrl) +await new Promise((r) => ws.addEventListener('open', r)) +let id = 0 +const pending = new Map() +const logs = [] +ws.addEventListener('message', (ev) => { + const m = JSON.parse(ev.data) + if (m.id && pending.has(m.id)) { pending.get(m.id)(m); pending.delete(m.id) } + if (m.method === 'Runtime.consoleAPICalled') logs.push(`${m.params.type}: ${m.params.args.map((a) => a.value ?? a.description ?? '').join(' ')}`.slice(0, 300)) + if (m.method === 'Runtime.exceptionThrown') logs.push(`EXC: ${m.params.exceptionDetails?.exception?.description || m.params.exceptionDetails?.text}`.slice(0, Number(process.env.EXC_LEN || 400))) +}) +const send = (method, params = {}) => new Promise((resolve) => { const i = ++id; pending.set(i, resolve); ws.send(JSON.stringify({ id: i, method, params })) }) +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) +const evalJs = async (expr) => (await send('Runtime.evaluate', { expression: expr, awaitPromise: true, returnByValue: true })).result?.result?.value +const shot = async (name) => { + const r = await send('Page.captureScreenshot', { format: 'png' }) + fs.writeFileSync(path.join(outDir, name), Buffer.from(r.result.data, 'base64')) + console.log('screenshot', path.join(outDir, name)) +} + +await send('Page.enable'); await send('Runtime.enable') +await send('Network.enable') +await send('Network.setCookie', { name: cookieName(authority), value: cookieVal, domain: '127.0.0.1', path: '/', httpOnly: true }) +await send('Page.navigate', { url: `http://${authority}/` }) +await sleep(9000) +// Click the Devices panel entry (its icon carries data-dsh-panel-entry). +const clicked = await evalJs(`(() => { const svg = document.querySelector('[data-dsh-panel-entry="dsh-remote-devices"]'); if (!svg) return 'no-entry'; (svg.closest('button,a,[role=button]') || svg).click(); return 'clicked' })()`) +console.log('devices entry:', clicked) +await sleep(3500) +console.log('panel text:', String(await evalJs(`(document.querySelector('[data-dsh-plugin="dsh-remote"]')?.innerText || 'NO PANEL').slice(0, 600)`)).replace(/\n+/g, ' | ')) +await shot('devices-panel.png') +const localErrors = logs.filter((l) => /EXC/.test(l)).length +console.log('local page exceptions:', localErrors, logs.filter((l) => /EXC/.test(l)).map((l) => l.slice(0, 90)).join(' || ')) +logs.length = 0 + +const remote = process.env.REMOTE_DEVICE || 'ai-pc' +const remoteSession = process.env.REMOTE_SESSION || '' +await send('Page.navigate', { url: `http://${authority}/r/${remote}/${remoteSession ? `#/session/${remoteSession}` : ''}` }) +await sleep(remoteSession ? 14000 : 10000) +console.log('remote title/banner:', await evalJs(`document.title + ' :: ' + (document.querySelector('a[title*="another device"]')?.innerText || 'no banner') + ' :: hash=' + location.hash`)) +if (remoteSession) { + console.log('remote session view:', String(await evalJs(`document.body.innerText.slice(0, 500)`)).replace(/\n+/g, ' | ')) +} +await shot('remote-ui.png') +console.log('console:', logs.filter((l) => /error|EXC|warn/i.test(l)).slice(0, 12).join('\n ')) +ws.close(); proc.kill('SIGKILL'); fs.rmSync(prof, { recursive: true, force: true }) +process.exit(0) diff --git a/deploy/e2e-ws.mjs b/deploy/e2e-ws.mjs new file mode 100644 index 0000000..51589e1 --- /dev/null +++ b/deploy/e2e-ws.mjs @@ -0,0 +1,28 @@ +#!/usr/bin/env node +// E2E probe: open the DSH Remote stream WebSocket of a device THROUGH the local +// /r/<device>/api/remote.mux proxy and check that the remote DSH answers. +// node deploy/e2e-ws.mjs <device> [port] +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { connect } from '../lib/ws.mjs' +import { mintCookie } from '../lib/auth-cookie.mjs' + +const device = process.argv[2] || 'ai-pc' +const port = process.argv[3] || process.env.DSH_PORT || '3081' +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const secret = fs.readFileSync(path.join(home, '.credentials.yaml'), 'utf8').match(/browser-session:[\s\S]*?secret:\s*([A-Za-z0-9_-]{43})/)[1] +const authority = `127.0.0.1:${port}` +const cookie = mintCookie(authority, secret, 1) +const t0 = Date.now() +const ws = await connect(`ws://${authority}/r/${device}/api/remote.mux`, { headers: { cookie, origin: `http://${authority}` } }) +console.log(`ws open through /r/${device}/ in ${Date.now() - t0} ms`) +const got = await new Promise((resolve) => { + const timer = setTimeout(() => resolve('(no frame within 4s: connection stays open, server waits for client frames)'), 4000) + ws.on('message', (m) => { clearTimeout(timer); resolve(`first frame: ${String(m).slice(0, 160)}`) }) + ws.on('close', (c, r) => { clearTimeout(timer); resolve(`closed ${c} ${r}`) }) +}) +console.log(got) +console.log(`still open: ${ws.open}`) +ws.close() +process.exit(0) diff --git a/deploy/install-local.mjs b/deploy/install-local.mjs new file mode 100644 index 0000000..fd82638 --- /dev/null +++ b/deploy/install-local.mjs @@ -0,0 +1,61 @@ +#!/usr/bin/env node +// Register a dsh-remote checkout as a `link:` bundle in the current user's DSH web profile +// (no pnpm/network needed) and optionally set its config. +// +// node deploy/install-local.mjs [--role slave|master|auto] [--name "My PC"] [--profile web] +// +// Effects (idempotent): +// <profile>/package.json dependencies["@local/dsh-remote"] = "link:<this dir>" +// dsh.profile.bundles += "@local/dsh-remote" +// <profile>/node_modules/@local/dsh-remote -> <this dir> +// <profile>/cordis.patch.yml appends a `- id: dsh-remote` config block if missing +// Restart `dsh web` afterwards (host code is loaded at boot). +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { fileURLToPath } from 'node:url' + +const args = process.argv.slice(2) +const opt = (k, d) => { const i = args.indexOf(`--${k}`); return i >= 0 ? args[i + 1] : d } +const pluginDir = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..') +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const prof = path.join(home, 'profiles', opt('profile', 'web')) +const NAME = '@local/dsh-remote' + +const pkgFile = path.join(prof, 'package.json') +const pkg = JSON.parse(fs.readFileSync(pkgFile, 'utf8')) +pkg.dependencies ||= {} +pkg.dependencies[NAME] = `link:${pluginDir}` +const bundles = (((pkg.dsh ||= {}).profile ||= {}).bundles ||= []) +if (!bundles.includes(NAME)) bundles.push(NAME) +fs.writeFileSync(pkgFile, JSON.stringify(pkg, null, 2) + '\n') + +const linkDir = path.join(prof, 'node_modules', '@local') +fs.mkdirSync(linkDir, { recursive: true }) +const link = path.join(linkDir, 'dsh-remote') +try { fs.rmSync(link, { recursive: true, force: true }) } catch { /* ignore */ } +fs.symlinkSync(pluginDir, link) + +const patchFile = path.join(prof, 'cordis.patch.yml') +let patch = '' +try { patch = fs.readFileSync(patchFile, 'utf8') } catch { patch = '' } +if (!/^- id: dsh-remote\b/m.test(patch)) { + const role = opt('role', 'auto') + const name = opt('name', '') + const block = [ + '# dsh-remote (installed by deploy/install-local.mjs)', + '- id: dsh-remote', + ' config:', + ` role: ${role}`, + ` deviceName: ${JSON.stringify(name)}`, + ' sync:', + ' auto: true', + ' allow: []', + ' deny: []', + ' secrets: []', + '', + ].join('\n') + // An empty/comment-only patch file fails DSH boot, so only append real entries. + fs.writeFileSync(patchFile, (patch.trim() && !/^\s*\[\]\s*$/m.test(patch.trim()) ? patch.replace(/\n*$/, '\n') : '') + block) +} +console.log(`dsh-remote linked into ${prof}; restart dsh web to load it.`) diff --git a/deploy/push-slave.sh b/deploy/push-slave.sh new file mode 100755 index 0000000..f83d92e --- /dev/null +++ b/deploy/push-slave.sh @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +# Dev helper: copy this checkout to a slave's ~/.dsh/plugins/dsh-remote and restart its dsh. +# deploy/push-slave.sh agent@192.168.0.110 +set -euo pipefail +HOST="${1:?usage: push-slave.sh user@host}" +DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +O=(-o BatchMode=yes -o ConnectTimeout=5 -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR) +tar --exclude=.git --exclude=node_modules -C "$DIR" -czf - . | timeout 20 ssh "${O[@]}" "$HOST" \ + 'mkdir -p ~/.dsh/plugins/dsh-remote && tar -xzf - -C ~/.dsh/plugins/dsh-remote' +timeout 45 ssh "${O[@]}" "$HOST" 'export PATH=$HOME/.local/bin:$PATH XDG_RUNTIME_DIR=/run/user/$(id -u) +systemctl --user restart dsh +for i in $(seq 1 40); do curl -s -m 2 -o /dev/null http://127.0.0.1:3081/ && break; sleep 0.5; done +echo "slave dsh: $(systemctl --user is-active dsh)"' diff --git a/index.js b/index.js new file mode 100644 index 0000000..efa7af1 --- /dev/null +++ b/index.js @@ -0,0 +1,727 @@ +// @local/dsh-remote — Host side. +// +// Master/slave remote control for DSH (Claude-Desktop-remote style): +// * slaves keep one authenticated WebSocket link to the master (outbound, NAT-friendly); +// * every device's DSH is reachable from every other device at /r/<device>/ (HTTP + +// WebSocket tunnelled over the link; sessions execute on the owning machine); +// slave→slave goes through the master's relay; +// * the master aggregates device + session lists for the Devices UI; +// * plugins and config are synced master → slave (managed bundle layer, allow/deny, +// opt-in per-secret credential sync). +// +// HTTP routes (all behind DSH's own browser auth via ctx.connection.requestRejection): +// GET /dsh-remote/state → role, self, link status, devices (+ online) +// GET /dsh-remote/sessions → sessions of every reachable device +// POST /dsh-remote/pairing → master: create a one-time pairing code +// POST /dsh-remote/pair → slave: pair with {masterUrl, code} +// POST /dsh-remote/unpair → slave: forget the master +// POST /dsh-remote/devices/<id>/revoke → master: revoke a device key +// POST /dsh-remote/devices/<id>/remove → master: delete a device +// POST /dsh-remote/devices/<id>/sync → master: set {enabled, allow, deny, secrets} +// POST /dsh-remote/sync → slave: pull + apply now +// GET /dsh-remote/sync/status → slave: last sync result / master: manifest summary +// POST /dsh-remote/new-session → {device, cwd?} create a session on a device +// * /r/<device>/... → proxy to that device's DSH +// Upgrade routes: +// /dsh-remote/link → slave → master link (device id + key headers) +// /r/<device>/api/remote.mux → DSH Remote stream WebSocket of another device +import os from 'node:os' +import { acceptUpgrade, connect, rejectUpgrade } from './lib/ws.mjs' +import { Mux } from './lib/mux.mjs' +import { Store, randomId, setProfileDir, slug } from './lib/store.mjs' +import { CookieJar } from './lib/auth-cookie.mjs' +import { ownHttp, ownUpgrade, viewHttp, viewUpgrade } from './lib/proxy.mjs' +import * as sync from './lib/sync.mjs' + +export const name = 'dsh-remote' +export const inject = { + webServer: { required: true }, + connection: { required: true }, + credentials: { required: false }, +} + +/** Optional services, read at call time (they may load after this plugin). */ +function svc(ctx, key) { + try { return ctx.get(key) } catch { return undefined } +} + +const MAX_BODY = 256 * 1024 +const PREFIX = '/r/' +const LINK_PATH = '/dsh-remote/link' +const MUX_PATH = '/api/remote.mux' + +function sendJson(res, status, payload) { + res.statusCode = status + res.setHeader('content-type', 'application/json; charset=utf-8') + res.setHeader('cache-control', 'no-store') + res.end(JSON.stringify(payload)) +} + +async function readJson(req) { + const chunks = [] + let size = 0 + for await (const c of req) { + size += c.length + if (size > MAX_BODY) throw Object.assign(new Error('body too large'), { status: 413 }) + chunks.push(c) + } + if (!size) return {} + const v = JSON.parse(Buffer.concat(chunks).toString('utf8')) + if (!v || typeof v !== 'object' || Array.isArray(v)) throw Object.assign(new Error('body must be an object'), { status: 400 }) + return v +} + +const httpErr = (status, message) => Object.assign(new Error(message), { status }) + +export function apply(ctx, config = {}) { + const prof = svc(ctx, 'profileContext') + if (prof?.dir) setProfileDir(prof.dir) + const store = new Store() + const self = store.identity(config.deviceName || os.hostname()) + if (config.deviceName && config.deviceName !== self.name) { + self.name = config.deviceName + store.save() + } + const role = config.role === 'master' ? 'master' : config.role === 'slave' ? 'slave' : (store.master ? 'slave' : 'master') + const log = ctx.logger ?? console + const cookies = new CookieJar(ctx) + const localTarget = () => ({ host: '127.0.0.1', port: ctx.webServer.port }) + + // device id -> { mux, info, connectedAt } (master: slaves; slave: { master }) + const links = new Map() + const listeners = new Set() + const notify = (event) => { for (const l of listeners) { try { l(event) } catch { /* ignore */ } } } + let lastSync = store.data.lastSync || null + let masterLink = null // slave only + + // ------------------------------------------------------------------ stream routing + // A stream's meta.to names the device that must serve it. "self" streams hit local DSH. + // The master relays streams whose target is another slave. + function serveStream(stream) { + const to = stream.meta.to + if (to && to !== self.id) { + if (role !== 'master') return stream.fail(`cannot relay to ${to}`) + const peer = links.get(to) + if (!peer) return stream.fail(`device ${to} offline`) + return relay(stream, peer.mux) + } + if (stream.kind === 'http') return ownHttp(stream, { target: localTarget(), cookie: (a) => cookies.get(a) }) + if (stream.kind === 'ws') return ownUpgrade(stream, { target: localTarget(), cookie: (a) => cookies.get(a) }) + if (stream.kind === 'blob') return serveBlob(stream) + return stream.fail(`unknown stream kind ${stream.kind}`) + } + + /** Splice an incoming stream onto a new stream to another link (master relay). */ + function relay(inbound, mux) { + const out = mux.open(inbound.kind, inbound.meta) + const pipe = (a, b) => { + a.on('head', (m) => b.head(m)) + a.on('data', (d) => b.write(d)) + a.on('end', () => b.end()) + a.on('error', (e) => b.fail(e.message)) + a.on('remote-pause', () => b.mux.sendControl({ t: 'pause', id: b.id })) + a.on('remote-resume', () => b.mux.sendControl({ t: 'resume', id: b.id })) + } + pipe(inbound, out) + pipe(out, inbound) + inbound.start() + } + + // ------------------------------------------------------------------ control handlers + const handlers = { + hello: async (p) => ({ id: self.id, name: self.name, role, version: config.version || '1' }), + sessions: async () => listLocalSessions(), + newSession: async (p) => createLocalSession(p), + devices: async (p, mux) => devicesView(), + route: async () => ({ ok: true }), + manifest: async (p, mux) => { + if (role !== 'master') throw httpErr(400, 'not a master') + const dev = store.devices()[mux.deviceId] + if (dev?.sync?.enabled === false) return { version: 1, bundles: [], patch: [], secrets: [], disabled: true } + const m = await buildManifestCached() + return { ...m, _dirs: undefined, secrets: offerSecretsFor(dev) } + }, + secret: async (p, mux) => { + if (role !== 'master') throw httpErr(400, 'not a master') + const dev = store.devices()[mux.deviceId] + const name = String(p.name || '') + if (!offerSecretsFor(dev).includes(name)) throw httpErr(403, `secret ${name} not offered to this device`) + const c = svc(ctx, 'credentials') + if (!c) throw httpErr(503, 'credentials service unavailable') + const r = await c.resolve(name) + if (!r?.value) throw httpErr(404, `secret ${name} not set on master`) + return { name, value: r.value } + }, + // Forwarded calls: master answers for another device by asking it. + forward: async (p) => { + if (role !== 'master') throw httpErr(400, 'not a master') + const peer = links.get(p.device) + if (!peer) throw httpErr(404, `device ${p.device} offline`) + return peer.mux.call(p.method, p.params || {}, 20000) + }, + } + + function offerSecretsFor(dev) { + const offered = new Set(config.sync?.offerSecrets || []) + return (dev?.sync?.secrets || []).filter((n) => offered.has(n)) + } + + let manifestCache = null + async function buildManifestCached() { + if (manifestCache && Date.now() - manifestCache.at < 15000) return manifestCache.value + const value = await sync.buildManifest({ + self, + include: config.sync || {}, + offerSecrets: config.sync?.offerSecrets || [], + }) + manifestCache = { at: Date.now(), value } + return value + } + + async function serveBlob(stream) { + // Only the master serves blobs: packed local plugins named in its manifest. + try { + if (role !== 'master') throw new Error('not a master') + const m = await buildManifestCached() + const dir = m._dirs?.[stream.meta.name] + if (!dir) throw new Error(`no local plugin ${stream.meta.name}`) + const dev = store.devices()[stream.mux.deviceId] + if (dev && !sync.allowed(stream.meta.name, dev.sync || {})) throw new Error('denied for this device') + const buf = await sync.packLocal(dir) + stream.head({ status: 200, size: buf.length }) + stream.write(buf) + stream.end() + } catch (e) { + stream.fail(e.message) + } + } + + // ------------------------------------------------------------------ sessions + async function listLocalSessions() { + const sc = svc(ctx, 'sessionController') + if (!sc) return { items: [], workspaces: [] } + const ac = new AbortController() + const t = setTimeout(() => ac.abort(), 10000) + try { + const { items } = await sc.list({}, ac.signal) + const wr = svc(ctx, 'workspaceRegistry') + const workspaces = wr ? wr.list().map((w) => ({ id: w.id, title: w.title, path: w.path })) : [] + return { + items: items + .filter((s) => !s.parentSessionId && !s.blank) + .slice(0, config.maxSessions || 200) + .map((s) => ({ + sessionId: s.sessionId, + updatedAt: s.updatedAt, + running: s.running, + cwd: s.cwd, + title: s.projections?.values?.title ?? null, + })), + workspaces, + } + } finally { + clearTimeout(t) + } + } + + async function createLocalSession({ cwd, workspaceId } = {}) { + const sc = svc(ctx, 'sessionController') + if (!sc) throw httpErr(503, 'session controller unavailable') + const req = {} + if (workspaceId) req.workspaceId = workspaceId + else if (cwd) req.cwd = cwd + const r = await sc.create(req) + return { sessionId: r.sessionId } + } + + // ------------------------------------------------------------------ device views + function devicesView() { + const out = [{ id: self.id, name: self.name, role, self: true, online: true }] + if (role === 'master') { + for (const d of Object.values(store.devices())) { + const l = links.get(d.id) + out.push({ + id: d.id, name: d.name, platform: d.platform, role: 'slave', revoked: !!d.revoked, + online: !!l, connectedAt: l?.connectedAt || null, lastSeenAt: d.lastSeenAt || null, + sync: d.sync, + }) + } + } else if (masterLink?.info) { + out.push({ ...masterLink.info, role: 'master', online: true }) + // Peers in the master's broadcast are by definition online right now. + for (const d of masterLink.peers || []) if (d.id !== self.id && d.role !== 'master') out.push({ ...d, online: true, via: 'master' }) + } + return out + } + + /** Mux that reaches `device` from here (null = local). */ + function routeTo(device) { + if (!device || device === self.id) return { local: true } + if (role === 'master') { + const l = links.get(device) + return l ? { mux: l.mux, meta: {} } : null + } + if (!masterLink?.mux || masterLink.mux.closed) return null + if (masterLink.info?.id === device) return { mux: masterLink.mux, meta: {} } + return { mux: masterLink.mux, meta: { to: device } } // relay through master + } + + async function callDevice(device, method, params) { + const r = routeTo(device) + if (!r) throw httpErr(404, `device ${device} unreachable`) + if (r.local) return handlers[method](params || {}) + if (r.meta.to) return r.mux.call('forward', { device, method, params }, 25000) + return r.mux.call(method, params || {}, 20000) + } + + async function allSessions() { + const devices = devicesView().filter((d) => d.online && !d.revoked) + const results = await Promise.allSettled(devices.map(async (d) => ({ device: d, ...(await callDevice(d.id, 'sessions', {})) }))) + return results.map((r, i) => r.status === 'fulfilled' + ? { device: devices[i].id, name: devices[i].name, self: !!devices[i].self, items: r.value.items, workspaces: r.value.workspaces } + : { device: devices[i].id, name: devices[i].name, self: !!devices[i].self, error: String(r.reason?.message ?? r.reason), items: [] }) + } + + // ------------------------------------------------------------------ session activity watch + // Poll local running states; emit "finished" events (for notifications) to all viewers. + const lastRunning = new Map() + async function pollActivity() { + try { + const { items } = await listLocalSessions() + for (const s of items) { + const was = lastRunning.get(s.sessionId) + if (was === true && !s.running) { + const ev = { type: 'session-finished', device: self.id, deviceName: self.name, sessionId: s.sessionId, title: s.title, at: Date.now() } + notify(ev) + broadcast('activity', ev) + } + lastRunning.set(s.sessionId, s.running) + } + } catch { /* ignore */ } + } + const activityTimer = setInterval(pollActivity, config.activityPollMs || 4000) + activityTimer.unref?.() + ctx.effect(() => () => clearInterval(activityTimer), 'dsh-remote: activity poll') + + function broadcast(name, data) { + if (role === 'master') for (const l of links.values()) l.mux.event(name, data) + else masterLink?.mux?.event(name, data) + } + + function onLinkEvent(name, data, fromDevice) { + if (name === 'activity') { + notify(data) + // Master fans slave activity out to the other slaves. + if (role === 'master') for (const [id, l] of links) if (id !== fromDevice) l.mux.event(name, data) + } else if (name === 'peers' && role === 'slave' && masterLink) { + masterLink.peers = data + notify({ type: 'devices' }) + } else if (name === 'sync-available' && role === 'slave') { + if (config.sync?.auto !== false) runSlaveSync().catch((e) => log.warn(`dsh-remote: sync failed: ${e.message}`)) + } + } + + function pushPeers() { + if (role !== 'master') return + const peers = devicesView().filter((d) => d.online).map(({ id, name, role: r, platform }) => ({ id, name, role: r, platform })) + for (const l of links.values()) l.mux.event('peers', peers) + notify({ type: 'devices' }) + } + + // ------------------------------------------------------------------ master: accept links + if (role === 'master') { + ctx.effect(() => ctx.webServer.registerUpgrade({ + path: LINK_PATH, + handler: (req, socket, head) => { + const id = String(req.headers['x-dsh-remote-device'] || '') + const key = String(req.headers['x-dsh-remote-key'] || '') + const dev = store.verifyDevice(id, key) + if (!dev) return rejectUpgrade(socket, 401, 'unknown or revoked device') + const ws = acceptUpgrade(req, socket, head) + if (!ws) return + const prev = links.get(id) + if (prev) prev.mux.close(4000, 'replaced by new connection') + const mux = new Mux(ws, { odd: false, handlers, onStream: serveStream }) + mux.deviceId = id + const entry = { mux, connectedAt: Date.now(), info: { id, name: dev.name } } + links.set(id, entry) + dev.lastSeenAt = Date.now() + dev.platform = String(req.headers['x-dsh-remote-platform'] || dev.platform || '') + store.save() + mux.on('event', (n, d) => onLinkEvent(n, d, id)) + mux.on('close', () => { + if (links.get(id) === entry) links.delete(id) + const d = store.devices()[id] + if (d) { d.lastSeenAt = Date.now(); store.save() } + log.info?.(`dsh-remote: ${id} disconnected`) + pushPeers() + }) + log.info?.(`dsh-remote: ${id} connected`) + pushPeers() + }, + }), 'dsh-remote: link endpoint') + } + + // ------------------------------------------------------------------ slave: keep link up + let stopped = false + let backoff = 1000 + async function connectMaster() { + const m = store.master + if (!m || stopped) return + const url = new URL(m.url) + url.protocol = url.protocol === 'https:' ? 'wss:' : url.protocol === 'http:' ? 'ws:' : url.protocol + url.pathname = url.pathname.replace(/\/+$/, '') + LINK_PATH + try { + const ws = await connect(url.href, { + headers: { + 'x-dsh-remote-device': m.deviceId, + 'x-dsh-remote-key': m.key, + 'x-dsh-remote-platform': `${process.platform}-${process.arch}`, + }, + }) + const mux = new Mux(ws, { odd: true, handlers, onStream: serveStream }) + mux.deviceId = 'master' + masterLink = { mux, info: null, peers: [] } + backoff = 1000 + mux.on('event', (n, d) => onLinkEvent(n, d, 'master')) + mux.on('close', (code, reason) => { + masterLink = null + notify({ type: 'devices' }) + if (code === 4401) { + lastLinkError = `master rejected this device (${reason || 'revoked'}); re-pair it` + log.warn(`dsh-remote: ${lastLinkError}`) + return + } + lastLinkError = `link closed (${code}${reason ? ' ' + reason : ''}); reconnecting` + if (!stopped) schedule() + }) + const info = await mux.call('hello', { id: self.id, name: self.name }) + masterLink.info = { id: info.id, name: info.name } + // Only devices that are online on the master are reachable through its relay. + masterLink.peers = (await mux.call('devices', {})).filter((d) => d.online && !d.revoked).map(({ id, name, role: r }) => ({ id, name, role: r })) + notify({ type: 'devices' }) + lastLinkError = null + log.info?.(`dsh-remote: linked to master ${info.name} (${m.url})`) + if (config.sync?.auto !== false) runSlaveSync().catch((e) => log.warn(`dsh-remote: sync failed: ${e.message}`)) + } catch (e) { + masterLink = null + if (e.status === 401) { log.warn(`dsh-remote: master refused key (${e.message}); re-pair this device`); lastLinkError = e.message; return } + lastLinkError = e.message + schedule() + } + } + let lastLinkError = null + let timer = null + function schedule() { + clearTimeout(timer) + timer = setTimeout(connectMaster, backoff) + timer.unref?.() + backoff = Math.min(backoff * 2, 30000) + } + if (role === 'slave') connectMaster() + ctx.effect(() => () => { + stopped = true + clearTimeout(timer) + masterLink?.mux?.close(1001, 'shutdown') + for (const l of links.values()) l.mux.close(1001, 'shutdown') + }, 'dsh-remote: links') + + // ------------------------------------------------------------------ slave: sync + let syncing = null + async function runSlaveSync() { + if (role !== 'slave') throw httpErr(400, 'only slaves sync') + if (syncing) return syncing + syncing = (async () => { + const mux = masterLink?.mux + if (!mux) throw httpErr(503, 'not linked to a master') + const local = config.sync || {} + const manifest = await mux.call('manifest', {}, 30000) + const result = { at: Date.now(), installed: [], updated: [], skipped: [], failed: [], secrets: [], patchEntries: manifest.patch?.length || 0 } + if (manifest.disabled) { result.skipped.push('sync disabled for this device on master'); return result } + const pm = svc(ctx, 'pluginManager') + const prev = store.data.synced || {} + const next = {} + let profileChanged = false + for (const b of manifest.bundles || []) { + if (!sync.allowed(b.name, local)) { result.skipped.push(`${b.name} (local deny/allow)`); continue } + try { + if (b.kind === 'local') { + if (prev[b.name]?.hash === b.hash) { next[b.name] = prev[b.name]; continue } + const stream = mux.open('blob', { name: b.name }) + const buf = await new Promise((resolve, reject) => { + const parts = [] + stream.on('data', (d) => parts.push(d)) + stream.on('end', () => { resolve(Buffer.concat(parts)); stream.end() }) + stream.on('error', reject) + }) + const dir = await sync.unpackLocal(b.name, b.version, buf) + profileChanged = sync.linkLocalBundle(b.name, dir) || profileChanged + next[b.name] = { kind: 'local', hash: b.hash, version: b.version, dir } + ;(prev[b.name] ? result.updated : result.installed).push(`${b.name}@${b.version}`) + } else { + if (!pm) throw new Error('plugin manager unavailable') + const have = (await pm.listBundles()).find((x) => x.name === b.name) + if (have?.installed && have.version === b.version) { + if (!have.enabled) await pm.setBundleEnabled(b.name, true) + next[b.name] = { kind: 'registry', version: b.version } + continue + } + const r = await pm.installBundle(`${b.name}@${b.version}`, { enabled: true }) + if (r.application === 'failed') throw new Error(r.error?.message || 'install failed') + next[b.name] = { kind: 'registry', version: b.version } + ;(have ? result.updated : result.installed).push(`${b.name}@${b.version}`) + } + } catch (e) { + result.failed.push(`${b.name}: ${e.message}`) + if (prev[b.name]) next[b.name] = prev[b.name] + } + } + // Config layer. + await sync.writeManagedBundle(manifest.patch || [], Object.keys(next)) + profileChanged = sync.ensureManagedBundleOrder() || profileChanged + // Opt-in secrets. + for (const name of manifest.secrets || []) { + if (!(local.secrets || []).includes(name)) continue + try { + const { value } = await mux.call('secret', { name }, 15000) + const creds = svc(ctx, 'credentials') + if (!creds) throw new Error('credentials service unavailable') + await creds.set(name, value) + result.secrets.push(name) + } catch (e) { + result.failed.push(`secret ${name}: ${e.message}`) + } + } + store.data.synced = next + store.data.lastSync = result + store.save() + lastSync = result + result.restartRequired = profileChanged + notify({ type: 'sync', result }) + return result + })() + try { return await syncing } finally { syncing = null } + } + + // ------------------------------------------------------------------ HTTP API + const route = (path, methods, handler) => ctx.effect(() => ctx.webServer.register({ + kind: 'exact', + path, + handler: async (req, res) => { + const rejection = ctx.connection.requestRejection(req) + if (rejection !== undefined) return sendJson(res, rejection, { error: 'rejected' }) + if (!methods.includes(req.method)) { res.statusCode = 405; res.setHeader('allow', methods.join(', ')); return res.end() } + try { + sendJson(res, 200, await handler(req, new URL(req.url || '/', 'http://x'))) + } catch (e) { + sendJson(res, e.status && e.status < 600 ? e.status : 500, { error: String(e.message || e) }) + } + }, + }), `dsh-remote: ${path}`) + + route('/dsh-remote/state', ['GET'], async () => ({ + role, + self, + master: role === 'slave' ? { url: store.master?.url || null, linked: !!masterLink?.mux, info: masterLink?.info || null, error: masterLink ? null : lastLinkError } : null, + devices: devicesView(), + publicUrl: config.publicUrl || null, + lanUrl: config.lanUrl || null, + lastSync, + sync: { auto: config.sync?.auto !== false, allow: config.sync?.allow || [], deny: config.sync?.deny || [], secrets: config.sync?.secrets || [], offerSecrets: config.sync?.offerSecrets || [] }, + })) + + route('/dsh-remote/sessions', ['GET'], async () => ({ devices: await allSessions() })) + + route('/dsh-remote/pairing', ['POST'], async (req) => { + if (role !== 'master') throw httpErr(400, 'pairing codes are created on the master') + const body = await readJson(req) + return store.createPairingCode({ label: String(body.label || '') }) + }) + + // Pair endpoint ON THE MASTER, called by a slave's DSH (not by a browser): exchanges a + // pairing code for a device key. Not behind browser auth (the code is the credential). + ctx.effect(() => ctx.webServer.register({ + kind: 'exact', + path: '/dsh-remote/enroll', + handler: async (req, res) => { + if (req.method !== 'POST') { res.statusCode = 405; return res.end() } + try { + if (role !== 'master') throw httpErr(400, 'not a master') + const body = await readJson(req) + const rec = store.consumePairingCode(body.code) + if (!rec) throw httpErr(403, 'invalid or expired pairing code') + const dev = store.addDevice({ id: body.id, name: body.name, platform: body.platform }) + sendJson(res, 200, { deviceId: dev.id, key: dev.secret, master: { id: self.id, name: self.name } }) + } catch (e) { + sendJson(res, e.status || 500, { error: String(e.message || e) }) + } + }, + }), 'dsh-remote: enroll') + + route('/dsh-remote/pair', ['POST'], async (req) => { + const body = await readJson(req) + const masterUrl = String(body.masterUrl || '').trim().replace(/\/+$/, '') + if (!/^https?:\/\//.test(masterUrl)) throw httpErr(400, 'masterUrl must start with http:// or https://') + const r = await fetch(`${masterUrl}/dsh-remote/enroll`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: body.code, id: self.id, name: self.name, platform: `${process.platform}-${process.arch}` }), + }) + const j = await r.json().catch(() => ({})) + if (!r.ok) throw httpErr(r.status, j.error || `enroll failed (HTTP ${r.status})`) + store.setMaster({ url: masterUrl, deviceId: j.deviceId, key: j.key, master: j.master, pairedAt: Date.now() }) + return { ok: true, deviceId: j.deviceId, master: j.master, restartRequired: role !== 'slave' } + }) + + route('/dsh-remote/unpair', ['POST'], async () => { + store.setMaster(null) + masterLink?.mux?.close(1000, 'unpaired') + return { ok: true } + }) + + ctx.effect(() => ctx.webServer.register({ + kind: 'prefix', + path: '/dsh-remote/devices', + handler: async (req, res) => { + const rejection = ctx.connection.requestRejection(req) + if (rejection !== undefined) return sendJson(res, rejection, { error: 'rejected' }) + try { + if (role !== 'master') throw httpErr(400, 'devices are managed on the master') + if (req.method !== 'POST') throw httpErr(405, 'POST only') + const m = new URL(req.url || '/', 'http://x').pathname.match(/^\/dsh-remote\/devices\/([a-z0-9-]+)\/(revoke|remove|sync)$/) + if (!m) throw httpErr(404, 'not found') + const [, id, action] = m + if (action === 'revoke') { store.revokeDevice(id); links.get(id)?.mux.close(4401, 'revoked') } + else if (action === 'remove') { store.removeDevice(id); links.get(id)?.mux.close(4401, 'removed') } + else { + const dev = store.devices()[id] + if (!dev) throw httpErr(404, 'unknown device') + const body = await readJson(req) + const asList = (v) => (Array.isArray(v) ? v.map(String).filter(Boolean) : []) + dev.sync = { enabled: body.enabled !== false, allow: asList(body.allow), deny: asList(body.deny), secrets: asList(body.secrets) } + store.save() + links.get(id)?.mux.event('sync-available', {}) + } + pushPeers() + sendJson(res, 200, { ok: true, devices: devicesView() }) + } catch (e) { + sendJson(res, e.status || 500, { error: String(e.message || e) }) + } + }, + }), 'dsh-remote: devices') + + route('/dsh-remote/sync', ['POST'], async () => { + if (role === 'master') { + manifestCache = null + for (const l of links.values()) l.mux.event('sync-available', {}) + return { ok: true, pushed: links.size } + } + return runSlaveSync() + }) + + route('/dsh-remote/sync/status', ['GET'], async () => { + if (role === 'master') { + const m = await buildManifestCached() + return { role, bundles: m.bundles, patchIds: m.patch.map((e) => e.id || (e.insert ? 'insert' : '?')), patchError: m.patchError || null, offerSecrets: m.secrets } + } + return { role, lastSync, synced: store.data.synced || {} } + }) + + route('/dsh-remote/new-session', ['POST'], async (req) => { + const body = await readJson(req) + const device = String(body.device || self.id) + const r = await callDevice(device, 'newSession', { cwd: body.cwd, workspaceId: body.workspaceId }) + return { device, sessionId: r.sessionId, url: device === self.id ? `./#/session/${r.sessionId}` : `./r/${device}/#/session/${r.sessionId}` } + }) + + // Server-sent events for the client: devices changes, activity, sync results. + route('/dsh-remote/events', ['GET'], async () => ({})) + ctx.effect(() => { + const dispose = ctx.webServer.register({ + kind: 'exact', + path: '/dsh-remote/stream', + handler: (req, res) => { + const rejection = ctx.connection.requestRejection(req) + if (rejection !== undefined) { res.statusCode = rejection; return res.end() } + res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-store', connection: 'keep-alive', 'x-accel-buffering': 'no' }) + res.write(': ok\n\n') + const l = (ev) => res.write(`data: ${JSON.stringify(ev)}\n\n`) + listeners.add(l) + const ka = setInterval(() => res.write(': ka\n\n'), 20000) + req.on('close', () => { listeners.delete(l); clearInterval(ka) }) + }, + }) + return dispose + }, 'dsh-remote: event stream') + + // ------------------------------------------------------------------ /r/<device>/ proxy + function parseRemote(urlPath) { + const m = urlPath.match(/^\/r\/([a-z0-9-]+)(\/.*)?$/) + if (!m) return null + return { device: m[1], rest: m[2] || '/' } + } + + ctx.effect(() => ctx.webServer.register({ + kind: 'prefix', + path: '/r', + handler: (req, res) => { + const rejection = ctx.connection.requestRejection(req) + if (rejection !== undefined) { + res.writeHead(rejection, { 'content-type': 'text/plain; charset=utf-8' }) + return res.end(rejection === 401 ? 'dsh-remote: open this DSH first (authentication required)\n' : 'forbidden\n') + } + const url = new URL(req.url || '/', 'http://x') + const p = parseRemote(url.pathname) + if (!p) { res.statusCode = 404; return res.end() } + // "/r/ai-pc" → "/r/ai-pc/" so relative URLs (<base href="./">) resolve under the mount. + if (url.pathname === `/r/${p.device}`) { + res.writeHead(301, { location: `/r/${p.device}/${url.search}${url.hash}` }) + return res.end() + } + if (p.device === self.id) { + res.writeHead(302, { location: `/${url.search}` }) + return res.end() + } + const r = routeTo(p.device) + if (!r) { + res.writeHead(502, { 'content-type': 'text/html; charset=utf-8', 'cache-control': 'no-store' }) + return res.end(`<!doctype html><meta charset=utf-8><title>offline +

Device ${p.device} is offline

Its DSH is not linked right now. Back

`) + } + viewHttp(req, res, (kind, meta) => r.mux.open(kind, { ...meta, ...r.meta }), { rest: p.rest + url.search, basePath: `/r/${p.device}` }) + }, + }), 'dsh-remote: /r proxy') + + // Upgrades: DSH's webserver matches upgrade routes exactly, so register the Remote mux + // path for each known device id (and re-register as devices appear). + const upgradeRoutes = new Map() + function ensureUpgradeRoute(device) { + if (device === self.id || upgradeRoutes.has(device)) return + const path = `/r/${device}${MUX_PATH}` + try { + const dispose = ctx.webServer.registerUpgrade({ + path, + handler: (req, socket, head) => { + if (ctx.connection.requestRejection(req) !== undefined) return rejectUpgrade(socket, 401, 'auth required') + const r = routeTo(device) + if (!r) return rejectUpgrade(socket, 502, 'device offline') + const url = new URL(req.url || '/', 'http://x') + viewUpgrade(req, socket, head, (kind, meta) => r.mux.open(kind, { ...meta, ...r.meta }), { rest: MUX_PATH + url.search }) + }, + }) + upgradeRoutes.set(device, dispose) + } catch (e) { + log.warn(`dsh-remote: upgrade route ${path}: ${e.message}`) + } + } + const upgradeSync = setInterval(() => { for (const d of devicesView()) ensureUpgradeRoute(d.id) }, 2000) + upgradeSync.unref?.() + for (const d of Object.keys(store.devices())) ensureUpgradeRoute(d) + ctx.effect(() => () => { + clearInterval(upgradeSync) + for (const dispose of upgradeRoutes.values()) { try { dispose() } catch { /* ignore */ } } + upgradeRoutes.clear() + }, 'dsh-remote: upgrade routes') + + log.info?.(`dsh-remote: ${role} "${self.name}" (${self.id}) ready`) +} diff --git a/lib/auth-cookie.mjs b/lib/auth-cookie.mjs new file mode 100644 index 0000000..0ec8255 --- /dev/null +++ b/lib/auth-cookie.mjs @@ -0,0 +1,58 @@ +// Mint DSH's own browser-session cookie for a given authority. +// +// Format (dsh-client-connection BrowserAuth): cookie name `dsh-auth-` + +// base64url(sha256(authority)), value `v1..` +// with json {version:1, authority, issuedAt, expiresAt}. The 32-byte secret is the +// credential record `client-connection/browser-session` (kind grant, payload.secret). +// +// The target DSH accepts proxied requests only when Host matches the cookie's authority, +// so the proxy rewrites Host to the loopback authority it forwards to and attaches a +// cookie minted for exactly that authority. The browser never sees this cookie. +import crypto from 'node:crypto' + +const DAY = 86400000 +const b64u = (b) => Buffer.from(b).toString('base64url') + +export function cookieName(authority) { + return 'dsh-auth-' + b64u(crypto.createHash('sha256').update(authority).digest()) +} + +export function mintCookie(authority, secretB64u, maxAgeDays = 1) { + const secret = Buffer.from(String(secretB64u || ''), 'base64url') + if (secret.length !== 32) return null + const issuedAt = Date.now() + const body = b64u(JSON.stringify({ version: 1, authority, issuedAt, expiresAt: issuedAt + maxAgeDays * DAY })) + const sig = b64u(crypto.createHmac('sha256', secret).update(body).digest()) + return `${cookieName(authority)}=v1.${body}.${sig}` +} + +/** Read the browser-session signing secret through ctx.credentials (no file access). */ +export async function readBrowserSecret(ctx) { + let creds + try { creds = typeof ctx.get === 'function' ? ctx.get('credentials') : ctx.credentials } catch { creds = undefined } + if (!creds || typeof creds.readRecord !== 'function') return '' + try { + const rec = await creds.readRecord('client-connection/browser-session') + return rec?.payload?.secret || '' + } catch { + return '' + } +} + +/** Small cache: re-mint a cookie per authority once an hour. */ +export class CookieJar { + constructor(ctx) { + this.ctx = ctx + this.cache = new Map() + this.secret = '' + } + + async get(authority) { + const hit = this.cache.get(authority) + if (hit && hit.at > Date.now() - 3600000) return hit.cookie + if (!this.secret) this.secret = await readBrowserSecret(this.ctx) + const cookie = mintCookie(authority, this.secret, 1) + if (cookie) this.cache.set(authority, { cookie, at: Date.now() }) + return cookie + } +} diff --git a/lib/mux.mjs b/lib/mux.mjs new file mode 100644 index 0000000..67b01ae --- /dev/null +++ b/lib/mux.mjs @@ -0,0 +1,245 @@ +// Stream multiplexer over one WsConn. +// +// Frames: +// text = JSON control message { t: , ...fields } +// binary = data frame [u32 stream id][payload] +// +// Streams are opened with a control message `{t:'open', id, kind, meta}` and carry raw +// bytes in both directions. They implement backpressure-lite: a stream pauses its local +// source while the underlying socket buffer is above HIGH_WATER. +// +// Request/response control calls: `{t:'call', cid, method, params}` → +// `{t:'reply', cid, ok, result | error}`. Unsolicited `{t:'event', name, data}`. +import { EventEmitter } from 'node:events' + +const HIGH_WATER = 8 * 1024 * 1024 + +export class Mux extends EventEmitter { + /** + * @param ws WsConn + * @param {{ odd: boolean, handlers?: Record, onStream?: Function }} opts + * odd: this side allocates odd stream ids (master = even, slave = odd) so ids never collide. + */ + constructor(ws, { odd, handlers = {}, onStream } = {}) { + super() + this.ws = ws + this.nextId = odd ? 1 : 2 + this.nextCid = 1 + this.streams = new Map() + this.calls = new Map() + this.handlers = handlers + this.onStream = onStream + this.closed = false + this.lastSeen = Date.now() + ws.on('message', (data, binary) => { + this.lastSeen = Date.now() + if (binary) this._data(data) + else this._control(data) + }) + ws.on('pong', () => { this.lastSeen = Date.now() }) + ws.on('close', (code, reason) => this._teardown(code, reason)) + this.pinger = setInterval(() => { + if (Date.now() - this.lastSeen > 45000) return ws.destroy(1001, 'ping timeout') + ws.ping() + }, 15000) + this.pinger.unref?.() + } + + _teardown(code, reason) { + if (this.closed) return + this.closed = true + clearInterval(this.pinger) + for (const s of this.streams.values()) s._remoteEnd(new Error('link closed')) + this.streams.clear() + for (const { reject, timer } of this.calls.values()) { clearTimeout(timer); reject(new Error('link closed')) } + this.calls.clear() + this.emit('close', code, reason) + } + + close(code, reason) { this.ws.close(code, reason) } + + sendControl(msg) { + if (this.closed) return false + return this.ws.send(JSON.stringify(msg)) + } + + _control(text) { + let msg + try { msg = JSON.parse(text) } catch { return } + switch (msg.t) { + case 'open': { + const s = new Stream(this, msg.id, msg.kind, msg.meta || {}) + this.streams.set(msg.id, s) + Promise.resolve() + .then(() => (this.onStream ? this.onStream(s) : s.fail('no stream handler'))) + .catch((e) => s.fail(String(e?.message ?? e))) + return + } + case 'head': return this.streams.get(msg.id)?._head(msg.meta || {}) + case 'end': return this.streams.get(msg.id)?._remoteEnd() + case 'fail': return this.streams.get(msg.id)?._remoteEnd(new Error(msg.error || 'stream failed')) + case 'pause': return this.streams.get(msg.id)?.emit('remote-pause') + case 'resume': return this.streams.get(msg.id)?.emit('remote-resume') + case 'call': return this._call(msg) + case 'reply': { + const c = this.calls.get(msg.cid) + if (!c) return + this.calls.delete(msg.cid) + clearTimeout(c.timer) + if (msg.ok) c.resolve(msg.result) + else c.reject(Object.assign(new Error(msg.error || 'remote error'), { status: msg.status })) + return + } + case 'event': return this.emit('event', msg.name, msg.data) + default: return this.emit('control', msg) + } + } + + async _call(msg) { + const h = this.handlers[msg.method] + try { + if (!h) throw Object.assign(new Error(`unknown method ${msg.method}`), { status: 404 }) + const result = await h(msg.params ?? {}, this) + this.sendControl({ t: 'reply', cid: msg.cid, ok: true, result: result ?? null }) + } catch (e) { + this.sendControl({ t: 'reply', cid: msg.cid, ok: false, error: String(e?.message ?? e), status: e?.status }) + } + } + + call(method, params = {}, timeoutMs = 30000) { + if (this.closed) return Promise.reject(new Error('link closed')) + const cid = this.nextCid++ + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.calls.delete(cid) + reject(new Error(`call ${method} timed out`)) + }, timeoutMs) + timer.unref?.() + this.calls.set(cid, { resolve, reject, timer }) + this.sendControl({ t: 'call', cid, method, params }) + }) + } + + event(name, data) { this.sendControl({ t: 'event', name, data }) } + + _data(buf) { + if (buf.length < 4) return + const id = buf.readUInt32BE(0) + this.streams.get(id)?._remoteData(buf.subarray(4)) + } + + /** Open an outgoing stream. */ + open(kind, meta = {}) { + const id = this.nextId + this.nextId += 2 + const s = new Stream(this, id, kind, meta) + this.streams.set(id, s) + this.sendControl({ t: 'open', id, kind, meta }) + // The opener attaches listeners synchronously; deliver from the next tick on. + queueMicrotask(() => s.start()) + return s + } + + _writeData(id, chunk) { + const header = Buffer.alloc(4) + header.writeUInt32BE(id, 0) + // Split large chunks so one stream cannot monopolise the link. + const MAX = 256 * 1024 + let ok = true + for (let off = 0; off < chunk.length; off += MAX) { + ok = this.ws.send(Buffer.concat([header, chunk.subarray(off, off + MAX)])) && ok + } + return ok && this.ws.bufferedAmount < HIGH_WATER + } +} + +/** + * One bidirectional byte stream. Events: 'head'(meta), 'data'(buf), 'end', 'error'(err), + * 'drain'. Writers should stop on write()===false and wait for 'drain'. + */ +export class Stream extends EventEmitter { + constructor(mux, id, kind, meta) { + super() + this.mux = mux + this.id = id + this.kind = kind + this.meta = meta + this.localEnded = false + this.remoteEnded = false + this.done = false + // Incoming events are queued until the consumer attaches its listeners and calls + // start(): handlers often await something (cookie, upstream connect) first, and + // data/end arriving in the meantime must not be lost. + this.paused = true + this.queue = [] + } + + /** Begin delivering queued + future incoming events. Idempotent. */ + start() { + if (!this.paused) return this + this.paused = false + const q = this.queue + this.queue = [] + for (const [ev, arg] of q) this._deliver(ev, arg) + return this + } + + _deliver(ev, arg) { + if (this.paused) { this.queue.push([ev, arg]); return } + if (ev === 'head') this.emit('head', arg) + else if (ev === 'data') this.emit('data', arg) + else if (ev === 'end') { this.emit('end'); this._maybeDone() } + else if (ev === 'error') { if (this.listenerCount('error')) this.emit('error', arg); this._finish() } + } + + head(meta) { this.mux.sendControl({ t: 'head', id: this.id, meta }) } + + write(chunk) { + if (this.localEnded || this.mux.closed) return false + const ok = this.mux._writeData(this.id, Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + if (!ok && !this._drainWait) { + this._drainWait = true + const sock = this.mux.ws.socket + sock.once('drain', () => { this._drainWait = false; this.emit('drain') }) + } + return ok + } + + end() { + if (this.localEnded) return + this.localEnded = true + this.mux.sendControl({ t: 'end', id: this.id }) + this._maybeDone() + } + + fail(error) { + if (this.done) return + this.mux.sendControl({ t: 'fail', id: this.id, error: String(error) }) + this.localEnded = true + this.remoteEnded = true + this._finish() + } + + _head(meta) { this._deliver('head', meta) } + _remoteData(buf) { if (!this.remoteEnded) this._deliver('data', buf) } + + _remoteEnd(err) { + if (this.remoteEnded) return + this.remoteEnded = true + if (err) { + this.localEnded = true + this._deliver('error', err) + return + } + this._deliver('end') + } + + _maybeDone() { if (this.localEnded && this.remoteEnded) this._finish() } + + _finish() { + if (this.done) return + this.done = true + this.mux.streams.delete(this.id) + this.emit('close') + } +} diff --git a/lib/proxy.mjs b/lib/proxy.mjs new file mode 100644 index 0000000..ca2e4f9 --- /dev/null +++ b/lib/proxy.mjs @@ -0,0 +1,199 @@ +// HTTP + WebSocket tunnelling between a browser-facing webserver and a remote DSH over +// the link mux. +// +// viewer side (where the browser is): /r// ──stream "http"|"ws"──► +// owner side (where the session runs): forward to 127.0.0.1:/ +// +// Owner side rewrites Host/Origin to its own loopback authority and attaches a freshly +// minted DSH auth cookie, so DSH's /api trust fence and cookie check pass. The viewer's +// own DSH cookie is stripped before forwarding (it would be meaningless remotely). +import http from 'node:http' + +const HOP = new Set([ + 'connection', 'keep-alive', 'proxy-connection', 'transfer-encoding', 'upgrade', 'te', 'trailer', + 'proxy-authorization', 'proxy-authenticate', 'host', 'content-length', +]) + +export function stripDshCookies(cookieHeader) { + return String(cookieHeader || '') + .split(';') + .map((s) => s.trim()) + .filter((s) => s && !/^dsh-auth-/.test(s) && !/^dsh_auth_session=/.test(s)) + .join('; ') +} + +function forwardHeaders(headers) { + const out = {} + for (const [k, v] of Object.entries(headers || {})) { + if (HOP.has(k.toLowerCase())) continue + out[k] = v + } + return out +} + +// --------------------------------------------------------------------------- +// viewer side +// --------------------------------------------------------------------------- + +/** + * Pipe one browser HTTP request through `openStream()` (returns a mux Stream). + * `rest` is the path+query to request on the remote side ("/api/remote/…"). + * `basePath` is the local mount ("/r/ai-pc") used to rewrite redirects/cookies. + */ +export function viewHttp(req, res, openStream, { rest, basePath, extra = {} }) { + const headers = forwardHeaders(req.headers) + headers.cookie = stripDshCookies(req.headers.cookie) + if (!headers.cookie) delete headers.cookie + const stream = openStream('http', { method: req.method, path: rest, headers, ...extra }) + let responded = false + + stream.on('head', (meta) => { + responded = true + const outHeaders = { ...meta.headers } + // Keep relative redirects inside the mount (DSH answers "./" which is already relative). + if (typeof outHeaders.location === 'string' && outHeaders.location.startsWith('/')) { + outHeaders.location = basePath + outHeaders.location + } + // The remote never sets cookies the browser should keep under our authority. + delete outHeaders['set-cookie'] + try { res.writeHead(meta.status || 502, outHeaders) } catch { /* headers already sent */ } + }) + stream.on('data', (chunk) => { + if (!res.write(chunk)) { + stream.mux.sendControl({ t: 'pause', id: stream.id }) + res.once('drain', () => stream.mux.sendControl({ t: 'resume', id: stream.id })) + } + }) + stream.on('end', () => res.end()) + stream.on('error', (err) => { + if (!responded) { + res.writeHead(502, { 'content-type': 'text/plain; charset=utf-8', 'cache-control': 'no-store' }) + res.end(`dsh-remote: ${err.message}\n`) + } else { + res.destroy() + } + }) + res.on('close', () => { if (!stream.done) stream.fail('client closed') }) + + req.on('data', (chunk) => { + if (!stream.write(chunk)) { + req.pause() + stream.once('drain', () => req.resume()) + } + }) + req.on('end', () => stream.end()) + req.on('error', () => stream.fail('request error')) +} + +/** Tunnel a browser WebSocket upgrade (raw socket) through a mux stream. */ +export function viewUpgrade(req, socket, head, openStream, { rest, extra = {} }) { + const headers = forwardHeaders(req.headers) + // Keep the WebSocket handshake headers (forwardHeaders dropped connection/upgrade). + headers.cookie = stripDshCookies(req.headers.cookie) + if (!headers.cookie) delete headers.cookie + const stream = openStream('ws', { path: rest, headers, ...extra }) + let open = false + stream.on('head', (meta) => { + // The owner side already completed the upstream handshake; replay its 101 to the + // browser with the browser's own Sec-WebSocket-Key accepted value. + if (meta.status !== 101) { + socket.end(`HTTP/1.1 ${meta.status || 502} Upgrade Failed\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`) + return + } + open = true + const lines = ['HTTP/1.1 101 Switching Protocols'] + for (const [k, v] of Object.entries(meta.headers || {})) { + for (const one of Array.isArray(v) ? v : [v]) lines.push(`${k}: ${one}`) + } + socket.write(lines.join('\r\n') + '\r\n\r\n') + if (head && head.length) stream.write(head) + }) + stream.on('data', (chunk) => { if (!socket.write(chunk)) { /* tcp backpressure handled by kernel */ } }) + stream.on('end', () => socket.end()) + stream.on('error', () => { + if (!open) { + try { socket.end('HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\nContent-Length: 0\r\n\r\n') } catch { socket.destroy() } + } else socket.destroy() + }) + socket.on('data', (chunk) => stream.write(chunk)) + socket.on('end', () => stream.end()) + socket.on('close', () => { if (!stream.done) stream.fail('client closed') }) + socket.on('error', () => { if (!stream.done) stream.fail('client error') }) +} + +// --------------------------------------------------------------------------- +// owner side +// --------------------------------------------------------------------------- + +/** + * Serve one incoming "http" stream against the local DSH. + * @param target {host, port}; authority = `${host}:${port}` + * @param cookie async () => "dsh-auth-…=v1…" for that authority + */ +export async function ownHttp(stream, { target, cookie }) { + const { method = 'GET', path = '/', headers = {} } = stream.meta + const authority = `${target.host}:${target.port}` + const h = { ...headers, host: authority } + if (h.origin) h.origin = `http://${authority}` + if (h.referer) h.referer = `http://${authority}/` + delete h['sec-fetch-site'] + const c = await cookie(authority) + h.cookie = [h.cookie, c].filter(Boolean).join('; ') + + const req = http.request({ host: target.host, port: target.port, method, path, headers: h }) + req.on('socket', (s) => s.on('error', () => {})) + req.on('response', (res) => { + stream.head({ status: res.statusCode, headers: forwardHeaders(res.headers) }) + res.on('data', (chunk) => { + if (!stream.write(chunk)) { + res.pause() + stream.once('drain', () => res.resume()) + } + }) + res.on('end', () => stream.end()) + res.on('error', () => stream.fail('upstream response error')) + stream.on('remote-pause', () => res.pause()) + stream.on('remote-resume', () => res.resume()) + }) + req.on('error', (err) => stream.fail(`upstream: ${err.message}`)) + stream.on('data', (chunk) => req.write(chunk)) + stream.on('end', () => req.end()) + stream.on('error', () => req.destroy()) + stream.start() +} + +/** Serve one incoming "ws" stream: perform the upgrade against local DSH, then pipe. */ +export async function ownUpgrade(stream, { target, cookie }) { + const { path = '/', headers = {} } = stream.meta + const authority = `${target.host}:${target.port}` + const h = { ...headers, host: authority, connection: 'Upgrade', upgrade: headers.upgrade || 'websocket' } + if (h.origin) h.origin = `http://${authority}` + delete h['sec-fetch-site'] + const c = await cookie(authority) + h.cookie = [h.cookie, c].filter(Boolean).join('; ') + + const req = http.request({ host: target.host, port: target.port, method: 'GET', path, headers: h }) + req.on('socket', (s) => s.on('error', () => {})) + req.on('upgrade', (res, sock, head) => { + const out = {} + for (const [k, v] of Object.entries(res.headers)) out[k] = v + stream.head({ status: 101, headers: out }) + if (head && head.length) stream.write(head) + sock.on('data', (chunk) => stream.write(chunk)) + sock.on('end', () => stream.end()) + sock.on('error', () => stream.fail('upstream socket error')) + stream.on('data', (chunk) => sock.write(chunk)) + stream.on('end', () => sock.end()) + stream.on('error', () => sock.destroy()) + stream.on('close', () => sock.destroy()) + // Browser frames sent before the upstream handshake finished were queued; flush now. + stream.start() + }) + req.on('response', (res) => { + stream.head({ status: res.statusCode || 502, headers: {} }) + res.resume() + stream.end() + }) + req.on('error', (err) => stream.fail(`upstream: ${err.message}`)) + req.end() +} diff --git a/lib/store.mjs b/lib/store.mjs new file mode 100644 index 0000000..8599427 --- /dev/null +++ b/lib/store.mjs @@ -0,0 +1,150 @@ +// Persistent state for dsh-remote: device identity, paired devices (master) or master +// link credentials (slave), sync bookkeeping. One JSON file, written atomically, 0600. +import crypto from 'node:crypto' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' + +let PROFILE_DIR = '' + +/** Set once from ctx.profileContext.dir at plugin start (authoritative). */ +export function setProfileDir(dir) { if (dir) PROFILE_DIR = dir } + +export function profileDir() { + if (PROFILE_DIR) return PROFILE_DIR + if (process.env.DSH_PROFILE_DIR) return process.env.DSH_PROFILE_DIR + const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') + return path.join(home, 'profiles', process.env.DSH_PROFILE || 'web') +} + +export function stateDir() { + return path.join(profileDir(), 'dsh-remote') +} + +const b64u = (buf) => Buffer.from(buf).toString('base64url') +export const randomId = (bytes = 16) => b64u(crypto.randomBytes(bytes)) +export const sha256 = (s) => crypto.createHash('sha256').update(s).digest('hex') + +export function safeEqual(a, b) { + const x = Buffer.from(String(a)) + const y = Buffer.from(String(b)) + return x.length === y.length && crypto.timingSafeEqual(x, y) +} + +/** Device id: lowercase slug, safe in URL paths (/r//). */ +export function slug(s) { + const v = String(s || '').toLowerCase().replace(/[^a-z0-9-]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 40) + return v || 'device' +} + +export class Store { + constructor(file = path.join(stateDir(), 'state.json')) { + this.file = file + this.data = this._load() + } + + _load() { + try { + const v = JSON.parse(fs.readFileSync(this.file, 'utf8')) + if (v && typeof v === 'object') return v + } catch { /* first run */ } + return {} + } + + save() { + fs.mkdirSync(path.dirname(this.file), { recursive: true, mode: 0o700 }) + const tmp = `${this.file}.${process.pid}.${Date.now()}.tmp` + fs.writeFileSync(tmp, JSON.stringify(this.data, null, 2), { mode: 0o600 }) + fs.renameSync(tmp, this.file) + } + + /** This installation's stable identity. */ + identity(defaultName) { + if (!this.data.self) { + const name = defaultName || os.hostname() + this.data.self = { id: slug(name), name, createdAt: Date.now() } + this.save() + } + return this.data.self + } + + // ---- master side: paired devices + pairing codes ------------------------- + devices() { return (this.data.devices ||= {}) } + + pairingCodes() { return (this.data.pairing ||= {}) } + + createPairingCode({ ttlMs = 15 * 60 * 1000, label = '' } = {}) { + // 4 groups of 4 from an unambiguous alphabet: easy to type, ~80 bits. + const alphabet = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789' + const raw = crypto.randomBytes(16) + const code = [...raw].map((b) => alphabet[b % alphabet.length]).join('').replace(/(.{4})(?=.)/g, '$1-') + this.pairingCodes()[sha256(code)] = { expiresAt: Date.now() + ttlMs, label, createdAt: Date.now() } + this._prunePairing() + this.save() + return { code, expiresAt: Date.now() + ttlMs } + } + + consumePairingCode(code) { + this._prunePairing() + const key = sha256(String(code || '').trim().toUpperCase()) + const rec = this.pairingCodes()[key] + if (!rec) return null + delete this.pairingCodes()[key] + this.save() + return rec + } + + _prunePairing() { + const now = Date.now() + for (const [k, v] of Object.entries(this.pairingCodes())) if (!v || v.expiresAt < now) delete this.pairingCodes()[k] + } + + /** Register (or re-register) a device; returns its secret key (shown once). */ + addDevice({ id, name, platform }) { + const devices = this.devices() + let devId = slug(id || name) + // Keep ids unique; a re-pair of the same machine replaces its key. + const secret = randomId(32) + devices[devId] = { + id: devId, + name: name || devId, + platform: platform || '', + keyHash: sha256(secret), + createdAt: devices[devId]?.createdAt || Date.now(), + revoked: false, + sync: devices[devId]?.sync || { enabled: true, allow: [], deny: [], secrets: [] }, + } + this.save() + return { id: devId, secret } + } + + verifyDevice(id, secret) { + const d = this.devices()[id] + if (!d || d.revoked) return null + return safeEqual(d.keyHash, sha256(String(secret || ''))) ? d : null + } + + revokeDevice(id) { + const d = this.devices()[id] + if (!d) return false + d.revoked = true + d.revokedAt = Date.now() + this.save() + return true + } + + removeDevice(id) { + if (!this.devices()[id]) return false + delete this.devices()[id] + this.save() + return true + } + + // ---- slave side: link to master ------------------------------------------ + get master() { return this.data.master || null } + + setMaster(m) { + this.data.master = m + this.save() + } +} diff --git a/lib/sync.mjs b/lib/sync.mjs new file mode 100644 index 0000000..01030c6 --- /dev/null +++ b/lib/sync.mjs @@ -0,0 +1,308 @@ +// Plugin + config sync (master → slaves). +// +// Manifest (built on the master, served over the link): +// { +// version, builtAt, master, +// bundles: [{ name, spec, kind: 'registry'|'local', version, hash }], +// patch: [ ...cordis patch entries from the master's cordis.patch.yml selected for sync ], +// secrets: [ names of credential refs the master offers for opt-in sync ] +// } +// +// Slave application: +// * registry bundles -> ctx.pluginManager.installBundle(`${name}@${version}`) unless already +// installed at that version, or excluded by the slave's deny list. +// * local (link:) bundles -> fetched as a tarball over the link, unpacked to +// /dsh-remote/plugins//, installed as `file:` spec. +// * patch entries -> written to /dsh-remote/managed-bundle/cordis.patch.yml, +// a bundle the slave enables BEFORE its own cordis.patch.yml, so +// anything the slave sets locally keeps precedence. +// * secrets -> only the names the slave explicitly opted into are requested; +// values travel over the authenticated link and are written with +// ctx.credentials.set(ref, value). +import crypto from 'node:crypto' +import { execFile } from 'node:child_process' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { profileDir, stateDir } from './store.mjs' + +export const MANAGED_BUNDLE = '@dsh-remote/managed' + +/** Patch entry ids never synced: they are machine-local by nature. */ +const LOCAL_ONLY_IDS = new Set(['dsh-remote', 'dsh-lanmode', 'host-webserver', 'webserver', 'client-connection']) + +function run(cmd, args, opts = {}) { + return new Promise((resolve) => { + execFile(cmd, args, { maxBuffer: 64 * 1024 * 1024, ...opts }, (err, stdout, stderr) => { + resolve({ code: err ? (err.code ?? 1) : 0, stdout: String(stdout), stderr: String(stderr) }) + }) + }) +} + +/** + * DSH's own entry-list YAML dialect (JSON schema + `!!js` expressions), loaded from the + * running installation so synced patches keep `!!js` nodes. Over the wire a `!!js` node is + * `{__jsExpr: ""}`; dumping with this schema writes it back as `!!js`. + */ +let yamlCache = null +async function loadYaml() { + if (yamlCache) return yamlCache + // Resolve DSH's own copies (not whatever the plugin dir could see): the dsh install's + // node_modules, found by following the real path of the running `dsh` bin. + const nm = dshNodeModules() + if (!nm) throw new Error('cannot locate the DSH installation (node_modules)') + const incPkg = JSON.parse(fs.readFileSync(path.join(nm, '@deepseek-ai', 'cordis-plugin-include', 'package.json'), 'utf8')) + const incMain = typeof incPkg.exports === 'string' ? incPkg.exports + : incPkg.exports?.['.']?.default || incPkg.exports?.['.']?.import || incPkg.exports?.['.'] || incPkg.main || 'lib/index.js' + const inc = await import(path.join(nm, '@deepseek-ai', 'cordis-plugin-include', String(incMain))) + const yaml = await import(path.join(nm, 'js-yaml', 'dist', 'js-yaml.mjs')) + const y = yaml.default || yaml + yamlCache = { + load: (s) => y.load(s, { schema: inc.entryListSchema }), + dump: (v, o = {}) => y.dump(v, { ...o, schema: inc.entryListSchema }), + } + return yamlCache +} + +export { loadYaml } + +function readJson(file) { + try { return JSON.parse(fs.readFileSync(file, 'utf8')) } catch { return null } +} + +function dirHash(dir) { + const h = crypto.createHash('sha256') + const walk = (d) => { + for (const ent of fs.readdirSync(d, { withFileTypes: true }).sort((a, b) => a.name.localeCompare(b.name))) { + if (ent.name === 'node_modules' || ent.name === '.git' || ent.name.startsWith('.')) continue + const p = path.join(d, ent.name) + if (ent.isDirectory()) walk(p) + else if (ent.isFile()) { h.update(path.relative(dir, p)); h.update(fs.readFileSync(p)) } + } + } + walk(dir) + return h.digest('hex').slice(0, 16) +} + +// --------------------------------------------------------------------------- +// master side +// --------------------------------------------------------------------------- + +export async function buildManifest({ self, include = {}, offerSecrets = [] }) { + const prof = profileDir() + const pkg = readJson(path.join(prof, 'package.json')) || {} + const deps = pkg.dependencies || {} + const enabled = new Set(pkg.dsh?.profile?.bundles || []) + const excludeBundles = new Set(include.excludeBundles || []) + const bundles = [] + for (const [name, spec] of Object.entries(deps)) { + if (!enabled.has(name) || excludeBundles.has(name) || name === '@local/dsh-remote' || name === MANAGED_BUNDLE) continue + const s = String(spec) + if (s.startsWith('link:') || s.startsWith('file:')) { + const dir = path.resolve(prof, s.replace(/^(link|file):/, '')) + const m = readJson(path.join(dir, 'package.json')) + if (!m) continue + bundles.push({ name, kind: 'local', version: m.version || '0.0.0', hash: dirHash(dir), dir }) + } else { + const installed = readJson(path.join(prof, 'node_modules', name, 'package.json')) + bundles.push({ name, kind: 'registry', spec: s, version: installed?.version || s.replace(/^[\^~]/, '') }) + } + } + + // Patch entries: everything in the master's cordis.patch.yml except machine-local ids. + let patch = [] + let patchError = null + try { + const yaml = await loadYaml() + const doc = yaml.load(fs.readFileSync(path.join(prof, 'cordis.patch.yml'), 'utf8')) || [] + const excludeIds = new Set([...(include.excludePatchIds || []), ...LOCAL_ONLY_IDS]) + patch = (Array.isArray(doc) ? doc : []).filter((e) => e && typeof e === 'object' && !excludeIds.has(e.id)) + } catch (e) { + patch = [] + patchError = String(e?.message ?? e) + } + + return { + version: 1, + builtAt: Date.now(), + master: { id: self.id, name: self.name }, + bundles: bundles.map(({ dir, ...b }) => b), + patch, + patchError, + secrets: offerSecrets, + _dirs: Object.fromEntries(bundles.filter((b) => b.dir).map((b) => [b.name, b.dir])), + } +} + +/** Pack a local plugin dir into a tgz Buffer (excludes .git, node_modules). */ +export async function packLocal(dir) { + const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'dsh-remote-pack-')) + const out = path.join(tmp, 'pkg.tgz') + const r = await run('tar', ['--exclude=.git', '--exclude=node_modules', '--exclude=*.bak', '-czf', out, '-C', dir, '.']) + if (r.code !== 0) throw new Error(`tar failed: ${r.stderr.trim()}`) + const buf = fs.readFileSync(out) + fs.rmSync(tmp, { recursive: true, force: true }) + return buf +} + +// --------------------------------------------------------------------------- +// slave side +// --------------------------------------------------------------------------- + +export function syncDir() { return path.join(stateDir(), 'sync') } + +export function managedBundleDir() { return path.join(stateDir(), 'managed-bundle') } + +/** Write the managed bundle (package.json + patch) that carries synced config. */ +export async function writeManagedBundle(patch, insertedLocalPlugins = []) { + const dir = managedBundleDir() + fs.mkdirSync(dir, { recursive: true }) + const yaml = await loadYaml() + const header = '# Managed by dsh-remote: config synced from the master. Do not edit;\n' + + '# your own profile cordis.patch.yml is applied after this layer and wins.\n' + const body = patch.length ? yaml.dump(patch, { lineWidth: 200, noRefs: true }) : '[]\n' + fs.writeFileSync(path.join(dir, 'cordis.patch.yml'), header + body) + fs.writeFileSync(path.join(dir, 'package.json'), JSON.stringify({ + name: MANAGED_BUNDLE, + version: `0.0.${Date.now()}`, + private: true, + description: 'dsh-remote managed layer: plugins/config synced from the master', + dsh: { bundle: { patch: './cordis.patch.yml' } }, + dsh_remote: { localPlugins: insertedLocalPlugins }, + }, null, 2)) + return dir +} + +/** + * Create/refresh /node_modules/ -> dir. DSH resolves bundle packages + * from the profile's node_modules; a `link:` dependency alone is only materialised by a + * pnpm install, so make the link ourselves (idempotent, no network). + */ +export function linkIntoNodeModules(name, dir) { + const link = path.join(profileDir(), 'node_modules', ...name.split('/')) + fs.mkdirSync(path.dirname(link), { recursive: true }) + try { + if (fs.readlinkSync(link) === dir) return false + } catch { /* missing or not a link */ } + fs.rmSync(link, { recursive: true, force: true }) + fs.symlinkSync(dir, link) + return true +} + +/** Ensure the managed bundle is linked and listed last, i.e. right before the user's own patch. */ +export function ensureManagedBundleOrder() { + const prof = profileDir() + const file = path.join(prof, 'package.json') + const pkg = readJson(file) + if (!pkg) return false + pkg.dependencies ||= {} + const spec = `link:${managedBundleDir()}` + let changed = linkIntoNodeModules(MANAGED_BUNDLE, managedBundleDir()) + let pkgChanged = false + if (pkg.dependencies[MANAGED_BUNDLE] !== spec) { pkg.dependencies[MANAGED_BUNDLE] = spec; pkgChanged = true } + const bundles = (((pkg.dsh ||= {}).profile ||= {}).bundles ||= []) + const idx = bundles.indexOf(MANAGED_BUNDLE) + if (idx !== bundles.length - 1) { + if (idx !== -1) bundles.splice(idx, 1) + bundles.push(MANAGED_BUNDLE) + pkgChanged = true + } + if (pkgChanged) fs.writeFileSync(file, JSON.stringify(pkg, null, 2) + '\n') + return changed || pkgChanged +} + +export function unpackLocal(name, version, tgz) { + const safe = name.replace(/[^a-zA-Z0-9._-]+/g, '_') + const dir = path.join(syncDir(), 'plugins', safe) + const tmp = `${dir}.new` + fs.rmSync(tmp, { recursive: true, force: true }) + fs.mkdirSync(tmp, { recursive: true }) + const tgzFile = path.join(syncDir(), `${safe}.tgz`) + fs.writeFileSync(tgzFile, tgz) + return run('tar', ['-xzf', tgzFile, '-C', tmp]).then((r) => { + fs.rmSync(tgzFile, { force: true }) + if (r.code !== 0) throw new Error(`untar ${name}: ${r.stderr.trim()}`) + fs.rmSync(dir, { recursive: true, force: true }) + fs.renameSync(tmp, dir) + linkDshDeps(dir) + return dir + }) +} + +/** DSH installation's own node_modules (where @deepseek-ai/* packages live). */ +export function dshNodeModules() { + const bin = process.argv[1] || '' + let real = bin + try { real = fs.realpathSync(bin) } catch { /* keep argv path */ } + const cand = [ + path.resolve(path.dirname(real), '..', 'node_modules'), + path.resolve(path.dirname(bin), '..', 'lib', 'node_modules', '@deepseek-ai', 'dsh', 'node_modules'), + ] + for (const c of cand) if (fs.existsSync(path.join(c, '@deepseek-ai'))) return c + return null +} + +/** + * Synced local plugins import `@deepseek-ai/*` modules that the master resolved through + * a node_modules symlink into its DSH install. Recreate those links against this + * machine's DSH so the plugin loads unchanged. + */ +export function linkDshDeps(pluginDir) { + const nm = dshNodeModules() + if (!nm) return [] + const wanted = new Set() + const scan = (d, depth = 0) => { + if (depth > 3) return + for (const ent of fs.readdirSync(d, { withFileTypes: true })) { + if (ent.name === 'node_modules' || ent.name.startsWith('.')) continue + const p = path.join(d, ent.name) + if (ent.isDirectory()) scan(p, depth + 1) + else if (/\.(m?js|cjs)$/.test(ent.name)) { + const src = fs.readFileSync(p, 'utf8') + for (const m of src.matchAll(/(?:from\s+|import\s*\(\s*|require\(\s*)['"](@deepseek-ai\/[a-z0-9-]+)/g)) wanted.add(m[1]) + } + } + } + scan(pluginDir) + const linked = [] + for (const mod of wanted) { + const target = path.join(nm, mod) + if (!fs.existsSync(target)) continue + const link = path.join(pluginDir, 'node_modules', mod) + fs.mkdirSync(path.dirname(link), { recursive: true }) + try { fs.rmSync(link, { recursive: true, force: true }) } catch { /* ignore */ } + fs.symlinkSync(target, link) + linked.push(mod) + } + return linked +} + +/** Decide whether a bundle passes the slave's allow/deny lists. */ +export function allowed(name, { allow = [], deny = [] } = {}) { + const match = (pat) => pat === name || (pat.endsWith('*') && name.startsWith(pat.slice(0, -1))) + if (deny.some(match)) return false + if (allow.length && !allow.some(match)) return false + return true +} + +/** Link-spec install of a local plugin dir into the profile (package.json + bundles list). */ +export function linkLocalBundle(name, dir) { + const prof = profileDir() + const file = path.join(prof, 'package.json') + const pkg = readJson(file) + if (!pkg) throw new Error('profile package.json unreadable') + pkg.dependencies ||= {} + const spec = `link:${dir}` + let changed = false + if (pkg.dependencies[name] !== spec) { pkg.dependencies[name] = spec; changed = true } + const bundles = (((pkg.dsh ||= {}).profile ||= {}).bundles ||= []) + if (!bundles.includes(name)) { + // Synced plugins go before the managed layer so the master's config for them applies. + const at = bundles.indexOf(MANAGED_BUNDLE) + if (at === -1) bundles.push(name) + else bundles.splice(at, 0, name) + changed = true + } + if (changed) fs.writeFileSync(file, JSON.stringify(pkg, null, 2) + '\n') + return linkIntoNodeModules(name, dir) || changed +} diff --git a/lib/ws.mjs b/lib/ws.mjs new file mode 100644 index 0000000..7988943 --- /dev/null +++ b/lib/ws.mjs @@ -0,0 +1,215 @@ +// Minimal dependency-free WebSocket (RFC 6455) for the dsh-remote link. +// +// Why not `ws`: DSH plugins loaded through `link:` resolve imports from their own +// folder, and slaves install the plugin from a tarball without node_modules. The link +// only needs binary + text frames, ping/pong and close, so a small implementation keeps +// the plugin installable anywhere with zero dependencies. +import crypto from 'node:crypto' +import { EventEmitter } from 'node:events' +import http from 'node:http' +import https from 'node:https' + +const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11' +const MAX_FRAME = 64 * 1024 * 1024 + +export const acceptKey = (key) => crypto.createHash('sha1').update(key + GUID).digest('base64') + +/** One WebSocket endpoint over a raw socket (server or client role). */ +export class WsConn extends EventEmitter { + constructor(socket, { client = false, head } = {}) { + super() + this.socket = socket + this.client = client + this.open = true + this.buf = head && head.length ? Buffer.from(head) : Buffer.alloc(0) + this.frag = null + socket.setNoDelay(true) + socket.on('data', (d) => this._onData(d)) + socket.on('close', () => this._closed(1006, 'socket closed')) + // Never throw from an unhandled 'error' event: a reset peer must not crash the host. + socket.on('error', (e) => { if (this.listenerCount('error')) this.emit('error', e); this._closed(1006, String(e?.message ?? e)) }) + if (this.buf.length) queueMicrotask(() => this._onData(Buffer.alloc(0))) + } + + get bufferedAmount() { return this.socket.writableLength } + + _closed(code, reason) { + if (!this.open) return + this.open = false + this.emit('close', code, reason) + } + + _onData(chunk) { + if (chunk.length) this.buf = this.buf.length ? Buffer.concat([this.buf, chunk]) : chunk + for (;;) { + const b = this.buf + if (b.length < 2) return + const fin = (b[0] & 0x80) !== 0 + const op = b[0] & 0x0f + const masked = (b[1] & 0x80) !== 0 + let len = b[1] & 0x7f + let off = 2 + if (len === 126) { + if (b.length < 4) return + len = b.readUInt16BE(2); off = 4 + } else if (len === 127) { + if (b.length < 10) return + const big = b.readBigUInt64BE(2) + if (big > BigInt(MAX_FRAME)) return this.destroy(1009, 'frame too large') + len = Number(big); off = 10 + } + if (len > MAX_FRAME) return this.destroy(1009, 'frame too large') + const maskOff = off + if (masked) off += 4 + if (b.length < off + len) return + let payload = b.subarray(off, off + len) + if (masked) { + const mask = b.subarray(maskOff, maskOff + 4) + payload = Buffer.from(payload) + for (let i = 0; i < payload.length; i++) payload[i] ^= mask[i & 3] + } + this.buf = b.subarray(off + len) + this._frame(fin, op, payload) + if (!this.open) return + } + } + + _frame(fin, op, payload) { + if (op === 0x8) { + const code = payload.length >= 2 ? payload.readUInt16BE(0) : 1005 + try { this._send(0x8, payload.subarray(0, 2)) } catch {} + this.socket.end() + return this._closed(code, payload.subarray(2).toString('utf8')) + } + if (op === 0x9) return this._send(0xa, payload) + if (op === 0xa) return this.emit('pong', payload) + if (op === 0x0) { + if (!this.frag) return this.destroy(1002, 'unexpected continuation') + this.frag.parts.push(payload) + if (!fin) return + const data = Buffer.concat(this.frag.parts) + const binary = this.frag.op === 0x2 + this.frag = null + return this.emit('message', binary ? data : data.toString('utf8'), binary) + } + if (op === 0x1 || op === 0x2) { + if (!fin) { this.frag = { op, parts: [payload] }; return } + return this.emit('message', op === 0x2 ? Buffer.from(payload) : payload.toString('utf8'), op === 0x2) + } + this.destroy(1002, 'bad opcode') + } + + _send(op, data) { + if (!this.socket.writable) return false + const payload = Buffer.isBuffer(data) ? data : Buffer.from(data) + const len = payload.length + let header + if (len < 126) { header = Buffer.alloc(2); header[1] = len } + else if (len < 65536) { header = Buffer.alloc(4); header[1] = 126; header.writeUInt16BE(len, 2) } + else { header = Buffer.alloc(10); header[1] = 127; header.writeBigUInt64BE(BigInt(len), 2) } + header[0] = 0x80 | op + if (this.client) { + header[1] |= 0x80 + const mask = crypto.randomBytes(4) + const body = Buffer.from(payload) + for (let i = 0; i < body.length; i++) body[i] ^= mask[i & 3] + return this.socket.write(Buffer.concat([header, mask, body])) + } + return this.socket.write(Buffer.concat([header, payload])) + } + + send(data) { + if (!this.open) return false + return typeof data === 'string' ? this._send(0x1, Buffer.from(data, 'utf8')) : this._send(0x2, data) + } + + ping(data = Buffer.alloc(0)) { if (this.open) this._send(0x9, data) } + + close(code = 1000, reason = '') { + if (!this.open) return + const p = Buffer.alloc(2 + Buffer.byteLength(reason)) + p.writeUInt16BE(code, 0); p.write(reason, 2) + try { this._send(0x8, p) } catch {} + this.socket.end() + setTimeout(() => this.socket.destroy(), 2000).unref() + this._closed(code, reason) + } + + destroy(code = 1006, reason = '') { + this._closed(code, reason) + this.socket.destroy() + } +} + +/** Complete a server-side upgrade on (req, socket, head). Returns a WsConn or null. */ +export function acceptUpgrade(req, socket, head, extraHeaders = {}) { + const key = req.headers['sec-websocket-key'] + if (!key || String(req.headers.upgrade).toLowerCase() !== 'websocket') { + socket.end('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n') + return null + } + const lines = [ + 'HTTP/1.1 101 Switching Protocols', + 'Upgrade: websocket', + 'Connection: Upgrade', + `Sec-WebSocket-Accept: ${acceptKey(key)}`, + ...Object.entries(extraHeaders).map(([k, v]) => `${k}: ${v}`), + ] + socket.write(lines.join('\r\n') + '\r\n\r\n') + return new WsConn(socket, { head }) +} + +export function rejectUpgrade(socket, status, message = '') { + const text = { 400: 'Bad Request', 401: 'Unauthorized', 403: 'Forbidden', 404: 'Not Found', 409: 'Conflict', 502: 'Bad Gateway' }[status] || 'Error' + try { + socket.end(`HTTP/1.1 ${status} ${text}\r\nConnection: close\r\nContent-Type: text/plain\r\nContent-Length: ${Buffer.byteLength(message)}\r\n\r\n${message}`) + } catch { socket.destroy() } +} + +/** Client connect: resolves a WsConn, rejects with {status, body} on refusal. */ +export function connect(url, { headers = {}, timeoutMs = 15000, rejectUnauthorized = true } = {}) { + return new Promise((resolve, reject) => { + const u = new URL(url) + const secure = u.protocol === 'wss:' || u.protocol === 'https:' + const key = crypto.randomBytes(16).toString('base64') + const req = (secure ? https : http).request({ + host: u.hostname, + port: u.port || (secure ? 443 : 80), + path: u.pathname + u.search, + method: 'GET', + servername: u.hostname, + rejectUnauthorized, + headers: { + ...headers, + Host: u.host, + Connection: 'Upgrade', + Upgrade: 'websocket', + 'Sec-WebSocket-Version': '13', + 'Sec-WebSocket-Key': key, + }, + timeout: timeoutMs, + }) + req.on('timeout', () => { req.destroy(new Error(`connect timeout after ${timeoutMs} ms`)) }) + req.on('error', reject) + req.on('response', (res) => { + let body = '' + res.setEncoding('utf8') + res.on('data', (d) => { if (body.length < 4096) body += d }) + res.on('end', () => { + const err = new Error(`upgrade refused: HTTP ${res.statusCode} ${body.trim()}`) + err.status = res.statusCode + err.body = body + reject(err) + }) + }) + req.on('upgrade', (res, socket, head) => { + if (res.headers['sec-websocket-accept'] !== acceptKey(key)) { + socket.destroy() + return reject(new Error('bad Sec-WebSocket-Accept')) + } + socket.setTimeout(0) + resolve(new WsConn(socket, { client: true, head })) + }) + req.end() + }) +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..4af2c03 --- /dev/null +++ b/package.json @@ -0,0 +1,27 @@ +{ + "name": "@local/dsh-remote", + "version": "0.1.0", + "description": "Claude-Desktop-style remote control for DSH: master/slave link, every device's sessions on every device (/r//), plugin + config sync with opt-in secrets.", + "private": true, + "type": "module", + "license": "MIT", + "exports": { + ".": "./index.js", + "./client": "./client.js", + "./package.json": "./package.json" + }, + "files": ["index.js", "client.js", "lib/*.mjs", "cordis.patch.yml", "README.md", "deploy/*"], + "scripts": { "test": "node --test test/*.test.mjs" }, + "dsh": { + "bundle": { "patch": "./cordis.patch.yml" }, + "client": { + "platform": "web", + "immediately": true, + "inject": [ + "@deepseek-ai/dsh-client-ui-layout", + "@deepseek-ai/dsh-client-ui-sidebar", + "@deepseek-ai/dsh-client-ui-workspace" + ] + } + } +} diff --git a/test/link.test.mjs b/test/link.test.mjs new file mode 100644 index 0000000..b761343 --- /dev/null +++ b/test/link.test.mjs @@ -0,0 +1,140 @@ +// Offline tests for the link stack: WebSocket framing, mux calls/streams, HTTP + WS +// tunnelling through the proxy helpers, cookie minting, sync allow/deny. +import test from 'node:test' +import assert from 'node:assert/strict' +import crypto from 'node:crypto' +import http from 'node:http' +import { acceptUpgrade, connect, rejectUpgrade } from '../lib/ws.mjs' +import { Mux } from '../lib/mux.mjs' +import { ownHttp, ownUpgrade, viewHttp, viewUpgrade, stripDshCookies } from '../lib/proxy.mjs' +import { cookieName, mintCookie } from '../lib/auth-cookie.mjs' +import { allowed } from '../lib/sync.mjs' + +const listen = (srv) => new Promise((r) => srv.listen(0, '127.0.0.1', () => r(srv.address().port))) + +async function linkPair({ handlersA = {}, handlersB = {}, onStreamA, onStreamB } = {}) { + const srv = http.createServer() + let serverMux + const ready = new Promise((resolve) => { + srv.on('upgrade', (req, socket, head) => { + if (req.headers['x-key'] !== 'k') return rejectUpgrade(socket, 401, 'nope') + const ws = acceptUpgrade(req, socket, head) + serverMux = new Mux(ws, { odd: false, handlers: handlersA, onStream: onStreamA }) + resolve() + }) + }) + const port = await listen(srv) + const ws = await connect(`ws://127.0.0.1:${port}/link`, { headers: { 'x-key': 'k' } }) + const clientMux = new Mux(ws, { odd: true, handlers: handlersB, onStream: onStreamB }) + await ready + return { srv, port, serverMux, clientMux, close: () => { clientMux.close(); srv.close() } } +} + +test('ws: refused upgrade surfaces HTTP status', async () => { + const srv = http.createServer() + srv.on('upgrade', (req, socket) => rejectUpgrade(socket, 401, 'bad key')) + const port = await listen(srv) + await assert.rejects(connect(`ws://127.0.0.1:${port}/`), (e) => e.status === 401 && /bad key/.test(e.message)) + srv.close() +}) + +test('mux: calls both ways, large binary stream, errors', async () => { + const big = crypto.randomBytes(3 * 1024 * 1024 + 17) + const p = await linkPair({ + handlersA: { echo: async (x) => ({ back: x.v }), boom: async () => { throw Object.assign(new Error('kaboom'), { status: 418 }) } }, + handlersB: { whoami: async () => 'slave' }, + onStreamA: (s) => { + const parts = [] + s.on('data', (d) => parts.push(d)) + s.on('end', () => { const buf = Buffer.concat(parts); s.head({ size: buf.length }); s.write(buf); s.end() }) + s.start() + }, + }) + assert.deepEqual(await p.clientMux.call('echo', { v: 42 }), { back: 42 }) + assert.equal(await p.serverMux.call('whoami'), 'slave') + await assert.rejects(p.clientMux.call('boom'), (e) => e.status === 418 && e.message === 'kaboom') + await assert.rejects(p.clientMux.call('missing'), /unknown method/) + + const s = p.clientMux.open('echo', {}) + const got = await new Promise((resolve, reject) => { + const parts = [] + s.on('data', (d) => parts.push(d)) + s.on('end', () => resolve(Buffer.concat(parts))) + s.on('error', reject) + s.write(big) + s.end() + }) + assert.equal(got.length, big.length) + assert.ok(got.equals(big)) + p.close() +}) + +test('proxy: HTTP + WebSocket through the link with cookie + host rewrite', async () => { + // "Remote DSH": checks host + cookie like dsh-client-connection would. + const secret = crypto.randomBytes(32).toString('base64url') + const upstream = http.createServer((req, res) => { + const authority = req.headers.host + const ok = (req.headers.cookie || '').includes(cookieName(authority) + '=v1.') + if (!ok) { res.writeHead(401); return res.end('auth') } + const body = [] + req.on('data', (d) => body.push(d)) + req.on('end', () => { + res.writeHead(200, { 'content-type': 'text/plain', 'x-path': req.url, 'set-cookie': 'leak=1' }) + res.end(`${req.method} ${req.url} ${Buffer.concat(body).toString()} origin=${req.headers.origin || '-'} viewerCookie=${/dsh-auth-viewer/.test(req.headers.cookie) ? 'leaked' : 'stripped'}`) + }) + }) + upstream.on('upgrade', (req, socket, head) => { + const ok = (req.headers.cookie || '').includes(cookieName(req.headers.host) + '=v1.') + if (!ok) return rejectUpgrade(socket, 401, 'auth') + const ws = acceptUpgrade(req, socket, head) + ws.on('error', () => {}) + ws.on('message', (m) => ws.send(`echo:${m}`)) + }) + const upPort = await listen(upstream) + const target = { host: '127.0.0.1', port: upPort } + const cookie = async (authority) => mintCookie(authority, secret) + + const p = await linkPair({ + onStreamA: (s) => (s.kind === 'http' ? ownHttp(s, { target, cookie }) : ownUpgrade(s, { target, cookie })), + }) + + // Viewer webserver that mounts the remote at /r/x/. + const viewer = http.createServer((req, res) => { + viewHttp(req, res, (kind, meta) => p.clientMux.open(kind, meta), { rest: req.url.replace(/^\/r\/x/, ''), basePath: '/r/x' }) + }) + viewer.on('upgrade', (req, socket, head) => { + viewUpgrade(req, socket, head, (kind, meta) => p.clientMux.open(kind, meta), { rest: req.url.replace(/^\/r\/x/, '') }) + }) + const vPort = await listen(viewer) + + const r = await fetch(`http://127.0.0.1:${vPort}/r/x/api/thing?q=1`, { + method: 'POST', + body: 'hello', + headers: { cookie: 'dsh-auth-viewer=abc; other=1', origin: `http://127.0.0.1:${vPort}` }, + }) + const text = await r.text() + assert.equal(r.status, 200) + assert.equal(r.headers.get('set-cookie'), null, 'remote set-cookie must not reach the browser') + assert.match(text, /^POST \/api\/thing\?q=1 hello/) + assert.match(text, new RegExp(`origin=http://127.0.0.1:${upPort}`)) + assert.match(text, /viewerCookie=stripped/) + + const ws = await connect(`ws://127.0.0.1:${vPort}/r/x/api/remote.mux`) + const reply = await new Promise((resolve) => { ws.on('message', (m) => resolve(m)); ws.send('ping-1') }) + assert.equal(reply, 'echo:ping-1') + ws.close() + + p.close(); viewer.close(); upstream.close() +}) + +test('cookie + allow/deny helpers', () => { + const secret = crypto.randomBytes(32).toString('base64url') + const c = mintCookie('127.0.0.1:3081', secret) + assert.match(c, /^dsh-auth-[A-Za-z0-9_-]+=v1\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$/) + assert.equal(mintCookie('a', 'short'), null) + assert.equal(stripDshCookies('dsh-auth-x=1; a=2; dsh_auth_session=3'), 'a=2') + assert.equal(allowed('@local/foo', {}), true) + assert.equal(allowed('@local/foo', { deny: ['@local/*'] }), false) + assert.equal(allowed('@local/foo', { allow: ['@other/*'] }), false) + assert.equal(allowed('@local/foo', { allow: ['@local/foo'], deny: [] }), true) +})