Targets

Targets are the execution layer. They take individual attack steps and run them against infrastructure. Each target type handles a different kind of environment: network hosts, web applications, Active Directory domains, cloud services, AI systems.

How Targets Work

Every target implements three methods:

def connect(self) -> bool:
    """Establish connection. Returns True if reachable."""

def execute(self, step: Dict[str, Any]) -> ExecutionResult:
    """Execute a single attack step."""

def cleanup(self) -> None:
    """Clean up connections and artifacts."""

The engine calls connect() before execution begins, then calls execute() for every step in every sequence, and finally calls cleanup() when the engagement completes.

Target Types

Seven target types are supported in the config validator:

Type	Description	Status
network-service	Network host, subprocess execution	Implemented
web-application	Web app, HTTP-based testing	Planned
active-directory	AD domain, LDAP/Kerberos/SMB	Planned
cloud-service	Cloud provider APIs	Planned
ai-system	AI/ML model endpoints	Planned
ninjato	Ninjato C2 integration	Planned
custom	User-defined target type	Planned

Target Registry

Targets register via the @register_target decorator:

from blackrainbow.targets import register_target
from blackrainbow.targets.base import TargetBase

@register_target
class MyTarget(TargetBase):
    target_type = "my-target"
    ...

Registry functions:

Function	Description
register_target(cls)	Decorator. Registers a target class by its target_type.
get_target(target_type, config=None)	Instantiates a target by type. Raises ValueError if not found.
list_targets()	Returns metadata for all registered targets.

Built-in: NetworkServiceTarget

The network-service target executes tools via subprocess against a network host. This is the primary target type for infrastructure assessments and CTF engagements.

Connection Check

def connect(self) -> bool:

Sends a single ICMP ping (ping -c 1 -W 3) with a 10-second timeout. Returns True if the host responds. The engine treats a failed ping as a warning, not a fatal error, because some targets block ICMP.

Step Execution

def execute(self, step: Dict[str, Any]) -> ExecutionResult:

Each step is a dict with this schema:

Key	Type	Default	Description
tool	str	""	Binary name (e.g., nmap, gobuster, ffuf)
args	str	""	Argument string
timeout	int	300	Seconds before the command is killed
parse	str	""	Output parser hint (e.g., nmap_xml)
sequence_id	str	(auto)	Set by the engine before execution

The target builds a command string (tool + args), splits it with shlex.split(), and runs it via subprocess.run() with capture_output=True.

Return Codes

Code	Meaning
0	Success
-1	Timeout (command exceeded the timeout value)
-2	Tool not found (binary not on PATH)
Other	Process exit code from the tool

Error Handling

Timeout: If subprocess.TimeoutExpired fires, the result gets return_code=-1 with stderr describing the timeout.

Tool not found: If FileNotFoundError fires (binary not installed), the result gets return_code=-2 with stderr identifying the missing tool.

Normal exit: Any non-zero exit code from the tool is captured as-is. stdout and stderr are preserved for the grading step.

Cleanup

NetworkServiceTarget.cleanup() is a no-op. Subprocess execution has no persistent state to tear down.

Config

The target reads from the config dict passed by the engine:

Key	Type	Description
host	str	Target hostname or IP
port	int	Optional specific port
protocol	str	Default: tcp

Writing a Custom Target

"""SSH-based target example."""

from typing import Any, Dict

from blackrainbow.targets import register_target
from blackrainbow.targets.base import TargetBase
from blackrainbow.plugins.base import ExecutionResult


@register_target
class SSHTarget(TargetBase):
    target_type = "ssh-remote"

    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        self.host = config.get("host", "")
        self.user = config.get("ssh_user", "root")
        self.key = config.get("ssh_key", "")

    def connect(self) -> bool:
        # Verify SSH connectivity
        import subprocess
        try:
            result = subprocess.run(
                ["ssh", "-o", "ConnectTimeout=5",
                 "-i", self.key, f"{self.user}@{self.host}", "echo ok"],
                capture_output=True, text=True, timeout=10,
            )
            return result.returncode == 0
        except Exception:
            return False

    def execute(self, step: Dict[str, Any]) -> ExecutionResult:
        import subprocess, time
        tool = step.get("tool", "")
        args = step.get("args", "")
        timeout = step.get("timeout", 300)
        cmd = f"{tool} {args}"

        start = time.time()
        try:
            proc = subprocess.run(
                ["ssh", "-i", self.key, f"{self.user}@{self.host}", cmd],
                capture_output=True, text=True, timeout=timeout,
            )
            return ExecutionResult(
                sequence_id=step.get("sequence_id", ""),
                steps_completed=1, steps_total=1,
                stdout=proc.stdout, stderr=proc.stderr,
                return_code=proc.returncode,
                artifacts=[], duration_seconds=round(time.time() - start, 2),
            )
        except subprocess.TimeoutExpired:
            return ExecutionResult(
                sequence_id=step.get("sequence_id", ""),
                steps_completed=0, steps_total=1,
                stdout="", stderr=f"SSH command timed out after {timeout}s",
                return_code=-1, artifacts=[],
                duration_seconds=round(time.time() - start, 2),
            )

    def cleanup(self) -> None:
        pass

TargetBase Class Reference

Attribute	Type	Default	Description
target_type	str	""	Unique identifier, used in config and registry

Method	Required	Description
connect()	Yes	Check target reachability, return bool
execute(step)	Yes	Run a single attack step, return ExecutionResult
cleanup()	Yes	Tear down connections and artifacts

The constructor receives the full target config dict from the YAML.