purple-team-automation

You are a purple team automation specialist. Automate adversary emulation and detection validation at scale.

Automated Campaign Framework

#!/usr/bin/env python3
"""Automated purple team campaign engine."""
import json
import yaml
from datetime import datetime
from pathlib import Path

class Campaign:
    def __init__(self, name: str, adversary: str):
        self.name = name
        self.adversary = adversary
        self.phases = []
        self.results = {}

    def add_phase(self, name: str, technique_id: str, atomic_test: str,
                  expected_detection: list[str] = None):
        self.phases.append({
            "phase": name,
            "technique": technique_id,
            "atomic": atomic_test,
            "expected_detection_sources": expected_detection or [],
            "status": "pending",
            "evidence": {}
        })

    def execute(self):
        for phase in self.phases:
            print(f"[*] Phase: {phase['phase']} ({phase['technique']})")
            # Execute Atomic Red Team test
            result = self._run_atomic(phase['atomic'])
            phase['evidence']['output'] = result

            # Check detection sources
            detections = self._check_detections(phase['expected_detection_sources'])
            phase['evidence']['detections'] = detections
            phase['status'] = 'detected' if detections else 'missed'

    def report(self) -> dict:
        return {
            "campaign": self.name,
            "adversary": self.adversary,
            "timestamp": datetime.utcnow().isoformat(),
            "phases": self.phases,
            "coverage": sum(1 for p in self.phases if p['status'] == 'detected'),
            "gaps": [p for p in self.phases if p['status'] == 'missed']
        }

mitre_cycle = Campaign("MCT-001", "FIN7")
mitre_cycle.add_phase("Initial Access", "T1566.001", "Spearphishing Attachment",
                       ["email_gateway", "edr"])
mitre_cycle.add_phase("Execution", "T1059.001", "PowerShell Download Cradle",
                       ["sysmon", "windows_event_log"])
mitre_cycle.add_phase("Persistence", "T1053.005", "Scheduled Task",
                       ["sysmon", "windows_event_log"])

Caldera Integration

import requests

caldera = "http://localhost:8888"
headers = {"KEY": "API_KEY"}

def run_operation(name: str, adversary_id: str, groups: list[str]):
    """Run Caldera operation."""
    payload = {
        "name": name,
        "adversary_id": adversary_id,
        "groups": groups,
        "planner_id": "atomic",
        "obfuscators": "base64"
    }
    r = requests.post(f"{caldera}/api/v2/operations", json=payload, headers=headers)
    operation_id = r.json()['id']
    return operation_id

def check_detection(operation_id: str) -> dict:
    """Correlate Caldera actions with SIEM alerts."""
    r = requests.get(f"{caldera}/api/v2/operations/{operation_id}", headers=headers)
    facts = r.json().get('facts', [])

    # Query SIEM for matching alerts
    siem_alerts = query_siem(operation_id)

    return {
        "actions": len(facts),
        "detected": len(siem_alerts),
        "coverage": f"{len(siem_alerts)}/{len(facts)}"
    }