Plugin Examples

This section contains practical examples of plugins for different use cases.

Base64 Decoding Plugin

Purpose: Decodes base64-embedded content in files

Structure

decodebase64/
├── plugin.toml
├── plugin.py
└── schema.json

plugin.toml

name = "decodebase64"
version = "1.0.0"
api_version = "1.x"
entry = "python3 plugin.py"
capabilities = ["transform"]
needs_content = true
reads_fs = false
timeout_ms = 5000
config_schema = "schema.json"

schema.json

{
  "type": "object",
  "properties": {
    "mode": {
      "type": "string",
      "enum": ["safe", "aggressive"],
      "default": "safe"
    },
    "min_length": {
      "type": "integer",
      "minimum": 16,
      "default": 64
    }
  }
}

plugin.py

#!/usr/bin/env python3
import sys
import json
import base64
import re

def send_response(id, result=None, error=None):
    obj = {"jsonrpc": "2.0", "id": id}
    obj["result" if error is None else "error"] = result if error is None else error
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()

config = {"mode": "safe", "min_length": 64}

for line in sys.stdin:
    msg = json.loads(line)
    mid, mth, p = msg.get("id"), msg.get("method"), msg.get("params", {})
    
    if mth == "plugin.init":
        config.update(p.get("options", {}))
        send_response(mid, {
            "ok": True,
            "capabilities": ["transform"],
            "plugin_version": "1.0.0"
        })
    elif mth == "file.transform":
        results, decoded = [], 0
        for f in p.get("files", []):
            content = base64.b64decode(f["content_b64"])
            pattern = rb"[A-Za-z0-9+/]{%d,}={0,2}" % config["min_length"]
            matches = re.findall(pattern, content)
            
            if matches:
                try:
                    decoded_content = b"".join([base64.b64decode(m) for m in matches])
                    results.append({
                        "path": f["path"],
                        "actions": ["decoded:base64"],
                        "content_b64": base64.b64encode(decoded_content).decode(),
                        "notes": [f"blocks:{len(matches)}"]
                    })
                    decoded += 1
                except Exception:
                    results.append({"path": f["path"], "actions": []})
            else:
                results.append({"path": f["path"], "actions": []})
        
        send_response(mid, {
            "files": results,
            "metrics": {"decoded": decoded, "ms": 0}
        })
    elif mth == "plugin.ping":
        send_response(mid, {"pong": True})
    elif mth == "plugin.shutdown":
        break
    else:
        send_response(mid, None, {
            "code": 1002,
            "message": "unknown method",
            "data": {"method": mth}
        })

Usage

rootcause ./code --plugin ./plugins/decodebase64 --plugin-opt decodebase64.mode=aggressive

TypeScript analyser

Purpose: Detects dangerous use of eval() in TypeScript files

plugin.toml

name = "ts-eval-detector"
version = "1.0.0"
api_version = "1.x"
entry = "python3 plugin.py"
capabilities = ["analyze"]
needs_content = true
reads_fs = false
timeout_ms = 3000

plugin.py

#!/usr/bin/env python3
import sys
import json
import base64
import hashlib

def send_response(id, result=None, error=None):
    obj = {"jsonrpc": "2.0", "id": id}
    obj["result" if error is None else "error"] = result if error is None else error
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()

for line in sys.stdin:
    msg = json.loads(line)
    mid, mth, p = msg.get("id"), msg.get("method"), msg.get("params", {})
    
    if mth == "plugin.init":
        send_response(mid, {
            "ok": True,
            "capabilities": ["analyze"],
            "plugin_version": "1.0.0"
        })
    elif mth == "file.analyze":
        findings = []
        for f in p.get("files", []):
            if f.get("language") != "typescript":
                continue
                
            content = base64.b64decode(f["content_b64"]).decode('utf-8')
            lines = content.split('\n')
            
            for i, line in enumerate(lines, 1):
                if "eval(" in line:
                    col = line.find("eval(") + 1
                    fingerprint = hashlib.sha256(f"{f['path']}:{i}:{line}".encode()).hexdigest()
                    
                    findings.append({
                        "rule_id": "ts.no-eval",
                        "severity": "MEDIUM",
                        "file": f["path"],
                        "line": i,
                        "column": col,
                        "excerpt": line.strip(),
                        "message": "Avoid using eval() in TypeScript",
                        "remediation": "Use Function() or an appropriate parser",
                        "fingerprint": f"sha256:{fingerprint}"
                    })
        
        send_response(mid, {"findings": findings})
    elif mth == "plugin.ping":
        send_response(mid, {"pong": True})
    elif mth == "plugin.shutdown":
        break
    else:
        send_response(mid, None, {
            "code": 1002,
            "message": "unknown method"
        })

Custom report generator

Purpose: Generates reports in custom HTML format

plugin.toml

name = "html-reporter"
version = "1.0.0"
api_version = "1.x"
entry = "python3 plugin.py"
capabilities = ["report"]
needs_content = false
reads_fs = false
timeout_ms = 5000

plugin.py

#!/usr/bin/env python3
import sys
import json
import base64
from datetime import datetime

def send_response(id, result=None, error=None):
    obj = {"jsonrpc": "2.0", "id": id}
    obj["result" if error is None else "error"] = result if error is None else error
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()

def generate_html_report(findings):
    html = f"""
    <!DOCTYPE html>
    <html>
    <head>
        <title>RootCause Report - {datetime.now().strftime('%Y-%m-%d %H:%M')}</title>
        <style>
            body { font-family: Arial, sans-serif; margin: 20px; }
            .finding { border: 1px solid #ccc; margin: 10px 0; padding: 15px; }
            .high { border-left: 5px solid #ff4444; }
            .medium { border-left: 5px solid #ffaa00; }
            .low { border-left: 5px solid #44aa44; }
            .file { font-weight: bold; color: #0066cc; }
            .message { margin: 10px 0; }
            .remediation { background: #f0f0f0; padding: 10px; margin: 10px 0; }
        </style>
    </head>
    <body>
        <h1>RootCause Analysis Report</h1>
        <p>Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
        <p>Total findings: {len(findings)}</p>
    """
    
    for finding in findings:
        severity_class = finding["severity"].lower()
        html += f"""
        <div class="finding {severity_class}">
            <div class="file">{finding["file"]}:{finding["line"]}:{finding["column"]}</div>
            <div class="message"><strong>{finding["message"]}</strong></div>
            <div class="remediation"><strong>Recommendation:</strong> {finding["remediation"]}</div>
            <pre>{finding["excerpt"]}</pre>
        </div>
        """
    
    html += "</body></html>"
    return html

for line in sys.stdin:
    msg = json.loads(line)
    mid, mth, p = msg.get("id"), msg.get("method"), msg.get("params", {})
    
    if mth == "plugin.init":
        send_response(mid, {
            "ok": True,
            "capabilities": ["report"],
            "plugin_version": "1.0.0"
        })
    elif mth == "report.generate":
        findings = p.get("findings", [])
        html_content = generate_html_report(findings)
        
        with open("rootcause-report.html", "w", encoding="utf-8") as f:
            f.write(html_content)
        
        send_response(mid, {
            "files": [{
                "path": "rootcause-report.html",
                "content_b64": base64.b64encode(html_content.encode()).decode(),
                "format": "html"
            }],
            "metrics": {
                "findings_processed": len(findings),
                "files_generated": 1
            }
        })
    elif mth == "plugin.ping":
        send_response(mid, {"pong": True})
    elif mth == "plugin.shutdown":
        break
    else:
        send_response(mid, None, {
            "code": 1002,
            "message": "unknown method"
        })

Using the examples

# Install plugins
mkdir -p ~/.config/rootcause/plugins
cp -r decodebase64 ~/.config/rootcause/plugins/
cp -r ts-eval-detector ~/.config/rootcause/plugins/
cp -r html-reporter ~/.config/rootcause/plugins/

# Run with multiple plugins
rootcause ./project \
  --plugin decodebase64 \
  --plugin ts-eval-detector \
  --plugin html-reporter \
  --plugin-opt decodebase64.mode=aggressive