From 8203de1f865f653d3818bed925d507cbfe11713f Mon Sep 17 00:00:00 2001 From: anikitenko Date: Sun, 18 Oct 2026 13:42:21 +0300 Subject: [PATCH 1/2] fix: address remote command feedback and persistence failures - Add host adapters for SSH status, live output, and cancellation - Validate combined output limits and sanitize service errors - Reject JSON flush failures and support recovery - Document FDO wiring, timeout semantics, and save acknowledgements --- CHANGELOG.md | 8 + README.md | 2 +- docs/API_STABILITY.md | 1 + docs/ARCHITECTURE.md | 2 + docs/FDO_HOST_REMOTE_COMMANDS_HANDOFF.md | 104 +++++++++++ docs/FDO_HOST_SECRETS_SSH_HANDOFF.md | 6 +- docs/SECRETS_AND_SSH.md | 15 +- src/PluginRegistry.ts | 4 +- src/StoreJson.ts | 9 +- src/types.ts | 34 +++- src/utils/capabilities.ts | 13 +- src/utils/contracts.ts | 2 +- src/utils/hostServices.ts | 214 ++++++++++++++++++----- tests/StoreJson.test.ts | 21 ++- tests/utils/sshLive.test.ts | 168 ++++++++++++++++++ 15 files changed, 542 insertions(+), 61 deletions(-) create mode 100644 docs/FDO_HOST_REMOTE_COMMANDS_HANDOFF.md create mode 100644 tests/utils/sshLive.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index e1efd93..a373126 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## Unreleased + +* Add optional host adapters for timestamped SSH status monitoring and streaming execution with explicit cancellation; retain buffered `ssh.exec` compatibility. +* Add opt-in `ssh.status.watch` capability and safe `CANCELLED` service error code. +* Reject JSON `flush()`/`dispose()` after failed queued writes; allow recovery after a successful full snapshot. Previously these failures were logged and swallowed. +* Document FDO wiring, output bounds, status freshness, timeout/cancellation evidence and persistent-save acknowledgements. + + All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), diff --git a/README.md b/README.md index cbcb342..9c3c9d8 100644 --- a/README.md +++ b/README.md @@ -286,7 +286,7 @@ For a dedicated authoring guide, see [docs/OPERATOR_PLUGIN_PATTERNS.md](./docs/O ## Universal Secrets and SSH -Plugin backend code can use `this.secrets` for host-managed universal secret storage and `this.ssh` for builtin SSH execution against authorized connection profiles. Declare the required secret operations and SSH capabilities through `declareCapabilities()`. The FDO host must configure the backend service transport; calls fail closed until host support is installed. Secrets are separate from JSON storage. +Plugin backend code can use `this.secrets` for host-managed universal secret storage and `this.ssh` for builtin SSH execution against authorized connection profiles. Declare the required secret operations and SSH capabilities through `declareCapabilities()`. The FDO host must configure the backend service transport; calls fail closed until host support is installed. Secrets are separate from JSON storage. Optional `ssh.watchStatus()` and `ssh.start()` adapters add fresh connection observations, live output and explicit cancellation when wired by FDO; see the [Remote Commands host handoff](docs/FDO_HOST_REMOTE_COMMANDS_HANDOFF.md). JSON `flush()` now rejects failed queued writes, so acknowledge persistent saves only after awaiting it. See [secrets and SSH usage and protocol](./docs/SECRETS_AND_SSH.md) and [FDO host implementation handoff](./docs/FDO_HOST_SECRETS_SSH_HANDOFF.md). diff --git a/docs/API_STABILITY.md b/docs/API_STABILITY.md index cb2c910..bd8b4ab 100644 --- a/docs/API_STABILITY.md +++ b/docs/API_STABILITY.md @@ -72,6 +72,7 @@ This document defines which SDK surfaces are stable contracts versus internal im - Backend secret and SSH service surfaces: - `FDO_SDK.secrets` / `SecretStore` - `FDO_SDK.ssh` / `SSHClient` + - `SSHClient.watchStatus`, `SSHClient.start`, `SSHExecution`, and optional host live adapters (host wiring required) - `HostServiceError` and exported service types - `createSecretsCapabilityBundle` and `createSSHCapabilityBundle` - base-plus-operation capability requirements and backend protocol v1 in [the service contract](./SECRETS_AND_SSH.md) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index f519f57..005c3c6 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -80,6 +80,8 @@ Store lifecycle hooks are supported: - `dispose()` - optional capability metadata and migration/version hooks +The public `useStore()` result includes optional lifecycle hooks. JSON `flush()`/`dispose()` reject failed queued writes; a later successful full snapshot allows recovery. Await `flush()` before acknowledging a persistent save. See the [Remote Commands host handoff](./FDO_HOST_REMOTE_COMMANDS_HANDOFF.md) for storage setup and shutdown requirements. + Runtime permission model: - privileged SDK features are capability-gated (`storage`, `storage.` such as `storage.json`, `system.network`, `system.network.` such as `system.network.https`, `sudo.prompt`, `system.hosts.write`, `system.fs.scope.`, `system.process.exec`, `system.process.scope.`) diff --git a/docs/FDO_HOST_REMOTE_COMMANDS_HANDOFF.md b/docs/FDO_HOST_REMOTE_COMMANDS_HANDOFF.md new file mode 100644 index 0000000..8344aef --- /dev/null +++ b/docs/FDO_HOST_REMOTE_COMMANDS_HANDOFF.md @@ -0,0 +1,104 @@ +# Remote Commands: SDK and FDO integration + +The SDK now exposes optional connection monitoring and streaming execution adapters in addition to buffered `ssh.exec`. These additions require a new SDK build and FDO implementation; installing the SDK alone does not enable them. Existing 1.8.1 hosts/plugins retain buffered behavior. This document is the implementation handoff for FDO and the Remote Commands plugin. No host or plugin UI implementation is included in this SDK repository. + +## Host bootstrap contract + +Configure adapters in the isolated backend worker before plugin initialization: + +```typescript +PluginRegistry.configureHostServices({ + request: requestHostService, // existing HostServiceRequest/result transport + watchSSHStatus: async ({ profileId }) => statusBridge.subscribe(profileId), + startSSHExecution: async options => executionBridge.start(options), +}); +``` + +`statusBridge` and `executionBridge` above are host implementations, not SDK exports. Exact TypeScript signatures are `HostServicesConfiguration` and `HostSSHExecution` in `src/types.ts`. Optional adapters are discovered by their presence; absent adapters reject with `HostServiceError.code === "UNAVAILABLE"`. Never silently run a buffered command as a fallback from `start`, because this can duplicate execution. + +Adapters carry async iterables and promises inside the worker. Do not send these objects over IPC. Implement authenticated request IDs, subscription/execution IDs, ordered event delivery, terminal replies and explicit unsubscribe/cancel messages on the existing worker bridge. A suitable host-internal mapping is: + +| Worker operation | Host responsibility | +| --- | --- | +| watch status `{ profileId }` | Authorize, allocate subscription ID, return initial status, stream heartbeats/transitions | +| status iterator `return()` | Unsubscribe that ID and release monitor references | +| start execution `{ profileId, command, timeoutMs, maxOutputBytes }` | Authorize, allocate execution ID, start once, return event iterable/result promise/cancel closure | +| execution output | Route `{ sequence, stream, data }` to its execution ID | +| execution terminal reply | Resolve result with `SSHExecResult`, or reject with a safe code | +| output iterator `return()` | Release output subscription; continue bounded capture/draining until terminal result | +| execution `cancel()` | Send cancellation for that execution ID, return `{ remoteProcessStopped }` | + +IDs are host allocated, scoped to authenticated plugin/worker identity and invalidated on teardown. A guessed ID must not allow observation or cancellation of another plugin's work. Install output routing before starting the command so early output cannot be lost. Bound queues and pending work. A slow or absent consumer must never create an unbounded queue or block command deadline/cancellation. Retain at most the approved capture budget, drain excess, and end the event iterator independently of whether its consumer awaits the result. Terminal events must wake pending `next()` calls. Keep `cancel()` idempotent through completion/cancellation races, and retain the terminal acknowledgement for the scoped handle until cleanup. + +The SDK invalidates stale responses and requests best-effort cleanup when host configuration is cleared/replaced. The host must independently release monitors, wake iterators, settle pending promises, and close channels/sockets on worker disconnect, plugin unload, profile change/deletion, credential rotation and grant revocation. View closure requires an explicit plugin-to-backend cancellation request if work should stop; a DOM disappearing is not that request. Worker teardown must not depend on JavaScript iterator finalizers running. + +## Connection feed and grants + +Add `ssh.status.watch` to capability discovery, installation/consent, preflight and revocation. The SDK helper `createSSHCapabilityBundle({ monitorStatus: true })` includes it; the default bundle is unchanged. Monitoring requires `ssh.use`, `ssh.status.watch`, `secrets.use`, `secrets.consume`, plus host authorization for the selected command profile. It authenticates using host-held credential references and verifies the pinned host key. It must not execute an arbitrary shell command as a health check or grant access to Git-only profiles. Streaming start/cancel uses the existing SSH execution grant set and command policy; no plaintext `secrets.get` grant is required. + +`watchSSHStatus` returns an async iterable of: + +```typescript +type Status = { profileId: string; state: "unknown" | "connecting" | "connected" | "disconnected"; + checkedAt: string | null; staleAfterMs: number }; +``` + +Send an initial observation promptly, transitions immediately, and timestamped heartbeats while subscribed. `checkedAt` is an ISO timestamp of an actual host observation, not a cached command completion time. Null is allowed only for `unknown`. `staleAfterMs` is a positive integer. Bound polling/handshake deadlines and monitor counts; use SSH keepalives and explicit liveness checks, reference-count shared authorized monitors, and stop probes when no subscriptions remain. Return `unknown` until a real observation exists. A connected observation means the host observed authenticated SSH reachability at that time. It cannot guarantee that the VM or every service remains available. Never derive this feed from the previous command's exit code; a nonzero exit does not imply a connection failure. + +Remote Commands must show the observation time and freshness. Set status to unknown/stale when `checkedAt + staleAfterMs` expires, on feed error/end, or when the view/backend loses its subscription. Only show “SSH connected” while observations are fresh. A stale/future timestamp must not keep a green badge indefinitely; account for clock skew or display unknown. Keep a command's last-attempt outcome separately. FDO must reauthorize active subscriptions when grants/profiles change and stop delivery on revocation. + +## Live execution, cancellation, timeouts and capture + +`startSSHExecution` returns `{ events, result, cancel }`. `events` yields `{ sequence, stream: "stdout" | "stderr", data }`; sequence starts at zero and increases contiguously across both streams. Decode each stream incrementally so UTF-8 characters split across network chunks survive. Sequence establishes host observation order only; independent stdout/stderr pipes cannot establish exact remote write order. Do not infer order by concatenating the two buffered result strings. `result` retains the buffered `SSHExecResult` shape for compatibility, including prominent `truncated` feedback. + +The default combined stdout/stderr budget remains 1 MiB; plugins can request another positive safe integer via `maxOutputBytes`. The host applies its own ceilings. Count UTF-8 bytes across both streams, retain/emit only the approved prefix, drain excess output, and set `truncated: true` if anything was discarded. Apply one shared capture budget: event delivery and the final buffered copy represent the same captured bytes, not two independently allowed prefixes. Keep the encoded delivered strings within the budget, including replacement characters for malformed UTF-8. The SDK validates event sequences and combined delivered byte limits and rejects oversized final results. No unlimited capture is promised. Once the budget is exhausted, execution may continue with live elapsed-time feedback until it completes or is cancelled. + +Remote Commands can continue to offer 1–120 seconds, but this is a plugin choice; the SDK accepts positive safe integers and defaults to 30 seconds. Enforce handshake and command deadlines independently in the host and apply lower policy ceilings when necessary. Explain the effective host policy in FDO settings/UI. On timeout, close the channel, attempt termination according to host policy, drain/release resources, settle result with `TIMEOUT`, and end events. Never automatically retry a command, including after a lost start acknowledgement. + +`cancel()` requests termination and returns `{ remoteProcessStopped: true | false | null }`. Return true only with affirmative evidence that the intended process/workload stopped; channel/socket closure or an SSH signal request alone is insufficient. Null means unknown. Keep process termination evidence and the cancellation acknowledgement distinct from the command's transport result. For an active execution stopped by cancellation, reject `result` with `CANCELLED`; if it already finished, preserve its terminal result. A cancellation racing completion must not overwrite a completed result. Provider errors must be mapped to safe SDK codes; never expose provider messages or log command/output bodies by default. + +If FDO needs guaranteed remote workload termination, it needs a separately designed remote supervisor/job protocol that owns process groups and can query job state. The SSH channel API alone cannot provide that guarantee. Do not label timeout, view closure or an unknown cancellation acknowledgement “remote process stopped.” + +Plugin backend usage (consume output and result concurrently): + +```typescript +const execution = await this.ssh.start({ profileId, command, timeoutMs: 60_000 }); +// Store the handle in the backend under a plugin-owned run ID so the Cancel handler can use it. +const outputDone = (async () => { + for await (const event of execution.events) { + // Forward a bounded batch to the plugin view; render command output as text. + publishOutput(event); + } +})(); +const [result] = await Promise.all([execution.result, outputDone]); +// Use result.truncated prominently. Do not append the buffered streams again after live output. +// In the Cancel handler: await execution.cancel(), then display its termination evidence. +``` + +Catch errors around this code, await/handle both tasks, release run handles in `finally`, and stop subscriptions when the plugin view is closed. Do not serialize the SDK execution object into the iframe. `events` allows one consumer; breaking its loop closes only the output subscription. Call `cancel()` explicitly to request termination. A result can complete before the consumer drains its bounded queued events. + +## Persistent JSON acknowledgement + +FDO must configure a writable, host-owned persistent root with `PluginRegistry.configureStorage({ rootDir })` (or `FDO_SDK_STORAGE_ROOT`) before the plugin selects JSON storage. Grant both `storage` and `storage.json`. Keep identity/scoped paths stable across restarts. Missing configuration and missing grants already fail explicitly; never substitute the memory store and acknowledge a persistent save. + +The revised SDK JSON store still queues writes without making `set()` async, but `flush()` and `dispose()` now reject when the latest queued snapshot could not be written. Failures are also logged to FDO diagnostics. The queue remains usable: a later successful full snapshot clears the write error, and a subsequent flush can succeed. This is a change from 1.8.1, which logged and swallowed write failures. A successful flush confirms the queued atomic write completed; it is not a guarantee against all future disk loss or external modification. Preserve diagnostics rather than exposing raw disk errors/paths to the view. + +A save handler must acknowledge only after the public lifecycle barrier: + +```typescript +try { + const store = PluginRegistry.useStore("json"); + store.set("remoteCommands", settings); + if (!store.flush) throw new Error("Persistent storage acknowledgement is unsupported."); + await store.flush(); + return { saved: true }; +} catch { + return { saved: false, error: "Settings could not be saved. Check storage configuration, grants and FDO diagnostics." }; +} +``` + +The identifiers `settings` and the surrounding handler are supplied by the plugin. Show the rejected-save state explicitly and retain unsaved edits. On older SDKs, a flush acknowledgement can still hide disk failures; do not claim the stronger contract until the host/plugin actually uses this updated build. Host shutdown should await lifecycle completion before terminating the worker, and surface flush failures. The existing registry's synchronous teardown initiates cleanup but does not itself provide an awaitable host shutdown barrier. + +## Acceptance checks in FDO + +Verify fresh initial status, heartbeat loss, reconnect, stale UI and teardown without shell probes. Test missing grants, cross-plugin IDs, revocation during monitoring/execution, pinned-key mismatch and credential rotation. Confirm output arrives before completion; exercise stdout/stderr ordering, split and malformed UTF-8, exact combined caps, truncation, slow/no consumers and commands that exceed deadlines. Test cancel before/after completion, timeout, disconnect and view closure; ensure no command retry and no false termination claim. Exercise missing/unwritable roots, disk full, flush rejection, later recovery and restart persistence. Test packaged Windows/macOS/Linux SSH without an external SSH executable. SDK mock tests cannot establish host/network integration. diff --git a/docs/FDO_HOST_SECRETS_SSH_HANDOFF.md b/docs/FDO_HOST_SECRETS_SSH_HANDOFF.md index 9e4da95..f543fa5 100644 --- a/docs/FDO_HOST_SECRETS_SSH_HANDOFF.md +++ b/docs/FDO_HOST_SECRETS_SSH_HANDOFF.md @@ -56,13 +56,17 @@ Current host alignment: Profile management and host-key enrollment remain host-owned. Profile edits/deletion, credential rotation and permission revocation invalidate affected work. Use the current FDO starter templates and host contract for supported CLI forms; do not claim every Git flag is authorized by the SDK's generic process payload type. +## Remote Commands follow-up + +Implement the optional status/streaming adapters, cancellation and persistent save acknowledgement described in [Remote Commands host handoff](./FDO_HOST_REMOTE_COMMANDS_HANDOFF.md). This extends the original buffered service without changing its request/reply actions. + ## Release compatibility Read [minor release scope and migration](./SECRETS_AND_SSH.md#minor-release-scope-and-migration). The SDK additions are additive, but the host must recognize all new capability names, enforce the base-plus-operation model and install the backend transport before enabling the feature. The SDK handshake alone does not establish host readiness. Existing plugins need no new grants unless adopting these services. Never infer secret/SSH grants from existing storage or process permissions, and never fall back to insecure storage or an unrestricted SSH process. ## Errors, audit and verification -Transport failures expose only a safe code: ACCESS_DENIED, UNAVAILABLE, LOCKED, TIMEOUT, HOST_KEY_REJECTED, CONNECTION_FAILED, or INVALID_REQUEST. Provider messages must not leak credentials. Audit identity, action, outcome and request ID without secret values or request/response bodies. Commands and stdout/stderr may also contain secrets, so do not log them by default. +Transport failures expose only a safe code: ACCESS_DENIED, UNAVAILABLE, LOCKED, TIMEOUT, CANCELLED, HOST_KEY_REJECTED, CONNECTION_FAILED, or INVALID_REQUEST. Provider messages must not leak credentials. Audit identity, action, outcome and request ID without secret values or request/response bodies. Commands and stdout/stderr may also contain secrets, so do not log them by default. Add host tests covering cross-plugin access and forged identities, each missing capability, revocation, unavailable/locked secure backends, absence of plaintext persistence/logging, missing and empty values, overwrites/deletion, malformed requests, unauthorized profiles, correct/mismatched/missing host keys, password and encrypted private-key authentication, output caps, UTF-8 chunk boundaries, nonzero/null exits, timeouts, disconnects and cleanup. Test the packaged host on all three platforms without an installed SSH executable. The SDK test suite mocks the bridge and does not verify real encryption or networking. diff --git a/docs/SECRETS_AND_SSH.md b/docs/SECRETS_AND_SSH.md index c9ea021..e1a0421 100644 --- a/docs/SECRETS_AND_SSH.md +++ b/docs/SECRETS_AND_SSH.md @@ -18,6 +18,7 @@ Declare permissions through the existing `declareCapabilities()` hook. Declarati | `secrets.consume` | Allow an authorized host service to consume a referenced secret (requires `secrets.use`) | | `ssh.use` | Base permission to access SSH services; grants no operation by itself | | `ssh.exec` | Execute a remote command on an authorized host profile | +| `ssh.status.watch` | Monitor authenticated SSH reachability with freshness; also requires secret consumption grants | | `ssh.profiles.list` | List authorized SSH command profile IDs and labels, plus optional host, port, username, authentication type and trusted host-key fingerprint; excludes credentials, secret references and command policies | Every secret operation requires **`secrets.use` plus the specific operation capability**: for example, reading requires `secrets.use` + `secrets.get`, and listing requires `secrets.use` + `secrets.list`. Neither the base permission alone nor an operation alone is sufficient. Permission to write never implicitly permits reading or deletion. @@ -72,6 +73,12 @@ The host should bundle `ssh2`, so users need no installed SSH executable on Wind This initial API supports buffered execution. Interactive terminals, SFTP, forwarding, agent integration, streaming events and explicit caller cancellation are future extensions. The host must enforce deadlines and dispose channels/connections on worker teardown. Closing a channel does not guarantee the remote process has stopped. +## Live connection status and command execution + +Optional host adapters now support `this.ssh.watchStatus(profileId)` (an async status iterable) and `this.ssh.start(options)` (live output events, a buffered result promise and explicit cancellation). Declare `createSSHCapabilityBundle({ monitorStatus: true })` to request monitoring in addition to existing SSH grants. Hosts without these adapters reject with `UNAVAILABLE`; `exec()` remains buffered and has no caller cancellation. Monitoring has the separate `ssh.status.watch` grant plus `ssh.use`, `secrets.use` and `secrets.consume`. These are additive SDK contracts that require host wiring, not features automatically enabled on an existing FDO installation. + +See the [Remote Commands host handoff](./FDO_HOST_REMOTE_COMMANDS_HANDOFF.md) for exact adapter contracts, event consumption, freshness, termination evidence, byte budgets, JSON save acknowledgements and acceptance checks. Fresh status is an observation of SSH reachability, not a guarantee of future VM availability. Cancellation/timeout can close a channel without proving remote process termination. Default capture remains 1 MiB combined; callers can request another limit subject to host ceilings. Output sequences reflect host observation order across streams. + ## Reusing SSH profiles for Git FDO now reuses plugin-owned SSH profiles for Git through the existing `requestOperatorTool("git", payload)` path. Configure the profile's Git consumer, exact repository paths and read/push policy in Settings → Credentials & SSH. Reuse is within the same plugin namespace; cross-plugin sharing is not supported. It never copies credentials into plugin code or repository settings. @@ -136,19 +143,19 @@ For packaged-host acceptance, verify clone/fetch/push against a controlled SSH G ## Minor release scope and migration -This is an additive SDK feature suitable for a minor release when included with the associated implementation. Public additions are `this.secrets`, `this.ssh`, `SecretStore`, `SSHClient`, `HostServiceError`, `PluginRegistry.configureHostServices`, capability bundle helpers and their exported types. Existing storage and privileged-action envelopes retain their behavior. +This is an additive SDK feature suitable for a minor release when included with the associated implementation. Public additions are `this.secrets`, `this.ssh`, `SecretStore`, `SSHClient`, `HostServiceError`, `PluginRegistry.configureHostServices`, capability bundle helpers and their exported types. Existing privileged-action envelopes retain their behavior. The Remote Commands follow-up also strengthens JSON flush failure reporting as documented in the linked handoff. Existing plugins need no changes unless adopting the new services. Adopting plugins must declare base and operation permissions, obtain host grants, and call services from backend handlers while keeping lifecycle hooks synchronous. Never migrate ordinary JSON store contents wholesale into secret storage or derive new grants from old storage/process grants. Explicitly move only user-selected secrets through an authorized host flow and remove plaintext copies as appropriate. Older hosts can continue running existing plugins, but cannot execute the new service APIs until their backend transport is installed. New capabilities must be recognized by the host's capability parser and consent UI. The current SDK handshake does not advertise host readiness for these services: a new SDK version or successful SDK handshake alone is insufficient. Before enabling feature UI, the host must provide its own readiness/availability signal or handle unavailable service calls explicitly. Do not silently fall back to plaintext storage or unrestricted system SSH. -Release notes must distinguish the implemented SDK facade and protocol from the required FDO host encryption/network implementation. Current FDO hosts can reuse profiles through the existing Git operator API. No dedicated Git client, transport helper or new Git permission family ships inside this SDK package; these host features require a compatible FDO version. Interactive SSH, SFTP, agent forwarding, streaming SSH events and caller cancellation also remain outside v1. End-to-end cross-platform claims require packaged host verification. +Release notes must distinguish the implemented SDK facade and protocol from the required FDO host encryption/network implementation. Current FDO hosts can reuse profiles through the existing Git operator API. No dedicated Git client, transport helper or new Git permission family ships inside this SDK package; these host features require a compatible FDO version. Interactive SSH, SFTP and agent forwarding remain outside v1. Streaming events, cancellation and connection monitoring use the optional live adapters documented above; they do not change the buffered v1 request/reply actions. End-to-end cross-platform claims require packaged host verification. Use the repository's Release Please workflow for the version and changelog update. A conventional `feat` commit describes the additive feature; do not mark it as a breaking change or manually promise Git integration in the release title. The safe profile metadata fields are an additive extension to the SDK 1.8 service contract; older hosts returning only id/label remain compatible. Let Release Please determine the actual next version from the release baseline and commit history. ## Backend transport protocol v1 -Configure one transport per isolated plugin worker/runtime. The host must bind it to a trusted plugin identity from worker bootstrap, never an identity supplied by the plugin or derived from mutable metadata. `PluginRegistry.clearPlugin()` clears the adapter; the host must also revoke its endpoint and close in-flight work. Reconfiguration invalidates in-flight SDK responses. A transport has a single asynchronous `request(HostServiceRequest)` method. It resolves with the result object below, without a privileged-action envelope. +Configure one transport per isolated plugin worker/runtime. The host must bind it to a trusted plugin identity from worker bootstrap, never an identity supplied by the plugin or derived from mutable metadata. `PluginRegistry.clearPlugin()` clears the adapter; the host must also revoke its endpoint and close in-flight work. Reconfiguration invalidates in-flight SDK responses. The base transport has an asynchronous `request(HostServiceRequest)` method; optional live adapters extend it as described above. It resolves with the result object below, without a privileged-action envelope. | Action | Payload | Result | | --- | --- | --- | @@ -159,7 +166,7 @@ Configure one transport per isolated plugin worker/runtime. The host must bind i | `ssh.profiles.list` | `{}` | `{ profiles: SSHProfileSummary[] }` | | `ssh.exec` | `{ profileId, command, timeoutMs, maxOutputBytes }` | `SSHExecResult` | -Failures reject with a safe `code`: `ACCESS_DENIED`, `UNAVAILABLE`, `LOCKED`, `TIMEOUT`, `HOST_KEY_REJECTED`, `CONNECTION_FAILED`, or `INVALID_REQUEST`. The SDK wraps provider failures in `HostServiceError`, preserving only these codes and suppressing arbitrary provider messages. Unknown failures become `UNAVAILABLE`. Validation and local capability failures also reject before dispatch. SDK validation errors never interpolate secret values or host response data. +Failures reject with a safe `code`: `ACCESS_DENIED`, `UNAVAILABLE`, `LOCKED`, `TIMEOUT`, `CANCELLED`, `HOST_KEY_REJECTED`, `CONNECTION_FAILED`, or `INVALID_REQUEST`. The SDK wraps provider failures in `HostServiceError`, preserving only these codes and suppressing arbitrary provider messages. Unknown failures become `UNAVAILABLE`. Validation and local capability failures also reject before dispatch. SDK validation errors never interpolate secret values or host response data. No request accepts a plugin ID, filesystem path, encryption key or raw SSH authentication material. The host must validate all requests independently, reject unknown fields, limit secret size and namespace quotas, enforce capabilities using its own policy, and avoid logging secret request/response bodies. This interface is intentionally backend-only; never route it through `window.createBackendReq` or the generic UI privileged-action handler. diff --git a/src/PluginRegistry.ts b/src/PluginRegistry.ts index 1dd28a4..76178cc 100644 --- a/src/PluginRegistry.ts +++ b/src/PluginRegistry.ts @@ -1,6 +1,6 @@ import path from "path"; import { configureHostServices } from "./utils/hostServices"; -import { FDO_SDK, StoreType } from "./index"; +import { FDO_SDK } from "./index"; import { PluginMetadata } from "./PluginMetadata"; import { PluginDiagnostics, @@ -109,7 +109,7 @@ export class PluginRegistry { this.handlers[name] = handler; } - public static useStore(name: string = "default") : StoreType { + public static useStore(name: string = "default") : StoreWithLifecycle { const storeName = this.storeFactories[name] ? name : "default"; if (!this.storeFactories[name]) { diff --git a/src/StoreJson.ts b/src/StoreJson.ts index f2a6d7c..140ab3b 100644 --- a/src/StoreJson.ts +++ b/src/StoreJson.ts @@ -18,6 +18,7 @@ export type JsonStore = StoreWithLifecycle & { _filePath: string; _logger: Logger; _writeQueue: Promise; + _writeError: unknown; }; export function createJsonStore(options: JsonStoreOptions): JsonStore { @@ -28,6 +29,7 @@ export function createJsonStore(options: JsonStoreOptions): JsonStore { _filePath: options.filePath, _logger: logger, _writeQueue: Promise.resolve(), + _writeError: undefined, capabilities: { persistent: true, supportsAsyncFlush: true, @@ -73,14 +75,19 @@ export function createJsonStore(options: JsonStoreOptions): JsonStore { .then(async () => { await fs.promises.mkdir(path.dirname(this._filePath), { recursive: true }); await atomicWriteFile(this._filePath, snapshot, { encoding: "utf-8" }); + this._writeError = undefined; }) .catch((error) => { + this._writeError = error ?? new Error("JSON storage write failed."); this._logger.warn(`[StoreJson] Failed to save ${this._fileName}:`, error); }); }, _flush() { - return this._writeQueue; + // Capture this barrier: later saves must not change the outcome observed by this flush. + return this._writeQueue.then(() => { + if (this._writeError !== undefined) throw new Error("JSON storage write failed. Check FDO diagnostics."); + }); }, flush() { return this._flush(); diff --git a/src/types.ts b/src/types.ts index 6322d86..2fd7071 100644 --- a/src/types.ts +++ b/src/types.ts @@ -772,7 +772,7 @@ export type DiagnosticFixTemplate = { /** Secret operations require secrets.use plus an operation grant; SSH consumes references without secrets.get. */ export type HostServiceCapability = | "secrets.set" | "secrets.get" | "secrets.delete" | "secrets.list" | "secrets.use" | "secrets.consume" - | "ssh.use" | "ssh.exec" | "ssh.profiles.list"; + | "ssh.use" | "ssh.exec" | "ssh.profiles.list" | "ssh.status.watch"; /** References are names in the authenticated plugin namespace, not bearer credentials. */ export type SecretReference = Readonly<{ kind: "secret"; key: string }>; @@ -819,6 +819,33 @@ export type SSHExecResult = { truncated: boolean; }; +/** A host observation of an authenticated SSH connection, not a guarantee of VM availability. */ +export type SSHConnectionStatus = { + profileId: string; + state: "unknown" | "connecting" | "connected" | "disconnected"; + /** Host observation time (ISO timestamp). Null only when state is unknown. */ + checkedAt: string | null; + /** UI must treat the observation as stale after this interval without an update. */ + staleAfterMs: number; +}; +/** Sequence is contiguous from zero in host observation order across both streams. */ +export type SSHOutputEvent = { sequence: number; stream: "stdout" | "stderr"; data: string }; +/** Null means remote termination could not be established. Channel closure alone is insufficient. */ +export type SSHCancelResult = { remoteProcessStopped: boolean | null }; +export type SSHExecution = { + /** Single-consumer stream; consume concurrently with result. Break closes the event subscription only. */ + events: AsyncIterable; + result: Promise; + /** Idempotent request to stop this execution. Never retries the command. */ + cancel(): Promise; +}; +/** Host-owned execution. All fields are untrusted at the SDK boundary. */ +export type HostSSHExecution = { + events: AsyncIterable; + result: Promise; + cancel(): Promise; +}; + /** Backend-only protocol v1. No plugin id, storage path or raw SSH credentials are accepted. */ export type HostServiceRequest = | { action: "secrets.set"; payload: { key: string; value: string } } @@ -833,8 +860,11 @@ export type HostServiceRequest = */ export type HostServicesConfiguration = { request(request: HostServiceRequest): Promise; + /** Optional live adapters. An absent adapter fails with UNAVAILABLE, without buffered fallback. */ + watchSSHStatus?(options: { profileId: string }): Promise>; + startSSHExecution?(options: Required): Promise; }; export type HostServiceErrorCode = - | "ACCESS_DENIED" | "UNAVAILABLE" | "LOCKED" | "TIMEOUT" + | "ACCESS_DENIED" | "UNAVAILABLE" | "LOCKED" | "TIMEOUT" | "CANCELLED" | "HOST_KEY_REJECTED" | "CONNECTION_FAILED" | "INVALID_REQUEST"; diff --git a/src/utils/capabilities.ts b/src/utils/capabilities.ts index 3569d38..2ce3ea1 100644 --- a/src/utils/capabilities.ts +++ b/src/utils/capabilities.ts @@ -74,7 +74,7 @@ function getCapabilityPrerequisites(capability: string): string[] { const prerequisites = new Set(); if (capability.startsWith("secrets.") && capability !== "secrets.use") prerequisites.add("secrets.use"); if (capability.startsWith("ssh.") && capability !== "ssh.use") prerequisites.add("ssh.use"); - if (capability === "ssh.exec") { + if (capability === "ssh.exec" || capability === "ssh.status.watch") { prerequisites.add("secrets.use"); prerequisites.add("secrets.consume"); } @@ -321,13 +321,13 @@ export function describeCapability(capability: PluginCapability | string): Capab category: "secrets", }; } - if (capability === "ssh.use" || capability === "ssh.exec" || capability === "ssh.profiles.list") { + if (capability === "ssh.use" || capability === "ssh.exec" || capability === "ssh.profiles.list" || capability === "ssh.status.watch") { return { capability, - label: capability === "ssh.use" ? "SSH Access" : capability === "ssh.exec" ? "SSH Execution" : "SSH Profiles List", + label: capability === "ssh.use" ? "SSH Access" : capability === "ssh.exec" ? "SSH Execution" : capability === "ssh.status.watch" ? "SSH Connection Monitoring" : "SSH Profiles List", description: capability === "ssh.use" ? "Base permission to access SSH services; pair with specific SSH operation capabilities." - : "Allows this operation on authorized profiles when paired with ssh.use. Execution also requires secrets.use and secrets.consume.", + : "Allows this operation on authorized profiles when paired with ssh.use. Execution and monitoring also require secrets.use and secrets.consume.", category: "ssh", }; } @@ -650,6 +650,7 @@ export function createSecretsCapabilityBundle( } /** SSH uses references inside the host, so plaintext secret reads are not required. */ -export function createSSHCapabilityBundle(): PluginCapability[] { - return createCapabilityBundle(["ssh.use", "ssh.exec", "ssh.profiles.list", "secrets.use", "secrets.consume"]); +export function createSSHCapabilityBundle(options: { monitorStatus?: boolean } = {}): PluginCapability[] { + return createCapabilityBundle(["ssh.use", "ssh.exec", "ssh.profiles.list", "secrets.use", "secrets.consume", + ...(options.monitorStatus ? ["ssh.status.watch" as const] : [])]); } diff --git a/src/utils/contracts.ts b/src/utils/contracts.ts index 6a456f7..77d39d2 100644 --- a/src/utils/contracts.ts +++ b/src/utils/contracts.ts @@ -26,7 +26,7 @@ export interface PluginInitPayload { const KNOWN_PLUGIN_CAPABILITIES = new Set([ "secrets.set", "secrets.get", "secrets.delete", "secrets.list", "secrets.use", "secrets.consume", - "ssh.use", "ssh.exec", "ssh.profiles.list", + "ssh.use", "ssh.exec", "ssh.profiles.list", "ssh.status.watch", "system.ai", "system.ai.assistants.list", "system.ai.request", diff --git a/src/utils/hostServices.ts b/src/utils/hostServices.ts index cd624b9..cbd7791 100644 --- a/src/utils/hostServices.ts +++ b/src/utils/hostServices.ts @@ -1,6 +1,7 @@ import type { HostServiceRequest, HostServicesConfiguration, SecretMetadata, SecretReference, SSHExecOptions, SSHExecResult, SSHProfileSummary, HostServiceErrorCode, + SSHExecution, SSHConnectionStatus, SSHOutputEvent, SSHCancelResult, HostServiceCapability, } from "../types"; import { requireCapability } from "./capabilities"; @@ -15,13 +16,23 @@ export class HostServiceError extends Error { } let configuration: Readonly | undefined; +const activeResources = new Set<() => void>(); /** Host bootstrap only. The transport must be bound to a trusted plugin identity. */ export function configureHostServices(value: HostServicesConfiguration | null): void { if (value !== null && (!value || typeof value.request !== "function")) { throw new Error("Host services require a request transport."); } - configuration = value === null ? undefined : Object.freeze({ request: value.request.bind(value) }); + for (const key of ["watchSSHStatus", "startSSHExecution"] as const) { + if (value?.[key] !== undefined && typeof value[key] !== "function") throw new Error("Invalid SSH host adapter."); + } + for (const cleanup of activeResources) cleanup(); + activeResources.clear(); + configuration = value === null ? undefined : Object.freeze({ + request: value.request.bind(value), + watchSSHStatus: value.watchSSHStatus?.bind(value), + startSSHExecution: value.startSSHExecution?.bind(value), + }); } function validKey(key: unknown): asserts key is string { @@ -40,36 +51,85 @@ function invalidResponse(): never { throw new Error("Invalid host service response."); } -async function request(request: HostServiceRequest): Promise { +function authorize(action: HostServiceCapability): void { if (typeof window !== "undefined") { throw new Error("Host services are only available in the plugin backend runtime."); } - if (request.action.startsWith("ssh.")) requireCapability("ssh.use", "access SSH services"); - if (request.action.startsWith("secrets.") || request.action === "ssh.exec") { + if (action.startsWith("ssh.")) requireCapability("ssh.use", "access SSH services"); + if (action.startsWith("secrets.") || action === "ssh.exec" || action === "ssh.status.watch") { requireCapability("secrets.use", "access the secret store"); } - requireCapability(request.action, "access host services"); - if (request.action === "ssh.exec") { - requireCapability("secrets.consume", "consume secret references for SSH execution"); + requireCapability(action, "access host services"); + if (action === "ssh.exec" || action === "ssh.status.watch") { + requireCapability("secrets.consume", "consume secret references for SSH"); } - const transport = configuration; - if (!transport) { +} + +function safeError(error: unknown): HostServiceError { + const codes: HostServiceErrorCode[] = ["ACCESS_DENIED", "UNAVAILABLE", "LOCKED", "TIMEOUT", "CANCELLED", "HOST_KEY_REJECTED", "CONNECTION_FAILED", "INVALID_REQUEST"]; + return new HostServiceError(record(error) && codes.includes(error.code as HostServiceErrorCode) + ? error.code as HostServiceErrorCode : "UNAVAILABLE"); +} + +function getTransport(action: HostServiceCapability): Readonly { + authorize(action); + if (!configuration) { throw new Error("Host services are unavailable. The FDO host must configure a backend transport before plugin initialization."); } - let result: unknown; - try { - result = await transport.request(request); - } catch (error) { - const codes: HostServiceErrorCode[] = ["ACCESS_DENIED", "UNAVAILABLE", "LOCKED", "TIMEOUT", "HOST_KEY_REJECTED", "CONNECTION_FAILED", "INVALID_REQUEST"]; - const code = record(error) && codes.includes(error.code as HostServiceErrorCode) - ? error.code as HostServiceErrorCode : "UNAVAILABLE"; - // Provider errors must not accidentally leak credentials or request payloads. - throw new HostServiceError(code); - } + return configuration; +} + +function assertCurrent(transport: Readonly, action: HostServiceCapability): void { if (configuration !== transport) throw new HostServiceError("UNAVAILABLE"); + authorize(action); +} + +async function request(request: HostServiceRequest): Promise { + const transport = getTransport(request.action); + let result: unknown; + try { result = await transport.request(request); } + catch (error) { throw safeError(error); } + assertCurrent(transport, request.action); return result; } +function execOptions(options: SSHExecOptions): Required { + if (!record(options) || typeof options.profileId !== "string" || !options.profileId.trim() + || typeof options.command !== "string" || !options.command.trim() || options.command.includes("\0")) { + throw new Error("SSH execution requires a profileId and a nonempty command without NUL characters."); + } + for (const value of [options.timeoutMs, options.maxOutputBytes]) { + if (value !== undefined && (!Number.isSafeInteger(value) || value <= 0)) { + throw new Error("SSH timeout and output limits must be positive safe integers."); + } + } + return { profileId: options.profileId, command: options.command, + timeoutMs: options.timeoutMs ?? 30_000, maxOutputBytes: options.maxOutputBytes ?? 1_048_576 }; +} + +function execResult(result: unknown, maxBytes: number): SSHExecResult { + if (!record(result) || typeof result.stdout !== "string" || typeof result.stderr !== "string" + || (result.exitCode !== null && !Number.isSafeInteger(result.exitCode)) + || (result.signal !== null && typeof result.signal !== "string") + || typeof result.durationMs !== "number" || !Number.isFinite(result.durationMs) || result.durationMs < 0 + || typeof result.truncated !== "boolean" + || Buffer.byteLength(result.stdout) + Buffer.byteLength(result.stderr) > maxBytes) invalidResponse(); + return { stdout: result.stdout, stderr: result.stderr, + exitCode: result.exitCode as number | null, signal: result.signal as string | null, + durationMs: result.durationMs, truncated: result.truncated }; +} + +function iteratorFor(source: unknown): AsyncIterator { + if (!source || typeof (source as AsyncIterable)[Symbol.asyncIterator] !== "function") invalidResponse(); + const iterator = (source as AsyncIterable)[Symbol.asyncIterator](); + if (!iterator || typeof iterator.next !== "function") invalidResponse(); + return iterator; +} + +function closeIterator(iterator: AsyncIterator): void { + try { void Promise.resolve(iterator.return?.()).catch(() => {}); } catch { /* Best-effort teardown. */ } +} + /** Universal plugin-scoped secret storage. Persistence and isolation belong to the host. */ export class SecretStore { /** Create or replace a UTF-8 string (tokens, PEM keys, passwords or serialized credentials). */ @@ -146,31 +206,101 @@ export class SSHClient { /** Buffered noninteractive execution; remote nonzero exits are returned, transport failures reject. */ async exec(options: SSHExecOptions): Promise { - if (!record(options) || typeof options.profileId !== "string" || !options.profileId.trim() - || typeof options.command !== "string" || !options.command.trim() || options.command.includes("\0")) { - throw new Error("SSH execution requires a profileId and a nonempty command without NUL characters."); - } - for (const value of [options.timeoutMs, options.maxOutputBytes]) { - if (value !== undefined && (!Number.isSafeInteger(value) || value <= 0)) { - throw new Error("SSH timeout and output limits must be positive safe integers."); + const payload = execOptions(options); + return execResult(await request({ action: "ssh.exec", payload }), payload.maxOutputBytes); + } + + /** Requires an optional host monitor adapter and ssh.status.watch; break/return unsubscribes. */ + async *watchStatus(profileId: string): AsyncIterableIterator { + if (typeof profileId !== "string" || !profileId.trim()) throw new Error("SSH monitoring requires a profileId."); + const transport = getTransport("ssh.status.watch"); + if (!transport.watchSSHStatus) throw new HostServiceError("UNAVAILABLE"); + let source: AsyncIterable; + try { source = await transport.watchSSHStatus({ profileId }); } + catch (error) { throw safeError(error); } + const iterator = iteratorFor(source); + const cleanup = () => closeIterator(iterator); + activeResources.add(cleanup); + try { + while (true) { + assertCurrent(transport, "ssh.status.watch"); + let next: IteratorResult; + try { next = await iterator.next(); } catch (error) { throw safeError(error); } + assertCurrent(transport, "ssh.status.watch"); + if (next.done) return; + const status = next.value; + if (!record(status) || status.profileId !== profileId + || !["unknown", "connecting", "connected", "disconnected"].includes(status.state as string) + || !(status.checkedAt === null && status.state === "unknown" + || typeof status.checkedAt === "string" && Number.isFinite(Date.parse(status.checkedAt))) + || !Number.isSafeInteger(status.staleAfterMs) || (status.staleAfterMs as number) <= 0) invalidResponse(); + yield { profileId, state: status.state as SSHConnectionStatus["state"], + checkedAt: status.checkedAt as string | null, staleAfterMs: status.staleAfterMs as number }; } + } finally { + activeResources.delete(cleanup); + cleanup(); } - // Project allowed fields so arbitrary caller properties cannot cross the boundary. - const result = await request({ action: "ssh.exec", payload: { - profileId: options.profileId, - command: options.command, - timeoutMs: options.timeoutMs ?? 30_000, - maxOutputBytes: options.maxOutputBytes ?? 1_048_576, - } }); - if (!record(result) || typeof result.stdout !== "string" || typeof result.stderr !== "string" - || (result.exitCode !== null && !Number.isSafeInteger(result.exitCode)) - || (result.signal !== null && typeof result.signal !== "string") - || typeof result.durationMs !== "number" || !Number.isFinite(result.durationMs) || result.durationMs < 0 - || typeof result.truncated !== "boolean") invalidResponse(); - return { - stdout: result.stdout as string, stderr: result.stderr as string, - exitCode: result.exitCode as number | null, signal: result.signal as string | null, - durationMs: result.durationMs as number, truncated: result.truncated as boolean, + } + + /** Live output and explicit cancellation; requires the optional host execution adapter. No retries. */ + async start(options: SSHExecOptions): Promise { + const payload = execOptions(options); + const transport = getTransport("ssh.exec"); + if (!transport.startSSHExecution) throw new HostServiceError("UNAVAILABLE"); + let host: Awaited>>; + try { host = await transport.startSSHExecution(payload); } + catch (error) { throw safeError(error); } + if (!host || typeof host.cancel !== "function" || !host.result || typeof host.result.then !== "function") invalidResponse(); + // Attach immediately: a caller may be consuming events before awaiting result. + const result = Promise.resolve(host.result).then(value => { + assertCurrent(transport, "ssh.exec"); + return execResult(value, payload.maxOutputBytes); + }, error => { throw safeError(error); }); + void result.catch(() => {}); + const iterator = iteratorFor(host.events); + let cancelPromise: Promise | undefined; + const cancel = async (): Promise => { + assertCurrent(transport, "ssh.exec"); + return cancelPromise ??= Promise.resolve().then(() => host.cancel()).catch(error => { throw safeError(error); }).then(value => { + assertCurrent(transport, "ssh.exec"); + if (!record(value) || (value.remoteProcessStopped !== null && typeof value.remoteProcessStopped !== "boolean")) invalidResponse(); + return { remoteProcessStopped: value.remoteProcessStopped as boolean | null }; + }); + }; + const cleanup = () => { + closeIterator(iterator); + // Host cancellation is required to be idempotent and scoped to this execution. + try { void Promise.resolve(host.cancel()).catch(() => {}); } catch { /* Best-effort teardown. */ } + }; + activeResources.add(cleanup); + try { assertCurrent(transport, "ssh.exec"); } + catch (error) { activeResources.delete(cleanup); cleanup(); throw error; } + let sequence = 0; + let bytes = 0; + let consumed = false; + const events: AsyncIterable = { + async *[Symbol.asyncIterator]() { + if (consumed) throw new Error("SSH output supports one consumer."); + consumed = true; + try { + while (true) { + assertCurrent(transport, "ssh.exec"); + let next: IteratorResult; + try { next = await iterator.next(); } catch (error) { throw safeError(error); } + assertCurrent(transport, "ssh.exec"); + if (next.done) return; + const event = next.value; + if (!record(event) || event.sequence !== sequence++ + || (event.stream !== "stdout" && event.stream !== "stderr") || typeof event.data !== "string") invalidResponse(); + bytes += Buffer.byteLength(event.data); + if (bytes > payload.maxOutputBytes) invalidResponse(); + yield { sequence: event.sequence as number, stream: event.stream, data: event.data }; + } + } finally { closeIterator(iterator); } + }, }; + void result.finally(() => { activeResources.delete(cleanup); }).catch(() => {}); + return { events, result, cancel }; } } diff --git a/tests/StoreJson.test.ts b/tests/StoreJson.test.ts index f8f9bcf..eae66d8 100644 --- a/tests/StoreJson.test.ts +++ b/tests/StoreJson.test.ts @@ -47,6 +47,7 @@ describe('StoreJson', () => { beforeEach(() => { vi.clearAllMocks(); + vi.mocked(atomicWriteFile).mockResolvedValue(undefined); store = createJsonStore({ filePath: mockFilePath, logger: new Logger() as any }); store._store = {}; }); @@ -145,7 +146,8 @@ describe('StoreJson', () => { (atomicWriteFile as vi.Mock).mockRejectedValue(new Error('Write error')); expect(() => store.set('testKey', 'testValue')).not.toThrow(); - await store._flush(); + await expect(store.flush!()).rejects.toThrow("JSON storage write failed"); + await expect(store.dispose!()).rejects.toThrow("JSON storage write failed"); expect(atomicWriteFile).toHaveBeenCalled(); // Verify that the logger was called with a warning @@ -211,6 +213,23 @@ describe('StoreJson', () => { }); describe('persistence', () => { + it('recovers after a failed write when a later full snapshot is persisted', async () => { + vi.mocked(atomicWriteFile).mockRejectedValueOnce(new Error('disk full')); + store.set('first', 1); + await expect(store.flush!()).rejects.toThrow('JSON storage write failed'); + store.set('second', 2); + await expect(store.flush!()).resolves.toBeUndefined(); + expect(atomicWriteFile).toHaveBeenLastCalledWith(mockFilePath, + JSON.stringify({ first: 1, second: 2 }, null, 2), { encoding: 'utf-8' }); + }); + + it('reports directory creation failures through the flush barrier', async () => { + vi.mocked(fs.promises.mkdir).mockRejectedValueOnce(new Error('permission denied')); + store.set('key', 'value'); + await expect(store.flush!()).rejects.toThrow('JSON storage write failed'); + expect(atomicWriteFile).not.toHaveBeenCalled(); + }); + it('should write to file when setting a value', async () => { store.set('testKey', 'testValue'); await store._flush(); diff --git a/tests/utils/sshLive.test.ts b/tests/utils/sshLive.test.ts new file mode 100644 index 0000000..00bd18d --- /dev/null +++ b/tests/utils/sshLive.test.ts @@ -0,0 +1,168 @@ +import { PluginRegistry, SSHClient, createSSHCapabilityBundle, runCapabilityPreflight, validatePluginInitPayload } from '../../src'; +import type { HostServicesConfiguration } from '../../src'; +import { resetCapabilityStateForTests } from '../../src/utils/capabilities'; + +const ssh = new SSHClient(); +const options = { profileId: 'vm', command: 'uptime' }; +const completed = { stdout: 'ok', stderr: '', exitCode: 0, signal: null, durationMs: 1, truncated: false }; +async function* values(...items: unknown[]) { yield* items; } +function setup(adapters: Partial = {}) { + PluginRegistry.configureCapabilities({ granted: createSSHCapabilityBundle({ monitorStatus: true }) }); + PluginRegistry.configureHostServices({ request: vi.fn(), ...adapters }); +} +beforeEach(() => { resetCapabilityStateForTests(); setup(); }); +afterEach(() => { PluginRegistry.configureHostServices(null); resetCapabilityStateForTests(); vi.unstubAllGlobals(); }); + +test('old hosts reject live APIs without running a buffered command', async () => { + await expect(ssh.start(options)).rejects.toMatchObject({ code: 'UNAVAILABLE' }); + await expect(ssh.watchStatus('vm').next()).rejects.toMatchObject({ code: 'UNAVAILABLE' }); +}); + +test('monitoring has a separate grant and secret consumption prerequisites', async () => { + const watchSSHStatus = vi.fn(); + setup({ watchSSHStatus }); + PluginRegistry.configureCapabilities({ granted: createSSHCapabilityBundle() }); + await expect(ssh.watchStatus('vm').next()).rejects.toThrow('ssh.status.watch'); + expect(watchSSHStatus).not.toHaveBeenCalled(); + expect(runCapabilityPreflight({ declared: ['ssh.status.watch'], granted: [] }).missing[0].requiredCapabilities) + .toEqual(['secrets.consume', 'secrets.use', 'ssh.status.watch', 'ssh.use']); + expect(() => validatePluginInitPayload({ capabilities: ['ssh.status.watch'] })).not.toThrow(); +}); + +test('status projection includes freshness and iterator return unsubscribes', async () => { + const close = vi.fn(); + const status = { profileId: 'vm', state: 'connected', checkedAt: '2026-10-18T00:00:00Z', staleAfterMs: 5000 }; + setup({ watchSSHStatus: async () => (async function* () { try { yield { ...status, password: 'SECRET' }; } finally { close(); } })() }); + const monitor = ssh.watchStatus('vm'); + try { + expect(await monitor.next()).toEqual({ value: status, done: false }); + } finally { + await monitor.return(); + } + expect(close).toHaveBeenCalledOnce(); +}); + +test.each([ + { profileId: 'other' }, { state: 'ready' }, { checkedAt: null }, { staleAfterMs: 0 }, { checkedAt: 'SECRET' }, +])('rejects malformed status %j', async invalid => { + setup({ watchSSHStatus: async () => values({ profileId: 'vm', state: 'connected', checkedAt: '2026-10-18', staleAfterMs: 1, ...invalid }) }); + await expect(ssh.watchStatus('vm').next()).rejects.toThrow('Invalid host service response'); +}); + +test('output arrives while execution is pending; cancel is scoped, sanitized and idempotent', async () => { + let finish!: (value: unknown) => void; + const cancel = vi.fn().mockResolvedValue({ remoteProcessStopped: null, secret: 'SECRET' }); + const startSSHExecution = vi.fn().mockResolvedValue({ + events: values({ sequence: 0, stream: 'stdout', data: 'o', secret: 'SECRET' }, { sequence: 1, stream: 'stderr', data: 'k' }), + result: new Promise(resolve => { finish = resolve; }), cancel, + }); + setup({ startSSHExecution }); + const execution = await ssh.start({ ...options, password: 'SECRET' } as never); + expect(startSSHExecution).toHaveBeenCalledWith({ ...options, timeoutMs: 30000, maxOutputBytes: 1048576 }); + const output = []; + for await (const event of execution.events) output.push(event); + expect(output).toEqual([{ sequence: 0, stream: 'stdout', data: 'o' }, { sequence: 1, stream: 'stderr', data: 'k' }]); + await expect(execution.cancel()).resolves.toEqual({ remoteProcessStopped: null }); + await expect(execution.cancel()).resolves.toEqual({ remoteProcessStopped: null }); + expect(cancel).toHaveBeenCalledOnce(); + finish(completed); + await expect(execution.result).resolves.toEqual(completed); +}); + +test.each([ + [{ sequence: 1, stream: 'stdout', data: 'x' }], + [{ sequence: 0, stream: 'stdin', data: 'x' }], + [{ sequence: 0, stream: 'stdout', data: 'éé' }], + [{ sequence: 0, stream: 'stdout', data: 'é' }, { sequence: 1, stream: 'stderr', data: 'é' }], +])('rejects output with invalid sequence, stream or combined UTF-8 budget: %j', async (...events) => { + setup({ startSSHExecution: async () => ({ events: values(...events), result: Promise.resolve({ ...completed, stdout: '' }), cancel: async () => ({ remoteProcessStopped: null }) }) }); + const execution = await ssh.start({ ...options, maxOutputBytes: 3 }); + await expect((async () => { for await (const _event of execution.events) { /* drain */ } })()).rejects.toThrow('Invalid host service response'); +}); + +test('completion before consumption preserves buffered events and truncation', async () => { + setup({ startSSHExecution: async () => ({ events: values({ sequence: 0, stream: 'stdout', data: 'ok' }), result: Promise.resolve({ ...completed, truncated: true }), cancel: async () => ({ remoteProcessStopped: true }) }) }); + const execution = await ssh.start(options); + expect((await execution.result).truncated).toBe(true); + const received = []; + for await (const event of execution.events) received.push(event.data); + expect(received).toEqual(['ok']); +}); + +test('teardown invalidates in-flight results and cancels pending execution', async () => { + let finish!: (value: unknown) => void; + const cancel = vi.fn().mockResolvedValue({ remoteProcessStopped: null }); + setup({ startSSHExecution: async () => ({ events: values(), result: new Promise(resolve => { finish = resolve; }), cancel }) }); + const execution = await ssh.start(options); + PluginRegistry.configureHostServices(null); + finish(completed); + await expect(execution.result).rejects.toMatchObject({ code: 'UNAVAILABLE' }); + expect(cancel).toHaveBeenCalledOnce(); +}); + +test('a start resolved after teardown is cancelled without returning a live handle', async () => { + let finish!: (value: any) => void; + const cancel = vi.fn().mockResolvedValue({ remoteProcessStopped: null }); + setup({ startSSHExecution: () => new Promise(resolve => { finish = resolve; }) }); + const pending = ssh.start(options); + PluginRegistry.configureHostServices(null); + finish({ events: values(), result: Promise.resolve(completed), cancel }); + await expect(pending).rejects.toMatchObject({ code: 'UNAVAILABLE' }); + expect(cancel).toHaveBeenCalledOnce(); +}); + +test('live adapters require backend execution and all existing exec grants', async () => { + const startSSHExecution = vi.fn(); + setup({ startSSHExecution }); + vi.stubGlobal('window', {}); + await expect(ssh.start(options)).rejects.toThrow('backend runtime'); + vi.unstubAllGlobals(); + PluginRegistry.configureCapabilities({ granted: ['ssh.use', 'ssh.exec', 'secrets.use'] }); + await expect(ssh.start(options)).rejects.toThrow('secrets.consume'); + expect(startSSHExecution).not.toHaveBeenCalled(); +}); + +test('host live errors are sanitized and commands are never retried', async () => { + const startSSHExecution = vi.fn().mockRejectedValue({ code: 'TIMEOUT', message: 'SECRET' }); + setup({ startSSHExecution }); + await expect(ssh.start(options)).rejects.toMatchObject({ code: 'TIMEOUT', message: 'Host service request failed (TIMEOUT).' }); + expect(startSSHExecution).toHaveBeenCalledOnce(); +}); + +test('event and cancellation failures preserve safe codes and hide provider messages', async () => { + setup({ startSSHExecution: async () => ({ + events: (async function* () { throw { code: 'CONNECTION_FAILED', message: 'SECRET' }; })(), + result: Promise.reject({ code: 'CANCELLED', message: 'SECRET' }), + cancel: async () => { throw { code: 'ACCESS_DENIED', message: 'SECRET' }; }, + }) }); + const execution = await ssh.start(options); + await expect(execution.result).rejects.toMatchObject({ code: 'CANCELLED', message: 'Host service request failed (CANCELLED).' }); + await expect(execution.events[Symbol.asyncIterator]().next()).rejects.toMatchObject({ code: 'CONNECTION_FAILED' }); + await expect(execution.cancel()).rejects.toMatchObject({ code: 'ACCESS_DENIED' }); +}); + +test('monitoring teardown closes subscriptions and suppresses later delivery', async () => { + const close = vi.fn(); + setup({ watchSSHStatus: async () => (async function* () { + try { + yield { profileId: 'vm', state: 'unknown', checkedAt: null, staleAfterMs: 1000 }; + yield { profileId: 'vm', state: 'connected', checkedAt: '2026-10-18', staleAfterMs: 1000 }; + } finally { close(); } + })() }); + const monitor = ssh.watchStatus('vm'); + await monitor.next(); + PluginRegistry.configureHostServices(null); + await expect(monitor.next()).rejects.toMatchObject({ code: 'UNAVAILABLE' }); + expect(close).toHaveBeenCalledOnce(); +}); + +test('grant revocation prevents further event delivery and cancel calls', async () => { + const cancel = vi.fn(); + setup({ startSSHExecution: async () => ({ events: values({ sequence: 0, stream: 'stdout', data: 'ok' }), result: Promise.resolve(completed), cancel }) }); + const execution = await ssh.start(options); + await execution.result; + PluginRegistry.configureCapabilities({ granted: [] }); + await expect(execution.events[Symbol.asyncIterator]().next()).rejects.toThrow('Capability'); + await expect(execution.cancel()).rejects.toThrow('Capability'); + expect(cancel).not.toHaveBeenCalled(); +}); From a02313d1d09c7a8c0e1f4382f9cc0e3cab302051 Mon Sep 17 00:00:00 2001 From: anikitenko Date: Sun, 18 Oct 2026 13:53:08 +0300 Subject: [PATCH 2/2] fix: address comments --- src/utils/hostServices.ts | 68 ++++++++++++++++++------ tests/utils/sshLive.test.ts | 100 +++++++++++++++++++++++++++++++++++- 2 files changed, 150 insertions(+), 18 deletions(-) diff --git a/src/utils/hostServices.ts b/src/utils/hostServices.ts index cbd7791..0cb4acd 100644 --- a/src/utils/hostServices.ts +++ b/src/utils/hostServices.ts @@ -126,6 +126,11 @@ function iteratorFor(source: unknown): AsyncIterator { return iterator; } +async function nextHostEvent(iterator: AsyncIterator): Promise> { + try { return await iterator.next(); } + catch (error) { throw safeError(error); } +} + function closeIterator(iterator: AsyncIterator): void { try { void Promise.resolve(iterator.return?.()).catch(() => {}); } catch { /* Best-effort teardown. */ } } @@ -224,8 +229,7 @@ export class SSHClient { try { while (true) { assertCurrent(transport, "ssh.status.watch"); - let next: IteratorResult; - try { next = await iterator.next(); } catch (error) { throw safeError(error); } + const next = await nextHostEvent(iterator); assertCurrent(transport, "ssh.status.watch"); if (next.done) return; const status = next.value; @@ -252,13 +256,50 @@ export class SSHClient { try { host = await transport.startSSHExecution(payload); } catch (error) { throw safeError(error); } if (!host || typeof host.cancel !== "function" || !host.result || typeof host.result.then !== "function") invalidResponse(); + let iterator: AsyncIterator | undefined; + let resultSettled = false; + let eventsClosed = false; + let closingEvents: Promise | undefined; + let cleaned = false; + const releaseIfFinished = () => { + if (resultSettled && eventsClosed) activeResources.delete(cleanup); + }; + const closeEvents = (): Promise => { + if (!closingEvents) { + let closing: Promise; + try { closing = Promise.resolve(iterator?.return?.()); } + catch { closing = Promise.resolve(); } + closingEvents = closing.catch(() => {}).then(() => { + eventsClosed = true; + releaseIfFinished(); + }); + } + return closingEvents; + }; + const cleanup = () => { + if (cleaned) return; + cleaned = true; + activeResources.delete(cleanup); + void closeEvents(); + // Register cancellation before validating events: start already launched the command. + try { void Promise.resolve(host.cancel()).catch(() => {}); } catch { /* Best-effort teardown. */ } + }; + activeResources.add(cleanup); // Attach immediately: a caller may be consuming events before awaiting result. const result = Promise.resolve(host.result).then(value => { assertCurrent(transport, "ssh.exec"); return execResult(value, payload.maxOutputBytes); - }, error => { throw safeError(error); }); - void result.catch(() => {}); - const iterator = iteratorFor(host.events); + }, error => { + assertCurrent(transport, "ssh.exec"); + throw safeError(error); + }); + const settleResult = () => { resultSettled = true; releaseIfFinished(); }; + void result.then(settleResult, settleResult); + let outputIterator: AsyncIterator; + try { + outputIterator = iterator = iteratorFor(host.events); + assertCurrent(transport, "ssh.exec"); + } catch (error) { cleanup(); throw error; } let cancelPromise: Promise | undefined; const cancel = async (): Promise => { assertCurrent(transport, "ssh.exec"); @@ -268,14 +309,6 @@ export class SSHClient { return { remoteProcessStopped: value.remoteProcessStopped as boolean | null }; }); }; - const cleanup = () => { - closeIterator(iterator); - // Host cancellation is required to be idempotent and scoped to this execution. - try { void Promise.resolve(host.cancel()).catch(() => {}); } catch { /* Best-effort teardown. */ } - }; - activeResources.add(cleanup); - try { assertCurrent(transport, "ssh.exec"); } - catch (error) { activeResources.delete(cleanup); cleanup(); throw error; } let sequence = 0; let bytes = 0; let consumed = false; @@ -286,8 +319,7 @@ export class SSHClient { try { while (true) { assertCurrent(transport, "ssh.exec"); - let next: IteratorResult; - try { next = await iterator.next(); } catch (error) { throw safeError(error); } + const next = await nextHostEvent(outputIterator); assertCurrent(transport, "ssh.exec"); if (next.done) return; const event = next.value; @@ -297,10 +329,12 @@ export class SSHClient { if (bytes > payload.maxOutputBytes) invalidResponse(); yield { sequence: event.sequence as number, stream: event.stream, data: event.data }; } - } finally { closeIterator(iterator); } + } catch (error) { + cleanup(); + throw error; + } finally { await closeEvents(); } }, }; - void result.finally(() => { activeResources.delete(cleanup); }).catch(() => {}); return { events, result, cancel }; } } diff --git a/tests/utils/sshLive.test.ts b/tests/utils/sshLive.test.ts index 00bd18d..23c6148 100644 --- a/tests/utils/sshLive.test.ts +++ b/tests/utils/sshLive.test.ts @@ -156,13 +156,111 @@ test('monitoring teardown closes subscriptions and suppresses later delivery', a expect(close).toHaveBeenCalledOnce(); }); -test('grant revocation prevents further event delivery and cancel calls', async () => { +test('grant revocation prevents delivery and caller cancellation while cleaning up execution', async () => { const cancel = vi.fn(); setup({ startSSHExecution: async () => ({ events: values({ sequence: 0, stream: 'stdout', data: 'ok' }), result: Promise.resolve(completed), cancel }) }); const execution = await ssh.start(options); await execution.result; PluginRegistry.configureCapabilities({ granted: [] }); await expect(execution.events[Symbol.asyncIterator]().next()).rejects.toThrow('Capability'); + expect(cancel).toHaveBeenCalledOnce(); await expect(execution.cancel()).rejects.toThrow('Capability'); + expect(cancel).toHaveBeenCalledOnce(); +}); + +test.each([undefined, null, {}, { [Symbol.asyncIterator]: () => ({}) }])( + 'cancels a started execution when its event source is malformed: %j', async events => { + let rejectResult!: (reason: unknown) => void; + const cancel = vi.fn(() => { + rejectResult({ code: 'CANCELLED' }); + return Promise.reject(new Error('provider cancellation failed')); + }); + setup({ startSSHExecution: async () => ({ + events: events as AsyncIterable, + result: new Promise((_, reject) => { rejectResult = reject; }), cancel, + }) }); + await expect(ssh.start(options)).rejects.toThrow('Invalid host service response'); + expect(cancel).toHaveBeenCalledOnce(); + PluginRegistry.configureHostServices(null); + expect(cancel).toHaveBeenCalledOnce(); + }, +); + +test.each(['clear', 'replace'] as const)('invalidates provider rejections after transport %s', async action => { + let rejectResult!: (reason: unknown) => void; + const cancel = vi.fn(async () => { + rejectResult({ code: 'CANCELLED', message: 'SECRET' }); + return { remoteProcessStopped: null }; + }); + setup({ startSSHExecution: async () => ({ + events: values(), result: new Promise((_, reject) => { rejectResult = reject; }), cancel, + }) }); + const execution = await ssh.start(options); + if (action === 'clear') PluginRegistry.configureHostServices(null); + else setup(); + await expect(execution.result).rejects.toMatchObject({ code: 'UNAVAILABLE', message: 'Host service request failed (UNAVAILABLE).' }); + expect(cancel).toHaveBeenCalledOnce(); +}); + +test.each([false, true])('teardown closes output after result settles (consumer started: %j)', async consume => { + const close = vi.fn().mockResolvedValue({ done: true }); + const cancel = vi.fn().mockResolvedValue({ remoteProcessStopped: null }); + setup({ startSSHExecution: async () => ({ + events: { [Symbol.asyncIterator]: () => ({ + next: async () => ({ value: { sequence: 0, stream: 'stdout', data: 'ok' }, done: false }), + return: close, + }) }, + result: Promise.resolve(completed), cancel, + }) }); + const execution = await ssh.start(options); + const consumer = consume ? execution.events[Symbol.asyncIterator]() : undefined; + if (consumer) await consumer.next(); + await execution.result; + expect(close).not.toHaveBeenCalled(); + PluginRegistry.configureHostServices(null); + expect(close).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledOnce(); + if (consumer) await consumer.return?.(); + expect(close).toHaveBeenCalledOnce(); +}); + +test.each([false, true])('releases completed execution after result and output close (result first: %j)', async resultFirst => { + let finish!: (value: unknown) => void; + const cancel = vi.fn(); + const close = vi.fn().mockResolvedValue({ done: true }); + setup({ startSSHExecution: async () => ({ + events: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true }), return: close }) }, + result: new Promise(resolve => { finish = resolve; }), cancel, + }) }); + const execution = await ssh.start(options); + if (resultFirst) { finish(completed); await execution.result; } + for await (const _event of execution.events) { /* drain */ } + if (!resultFirst) { finish(completed); await execution.result; } + PluginRegistry.configureHostServices(null); + expect(close).toHaveBeenCalledOnce(); expect(cancel).not.toHaveBeenCalled(); }); + +test('retains teardown cleanup while asynchronous output closure is pending', async () => { + let finishClose!: () => void; + let notifyCloseStarted!: () => void; + const closeStarted = new Promise(resolve => { notifyCloseStarted = resolve; }); + const close = vi.fn(() => new Promise<{ done: true }>(resolve => { + finishClose = () => resolve({ done: true }); + notifyCloseStarted(); + })); + const cancel = vi.fn().mockResolvedValue({ remoteProcessStopped: null }); + setup({ startSSHExecution: async () => ({ + events: { [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true }), return: close }) }, + result: Promise.resolve(completed), cancel, + }) }); + const execution = await ssh.start(options); + await execution.result; + const outputDone = execution.events[Symbol.asyncIterator]().next(); + await closeStarted; + PluginRegistry.configureHostServices(null); + expect(cancel).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + finishClose(); + await outputDone; +});