mirror of
https://github.com/KeygraphHQ/shannon.git
synced 2026-10-11 18:38:44 +02:00
feat: surface startup and preflight failures (#454)
* fix: surface pre-workflow worker failures instead of dying silently * feat: clearer, aggregated config rule validation errors * fix: preserve blank lines when printing startup errors * feat: hold start until preflight passes and surface its failure * fix: cleaner formatting for scan-start failure messages
This commit is contained in:
1 parent
d41d52f17d
commit
2786f9aa2d
6 files changed
+311
-138
No files matched your search
@@ -28,6 +28,7 @@
|
||||
* TEMPORAL_ADDRESS - Temporal server address (default: localhost:7233)
|
||||
*/
|
||||
|
||||
import { mkdirSync, writeFileSync } from 'node:fs';
|
||||
import path from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { Client, Connection, type WorkflowHandle, WorkflowNotFoundError } from '@temporalio/client';
|
||||
@@ -40,7 +41,8 @@ import { CAPELLA_FORMAT_VERSION, CAPELLA_PROMPT_SET_VERSION } from '../ai/sast/c
|
||||
import { summarizeOperationalMetrics } from '../audit/operational-summary.js';
|
||||
import { sanitizeHostname } from '../audit/utils.js';
|
||||
import { distributeConfig, parseConfig } from '../config-parser.js';
|
||||
import { deliverablesDir, resolveSessionJsonPath } from '../paths.js';
|
||||
import { deliverablesDir, INTERNAL_DIR, resolveSessionJsonPath, STARTUP_ERROR_FILENAME } from '../paths.js';
|
||||
import { PentestError } from '../services/error-handling.js';
|
||||
import { isProviderFailureCategory } from '../types/errors.js';
|
||||
import {
|
||||
ACCEPTED_CAPELLA_FAILURE_STAGES,
|
||||
@@ -511,31 +513,61 @@ interface OrchestrationConfig {
|
||||
exploit?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse the scan config into orchestration values, or throw on a broken config. Failing (rather
|
||||
* than falling back to defaults that quietly change scope) lets the caller persist parseConfig's
|
||||
* error for the CLI instead of running a misconfigured scan.
|
||||
*/
|
||||
async function loadOrchestrationConfig(configPath: string | undefined): Promise<OrchestrationConfig> {
|
||||
if (!configPath) return {};
|
||||
try {
|
||||
const config = await parseConfig(configPath);
|
||||
const distributed = distributeConfig(config);
|
||||
const codePathAvoids = distributed.avoid.filter((rule) => rule.type === 'code_path').map((rule) => rule.value);
|
||||
const codePathFocus = distributed.focus.filter((rule) => rule.type === 'code_path').map((rule) => rule.value);
|
||||
const config = await parseConfig(configPath);
|
||||
const distributed = distributeConfig(config);
|
||||
const codePathAvoids = distributed.avoid.filter((rule) => rule.type === 'code_path').map((rule) => rule.value);
|
||||
const codePathFocus = distributed.focus.filter((rule) => rule.type === 'code_path').map((rule) => rule.value);
|
||||
|
||||
return {
|
||||
...(distributed.agenticSast && {
|
||||
agenticSast: {
|
||||
codePathAvoids,
|
||||
codePathFocus,
|
||||
modelSpec: process.env.SHANNON_AI_MODEL?.trim() || DEFAULT_MODEL_SPEC,
|
||||
capellaFormatVersion: CAPELLA_FORMAT_VERSION,
|
||||
promptSetVersion: CAPELLA_PROMPT_SET_VERSION,
|
||||
},
|
||||
}),
|
||||
exploit: distributed.exploit,
|
||||
};
|
||||
} catch (error) {
|
||||
// A broken config must fail the run, not silently fall back to empty
|
||||
// defaults that quietly change scope (vuln classes, exploit, retries).
|
||||
console.error('Worker configuration could not be loaded. Reference code: CONFIG_VALIDATION_FAILED');
|
||||
process.exit(1);
|
||||
return {
|
||||
...(distributed.agenticSast && {
|
||||
agenticSast: {
|
||||
codePathAvoids,
|
||||
codePathFocus,
|
||||
modelSpec: process.env.SHANNON_AI_MODEL?.trim() || DEFAULT_MODEL_SPEC,
|
||||
capellaFormatVersion: CAPELLA_FORMAT_VERSION,
|
||||
promptSetVersion: CAPELLA_PROMPT_SET_VERSION,
|
||||
},
|
||||
}),
|
||||
exploit: distributed.exploit,
|
||||
};
|
||||
}
|
||||
|
||||
// === Startup Failure Persistence ===
|
||||
|
||||
/** Reason for a failure that happens before the workflow is created. */
|
||||
interface StartupErrorRecord {
|
||||
phase: string;
|
||||
code?: string;
|
||||
message: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist a pre-workflow failure to the bind-mounted workspace so the CLI can surface it. The
|
||||
* worker exits before the workflow exists, so Temporal has no record and `--rm` removes the
|
||||
* container; the file under INTERNAL_DIR outlives it on the host mount. `workspace` is the CLI's
|
||||
* `--workspace` name (the run directory); absent only when the worker is run off the CLI path.
|
||||
* Best-effort — a persist failure must not mask the original error.
|
||||
*/
|
||||
function persistStartupError(workspace: string | undefined, error: unknown, phase: string): void {
|
||||
if (!workspace) return;
|
||||
const record: StartupErrorRecord = {
|
||||
phase,
|
||||
...(error instanceof PentestError && error.code !== undefined && { code: error.code }),
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
};
|
||||
try {
|
||||
const dir = path.join('./workspaces', workspace, INTERNAL_DIR);
|
||||
mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(path.join(dir, STARTUP_ERROR_FILENAME), JSON.stringify(record, null, 2), 'utf8');
|
||||
} catch {
|
||||
// A broken bind mount must not compound the failure; the caller's console.error still fires.
|
||||
}
|
||||
}
|
||||
|
||||
@@ -655,24 +687,26 @@ async function waitForWorkflowResult(
|
||||
|
||||
// === Main Entry Point ===
|
||||
|
||||
async function run(): Promise<void> {
|
||||
// 1. Parse CLI args
|
||||
const args = parseCliArgs(process.argv.slice(2));
|
||||
|
||||
// 2. Connect to Temporal server
|
||||
const address = process.env.TEMPORAL_ADDRESS || 'localhost:7233';
|
||||
console.log(`Connecting to Temporal at ${address}...`);
|
||||
|
||||
const connection = await NativeConnection.connect({ address });
|
||||
const clientConnection = await Connection.connect({ address });
|
||||
const client = new Client({ connection: clientConnection });
|
||||
/** A scan whose workflow is durably submitted, with the handles run() needs to await it. */
|
||||
interface StartedScan {
|
||||
handle: WorkflowHandle<(input: PipelineInput) => Promise<PipelineState>>;
|
||||
workspace: WorkspaceResolution;
|
||||
worker: Worker;
|
||||
workerDone: Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run every step that precedes the durable creation of the workflow: config parsing, workspace
|
||||
* resolution, worker setup, and workflow submission. A failure anywhere here is a startup failure
|
||||
* — Temporal holds no record yet — so the reason is persisted for the CLI before it propagates.
|
||||
*/
|
||||
async function startScan(client: Client, connection: NativeConnection, args: CliArgs): Promise<StartedScan> {
|
||||
try {
|
||||
// 3. Validate orchestration and resume state before terminating any workflow.
|
||||
// 1. Validate orchestration and resume state before terminating any workflow.
|
||||
const orchestration = await loadOrchestrationConfig(args.configPath);
|
||||
const workspace = await resolveWorkspace(client, args, orchestration.exploit ?? true);
|
||||
|
||||
// 4. Bundle workflows and create the worker with the collision-checked activity registry.
|
||||
// 2. Bundle workflows and create the worker with the collision-checked activity registry.
|
||||
console.log('Preparing scan...');
|
||||
const workflowBundle = await bundleWorkflowCode({
|
||||
workflowsPath: path.join(__dirname, 'workflows.js'),
|
||||
@@ -695,13 +729,11 @@ async function run(): Promise<void> {
|
||||
maxConcurrentActivityTaskExecutions: 25,
|
||||
});
|
||||
|
||||
// 5. Build the fixed-scope pipeline input.
|
||||
// 3. Build the fixed-scope pipeline input and start worker polling in the background.
|
||||
const input = buildPipelineInput(args, workspace, orchestration);
|
||||
|
||||
// 6. Start worker polling in the background.
|
||||
const workerDone = worker.run();
|
||||
|
||||
// 7. Submit workflow to the same task queue.
|
||||
// 4. Submit workflow to the same task queue. Past this point the run exists in Temporal.
|
||||
const handle = await client.workflow.start<(input: PipelineInput) => Promise<PipelineState>>(
|
||||
'pentestPipelineWorkflow',
|
||||
{
|
||||
@@ -711,10 +743,33 @@ async function run(): Promise<void> {
|
||||
},
|
||||
);
|
||||
|
||||
// 8. Wait for workflow result.
|
||||
return { handle, workspace, worker, workerDone };
|
||||
} catch (startupError) {
|
||||
persistStartupError(args.resumeFromWorkspace, startupError, 'startup');
|
||||
throw startupError;
|
||||
}
|
||||
}
|
||||
|
||||
async function run(): Promise<void> {
|
||||
// 1. Parse CLI args
|
||||
const args = parseCliArgs(process.argv.slice(2));
|
||||
|
||||
// 2. Connect to Temporal server
|
||||
const address = process.env.TEMPORAL_ADDRESS || 'localhost:7233';
|
||||
console.log(`Connecting to Temporal at ${address}...`);
|
||||
|
||||
const connection = await NativeConnection.connect({ address });
|
||||
const clientConnection = await Connection.connect({ address });
|
||||
const client = new Client({ connection: clientConnection });
|
||||
|
||||
try {
|
||||
// 3. Start the scan: parse config, resolve the workspace, and submit the workflow.
|
||||
const { handle, workspace, worker, workerDone } = await startScan(client, connection, args);
|
||||
|
||||
// 4. Wait for workflow result.
|
||||
await waitForWorkflowResult(handle, workspace);
|
||||
|
||||
// 9. Shut down worker gracefully. Final customer copies are workflow-owned.
|
||||
// 5. Shut down worker gracefully. Final customer copies are workflow-owned.
|
||||
worker.shutdown();
|
||||
await workerDone;
|
||||
} finally {
|
||||
@@ -725,8 +780,10 @@ async function run(): Promise<void> {
|
||||
|
||||
const invokedPath = process.argv[1] ? path.resolve(process.argv[1]) : undefined;
|
||||
if (invokedPath === fileURLToPath(import.meta.url)) {
|
||||
run().catch(() => {
|
||||
console.error('Worker failed. Reference code: WORKER_FAILED');
|
||||
run().catch((error) => {
|
||||
// startScan persists pre-workflow failures for the CLI; this also logs them in the container.
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
console.error(`Worker failed: ${message}`);
|
||||
process.exit(1);
|
||||
});
|
||||
}
|
||||
@@ -131,10 +131,6 @@ export function formatWorkflowError(error: unknown, currentPhase: string | null,
|
||||
|
||||
const segments: string[] = [phaseContext];
|
||||
|
||||
if (unwrapped.type) {
|
||||
segments.push(unwrapped.type);
|
||||
}
|
||||
|
||||
segments.push(
|
||||
unwrapped.type === null
|
||||
? 'The scan could not be completed.'
|
||||
@@ -146,6 +142,7 @@ export function formatWorkflowError(error: unknown, currentPhase: string | null,
|
||||
if (hint) {
|
||||
segments.push(`Hint: ${hint}`);
|
||||
}
|
||||
segments.push(`Reference code: ${unwrapped.type}`);
|
||||
}
|
||||
|
||||
return segments.join('|');
|
||||
|
||||
Reference in new issue
Block a user