feat(workflow-executor): add JWT auth and secret validation to HTTP server

packages/workflow-executor/package.json

@@ -23,14 +23,17 @@
     "test": "jest"
   },
   "dependencies": {
-    "@forestadmin/ai-proxy": "1.6.1",
     "@forestadmin/agent-client": "1.4.13",
+    "@forestadmin/ai-proxy": "1.6.1",
     "@forestadmin/forestadmin-client": "1.37.17",
     "@koa/router": "^13.1.0",
+    "jsonwebtoken": "^9.0.3",
     "koa": "^3.0.1",
+    "koa-jwt": "^4.0.4",
     "zod": "4.3.6"
   },
   "devDependencies": {
+    "@types/jsonwebtoken": "^9.0.10",
     "@types/koa": "^2.13.5",
     "@types/koa__router": "^12.0.4",
     "supertest": "^7.1.3"

packages/workflow-executor/src/adapters/forest-server-workflow-port.ts

@@ -60,4 +60,12 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
   async getMcpServerConfigs(): Promise<McpConfiguration[]> {
     return ServerUtils.query<McpConfiguration[]>(this.options, 'get', ROUTES.mcpServerConfigs);
   }
+
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  async hasRunAccess(_runId: string, _userToken: string): Promise<boolean> {
+    // TODO: implement once GET /liana/v1/workflow-runs/:runId/access is available.
+    // When live: call ServerUtils.query with extra header 'forest-user-token': userToken
+    // to let the orchestrator verify ownership.
+    return true;
+  }
 }

packages/workflow-executor/src/errors.ts

@@ -199,6 +199,13 @@ export class McpToolInvocationError extends WorkflowExecutorError {
   }
 }
 
+export class ConfigurationError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'ConfigurationError';
+  }
+}
+
 export class RunNotFoundError extends Error {
   cause?: unknown;
 

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

@@ -1,18 +1,22 @@
 import type { Logger } from '../ports/logger-port';
 import type { RunStore } from '../ports/run-store';
+import type { WorkflowPort } from '../ports/workflow-port';
 import type Runner from '../runner';
 import type { Server } from 'http';
 
 import Router from '@koa/router';
 import http from 'http';
 import Koa from 'koa';
+import koaJwt from 'koa-jwt';
 
 import { RunNotFoundError } from '../errors';
 
 export interface ExecutorHttpServerOptions {
   port: number;
   runStore: RunStore;
   runner: Runner;
+  authSecret: string;
+  workflowPort: WorkflowPort;
   logger?: Logger;
 }
 
@@ -25,11 +29,20 @@ export default class ExecutorHttpServer {
     this.options = options;
     this.app = new Koa();
 
-    // Error middleware — catches all async handler errors and returns structured JSON
+    // Error middleware — catches all errors (including JWT 401) and returns structured JSON
     this.app.use(async (ctx, next) => {
       try {
         await next();
       } catch (err: unknown) {
+        const { status } = err as { status?: number };
+
+        if (status === 401) {
+          ctx.status = 401;
+          ctx.body = { error: 'Unauthorized' };
+
+          return;
+        }
+
         this.options.logger?.error('Unhandled HTTP error', {
           method: ctx.method,
           path: ctx.path,
@@ -41,7 +54,46 @@ export default class ExecutorHttpServer {
       }
     });
 
+    // JWT middleware — validates Bearer token using authSecret
+    // tokenKey: 'rawToken' exposes the raw token string on ctx.state.rawToken for downstream use
+    this.app.use(
+      koaJwt({ secret: options.authSecret, cookie: 'forest_session_token', tokenKey: 'rawToken' }),
+    );
+
     const router = new Router();
+
+    // Authorization middleware — verifies that the authenticated user owns the requested run.
+    // Applied to all /runs/:runId routes so future routes are automatically protected.
+    router.use('/runs/:runId', async (ctx, next) => {
+      // Raw token is always present here: koa-jwt already rejected the request if missing.
+      const userToken = ctx.state.rawToken as string;
+
+      try {
+        const allowed = await this.options.workflowPort.hasRunAccess(ctx.params.runId, userToken);
+
+        if (!allowed) {
+          ctx.status = 403;
+          ctx.body = { error: 'Forbidden' };
+
+          return;
+        }
+      } catch (err) {
+        this.options.logger?.error('Failed to check run access', {
+          runId: ctx.params.runId,
+          method: ctx.method,
+          path: ctx.path,
+          error: err instanceof Error ? err.message : String(err),
+          stack: err instanceof Error ? err.stack : undefined,
+        });
+        ctx.status = 503;
+        ctx.body = { error: 'Service unavailable' };
+
+        return;
+      }
+
+      await next();
+    });
+
     router.get('/runs/:runId', this.handleGetRun.bind(this));
     router.post('/runs/:runId/trigger', this.handleTrigger.bind(this));
 

packages/workflow-executor/src/index.ts

@@ -86,6 +86,7 @@ export {
   McpToolNotFoundError,
   McpToolInvocationError,
   AgentPortError,
+  ConfigurationError,
 } from './errors';
 export { default as BaseStepExecutor } from './executors/base-step-executor';
 export { default as ConditionStepExecutor } from './executors/condition-step-executor';
@@ -100,3 +101,4 @@ export { default as ExecutorHttpServer } from './http/executor-http-server';
 export type { ExecutorHttpServerOptions } from './http/executor-http-server';
 export { default as Runner } from './runner';
 export type { RunnerConfig } from './runner';
+export { default as validateSecrets } from './validate-secrets';

packages/workflow-executor/src/ports/workflow-port.ts

@@ -13,4 +13,5 @@ export interface WorkflowPort {
   updateStepExecution(runId: string, stepOutcome: StepOutcome): Promise<void>;
   getCollectionSchema(collectionName: string): Promise<CollectionSchema>;
   getMcpServerConfigs(): Promise<McpConfiguration[]>;
+  hasRunAccess(runId: string, userToken: string): Promise<boolean>;
 }

packages/workflow-executor/src/runner.ts

@@ -10,13 +10,16 @@ import ConsoleLogger from './adapters/console-logger';
 import { RunNotFoundError, causeMessage } from './errors';
 import StepExecutorFactory from './executors/step-executor-factory';
 import ExecutorHttpServer from './http/executor-http-server';
+import validateSecrets from './validate-secrets';
 
 export interface RunnerConfig {
   agentPort: AgentPort;
   workflowPort: WorkflowPort;
   runStore: RunStore;
   pollingIntervalMs: number;
   aiClient: AiClient;
+  envSecret: string;
+  authSecret: string;
   logger?: Logger;
   httpPort?: number;
 }
@@ -50,6 +53,9 @@ export default class Runner {
 
   async start(): Promise<void> {
     if (this.isRunning) return;
+
+    validateSecrets({ envSecret: this.config.envSecret, authSecret: this.config.authSecret });
+
     this.isRunning = true;
 
     try {
@@ -58,6 +64,9 @@ export default class Runner {
           port: this.config.httpPort,
           runStore: this.config.runStore,
           runner: this,
+          authSecret: this.config.authSecret,
+          workflowPort: this.config.workflowPort,
+          logger: this.logger,
         });
         await server.start();
         this.httpServer = server;

packages/workflow-executor/src/validate-secrets.ts

@@ -0,0 +1,13 @@
+import { ConfigurationError } from './errors';
+
+const ENV_SECRET_PATTERN = /^[0-9a-f]{64}$/;
+
+export default function validateSecrets(params: { envSecret: string; authSecret: string }): void {
+  if (!params.authSecret || typeof params.authSecret !== 'string') {
+    throw new ConfigurationError('authSecret must be a non-empty string');
+  }
+
+  if (!ENV_SECRET_PATTERN.test(params.envSecret)) {
+    throw new ConfigurationError('envSecret must be a 64-character hex string');
+  }
+}

packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts

@@ -123,6 +123,15 @@ describe('ForestServerWorkflowPort', () => {
     });
   });
 
+  describe('hasRunAccess', () => {
+    it('always returns true (stub until orchestrator endpoint is available)', async () => {
+      const result = await port.hasRunAccess('run-42', 'some-token');
+
+      expect(result).toBe(true);
+      expect(mockQuery).not.toHaveBeenCalled();
+    });
+  });
+
   describe('error propagation', () => {
     it('should propagate errors from ServerUtils.query', async () => {
       mockQuery.mockRejectedValue(new Error('Network error'));

packages/workflow-executor/test/executors/load-related-record-step-executor.test.ts

@@ -90,6 +90,7 @@ function makeMockWorkflowPort(
         ),
       ),
     getMcpServerConfigs: jest.fn().mockResolvedValue([]),
+    hasRunAccess: jest.fn().mockResolvedValue(true),
   };
 }
 

packages/workflow-executor/test/executors/mcp-task-step-executor.test.ts

@@ -59,6 +59,7 @@ function makeMockWorkflowPort(): WorkflowPort {
       actions: [],
     }),
     getMcpServerConfigs: jest.fn().mockResolvedValue([]),
+    hasRunAccess: jest.fn().mockResolvedValue(true),
   };
 }
 

packages/workflow-executor/test/executors/read-record-step-executor.test.ts

@@ -83,6 +83,7 @@ function makeMockWorkflowPort(
         ),
       ),
     getMcpServerConfigs: jest.fn().mockResolvedValue([]),
+    hasRunAccess: jest.fn().mockResolvedValue(true),
   };
 }
 

packages/workflow-executor/test/executors/trigger-record-action-step-executor.test.ts

@@ -78,6 +78,7 @@ function makeMockWorkflowPort(
         ),
       ),
     getMcpServerConfigs: jest.fn().mockResolvedValue([]),
+    hasRunAccess: jest.fn().mockResolvedValue(true),
   };
 }
 

packages/workflow-executor/test/executors/update-record-step-executor.test.ts

@@ -83,6 +83,7 @@ function makeMockWorkflowPort(
         ),
       ),
     getMcpServerConfigs: jest.fn().mockResolvedValue([]),
+    hasRunAccess: jest.fn().mockResolvedValue(true),
   };
 }