feat(plugin): AEGIS scan-pipe on all inbound Pilot messages

plugin/src/aegis-scan.ts

@@ -0,0 +1,79 @@
+// AEGIS scan-pipe — prompt-injection screening for inbound message text.
+//
+// Runs the external `aegis scan-pipe` tool, feeding the candidate text on
+// stdin. AEGIS signals a verdict via process exit code:
+//   0 → clean (allow)
+//   2 → flagged (block); the matched rule is printed on stdout
+//   * → any other status / spawn error / timeout → treated as inconclusive,
+//        and the caller fails OPEN (allows the message). A scanner outage must
+//        not silently swallow every inbound DM.
+//
+// This is intentionally async: the inbound pipeline is fully `await`-based and
+// runs one scan per message. A synchronous `spawnSync` here would block the
+// Node event loop for the whole subprocess lifetime on every message, starving
+// every other connection's I/O. We spawn non-blocking and await the result.
+
+import { execFile } from "node:child_process";
+
+/** Outcome of an AEGIS scan. */
+export type AegisScanResult = {
+  /** True when AEGIS flagged the text (exit code 2). */
+  blocked: boolean;
+  /** The matched rule (stdout) when blocked; empty otherwise. */
+  rule: string;
+};
+
+/** Pluggable scanner signature so the pipeline can be tested without execing. */
+export type AegisScan = (text: string) => Promise<AegisScanResult>;
+
+export type AegisScanOptions = {
+  /** Subprocess hard timeout in ms. Default 500. */
+  timeoutMs?: number;
+  /** Binary to invoke. Default "aegis". */
+  command?: string;
+};
+
+/**
+ * Default scanner: spawns `aegis scan-pipe` and writes `text` to its stdin.
+ *
+ * Non-blocking — resolves once the child exits (or the timeout kills it).
+ * Equivalent allow/block contract to the previous spawnSync implementation
+ * (exit 2 = block), but without parking the event loop.
+ */
+export function createAegisScan(opts: AegisScanOptions = {}): AegisScan {
+  const timeout = opts.timeoutMs ?? 500;
+  const command = opts.command ?? "aegis";
+
+  return (text: string): Promise<AegisScanResult> =>
+    new Promise<AegisScanResult>((resolve) => {
+      const child = execFile(
+        command,
+        ["scan-pipe"],
+        { timeout, encoding: "utf8", maxBuffer: 1024 * 1024 },
+        (err, stdout) => {
+          // `err.code` carries the exit status for a non-zero exit; on a
+          // timeout the child is killed and `err.killed` is set with no code.
+          const status =
+            err && typeof (err as { code?: unknown }).code === "number"
+              ? (err as { code: number }).code
+              : err
+                ? null // spawn error, timeout, or signal — inconclusive
+                : 0;
+
+          if (status === 2) {
+            resolve({ blocked: true, rule: (stdout ?? "").trim() });
+            return;
+          }
+          // Clean (0) or inconclusive (null / other) → fail open.
+          resolve({ blocked: false, rule: "" });
+        },
+      );
+
+      // Feed the candidate text and close stdin so AEGIS sees EOF.
+      child.stdin?.on("error", () => {
+        // A write race (e.g. child already exited) must not crash us; the
+        // exec callback above still resolves the promise.
+      });
+      child.stdin?.end(text);
+    });
+}

plugin/src/inbound.ts

@@ -12,6 +12,7 @@ import { join } from "node:path";
 import { tmpdir } from "node:os";
 
 import type { ResolvedPilotAccount } from "./config.js";
+import { createAegisScan, type AegisScan } from "./aegis-scan.js";
 import { decideAllowlist } from "./allowlist.js";
 import type { PeerAddressCache } from "./peer-address.js";
 import type { IncomingDatagram, Transport } from "./transport.js";
@@ -102,6 +103,13 @@ export type InboundDeps = {
    * common case but wrong for multi-network deployments.
    */
   peerAddressCache?: PeerAddressCache;
+  /**
+   * AEGIS prompt-injection scanner. Runs on every inbound text / media
+   * caption before dispatch; a `blocked` result drops the message. Async and
+   * non-blocking so one message's scan never parks the event loop. Defaults
+   * to the real `aegis scan-pipe` subprocess; tests inject a stub.
+   */
+  aegisScan?: AegisScan;
 };
 
 export class InboundPipeline {
@@ -114,12 +122,14 @@ export class InboundPipeline {
   private recentOrder: Array<{ id: string; ts: number }> = [];
   private readonly mediaDir: string;
   private readonly maxMediaBytes: number;
+  private readonly aegisScan: AegisScan;
 
   constructor(deps: InboundDeps) {
     this.deps = deps;
     this.recent = deps.recentIds ?? new Set();
     this.mediaDir = deps.mediaDir ?? join(tmpdir(), "claw-pilot-inbound");
     this.maxMediaBytes = deps.maxMediaBytes ?? 25 * 1024 * 1024;
+    this.aegisScan = deps.aegisScan ?? createAegisScan();
     try {
       mkdirSync(this.mediaDir, { recursive: true });
     } catch (e) {
@@ -272,6 +282,21 @@ export class InboundPipeline {
     // class of UI mystery.
     void this.sendAck(reassembled.id, peer);
 
+    // AEGIS scan: check message text for prompt injection before dispatching
+    // to the agent. Async + non-blocking so one slow scan can't stall the
+    // recv loop for every other peer. A scanner outage fails open.
+    if (reassembled.text) {
+      const scan = await this.aegisScan(reassembled.text);
+      if (scan.blocked) {
+        this.deps.logger.warn("pilot inbound: AEGIS blocked message", {
+          id: reassembled.id,
+          peer,
+          rule: scan.rule,
+        });
+        return;
+      }
+    }
+
     try {
       await this.deps.dispatch({
         accountId: this.deps.account.accountId,
@@ -343,6 +368,20 @@ export class InboundPipeline {
     // comment in handleDatagram above.
     void this.sendAck(out.id, peer);
 
+    // AEGIS scan: check caption text for injection before dispatch. Same
+    // async, fail-open contract as the text path above.
+    if (out.caption) {
+      const scan = await this.aegisScan(out.caption);
+      if (scan.blocked) {
+        this.deps.logger.warn("pilot inbound: AEGIS blocked media caption", {
+          id: out.id,
+          peer,
+          rule: scan.rule,
+        });
+        return;
+      }
+    }
+
     try {
       await this.deps.dispatch({
         accountId: this.deps.account.accountId,

plugin/src/lifecycle.ts

@@ -9,6 +9,7 @@ import type { OpenClawConfig } from "./openclaw-types.js";
 import type { PilotAccountConfig, ResolvedPilotAccount } from "./config.js";
 import { DEFAULT_ACCOUNT_ID, resolveAccount } from "./config.js";
 import { InboundPipeline, type InboundLogger } from "./inbound.js";
+import type { AegisScan } from "./aegis-scan.js";
 import { Outbox } from "./outbox.js";
 import { PeerAddressCache } from "./peer-address.js";
 import { getPilotRuntime } from "./runtime-api.js";
@@ -49,6 +50,13 @@ export type LifecycleDeps = {
    * Periodic drain interval (ms). Defaults to 30s. Tests can shorten.
    */
   outboxDrainIntervalMs?: number;
+  /**
+   * AEGIS prompt-injection scanner passed through to each account's
+   * InboundPipeline. Defaults (in the pipeline) to the real `aegis scan-pipe`
+   * subprocess; tests inject a stub so the integration path stays
+   * deterministic and never execs a real binary.
+   */
+  aegisScan?: AegisScan;
 };
 
 export class PilotLifecycle {
@@ -213,6 +221,7 @@ export class PilotLifecycle {
       account,
       dispatch,
       logger: this.deps.logger,
+      aegisScan: this.deps.aegisScan,
       // Reuse the same transport for ACKs — the sender Driver inside it is
       // the one with permission to send to the peer.
       ackTransport: transport,

plugin/tests/ack.test.ts

@@ -51,6 +51,7 @@ describe("inbound ACK", () => {
         dispatched.push(m);
       },
       logger: silentLogger(),
+      aegisScan: async () => ({ blocked: false, rule: "" }),
       ackTransport: transport,
     });
     pipeline.attach(transport);
@@ -85,6 +86,7 @@ describe("inbound ACK", () => {
         dispatched.push(m);
       },
       logger: silentLogger(),
+      aegisScan: async () => ({ blocked: false, rule: "" }),
       ackTransport: transport,
     });
     pipeline.attach(transport);
@@ -125,6 +127,7 @@ describe("inbound ACK", () => {
         /* noop */
       },
       logger: silentLogger(),
+      aegisScan: async () => ({ blocked: false, rule: "" }),
     });
     pipeline.attach(transport);
 
@@ -148,6 +151,7 @@ describe("inbound ACK", () => {
         /* noop */
       },
       logger: silentLogger(),
+      aegisScan: async () => ({ blocked: false, rule: "" }),
       ackTransport: transport,
     });
     pipeline.attach(transport);