Writing Custom Plugins

Build your own attack module. A plugin needs two methods: generate() and grade().

Minimal Example

from blackrainbow.plugins import register_plugin, PluginBase

@register_plugin
class PortScanPlugin(PluginBase):
    plugin_id = "my-port-scan"
    description = "Custom port scanning plugin"

    def generate(self, context):
        host = context.target.get("host", "")
        return [
            {
                "plugin_id": self.plugin_id,
                "description": f"TCP port scan of {host}",
                "steps": [
                    {
                        "tool": "nmap",
                        "args": f"-sV -p 1-1000 {host}",
                        "timeout": 120,
                    }
                ],
                "target_type": "network-service",
                "mitre_techniques": ["T1046"],
                "severity": "medium",
            }
        ]

    def grade(self, sequence, result):
        found_services = "open" in result.stdout.lower()
        return {
            "plugin_id": self.plugin_id,
            "passed": found_services,
            "score": 1.0 if found_services else 0.0,
            "evidence": [result.stdout[:500]] if found_services else [],
            "mitre_techniques": sequence.get("mitre_techniques", []),
            "reasoning": "Services found" if found_services else "No open ports",
        }

That is a complete, working plugin. Save it to a file and reference it in your config.

The Plugin Interface

generate(context)

Takes an engagement context. Returns a list of attack sequences.

def generate(self, context) -> list:

The context gives you everything you need:

Field	What It Contains
context.target	Target config (host, type, port, scope)
context.purpose	Engagement objective
context.scope	In-scope networks and exclusions
context.discovered_services	Services found by previous plugins
context.previous_results	Grade results from earlier in the engagement
context.credentials	Credentials discovered so far

grade(sequence, result)

Takes the original sequence and the execution result. Returns a grade.

def grade(self, sequence, result) -> dict:

The result contains:

Field	What It Contains
result.stdout	Standard output from the tool
result.stderr	Standard error
result.return_code	Exit code (0 = success, -1 = timeout, -2 = tool not found)
result.duration_seconds	How long the tool ran
result.artifacts	List of output file paths

Registering Your Plugin

Option 1: File reference in config

Save your plugin to a Python file and reference it directly:

plugins:
  - file://./plugins/my-port-scan.py

Option 2: Install as a package

Place your plugin in an importable Python package. The @register_plugin decorator handles registration at import time.

Full Example: Web Directory Scanner

from blackrainbow.plugins import register_plugin, PluginBase

@register_plugin
class DirScanPlugin(PluginBase):
    plugin_id = "dir-scan"
    description = "Web directory enumeration"

    def generate(self, context):
        host = context.target.get("host", "")
        url = context.target.get("url", f"http://{host}")
        wordlist = self.config.get("wordlist", "/usr/share/wordlists/dirb/common.txt")

        sequences = [
            {
                "plugin_id": self.plugin_id,
                "description": f"Directory brute-force of {url}",
                "steps": [
                    {
                        "tool": "gobuster",
                        "args": f"dir -u {url} -w {wordlist} -t 50 -o gobuster-out.txt",
                        "timeout": 300,
                    }
                ],
                "target_type": "web-application",
                "mitre_techniques": ["T1083"],
                "severity": "medium",
                "metadata": {"phase": "enumeration"},
            }
        ]

        # If previous recon found specific ports with HTTP, scan those too
        for key, svc in context.discovered_services.items():
            if svc.get("service") in ("http", "https") and svc.get("port") != 80:
                port_url = f"http://{host}:{svc['port']}"
                sequences.append({
                    "plugin_id": self.plugin_id,
                    "description": f"Directory brute-force of {port_url}",
                    "steps": [
                        {
                            "tool": "gobuster",
                            "args": f"dir -u {port_url} -w {wordlist} -t 50",
                            "timeout": 300,
                        }
                    ],
                    "target_type": "web-application",
                    "mitre_techniques": ["T1083"],
                    "severity": "medium",
                })

        return sequences

    def grade(self, sequence, result):
        lines = result.stdout.strip().split("\n")
        found_dirs = [l for l in lines if "Status: 200" in l or "Status: 301" in l]

        return {
            "plugin_id": self.plugin_id,
            "passed": len(found_dirs) > 0,
            "score": min(1.0, len(found_dirs) / 10),
            "evidence": found_dirs[:20],
            "mitre_techniques": ["T1083"],
            "reasoning": f"Found {len(found_dirs)} accessible directories",
        }

Plugin Configuration

Plugins receive a config dict from the YAML:

plugins:
  - id: dir-scan
    numTests: 5
    severity: medium
    config:
      wordlist: /opt/wordlists/big.txt

Access config values in your plugin:

wordlist = self.config.get("wordlist", "/usr/share/wordlists/dirb/common.txt")

Field	Type	Default	Passed To
numTests	integer	5	self.num_tests
severity	string	medium	self.severity
config	object	{}	self.config

Testing Your Plugin

Test it with a dry run:

br run --dry-run --plugins my-port-scan

This runs generate() without executing, so you can verify the attack sequences look right.

Then run it for real against a test target:

br run --plugins my-port-scan --target 192.168.1.100

Tips

• Read the context. Check discovered_services and previous_results to build on earlier findings.
• Set timeouts. Every step should have a reasonable timeout. Hanging tools will block the engagement.
• Map to MITRE. Include ATT&CK technique IDs. They show up in reports and help with coverage tracking.
• Collect evidence. Put useful output in the evidence field. Reports pull directly from it.
• Keep it focused. One plugin, one attack surface. Recon and exploitation should be separate plugins that chain through the context.