diff --git a/docs/features/cloudflare-tunnel-sync/plan.md b/docs/features/cloudflare-tunnel-sync/plan.md new file mode 100644 index 000000000..ec101a5aa --- /dev/null +++ b/docs/features/cloudflare-tunnel-sync/plan.md @@ -0,0 +1,290 @@ +# Cloudflare Tunnel Host Sync Plan + +Spec: [spec.md](./spec.md). + +## Architecture + +Two boundaries, one data path: + +1. `SyncHostService` (core, `src/main/sync/host/`) owns the loopback sync endpoint, pairing, device + token authority, snapshot streaming, push assembly, rate limits, and audit. It consumes the + existing `SyncService` backup pipeline and `importFromSync`; it does not implement export/import + itself. +2. The Cloudflare Tunnel helper is **core-supervised** using the binary resolved from the plugin's + declared runtime (`PluginServicePort.getPlugin(id).runtime.command`). The plugin package owns the + bundled `cloudflared` binaries, their manifest declarations, and the tunnel settings UI; it does + not own the process (slice 0 finding, recorded in `spec.md`). + +Renderer owns Settings → Data UI for both layers (host mode, pairing, devices, status, progress, +errors). New canonical shared contracts live in `src/shared/contracts/routes/`; the renderer reaches +them through existing `api/*Client` patterns. + +## Current status + +Landed and verified (typecheck node+web, lint, format, i18n, `test/main/sync`, `test/main/contracts`, +`test/main/cli/surface.test.ts`): + +- `src/shared/contracts/syncHost.ts` — wire protocol, shared limits, DTO schemas. +- `src/shared/contracts/routes/syncHost.routes.ts` + catalog registration — `syncHost.*` IPC surface. +- `src/main/sync/host/{index,endpoint,devices,pairing,snapshot,state,routes}.ts` — host mode, + loopback endpoint (`handshake`, `pair`, `status`, `snapshot` with Range), device token authority, + pairing codes, audit, private machine-local state, and the endpoint descriptor. +- Composition wiring: service construction, `syncHostRoutes` in the route map, boot-time + `startIfEnabled()`, and a `syncHostService.stop` destroy step. +- `test/main/sync/host/hostEndpoint.test.ts` — 28 real-listener tests: uniform pre-auth 401s, + authenticated 404/405/501, pairing single-use and failure accounting, revocation (including across + a state reload), token-hash containment, byte-exact Range resume, abort-then-resume, abort during + snapshot resolution, corrupt archive handling, unreadable backup list, loopback-only reachability, + stalled-request reaping, connection-ceiling eviction, anonymous-audit coalescing, package-name + filtering, host-identity stability, oversized-body 413, keep-alive sockets staying evictable + after a download, per-device limiting of unknown paths, pre-`initialize()` state preservation, + lifecycle consistency under interleaved transitions, and teardown. +- `test/main/sync/host/state.test.ts` — 3 unit tests for the machine-local state store: rollback of + concurrent updates whose writes fail, a failed mutation staying out of the next successful write, + and a read/parse failure failing closed instead of resetting to defaults (then recovering). +- `test/main/sync/host/routes.test.ts` — 12 route-level tests for the seven `syncHost.*` handlers: + handler coverage, status/pairing shapes, `setEnabled` pass-through and failure propagation, pairing + creation and fallback, device-list and audit redaction, revoke/rename pass-through, input + validation, and renderer-caller enforcement. + +Not yet landed: push, change events, renderer UI + i18n copy, the tunnel supervisor, the plugin +package, snapshot production, and everything on the slave side. + +## Review record + +Independent security, correctness/lifecycle, and spec-conformance reviews were run against the first +implementation. Findings and disposition: + +| Finding | Severity | Disposition | +| --- | --- | --- | +| Device records, host identity and the enabled flag lived in the synced settings blob, so token hashes would ship in every backup/S3 upload and be merged on import (importer accepts the original host's tokens, revoked devices resurrect, host mode enabled without consent) | high | **Fixed** — all host state moved to `/sync-host/host-state.json` (`0600`, atomic); no settings keys involved. Covered by a test asserting the token never appears on disk. | +| `'sync_host'` was added to the log-event type union but not to the runtime `STARTUP_COMPONENTS` allowlist, so boot-failure reporting was silently rejected | high | **Fixed** — allowlist updated. | +| Unauthenticated method/path handling revealed the route surface (405/404 before auth) | medium | **Fixed** — uniform 401 before path/method handling; authenticated callers still get 404/405. | +| Stalled unauthenticated request could hold one of 16 connection slots for the 30-minute request timeout | medium | **Fixed** — request-receive timeout is now 60 s (configurable) and does not bound response streaming; the ceiling is 32 and full endpoints evict a non-streaming connection rather than refusing the newcomer; covered by a stalled-socket test and a ceiling-eviction test. | +| An anonymous caller could permanently kill pairing by burning the attempt budget | medium | **Fixed** — failures now impose backoff and never destroy the code; covered by a test. | +| Whole-archive `readFile` for the manifest, with no in-flight dedupe, multiplied memory under concurrent requests | medium | **Fixed** — the manifest is streamed (bounded memory) and concurrent `current()` calls share one digest pass. | +| No start/stop serialization: interleaved enable/disable could leave a listener running while host mode read as disabled | medium | **Fixed** — lifecycle transitions are serialized and `start()` is idempotent; invariant asserted in tests. | +| `stop()` could report stopped while the port was still bound (2 s fallback timer) | medium | **Fixed** — `stop()` awaits the real close, escalates, and warns if the listener survives. | +| Mid-stream read error never ended or destroyed the response, hanging the client until the request timeout | medium | **Fixed** — the connection is aborted instead. | +| Zero-byte snapshot produced an invalid range read | low | **Fixed** — explicit 409 for an empty package. | +| `push` path was missing from the handled set, so an authenticated push returned 404 while the comment claimed 501 | low | **Fixed** — returns 501 `not_implemented`. | +| Whitespace-only device name was accepted and replaced by a placeholder | low | **Fixed** — request schema trims and rejects empty names. | +| Rate-limit map was never pruned and keyed on an unvalidated header | low | **Fixed** — pruned on insert; the header is only trusted when it looks like an IP literal. | +| `databaseEncrypted` was computed from the manifest and then dropped by the schema | low | **Fixed** — reported in `status` so a slave fails clearly instead of hitting a decrypt error. | +| The traversal test was vacuous (`/sync/v1/../secret` is normalized client-side) | low | **Fixed** — replaced with uniform-401 and authenticated-404/405 assertions. | +| Provider API keys travel inside `agent.db` in every package; the spec claimed credentials never leave the machine | high | **Open — needs a product decision** (spec Open Questions): redact provider credentials from the export, or ship explicit consent + warning. | +| Host does not produce a snapshot on demand; a fresh host answers 404 and `startBackup` refuses while the legacy S3 toggle is off | medium | **Open** — added to slice 4. | +| Tokens are unscoped and non-expiring in practice though the spec implied scope and expiry | medium | **Open** — spec wording corrected; scope/expiry model is phase-2 or an explicit slice. | +| Descriptor pid/host-identity is never verified, so a stale descriptor could aim the tunnel at a reassigned port | medium | **Open** — slice 7 must verify the descriptor before launching the tunnel. | +| User-visible failures from main are not translatable copy yet (bind failure now throws a key-shaped error) | medium | **Open** — slice 1 adds the copy. | +| Cloudflare Access service token: no handling, no UI, no "no-Access" warning | medium | **Tracked** — slice 7. | +| Explicit confirmation + risk notice for enabling host mode is UI-only and unenforced in main | medium | **Tracked** — slice 1. | +| A mutation arriving before the initial state read persisted the empty default state over the real file, discarding every device record and the enabled flag | high | **Fixed** — `update()` loads first and concurrent loads share one read; regression test covers pre-`initialize()` mutation. | +| Persist failures were swallowed, so the documented enable rollback was dead code and a revocation could report success without being durable | medium/high | **Fixed** — write failures propagate to callers (the enable path rolls back, revocation surfaces an error); only last-seen is best-effort. | +| An unparseable archive threw out of a stream handler, escaping to the main process and hanging the request | medium | **Fixed** — archive errors are contained and report a null format version; covered by a corrupt-archive test. | +| The pairing backoff was global, so an anonymous caller could keep the legitimate user from pairing | medium | **Fixed** — failures are charged to the calling source; the user's code is never invalidated or blocked. | +| The audit recorded the planned status for an aborted transfer and dropped the device id on failures | low/medium | **Fixed** — aborted transfers are recorded as 499 and authenticated requests keep their device id. | +| An oversized pairing body reset the connection instead of answering | low | **Fixed** — the body is drained and the caller receives 413. | +| A package rewritten during hashing could be cached under a stale identity | low | **Fixed** — identity is re-verified after hashing; a second change reports no snapshot rather than a mismatched hash. | +| A client that disconnected while the snapshot was being resolved left the handler hanging forever: the abort hook was attached after the await, when `close` had already fired, so the request was never audited and the read stream was never released | high | **Fixed** — the hook is armed before the first await and the handler bails with a 499 audit entry when the peer is gone; covered by a test that aborts inside the resolution window. | +| A failed state read (EACCES/EIO, a lock, a corrupt file) silently degraded to the empty default state, and `initialize()` then persisted that default over the real file, discarding the enabled flag and every device record | high | **Fixed** — only `ENOENT` means "no state"; any other read/parse failure throws, leaves the cache unloaded and retries on the next call. | +| A failed state write left the cache ahead of disk: a failed enable could be persisted by a later unrelated write, and a failed disable skipped the stop path while the flag read as disabled | medium | **Fixed** — `update()` rolls the cache back when the write rejects and rethrows. | +| Snapshot resolution propagated storage failures as 500s (a package vanishing between `readdir` and `stat`, or during the digest) | medium | **Fixed** — resolution degrades to "no snapshot" (status `null`, snapshot 404) and a package that disappears mid-digest is no longer re-read from a stale stat; covered by a test. | +| Every inflated chunk of a `manifest.json` entry was buffered without a cap, so a deflate bomb in the sync folder could inflate into the main-process heap | medium | **Fixed** — the manifest buffer is capped at 1 MiB. | +| Anyone who learned the hostname could hold all 16 connection slots with stalled sockets and deny the legitimate device, including the user's own pairing | medium | **Fixed** — the ceiling is 32 and a full endpoint evicts the oldest connection that is not streaming a response instead of refusing the newcomer; covered by a test that fills the ceiling and still pairs. | +| The limiter maps only pruned expired windows, so inside one window nothing was freed and both maps could grow without bound (a local caller can also spoof `cf-connecting-ip`) | low | **Fixed** — both windowed maps are pruned and then hard-capped by evicting the oldest entry. | +| Unauthenticated traffic could flush the 500-entry audit ring in minutes, erasing the evidence of earlier probes | low | **Fixed** — anonymous traffic is coalesced into one entry per source that counts repeats (`suppressed`), and `handshake` is rate-limited like the other anonymous routes. | +| `attemptsRemaining` was a global counter driven by anonymous failures, so anyone could make the UI report a valid code as exhausted | low | **Fixed** — the field and the global counter are gone; enforcement is the per-source failure budget only. | +| `getHostId()` could mint two different identities before `initialize()` (routes are registered long before the boot start), so a pairing payload could disagree with a later handshake | low | **Fixed** — the pre-initialize identity is memoized and reused by `initialize()`; covered by a test. | +| Digest cache identity was size+mtime only, so a replaced file with preserved timestamps could serve a stale hash | low | **Fixed** — identity now includes inode and `ctimeMs`. | +| Failed atomic writes left `.tmp` debris next to the state file and the endpoint descriptor | low | **Fixed** — the temp file is removed on the failure path. | +| A post-listen server error (EMFILE/ENFILE/ECONNABORTED) had no listener and would surface as an unhandled error in the main process | low | **Fixed** — a permanent handler logs it and stops the endpoint if the listener is gone. | +| `stop()`'s "listener still bound" check was dead code (`server.listening` is false the moment `close()` is called), so a listener that survived the deadline was never reported | low | **Fixed** — `closeServer` resolves whether the close callback actually fired and warns when it did not. | +| Any `*.zip` in the sync folder was served as the host snapshot, while every other call site validates `backup-.zip` | low | **Fixed** — the snapshot source applies the same package-name filter; covered by a test. | +| The seven `syncHost.*` routes had no tests and did not assert a renderer caller | low | **Fixed** — `test/main/sync/host/routes.test.ts` (12 tests) and `requireRendererCaller` in every handler. | +| The loopback reachability test skipped its negative probe on hosts without an external interface | low | **Fixed** — an unconditional `127.0.0.2` probe proves the bind is address-specific; the external-interface probe remains as an extra. | +| The pre-`initialize()` state test never performed a mutation, so it passed with the guard removed | low | **Fixed** — the second instance now mutates before `initialize()`; removing the load-first guard fails the test. | +| A paired device could probe unknown paths without ever being charged: the per-device limiter ran after the 404/405 branches, and each of those responses evicted a legitimate audit entry | medium | **Fixed** — the limiter runs before path and method handling; covered by a test that probes unknown paths until it is throttled. | +| A finished keep-alive download left its socket marked as streaming, so it could never be evicted and 32 of them turned the connection ceiling into a wall | medium | **Fixed** — the flag is cleared when the stream settles, aborts included; covered by a test that fills the ceiling with finished downloads and still admits a newcomer. | +| Only filesystem writes were serialized, so a second update could snapshot a first update's mutation before it failed: the rollback was skipped and the change whose caller was told it failed could still be persisted | medium | **Fixed** — the whole transaction (load, mutate, write, rollback) is serialized and the rollback is unconditional; covered by a state-store test with two concurrent failing updates. | +| `getHostId()` wrote the pre-initialize identity fire-and-forget, so a failed write could leave the cache rolled back while `initialize()` had already adopted the value | low | **Fixed** — the value stays in memory until `initialize()` persists it through its awaited update. | +| Pairing consumed the code before issuing the device, so an issuance failure burned the user's code | low | **Fixed** — the code is restored when issuance fails. | +| Pairing code generation used `% 31` over random bytes, biasing the first alphabet entries ~1.4% | low | **Fixed** — rejection sampling. | +| `plan.md` claimed `handshake` was the only unauthenticated route, and `spec.md`'s default-scope row claimed provider credentials are excluded | low | **Fixed** — both statements now match the implementation. | +| `SYNC_HOST_MAX_PUSH_PART_BYTES` is declared but unused until push ships | low | **Tracked** — kept as the forward declaration for slice 5. | + +## Slice 0 — Gates before implementation + +Objective: remove the two unknowns that decide the plugin/core split. + +- [x] Verify declared-process capability semantics. **Result: a plugin cannot own a long-lived + non-MCP child.** An official plugin can only run a binary as an MCP stdio server; the SDK + kills only the direct child (SIGTERM→SIGKILL) and closes the transport before tree + termination, so a grandchild is reparented and survives. Plugin-owned supervision fails the + "no leftover process" acceptance criterion. +- [x] Decide how the endpoint port reaches the tunnel layer. **Result: a private descriptor file** + (`/sync-host/endpoint.json`, `0600`, temp+rename), written on start and removed on + stop. +- [x] Adopt the fallback: tunnel supervision lives in core; the plugin supplies the binary and UI. + +Completion: met — both questions are answered in `spec.md` and the fallback is adopted. + +## Slice 1 — Shared contracts and settings surface + +Objective: freeze the interface before handlers exist. + +- [x] Add canonical contracts for host enable/disable, status, pairing code, device list/rename/revoke + with redacted public DTOs (no tokens, no secrets) — `src/shared/contracts/routes/syncHost.routes.ts`. +- [ ] Add slave-side contracts (configure, transfer trigger) when slice 8 starts. +- [ ] Add i18n keys for all new user copy, including the enable risk notice, the Quick Tunnel + re-pairing warning, the "increment does not propagate updates or deletions" limitation, and + the `syncHost.error.bindFailed` key thrown by the host start path. +- [ ] Decide and surface the provider-credential situation (redact on export, or explicit consent + + warning) before host mode can be enabled. +- [ ] Extend Settings → Data with a host section and a slave section. + +Completion: contracts and copy exist, typecheck passes, no handlers yet. Host-side contracts landed; +copy and UI are pending, so this slice is not closed. + +## Slice 2 — Endpoint, auth and audit + +Objective: a loopback endpoint that rejects everything it should. + +- [x] `SyncHostService` binds `127.0.0.1:` while host mode is enabled. +- [x] Reuse control-plane hardening patterns: descriptor file `0600` via temp+rename, header cap, + connection cap, a per-request *receive* timeout (response streaming stays unbounded), and a + body cap on the pair request. +- [x] Bearer device-token middleware with hash-only storage, expiry, immediate revocation; 401 on + every failure path, authentication before method/path handling. +- [x] Bounded audit log: device, method, bytes, result, client IP; never tokens or payload contents. +- [x] `GET /sync/v1/handshake` reports protocol, host identity, app version, capabilities and + encryption mode; snapshot format version is reported by `status` instead, since it comes from + the backup manifest rather than a global constant. + +Completion: met — unauthorized paths cannot return 200, and `handshake` and `pair` are the only +unauthenticated routes (`pair` exchanges the one-time code for a device token, so it cannot require +one; everything else answers a uniform 401). Covered by `hostEndpoint.test.ts`. + +## Slice 3 — Pairing and device lifecycle + +Objective: devices get tokens, and can lose them. + +- [x] Pairing code: short TTL, single use, bounded attempts, carries the host identity. QR rendering + is UI work and remains pending. +- [x] `POST /sync/v1/pair` exchanges the code for a per-device token; the host persists only the + token hash plus metadata. +- [x] Device list, rename, revoke; revocation takes effect on the next request. +- [ ] Slave side stores `{hostUrl, deviceId, token}` with `safeStorage`, never in synced settings or + backup packages. + +Completion: host side met (pair → request → revoke → rejected, covered by tests). End-to-end +verification on two instances waits for the slave side. + +## Slice 4 — Snapshot pull + +Objective: resumable, integrity-checked pull. + +- [x] `GET /sync/v1/status` reports snapshot file name, size, hash and backup format version. +- [x] `GET /sync/v1/snapshot` streams the latest backup with `Content-Length` and hash headers; + `Range` returns 206 with `Content-Range`, unsatisfiable ranges return 416, and no snapshot + returns 404. +- [ ] Produce a snapshot on demand when the host has none (or none newer than its own data), and + define behavior while the legacy S3 sync toggle is off instead of silently serving 404 or an + arbitrarily old package. +- [ ] Slave resumes by offset, verifies the assembled hash, then calls the existing import path. + A partial download never reaches import. + +Completion: host side met (whole-file and ranged reads are byte-exact). The end-to-end resume → +import path waits for the slave side. + +## Slice 5 — Push + +Objective: host-side import that cannot half-apply. + +- [ ] `POST /sync/v1/push` accepts bounded parts (target ≤ 32 MiB), each independently retryable and + idempotent, staged outside the sync folder. +- [ ] Reassembly verifies part set, size and hash before import; incomplete pushes are discarded and + staged files cleaned up. +- [ ] Import uses existing `increment` | `overwrite` semantics; `overwrite` is an explicit, + confirmed action. + +Completion: a push interrupted between parts leaves no import side effect and can be resumed. + +## Slice 6 — Change events + +Objective: slaves can notice host changes without polling hard. + +- [ ] `GET /sync/v1/events` SSE stream with bounded keepalive and client count, no payload contents. +- [ ] Slave degrades to manual/scheduled pull when the stream is unavailable. + +Completion: stream works through the tunnel and its absence never breaks sync. + +## Slice 7 — Tunnel helper (core supervisor + plugin package) + +Objective: the tunnel layer, without the data plane. Split by slice 0's decision. + +Core supervisor: + +- [ ] `SyncHostTunnelService` spawns `cloudflared` using the binary path from + `PluginServicePort.getPlugin(id).runtime.command`, and stops it via + `terminateProcessTree` with `detached: true` on POSIX so the whole group dies. +- [ ] Add a subsystem value to `ChildProcessSubsystem` and reap stale tunnel processes at startup, + mirroring the MCP server reaping path. +- [ ] Crash-orphan watchdog or parent-liveness strategy (see `spec.md` open questions); until then + "no leftover process" holds at next boot rather than at crash time. +- [ ] Protocol selection with `http2` default, precheck `suggested_protocol` surfaced, and clear + errors for UDP-blocked networks. +- [ ] Read the endpoint port from `/sync-host/endpoint.json` and verify its pid and host + identity before launch, rejecting a missing or stale descriptor instead of aiming the tunnel + at a port the OS may have reassigned to another local service. +- [ ] Lifecycle wired into the same destroy path as `syncHostService.stop`. + +Plugin package: + +- [ ] `plugins/cloudflare-tunnel-sync/` with `plugin.json`, official source metadata, per-target + packages, bundled binaries (~42 MB per target), and `plugin:` detect candidates per platform. +- [ ] Named-tunnel configuration (user domain, config-file ingress) plus Quick Tunnel debug mode + with the re-pairing warning; optional `service: unix:` ingress for named tunnels. +- [ ] Settings contribution page: tunnel state, transport warnings, Cloudflare Access + service-token fields (host mode, pairing and device list live in Settings → Data, in core). + +Completion: a host exposes the endpoint over a real tunnel, and disabling host mode cleans up fully. + +## Slice 8 — Slave transfer UX + +Objective: the flow a user actually runs. + +- [ ] Slave-side pull and push with progress, cancel, retry and readable errors. +- [ ] On-demand and scheduled triggers; last-sync time and result in Settings → Data. +- [ ] Host-side view: connected devices, last seen, transfer history summary. + +Completion: two machines stay in sync across a restart, a tunnel outage, and a revoked device. + +## Slice 9 — Whole-change review + +Objective: verify the change against the spec before handoff. + +- [ ] Review hidden side effects, compatibility, failure behavior, performance, security, naming and + maintenance cost against `spec.md`. +- [ ] Confirm invariants: loopback-only binding, no credential-class data in any transfer, no + partial imports, no leftover process or port. +- [ ] Confirm known limitations are user-visible copy, not silent behavior. + +Completion: every acceptance criterion in the spec has a check or a recorded reason it is deferred. + +## Slice 10 — Validation and quality gates + +Objective: the smallest durable verification, then repo gates. + +- [ ] Durable tests only for qualifying behavior: auth rejection and revocation, credential + exclusion, resumable transfer integrity, push reassembly atomicity, and host-mode teardown. +- [ ] Remove every temporary probe and spike artifact before handoff. +- [ ] Run `pnpm run format`, `pnpm run i18n`, `pnpm run lint`, `pnpm run typecheck`, and the + relevant `test/main` and `test/renderer` suites. + +Completion: gates pass, and the durable tests fail if an invariant regresses. + +## Notes + +- Validation evidence and transport measurements are recorded in `spec.md`; the spike used synthetic + data only and has been torn down. diff --git a/docs/features/cloudflare-tunnel-sync/spec.md b/docs/features/cloudflare-tunnel-sync/spec.md new file mode 100644 index 000000000..88380b573 --- /dev/null +++ b/docs/features/cloudflare-tunnel-sync/spec.md @@ -0,0 +1,300 @@ +# Cloudflare Tunnel Host Sync + +Status: proposed. Transport validated end-to-end; host-side core implemented (pairing, device +tokens, status, snapshot pull with resume). Tunnel supervision, push, change events, settings UI +and the slave side are not implemented yet. + +A DeepChat instance becomes the **host** (device A) and exposes a sync endpoint through the user's +own Cloudflare Tunnel, so other devices (B/C/D) can pull from or push to it over a public HTTPS +address without a third-party bucket, inbound port, or working NAT. Topology is star-shaped: slaves +talk only to the host; devices never connect to each other. + +The existing S3/R2 cloud backup flow stays as-is and is not replaced. + +## Goals + +1. Host mode: a host exposes an authenticated sync endpoint through a user-operated Cloudflare + Tunnel; no inbound port, works behind CGNAT. +2. Pairing: the host shows a short-lived pairing code (text + QR); a slave exchanges it for its own + device token. The host can list, rename, and revoke devices at any time. +3. Transfer: slaves pull the host snapshot or push their own backup to the host, reusing the + existing backup/import pipeline with `increment` and `overwrite` semantics unchanged. +4. Visibility: tunnel state, transport protocol, last sync time, progress, and errors live in + Settings → Data. +5. Security: every request except `handshake` is authenticated; Cloudflare provides TLS; optional + Cloudflare Access service token; loopback-only origin binding. + +## Non-Goals + +- No DeepChat-operated relay, no multi-tenant account system. +- No queueing or forwarding while the host is offline; slaves wait or skip. +- No programmatic manipulation of the user's Cloudflare account. +- No replacement of the existing S3/R2 backup path. +- No concurrent-edit merge in this phase (see Known Limitations). +- No incremental or delete-propagation improvement to the backup format itself. + +## Ownership + +Two layers, deliberately split by capability rather than by convenience: + +- **Core** owns the data plane: the loopback-bound sync HTTP surface (`/sync/v1/*`), pairing, device + token authority, snapshot streaming, push assembly, audit, and the Settings → Data UI. Core + already owns `agent.db`, the backup pipeline and the import semantics; the endpoint must not + duplicate them. +- **Core also supervises the tunnel process** (Plan slice 0 closed this question against the + original assumption). Slice 0 established that a plugin cannot own a long-lived `cloudflared` + child: an official plugin can only run a binary as an MCP stdio server, the SDK kills only that + direct child (SIGTERM then SIGKILL) and closes the transport *before* tree termination, so a + reparented grandchild survives — which fails the "disabling leaves no leftover process" + acceptance criterion. Core supervision is therefore required, not preferred. +- **Plugin** (`com.deepchat.plugins.cloudflare-tunnel-sync`, official package) owns binary + provisioning and user-facing tunnel configuration: the bundled `cloudflared` per target, its + declared runtime manifest, the tunnel settings/status page, and the configuration values core + needs to launch the tunnel. + +The plugin still cannot carry the data plane for the reasons below: an official plugin cannot +register HTTP routes on the app server, cannot reach the DB or backup pipeline (`sync.*` contracts +are renderer-IPC only and absent from the CLI surface), and the only host-owned runtime adapter +(`cua-embedded-v1`) is reserved for the CUA plugin. + +## Host Endpoint + +A new server in the main process owns `127.0.0.1:`, started only while host mode is +enabled: + +- **Binding invariant**: loopback only. Never `0.0.0.0`, never a LAN address. This is what the + issue's security baseline permits and what the transport evidence forces: `cloudflared` Quick + Tunnels cannot target a unix socket, so a TCP origin is required on every platform. +- **Platform coverage**: POSIX and Windows both use the loopback listener. Windows cannot use a + named pipe for this, because `cloudflared` cannot dial one. +- **Hardening** reuses the control-plane patterns rather than inventing new ones: descriptor file + `0600` written by temp+rename, `maxHeaderSize` cap, connection cap, a bounded *request-receive* + timeout (which does not limit response streaming, so long downloads stay possible while a stalled + request cannot hold a connection slot), per-route body caps, and a bounded audit log. +- Authentication runs before path and method handling: every unauthenticated request other than + `handshake` and `pair` receives one uniform 401, so callers cannot map the route surface. Unknown + paths and unsupported methods are only distinguished for authenticated devices. +- The endpoint is **not** the local control plane. That surface explicitly non-goals TCP, loopback, + remote access and network callers, and its bearer token is a same-user file-readable secret. Host + sync gets its own listener, its own token authority, and its own principal model. +- Core publishes the bound port and host identity to a private descriptor + (`/sync-host/endpoint.json`, `0600`, temp+rename), which the tunnel supervisor reads + instead of guessing the port. The descriptor is removed when host mode is disabled. +- Renderer access is IPC-only (`syncHost.*` routes); remote devices never touch those routes. + +## Pairing and Device Tokens + +- Enabling host mode is off by default and requires an explicit confirmation with a risk notice. +- The host generates a pairing code: short TTL (single-digit minutes), single use, rate-limited + attempts. It carries the host identity, and the UI renders it as text plus QR alongside the + current tunnel URL, so a slave can confirm it reached the intended host. +- Failed attempts impose a backoff window but **never destroy the code**. Anyone who learns the + tunnel hostname can call `pair` unauthenticated, so letting failures invalidate the code would + hand an anonymous caller a permanent denial of pairing. +- Phase 1 uses an opaque host identity (`hostId`) rather than a cryptographic host key: the value is + generated once per profile and is what a slave compares against the pairing payload. Signature + verification and end-to-end payload encryption remain outside this phase and must not be implied + in UI copy. +- `POST /sync/v1/pair` exchanges the code for a per-device token. The host stores only the token + hash plus device metadata (id, name, created/expires, last seen). +- Tokens are per device and revocable immediately; revocation is enforced on the next request, not + on restart. Phase 1 issues unscoped, non-expiring tokens — the store supports expiry but pairing + does not set one yet, and there is no scope model. UI copy must not imply otherwise. +- Every other route requires `Authorization: Bearer `; failures return 401 and never + a success status, and authentication runs before method or path handling so unauthenticated + callers learn nothing about the route surface. + +## Snapshot Pull + +`GET /sync/v1/snapshot` streams the latest backup package: + +- `Content-Length`, snapshot id, and content hash headers are required; `Range` requests return + `206` with `Content-Range` (validated end-to-end, see Validation Evidence). +- The slave resumes by offset, then verifies the assembled hash before importing. A partial + download never reaches `importFromSync`. +- Payload reuse is the existing pipeline (`startBackup` producing `backup-.zip`, + `importFromSync` with `increment` | `overwrite`), not a parallel export implementation. +- The host currently serves whatever the newest package in the sync folder is; it does not yet + produce one on demand. A fresh or stale host therefore answers 404 or serves an old package, and + `startBackup` additionally refuses while the legacy S3 sync toggle is off. Closing this gap (a + host-triggered snapshot or an explicit "no snapshot yet" state) is required before the acceptance + criteria can pass. + +## Push + +`POST /sync/v1/push` accepts a slave's backup for host-side import, split into bounded parts: + +- Cloudflare documents a **100 MB proxied request body limit** on free plans. Our validation pushed + 125,829,120 bytes through a Quick Tunnel successfully, so enforcement varies by tunnel mode, plan + and edge; the design must not depend on it. Parts are therefore capped well below that + (target ≤ 32 MiB), each part is independently retryable and idempotent, and the host reassembles + into a staging file before import. +- Import runs only after full assembly and hash/identity verification. An interrupted or partial + push leaves no import side effect; staging is discarded and cleaned up. +- Imported data uses the existing `increment` | `overwrite` modes. `increment` only inserts missing + rows; it does not propagate updates or deletions. This limitation is user-visible copy, not a + hidden surprise. + +## Change Events + +`GET /sync/v1/events` is a long-lived SSE stream used to tell slaves that host data changed, so a +slave can decide to pull. Bounded keepalive, bounded client count, no payload contents, no +credential material. Slaves must treat it as an optimization: absence of the stream must degrade to +manual or scheduled pulls, never to a broken sync. + +## Slave Device Side + +- Slaves store `{hostUrl, deviceId, token}` locally; the token is protected with `safeStorage`, + never written to synced settings or backup packages. +- Slaves trigger pull or push on demand or on a schedule, with progress, cancel, and clear errors. +- Host URL changes (for example after a Quick Tunnel restart mints a new hostname) invalidate + pairing for that device until re-paired; the UI must say so instead of failing opaquely. + +## Tunnel Helper Plugin + +- Package id `com.deepchat.plugins.cloudflare-tunnel-sync`, official source, per-target packages + following the existing `deepchat-plugin-*` release naming. +- `cloudflared` ships **inside the plugin package** (measured `darwin-amd64` binary: 41.7 MB), so + no download infrastructure is required. Detection uses manifest `plugin:` relative candidates and + the host applies the executable bit; version comes from `--version`. Note that + `runtime.install.provider`/`strategy` are only recorded as labels by the host today; there is no + generic manifest-driven downloader to lean on. +- **Transport protocol**: `http2` is the default and is user-selectable. QUIC/UDP 7844 is blocked on + real networks (including the validation network): with the default `auto`, `cloudflared` retries + QUIC indefinitely and every request fails with 502 while TCP/HTTP2 passes the precheck. The plugin + surfaces transport state and the precheck's `suggested_protocol` instead of failing silently. +- Tunnel modes: a named tunnel on the user's own domain (fixed hostname, the supported product path) + and Quick Tunnel (random `*.trycloudflare.com`, debug/fallback, no SLA, new hostname per start). + Quick Tunnel mode must warn that pairing does not survive a restart. +- Optional hardening: for named-tunnel users, a config-file ingress with + `service: unix://sync.sock` is supported and validates cleanly, but it must remain + optional because Quick Tunnel cannot use it. +- Host mode disable must stop the tunnel process and close the listener: no leftover `cloudflared`, + no listening port, no stale socket file. + +## Excluded Data (Invariants) + +These must never appear in any transfer, and this is verified rather than assumed: + +- **Host-mode credentials and identity**: device token hashes, the host identity and the enabled + flag. These are satisfied by construction: host state lives in a private machine-local file + (`/sync-host/host-state.json`, `0600`), not in the settings blob. This matters because + settings travel inside backup packages (`configs/app-settings.json`), into S3/R2 uploads, and are + merged wholesale on import — a settings-backed device list would have let an importer accept + another host's device tokens, resurrect revoked devices, and enable host mode without consent. +- tunnel credentials and Cloudflare Access secrets — nothing in the sync path touches them yet; + this becomes an implementation obligation when the tunnel layer lands. +- machine-local values excluded by design today (`cloudSyncSecret`, `agentCommandShell`); +- memory vector data — vectors are not in `agent.db`; slaves regenerate them locally. + +**Not yet satisfied — provider credentials.** The existing backup format packages `database/agent.db` +whole, and `providers.api_key` is a plaintext column in that database. Today's S3/R2 flow already +uploads it; serving the same package over a tunnel extends that exposure to every paired device and +to the network path in between. "Provider API keys never leave the machine" is therefore **false** +until either the export redacts provider credentials or the feature ships with an explicit, +user-visible warning and consent. This is a product decision, not an implementation detail, because +redacting keys changes what an imported backup restores. + +Session, message and settings data are the baseline; skills, MCP configuration and knowledge-base +files are in scope because the existing backup package already carries them. + +## Security Baseline + +- Default off; enabling requires explicit confirmation and a risk notice. +- Loopback-only origin binding on every platform; Unix socket is an optional extra for named + tunnels, never a requirement. +- Per-device tokens: hash-only at rest, immediate revocation. Phase 1 issues **unscoped, + non-expiring** tokens — the store supports expiry but pairing does not set one, and there is no + scope model, so a leaked device token stays valid on every route until a human revokes it. Pairing + codes are one-time, short-lived, and rate-limited per source. +- Request size caps, per-device rate limits, and path validation on both ends. +- Audit log records device, method, bytes, result, and client IP (`cf-connecting-ip` is forwarded by + Cloudflare and was confirmed present at the origin), never tokens or payload contents. +- Cloudflare Access service token (`CF-Access-Client-Id`/`CF-Access-Client-Secret`) is **strongly + recommended** rather than mandatory: the Settings UI provides a configuration entry and warns when + a public hostname runs without it, but application-level device tokens remain the enforced layer. + +## Compatibility + +- The endpoint is versioned (`/sync/v1`) and `handshake` reports protocol, app version, database + version, capabilities and encryption mode so a slave can refuse an incompatible host instead of + corrupting data. +- Import compatibility is the existing backup contract; a host and slave on incompatible database + versions must fail handshake, not attempt import. +- Nothing in this feature changes existing S3/R2 behavior, existing sync IPC contracts, or the local + control plane's surface. + +## Known Limitations + +- Whole-database transfer every time; no incremental and no delete propagation. +- `increment` inserts missing rows only, so updates and deletions made on the host do not reach + slaves. +- Memory vectors must be regenerated on the receiving side. +- Measured throughput on the validation network was ~7.2 MB/s, so a multi-hundred-MB database means + minutes, not seconds. +- The host is a single point of failure by design; slaves cannot reach each other. + +## Validation Evidence + +Transport was validated end-to-end before writing this spec (bundled `cloudflared` 2026.9.1, Quick +Tunnel, synthetic data only, bearer-gated endpoint): + +| Probe | Outcome | +| --- | --- | +| Quick Tunnel + public edge request | 200, TLS 490 ms, TTFB 1.29 s | +| Missing/invalid token | 401, never 200 | +| `Range` request | 206 with correct `Content-Range` | +| Two ranged halves vs single fetch | byte-identical | +| Abort at 37,993,254 bytes then `curl -C -` | resumed to full length, byte-identical | +| 120 MiB download | 17.4 s, ~7.2 MB/s | +| SSE `/events` | 5 events over ~3 s, clean close | +| 120 MiB upload | accepted (but see the 100 MB documented proxy limit) | +| `cf-connecting-ip` at origin | present | +| QUIC/UDP 7844 | blocked; precheck `suggested_protocol=http2` | +| `cloudflared --url unix:/path` | fails (`http://unix:` → DNS lookup of `unix`) | +| Config-file `service: unix:` ingress | `ingress validate` OK, routes correctly | +| Process teardown | no leftover processes; stale socket file survived SIGTERM | + +## Acceptance Criteria + +1. Two devices pair, then pull and push successfully; `increment` import neither loses data nor + duplicates sessions. +2. A transfer interrupted mid-flight resumes without corruption and without a partial import. +3. A revoked or expired token is rejected immediately; no request returns 200 on a failed + validation. +4. Disabling the feature leaves no listening port, no `cloudflared` process, and no stale socket. +5. Credential-class data (provider keys, tunnel credentials, machine-local values) is provably + absent from every transfer. +6. Host mode works on a network where QUIC/UDP 7844 is blocked. +7. Host mode works on macOS, Linux and Windows hosts. + +## Resolved Questions + +| Question | Decision | +| --- | --- | +| Windows host: loopback-only or macOS/Linux-only in phase 1? | Support Windows hosts in phase 1, using the same loopback listener as POSIX. | +| `cloudflared` managed by the app or user-run? | Bundled inside the plugin package; core supervises the process using the plugin-resolved binary path. | +| Default sync scope | Sessions, messages, settings, plus skills, MCP configuration and knowledge-base files. Memory vectors excluded (not in `agent.db`; slaves regenerate them). **Provider credentials are currently included, not excluded** — they are plaintext columns in the `agent.db` that every package carries, so this row stays unresolved until the export redacts them or the feature ships explicit consent (see Excluded Data). | +| Access service token mandatory? | Strongly recommended in the UI, not mandatory; device tokens remain enforced. | +| Endpoint transport (added) | Loopback TCP listener on every platform; Unix socket optional for named tunnels only. | +| Push framing (added) | Bounded, independently retryable parts; no reliance on large single-body uploads. | +| Who supervises the tunnel process? (Plan slice 0) | Core, after slice 0 proved a plugin cannot guarantee grandchild teardown: the MCP SDK kills only its direct child and closes the transport before tree termination, so a reparented `cloudflared` survives plugin disable. The plugin supplies the binary and UI. | + +## Open Questions + +- **Provider credentials in the served package** (see Excluded Data): decide between redacting + provider credentials from the export, or shipping an explicit consent + warning. This blocks the + "credentials provably absent" acceptance criterion. +- **Snapshot production**: whether the host creates a snapshot on demand or only serves an existing + one, and how it behaves when the legacy S3 sync toggle is off. +- Descriptor ownership verification: the endpoint descriptor carries a pid and host identity but + nothing verifies them. Once the tunnel supervisor lands, a stale descriptor could aim the tunnel + at a port the OS later reassigned to an unrelated local service. +- Crash-orphan handling for the tunnel: a core-spawned `cloudflared` is reparented after an app + crash or SIGKILL and holds its edge connection until the next launch reaps it. Closing this needs + a watchdog or a parent-liveness mechanism, since `cloudflared` has no equivalent of the CUA + driver's `--parent-liveness-stdio` flag. The acceptance criterion currently holds at next boot, + not at crash time. +- Whether the slave must be told that a host's database is encrypted before it attempts an import. + The snapshot already reports `backupFormatVersion`; a companion "database is encrypted" flag may + be required so the slave fails with a clear message instead of a decrypt error. diff --git a/src/main/app/composition.ts b/src/main/app/composition.ts index e1b392fee..db6cb5534 100644 --- a/src/main/app/composition.ts +++ b/src/main/app/composition.ts @@ -125,6 +125,8 @@ import { createDeviceRoutes } from '../device/routes' import { createOnboardingRoutes } from '../onboarding/routes' import { createUpgradeRoutes } from '../upgrade/routes' import { createSyncRoutes } from '../sync/routes' +import { SyncHostService } from '../sync/host' +import { createSyncHostRoutes } from '../sync/host/routes' import { createPlatformRoutes } from '../platform/routes' import { createHookRoutes } from '../hook/routes' import { createAppSettingsRoutes } from './settingsRoutes' @@ -525,6 +527,7 @@ export async function createMainProcessControl(dependencies: { let ocrSettings: OcrSettings let mcpService: McpService let syncService: SyncService + let syncHostService: SyncHostService let deeplinkService: DeeplinkService let notificationService: NotificationService let tabPresenter: TabPresenter @@ -1318,6 +1321,13 @@ export async function createMainProcessControl(dependencies: { providerDatabase, publishDeepchatEvent ) + syncHostService = new SyncHostService({ + listBackups: () => syncService.listBackups(), + getFolderPath: () => syncSettings.getFolderPath(), + getUserDataPath: () => app.getPath('userData'), + getAppVersion: () => app.getVersion(), + logger + }) notificationService = new NotificationService(desktopSettings, publishDeepchatEvent) trayPresenter = new TrayPresenter(desktopSettings, windowPresenter) dialogService = new DialogService(publishDeepchatEvent) @@ -2635,6 +2645,7 @@ export async function createMainProcessControl(dependencies: { async function destroy(): Promise { await runDestroyStep('agentCliTokenAuthority.clear', () => agentCliTokenAuthority.clear()) await runDestroyStep('cliServer.stop', () => cliServer.stop()) + await runDestroyStep('syncHostService.stop', () => syncHostService.stop()) await runDestroyStep('tapeInspectorHeadWatcher.close', () => tapeInspectorHeadWatcher.close()) await runDestroyStep('typedEventHub.close', () => typedEventHub.close()) await runDestroyStep('cliMutationGuard.clear', () => cliMutationGuard.clear()) @@ -2933,6 +2944,7 @@ export async function createMainProcessControl(dependencies: { }) } }) + const syncHostRoutes = createSyncHostRoutes({ host: syncHostService }) const platformRoutes = createPlatformRoutes({ proxySettings: dependencies.proxySettings, applyProxyMode: (mode) => { @@ -3086,6 +3098,7 @@ export async function createMainProcessControl(dependencies: { upgradeRoutes, exporterRoutes, syncRoutes, + syncHostRoutes, platformRoutes, hookRoutes, notificationRoutes, @@ -3583,6 +3596,12 @@ export async function createMainProcessControl(dependencies: { reportMainStartupComponentFailure(dependencies.startupRunId, 'cli_control', 'unknown') logger.error('[CLI] Failed to start local control server', error) } + try { + await syncHostService.startIfEnabled() + } catch (error) { + reportMainStartupComponentFailure(dependencies.startupRunId, 'sync_host', 'unknown') + logger.error('[SyncHost] Failed to start host mode', error) + } if (cliServer.getStatus().running) { try { await cliLauncherService.ensureInstalled() diff --git a/src/main/logging/mainLogEvents.ts b/src/main/logging/mainLogEvents.ts index 52d7097a3..92758614e 100644 --- a/src/main/logging/mainLogEvents.ts +++ b/src/main/logging/mainLogEvents.ts @@ -82,6 +82,7 @@ export type MainLogStartupComponent = | 'rtk_health_check' | 'skill_sync' | 'sqlite_mainline_normalization' + | 'sync_host' | 'toolchain_gc' | 'usage_stats_backfill' @@ -453,6 +454,7 @@ const STARTUP_COMPONENTS = [ 'rtk_health_check', 'skill_sync', 'sqlite_mainline_normalization', + 'sync_host', 'toolchain_gc', 'usage_stats_backfill' ] as const satisfies readonly MainLogStartupComponent[] diff --git a/src/main/sync/host/devices.ts b/src/main/sync/host/devices.ts new file mode 100644 index 000000000..c2a4caf33 --- /dev/null +++ b/src/main/sync/host/devices.ts @@ -0,0 +1,129 @@ +import { createHash, randomBytes, timingSafeEqual } from 'node:crypto' +import { + SYNC_HOST_DEVICE_NAME_MAX_LENGTH, + SYNC_HOST_DEVICE_TOKEN_BYTES, + type SyncHostDeviceView +} from '@shared/contracts/syncHost' +import type { SyncHostDeviceRecord, SyncHostStateStore } from './state' + +const LAST_SEEN_PERSIST_INTERVAL_MS = 60_000 + +export interface IssuedSyncHostDevice { + device: SyncHostDeviceView + token: string +} + +function hashToken(token: string): string { + return createHash('sha256').update(token).digest('hex') +} + +function toView(record: SyncHostDeviceRecord): SyncHostDeviceView { + return { + deviceId: record.deviceId, + name: record.name, + createdAt: record.createdAt, + expiresAt: record.expiresAt, + lastSeenAt: record.lastSeenAt, + revoked: record.revokedAt !== null + } +} + +/** + * Owns per-device bearer tokens for the sync host endpoint: issuance, authentication, revocation + * and expiry. Records live in the machine-local host state, never in the synced settings blob, and + * only token hashes are stored. + */ +export class SyncHostDeviceStore { + private readonly lastSeenPersistedAt = new Map() + + constructor(private readonly state: SyncHostStateStore) {} + + list(): SyncHostDeviceView[] { + return this.state + .snapshot() + .devices.map(toView) + .sort((left, right) => right.createdAt - left.createdAt) + } + + count(): number { + return this.state.snapshot().devices.length + } + + async issue(input: { + name: string + expiresAt?: number | null + now?: number + }): Promise { + const now = input.now ?? Date.now() + const token = randomBytes(SYNC_HOST_DEVICE_TOKEN_BYTES).toString('base64url') + const record: SyncHostDeviceRecord = { + deviceId: `dev_${randomBytes(9).toString('hex')}`, + name: input.name.trim().slice(0, SYNC_HOST_DEVICE_NAME_MAX_LENGTH), + tokenHash: hashToken(token), + createdAt: now, + lastSeenAt: null, + expiresAt: input.expiresAt ?? null, + revokedAt: null + } + await this.state.update((state) => { + state.devices.push(record) + }) + return { device: toView(record), token } + } + + /** + * Verifies a presented bearer token. Returns the device view on success and `null` for + * unknown, malformed, revoked or expired tokens. + */ + authenticate(token: string, now: number = Date.now()): SyncHostDeviceView | null { + if (!token) return null + const presented = Buffer.from(hashToken(token), 'hex') + for (const record of this.state.snapshot().devices) { + const expected = Buffer.from(record.tokenHash, 'hex') + if (presented.length !== expected.length) continue + if (!timingSafeEqual(presented, expected)) continue + if (record.revokedAt !== null) return null + if (record.expiresAt !== null && record.expiresAt <= now) return null + this.touchLastSeen(record.deviceId, now) + return toView(record) + } + return null + } + + async revoke(deviceId: string, now: number = Date.now()): Promise { + let revoked = false + await this.state.update((state) => { + const target = state.devices.find((record) => record.deviceId === deviceId) + if (!target || target.revokedAt !== null) return + target.revokedAt = now + revoked = true + }) + return revoked + } + + async rename(deviceId: string, name: string): Promise { + const next = name.trim().slice(0, SYNC_HOST_DEVICE_NAME_MAX_LENGTH) + if (!next) return false + let renamed = false + await this.state.update((state) => { + const target = state.devices.find((record) => record.deviceId === deviceId) + if (!target) return + target.name = next + renamed = true + }) + return renamed + } + + private touchLastSeen(deviceId: string, now: number): void { + const lastPersisted = this.lastSeenPersistedAt.get(deviceId) ?? 0 + if (now - lastPersisted < LAST_SEEN_PERSIST_INTERVAL_MS) return + this.lastSeenPersistedAt.set(deviceId, now) + // Fire-and-forget: a failed last-seen write must never fail an authorized request. + void this.state + .update((state) => { + const target = state.devices.find((record) => record.deviceId === deviceId) + if (target) target.lastSeenAt = now + }) + .catch(() => undefined) + } +} diff --git a/src/main/sync/host/endpoint.ts b/src/main/sync/host/endpoint.ts new file mode 100644 index 000000000..6e59feeb8 --- /dev/null +++ b/src/main/sync/host/endpoint.ts @@ -0,0 +1,790 @@ +import http from 'node:http' +import type net from 'node:net' +import fs from 'node:fs' +import { + SYNC_HOST_AUDIT_LIMIT, + SYNC_HOST_DEVICE_HEADER, + SYNC_HOST_EVENTS_PATH, + SYNC_HOST_HANDSHAKE_PATH, + SYNC_HOST_MAX_CONNECTIONS, + SYNC_HOST_MAX_HEADER_BYTES, + SYNC_HOST_PAIR_BODY_MAX_BYTES, + SYNC_HOST_PAIR_FAILURE_MAX_KEYS, + SYNC_HOST_PAIR_FAILURE_WINDOW_MS, + SYNC_HOST_PAIR_MAX_FAILURES_PER_WINDOW, + SYNC_HOST_PAIR_PATH, + SYNC_HOST_PROTOCOL_NAME, + SYNC_HOST_PROTOCOL_VERSION, + SYNC_HOST_PUSH_PATH, + SYNC_HOST_RATE_LIMIT_MAX_KEYS, + SYNC_HOST_RATE_LIMIT_REQUESTS_PER_WINDOW, + SYNC_HOST_RATE_LIMIT_WINDOW_MS, + SYNC_HOST_REQUEST_RECEIVE_TIMEOUT_MS, + SYNC_HOST_SNAPSHOT_HASH_HEADER, + SYNC_HOST_SNAPSHOT_ID_HEADER, + SYNC_HOST_SNAPSHOT_PATH, + SYNC_HOST_STATUS_PATH, + SyncHostHandshakeSchema, + SyncHostPairRequestSchema, + SyncHostPairResponseSchema, + SyncHostStatusSchema, + type SyncHostAuditEntry, + type SyncHostCapability +} from '@shared/contracts/syncHost' +import type { IssuedSyncHostDevice, SyncHostDeviceStore } from './devices' +import type { SyncHostPairingAuthority } from './pairing' +import type { SyncHostSnapshotSource } from './snapshot' + +/** Capabilities this build actually serves; the handshake must not advertise more. */ +const HOST_CAPABILITIES: SyncHostCapability[] = ['snapshot', 'range'] +const HANDLED_PATHS = new Set([ + SYNC_HOST_HANDSHAKE_PATH, + SYNC_HOST_PAIR_PATH, + SYNC_HOST_STATUS_PATH, + SYNC_HOST_SNAPSHOT_PATH, + SYNC_HOST_PUSH_PATH, + SYNC_HOST_EVENTS_PATH +]) + +export interface SyncHostEndpointLogger { + warn(message: string, meta?: unknown): void +} + +export interface SyncHostEndpointDeps { + devices: SyncHostDeviceStore + pairing: SyncHostPairingAuthority + snapshotSource: SyncHostSnapshotSource + getHostId: () => string + getAppVersion: () => string + logger?: SyncHostEndpointLogger + /** Overridable for tests; production uses the shared contract default. */ + requestReceiveTimeoutMs?: number +} + +interface RangeSelection { + start: number + end: number +} + +/** + * Loopback-only HTTP endpoint served through the user's Cloudflare Tunnel. + * + * Binding is hard-coded to `127.0.0.1`: the tunnel connector dials this listener from the same + * machine, so no other interface must ever be reachable. Only `handshake` is unauthenticated. + */ +export class SyncHostEndpoint { + private server: http.Server | null = null + private boundPort = 0 + private readonly sockets = new Map() + private readonly auditEntries: SyncHostAuditEntry[] = [] + private readonly anonymousAudit = new Map() + private readonly rateWindows = new Map() + private readonly requestGuards = new Map() + private readonly pairFailures = new Map() + + private readonly requestReceiveTimeoutMs: number + + constructor(private readonly deps: SyncHostEndpointDeps) { + this.requestReceiveTimeoutMs = + deps.requestReceiveTimeoutMs ?? SYNC_HOST_REQUEST_RECEIVE_TIMEOUT_MS + } + + isRunning(): boolean { + return this.server !== null + } + + getPort(): number { + return this.boundPort + } + + getAuditEntries(): SyncHostAuditEntry[] { + return [...this.auditEntries] + } + + async start(input: { port?: number } = {}): Promise<{ port: number }> { + if (this.server) return { port: this.boundPort } + const server = http.createServer( + { maxHeaderSize: SYNC_HOST_MAX_HEADER_BYTES, requestTimeout: this.requestReceiveTimeoutMs }, + (request, response) => { + void this.handle(request, response) + } + ) + server.headersTimeout = 15_000 + + server.on('connection', (socket) => { + if (this.sockets.size >= SYNC_HOST_MAX_CONNECTIONS) { + // Refusing the newcomer would let anyone who learns the hostname hold every slot with + // stalled connections and deny the legitimate device — including the user's own pairing. + // Drop the oldest connection that is not streaming a response instead: an in-flight + // download is never sacrificed, and the newcomer always gets in. + const victim = this.oldestNonStreamingSocket() + if (!victim) { + socket.destroy() + return + } + this.sockets.delete(victim) + victim.destroy() + } + this.sockets.set(socket, { streaming: false }) + socket.on('close', () => this.sockets.delete(socket)) + }) + + const port = input.port ?? 0 + await new Promise((resolve, reject) => { + server.once('error', reject) + server.listen(port, '127.0.0.1', () => { + server.removeListener('error', reject) + resolve() + }) + }) + + // A post-listen server error (accept failure: EMFILE/ENFILE/ECONNABORTED) would otherwise be an + // unhandled EventEmitter error and take down the main process. + server.on('error', (error) => { + this.deps.logger?.warn('[SyncHost] Endpoint server error', { + error: error instanceof Error ? error.message : String(error) + }) + if (!server.listening) void this.stop() + }) + + const address = server.address() + if (!address || typeof address === 'string') { + await this.closeServer(server) + throw new Error('Sync host endpoint did not bind to a TCP port') + } + this.server = server + this.boundPort = address.port + this.deps.logger?.warn('[SyncHost] Endpoint listening', { port: this.boundPort }) + return { port: this.boundPort } + } + + async stop(): Promise { + const server = this.server + this.rateWindows.clear() + this.pairFailures.clear() + this.anonymousAudit.clear() + if (!server) { + this.boundPort = 0 + return + } + for (const guard of this.requestGuards.values()) clearTimeout(guard) + this.requestGuards.clear() + for (const socket of this.sockets.keys()) socket.destroy() + this.sockets.clear() + const closed = await this.closeServer(server) + // `server.listening` is already false the moment close() is called, even with connections still + // open, so the only honest signal that the listener survived the deadline is the close callback. + if (!closed) { + this.deps.logger?.warn('[SyncHost] Listener did not close within the deadline') + } + this.server = null + this.boundPort = 0 + } + + /** Resolves true when the listener actually closed, false when the deadline was reached first. */ + private async closeServer(server: http.Server): Promise { + return await new Promise((resolve) => { + let settled = false + let closed = false + const done = (): void => { + if (settled) return + settled = true + clearTimeout(escalation) + clearTimeout(giveUp) + resolve(closed) + } + const escalation = setTimeout(() => { + server.closeAllConnections?.() + }, 1_000) + escalation.unref?.() + const giveUp = setTimeout(done, 5_000) + giveUp.unref?.() + server.close(() => { + closed = true + done() + }) + }) + } + + private async handle( + request: http.IncomingMessage, + response: http.ServerResponse + ): Promise { + const url = request.url ?? '/' + const path = url.split('?')[0] + const method = request.method ?? 'GET' + const clientIp = this.resolveClientIp(request) + const socket = request.socket + let authenticatedDeviceId: string | null = null + this.armRequestGuard(socket) + if (request.complete) { + this.clearRequestGuard(socket) + } else { + const clear = (): void => this.clearRequestGuard(socket) + request.once('end', clear) + request.once('aborted', clear) + request.once('error', clear) + request.once('close', clear) + } + + try { + if (path === SYNC_HOST_HANDSHAKE_PATH && method === 'GET') { + if (!this.consumeRateLimit(`anon:${clientIp ?? 'unknown'}`)) { + this.respondJson(response, 429, { error: 'rate_limited' }) + this.auditAnonymous({ method, path, status: 429, bytes: 0, deviceId: null, clientIp }) + return + } + const payload = SyncHostHandshakeSchema.parse({ + protocol: SYNC_HOST_PROTOCOL_NAME, + protocolVersion: SYNC_HOST_PROTOCOL_VERSION, + hostId: this.deps.getHostId(), + appVersion: this.deps.getAppVersion(), + capabilities: HOST_CAPABILITIES, + encryption: { payload: 'none', transport: 'tls' } + }) + const bytes = this.respondJson(response, 200, payload) + this.auditAnonymous({ method, path, status: 200, bytes, deviceId: null, clientIp }) + return + } + + if (path === SYNC_HOST_PAIR_PATH) { + await this.handlePair(request, response, method, path, clientIp) + return + } + + // Authentication comes before path and method handling: an unauthenticated caller must not be + // able to tell which routes exist, or which methods they accept, from the response. + const device = this.authenticate(request) + if (!device) { + if (!this.consumeRateLimit(`anon:${clientIp ?? 'unknown'}`)) { + this.respondJson(response, 429, { error: 'rate_limited' }) + this.auditAnonymous({ method, path, status: 429, bytes: 0, deviceId: null, clientIp }) + return + } + this.respondJson(response, 401, { error: 'unauthorized' }) + this.auditAnonymous({ method, path, status: 401, bytes: 0, deviceId: null, clientIp }) + return + } + authenticatedDeviceId = device.deviceId + response.setHeader(SYNC_HOST_DEVICE_HEADER, device.deviceId) + + // The limiter runs before path and method handling: a paired device hammering unknown paths + // would otherwise never be charged, and every 404 it earns evicts a legitimate audit entry. + if (!this.consumeRateLimit(device.deviceId)) { + this.respondJson(response, 429, { error: 'rate_limited' }) + this.audit({ method, path, status: 429, bytes: 0, deviceId: device.deviceId, clientIp }) + return + } + + if (!HANDLED_PATHS.has(path)) { + this.respondJson(response, 404, { error: 'not_found' }) + this.audit({ method, path, status: 404, bytes: 0, deviceId: device.deviceId, clientIp }) + return + } + if (method !== 'GET' && method !== 'POST') { + this.respondJson(response, 405, { error: 'method_not_allowed' }) + this.audit({ method, path, status: 405, bytes: 0, deviceId: device.deviceId, clientIp }) + return + } + + if (path === SYNC_HOST_STATUS_PATH) { + if (method !== 'GET') { + this.respondJson(response, 405, { error: 'method_not_allowed' }) + this.audit({ method, path, status: 405, bytes: 0, deviceId: device.deviceId, clientIp }) + return + } + await this.handleStatus(response, method, path, device.deviceId, clientIp) + return + } + + if (path === SYNC_HOST_SNAPSHOT_PATH) { + if (method !== 'GET') { + this.respondJson(response, 405, { error: 'method_not_allowed' }) + this.audit({ method, path, status: 405, bytes: 0, deviceId: device.deviceId, clientIp }) + return + } + await this.handleSnapshot(request, response, method, path, device.deviceId, clientIp) + return + } + + // Declared by the protocol but not served by this build yet (push, events). + this.respondJson(response, 501, { error: 'not_implemented' }) + this.audit({ method, path, status: 501, bytes: 0, deviceId: device.deviceId, clientIp }) + } catch (error) { + this.deps.logger?.warn('[SyncHost] Request failed', { + method, + path, + error: error instanceof Error ? error.message : String(error) + }) + if (!response.headersSent) this.respondJson(response, 500, { error: 'internal_error' }) + else response.destroy() + this.audit({ method, path, status: 500, bytes: 0, deviceId: authenticatedDeviceId, clientIp }) + } + } + + private async handlePair( + request: http.IncomingMessage, + response: http.ServerResponse, + method: string, + path: string, + clientIp: string | null + ): Promise { + if (method !== 'POST') { + this.respondJson(response, 405, { error: 'method_not_allowed' }) + this.auditAnonymous({ method, path, status: 405, bytes: 0, deviceId: null, clientIp }) + return + } + if (!this.consumeRateLimit(`pair:${clientIp ?? 'unknown'}`)) { + this.respondJson(response, 429, { error: 'rate_limited' }) + this.auditAnonymous({ method, path, status: 429, bytes: 0, deviceId: null, clientIp }) + return + } + + const read = await this.readBody(request, SYNC_HOST_PAIR_BODY_MAX_BYTES) + if (!read.ok) { + const status = read.reason === 'overflow' ? 413 : 400 + const error = read.reason === 'overflow' ? 'payload_too_large' : 'invalid_request' + this.respondJson(response, status, { error }) + this.auditAnonymous({ method, path, status, bytes: 0, deviceId: null, clientIp }) + return + } + + let parsed: unknown + try { + parsed = JSON.parse(read.body) + } catch { + this.respondJson(response, 400, { error: 'invalid_request' }) + this.auditAnonymous({ method, path, status: 400, bytes: 0, deviceId: null, clientIp }) + return + } + const validation = SyncHostPairRequestSchema.safeParse(parsed) + if (!validation.success) { + this.respondJson(response, 400, { error: 'invalid_request' }) + this.auditAnonymous({ method, path, status: 400, bytes: 0, deviceId: null, clientIp }) + return + } + + const failureKey = clientIp ?? 'unknown' + if (!this.consumePairFailureBudget(failureKey, false)) { + this.respondJson(response, 429, { error: 'rate_limited' }) + this.auditAnonymous({ method, path, status: 429, bytes: 0, deviceId: null, clientIp }) + return + } + + // Captured before consuming so a failure after the code was spent can hand it back. + const outstanding = this.deps.pairing.current() + const outcome = this.deps.pairing.consume(validation.data.code) + // Invalid and expired codes share one response so a caller cannot probe code state. Repeated + // failures cost the caller's own budget, never the user's code. + if (outcome !== 'accepted') { + this.consumePairFailureBudget(failureKey, true) + this.respondJson(response, 401, { error: 'pairing_failed' }) + this.auditAnonymous({ method, path, status: 401, bytes: 0, deviceId: null, clientIp }) + return + } + + this.pairFailures.delete(failureKey) + let issued: IssuedSyncHostDevice + try { + issued = await this.deps.devices.issue({ name: validation.data.deviceName }) + } catch (error) { + // The code was spent but no device exists. Burning it would force the user to generate a new + // one for a failure that was not theirs, so it is restored and the error still surfaces. + if (outstanding) this.deps.pairing.restore(outstanding.code, outstanding.expiresAt) + throw error + } + const payload = SyncHostPairResponseSchema.parse({ + deviceId: issued.device.deviceId, + deviceName: issued.device.name, + token: issued.token + }) + const bytes = this.respondJson(response, 200, payload) + this.audit({ method, path, status: 200, bytes, deviceId: issued.device.deviceId, clientIp }) + } + + private async handleStatus( + response: http.ServerResponse, + method: string, + path: string, + deviceId: string, + clientIp: string | null + ): Promise { + const snapshot = await this.deps.snapshotSource.current() + const payload = SyncHostStatusSchema.parse({ + snapshot: snapshot + ? { + fileName: snapshot.fileName, + size: snapshot.size, + sha256: snapshot.sha256, + backupFormatVersion: snapshot.backupFormatVersion, + databaseEncrypted: snapshot.databaseEncrypted + } + : null, + serverTime: Date.now() + }) + const bytes = this.respondJson(response, 200, payload) + this.audit({ method, path, status: 200, bytes, deviceId, clientIp }) + } + + private async handleSnapshot( + request: http.IncomingMessage, + response: http.ServerResponse, + method: string, + path: string, + deviceId: string, + clientIp: string | null + ): Promise { + // The abort hook must be attached before the first await. Resolving the snapshot can take + // seconds on a large package, and if the client disconnects in that window the response's + // `close` has already fired: a listener attached afterwards never runs, so the handler would + // hang forever, leak the read stream and never audit the request. + let gone = this.isResponseGone(response) + const markGone = (): void => { + gone = true + } + response.once('close', markGone) + + const snapshot = await this.deps.snapshotSource.current() + if (gone) { + this.audit({ method, path, status: 499, bytes: 0, deviceId, clientIp }) + return + } + if (!snapshot) { + this.respondJson(response, 404, { error: 'no_snapshot' }) + this.audit({ method, path, status: 404, bytes: 0, deviceId, clientIp }) + return + } + + const rangeHeader = request.headers.range + let selection: RangeSelection | null = null + let status = 200 + if (typeof rangeHeader === 'string' && rangeHeader.trim()) { + selection = this.parseRange(rangeHeader, snapshot.size) + if (!selection) { + response.writeHead(416, { + 'content-range': `bytes */${snapshot.size}`, + 'accept-ranges': 'bytes' + }) + response.end() + this.audit({ method, path, status: 416, bytes: 0, deviceId, clientIp }) + return + } + status = 206 + } + + if (snapshot.size === 0) { + // An empty package cannot be streamed with a range; report it as an empty body rather than + // letting createReadStream reject on a negative end offset. + const bytes = this.respondJson(response, 409, { error: 'empty_snapshot' }) + this.audit({ method, path, status: 409, bytes, deviceId, clientIp }) + return + } + + const start = selection?.start ?? 0 + const end = selection?.end ?? snapshot.size - 1 + const headers: http.OutgoingHttpHeaders = { + 'content-type': 'application/octet-stream', + 'content-length': String(end - start + 1), + 'accept-ranges': 'bytes', + [SYNC_HOST_SNAPSHOT_ID_HEADER]: snapshot.fileName, + [SYNC_HOST_SNAPSHOT_HASH_HEADER]: snapshot.sha256, + 'cache-control': 'no-store' + } + if (status === 206) { + headers['content-range'] = `bytes ${start}-${end}/${snapshot.size}` + } + response.writeHead(status, headers) + if (this.isResponseGone(response)) { + // The peer vanished while the headers were being written; nothing below would ever settle. + this.audit({ method, path, status: 499, bytes: 0, deviceId, clientIp }) + return + } + + let bytesWritten = 0 + let completed = false + response.once('finish', () => { + completed = true + }) + try { + await new Promise((resolve) => { + const stream = fs.createReadStream(snapshot.filePath, { start, end }) + const finish = (): void => { + stream.destroy() + resolve() + } + response.on('close', finish) + stream.on('data', (chunk) => { + bytesWritten += chunk.length + }) + stream.on('error', () => { + // Headers and Content-Length are already sent, so the body cannot be completed honestly. + // Abort the connection instead of leaving the client to wait for the request timeout. + response.destroy() + finish() + }) + stream.on('end', () => { + response.end() + }) + // Protect this connection from slot eviction for as long as the body is being written. + this.markStreaming(request.socket) + stream.pipe(response) + }) + } finally { + // The socket outlives the response on keep-alive: leaving it marked streaming would make it + // unevictable, and 32 such sockets would turn the ceiling into a wall for every newcomer. + this.clearStreaming(request.socket) + } + + this.audit({ + method, + path, + // 499 (client closed request) records an aborted transfer rather than claiming success. + status: completed ? status : 499, + bytes: bytesWritten, + deviceId, + clientIp + }) + } + + private parseRange(header: string, size: number): RangeSelection | null { + const match = /^bytes=(\d*)-(\d*)$/.exec(header.trim()) + if (!match) return null + const [, rawStart, rawEnd] = match + if (rawStart === '' && rawEnd === '') return null + + if (rawStart === '') { + const suffixLength = Number(rawEnd) + if (!Number.isInteger(suffixLength) || suffixLength <= 0) return null + const start = Math.max(0, size - suffixLength) + return { start, end: size - 1 } + } + + const start = Number(rawStart) + const end = rawEnd === '' ? size - 1 : Number(rawEnd) + if (!Number.isInteger(start) || !Number.isInteger(end)) return null + if (start > end || start >= size) return null + return { start, end: Math.min(end, size - 1) } + } + + /** + * Cloudflare sets `cf-connecting-ip` on tunneled requests; a loopback peer may also spoof it, so + * only an IP-literal shaped value is trusted as a limiter/audit key. + */ + private resolveClientIp(request: http.IncomingMessage): string | null { + const header = request.headers['cf-connecting-ip'] + if (typeof header === 'string' && /^[0-9a-fA-F:.]{3,45}$/.test(header.trim())) { + return header.trim() + } + return request.socket.remoteAddress ?? null + } + + private authenticate(request: http.IncomingMessage) { + const header = request.headers.authorization + if (typeof header !== 'string' || !header.startsWith('Bearer ')) return null + return this.deps.devices.authenticate(header.slice('Bearer '.length)) + } + + private consumeRateLimit(key: string): boolean { + const now = Date.now() + this.boundWindowMap( + this.rateWindows, + SYNC_HOST_RATE_LIMIT_MAX_KEYS, + SYNC_HOST_RATE_LIMIT_WINDOW_MS, + now + ) + const window = this.rateWindows.get(key) + if (!window || now - window.windowStart >= SYNC_HOST_RATE_LIMIT_WINDOW_MS) { + this.rateWindows.set(key, { windowStart: now, count: 1 }) + return true + } + window.count += 1 + return window.count <= SYNC_HOST_RATE_LIMIT_REQUESTS_PER_WINDOW + } + + /** + * Per-source pairing failure budget. `charge` records a failure; without it the call is a + * read-only check. A successful pairing clears the budget. + */ + private consumePairFailureBudget(key: string, charge: boolean): boolean { + const now = Date.now() + this.boundWindowMap( + this.pairFailures, + SYNC_HOST_PAIR_FAILURE_MAX_KEYS, + SYNC_HOST_PAIR_FAILURE_WINDOW_MS, + now + ) + const entry = this.pairFailures.get(key) + if (charge) { + if (!entry || now - entry.windowStart >= SYNC_HOST_PAIR_FAILURE_WINDOW_MS) { + this.pairFailures.set(key, { windowStart: now, count: 1 }) + return true + } + entry.count += 1 + return entry.count <= SYNC_HOST_PAIR_MAX_FAILURES_PER_WINDOW + } + if (!entry || now - entry.windowStart >= SYNC_HOST_PAIR_FAILURE_WINDOW_MS) return true + return entry.count < SYNC_HOST_PAIR_MAX_FAILURES_PER_WINDOW + } + + /** + * Keeps a windowed limiter map bounded: expired windows are dropped first, and if the map is still + * full the oldest entry is evicted. Pruning alone is not enough — inside one window nothing is + * expired, so a caller could otherwise grow the map without bound. + */ + private boundWindowMap( + map: Map, + limit: number, + windowMs: number, + now: number + ): void { + if (map.size < limit) return + for (const [key, value] of map) { + if (now - value.windowStart >= windowMs) map.delete(key) + } + while (map.size >= limit) { + const oldest = map.keys().next().value + if (oldest === undefined) break + map.delete(oldest) + } + } + + /** Oldest connection that is not currently streaming a response body, or null if all are busy. */ + private oldestNonStreamingSocket(): net.Socket | null { + for (const [socket, state] of this.sockets) { + if (!state.streaming) return socket + } + return null + } + + private markStreaming(socket: net.Socket): void { + const state = this.sockets.get(socket) + if (state) state.streaming = true + } + + private clearStreaming(socket: net.Socket): void { + const state = this.sockets.get(socket) + if (state) state.streaming = false + } + + /** + * Reads a bounded body. An oversized body is drained but not buffered, so the caller can still + * send a real 413 instead of resetting the connection under the client. + */ + private readBody( + request: http.IncomingMessage, + limit: number + ): Promise<{ ok: true; body: string } | { ok: false; reason: 'overflow' | 'invalid' }> { + return new Promise((resolve) => { + const chunks: Buffer[] = [] + let total = 0 + let settled = false + const done = ( + result: { ok: true; body: string } | { ok: false; reason: 'overflow' | 'invalid' } + ): void => { + if (settled) return + settled = true + resolve(result) + } + request.on('data', (chunk: Buffer) => { + if (settled) return + total += chunk.length + if (total > limit) { + done({ ok: false, reason: 'overflow' }) + return + } + chunks.push(chunk) + }) + request.on('end', () => done({ ok: true, body: Buffer.concat(chunks).toString('utf8') })) + request.on('error', () => done({ ok: false, reason: 'invalid' })) + request.on('aborted', () => done({ ok: false, reason: 'invalid' })) + }) + } + + /** + * Enforces the request-receive deadline per request. + * + * Node's own `requestTimeout` is only evaluated on the connections-checking interval (default + * 30 s), so a stalled body could keep a connection slot for far longer than the configured + * budget. This guard destroys the socket on the deadline; it is cleared once the request has + * been fully received, so it never interferes with a long response stream. + */ + private armRequestGuard(socket: net.Socket): void { + this.clearRequestGuard(socket) + const guard = setTimeout(() => { + this.requestGuards.delete(socket) + socket.destroy() + }, this.requestReceiveTimeoutMs) + guard.unref?.() + this.requestGuards.set(socket, guard) + } + + private clearRequestGuard(socket: net.Socket): void { + const guard = this.requestGuards.get(socket) + if (!guard) return + clearTimeout(guard) + this.requestGuards.delete(socket) + } + + /** + * True when the peer is already gone, so nothing written below could reach it or settle. + * + * `destroyed`/`writableEnded` are checked together with the underlying socket because an aborted + * request can leave the response object alive but unwritable. + */ + private isResponseGone(response: http.ServerResponse): boolean { + return ( + response.destroyed || + response.writableEnded || + response.socket === null || + response.socket?.destroyed === true + ) + } + + private respondJson(response: http.ServerResponse, status: number, body: unknown): number { + if (response.destroyed || response.writableEnded) return 0 + const payload = JSON.stringify(body) + response.writeHead(status, { + 'content-type': 'application/json', + 'content-length': String(Buffer.byteLength(payload)), + 'cache-control': 'no-store' + }) + response.end(payload) + return Buffer.byteLength(payload) + } + + private audit(entry: Omit): void { + this.auditEntries.push({ at: Date.now(), suppressed: 0, ...entry }) + this.trimAudit() + } + + /** + * Records unauthenticated traffic, coalescing repeats from the same source into one entry. + * + * Anyone who learns the tunnel hostname can generate these (handshake included), and at the anon + * request budget a single source would otherwise evict the whole ring in minutes — erasing the + * evidence of earlier probes. One entry per source keeps the signal and bounds the noise. + */ + private auditAnonymous(entry: Omit): void { + const key = entry.clientIp ?? 'unknown' + const existing = this.anonymousAudit.get(key) + if (existing && this.auditEntries.includes(existing)) { + existing.suppressed += 1 + existing.status = entry.status + existing.at = Date.now() + return + } + const created: SyncHostAuditEntry = { at: Date.now(), suppressed: 0, ...entry } + this.auditEntries.push(created) + this.anonymousAudit.set(key, created) + this.trimAudit() + } + + private trimAudit(): void { + if (this.auditEntries.length <= SYNC_HOST_AUDIT_LIMIT) return + const removed = this.auditEntries.splice(0, this.auditEntries.length - SYNC_HOST_AUDIT_LIMIT) + for (const entry of removed) { + const key = entry.clientIp ?? 'unknown' + if (this.anonymousAudit.get(key) === entry) this.anonymousAudit.delete(key) + } + } +} diff --git a/src/main/sync/host/index.ts b/src/main/sync/host/index.ts new file mode 100644 index 000000000..248bab628 --- /dev/null +++ b/src/main/sync/host/index.ts @@ -0,0 +1,284 @@ +import { randomBytes } from 'node:crypto' +import { chmod, mkdir, open, rename, unlink } from 'node:fs/promises' +import path from 'node:path' +import { + SYNC_HOST_BIND_FAILED_ERROR, + SYNC_HOST_PROTOCOL_VERSION, + type SyncHostAuditEntry, + type SyncHostDeviceView +} from '@shared/contracts/syncHost' +import type { SyncBackupInfo } from '@shared/types/sync' +import { SyncHostDeviceStore } from './devices' +import { SyncHostEndpoint, type SyncHostEndpointLogger } from './endpoint' +import { SyncHostPairingAuthority, type SyncHostPairingCode } from './pairing' +import { SyncHostSnapshotSource } from './snapshot' +import { SyncHostStateStore } from './state' + +const ENDPOINT_DIRECTORY = 'sync-host' +const ENDPOINT_DESCRIPTOR_FILENAME = 'endpoint.json' + +export interface SyncHostServiceStatus { + enabled: boolean + running: boolean + port: number | null + hostId: string + deviceCount: number + hasSnapshot: boolean +} + +export interface SyncHostServiceDeps { + listBackups: () => Promise + getFolderPath: () => string + getUserDataPath: () => string + getAppVersion: () => string + logger?: SyncHostEndpointLogger + /** Overridable for tests. */ + requestReceiveTimeoutMs?: number +} + +/** + * Owns host mode: whether the endpoint runs, who may talk to it, and what a slave can fetch. + * + * The endpoint is deliberately loopback-only and started on demand. All host state (enabled flag, + * host identity, device records) lives in a private machine-local file rather than the settings + * store, because settings travel inside backup packages and are merged on import. + * + * Lifecycle transitions are serialized: an enable racing a disable can otherwise leave a listener + * running while host mode reads as disabled. + */ +export class SyncHostService { + private readonly state: SyncHostStateStore + private readonly devices: SyncHostDeviceStore + private readonly pairing: SyncHostPairingAuthority + private readonly snapshotSource: SyncHostSnapshotSource + private readonly endpoint: SyncHostEndpoint + private lifecycle: Promise = Promise.resolve() + /** Identity minted before the first load, so every caller sees the same value. */ + private pendingHostId: string | null = null + + constructor(private readonly deps: SyncHostServiceDeps) { + this.state = new SyncHostStateStore(path.join(deps.getUserDataPath(), ENDPOINT_DIRECTORY)) + this.devices = new SyncHostDeviceStore(this.state) + this.pairing = new SyncHostPairingAuthority(() => this.getHostId()) + this.snapshotSource = new SyncHostSnapshotSource({ + listBackups: deps.listBackups, + getFolderPath: deps.getFolderPath, + logger: deps.logger + }) + this.endpoint = new SyncHostEndpoint({ + devices: this.devices, + pairing: this.pairing, + snapshotSource: this.snapshotSource, + getHostId: () => this.getHostId(), + getAppVersion: deps.getAppVersion, + logger: deps.logger, + requestReceiveTimeoutMs: deps.requestReceiveTimeoutMs + }) + } + + /** Loads machine-local state. Safe to call repeatedly. */ + async initialize(): Promise { + await this.serialize(async () => { + await this.state.load() + if (this.state.snapshot().hostId) { + this.pendingHostId = null + return + } + // Reuse the identity a pre-initialize caller already saw, so the handshake, the pairing + // payload and the persisted file never disagree about who the host is. + const created = this.pendingHostId ?? randomBytes(16).toString('hex') + await this.state.update((state) => { + state.hostId = state.hostId ?? created + }) + this.pendingHostId = null + }) + } + + getEnabled(): boolean { + return this.state.snapshot().enabled + } + + getHostId(): string { + const existing = this.state.snapshot().hostId + if (existing) return existing + // Before `initialize()` completes there is no persisted identity yet; generate one in memory so + // the handshake and pairing authority never observe an empty value. It is memoized (two callers + // in this window must not see two different identities) and deliberately not written here: + // `initialize()` persists it, so a failed write surfaces to its caller instead of leaving the + // cache and the file disagreeing about who this host is. + const created = this.pendingHostId ?? randomBytes(16).toString('hex') + this.pendingHostId = created + return created + } + + async setEnabled(enabled: boolean): Promise { + await this.serialize(async () => { + if (enabled) { + // Start before persisting so a failed bind never leaves host mode marked enabled without a + // listener, and roll the listener back if the flag itself cannot be written. + await this.startInternal() + try { + await this.state.update((state) => { + state.enabled = true + }) + } catch (error) { + await this.stopInternal() + throw error + } + return + } + // Stop after persisting so a crash mid-stop cannot resurrect the listener on next boot. + await this.state.update((state) => { + state.enabled = false + }) + await this.stopInternal() + }) + return this.getStatus() + } + + async start(): Promise { + await this.serialize(() => this.startInternal()) + } + + async stop(): Promise { + await this.serialize(() => this.stopInternal()) + } + + /** Starts the endpoint only when host mode is enabled; safe to call unconditionally at boot. */ + async startIfEnabled(): Promise { + await this.initialize() + if (!this.getEnabled()) return + await this.serialize(async () => { + await this.startInternal() + try { + await this.state.update((state) => { + state.enabled = true + }) + } catch (error) { + await this.stopInternal() + throw error + } + }) + } + + async getStatus(): Promise { + const snapshot = await this.snapshotSource.current() + return { + enabled: this.getEnabled(), + running: this.endpoint.isRunning(), + port: this.endpoint.isRunning() ? this.endpoint.getPort() : null, + hostId: this.getHostId(), + deviceCount: this.devices.count(), + hasSnapshot: snapshot !== null + } + } + + /** Creates a pairing code. Host mode must be running, otherwise there is nothing to pair with. */ + createPairingCode(): SyncHostPairingCode | null { + if (!this.endpoint.isRunning()) return null + return this.pairing.create() + } + + getPairingCode(): SyncHostPairingCode | null { + return this.pairing.current() + } + + listDevices(): SyncHostDeviceView[] { + return this.devices.list() + } + + revokeDevice(deviceId: string): Promise { + return this.devices.revoke(deviceId) + } + + renameDevice(deviceId: string, name: string): Promise { + return this.devices.rename(deviceId, name) + } + + getAuditEntries(): SyncHostAuditEntry[] { + return this.endpoint.getAuditEntries() + } + + private async startInternal(): Promise { + if (this.endpoint.isRunning()) { + // The descriptor may be missing if a previous write failed; rewrite it before returning. + await this.writeEndpointDescriptor() + return + } + try { + await this.endpoint.start() + } catch (error) { + this.deps.logger?.warn('[SyncHost] Failed to bind endpoint', { + error: error instanceof Error ? error.message : String(error) + }) + throw new Error(SYNC_HOST_BIND_FAILED_ERROR) + } + try { + await this.writeEndpointDescriptor() + } catch (error) { + // A listener without a descriptor is unreachable by the tunnel and must not survive. + await this.endpoint.stop() + throw error + } + } + + private async stopInternal(): Promise { + this.pairing.clear() + await this.endpoint.stop() + await this.removeEndpointDescriptor() + await this.state.flush() + } + + private serialize(step: () => Promise): Promise { + const next = this.lifecycle.then(step, step) + this.lifecycle = next.catch(() => undefined) + return next + } + + private descriptorPath(): string { + return path.join(this.deps.getUserDataPath(), ENDPOINT_DIRECTORY, ENDPOINT_DESCRIPTOR_FILENAME) + } + + /** + * Publishes the bound loopback port so the tunnel layer can point `cloudflared` at it without + * guessing. The file is private to the user, like every other local endpoint descriptor. + */ + private async writeEndpointDescriptor(): Promise { + const directory = path.dirname(this.descriptorPath()) + await mkdir(directory, { recursive: true, mode: 0o700 }) + await chmod(directory, 0o700) + const tempPath = `${this.descriptorPath()}.${randomBytes(6).toString('hex')}.tmp` + try { + const handle = await open(tempPath, 'wx', 0o600) + try { + await handle.writeFile( + `${JSON.stringify({ + port: this.endpoint.getPort(), + hostId: this.getHostId(), + protocolVersion: SYNC_HOST_PROTOCOL_VERSION, + pid: process.pid, + startedAt: Date.now() + })}\n`, + 'utf8' + ) + await handle.sync() + } finally { + await handle.close() + } + await chmod(tempPath, 0o600) + await rename(tempPath, this.descriptorPath()) + } catch (error) { + // A failed write must not leave `.tmp` debris next to the descriptor. + await unlink(tempPath).catch(() => undefined) + throw error + } + } + + private async removeEndpointDescriptor(): Promise { + try { + await unlink(this.descriptorPath()) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error + } + } +} diff --git a/src/main/sync/host/pairing.ts b/src/main/sync/host/pairing.ts new file mode 100644 index 000000000..2ae1e4c59 --- /dev/null +++ b/src/main/sync/host/pairing.ts @@ -0,0 +1,108 @@ +import { randomBytes, timingSafeEqual } from 'node:crypto' +import { SYNC_HOST_PAIRING_CODE_TTL_MS } from '@shared/contracts/syncHost' + +/** Unambiguous alphabet: no 0/O/1/I/L so codes survive being read aloud or retyped. */ +const CODE_ALPHABET = 'ABCDEFGHJKMNPQRSTUVWXYZ23456789' +const CODE_LENGTH = 8 + +export interface SyncHostPairingCode { + code: string + hostId: string + expiresAt: number +} + +function createCode(): string { + // Rejection sampling: 248 is the largest multiple of the alphabet length below 256, so discarding + // the top eight byte values removes the modulo bias entirely (the bias was ~1.4% and irrelevant + // against the failure budget, but it is free to avoid). + const limit = Math.floor(256 / CODE_ALPHABET.length) * CODE_ALPHABET.length + let code = '' + while (code.length < CODE_LENGTH) { + for (const byte of randomBytes(CODE_LENGTH)) { + if (byte >= limit) continue + code += CODE_ALPHABET[byte % CODE_ALPHABET.length] + if (code.length === CODE_LENGTH) break + } + } + return code +} + +function normalize(code: string): string { + return code.trim().toUpperCase().replace(/[\s-]/g, '') +} + +function codesEqual(left: string, right: string): boolean { + const leftBuffer = Buffer.from(left) + const rightBuffer = Buffer.from(right) + if (leftBuffer.length !== rightBuffer.length) return false + return timingSafeEqual(leftBuffer, rightBuffer) +} + +/** + * One-time, short-lived pairing codes. Codes live in memory only: a host restart invalidates + * every outstanding code, which is the intended failure direction. + */ +export class SyncHostPairingAuthority { + private code: string | null = null + private expiresAt = 0 + + constructor(private readonly getHostId: () => string) {} + + create(input: { now?: number; ttlMs?: number } = {}): SyncHostPairingCode { + const now = input.now ?? Date.now() + const ttl = input.ttlMs ?? SYNC_HOST_PAIRING_CODE_TTL_MS + this.code = createCode() + this.expiresAt = now + ttl + return this.describe() + } + + current(now: number = Date.now()): SyncHostPairingCode | null { + if (!this.code || this.expiresAt <= now) return null + return this.describe() + } + + /** + * Consumes a presented code. Success is single-use. + * + * Failed attempts never invalidate or block the code. The endpoint is reachable by anyone who + * learns the tunnel hostname, so any global penalty would hand an anonymous caller a permanent + * denial of pairing; brute force is instead bounded per source by the endpoint's failure budget, + * against roughly 40 bits of code entropy. + */ + consume(presented: string, now: number = Date.now()): 'accepted' | 'invalid' | 'expired' { + if (!this.code) return 'expired' + if (this.expiresAt <= now) { + this.clear() + return 'expired' + } + + const candidate = normalize(presented) + if (!candidate || !codesEqual(candidate, this.code)) return 'invalid' + this.clear() + return 'accepted' + } + + /** + * Puts a consumed code back when the pairing it authorized could not be completed (device + * issuance failed). Never clobbers a code the user created in the meantime, and never revives an + * expired one. + */ + restore(code: string, expiresAt: number, now: number = Date.now()): void { + if (this.code || expiresAt <= now) return + this.code = code + this.expiresAt = expiresAt + } + + clear(): void { + this.code = null + this.expiresAt = 0 + } + + private describe(): SyncHostPairingCode { + return { + code: this.code as string, + hostId: this.getHostId(), + expiresAt: this.expiresAt + } + } +} diff --git a/src/main/sync/host/routes.ts b/src/main/sync/host/routes.ts new file mode 100644 index 000000000..b76134fc8 --- /dev/null +++ b/src/main/sync/host/routes.ts @@ -0,0 +1,102 @@ +import { + syncHostCreatePairingCodeRoute, + syncHostGetAuditRoute, + syncHostGetStatusRoute, + syncHostListDevicesRoute, + syncHostRenameDeviceRoute, + syncHostRevokeDeviceRoute, + syncHostSetEnabledRoute +} from '@shared/contracts/routes' +import { + createRouteMap, + requireRendererCaller, + type DeepchatRouteMap +} from '@/routes/routeRegistry' +import type { SyncHostService } from './index' + +/** + * Renderer-facing control surface for host mode. Remote device traffic never uses these routes: + * it arrives on the loopback endpoint and is authorized by device tokens. Every handler asserts a + * renderer caller: enabling host mode opens a network listener and pairing mints device tokens, so + * these must not be reachable from the local control plane even if the surface list changes. + */ +export type SyncHostRoutePort = Pick< + SyncHostService, + | 'getStatus' + | 'getPairingCode' + | 'setEnabled' + | 'createPairingCode' + | 'listDevices' + | 'revokeDevice' + | 'renameDevice' + | 'getAuditEntries' +> + +export function createSyncHostRoutes(deps: { host: SyncHostRoutePort }): DeepchatRouteMap { + return createRouteMap([ + [ + syncHostGetStatusRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + syncHostGetStatusRoute.input.parse(rawInput) + const status = await deps.host.getStatus() + const pairing = deps.host.getPairingCode() + return syncHostGetStatusRoute.output.parse({ status, pairing }) + } + ], + [ + syncHostSetEnabledRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + const input = syncHostSetEnabledRoute.input.parse(rawInput) + const status = await deps.host.setEnabled(input.enabled) + return syncHostSetEnabledRoute.output.parse({ status }) + } + ], + [ + syncHostCreatePairingCodeRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + syncHostCreatePairingCodeRoute.input.parse(rawInput) + const pairing = deps.host.createPairingCode() ?? deps.host.getPairingCode() + return syncHostCreatePairingCodeRoute.output.parse({ pairing }) + } + ], + [ + syncHostListDevicesRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + syncHostListDevicesRoute.input.parse(rawInput) + return syncHostListDevicesRoute.output.parse({ devices: deps.host.listDevices() }) + } + ], + [ + syncHostRevokeDeviceRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + const input = syncHostRevokeDeviceRoute.input.parse(rawInput) + return syncHostRevokeDeviceRoute.output.parse({ + revoked: await deps.host.revokeDevice(input.deviceId) + }) + } + ], + [ + syncHostRenameDeviceRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + const input = syncHostRenameDeviceRoute.input.parse(rawInput) + return syncHostRenameDeviceRoute.output.parse({ + renamed: await deps.host.renameDevice(input.deviceId, input.name) + }) + } + ], + [ + syncHostGetAuditRoute.name, + async (rawInput, context) => { + requireRendererCaller(context) + syncHostGetAuditRoute.input.parse(rawInput) + return syncHostGetAuditRoute.output.parse({ entries: deps.host.getAuditEntries() }) + } + ] + ]) +} diff --git a/src/main/sync/host/snapshot.ts b/src/main/sync/host/snapshot.ts new file mode 100644 index 000000000..a823550a8 --- /dev/null +++ b/src/main/sync/host/snapshot.ts @@ -0,0 +1,264 @@ +import { createHash } from 'node:crypto' +import fs from 'node:fs' +import path from 'node:path' +import { Unzip, UnzipInflate } from 'fflate' +import type { SyncBackupInfo } from '@shared/types/sync' + +export interface SyncHostSnapshot { + fileName: string + filePath: string + size: number + sha256: string + /** `version` from the backup package's own manifest.json; null when it cannot be read. */ + backupFormatVersion: number | null + /** Whether the packaged database is encrypted, which a slave cannot import without the password. */ + databaseEncrypted: boolean +} + +interface CachedDigest { + size: number + mtimeMs: number + /** Inode and status-change time: size+mtime alone cannot tell a replaced file from the same one. */ + ino: number + ctimeMs: number + sha256: string + backupFormatVersion: number | null + databaseEncrypted: boolean +} + +interface BackupManifestShape { + version?: unknown + databaseEncrypted?: unknown +} + +/** + * Ceiling on the manifest entry we are willing to buffer. A real manifest is well under a kilobyte; + * anything larger is a hostile or broken archive, and buffering it would let a small deflate bomb + * inflate into the main process's heap. + */ +const MANIFEST_MAX_BYTES = 1024 * 1024 + +/** + * Only real backup packages may be served. Every other call site in the sync pipeline validates the + * name this way (`sync/index.ts`, `cloudStorageService.ts`); without it, any `*.zip` that lands in + * the (cloud-synced, user-configurable) sync folder would be handed to every paired device as the + * host's snapshot. + */ +const BACKUP_FILE_NAME_REGEX = /^backup-\d+\.zip$/ + +/** + * Extracts only manifest.json from the archive. + * + * The archive is streamed rather than buffered: a backup package can be hundreds of megabytes and a + * transient full-archive read in the main process is a real memory risk. fflate cannot skip an + * entry, so the remaining entries are inflated and discarded — bounded memory at the cost of one + * decompression pass, which only happens when the digest cache misses. + */ +function readManifestEntry(filePath: string): Promise { + return new Promise((resolve) => { + let settled = false + let manifest: BackupManifestShape | null = null + let stream: fs.ReadStream | null = null + const done = (value: BackupManifestShape | null): void => { + if (settled) return + settled = true + resolve(value) + } + + const unzip = new Unzip((file) => { + if (!file.name.endsWith('manifest.json')) { + file.ondata = () => undefined + file.start() + return + } + const chunks: Uint8Array[] = [] + let buffered = 0 + file.ondata = (error, data, final) => { + if (error) return + buffered += data.length + if (buffered > MANIFEST_MAX_BYTES) { + done(null) + stream?.destroy() + return + } + chunks.push(data) + if (!final) return + try { + const parsed = JSON.parse(Buffer.concat(chunks).toString('utf8')) as unknown + manifest = parsed && typeof parsed === 'object' ? (parsed as BackupManifestShape) : null + } catch { + manifest = null + } + } + file.start() + }) + + // fflate only auto-registers stored entries; deflate entries need a codec or `start()` throws. + unzip.register(UnzipInflate) + + stream = fs.createReadStream(filePath, { highWaterMark: 1024 * 1024 }) + stream.on('data', (chunk: string | Buffer) => { + const bytes = typeof chunk === 'string' ? Buffer.from(chunk) : chunk + // A corrupt or hostile archive must not throw out of a stream handler: that would escape to + // the main process and leave the request hanging. + try { + unzip.push(new Uint8Array(bytes.buffer, bytes.byteOffset, bytes.byteLength), false) + } catch { + done(null) + stream?.destroy() + } + }) + stream.on('error', () => done(null)) + stream.on('end', () => { + // The trailing central directory is what finalizes the last entry, so the result is only + // known after the final push has been processed. + try { + unzip.push(new Uint8Array(0), true) + } catch { + done(null) + return + } + setImmediate(() => done(manifest)) + }) + }) +} + +async function digestFile( + filePath: string +): Promise<{ sha256: string; backupFormatVersion: number | null; databaseEncrypted: boolean }> { + const sha256 = await new Promise((resolve, reject) => { + const hash = createHash('sha256') + const stream = fs.createReadStream(filePath, { highWaterMark: 512 * 1024 }) + stream.on('data', (chunk) => hash.update(chunk)) + stream.on('error', reject) + stream.on('end', () => resolve(hash.digest('hex'))) + }) + + const manifest = await readManifestEntry(filePath) + + const version = manifest?.version + return { + sha256, + backupFormatVersion: typeof version === 'number' && Number.isInteger(version) ? version : null, + databaseEncrypted: manifest?.databaseEncrypted === true + } +} + +/** + * Resolves the snapshot a slave would receive: the most recent backup package in the sync folder, + * with its size, content hash and format metadata. Digests are cached per file identity so repeated + * status requests do not re-hash a large archive. + */ +export class SyncHostSnapshotSource { + private readonly digests = new Map() + private readonly inFlight = new Map>() + + constructor( + private readonly deps: { + listBackups: () => Promise + getFolderPath: () => string + logger?: { warn(message: string, meta?: unknown): void } + } + ) {} + + /** + * Concurrent callers (several slaves, or a status poll during a download) share one digest pass, + * so N simultaneous requests cannot each hash the archive and multiply memory and I/O. + */ + async current(): Promise { + const pending = this.inFlight.get('current') + if (pending) return pending + const run = this.resolveCurrent().finally(() => { + this.inFlight.delete('current') + }) + this.inFlight.set('current', run) + return run + } + + /** + * Storage failures are "no snapshot", not a failed request. A package that vanishes or is locked + * mid-scan is a transient condition the slave can retry: it must see `snapshot: null` / 404, not + * a 500 from the host's own filesystem. + */ + private async resolveCurrent(): Promise { + try { + return await this.resolveCurrentInner() + } catch (error) { + this.deps.logger?.warn('[SyncHost] Snapshot resolution failed', { + error: error instanceof Error ? error.message : String(error) + }) + return null + } + } + + private async resolveCurrentInner(): Promise { + const backups = (await this.deps.listBackups()).filter((backup) => + BACKUP_FILE_NAME_REGEX.test(backup.fileName) + ) + if (backups.length === 0) return null + const latest = [...backups].sort((left, right) => right.createdAt - left.createdAt)[0] + const filePath = path.join(this.deps.getFolderPath(), latest.fileName) + + let stat: fs.Stats + try { + stat = await fs.promises.stat(filePath) + } catch { + return null + } + + const cached = this.digests.get(latest.fileName) + if ( + cached && + cached.size === stat.size && + cached.mtimeMs === stat.mtimeMs && + cached.ino === stat.ino && + cached.ctimeMs === stat.ctimeMs + ) { + return { + fileName: latest.fileName, + filePath, + size: stat.size, + sha256: cached.sha256, + backupFormatVersion: cached.backupFormatVersion, + databaseEncrypted: cached.databaseEncrypted + } + } + + for (let attempt = 0; attempt < 2; attempt += 1) { + const digest = await digestFile(filePath) + const after = await fs.promises.stat(filePath).catch(() => null) + if (!after) { + // The package disappeared while it was read; retrying against a stale stat would only + // digest a missing file again. + return null + } + if ( + after.size !== stat.size || + after.mtimeMs !== stat.mtimeMs || + after.ino !== stat.ino || + after.ctimeMs !== stat.ctimeMs + ) { + // The package changed while it was read: never report a hash for bytes we did not measure. + if (attempt === 1) return null + stat = after + continue + } + this.digests.set(latest.fileName, { + size: stat.size, + mtimeMs: stat.mtimeMs, + ino: stat.ino, + ctimeMs: stat.ctimeMs, + ...digest + }) + this.forgetOtherEntries(latest.fileName) + return { fileName: latest.fileName, filePath, size: stat.size, ...digest } + } + return null + } + + private forgetOtherEntries(keepFileName: string): void { + for (const key of this.digests.keys()) { + if (key !== keepFileName) this.digests.delete(key) + } + } +} diff --git a/src/main/sync/host/state.ts b/src/main/sync/host/state.ts new file mode 100644 index 000000000..ff24a99fb --- /dev/null +++ b/src/main/sync/host/state.ts @@ -0,0 +1,176 @@ +import { randomBytes } from 'node:crypto' +import fs from 'node:fs' +import path from 'node:path' + +const STATE_FILENAME = 'host-state.json' + +export interface SyncHostDeviceRecord { + deviceId: string + name: string + /** SHA-256 of the device bearer token. The token itself is never persisted. */ + tokenHash: string + createdAt: number + lastSeenAt: number | null + expiresAt: number | null + revokedAt: number | null +} + +export interface SyncHostState { + enabled: boolean + hostId: string | null + devices: SyncHostDeviceRecord[] +} + +const DEFAULT_STATE: SyncHostState = { enabled: false, hostId: null, devices: [] } + +/** + * Machine-local state for host mode, stored as a private file instead of a settings key. + * + * This is deliberate. Settings flow into backup packages (`configs/app-settings.json`), into + * S3/R2 uploads, and into every peer that imports a package, and import merges settings wholesale. + * Device token hashes, the host identity and the enabled flag must never travel: an imported copy + * would let an importer accept another host's device tokens, resurrect revoked devices, and enable + * host mode without the user's consent. Keeping them in their own file removes that path entirely. + * + * Reads are served from an in-memory cache so request-time authentication stays synchronous; writes + * are serialized, atomic (temp + rename), and `0600`. + */ +export class SyncHostStateStore { + private state: SyncHostState = { ...DEFAULT_STATE } + private loaded = false + private loadChain: Promise | null = null + /** + * Serializes whole update transactions — load, snapshot, mutation, write, rollback — not just the + * filesystem writes. Serializing writes alone is not enough: a second update could snapshot the + * first one's mutation before it failed, which would both block the rollback and persist the + * change whose caller was told it failed. + */ + private updateChain: Promise = Promise.resolve() + + constructor(private readonly directory: string) {} + + get filePath(): string { + return path.join(this.directory, STATE_FILENAME) + } + + /** + * Reads state from disk once. Concurrent callers share the same read. + */ + async load(): Promise { + if (this.loaded) return this.snapshot() + if (!this.loadChain) this.loadChain = this.loadFromDisk() + return this.loadChain + } + + private async loadFromDisk(): Promise { + let parsed: unknown = null + try { + parsed = JSON.parse(await fs.promises.readFile(this.filePath, 'utf8')) as unknown + } catch (error) { + const code = (error as NodeJS.ErrnoException).code + if (code !== 'ENOENT') { + // A read or parse failure must never degrade to the empty default state: `initialize()` + // would then persist that default over a file that still holds every device record and the + // enabled flag, destroying pairing because of a transient EACCES/EIO/AV lock. Fail closed, + // keep `loaded` false, and let the next call retry. + this.loadChain = null + throw error + } + } + this.state = this.normalize(parsed) + this.loaded = true + return this.snapshot() + } + + isLoaded(): boolean { + return this.loaded + } + + snapshot(): SyncHostState { + return { + ...this.state, + devices: this.state.devices.map((record) => ({ ...record })) + } + } + + /** + * Applies a mutation to the cached state and persists it atomically. + * + * Loading first is not an optimisation, it is a correctness requirement: a mutation that arrives + * before the initial read (a renderer call landing between route registration and app boot) would + * otherwise persist the empty default state over the real file, discarding every device record + * and the enabled flag. + */ + async update(mutator: (state: SyncHostState) => void): Promise { + const update = this.updateChain.then(async () => { + if (!this.loaded) await this.load() + const previous = this.state + const next = this.snapshot() + mutator(next) + this.state = next + try { + // The caller must see write failures: a revocation that silently failed to persist would + // come back to life after a restart. + await this.writeAtomic(`${JSON.stringify(next, null, 2)}\n`) + } catch (error) { + // Roll the cache back: memory must never claim a change that is not on disk, or a later + // successful write (a last-seen touch, a rename) would silently persist a mutation whose + // caller was told it failed. Safe unconditionally because updates are serialized. + this.state = previous + throw error + } + }) + // Keep the chain usable after a rejection while still surfacing the failure to this caller. + this.updateChain = update.catch(() => undefined) + return update + } + + /** Resolves once every queued update has settled; used by teardown and tests. */ + async flush(): Promise { + await this.updateChain + } + + private async writeAtomic(payload: string): Promise { + await fs.promises.mkdir(this.directory, { recursive: true, mode: 0o700 }) + try { + await fs.promises.chmod(this.directory, 0o700) + } catch { + // Best effort: the directory may live on a filesystem without POSIX modes. + } + const tempPath = `${this.filePath}.${randomBytes(6).toString('hex')}.tmp` + try { + const handle = await fs.promises.open(tempPath, 'wx', 0o600) + try { + await handle.writeFile(payload, 'utf8') + await handle.sync() + } finally { + await handle.close() + } + await fs.promises.chmod(tempPath, 0o600) + await fs.promises.rename(tempPath, this.filePath) + } catch (error) { + // A failed write must not leave `.tmp` debris next to the state file. + await fs.promises.rm(tempPath, { force: true }).catch(() => undefined) + throw error + } + } + + private normalize(parsed: unknown): SyncHostState { + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return { ...DEFAULT_STATE } + const record = parsed as Partial + return { + enabled: record.enabled === true, + hostId: typeof record.hostId === 'string' && record.hostId.length > 0 ? record.hostId : null, + devices: Array.isArray(record.devices) + ? record.devices.filter( + (device): device is SyncHostDeviceRecord => + Boolean(device) && + typeof device.deviceId === 'string' && + typeof device.name === 'string' && + typeof device.tokenHash === 'string' && + typeof device.createdAt === 'number' + ) + : [] + } + } +} diff --git a/src/shared/contracts/routes.ts b/src/shared/contracts/routes.ts index 75533556c..455d6b832 100644 --- a/src/shared/contracts/routes.ts +++ b/src/shared/contracts/routes.ts @@ -570,6 +570,15 @@ import { skillSyncGetRegisteredToolsRoute, skillSyncScanExternalToolsRoute } from './routes/skillSync.routes' +import { + syncHostCreatePairingCodeRoute, + syncHostGetAuditRoute, + syncHostGetStatusRoute, + syncHostListDevicesRoute, + syncHostRenameDeviceRoute, + syncHostRevokeDeviceRoute, + syncHostSetEnabledRoute +} from './routes/syncHost.routes' import { syncGetBackupStatusRoute, syncImportRoute, @@ -682,6 +691,7 @@ export * from './routes/sessions.routes' export * from './routes/skills.routes' export * from './routes/skillSync.routes' export * from './routes/sync.routes' +export * from './routes/syncHost.routes' export * from './routes/system.routes' export * from './routes/toolchains.routes' export * from './routes/tab.routes' @@ -1279,7 +1289,14 @@ const DEEPCHAT_ROUTE_CATALOG_PART_5 = { [toolchainsCancelInstallRoute.name]: toolchainsCancelInstallRoute, [toolchainsRepairRoute.name]: toolchainsRepairRoute, [toolchainsRevertRoute.name]: toolchainsRevertRoute, - [toolchainsPickCustomRoute.name]: toolchainsPickCustomRoute + [toolchainsPickCustomRoute.name]: toolchainsPickCustomRoute, + [syncHostGetStatusRoute.name]: syncHostGetStatusRoute, + [syncHostSetEnabledRoute.name]: syncHostSetEnabledRoute, + [syncHostCreatePairingCodeRoute.name]: syncHostCreatePairingCodeRoute, + [syncHostListDevicesRoute.name]: syncHostListDevicesRoute, + [syncHostRevokeDeviceRoute.name]: syncHostRevokeDeviceRoute, + [syncHostRenameDeviceRoute.name]: syncHostRenameDeviceRoute, + [syncHostGetAuditRoute.name]: syncHostGetAuditRoute } satisfies Record export type DeepchatRouteCatalog = typeof DEEPCHAT_ROUTE_CATALOG_PART_1 & diff --git a/src/shared/contracts/routes/syncHost.routes.ts b/src/shared/contracts/routes/syncHost.routes.ts new file mode 100644 index 000000000..dae57ad35 --- /dev/null +++ b/src/shared/contracts/routes/syncHost.routes.ts @@ -0,0 +1,82 @@ +import { z } from 'zod' +import { SyncHostAuditEntrySchema, SyncHostDeviceViewSchema } from '../syncHost' +import { defineRouteContract } from '../common' + +const SyncHostStatusViewSchema = z.object({ + enabled: z.boolean(), + running: z.boolean(), + port: z.number().int().positive().nullable(), + hostId: z.string(), + deviceCount: z.number().int().nonnegative(), + hasSnapshot: z.boolean() +}) + +const SyncHostPairingViewSchema = z.object({ + code: z.string(), + hostId: z.string(), + expiresAt: z.number().int().nonnegative() +}) + +export const syncHostGetStatusRoute = defineRouteContract({ + name: 'syncHost.getStatus', + input: z.object({}).default({}), + output: z.object({ + status: SyncHostStatusViewSchema, + pairing: SyncHostPairingViewSchema.nullable() + }) +}) + +export const syncHostSetEnabledRoute = defineRouteContract({ + name: 'syncHost.setEnabled', + input: z.object({ + enabled: z.boolean() + }), + output: z.object({ + status: SyncHostStatusViewSchema + }) +}) + +export const syncHostCreatePairingCodeRoute = defineRouteContract({ + name: 'syncHost.createPairingCode', + input: z.object({}).default({}), + output: z.object({ + pairing: SyncHostPairingViewSchema.nullable() + }) +}) + +export const syncHostListDevicesRoute = defineRouteContract({ + name: 'syncHost.listDevices', + input: z.object({}).default({}), + output: z.object({ + devices: z.array(SyncHostDeviceViewSchema) + }) +}) + +export const syncHostRevokeDeviceRoute = defineRouteContract({ + name: 'syncHost.revokeDevice', + input: z.object({ + deviceId: z.string().min(1) + }), + output: z.object({ + revoked: z.boolean() + }) +}) + +export const syncHostRenameDeviceRoute = defineRouteContract({ + name: 'syncHost.renameDevice', + input: z.object({ + deviceId: z.string().min(1), + name: z.string().min(1) + }), + output: z.object({ + renamed: z.boolean() + }) +}) + +export const syncHostGetAuditRoute = defineRouteContract({ + name: 'syncHost.getAudit', + input: z.object({}).default({}), + output: z.object({ + entries: z.array(SyncHostAuditEntrySchema) + }) +}) diff --git a/src/shared/contracts/syncHost.ts b/src/shared/contracts/syncHost.ts new file mode 100644 index 000000000..c6c56a446 --- /dev/null +++ b/src/shared/contracts/syncHost.ts @@ -0,0 +1,135 @@ +import { z } from 'zod' + +/** + * Wire contract for the Cloudflare Tunnel host sync endpoint. + * + * The host serves these routes on a loopback listener that a user-operated `cloudflared` + * tunnel forwards to. `handshake` is the only unauthenticated route; every other route + * requires a per-device bearer token issued through pairing. + */ +export const SYNC_HOST_PROTOCOL_VERSION = 1 as const +export const SYNC_HOST_PROTOCOL_NAME = 'sync/v1' as const + +export const SYNC_HOST_PATH_PREFIX = '/sync/v1' +export const SYNC_HOST_HANDSHAKE_PATH = `${SYNC_HOST_PATH_PREFIX}/handshake` +export const SYNC_HOST_PAIR_PATH = `${SYNC_HOST_PATH_PREFIX}/pair` +export const SYNC_HOST_STATUS_PATH = `${SYNC_HOST_PATH_PREFIX}/status` +export const SYNC_HOST_SNAPSHOT_PATH = `${SYNC_HOST_PATH_PREFIX}/snapshot` +export const SYNC_HOST_PUSH_PATH = `${SYNC_HOST_PATH_PREFIX}/push` +export const SYNC_HOST_EVENTS_PATH = `${SYNC_HOST_PATH_PREFIX}/events` + +export const SYNC_HOST_MAX_HEADER_BYTES = 8 * 1024 +/** + * Connection ceiling. It is a memory bound, not a quota: when it is reached the endpoint evicts the + * oldest connection that is not streaming a response rather than refusing the newcomer, so a caller + * holding stalled connections cannot deny the legitimate device (or the user's own pairing). + */ +export const SYNC_HOST_MAX_CONNECTIONS = 32 +/** + * Budget for *receiving* a request (headers plus body). It does not bound how long a response may + * stream, so a large snapshot download is unaffected, while a stalled request is discarded long + * before it can hold a connection slot indefinitely. + */ +export const SYNC_HOST_REQUEST_RECEIVE_TIMEOUT_MS = 60_000 +export const SYNC_HOST_PAIR_BODY_MAX_BYTES = 4 * 1024 +export const SYNC_HOST_RATE_LIMIT_WINDOW_MS = 60_000 +export const SYNC_HOST_RATE_LIMIT_REQUESTS_PER_WINDOW = 120 +export const SYNC_HOST_RATE_LIMIT_MAX_KEYS = 1024 +export const SYNC_HOST_PAIRING_CODE_TTL_MS = 5 * 60_000 +/** + * Per-source pairing failure budget. There is deliberately no global attempt cap and no + * `attemptsRemaining` in the pairing payload: anyone who learns the tunnel hostname can call + * `pair`, so a global counter would both hand them a denial of pairing and let them drive a + * number the UI shows. Brute force is bounded per source against ~40 bits of code entropy. + */ +export const SYNC_HOST_PAIR_FAILURE_WINDOW_MS = 5 * 60_000 +export const SYNC_HOST_PAIR_MAX_FAILURES_PER_WINDOW = 20 +export const SYNC_HOST_PAIR_FAILURE_MAX_KEYS = 1024 +export const SYNC_HOST_DEVICE_TOKEN_BYTES = 32 +export const SYNC_HOST_DEVICE_NAME_MAX_LENGTH = 120 +export const SYNC_HOST_MAX_PUSH_PART_BYTES = 32 * 1024 * 1024 +export const SYNC_HOST_AUDIT_LIMIT = 500 + +/** User-visible failure codes surfaced to the renderer; copy is added with the Settings UI. */ +export const SYNC_HOST_BIND_FAILED_ERROR = 'syncHost.error.bindFailed' + +export const SYNC_HOST_SNAPSHOT_ID_HEADER = 'x-deepchat-snapshot-id' +export const SYNC_HOST_SNAPSHOT_HASH_HEADER = 'x-deepchat-snapshot-sha256' +export const SYNC_HOST_DEVICE_HEADER = 'x-deepchat-device-id' + +export const SyncHostCapabilitySchema = z.enum(['snapshot', 'range', 'push', 'events']) +export type SyncHostCapability = z.infer + +export const SyncHostHandshakeSchema = z.object({ + protocol: z.literal(SYNC_HOST_PROTOCOL_NAME), + protocolVersion: z.number().int().positive(), + /** Stable opaque host identity, also carried by pairing payloads for out-of-band verification. */ + hostId: z.string(), + appVersion: z.string(), + capabilities: z.array(SyncHostCapabilitySchema), + encryption: z.object({ + payload: z.literal('none'), + transport: z.literal('tls') + }) +}) +export type SyncHostHandshake = z.infer + +export const SyncHostSnapshotInfoSchema = z.object({ + fileName: z.string(), + size: z.number().int().nonnegative(), + sha256: z.string(), + /** `version` from the backup package manifest; null when the archive cannot be read. */ + backupFormatVersion: z.number().int().nonnegative().nullable(), + /** + * Whether the packaged database is encrypted. A slave cannot import an encrypted package without + * the password, so this must be reported rather than discovered as a decrypt failure. + */ + databaseEncrypted: z.boolean() +}) +export type SyncHostSnapshotInfo = z.infer + +export const SyncHostStatusSchema = z.object({ + snapshot: SyncHostSnapshotInfoSchema.nullable(), + serverTime: z.number().int().nonnegative() +}) +export type SyncHostStatus = z.infer + +export const SyncHostPairRequestSchema = z.object({ + code: z.string().min(1).max(256), + deviceName: z.string().trim().min(1).max(SYNC_HOST_DEVICE_NAME_MAX_LENGTH) +}) +export type SyncHostPairRequest = z.infer + +export const SyncHostPairResponseSchema = z.object({ + deviceId: z.string(), + deviceName: z.string(), + token: z.string() +}) +export type SyncHostPairResponse = z.infer + +/** Device view exposed to the renderer. Never contains token material or token hashes. */ +export const SyncHostDeviceViewSchema = z.object({ + deviceId: z.string(), + name: z.string(), + createdAt: z.number().int().nonnegative(), + expiresAt: z.number().int().nonnegative().nullable(), + lastSeenAt: z.number().int().nonnegative().nullable(), + revoked: z.boolean() +}) +export type SyncHostDeviceView = z.infer + +export const SyncHostAuditEntrySchema = z.object({ + at: z.number().int().nonnegative(), + method: z.string(), + path: z.string(), + status: z.number().int(), + bytes: z.number().int().nonnegative(), + deviceId: z.string().nullable(), + clientIp: z.string().nullable(), + /** + * Rejections from one anonymous source are coalesced into a single entry that counts the repeats, + * so unauthenticated traffic cannot flush the audit ring by evicting everything else. + */ + suppressed: z.number().int().nonnegative() +}) +export type SyncHostAuditEntry = z.infer diff --git a/test/main/sync/host/hostEndpoint.test.ts b/test/main/sync/host/hostEndpoint.test.ts new file mode 100644 index 000000000..e5691c8e1 --- /dev/null +++ b/test/main/sync/host/hostEndpoint.test.ts @@ -0,0 +1,758 @@ +import { randomBytes } from 'node:crypto' +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises' +import http from 'node:http' +import { connect } from 'node:net' +import os from 'node:os' +import path from 'node:path' +import { strToU8, zipSync } from 'fflate' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +// This suite exercises a real loopback listener against real files on disk, so the global +// partial `fs` mock from test/setup.ts must not apply here. +vi.unmock('fs') +vi.unmock('node:fs') + +import { SYNC_HOST_MAX_CONNECTIONS, SYNC_HOST_PATH_PREFIX } from '@shared/contracts/syncHost' +import type { SyncBackupInfo } from '@shared/types/sync' +import { SyncHostService } from '@/sync/host' +import { SyncHostPairingAuthority } from '@/sync/host/pairing' + +const BACKUP_FILE_NAME = 'backup-1700000000000.zip' +const BACKUP_FORMAT_VERSION = 3 +const DB_PAYLOAD_BYTES = 1_500_000 +const REQUEST_RECEIVE_TIMEOUT_MS = 400 + +/** + * Exercises the real loopback listener: host mode is a security boundary, so these tests assert + * observable HTTP behavior rather than internal wiring. + */ +describe('SyncHostService endpoint', () => { + let tempDir: string + let syncDir: string + let service: SyncHostService + let baseUrl: string + let backupBytes: Buffer + let listBackups: () => Promise + + beforeEach(async () => { + tempDir = await mkdtemp(path.join(os.tmpdir(), 'deepchat-sync-host-')) + syncDir = path.join(tempDir, 'sync') + await mkdir(syncDir, { recursive: true }) + + // Random (and therefore incompressible) so the package keeps its real size; a repetitive + // payload compresses to a few kilobytes and hides streaming and resume behaviour. + const payload = randomBytes(DB_PAYLOAD_BYTES) + const archive = zipSync({ + 'manifest.json': strToU8( + JSON.stringify({ version: BACKUP_FORMAT_VERSION, databaseEncrypted: false }) + ), + 'agent.db': new Uint8Array(payload) + }) + backupBytes = Buffer.from(archive) + await writeFile(path.join(syncDir, BACKUP_FILE_NAME), backupBytes) + + listBackups = async () => [ + { fileName: BACKUP_FILE_NAME, createdAt: 1_700_000_000_000, size: backupBytes.length } + ] + service = new SyncHostService({ + listBackups: () => listBackups(), + getFolderPath: () => syncDir, + getUserDataPath: () => tempDir, + getAppVersion: () => '9.9.9', + requestReceiveTimeoutMs: REQUEST_RECEIVE_TIMEOUT_MS + }) + await service.initialize() + await service.start() + const started = await service.getStatus() + baseUrl = `http://127.0.0.1:${started.port}` + }) + + afterEach(async () => { + await service.stop() + await rm(tempDir, { recursive: true, force: true }) + }) + + async function pairDevice(name = 'Laptop'): Promise<{ deviceId: string; token: string }> { + const pairing = service.createPairingCode() + expect(pairing).not.toBeNull() + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: pairing?.code, deviceName: name }) + }) + expect(response.status).toBe(200) + return (await response.json()) as { deviceId: string; token: string } + } + + it('serves handshake without authentication and advertises only real capabilities', async () => { + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/handshake`) + expect(response.status).toBe(200) + const body = (await response.json()) as { + protocol: string + hostId: string + capabilities: string[] + encryption: { transport: string } + } + expect(body.protocol).toBe('sync/v1') + expect(body.hostId).toBe(service.getHostId()) + expect(body.capabilities).toEqual(['snapshot', 'range']) + expect(body.encryption.transport).toBe('tls') + }) + + it('answers every unauthenticated request with one uniform 401 regardless of path or method', async () => { + const probes: Array<[string, string]> = [ + ['GET', `${SYNC_HOST_PATH_PREFIX}/status`], + ['GET', `${SYNC_HOST_PATH_PREFIX}/snapshot`], + ['PUT', `${SYNC_HOST_PATH_PREFIX}/status`], + ['DELETE', `${SYNC_HOST_PATH_PREFIX}/snapshot`], + ['GET', `${SYNC_HOST_PATH_PREFIX}/push`], + ['GET', '/secret'], + ['POST', `${SYNC_HOST_PATH_PREFIX}/handshake`] + ] + for (const [method, url] of probes) { + const response = await fetch(`${baseUrl}${url}`, { method }) + expect({ method, url, status: response.status }).toEqual({ method, url, status: 401 }) + } + + const forged = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${'a'.repeat(43)}` } + }) + expect(forged.status).toBe(401) + const wrongScheme = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { authorization: `Basic ${'a'.repeat(43)}` } + }) + expect(wrongScheme.status).toBe(401) + }) + + it('distinguishes unknown routes and methods only after authentication', async () => { + const { token } = await pairDevice() + const headers = { authorization: `Bearer ${token}` } + + const unknown = await fetch(`${baseUrl}/secret`, { headers }) + expect(unknown.status).toBe(404) + const wrongMethod = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + method: 'PUT', + headers + }) + expect(wrongMethod.status).toBe(405) + const notImplemented = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/push`, { + method: 'POST', + headers + }) + expect(notImplemented.status).toBe(501) + }) + + it('issues a single-use pairing code and reports snapshot metadata to the paired device', async () => { + const pairing = service.createPairingCode() + expect(pairing).not.toBeNull() + + const wrongCode = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: 'WRONGCODE', deviceName: 'Laptop' }) + }) + expect(wrongCode.status).toBe(401) + + const blankName = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: pairing?.code, deviceName: ' ' }) + }) + expect(blankName.status).toBe(400) + + const paired = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: pairing?.code, deviceName: 'Laptop' }) + }) + expect(paired.status).toBe(200) + const issued = (await paired.json()) as { deviceId: string; token: string } + expect(issued.token).toHaveLength(43) + + const reused = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: pairing?.code, deviceName: 'Second' }) + }) + expect(reused.status).toBe(401) + + const status = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${issued.token}` } + }) + expect(status.status).toBe(200) + const body = (await status.json()) as { + snapshot: { + fileName: string + size: number + sha256: string + backupFormatVersion: number + databaseEncrypted: boolean + } + } + expect(body.snapshot.fileName).toBe(BACKUP_FILE_NAME) + expect(body.snapshot.size).toBe(backupBytes.length) + expect(body.snapshot.backupFormatVersion).toBe(BACKUP_FORMAT_VERSION) + expect(body.snapshot.databaseEncrypted).toBe(false) + expect(body.snapshot.sha256).toMatch(/^[0-9a-f]{64}$/) + }) + + it('rejects a revoked device token immediately', async () => { + const issued = await pairDevice() + const before = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${issued.token}` } + }) + expect(before.status).toBe(200) + + expect(await service.revokeDevice(issued.deviceId)).toBe(true) + + const after = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${issued.token}` } + }) + expect(after.status).toBe(401) + expect(service.listDevices()).toEqual([ + expect.objectContaining({ deviceId: issued.deviceId, revoked: true }) + ]) + }) + + it('keeps device token hashes out of the settings store and out of plaintext state', async () => { + const issued = await pairDevice() + const stateFile = path.join(tempDir, 'sync-host', 'host-state.json') + const state = await readFile(stateFile, 'utf8') + const mode = (await stat(stateFile)).mode & 0o777 + + expect(state).not.toContain(issued.token) + expect(JSON.parse(state).devices[0].tokenHash).toMatch(/^[0-9a-f]{64}$/) + expect(mode).toBe(0o600) + }) + + it('streams the snapshot whole and resumes it with byte-exact ranges', async () => { + const { token } = await pairDevice() + const headers = { authorization: `Bearer ${token}` } + + const full = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { headers }) + expect(full.status).toBe(200) + expect(full.headers.get('accept-ranges')).toBe('bytes') + const fullBytes = Buffer.from(await full.arrayBuffer()) + expect(fullBytes.equals(backupBytes)).toBe(true) + + const ranged = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { ...headers, range: 'bytes=100-199' } + }) + expect(ranged.status).toBe(206) + expect(ranged.headers.get('content-range')).toBe(`bytes 100-199/${backupBytes.length}`) + const rangedBytes = Buffer.from(await ranged.arrayBuffer()) + expect(rangedBytes.equals(backupBytes.subarray(100, 200))).toBe(true) + + const resumed = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { ...headers, range: `bytes=${backupBytes.length - 10}-` } + }) + expect(resumed.status).toBe(206) + const resumedBytes = Buffer.from(await resumed.arrayBuffer()) + expect(resumedBytes.equals(backupBytes.subarray(backupBytes.length - 10))).toBe(true) + + const unsatisfiable = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { ...headers, range: `bytes=${backupBytes.length + 5}-` } + }) + expect(unsatisfiable.status).toBe(416) + expect(unsatisfiable.headers.get('content-range')).toBe(`bytes */${backupBytes.length}`) + }) + + it('resumes a download that was aborted mid-stream', async () => { + const { token } = await pairDevice() + const headers = { authorization: `Bearer ${token}` } + + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { headers }) + const reader = response.body?.getReader() + expect(reader).toBeDefined() + const first = await reader!.read() + expect(first.done).toBe(false) + const partial = Buffer.from(first.value) + const received = partial.length + await reader!.cancel() + expect(received).toBeLessThan(backupBytes.length) + + const resumed = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { ...headers, range: `bytes=${received}-` } + }) + expect(resumed.status).toBe(206) + const rest = Buffer.from(await resumed.arrayBuffer()) + expect(Buffer.concat([partial, rest]).equals(backupBytes)).toBe(true) + }) + + it('is reachable only on loopback', async () => { + const { port } = await service.getStatus() + + const loopback = await fetch(`http://127.0.0.1:${port}${SYNC_HOST_PATH_PREFIX}/handshake`) + expect(loopback.status).toBe(200) + + // 127.0.0.2 is loopback but a different address, so it proves the bind is address-specific + // rather than a wildcard bind. Unlike an external-interface probe it exists on every host, so + // the assertion can never silently skip. + await expect( + fetch(`http://127.0.0.2:${port}${SYNC_HOST_PATH_PREFIX}/handshake`, { + signal: AbortSignal.timeout(2_000) + }) + ).rejects.toThrow() + + const external = Object.values(os.networkInterfaces()) + .flat() + .find((entry) => entry && entry.family === 'IPv4' && !entry.internal) + if (external) { + await expect( + fetch(`http://${external.address}:${port}${SYNC_HOST_PATH_PREFIX}/handshake`, { + signal: AbortSignal.timeout(1_000) + }) + ).rejects.toThrow() + } + }) + + it('lets a new caller in when the connection ceiling is full of stalled sockets', async () => { + const { port } = await service.getStatus() + const stalled: ReturnType[] = [] + for (let index = 0; index < SYNC_HOST_MAX_CONNECTIONS; index += 1) { + const socket = connect({ host: '127.0.0.1', port }) + stalled.push(socket) + await new Promise((resolve) => socket.once('connect', () => resolve())) + // Announce a body that never arrives: these occupy slots without ever completing a request. + socket.write( + `POST ${SYNC_HOST_PATH_PREFIX}/pair HTTP/1.1\r\nHost: 127.0.0.1\r\n` + + `Content-Type: application/json\r\nContent-Length: 100000\r\n\r\n` + ) + } + try { + // The newcomer must be admitted by evicting a stalled connection, not refused: otherwise + // anyone who learns the hostname can deny the user's own pairing with idle sockets. + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/handshake`, { + signal: AbortSignal.timeout(3_000) + }) + expect(response.status).toBe(200) + } finally { + for (const socket of stalled) socket.destroy() + } + }) + + it('lets a newcomer in after the ceiling is filled with finished keep-alive downloads', async () => { + const { token } = await pairDevice() + // Range 0-0 keeps each download one byte while still exercising the streaming path. + const agent = new http.Agent({ keepAlive: true, maxSockets: SYNC_HOST_MAX_CONNECTIONS }) + try { + await Promise.all( + Array.from({ length: SYNC_HOST_MAX_CONNECTIONS }, async () => { + const { port } = await service.getStatus() + await new Promise((resolve, reject) => { + const request = http.get( + { + host: '127.0.0.1', + port, + path: `${SYNC_HOST_PATH_PREFIX}/snapshot`, + agent, + headers: { authorization: `Bearer ${token}`, range: 'bytes=0-0' } + }, + (response) => { + response.resume() + response.on('end', () => resolve()) + response.on('error', reject) + } + ) + request.on('error', reject) + }) + }) + ) + + // The download sockets are still open (keep-alive). If a finished download left its socket + // marked as streaming, every slot would be unevictable and this request would be refused. + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/handshake`, { + signal: AbortSignal.timeout(3_000) + }) + expect(response.status).toBe(200) + } finally { + agent.destroy() + } + }) + + it('charges an authenticated device for unknown paths instead of letting it flush the audit ring', async () => { + const { token } = await pairDevice() + const headers = { authorization: `Bearer ${token}` } + + let throttled = false + for (let attempt = 0; attempt < 200; attempt += 1) { + const response = await fetch(`${baseUrl}/unknown-${attempt}`, { headers }) + if (response.status === 429) { + throttled = true + break + } + expect(response.status).toBe(404) + } + + // A paired device must not be able to append unbounded 404 entries by probing unknown paths. + expect(throttled).toBe(true) + }) + + it('coalesces anonymous rejections instead of letting them flush the audit ring', async () => { + const before = service.getAuditEntries().length + for (let attempt = 0; attempt < 25; attempt += 1) { + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`) + expect(response.status).toBe(401) + } + + const entries = service.getAuditEntries() + const unauthorized = entries.filter((entry) => entry.status === 401) + // Every repeat is counted, not appended: an anonymous caller must not be able to evict the rest + // of the ring by hammering the endpoint. + expect(unauthorized).toHaveLength(1) + expect(unauthorized[0].suppressed).toBe(24) + expect(entries.length - before).toBe(1) + }) + + it('serves only real backup packages, not any zip in the sync folder', async () => { + const impostor = 'not-a-backup.zip' + const archive = zipSync({ 'manifest.json': strToU8(JSON.stringify({ version: 3 })) }) + await writeFile(path.join(syncDir, impostor), Buffer.from(archive)) + listBackups = async () => [ + { fileName: impostor, createdAt: 1_800_000_000_000, size: archive.length }, + { fileName: BACKUP_FILE_NAME, createdAt: 1_700_000_000_000, size: backupBytes.length } + ] + + const { token } = await pairDevice() + const status = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${token}` } + }) + expect(status.status).toBe(200) + const body = (await status.json()) as { snapshot: { fileName: string } | null } + expect(body.snapshot?.fileName).toBe(BACKUP_FILE_NAME) + }) + + it('returns one host identity before initialize() and persists it', async () => { + const userData = await mkdtemp(path.join(os.tmpdir(), 'deepchat-sync-host-id-')) + try { + const uninitialized = new SyncHostService({ + listBackups: async () => [], + getFolderPath: () => syncDir, + getUserDataPath: () => userData, + getAppVersion: () => '9.9.9' + }) + // Two callers before the first load must not observe two different host identities: this is + // the value a slave compares against the pairing payload. + const first = uninitialized.getHostId() + const second = uninitialized.getHostId() + expect(second).toBe(first) + + await uninitialized.initialize() + expect(uninitialized.getHostId()).toBe(first) + // Flush the queued write before asserting durability from a second instance. + await uninitialized.stop() + + const reloaded = new SyncHostService({ + listBackups: async () => [], + getFolderPath: () => syncDir, + getUserDataPath: () => userData, + getAppVersion: () => '9.9.9' + }) + await reloaded.initialize() + expect(reloaded.getHostId()).toBe(first) + } finally { + await rm(userData, { recursive: true, force: true }) + } + }) + + it('reaps a stalled request instead of holding a connection slot', async () => { + const { port } = await service.getStatus() + const socket = connect({ host: '127.0.0.1', port }) + const closed = new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('stalled socket was never reaped')), 5_000) + socket.on('close', () => { + clearTimeout(timer) + resolve() + }) + socket.on('error', () => { + clearTimeout(timer) + resolve() + }) + }) + await new Promise((resolve) => socket.once('connect', () => resolve())) + // Announce a body that never arrives. + socket.write( + `POST ${SYNC_HOST_PATH_PREFIX}/pair HTTP/1.1\r\nHost: 127.0.0.1\r\n` + + `Content-Type: application/json\r\nContent-Length: 100000\r\n\r\n` + ) + + await closed + + const healthy = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/handshake`) + expect(healthy.status).toBe(200) + }) + + it('settles and audits a snapshot request aborted while the snapshot is being resolved', async () => { + const { token } = await pairDevice() + + // Hold the snapshot resolution open (the real digest pass takes seconds on a large package) and + // drop the connection inside that window. The handler must still settle, audit the request and + // release the read stream; a hook attached only after the await would hang forever and leave no + // audit entry at all. + let release = (): void => undefined + const gate = new Promise((resolve) => { + release = resolve + }) + const gated = listBackups + listBackups = async () => { + await gate + return gated() + } + + const controller = new AbortController() + const pending = fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { authorization: `Bearer ${token}` }, + signal: controller.signal + }).catch(() => undefined) + + await new Promise((resolve) => setTimeout(resolve, 100)) + controller.abort() + await pending + release() + await new Promise((resolve) => setTimeout(resolve, 300)) + + const audit = service.getAuditEntries().filter((entry) => entry.path.endsWith('/snapshot')) + expect(audit.map((entry) => entry.status)).toEqual([499]) + }) + + it('reports no snapshot instead of failing when the backup list cannot be read', async () => { + const { token } = await pairDevice() + listBackups = async () => { + throw new Error('sync folder unavailable') + } + + // A storage failure is a transient condition a slave retries: it must see "no snapshot", never + // a 500 that looks like a broken host. + const status = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${token}` } + }) + expect(status.status).toBe(200) + expect(((await status.json()) as { snapshot: unknown }).snapshot).toBeNull() + + const snapshot = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/snapshot`, { + headers: { authorization: `Bearer ${token}` } + }) + expect(snapshot.status).toBe(404) + }) + + it('stops listening and removes its descriptor when host mode is disabled', async () => { + const descriptorPath = path.join(tempDir, 'sync-host', 'endpoint.json') + const descriptor = JSON.parse(await readFile(descriptorPath, 'utf8')) as { port: number } + expect(descriptor.port).toBe((await service.getStatus()).port) + + await service.setEnabled(false) + + expect((await service.getStatus()).port).toBeNull() + expect(service.getEnabled()).toBe(false) + await expect(stat(descriptorPath)).rejects.toThrow() + await expect(fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/handshake`)).rejects.toThrow() + }) + + it('keeps the listener and the enabled flag consistent under interleaved enable/disable', async () => { + await Promise.all([ + service.setEnabled(false), + service.setEnabled(true), + service.setEnabled(true), + service.setEnabled(false), + service.setEnabled(true) + ]) + + const enabled = await service.getStatus() + expect(enabled.enabled).toBe(true) + expect(enabled.running).toBe(true) + expect(enabled.port).not.toBeNull() + + await Promise.all([service.setEnabled(false), service.setEnabled(false)]) + const disabled = await service.getStatus() + expect(disabled.enabled).toBe(false) + expect(disabled.running).toBe(false) + expect(disabled.port).toBeNull() + }) + + it('stops the listener on teardown without disabling host mode', async () => { + await service.setEnabled(true) + + await service.stop() + + const status = await service.getStatus() + expect(status.running).toBe(false) + expect(status.port).toBeNull() + // Host mode must survive a restart: teardown is not a disable. + expect(status.enabled).toBe(true) + }) + + it('expires pairing codes without accepting them', () => { + const authority = new SyncHostPairingAuthority(() => 'host-1') + const created = authority.create({ now: 1_000, ttlMs: 5_000 }) + expect(authority.consume(created.code, 5_999)).toBe('accepted') + + const expiring = authority.create({ now: 1_000, ttlMs: 5_000 }) + expect(authority.consume(expiring.code, 6_000)).toBe('expired') + }) + + it('does not persist the host identity until initialize() writes it', async () => { + const userData = await mkdtemp(path.join(os.tmpdir(), 'deepchat-sync-host-lazy-')) + try { + const uninitialized = new SyncHostService({ + listBackups: async () => [], + getFolderPath: () => syncDir, + getUserDataPath: () => userData, + getAppVersion: () => '9.9.9' + }) + const identity = uninitialized.getHostId() + // Reading the identity must not write: a fire-and-forget write here can fail and roll the + // cache back after initialize() already adopted the value, leaving two identities in play. + await new Promise((resolve) => setTimeout(resolve, 100)) + await expect(stat(path.join(userData, 'sync-host', 'host-state.json'))).rejects.toThrow() + + await uninitialized.initialize() + const persisted = JSON.parse( + await readFile(path.join(userData, 'sync-host', 'host-state.json'), 'utf8') + ) as { hostId: string } + expect(persisted.hostId).toBe(identity) + await uninitialized.stop() + } finally { + await rm(userData, { recursive: true, force: true }) + } + }) + + it('restores a consumed pairing code when pairing could not be completed', () => { + const authority = new SyncHostPairingAuthority(() => 'host-1') + const created = authority.create({ now: 1_000, ttlMs: 600_000 }) + expect(authority.consume(created.code, 1_001)).toBe('accepted') + expect(authority.current(1_002)).toBeNull() + + authority.restore(created.code, created.expiresAt, 1_003) + expect(authority.consume(created.code, 1_004)).toBe('accepted') + + // Never clobbers a code the user generated in the meantime, and never revives an expired one. + const fresh = authority.create({ now: 2_000, ttlMs: 600_000 }) + authority.restore(created.code, created.expiresAt, 2_001) + expect(authority.current(2_002)?.code).toBe(fresh.code) + + authority.clear() + authority.restore(created.code, 5_000, 6_000) + expect(authority.current(6_001)).toBeNull() + }) + + it('does not let failed attempts destroy or block the user pairing code', () => { + const authority = new SyncHostPairingAuthority(() => 'host-1') + const created = authority.create({ now: 1_000, ttlMs: 600_000 }) + + // Far more failures than any per-code budget: the code must still be intact and usable. + for (let attempt = 0; attempt < 200; attempt += 1) { + expect(authority.consume('WRONGCODE', 1_000 + attempt)).toBe('invalid') + } + expect(authority.consume(created.code, 2_000)).toBe('accepted') + }) + + it('charges pairing failures to the calling source, not to everyone', async () => { + const code = service.createPairingCode() + expect(code).not.toBeNull() + + let throttled = false + for (let attempt = 0; attempt < 40; attempt += 1) { + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: 'WRONGCODE', deviceName: 'Attacker' }) + }) + if (response.status === 429) { + throttled = true + break + } + expect(response.status).toBe(401) + } + expect(throttled).toBe(true) + + // A fresh code minted for the user must still pair: the attacker only spent their own budget. + const fresh = service.createPairingCode() + expect(fresh).not.toBeNull() + }) + + it('rejects an oversized pairing body with 413 instead of resetting the connection', async () => { + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/pair`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ code: 'x'.repeat(8_192), deviceName: 'Big' }) + }) + expect(response.status).toBe(413) + }) + + it('survives a corrupt archive without hanging or throwing', async () => { + await writeFile(path.join(syncDir, BACKUP_FILE_NAME), Buffer.from('not a zip at all')) + + const { token } = await pairDevice() + const response = await fetch(`${baseUrl}${SYNC_HOST_PATH_PREFIX}/status`, { + headers: { authorization: `Bearer ${token}` } + }) + expect(response.status).toBe(200) + const body = (await response.json()) as { + snapshot: { backupFormatVersion: number | null; sha256: string } + } + expect(body.snapshot.backupFormatVersion).toBeNull() + expect(body.snapshot.sha256).toMatch(/^[0-9a-f]{64}$/) + }) + + it('preserves existing state when a mutation arrives before initialize()', async () => { + const userData = await mkdtemp(path.join(os.tmpdir(), 'deepchat-sync-host-uninit-')) + try { + const seed = new SyncHostService({ + listBackups: async () => [], + getFolderPath: () => syncDir, + getUserDataPath: () => userData, + getAppVersion: () => '9.9.9' + }) + await seed.initialize() + const issued = await seed.listDevices() + expect(issued).toEqual([]) + const firstToken = await seed.setEnabled(true) + expect(firstToken.enabled).toBe(true) + // Release the listener seed.setEnabled(true) started before the files are removed. + await seed.stop() + + // A second instance that mutates before any explicit initialize() must not wipe the file. + // The mutation has to be a real one: with the read-modify-write bug this persisted the empty + // default state over the file, discarding the enabled flag and every device record. + const late = new SyncHostService({ + listBackups: async () => [], + getFolderPath: () => syncDir, + getUserDataPath: () => userData, + getAppVersion: () => '9.9.9' + }) + expect(await late.renameDevice('dev_missing', 'Renamed')).toBe(false) + await late.stop() + + const reloaded = new SyncHostService({ + listBackups: async () => [], + getFolderPath: () => syncDir, + getUserDataPath: () => userData, + getAppVersion: () => '9.9.9' + }) + await reloaded.initialize() + expect(reloaded.getEnabled()).toBe(true) + expect(reloaded.listDevices()).toEqual([]) + } finally { + await rm(userData, { recursive: true, force: true }) + } + }) + + it('keeps a revoked device revoked across a state reload', async () => { + const issued = await pairDevice() + expect(await service.revokeDevice(issued.deviceId)).toBe(true) + + const reloaded = new SyncHostService({ + listBackups: async () => [ + { fileName: BACKUP_FILE_NAME, createdAt: 1_700_000_000_000, size: backupBytes.length } + ], + getFolderPath: () => syncDir, + getUserDataPath: () => tempDir, + getAppVersion: () => '9.9.9' + }) + await reloaded.initialize() + + expect(reloaded.listDevices()).toEqual([ + expect.objectContaining({ deviceId: issued.deviceId, revoked: true }) + ]) + }) +}) diff --git a/test/main/sync/host/routes.test.ts b/test/main/sync/host/routes.test.ts new file mode 100644 index 000000000..e44e2195a --- /dev/null +++ b/test/main/sync/host/routes.test.ts @@ -0,0 +1,217 @@ +import { describe, expect, it, vi } from 'vitest' +import { createRendererRouteContext } from '@/routes/routeRegistry' +import { + syncHostCreatePairingCodeRoute, + syncHostGetAuditRoute, + syncHostGetStatusRoute, + syncHostListDevicesRoute, + syncHostRenameDeviceRoute, + syncHostRevokeDeviceRoute, + syncHostSetEnabledRoute +} from '@shared/contracts/routes' +import { + SyncHostAuditEntrySchema, + type SyncHostAuditEntry, + type SyncHostDeviceView +} from '@shared/contracts/syncHost' +import { createSyncHostRoutes, type SyncHostRoutePort } from '@/sync/host/routes' + +const STATUS = { + enabled: true, + running: true, + port: 43117, + hostId: 'host-abc', + deviceCount: 2, + hasSnapshot: true +} + +const PAIRING = { + code: 'ABCD2345', + hostId: 'host-abc', + expiresAt: 1_700_000_300_000 +} + +const DEVICE: SyncHostDeviceView = { + deviceId: 'device-1', + name: 'Laptop', + createdAt: 1_700_000_000_000, + expiresAt: null, + lastSeenAt: 1_700_000_100_000, + revoked: false +} + +/** Built through the contract so an evolving audit schema cannot leave the fixture malformed. */ +function auditEntry(overrides: Partial = {}): SyncHostAuditEntry { + return SyncHostAuditEntrySchema.parse({ + at: 1_700_000_000_000, + method: 'POST', + path: '/sync/v1/pair', + status: 401, + bytes: 42, + deviceId: null, + clientIp: '203.0.113.9', + suppressed: 0, + ...overrides + }) +} + +function createHostPort(overrides: Partial = {}): SyncHostRoutePort { + return { + getStatus: vi.fn(async () => STATUS), + getPairingCode: vi.fn(() => null), + setEnabled: vi.fn(async () => STATUS), + createPairingCode: vi.fn(() => null), + listDevices: vi.fn(() => []), + revokeDevice: vi.fn(async () => true), + renameDevice: vi.fn(async () => true), + getAuditEntries: vi.fn(() => []), + ...overrides + } +} + +const context = createRendererRouteContext(1, null) + +describe('sync host routes', () => { + it('exposes exactly the seven renderer-facing routes as handlers', () => { + const routes = createSyncHostRoutes({ host: createHostPort() }) + + expect([...routes.keys()].sort()).toEqual( + [ + syncHostCreatePairingCodeRoute.name, + syncHostGetAuditRoute.name, + syncHostGetStatusRoute.name, + syncHostListDevicesRoute.name, + syncHostRenameDeviceRoute.name, + syncHostRevokeDeviceRoute.name, + syncHostSetEnabledRoute.name + ].sort() + ) + expect(routes.size).toBe(7) + for (const handler of routes.values()) { + expect(typeof handler).toBe('function') + } + }) + + it('returns the host status with the current pairing code', async () => { + const host = createHostPort({ getPairingCode: () => PAIRING }) + const handler = createSyncHostRoutes({ host }).get(syncHostGetStatusRoute.name)! + + await expect(handler({}, context)).resolves.toEqual({ status: STATUS, pairing: PAIRING }) + }) + + it('reports a null pairing when no code is outstanding', async () => { + const host = createHostPort({ getPairingCode: () => null }) + const handler = createSyncHostRoutes({ host }).get(syncHostGetStatusRoute.name)! + + await expect(handler({}, context)).resolves.toEqual({ status: STATUS, pairing: null }) + }) + + it('passes the enabled flag through and returns the resulting status', async () => { + const setEnabled = vi.fn(async () => ({ ...STATUS, enabled: false })) + const host = createHostPort({ setEnabled }) + const handler = createSyncHostRoutes({ host }).get(syncHostSetEnabledRoute.name)! + + await expect(handler({ enabled: false }, context)).resolves.toEqual({ + status: { ...STATUS, enabled: false } + }) + expect(setEnabled).toHaveBeenCalledWith(false) + }) + + it('surfaces a setEnabled failure instead of reporting a status', async () => { + const host = createHostPort({ + setEnabled: async () => { + throw new Error('sync host bind failed') + } + }) + const handler = createSyncHostRoutes({ host }).get(syncHostSetEnabledRoute.name)! + + await expect(handler({ enabled: true }, context)).rejects.toThrow('sync host bind failed') + }) + + it('returns the freshly created pairing code', async () => { + const host = createHostPort({ + createPairingCode: () => PAIRING, + getPairingCode: () => ({ ...PAIRING, code: 'STALE234' }) + }) + const handler = createSyncHostRoutes({ host }).get(syncHostCreatePairingCodeRoute.name)! + + await expect(handler({}, context)).resolves.toEqual({ pairing: PAIRING }) + }) + + it('falls back to the current pairing code when creation returns null', async () => { + const host = createHostPort({ createPairingCode: () => null, getPairingCode: () => PAIRING }) + const handler = createSyncHostRoutes({ host }).get(syncHostCreatePairingCodeRoute.name)! + + await expect(handler({}, context)).resolves.toEqual({ pairing: PAIRING }) + }) + + it('lists device views without token material', async () => { + const storedDevice = { + ...DEVICE, + tokenHash: 'sha256:deadbeef', + token: 'device-token-secret' + } + const host = createHostPort({ listDevices: () => [storedDevice] }) + const handler = createSyncHostRoutes({ host }).get(syncHostListDevicesRoute.name)! + + const result = (await handler({}, context)) as { devices: Record[] } + + expect(result.devices).toEqual([DEVICE]) + expect(Object.keys(result.devices[0])).not.toContain('tokenHash') + expect(Object.keys(result.devices[0])).not.toContain('token') + }) + + it('passes the device id through and returns the revoke result', async () => { + const revokeDevice = vi.fn(async () => true) + const host = createHostPort({ revokeDevice }) + const handler = createSyncHostRoutes({ host }).get(syncHostRevokeDeviceRoute.name)! + + await expect(handler({ deviceId: DEVICE.deviceId }, context)).resolves.toEqual({ + revoked: true + }) + expect(revokeDevice).toHaveBeenCalledWith(DEVICE.deviceId) + }) + + it('passes the device id and name through and returns the rename result', async () => { + const renameDevice = vi.fn(async () => false) + const host = createHostPort({ renameDevice }) + const handler = createSyncHostRoutes({ host }).get(syncHostRenameDeviceRoute.name)! + + await expect( + handler({ deviceId: DEVICE.deviceId, name: 'Workstation' }, context) + ).resolves.toEqual({ renamed: false }) + expect(renameDevice).toHaveBeenCalledWith(DEVICE.deviceId, 'Workstation') + }) + + it('rejects empty device ids and names at the route boundary', async () => { + const revokeDevice = vi.fn(async () => true) + const renameDevice = vi.fn(async () => true) + const host = createHostPort({ revokeDevice, renameDevice }) + const routes = createSyncHostRoutes({ host }) + + await expect( + routes.get(syncHostRevokeDeviceRoute.name)!({ deviceId: '' }, context) + ).rejects.toThrow() + await expect( + routes.get(syncHostRenameDeviceRoute.name)!({ deviceId: '', name: 'Workstation' }, context) + ).rejects.toThrow() + await expect( + routes.get(syncHostRenameDeviceRoute.name)!({ deviceId: DEVICE.deviceId, name: '' }, context) + ).rejects.toThrow() + expect(revokeDevice).not.toHaveBeenCalled() + expect(renameDevice).not.toHaveBeenCalled() + }) + + it('returns the audit entries without token or payload fields', async () => { + const entry = auditEntry() + const storedEntry = { ...entry, token: 'device-token-secret', payload: { code: PAIRING.code } } + const host = createHostPort({ getAuditEntries: () => [storedEntry] }) + const handler = createSyncHostRoutes({ host }).get(syncHostGetAuditRoute.name)! + + const result = (await handler({}, context)) as { entries: Record[] } + + expect(result.entries).toEqual([entry]) + expect(Object.keys(result.entries[0])).not.toContain('token') + expect(Object.keys(result.entries[0])).not.toContain('payload') + }) +}) diff --git a/test/main/sync/host/state.test.ts b/test/main/sync/host/state.test.ts new file mode 100644 index 000000000..5ebe0773a --- /dev/null +++ b/test/main/sync/host/state.test.ts @@ -0,0 +1,99 @@ +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import os from 'node:os' +import path from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +// This suite exercises real files on disk, so the global partial `fs` mock from test/setup.ts must +// not apply here. +vi.unmock('fs') +vi.unmock('node:fs') + +import { SyncHostStateStore } from '@/sync/host/state' + +/** + * Durability rules for the machine-local host state. These are unit tests because the failure modes + * they pin (a write that cannot land, a read that cannot complete) are not reachable through the + * service without an unwritable filesystem. + */ +describe('SyncHostStateStore', () => { + let directory: string + let store: SyncHostStateStore + + beforeEach(async () => { + directory = await mkdtemp(path.join(os.tmpdir(), 'deepchat-sync-host-state-')) + store = new SyncHostStateStore(directory) + }) + + afterEach(async () => { + await rm(directory, { recursive: true, force: true }) + }) + + /** Makes every write fail: the atomic rename cannot replace a non-empty directory. */ + async function blockWrites(): Promise { + await mkdir(store.filePath, { recursive: true }) + await writeFile(path.join(store.filePath, 'blocker'), 'x') + } + + it('rolls back concurrent updates when their writes fail', async () => { + await store.load() + await blockWrites() + + // Two updates in flight at once. Serializing only the filesystem writes would let the second + // one snapshot the first one's mutation, which would both block the rollback and persist a + // change whose caller was told it failed. + const results = await Promise.allSettled([ + store.update((state) => { + state.enabled = true + }), + store.update((state) => { + state.devices.push({ + deviceId: 'dev_blocked', + name: 'Blocked', + tokenHash: 'a'.repeat(64), + createdAt: 1, + lastSeenAt: null, + expiresAt: null, + revokedAt: null + }) + }) + ]) + + expect(results.map((result) => result.status)).toEqual(['rejected', 'rejected']) + expect(store.snapshot()).toMatchObject({ enabled: false, devices: [] }) + }) + + it('keeps a failed mutation out of the next successful write', async () => { + await store.load() + await blockWrites() + await expect( + store.update((state) => { + state.enabled = true + }) + ).rejects.toThrow() + + // Unblock the path and write something legitimate. + await rm(store.filePath, { recursive: true, force: true }) + await store.update((state) => { + state.hostId = 'host-after-failure' + }) + + const persisted = JSON.parse(await readFile(store.filePath, 'utf8')) as { enabled: boolean } + expect(persisted.enabled).toBe(false) + }) + + it('fails closed on an unreadable state file instead of resetting to defaults', async () => { + await writeFile(store.filePath, '{ not json') + + await expect(store.load()).rejects.toThrow() + expect(store.isLoaded()).toBe(false) + // The damaged file is left alone: a transient read or parse failure must never be turned into + // an empty default state that the next write persists over the real one. + expect(await readFile(store.filePath, 'utf8')).toBe('{ not json') + + await writeFile( + store.filePath, + `${JSON.stringify({ enabled: true, hostId: 'host-1', devices: [] })}\n` + ) + expect((await store.load()).enabled).toBe(true) + }) +})