workflow-orchestrator

Files (11)

• SKILL.md
• api-testing.yaml · assets/workflows/
• standard-web.yaml · assets/workflows/
• dependency_patterns.md · references/
• state_machine.md · references/
• __init__.py · scripts/
• adaptive_strategy.py · scripts/
• dependency_chain.py · scripts/
• parallel_executor.py · scripts/
• state_manager.py · scripts/
• workflow_engine.py · scripts/

SKILL.md (first lines)

# Workflow Orchestrator Skill

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

## Quick Start

```python
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:

```python
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)
```

See [references/dependency_patterns.md](references/dependency_patterns.md) for complete patterns.

### 2. Adaptive Re-evaluation (`scripts/adaptive_strategy.py`)

Switches strategies based on results:

| Result | Action |
|--------|--------|
| Success | Continue to dependent modules |
| Empty | Trigger fallback module |
| Error | Retry with reduced intensity |
| Timeout | Queue for later retry |

```python
# Define fallback strategies
fallbacks = {
    "web_scan": {
        "empty_result": "service_deep_scan",
        "error": "reduced_web_scan",
        "timeout": "queue_retry"
    },
    "vuln_scan": {
        "empty_result": "full_port_scan",
        "blocked": "stealth_scan"
    }
}

strategy = AdaptiveStrategy(fallbacks)
new_module = strategy.decide("web_scan", result)
```

### 3. State Management (`scripts/state_manager.py`)

Prevents redundant scans using Smart Memory:

```python
state = StateManager(target="example.com")

# Check before scan
if state.is_tested("nmap", ports="80,443"):
    return state.get_result("nmap")

# Execute and store
result = await nmap.scan("example.com", ports="80,443")
state.store("nmap", result, metadata={"ports": "80,443", "timestamp": now()})

# Track tested paths
state.add_tested_path("/admin", method="GET", params={"id": "1"})

From SHAdd0WTAka/Zen-Ai-Pentest, MIT.

More from SHAdd0WTAka/Zen-Ai-Pentest