Lab 15: Purple Team Automation¶

#!/usr/bin/env python3
"""
SYNTHETIC — Purple Team Detection Validation Correlator
File: C:\PurpleTeam\scripts\correlate_results.py
Organization: ACME Security Corp (fictional)

Correlates Atomic Red Team execution reports with SIEM alert exports
to identify detection gaps.
"""

import json
import csv
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional


def load_art_report(report_path: str) -> dict:
    """Load Atomic Red Team JSON execution report."""
    with open(report_path, "r") as f:
        return json.load(f)


def load_siem_alerts(alerts_path: str) -> list[dict]:
    """Load SIEM alert export (CSV format)."""
    alerts = []
    with open(alerts_path, "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            alerts.append(row)
    return alerts


def correlate(art_report: dict, siem_alerts: list[dict],
              time_window_minutes: int = 10) -> dict:
    """
    Correlate ART test executions with SIEM alerts.

    For each technique executed, check if a corresponding alert
    was generated within the time window.
    """
    results = {
        "run_id": art_report["run_id"],
        "timestamp": art_report["timestamp"],
        "threat_actor": art_report["threat_actor"],
        "host": art_report["host"],
        "total_techniques": 0,
        "detected": 0,
        "gaps": 0,
        "coverage_pct": 0.0,
        "technique_results": []
    }

    for phase in art_report["phases"]:
        for technique in phase["techniques"]:
            results["total_techniques"] += 1
            tech_id = technique["technique_id"]
            tech_name = technique["technique_name"]
            exec_time = technique.get("start_time", "")

            # Search for matching alert
            matching_alert = find_matching_alert(
                tech_id, exec_time, siem_alerts, time_window_minutes
            )

            if matching_alert:
                results["detected"] += 1
                status = "DETECTED"
                alert_name = matching_alert.get("AlertName",
                              matching_alert.get("rule_name", "Unknown"))
                alert_severity = matching_alert.get("AlertSeverity",
                                  matching_alert.get("urgency", "Unknown"))
                time_to_detect = calculate_ttd(exec_time,
                                  matching_alert.get("TimeGenerated",
                                  matching_alert.get("_time", "")))
            else:
                results["gaps"] += 1
                status = "GAP"
                alert_name = None
                alert_severity = None
                time_to_detect = None

            results["technique_results"].append({
                "technique_id": tech_id,
                "technique_name": tech_name,
                "tactic": phase["tactic"],
                "phase": phase["phase"],
                "status": status,
                "alert_name": alert_name,
                "alert_severity": alert_severity,
                "time_to_detect_seconds": time_to_detect,
                "tests_executed": len(technique.get("tests", []))
            })

    if results["total_techniques"] > 0:
        results["coverage_pct"] = round(
            (results["detected"] / results["total_techniques"]) * 100, 1
        )

    return results


def find_matching_alert(technique_id: str, exec_time: str,
                         alerts: list[dict],
                         window_minutes: int) -> Optional[dict]:
    """Find a SIEM alert matching the technique within the time window."""
    for alert in alerts:
        alert_techniques = alert.get("AttackTechniques",
                            alert.get("attack_technique", ""))
        if technique_id in str(alert_techniques):
            return alert
    return None


def calculate_ttd(exec_time: str, alert_time: str) -> Optional[float]:
    """Calculate time-to-detect in seconds."""
    try:
        exec_dt = datetime.fromisoformat(exec_time.replace("Z", "+00:00"))
        alert_dt = datetime.fromisoformat(alert_time.replace("Z", "+00:00"))
        return round((alert_dt - exec_dt).total_seconds(), 1)
    except (ValueError, AttributeError):
        return None


def generate_report(correlation: dict, output_path: str) -> None:
    """Generate the detection validation report."""
    with open(output_path, "w") as f:
        json.dump(correlation, f, indent=2)

    print(f"\n{'='*70}")
    print(f"  PURPLE TEAM DETECTION VALIDATION REPORT")
    print(f"  Run ID: {correlation['run_id']}")
    print(f"  Threat Actor Profile: {correlation['threat_actor']}")
    print(f"  Target Host: {correlation['host']}")
    print(f"{'='*70}")
    print(f"\n  Total Techniques Tested:  {correlation['total_techniques']}")
    print(f"  Detected:                 {correlation['detected']}")
    print(f"  Detection Gaps:           {correlation['gaps']}")
    print(f"  Coverage:                 {correlation['coverage_pct']}%")
    print(f"\n{'─'*70}")
    print(f"  {'Technique':<14} {'Name':<35} {'Status':<10} {'TTD (s)'}")
    print(f"  {'─'*14} {'─'*35} {'─'*10} {'─'*10}")

    for t in correlation["technique_results"]:
        ttd = f"{t['time_to_detect_seconds']:.1f}" if t["time_to_detect_seconds"] else "N/A"
        status_color = "DETECTED" if t["status"] == "DETECTED" else "** GAP **"
        print(f"  {t['technique_id']:<14} {t['technique_name']:<35} {status_color:<10} {ttd}")

    print(f"\n{'='*70}")


if __name__ == "__main__":
    art_report = load_art_report(sys.argv[1])
    siem_alerts = load_siem_alerts(sys.argv[2])
    output_path = sys.argv[3] if len(sys.argv) > 3 else "validation_report.json"

    correlation = correlate(art_report, siem_alerts)
    generate_report(correlation, output_path)

#!/usr/bin/env python3
"""
SYNTHETIC — ATT&CK Navigator Layer Generator
File: C:\PurpleTeam\scripts\generate_navigator_layer.py
Organization: ACME Security Corp (fictional)

Converts purple team validation reports into ATT&CK Navigator layer files.
"""

import json
import sys
from datetime import datetime


def generate_layer(validation_report: dict) -> dict:
    """Generate an ATT&CK Navigator layer from validation results."""

    techniques = []
    for result in validation_report["technique_results"]:
        tech_id = result["technique_id"]

        # Determine score and color based on detection status
        if result["status"] == "DETECTED":
            score = 100
            color = "#8ec843"  # Green
            comment = (
                f"DETECTED — Alert: {result.get('alert_name', 'N/A')} | "
                f"Severity: {result.get('alert_severity', 'N/A')} | "
                f"TTD: {result.get('time_to_detect_seconds', 'N/A')}s"
            )
        else:
            score = 0
            color = "#ff6666"  # Red
            comment = (
                f"GAP — No detection alert fired during test. "
                f"Technique: {result['technique_name']} | "
                f"Phase: {result.get('phase', 'N/A')}"
            )

        # Handle sub-techniques (e.g., T1059.001 -> T1059 with tactic)
        if "." in tech_id:
            base_id, sub_id = tech_id.rsplit(".", 1)
            techniques.append({
                "techniqueID": base_id,
                "tactic": map_tactic(result.get("tactic", "")),
                "score": score,
                "color": color,
                "comment": comment,
                "enabled": True,
                "showSubtechniques": True,
                "metadata": [
                    {"name": "run_id", "value": validation_report["run_id"]},
                    {"name": "test_date", "value": validation_report["timestamp"]},
                    {"name": "tests_executed", "value": str(result.get("tests_executed", 0))}
                ]
            })
            # Also add the sub-technique entry
            techniques.append({
                "techniqueID": tech_id,
                "tactic": map_tactic(result.get("tactic", "")),
                "score": score,
                "color": color,
                "comment": comment,
                "enabled": True,
                "metadata": [
                    {"name": "run_id", "value": validation_report["run_id"]},
                    {"name": "test_date", "value": validation_report["timestamp"]}
                ]
            })
        else:
            techniques.append({
                "techniqueID": tech_id,
                "tactic": map_tactic(result.get("tactic", "")),
                "score": score,
                "color": color,
                "comment": comment,
                "enabled": True,
                "metadata": [
                    {"name": "run_id", "value": validation_report["run_id"]},
                    {"name": "test_date", "value": validation_report["timestamp"]},
                    {"name": "tests_executed", "value": str(result.get("tests_executed", 0))}
                ]
            })

    layer = {
        "name": f"ACME Security Corp — {validation_report['threat_actor']} Coverage ({datetime.now().strftime('%Y-%m-%d')})",
        "versions": {
            "attack": "14",
            "navigator": "4.9.1",
            "layer": "4.5"
        },
        "domain": "enterprise-attack",
        "description": (
            f"Detection coverage for {validation_report['threat_actor']} threat profile. "
            f"Coverage: {validation_report['coverage_pct']}% "
            f"({validation_report['detected']}/{validation_report['total_techniques']} techniques). "
            f"Generated: {validation_report['timestamp']}. "
            f"SYNTHETIC — ACME Security Corp (fictional)."
        ),
        "sorting": 3,
        "layout": {
            "layout": "side",
            "aggregateFunction": "average",
            "showID": True,
            "showName": True,
            "showAggregateScores": True,
            "countUnscored": False
        },
        "hideDisabled": False,
        "gradient": {
            "colors": ["#ff6666", "#ffe766", "#8ec843"],
            "minValue": 0,
            "maxValue": 100
        },
        "legendItems": [
            {"label": "Detected (alert fired)", "color": "#8ec843"},
            {"label": "Partial (logged, no alert)", "color": "#ffe766"},
            {"label": "Gap (not detected)", "color": "#ff6666"},
            {"label": "Not tested", "color": "#ffffff"}
        ],
        "techniques": techniques
    }

    return layer


def map_tactic(tactic_id: str) -> str:
    """Map ATT&CK tactic ID to Navigator tactic name."""
    mapping = {
        "TA0001": "initial-access",
        "TA0002": "execution",
        "TA0003": "persistence",
        "TA0004": "privilege-escalation",
        "TA0005": "defense-evasion",
        "TA0006": "credential-access",
        "TA0007": "discovery",
        "TA0008": "lateral-movement",
        "TA0009": "collection",
        "TA0010": "exfiltration",
        "TA0011": "command-and-control",
        "TA0040": "impact",
    }
    return mapping.get(tactic_id, "")


if __name__ == "__main__":
    with open(sys.argv[1]) as f:
        validation = json.load(f)

    layer = generate_layer(validation)
    output = sys.argv[2] if len(sys.argv) > 2 else "navigator_layer.json"

    with open(output, "w") as f:
        json.dump(layer, f, indent=2)

    print(f"Navigator layer written to {output}")
    print(f"Coverage: {validation['coverage_pct']}% "
          f"({validation['detected']}/{validation['total_techniques']})")
    print(f"Load this file in ATT&CK Navigator: https://mitre-attack.github.io/attack-navigator/")

# SYNTHETIC — Multi-run comparison layer generator (excerpt)
# File: C:\PurpleTeam\scripts\compare_runs.py

def generate_comparison_layer(run_reports: list[dict]) -> dict:
    """
    Generate a comparison layer showing coverage trends.

    Color coding:
    - Dark green (#2d6a2e): Detected in ALL runs (stable detection)
    - Light green (#8ec843): Detected in latest run (new detection)
    - Yellow (#ffe766): Detected in past but NOT latest (regression)
    - Red (#ff6666): Never detected (persistent gap)
    """
    technique_history = {}

    for i, report in enumerate(run_reports):
        for result in report["technique_results"]:
            tid = result["technique_id"]
            if tid not in technique_history:
                technique_history[tid] = {
                    "name": result["technique_name"],
                    "tactic": result.get("tactic", ""),
                    "detections": []
                }
            technique_history[tid]["detections"].append(
                result["status"] == "DETECTED"
            )

    # Assign colors based on history
    techniques = []
    for tid, history in technique_history.items():
        all_detected = all(history["detections"])
        latest_detected = history["detections"][-1] if history["detections"] else False
        ever_detected = any(history["detections"])

        if all_detected:
            color = "#2d6a2e"
            score = 100
            label = "Stable Detection"
        elif latest_detected:
            color = "#8ec843"
            score = 75
            label = "New Detection"
        elif ever_detected:
            color = "#ffe766"
            score = 25
            label = "REGRESSION"
        else:
            color = "#ff6666"
            score = 0
            label = "Persistent Gap"

        techniques.append({
            "techniqueID": tid,
            "tactic": map_tactic(history["tactic"]),
            "score": score,
            "color": color,
            "comment": f"{label} — {sum(history['detections'])}/{len(history['detections'])} runs detected"
        })

    # ... (layer construction same as 3.2)
    return {"techniques": techniques}