feat(workflow-executor): add OAuth credential store + deposit endpoint (PRD-367 PR1)

packages/_example/.env.example

@@ -21,6 +21,8 @@ FOREST_AUTH_SECRET=
 EXECUTOR_AGENT_URL=http://localhost:3351
 WORKFLOW_EXECUTOR_URL=http://localhost:3400
 EXECUTOR_DATABASE_URL=postgresql://executor:password@localhost:5459/workflow_executor
+# At-rest encryption key for OAuth MCP credentials (openssl rand -hex 32; same value across instances)
+FOREST_EXECUTOR_ENCRYPTION_KEY=
 # when start:with-executor:multiple-instances command
 # EXECUTOR_AGENT_URL=http://host.docker.internal:3351
 # WORKFLOW_EXECUTOR_URL=http://localhost:3400

packages/workflow-executor/CLAUDE.md

@@ -59,6 +59,7 @@ src/
 ├── stores/                 # RunStore implementations
 │   ├── in-memory-store.ts  # InMemoryStore — Map-based, for tests
 │   ├── database-store.ts   # DatabaseStore — Sequelize + umzug migrations
+│   ├── mcp-oauth-credentials-store.ts  # McpOAuthCredentialsStore — ai_mcp_oauth_credentials (002 migration)
 │   └── build-run-store.ts  # Factory functions: buildDatabaseRunStore, buildInMemoryRunStore
 ├── adapters/               # Port implementations
 │   ├── agent-client-agent-port.ts      # AgentPort via @forestadmin/agent-client
@@ -72,7 +73,10 @@ src/
 │   ├── load-related-record-step-executor.ts  # AI-powered relation loading step (with confirmation flow)
 │   └── guidance-step-executor.ts  # Manual guidance step (saves user input, no AI)
 ├── http/                   # HTTP server (optional, for frontend data access)
-│   └── executor-http-server.ts  # Koa server: GET /runs/:runId, POST /runs/:runId/trigger
+│   ├── executor-http-server.ts  # Koa server: GET /runs/:runId, POST /runs/:runId/trigger, POST+DELETE /mcp-oauth-credentials
+│   └── mcp-oauth-credentials.ts  # Deposit-body zod schema (.strict()) + buildMcpOAuthCredentialInput mapper
+├── crypto/                 # At-rest encryption
+│   └── credential-encryption.ts  # CredentialEncryption — HKDF (FOREST_EXECUTOR_ENCRYPTION_KEY) + AES-GCM, lazy key, fail-closed
 └── index.ts                # Barrel exports
 ```
 

packages/workflow-executor/README.md

@@ -26,6 +26,7 @@ npm install -g @forestadmin/workflow-executor
 | `FOREST_SERVER_URL` | — | `https://api.forestadmin.com` | Orchestrator URL |
 | `POLLING_INTERVAL_MS` | — | `5000` | Poll cadence for pending steps |
 | `STOP_TIMEOUT_MS` | — | `30000` | Graceful shutdown deadline |
+| `FOREST_EXECUTOR_ENCRYPTION_KEY` | —† | — | At-rest key for OAuth MCP credentials (HKDF-derived, AES-256-GCM). Generate with `openssl rand -hex 32`; set the **same** value across all executor instances. (†Read lazily — required only once OAuth-protected MCP servers are used.) |
 
 Optional AI configuration (all-or-nothing — falls back to server AI if any is missing):
 

packages/workflow-executor/example/.env.example

@@ -2,6 +2,11 @@
 FOREST_ENV_SECRET=
 FOREST_AUTH_SECRET=
 
+# At-rest encryption key for OAuth MCP credentials (HKDF-derived, AES-256-GCM).
+# Generate: openssl rand -hex 32. Set the SAME value across all executor instances.
+# Read lazily — only required once OAuth-protected MCP servers are used.
+FOREST_EXECUTOR_ENCRYPTION_KEY=
+
 # Your locally running Forest Admin agent
 AGENT_URL=http://localhost:3351
 

packages/workflow-executor/src/build-workflow-executor.ts

@@ -13,6 +13,7 @@ import ConsoleLogger from './adapters/console-logger';
 import ForestServerWorkflowPort from './adapters/forest-server-workflow-port';
 import ForestadminClientActivityLogPortFactory from './adapters/forestadmin-client-activity-log-port-factory';
 import ServerAiAdapter from './adapters/server-ai-adapter';
+import CredentialEncryption from './crypto/credential-encryption';
 import {
   DEFAULT_AI_INVOKE_TIMEOUT_MS,
   DEFAULT_FOREST_SERVER_URL,
@@ -25,6 +26,7 @@ import Runner from './runner';
 import SchemaCache from './schema-cache';
 import DatabaseStore from './stores/database-store';
 import InMemoryStore from './stores/in-memory-store';
+import McpOAuthCredentialsStore from './stores/mcp-oauth-credentials-store';
 
 const FORCE_EXIT_DELAY_MS = 5000;
 
@@ -237,6 +239,8 @@ export function buildDatabaseExecutor(options: DatabaseExecutorOptions): Workflo
     authSecret: options.authSecret,
     workflowPort: deps.workflowPort,
     logger: deps.logger,
+    mcpOAuthCredentialsStore: new McpOAuthCredentialsStore({ sequelize }),
+    credentialEncryption: new CredentialEncryption(),
   });
 
   return createWorkflowExecutor(runner, server, deps.logger);

packages/workflow-executor/src/crypto/credential-encryption.ts

@@ -0,0 +1,89 @@
+import { createCipheriv, createDecipheriv, hkdfSync, randomFillSync } from 'crypto';
+
+import { ExecutorEncryptionKeyMissingError } from '../errors';
+
+const ENV_KEY = 'FOREST_EXECUTOR_ENCRYPTION_KEY';
+// Fixed context label bound into the HKDF derivation — domain-separates this key from any other
+// use of the same secret. Changing it would make every existing row undecryptable.
+const HKDF_INFO = 'forest-executor:mcp-oauth-credentials';
+const HKDF_DIGEST = 'sha256';
+const KEY_BYTES = 32; // AES-256
+const IV_BYTES = 12; // GCM standard nonce length
+const AUTH_TAG_BYTES = 16;
+const ALGORITHM = 'aes-256-gcm';
+const CURRENT_ENC_KEY_VERSION = 1;
+
+export interface EncryptedValue {
+  // Packed layout: iv | authTag | ciphertext — stored as a single BLOB column.
+  ciphertext: Buffer;
+  encKeyVersion: number;
+}
+
+// Concatenate byte arrays without going through Buffer.concat — keeps everything in the concrete
+// Uint8Array<ArrayBuffer> domain the Node crypto types expect.
+function concatBytes(parts: Uint8Array[]): Uint8Array {
+  const total = parts.reduce((length, part) => length + part.length, 0);
+  const out = new Uint8Array(total);
+  let offset = 0;
+
+  for (const part of parts) {
+    out.set(part, offset);
+    offset += part.length;
+  }
+
+  return out;
+}
+
+// At-rest encryption for OAuth credentials. The HKDF key (from FOREST_EXECUTOR_ENCRYPTION_KEY) is
+// read lazily — an OAuth-less executor boots without it — and fails closed: a missing key throws
+// rather than persisting or returning an unprotected value.
+export default class CredentialEncryption {
+  private readonly encKeyVersion: number;
+
+  constructor(encKeyVersion: number = CURRENT_ENC_KEY_VERSION) {
+    this.encKeyVersion = encKeyVersion;
+  }
+
+  encrypt(plaintext: string): EncryptedValue {
+    const iv = randomFillSync(new Uint8Array(IV_BYTES));
+    const cipher = createCipheriv(ALGORITHM, this.deriveKey(), iv);
+    const encrypted = concatBytes([
+      new Uint8Array(cipher.update(plaintext, 'utf8')),
+      new Uint8Array(cipher.final()),
+    ]);
+    const authTag = new Uint8Array(cipher.getAuthTag());
+
+    return {
+      ciphertext: Buffer.from(concatBytes([iv, authTag, encrypted])),
+      encKeyVersion: this.encKeyVersion,
+    };
+  }
+
+  decrypt(value: Buffer): string {
+    const bytes = new Uint8Array(value);
+    const iv = bytes.subarray(0, IV_BYTES);
+    const authTag = bytes.subarray(IV_BYTES, IV_BYTES + AUTH_TAG_BYTES);
+    const encrypted = bytes.subarray(IV_BYTES + AUTH_TAG_BYTES);
+
+    const decipher = createDecipheriv(ALGORITHM, this.deriveKey(), iv);
+    decipher.setAuthTag(authTag);
+
+    const decrypted = concatBytes([
+      new Uint8Array(decipher.update(encrypted)),
+      new Uint8Array(decipher.final()),
+    ]);
+
+    return Buffer.from(decrypted).toString('utf8');
+  }
+
+  private deriveKey(): Uint8Array {
+    const secret = process.env[ENV_KEY];
+
+    if (!secret) throw new ExecutorEncryptionKeyMissingError();
+
+    // Empty salt is intentional: the fixed HKDF_INFO label gives domain separation and the
+    // single high-entropy secret needs no salt. Wrap hkdfSync's ArrayBuffer as a concrete
+    // Uint8Array<ArrayBuffer> to satisfy CipherKey (Buffer's ArrayBufferLike backing does not).
+    return new Uint8Array(hkdfSync(HKDF_DIGEST, secret, new Uint8Array(0), HKDF_INFO, KEY_BYTES));
+  }
+}

packages/workflow-executor/src/errors.ts

@@ -312,6 +312,17 @@ export class ConfigurationError extends Error {
   }
 }
 
+// Boundary error — the deposit endpoint maps it to a typed HTTP response so the frontend can tell
+// an operator to provision the key, not a generic or re-consent failure.
+export class ExecutorEncryptionKeyMissingError extends Error {
+  readonly code = 'executor_encryption_key_missing';
+
+  constructor() {
+    super('FOREST_EXECUTOR_ENCRYPTION_KEY is not set');
+    this.name = 'ExecutorEncryptionKeyMissingError';
+  }
+}
+
 export class RunNotFoundError extends Error {
   cause?: unknown;
 

packages/workflow-executor/src/http/executor-http-server.ts

@@ -1,6 +1,8 @@
+import type CredentialEncryption from '../crypto/credential-encryption';
 import type { Logger } from '../ports/logger-port';
 import type { WorkflowPort } from '../ports/workflow-port';
 import type Runner from '../runner';
+import type McpOAuthCredentialsStore from '../stores/mcp-oauth-credentials-store';
 import type { StepUser } from '../types/execution-context';
 import type { Server } from 'http';
 
@@ -10,9 +12,14 @@ import http from 'http';
 import Koa from 'koa';
 import koaJwt from 'koa-jwt';
 
+import {
+  buildMcpOAuthCredentialInput,
+  depositCredentialsBodySchema,
+} from './mcp-oauth-credentials';
 import serializeStepForWire from './step-serializer';
 import ConsoleLogger from '../adapters/console-logger';
 import {
+  ExecutorEncryptionKeyMissingError,
   RunAlreadyInFlightError,
   RunNotFoundError,
   UserMismatchError,
@@ -26,17 +33,21 @@ export interface ExecutorHttpServerOptions {
   authSecret: string;
   workflowPort: WorkflowPort;
   logger?: Logger;
+  mcpOAuthCredentialsStore?: McpOAuthCredentialsStore;
+  credentialEncryption?: CredentialEncryption;
 }
 
 export default class ExecutorHttpServer {
   private readonly app: Koa;
   private readonly options: ExecutorHttpServerOptions;
   private readonly logger: Logger;
+  private readonly mcpOAuthCredentialsStore?: McpOAuthCredentialsStore;
   private server: Server | null = null;
 
   constructor(options: ExecutorHttpServerOptions) {
     this.options = options;
     this.logger = options.logger ?? new ConsoleLogger();
+    this.mcpOAuthCredentialsStore = options.mcpOAuthCredentialsStore;
     this.app = new Koa();
 
     // Error middleware — catches all errors (including JWT 401) and returns structured JSON
@@ -96,11 +107,28 @@ export default class ExecutorHttpServer {
     );
     router.post('/runs/:runId/trigger', this.handleTrigger.bind(this));
 
+    // Registered only when both dependencies are wired (a real executor with a database) — keeps
+    // the OAuth deposit surface absent (and dormant) on in-memory / OAuth-less deployments.
+    const { mcpOAuthCredentialsStore: credentialsStore, credentialEncryption } = this.options;
+
+    if (credentialsStore && credentialEncryption) {
+      router.post('/mcp-oauth-credentials', ctx =>
+        this.handleDepositCredentials(ctx, credentialsStore, credentialEncryption),
+      );
+      router.delete('/mcp-oauth-credentials/:mcpServerId', ctx =>
+        this.handleDeleteCredentials(ctx, credentialsStore),
+      );
+    }
+
     this.app.use(router.routes());
     this.app.use(router.allowedMethods());
   }
 
   async start(): Promise<void> {
+    if (this.mcpOAuthCredentialsStore) {
+      await this.mcpOAuthCredentialsStore.init(this.logger);
+    }
+
     return new Promise((resolve, reject) => {
       this.server = http.createServer(this.app.callback());
       this.server.once('error', reject);
@@ -167,15 +195,8 @@ export default class ExecutorHttpServer {
 
   private async handleTrigger(ctx: Koa.Context): Promise<void> {
     const { runId } = ctx.params;
-    const rawId = (ctx.state.user as { id?: unknown })?.id;
-    const bearerUserId = typeof rawId === 'number' ? rawId : Number(rawId);
-
-    if (!Number.isFinite(bearerUserId)) {
-      ctx.status = 400;
-      ctx.body = { error: 'Missing or invalid user id in token' };
-
-      return;
-    }
+    const bearerUserId = this.requireBearerUserId(ctx);
+    if (bearerUserId === null) return;
 
     const pendingData = (ctx.request.body as { pendingData?: unknown })?.pendingData;
 
@@ -226,4 +247,70 @@ export default class ExecutorHttpServer {
     ctx.status = 200;
     ctx.body = { triggered: true };
   }
+
+  // Resolves the authenticated user id from the validated JWT. Writes a 400 and returns null when
+  // the token carries no usable id, so the three deposit/trigger handlers share one guard rather
+  // than repeating it. A Forest user id is always a positive integer — 0, negatives, and non-numeric
+  // ids are rejected here rather than reaching the store.
+  private requireBearerUserId(ctx: Koa.Context): number | null {
+    const rawId = (ctx.state.user as { id?: unknown })?.id;
+    const userId = typeof rawId === 'number' ? rawId : Number(rawId);
+
+    if (!Number.isInteger(userId) || userId <= 0) {
+      ctx.status = 400;
+      ctx.body = { error: 'Missing or invalid user id in token' };
+
+      return null;
+    }
+
+    return userId;
+  }
+
+  private async handleDepositCredentials(
+    ctx: Koa.Context,
+    store: McpOAuthCredentialsStore,
+    encryption: CredentialEncryption,
+  ): Promise<void> {
+    const userId = this.requireBearerUserId(ctx);
+    if (userId === null) return;
+
+    const parsed = depositCredentialsBodySchema.safeParse(ctx.request.body ?? {});
+
+    if (!parsed.success) {
+      const details = parsed.error.issues
+        .map(issue => `${issue.path.join('.') || 'body'}: ${issue.message}`)
+        .join('; ');
+      ctx.status = 400;
+      ctx.body = { error: `Invalid request body — ${details}` };
+
+      return;
+    }
+
+    try {
+      await store.upsert(buildMcpOAuthCredentialInput({ body: parsed.data, userId, encryption }));
+    } catch (err) {
+      if (err instanceof ExecutorEncryptionKeyMissingError) {
+        ctx.status = 503;
+        ctx.body = { code: err.code };
+
+        return;
+      }
+
+      throw err;
+    }
+
+    ctx.status = 200;
+    ctx.body = { stored: true };
+  }
+
+  private async handleDeleteCredentials(
+    ctx: Koa.Context,
+    store: McpOAuthCredentialsStore,
+  ): Promise<void> {
+    const userId = this.requireBearerUserId(ctx);
+    if (userId === null) return;
+
+    await store.delete(userId, ctx.params.mcpServerId);
+    ctx.status = 204;
+  }
 }