Add cancellation, tunnel recovery, and typed beacon tasks

README.md

@@ -29,6 +29,45 @@ same path on Linux/amd64, Windows/amd64, and macOS/arm64. See
 
 Operator connections use direct mTLS. Configurations containing a WireGuard operator profile are rejected before credentials or channels are created. WireGuard listener and implant C2 APIs remain available.
 
+### Cancellation and deadlines
+
+Existing numeric timeout arguments remain supported. Convenience RPCs, `connect`,
+and shell startup also accept `{ timeoutSeconds, signal }` in the same argument
+position. Omitting `timeoutSeconds` preserves the method's default; zero disables
+the local deadline while still allowing cancellation. `OperationAbortedError`
+and `OperationTimeoutError` distinguish caller cancellation from deadline expiry.
+Shell startup shares one deadline across tunnel creation, the server's bind
+acknowledgement, and shell creation. Cancellation stops waiting and requests
+transport cancellation; it does not undo a remote operation already accepted.
+
+```typescript
+const controller = new AbortController()
+const sessions = await client.getSessions({ timeoutSeconds: 15, signal: controller.signal })
+```
+
+### Typed beacon tasks
+
+Existing beacon commands keep their queued-response return values. Attach a
+typed handle using the task ID and the generated response codec:
+
+```typescript
+import { sliverpb } from "sliver-script"
+
+const queued = await client.pingBeacon(beaconId, 42)
+const task = client.beaconTask(queued.Response!.TaskID, sliverpb.Ping, { beaconId })
+const result = await task.wait({ timeoutSeconds: 120, signal: controller.signal })
+console.log(result.Nonce)
+const metadata = await task.status()
+```
+
+`wait()` uses events and polling, checks task identity and target errors, and
+supports repeated waits or attachment to a persisted ID. A handle created through
+`SliverClient.beaconTask` resolves the current channel after reconnect. Canceling
+a wait leaves the remote task queued; `task.cancel()` explicitly requests server
+cancellation. `InteractiveBeacon.beaconTask` binds the expected beacon ID, and
+the existing `lsTask()` handle also exposes `status()` and `cancel()`. Decoded task
+responses retain the existing 64 KiB limit.
+
 ### Stateful forwarding
 
 The client provides stateful TCP port-forward, reverse-port-forward, and SOCKS5
@@ -68,6 +107,12 @@ idempotent `close()`. Use `listPortForwards()` / `stopPortForward(id)` and
 inventory, including safe cleanup of legacy listeners whose target metadata is
 not trusted. Sliver's generic forwarding protocol currently supports TCP; the
 application traffic carried over it can be HTTP, RDP, or any other TCP protocol.
+`tunnelStreamState$` reports `idle`, `connecting`, `connected`, `failed`, and
+`stopped`, with a stream generation number. After a tunnel transport failure,
+existing tunnels fail and a new tunnel starts a fresh stream lazily. Old data is
+never replayed. `isConnected` continues to describe the control connection.
+Tunnel identities are retained to reject stale handles, bounded to 65,536 IDs
+per connection; reaching that limit requires disconnecting and reconnecting.
 These tunnels use full-close TCP semantics: a local FIN retires both directions,
 so protocols that send a delayed response only after the client half-closes are
 not supported. A SOCKS connection's `open` event means its Sliver lifecycle

e2e/README.md

@@ -19,6 +19,7 @@ keeping each area of client behavior independently identifiable in diagnostics.
 | `04-session-core` | Session inventory, direct and interactive ping, rename/restore, and environment read/set/unset |
 | `05-session-filesystem` | Working-directory changes, directory listing and globbing, upload/download ranges, recursive archives and grep, copy/move/timestamps/removal, and Unix permission/ownership calls |
 | `06-session-process-network` | Interfaces, live TCP/UDP socket inventory, mounts, basic/full process inventory, synchronous/background execution, child tracking and termination, Linux memfiles, and Windows token/privilege/service reads plus isolated HKCU registry CRUD |
+| `06-session-shell` | Native managed shell output and ordered writes, Unix PTY sizing and resize, Windows piped-shell resize rejection, natural exit, disconnect teardown, owned-child cleanup, and client reconnect |
 | `07-beacon-core` | Beacon metadata and rename, direct and interactive queued pings, derived task-result events, 10-second reconfiguration, task cancellation, fetch, and history/count reconciliation |
 | `08-beacon-filesystem` | Queued working-directory, mkdir, cd, upload/download, list, and recursive removal calls with completed-task verification |
 | `09-beacon-process-network` | Queued process, interface, socket, and environment calls plus synchronous/background execution, child tracking, termination, and task-history verification |
@@ -97,9 +98,11 @@ so diagnostic logs do not inherit unrelated CI secrets. Generated binary bytes
 are cleared after being written, and failure cleanup removes profiles, builds,
 listeners, sessions, beacons, processes, and the temporary root.
 
-The suite does not run interactive shell tunnels, screenshots, process dumps,
-service mutation, token mutation, registry-hive export, injection, or arbitrary
-payload execution. Forwarding coverage uses deterministic, bounded loopback TCP
+The suite does not run screenshots, process dumps, service mutation, token
+mutation, registry-hive export, injection, or arbitrary payload execution.
+Interactive shell coverage uses bounded native fixture commands and verifies
+cleanup of only the shell children it created. Forwarding coverage uses
+deterministic, bounded loopback TCP
 fixtures only. In particular, hosted CI does not install an RDP client, use RDP
 credentials, or connect to port 3389; RDP interoperability is validated in the
 separate Proxmox/GOAD range.

e2e/groups/06-session-shell.ts

@@ -0,0 +1,207 @@
+import assert from "node:assert/strict";
+import { access, readFile } from "node:fs/promises";
+import path from "node:path";
+import { setTimeout as delay } from "node:timers/promises";
+
+import type { ShellSessionHandle } from "../../lib";
+import type { E2ESuiteContext, LiveImplant } from "../context";
+
+export const name = "06-session-shell";
+const operationSeconds = 30;
+const outputLimitBytes = 128 * 1024;
+const waitMilliseconds = 20_000;
+
+interface OutputObservation {
+  text: string;
+  finished: boolean;
+  error?: unknown;
+  done: Promise<void>;
+}
+
+interface OwnedShell {
+  handle: ShellSessionHandle;
+  output: OutputObservation;
+  executable: string;
+}
+
+export async function run(context: E2ESuiteContext): Promise<void> {
+  const implant = context.session;
+  assert.ok(implant?.session?.ID, "live session fixture");
+  const sessionId = implant.session.ID;
+  const sentinel = path.join(implant.root, "outside-test-sentinel.txt");
+  assert.equal(await readFile(sentinel, "utf8"), "must-survive\n", "shell fixture sentinel");
+  const windows = context.environment.expectedOS === "windows";
+  const systemRoot = process.env.SystemRoot ?? process.env.SYSTEMROOT;
+  if (windows) assert.ok(systemRoot && path.win32.isAbsolute(systemRoot), "absolute Windows system root");
+  const shellPath = windows ? path.win32.join(systemRoot!, "System32", "cmd.exe") : "/bin/sh";
+  await access(shellPath); // Do not let Sliver silently select its fallback shell.
+  const prefix = `SSE2E_${process.pid}_`;
+  const markerCommand = (suffix: string): string => windows
+    ? `echo %SSE2E_PREFIX%${suffix}\r\n`
+    : `printf '%s%s\\n' '${prefix}' '${suffix}'\n`;
+  const owned: OwnedShell[] = [];
+  let failure: unknown;
+  try {
+    const first = await startOwnedShell(context, implant, shellPath, !windows);
+    owned.push(first);
+    await write(first.handle, windows
+      ? `@echo off\r\nset "SSE2E_PREFIX=${prefix}"\r\n`
+      : "PS1=''; PS2=''; /bin/stty -echo\n");
+    await write(first.handle, markerCommand("READY"));
+    await waitForOutput(first.output, `${prefix}READY\n`);
+
+    // Queue separate writes together to exercise the handle's ordering guarantee.
+    await bounded(Promise.all(["FIRST", "SECOND", "THIRD"].map((part) =>
+      write(first.handle, markerCommand(part)),
+    )), "ordered shell writes");
+    await waitForOutput(first.output, `${prefix}THIRD\n`);
+    const positions = ["FIRST", "SECOND", "THIRD"].map((part) => first.output.text.indexOf(`${prefix}${part}\n`));
+    assert.ok(positions[0]! >= 0 && positions[0]! < positions[1]! && positions[1]! < positions[2]!, "shell output follows write order");
+
+    if (windows) {
+      await assert.rejects(bounded(first.handle.resize(41, 101), "piped shell resize"), /requires a PTY/u, "piped Windows shell cannot resize");
+    } else {
+      await write(first.handle, `printf '%s%s' '${prefix}' 'INITIAL:'; /bin/stty size\n`);
+      await waitForOutput(first.output, `${prefix}INITIAL:24 80\n`);
+      await bounded(first.handle.resize(41, 101), "native PTY resize");
+      await write(first.handle, `printf '%s%s' '${prefix}' 'RESIZED:'; /bin/stty size\n`);
+      await waitForOutput(first.output, `${prefix}RESIZED:41 101\n`);
+    }
+
+    await write(first.handle, windows ? "exit\r\n" : "exit\n");
+    await bounded(first.output.done, "natural shell EOF");
+    assert.equal(first.output.error, undefined, "natural shell exit completes the output iterator");
+    await waitForShellExit(context, implant, first);
+    await assert.rejects(write(first.handle, "already closed"), /closed/u, "exited shell rejects writes");
+    await bounded(first.handle.close(), "exited shell close");
+    await bounded(first.handle.close(), "idempotent shell close");
+
+    const second = await startOwnedShell(context, implant, shellPath, !windows);
+    owned.push(second);
+    await write(second.handle, windows
+      ? `@echo off\r\nset "SSE2E_PREFIX=${prefix}"\r\n`
+      : "PS1=''; PS2=''; /bin/stty -echo\n");
+    await write(second.handle, markerCommand("DISCONNECT_READY"));
+    await waitForOutput(second.output, `${prefix}DISCONNECT_READY\n`);
+    await bounded(context.client.disconnect(), "client disconnect with an open shell");
+    await bounded(second.output.done, "disconnected shell output cleanup");
+    assert.ok(second.output.error instanceof Error, "disconnect fails the open output iterator");
+    assert.match(second.output.error.message, /Tunnel transport disconnected/u);
+    await assert.rejects(write(second.handle, "already disconnected"), /closed/u);
+    assert.equal(context.client.isConnected, false, "client disconnected after shell teardown");
+
+    await context.client.connect();
+    const ping = await context.client.pingSession(sessionId, 0x53e110, operationSeconds);
+    assert.equal(ping.Response?.Err ?? "", "", "session remains usable after shell reconnect");
+    assert.equal(ping.Nonce, 0x53e110);
+    await waitForShellExit(context, implant, second);
+  } catch (error) {
+    failure = error;
+  }
+
+  const cleanupErrors: unknown[] = [];
+  try {
+    if (!context.client.isConnected) await context.client.connect();
+    for (const shell of owned) {
+      try {
+        await bounded(shell.handle.close(), "owned shell cleanup");
+        const current = await ownedProcess(context, implant, shell.handle.pid);
+        if (current) {
+          assert.equal(current.Executable, shell.executable, "cleanup shell executable identity");
+          const terminated = await context.client.terminateSessionProcess(sessionId, shell.handle.pid, false, operationSeconds);
+          assert.equal(terminated.Response?.Err ?? "", "", "owned shell fallback termination");
+          assert.equal(terminated.Pid, shell.handle.pid, "only the returned shell PID is terminated");
+        }
+        await waitForShellExit(context, implant, shell);
+        await bounded(shell.output.done, "owned shell output cleanup");
+      } catch (error) {
+        cleanupErrors.push(error);
+      }
+    }
+    assert.equal(await readFile(sentinel, "utf8"), "must-survive\n", "shell fixture sentinel preserved");
+  } catch (error) {
+    cleanupErrors.push(error);
+  }
+  if (failure && cleanupErrors.length) throw new AggregateError([failure, ...cleanupErrors], "shell E2E and cleanup failed");
+  if (failure) throw failure;
+  if (cleanupErrors.length) throw new AggregateError(cleanupErrors, "shell E2E cleanup failed");
+}
+
+async function startOwnedShell(context: E2ESuiteContext, implant: LiveImplant, shellPath: string, pty: boolean): Promise<OwnedShell> {
+  const handle = await context.client.startShellSession(implant.session!.ID, {
+    path: shellPath, pty, rows: 24, cols: 80, outputBufferBytes: outputLimitBytes,
+  }, operationSeconds);
+  const output = observeOutput(handle);
+  try {
+    assert.ok(handle.pid > 1 && handle.pid !== implant.pid, "distinct owned shell child PID");
+    const process = await ownedProcess(context, implant, handle.pid);
+    assert.ok(process, "new shell is in the native process inventory");
+    return { handle, output, executable: process.Executable };
+  } catch (error) {
+    await bounded(handle.close(), "failed shell startup cleanup");
+    throw error;
+  }
+}
+
+function observeOutput(handle: ShellSessionHandle): OutputObservation {
+  const observed: OutputObservation = { text: "", finished: false, done: Promise.resolve() };
+  observed.done = (async () => {
+    let bytes = 0;
+    try {
+      for await (const chunk of handle.output) {
+        bytes += chunk.byteLength;
+        assert.ok(bytes <= outputLimitBytes, "bounded shell fixture output");
+        observed.text += Buffer.from(chunk).toString("utf8").replace(/\r/gu, "");
+      }
+    } catch (error) {
+      observed.error = error;
+    } finally {
+      observed.finished = true;
+    }
+  })();
+  return observed;
+}
+
+async function waitForOutput(output: OutputObservation, marker: string): Promise<void> {
+  const deadline = Date.now() + waitMilliseconds;
+  while (Date.now() < deadline) {
+    if (output.text.includes(marker)) return;
+    if (output.error) throw output.error;
+    assert.equal(output.finished, false, `shell closed before ${marker.trim()}`);
+    await delay(25);
+  }
+  throw new Error(`Timed out waiting for shell marker ${marker.trim()}; output=${JSON.stringify(output.text.slice(-2_000))}`);
+}
+
+async function ownedProcess(context: E2ESuiteContext, implant: LiveImplant, pid: number) {
+  assert.ok(pid > 1 && pid !== implant.pid, "safe shell child PID");
+  const listed = await context.client.psSession(implant.session!.ID, true, 5);
+  assert.equal(listed.Response?.Err ?? "", "", "shell process inventory");
+  const current = listed.Processes.find((process) => process.Pid === pid);
+  if (current) assert.equal(current.Ppid, implant.pid, "shell process belongs to this fixture implant");
+  return current;
+}
+
+async function waitForShellExit(context: E2ESuiteContext, implant: LiveImplant, shell: OwnedShell): Promise<void> {
+  const deadline = Date.now() + waitMilliseconds;
+  while (Date.now() < deadline) {
+    if (!await ownedProcess(context, implant, shell.handle.pid)) return;
+    await delay(100);
+  }
+  throw new Error(`Owned shell PID ${shell.handle.pid} did not exit`);
+}
+
+function write(handle: ShellSessionHandle, data: string): Promise<void> {
+  return bounded(handle.write(data), "shell input");
+}
+
+async function bounded<T>(operation: Promise<T>, label: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([operation, new Promise<never>((_resolve, reject) => {
+      timer = setTimeout(() => reject(new Error(`Timed out during ${label}`)), waitMilliseconds);
+    })]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}

e2e/groups/07-beacon-core.ts

@@ -79,14 +79,22 @@ export async function run(context: E2ESuiteContext): Promise<void> {
   }
 
   const interactiveNonce = nonce + 7;
-  const interactivePing = await context.runBeaconTask(
-    beacon.ID,
-    () => context.client.interactBeacon(beacon.ID).ping(interactiveNonce, rpcTimeoutSeconds),
-    sliverScript.sliverpb.Ping.decode,
-    "interactive beacon ping",
-  );
-  assert.equal(interactivePing.response.Nonce, interactiveNonce, "interactive beacon ping nonce");
-  assert.equal(interactivePing.response.Response?.Err ?? "", "", "interactive beacon ping response error");
+  const interactivePing = await context.client.interactBeacon(beacon.ID).ping(interactiveNonce, rpcTimeoutSeconds);
+  const interactiveMetadata = requireQueuedResponse(interactivePing.Response, beacon.ID, "interactive beacon ping");
+  const interactiveHandle = context.client.beaconTask(interactiveMetadata.TaskID, sliverScript.sliverpb.Ping, {
+    beaconId: beacon.ID,
+  });
+  const interactiveResult = await interactiveHandle.wait({ timeoutSeconds: 120 });
+  assert.equal(interactiveResult.Nonce, interactiveNonce, "interactive beacon ping nonce");
+  assert.equal(interactiveResult.Response?.Err ?? "", "", "interactive beacon ping response error");
+  const interactiveStatus = await interactiveHandle.status(rpcTimeoutSeconds);
+  assert.equal(interactiveStatus.ID, interactiveMetadata.TaskID, "interactive handle task ID");
+  assert.equal(interactiveStatus.BeaconID, beacon.ID, "interactive handle beacon ID");
+  assert.equal(interactiveStatus.State, "completed", "interactive handle completed state");
+  const reattachedResult = await context.client.beaconTask(
+    interactiveMetadata.TaskID, sliverScript.sliverpb.Ping, { beaconId: beacon.ID },
+  ).wait(rpcTimeoutSeconds);
+  assert.equal(reattachedResult.Nonce, interactiveNonce, "reattached completed beacon task result");
 
   const reconfigured = await context.runBeaconTask(
     beacon.ID,
@@ -113,7 +121,13 @@ export async function run(context: E2ESuiteContext): Promise<void> {
   // task remains deterministically pending on the loopback fixture.
   const canceledQueue = await context.client.pingBeacon(beacon.ID, nonce + 1, rpcTimeoutSeconds);
   const canceledMetadata = requireQueuedResponse(canceledQueue.Response, beacon.ID, "canceled beacon ping");
-  const canceled = await context.client.cancelBeaconTask(canceledMetadata.TaskID, rpcTimeoutSeconds);
+  const canceledHandle = context.client.beaconTask(canceledMetadata.TaskID, sliverScript.sliverpb.Ping, {
+    beaconId: beacon.ID,
+  });
+  const localWaitAbort = new AbortController();
+  localWaitAbort.abort();
+  await assert.rejects(canceledHandle.wait({ signal: localWaitAbort.signal }), sliverScript.OperationAbortedError);
+  const canceled = await canceledHandle.cancel(rpcTimeoutSeconds);
   assert.equal(canceled.ID, canceledMetadata.TaskID, "canceled beacon task ID");
   assert.equal(canceled.BeaconID, beacon.ID, "canceled beacon ID");
   assert.equal(canceled.State, "canceled", "canceled beacon task state");
@@ -128,7 +142,7 @@ export async function run(context: E2ESuiteContext): Promise<void> {
 
   const history: clientpb.BeaconTasks = await context.client.getBeaconTasks(beacon.ID, rpcTimeoutSeconds);
   assertTaskSummary(history, pingTaskId, "completed");
-  assertTaskSummary(history, interactivePing.task.ID, "completed");
+  assertTaskSummary(history, interactiveMetadata.TaskID, "completed");
   assertTaskSummary(history, reconfigured.task.ID, "completed");
   assertTaskSummary(history, canceled.ID, "canceled");