Component Development

worker/src/components/
├── security/        # Security tools (dnsx, subfinder, nuclei)
├── core/            # Core utilities (http-request, file-loader)
├── ai/              # AI components (llm, agents)
├── notification/    # Notifications (slack, email)
└── manual-action/   # Human-in-the-loop (approvals, forms)

<namespace>.<tool>.<action>

Examples:
  shipsec.dnsx.run       # Security tool
  core.http.request      # Core utility  
  ai.llm.generate        # AI component
  notification.slack     # Notification

import { z } from 'zod';
import { 
  componentRegistry, 
  defineComponent,
  inputs,
  outputs,
  port,
} from '@shipsec/component-sdk';

const definition = defineComponent({
  id: 'core.http.request',
  label: 'HTTP Request',
  category: 'transform',
  runner: { kind: 'inline' },
  
  inputs: inputs({
    url: port(z.string().url(), {
      label: 'URL',
      description: 'The URL to fetch',
    }),
    method: port(z.enum(['GET', 'POST']).default('GET'), {
      label: 'Method',
      description: 'HTTP verb to use',
    }),
  }),
  
  outputs: outputs({
    status: port(z.number(), { label: 'Status Code' }),
    body: port(z.string(), { label: 'Response Body' }),
  }),

  async execute({ inputs }, context) {
    context.logger.info(`Fetching ${inputs.url}`);
    const response = await fetch(inputs.url, { method: inputs.method });
    return { status: response.status, body: await response.text() };
  }
});

export default definition;

import { z } from 'zod';
import { 
  defineComponent,
  inputs,
  outputs,
  port,
  runComponentWithRunner,
  DockerRunnerConfig
} from '@shipsec/component-sdk';

const definition = defineComponent({
  id: 'shipsec.tool.scan',
  label: 'Tool Scanner',
  category: 'security',
  runner: {
    kind: 'docker',
    image: 'tool:latest',
    entrypoint: 'sh',
    command: ['-c', 'tool "$@"', '--'],
    network: 'bridge'
  },
  
  inputs: inputs({
    target: port(z.string(), { label: 'Target Host' }),
  }),
  
  outputs: outputs({
    results: port(z.any(), { label: 'Scan Results' }),
  }),

  async execute({ inputs }, context) {
    const args = ['-json', '-target', inputs.target];
    
    const runnerConfig: DockerRunnerConfig = {
      ...this.runner,
      command: [...(this.runner.command ?? []), ...args],
    };

    const rawOutput = await runComponentWithRunner(
      runnerConfig,
      async (stdout) => {
         // Logic to parse tool stdout
         return JSON.parse(stdout);
      },
      inputs,
      context
    );
    
    return { results: rawOutput };
  }
});

export default definition;

async execute({ inputs, params }, context) {
  // Logging (shows in UI timeline)
  context.logger.info('Starting scan...');
  context.logger.warn('Rate limit approaching');
  context.logger.error('Failed to connect');
  
  // Progress events (shows in UI)
  context.emitProgress('Processing 50 targets...');
  
  // Secrets (encrypted, from secret manager)
  const apiKey = await context.secrets?.get('API_KEY');
  
  // File downloads (from MinIO)
  const file = await context.storage?.downloadFile(inputs.fileId);
  
  // Artifact uploads (saved to MinIO, shown in UI)
  await context.artifacts?.upload({
    name: 'report.json',
    content: Buffer.from(JSON.stringify(results)),
    mimeType: 'application/json',
  });
  
  // Run metadata
  const { runId, componentRef } = context;
}

inputs: inputs({
  ipAddress: port(z.string().ip(), { 
    label: 'IP Address',
    description: 'The IP address to scan',
    valuePriority: 'connection-first', // default: connection-first
  }),
})

parameters: parameters({
  threads: param(z.number().default(10), {
    label: 'Threads',
    editor: 'number',
    min: 1,
    max: 100,
    description: 'Number of concurrent threads to use.'
  }),
  mode: param(z.enum(['fast', 'thorough']).default('fast'), {
    label: 'Scan Mode',
    editor: 'select',
    options: [
      { label: 'Fast Scan', value: 'fast' },
      { label: 'Thorough Scan', value: 'thorough' },
    ],
  }),
})

parameters: parameters({
  useProxy: param(z.boolean().default(false), { editor: 'boolean', label: 'Use Proxy' }),
  proxyUrl: param(z.string().optional(), { 
    editor: 'text', 
    label: 'Proxy URL',
    visibleWhen: { useProxy: true } 
  }),
})

port(z.string(), {
  label: 'API Key',
  connectionType: { kind: 'primitive', name: 'secret' }
})

// Entry point configuration
const runtimeInputs = [
  { id: 'apiKey', label: 'API Key', type: 'secret', required: true },
  { id: 'target', label: 'Target URL', type: 'text', required: true },
];

import { inputs, port } from '@shipsec/component-sdk';

export default defineComponent({
  // ...
  parameters: parameters({
    variables: param(z.array(VariableSchema), { ... }),
  }),

  resolvePorts(params) {
    const dynamicInputShape: Record<string, any> = {};
    
    for (const v of params.variables) {
      dynamicInputShape[v.name] = port(z.string(), { 
        label: v.label || v.name 
      });
    }
    
    return {
      inputs: inputs(dynamicInputShape),
    };
  }
});

const definition: ComponentDefinition<Input, Output> = {
  id: 'shipsec.api.call',
  // ... other fields ...
  
  retryPolicy: {
    maxAttempts: 5,              // Max retries (0 = unlimited, 1 = no retry)
    initialIntervalSeconds: 2,   // Initial delay
    maximumIntervalSeconds: 120, // Max delay
    backoffCoefficient: 2.0,     // Exponential backoff
    nonRetryableErrorTypes: [    // Errors that should NOT retry
      'AuthenticationError',
      'ValidationError',
    ],
  },
};

import { 
  NetworkError,        // Retryable - network issues
  RateLimitError,      // Retryable - with delay
  ServiceError,        // Retryable - 5xx errors
  AuthenticationError, // Non-retryable - bad credentials
  ValidationError,     // Non-retryable - bad input
  NotFoundError,       // Non-retryable - resource missing
  fromHttpResponse,    // Convert HTTP response to error
  wrapError,           // Wrap unknown errors
} from '@shipsec/component-sdk';

async execute({ inputs }, context) {
  try {
    const response = await fetch(inputs.url);
    
    if (!response.ok) {
      throw fromHttpResponse(response, await response.text());
    }
    
    return await response.json();
  } catch (error) {
    throw wrapError(error, 'Failed to call API');
  }
}

outputs: outputs({
  // ... other outputs ...

  results: port(z.array(z.record(z.string(), z.unknown())), {
    label: 'Results',
    description:
      'Analytics-ready findings array. Each item includes scanner name and asset key. Connect to Analytics Sink.',
    connectionType: { kind: 'list', element: { kind: 'primitive', name: 'json' } },
  }),
}),

import { generateFindingHash } from '@shipsec/component-sdk';

// Usage
const hash = generateFindingHash(finding.templateId, finding.host, finding.matchedAt);

import { generateFindingHash } from '@shipsec/component-sdk';

async execute({ inputs, params }, context) {
  // ... run scanner and get findings ...

  // Build analytics-ready results with scanner metadata
  const results: Record<string, unknown>[] = findings.map((finding) => ({
    ...finding,                           // Spread all finding fields
    scanner: 'my-scanner',               // Scanner identifier
    asset_key: finding.host ?? inputs.target,  // Primary asset
    finding_hash: generateFindingHash(   // Stable deduplication hash
      finding.ruleId,
      finding.host,
      finding.matchedAt
    ),
  }));

  return {
    findings,      // Original findings array
    results,       // Analytics-ready array for Analytics Sink
    rawOutput,     // Raw output for debugging
  };
}

{
  "check_id": "DB_RLS_DISABLED",
  "severity": "CRITICAL",
  "title": "RLS Disabled",
  "metadata": "{\"table\":\"users\"}",
  "scanner": "supabase-scanner",
  "asset_key": "abc123xyz",
  "finding_hash": "a1b2c3d4e5f67890",
  "shipsec": {
    "organization_id": "org_123",
    "run_id": "run_abc123",
    "workflow_id": "wf_xyz789",
    "workflow_name": "Supabase Security Audit",
    "component_id": "core.analytics.sink",
    "node_ref": "analytics-sink-1"
  },
  "@timestamp": "2024-01-21T10:30:00Z"
}

# Find all findings for an asset
asset_key: "api.example.com"

# Find new findings (first seen today)
finding_hash: X AND @timestamp: [now-1d TO now] AND NOT (finding_hash: X AND @timestamp: [* TO now-1d])

# All findings from a specific workflow run
shipsec.run_id: "run_abc123"

# Aggregate findings by scanner
scanner: * | stats count() by scanner

# Track recurring findings
finding_hash: "a1b2c3d4" | sort @timestamp

// ✅ CORRECT - Shell wrapper pattern
runner: {
  kind: 'docker',
  image: 'tool:latest',
  entrypoint: 'sh',                      // Shell wrapper
  command: ['-c', 'tool "$@"', '--'],    // Wraps CLI execution
  network: 'bridge',
}

// ❌ WRONG - Direct binary execution
runner: {
  kind: 'docker',
  image: 'tool:latest',
  entrypoint: 'tool',                    // No shell wrapper - will hang
  command: ['-read-stdin', '-output'],
}

Does your Docker image have a shell (/bin/sh)?
├─ YES → Use Shell Wrapper Pattern
│         entrypoint: 'sh', command: ['-c', 'tool "$@"', '--']
│
└─ NO (Distroless) → Use Default Entrypoint Pattern
          Omit entrypoint, pass args via command: []
          The image's built-in ENTRYPOINT handles execution.

// ✅ CORRECT - Distroless image (no shell available)
runner: {
  kind: 'docker',
  image: 'ghcr.io/shipsecai/subfinder:latest',
  // No entrypoint — uses image default (/usr/local/bin/subfinder)
  command: [],
  network: 'bridge',
}

// ❌ WRONG - Distroless image with shell wrapper (exit code 127)
runner: {
  kind: 'docker',
  image: 'ghcr.io/shipsecai/subfinder:latest',
  entrypoint: 'sh',  // sh does not exist in distroless images!
  command: ['-c', 'subfinder "$@"', '--'],
}

const runnerConfig: DockerRunnerConfig = {
  ...baseRunner,
  command: [...(baseRunner.command ?? []), ...toolArgs],
};

import { IsolatedContainerVolume } from '../../utils/isolated-volume';

async execute({ inputs }, context) {
  const tenantId = (context as any).tenantId ?? 'default-tenant';
  const volume = new IsolatedContainerVolume(tenantId, context.runId);

  try {
    await volume.initialize({
      'targets.txt': inputs.targets.join('\n')
    });

    const runnerConfig: DockerRunnerConfig = {
      ...this.runner,
      command: [...(this.runner.command ?? []), '-l', '/inputs/targets.txt'],
      volumes: [volume.getVolumeConfig('/inputs', true)]  // read-only
    };

    return await runComponentWithRunner(runnerConfig, parseOutput, inputs, context);
  } finally {
    await volume.cleanup();  // ALWAYS cleanup
  }
}

const definition: ComponentDefinition<Input, void> = {
  id: 'core.ui.text',
  label: 'Text Block',
  category: 'input',
  runner: { kind: 'inline' },
  inputSchema,
  outputSchema: z.void(),
  metadata: {
    uiOnly: true,  // Excluded from workflow execution
  },
  async execute() {
    // No-op for UI-only components
  }
};

import { describe, it, expect, vi } from 'bun:test';
import * as sdk from '@shipsec/component-sdk';
import { componentRegistry } from '../../index';

describe('my-component', () => {
  it('should process input correctly', async () => {
    const component = componentRegistry.get('my.component.id');
    
    const context = sdk.createExecutionContext({
      runId: 'test-run',
      componentRef: 'test-node',
    });
    
    // Mock the runner for Docker components
    vi.spyOn(sdk, 'runComponentWithRunner').mockResolvedValue('mock output');
    
    const result = await component!.execute({
      inputs: { target: 'example.com' },
      params: {}
    }, context);
    
    expect(result.success).toBe(true);
  });
});

const enableDocker = process.env.ENABLE_DOCKER_TESTS === 'true';
const dockerDescribe = enableDocker ? describe : describe.skip;

dockerDescribe('Component Integration', () => {
  // Tests that run real Docker containers
});

# Test with PTY mode (what workflows use)
docker run --rm -t your-image:latest sh -c 'tool "$@"' -- -flag value

# Verify it doesn't wait for stdin
timeout 5 docker run --rm -t your-image:latest sh -c 'tool "$@"' -- --help

# Start full local environment
just dev

# Verify services are running (via nginx)
curl http://localhost/api/v1/health -H "x-internal-token: local-internal-token"

RUN_E2E=true bun --cwd e2e-tests test

import { describe, test, expect } from 'bun:test';

const API_BASE = 'http://localhost/api/v1';
const HEADERS = {
  'Content-Type': 'application/json',
  'x-internal-token': 'local-internal-token',
};

// Only run when RUN_E2E=true and services are available
const runE2E = process.env.RUN_E2E === 'true';
const e2eDescribe = runE2E ? describe : describe.skip;

// Helper to poll workflow status until completion
async function pollRunStatus(runId: string, timeoutMs = 180000) {
  const startTime = Date.now();
  while (Date.now() - startTime < timeoutMs) {
    const res = await fetch(`${API_BASE}/workflows/runs/${runId}/status`, { headers: HEADERS });
    const status = await res.json();
    if (['COMPLETED', 'FAILED', 'CANCELLED'].includes(status.status)) {
      return status;
    }
    await new Promise(r => setTimeout(r, 1000));
  }
  throw new Error(`Timeout waiting for workflow ${runId}`);
}

e2eDescribe('My Component E2E', () => {
  test('should execute in a real workflow', async () => {
    // 1. Create workflow with your component
    const workflow = {
      name: 'Test: My Component',
      nodes: [
        {
          id: 'start',
          type: 'core.workflow.entrypoint',
          position: { x: 0, y: 0 },
          data: { label: 'Start', config: { runtimeInputs: [] } },
        },
        {
          id: 'my-node',
          type: 'my.component.id',  // Your component ID
          position: { x: 200, y: 0 },
          data: {
            label: 'My Component',
            config: { target: 'example.com' },
          },
        },
      ],
      edges: [{ id: 'e1', source: 'start', target: 'my-node' }],
    };

    // 2. Create workflow via API
    const createRes = await fetch(`${API_BASE}/workflows`, {
      method: 'POST',
      headers: HEADERS,
      body: JSON.stringify(workflow),
    });
    const { id: workflowId } = await createRes.json();

    // 3. Execute workflow
    const runRes = await fetch(`${API_BASE}/workflows/${workflowId}/run`, {
      method: 'POST',
      headers: HEADERS,
      body: JSON.stringify({ inputs: {} }),
    });
    const { runId } = await runRes.json();

    // 4. Poll until completion
    const result = await pollRunStatus(runId);
    
    // 5. Assert results
    expect(result.status).toBe('COMPLETED');
  }, 180000);  // 3 minute timeout for workflow execution
});

import { z } from 'zod';
import { 
  componentRegistry, 
  ComponentDefinition,
  DockerRunnerConfig,
  runComponentWithRunner,
  port,
} from '@shipsec/component-sdk';
import { IsolatedContainerVolume } from '../../utils/isolated-volume';

const inputSchema = z.object({
  domains: z.array(z.string()).min(1),
  threads: z.number().optional().default(10),
});

const outputSchema = z.object({
  results: z.array(z.object({
    domain: z.string(),
    records: z.array(z.string()),
  })),
  rawOutput: z.string(),
});

type Input = z.infer<typeof inputSchema>;
type Output = z.infer<typeof outputSchema>;

export default defineComponent({
  id: 'shipsec.dnsx.scan',
  label: 'DNSX Scanner',
  category: 'security',
  runner: {
    kind: 'docker',
    image: 'ghcr.io/shipsecai/dnsx:latest',
    entrypoint: 'sh',
    command: ['-c', 'dnsx "$@"', '--'],
    network: 'bridge',
  },
  
  inputs: inputs({
    domains: port(z.array(z.string()).min(1), { 
      label: 'Domains',
      connectionType: { kind: 'list', element: { kind: 'primitive', name: 'text' } }
    }),
  }),

  parameters: parameters({
    threads: param(z.number().default(10), {
      label: 'Threads',
      editor: 'number',
      min: 1,
      max: 100,
    }),
  }),

  outputs: outputs({
    results: port(z.array(z.object({
      domain: z.string(),
      records: z.array(z.string()),
    })), { label: 'Scan Results' }),
    rawOutput: port(z.string(), { label: 'Raw Stdout' }),
  }),
  
  async execute({ inputs, params }, context) {
    const tenantId = (context as any).tenantId ?? 'default-tenant';
    const volume = new IsolatedContainerVolume(tenantId, context.runId);

    try {
      context.emitProgress('Preparing input files...');
      await volume.initialize({
        'domains.txt': inputs.domains.join('\n')
      });

      const args = [
        '-l', '/inputs/domains.txt',
        '-json',
        '-t', String(params.threads),
        '-stream',
      ];

      const runnerConfig: DockerRunnerConfig = {
        ...this.runner,
        command: [...(this.runner.command ?? []), ...args],
        volumes: [volume.getVolumeConfig('/inputs', true)],
      };

      context.emitProgress('Running DNSX...');
      const rawOutput = await runComponentWithRunner(
        runnerConfig,
        async (stdout) => {
          const lines = stdout.split('\n').filter(Boolean);
          const results = lines.map(line => {
            const parsed = JSON.parse(line);
            return {
              domain: parsed.host,
              records: parsed.a || [],
            };
          });
          return { results, rawOutput: stdout };
        },
        inputs,
        context
      );

      context.logger.info(`Found ${rawOutput.results.length} results`);
      return rawOutput;

    } finally {
      await volume.cleanup();
    }
  }
});

componentRegistry.register(definition);
export default definition;