Skip to main content

Pipeline

The BlackRainbow engine runs a 5-stage pipeline for every engagement. Each stage feeds the next. Results accumulate in a shared EngagementContext that all plugins can read.

Stages​

┌─────────┐    ┌───────────┐    ┌─────────┐    ┌───────┐    ┌────────┐
│ Generate │ →  │ Transform │ →  │ Execute │ →  │ Grade │ →  │ Report │
└─────────┘    └───────────┘    └─────────┘    └───────┘    └────────┘
  plugins        strategies       targets       plugins       (planned)

Stage 1: Generate​

Each plugin produces a list of AttackSequence objects based on the engagement context.

sequences = plugin.generate(context)

An AttackSequence contains:

  • plugin_id: which plugin created it
  • description: human-readable summary
  • steps: list of step dicts (tool, args, timeout, parse hint)
  • target_type: what kind of target this applies to
  • mitre_techniques: ATT&CK technique IDs
  • colors: color categories for the sequence
  • severity: INFO, LOW, MEDIUM, HIGH, CRITICAL
  • metadata: arbitrary plugin-specific data

Example output from the recon plugin:

AttackSequence(
    plugin_id="recon",
    description="Full TCP port scan of 10.10.10.5",
    steps=[{
        "tool": "nmap",
        "args": "-sC -sV -p- -oX /tmp/br-nmap-full-10.10.10.5.xml 10.10.10.5",
        "parse": "nmap_xml",
        "timeout": 600,
    }],
    target_type="network-service",
    mitre_techniques=["T1046"],
    colors=["orange"],
    severity=Severity.INFO,
    metadata={"phase": "port_scan"},
)

Stage 2: Transform​

Strategies reorder, filter, or modify the generated sequences. Each strategy receives the full sequence list and the engagement context, then returns a (potentially modified) list.

for strategy in strategies:
    sequences = strategy.transform(sequences, context)

Strategies are optional. If no strategies are configured, sequences pass through unchanged.

Stage 3: Execute​

The target adapter runs each step in every sequence. For NetworkServiceTarget, this means spawning a subprocess:

result = target.execute(step)

The ExecutionResult contains:

  • stdout and stderr: raw command output
  • return_code: 0 (success), -1 (timeout), -2 (tool not found), or the process exit code
  • duration_seconds: wall-clock execution time
  • artifacts: list of output file paths
  • raw_output: dict with the original command and parse hint

Before execution begins, the engine calls target.connect() to verify reachability (ping for network targets). If the target is unreachable, a warning is printed but execution continues, because some tools handle their own connectivity.

Stage 4: Grade​

After each step executes, the originating plugin grades the result:

grade = plugin.grade(sequence, result)

The GradeResult contains:

  • passed: boolean, did this step achieve its objective
  • score: 0.0 to 1.0 float
  • objective_achieved: boolean
  • evidence: list of strings (discovered services, vulns, etc.)
  • mitre_techniques: ATT&CK IDs from the sequence
  • reasoning: human-readable explanation
  • metadata: plugin-specific data (services, vulns, phase)

Stage 5: Report​

Planned for Sprint 2. Will generate markdown, JSON, HTML, or PDF reports from accumulated grades and sequences.

Service Accumulation​

This is the key feedback mechanism. When a grading result includes discovered services in its metadata, the engine writes them into EngagementContext.discovered_services:

if grade.metadata.get("services"):
    for svc in grade.metadata["services"]:
        key = f"{svc['port']}/{svc['protocol']}"
        context.discovered_services[key] = svc

Downstream plugins can read context.discovered_services to generate targeted sequences. For example, if recon discovers an HTTP server on port 8080, a web exploitation plugin can generate sequences specifically for that service.

Similarly, context.previous_results accumulates all GradeResult objects, giving every plugin access to the full history of what has been tried and what worked.

EngagementContext​

The shared state object passed through the entire pipeline:

@dataclass
class EngagementContext:
    target: Dict[str, Any]           # Target config as dict
    purpose: str                      # Engagement purpose from config
    scope: Dict[str, Any]            # Network scope constraints
    previous_results: List[GradeResult]     # All grades so far
    discovered_services: Dict[str, Any]     # Accumulated service map
    credentials: List[Dict[str, str]]       # Discovered credentials
    model: Optional[Any]                    # AI model reference
    strategies: List[str]                   # Active strategy IDs

Engine Pseudocode​

The complete engine flow:

load config from YAML
build EngagementContext from config
instantiate plugins, strategies, target

if not dry_run:
    target.connect()

for each plugin:
    sequences = plugin.generate(context)

    for each strategy:
        sequences = strategy.transform(sequences, context)

    if dry_run:
        print each sequence and its steps
        continue

    for each sequence:
        for each step in sequence:
            result = target.execute(step)
            grade = plugin.grade(sequence, result)
            context.previous_results.append(grade)

            if grade has services:
                update context.discovered_services

if not dry_run:
    target.cleanup()

return summary {sequences_total, grades_total, passed, failed}

Dry Run Mode​

When --dry-run is passed, the engine runs stages 1 and 2 (generate and transform) but skips execution and grading. Each sequence is printed with its steps, showing what tools and arguments would be used. This is useful for reviewing the attack plan before committing to execution.