workflow-orchestrator

Workflow Orchestrator Skill

Manages cyclic testing workflows with dependency chaining, adaptive re-evaluation, and intelligent state management.

Quick Start

from scripts.workflow_engine import WorkflowEngine
from scripts.dependency_chain import DependencyChain
from scripts.state_manager import StateManager

# Initialize workflow
state = StateManager(target="example.com")
engine = WorkflowEngine(state)

# Define dependency chain
chain = DependencyChain()
chain.add_step("recon", NmapModule(), required=True)
chain.add_step("vuln_scan", NucleiModule(),
               depends_on=["recon"],
               input_mapper=lambda r: {"ports": r.open_ports})
chain.add_step("exploit", ExploitModule(),
               depends_on=["vuln_scan"],
               fallback=EnumModule())  # Fallback if no vulns

# Execute with iterations
results = await engine.execute(chain, max_iterations=3)

Workflow Architecture

┌─────────────────────────────────────────────────────────────┐
│                    WORKFLOW ITERATION                        │
├─────────────────────────────────────────────────────────────┤
│  1. CHECK STATE → Skip if already tested                     │
│  2. EXECUTE MODULE → Run with dependencies                   │
│  3. EVALUATE RESULTS → Success / Empty / Error               │
│  4. ADAPT STRATEGY → Fallback if empty                       │
│  5. CHAIN OUTPUTS → Feed into next modules                   │
│  6. UPDATE STATE → Mark tested, store findings               │
│  7. ITERATE → Continue until max_iterations or complete      │
└─────────────────────────────────────────────────────────────┘

Core Components

1. Dependency Chaining (scripts/dependency_chain.py)

Manages execution order and data flow between modules:

chain = DependencyChain()

# Reconnaissance phase
chain.add_step("port_scan", NmapScanner(), required=True)
chain.add_step("service_enum", ServiceEnumerator(),
               depends_on=["port_scan"])

# Vulnerability phase  
chain.add_step("web_scan", NucleiScanner(),
               depends_on=["port_scan"],
               condition=lambda ctx: 80 in ctx.get("port_scan", {}).ports or
                                    443 in ctx.get("port_scan", {}).ports)

# Exploitation phase
chain.add_step("sql_exploit", SQLMapModule(),
               depends_on=["web_scan"],
               condition=lambda ctx: "sql_injection" in ctx.findings)