performing-purple-team-atomic-testing

执行紫队原子测试

何时使用

• 针对特定 MITRE ATT&CK 技术验证检测覆盖率
• 使用 Atomic Red Team 测试库开展紫队演练
• 执行 ATT&CK 覆盖率差距分析，识别 SIEM/EDR 中的盲点
• 构建检测验证闭环：执行原子测试、检查 SIEM、调整规则、重新测试
• 生成 ATT&CK Navigator 热力图图层，用于管理层汇报
• 在 CI/CD 或定时流水线中自动化持续原子测试
• 将威胁情报报告映射为可执行的原子测试

请勿使用于需要自定义植入物或超出原子测试范围的实况对手模拟的全范围红队项目；如需高级对手模拟，请使用 Caldera、SCYTHE 或 Cobalt Strike。

免责声明：Atomic Red Team 测试会执行真实的攻击技术。请仅在你拥有或已获得明确书面授权测试的系统上运行。许多测试会修改系统状态、创建产物或触发安全告警。测试后务必执行清理命令。未经利益相关方接受风险，切勿在生产环境中运行原子测试。

前提条件

• Windows 主机，配备 PowerShell 5.1+ 或 PowerShell Core 7+（跨平台原子测试支持 Linux/macOS）
• 从 PSGallery 安装 Invoke-AtomicRedTeam PowerShell 模块
• 本地克隆 Atomic Red Team atomics 仓库
• SIEM/EDR 可从测试端点摄取日志（Splunk、Elastic、Microsoft Sentinel、CrowdStrike）
• MITRE ATT&CK Navigator（基于 Web 或本地实例），用于图层可视化
• Python 3.9+，配备 mitreattack-python、pyyaml 和 requests，用于自动化脚本
• Sigma 规则仓库，用于检测关联
• 测试端点上的管理员/root 访问权限
• 隔离的测试环境（实验室、沙箱或专用测试靶场）

工作流

步骤 1：安装并配置 Invoke-AtomicRedTeam

搭建执行框架并下载 atomics 库：

# Install the PowerShell execution module
Install-Module -Name invoke-atomicredteam -Scope CurrentUser -Force
Install-Module -Name powershell-yaml -Scope CurrentUser -Force

# Import the module
Import-Module invoke-atomicredteam

# Install atomics to default location (C:\AtomicRedTeam\atomics)
IEX (IEX (New-Object System.Net.WebClient).DownloadString(
    'https://raw.githubusercontent.com/redcanaryco/invoke-atomicredteam/master/install-atomicredteam.ps1'
)); Install-AtomicRedTeam -getAtomics -Force

# Verify installation - list available techniques
$atomicsPath = "C:\AtomicRedTeam\atomics"
$techniques = Get-ChildItem $atomicsPath -Directory | Where-Object { $_.Name -match '^T\d{4}' }
Write-Host "Available techniques: $($techniques.Count)"

# Configure execution logging
$env:ARTLOG = "C:\AtomicRedTeam\logs"
if (-not (Test-Path $env:ARTLOG)) { New-Item -Path $env:ARTLOG -ItemType Directory }

步骤 2：枚举并选择原子测试

盘点可用测试，并根据威胁情报或差距分析选择目标：

# List all tests for a specific technique
Invoke-AtomicTest T1059.001 -ShowDetailsBrief

# Show full details including attack commands and cleanup
Invoke-AtomicTest T1059.001 -ShowDetails

# List tests for a tactic (e.g., Persistence)
$persistenceTechniques = @(
    "T1547.001",  # Boot or Logon Autostart - Registry Run Keys
    "T1053.005",  # Scheduled Task
    "T1136.001",  # Create Account - Local Account
    "T1543.003",  # Create or Modify System Process - Windows Service
    "T1546.001",  # Event Triggered Execution - Change Default File Association
    "T1574.001",  # Hijack Execution Flow - DLL Search Order Hijacking
    "T1197"       # BITS Jobs
)

foreach ($tech in $persistenceTechniques) {
    Write-Host "`n=== $tech ===" -ForegroundColor Cyan
    try {
        Invoke-AtomicTest $tech -ShowDetailsBrief
    } catch {
        Write-Host "  No tests available" -ForegroundColor Yellow
    }
}

# Get all atomic techniques from YAML files programmatically
$allAtomics = Get-ChildItem "$atomicsPath\T*\T*.yaml" -Recurse |
    ForEach-Object {
        $yaml = Get-Content $_.FullName -Raw | ConvertFrom-Yaml
        [PSCustomObject]@{
            TechniqueId   = $yaml.attack_technique
            TechniqueName = $yaml.display_name
            TestCount     = $yaml.atomic_tests.Count
            Platforms     = ($yaml.atomic_tests.supported_platforms | Sort-Object -Unique) -join ", "
        }
    }

$allAtomics | Sort-Object TechniqueId | Format-Table -AutoSize
Write-Host "Total techniques with tests: $($allAtomics.Count)"
Write-Host "Total individual tests: $(($allAtomics | Measure-Object -Property TestCount -Sum).Sum)"

步骤 3：带日志记录执行原子测试

运行测试并记录执行前/后日志，用于检测验证：

# Execute a single test by technique ID (runs all tests for that technique)
Invoke-AtomicTest T1059.001

# Execute a specific test by number
Invoke-AtomicTest T1059.001 -TestNumbers 1

# Execute by test name
Invoke-AtomicTest T1059.001 -TestNames "Mimikatz - Cradled Invoke Expression"

# Execute by GUID
Invoke-AtomicTest T1059.001 -TestGuids "2e803f96-4e33-4c2c-b0c8-1c10cbb3945f"

# Execute with prerequisite check and installation
Invoke-AtomicTest T1059.001 -TestNumbers 1 -CheckPrereqs
Invoke-AtomicTest T1059.001 -TestNumbers 1 -GetPrereqs
Invoke-AtomicTest T1059.001 -TestNumbers 1

# Execute with timeout (seconds)
Invoke-AtomicTest T1003.001 -TimeoutSeconds 120

# Cleanup after testing
Invoke-AtomicTest T1059.001 -TestNumbers 1 -Cleanup

# Execute with full logging wrapper
function Invoke-AtomicWithLogging {
    param(
        [string]$TechniqueId,
        [int[]]$TestNumbers,
        [string]$LogPath = "C:\AtomicRedTeam\logs"
    )

    $timestamp = Get-Date -Format "yyyyMMdd_HHmmss"
    $logFile = Join-Path $LogPath "${TechniqueId}_${timestamp}.json"

    $result = @{
        technique_id  = $TechniqueId
        test_numbers  = $TestNumbers
        start_time    = (Get-Date).ToString("o")
        hostname      = $env:COMPUTERNAME
        username      = $env:USERNAME
        results       = @()
    }

    foreach ($testNum in $TestNumbers) {
        $testResult = @{
            test_number = $testNum
            status      = "unknown"
            start_time  = (Get-Date).ToString("o")
        }

        try {
            # Show what will execute
            $details = Invoke-AtomicTest $TechniqueId -TestNumbers $testNum -ShowDetails 2>&1
            $testResult["details"] = $details | Out-String

            # Execute the test
            Invoke-AtomicTest $TechniqueId -TestNumbers $testNum -Confirm:$false
            $testResult["status"] = "executed"
        } catch {
            $testResult["status"] = "failed"
            $testResult["error"] = $_.Exception.Message
        }

        $testResult["end_time"] = (Get-Date).ToString("o")
        $result.results += $testResult

        # Wait for SIEM ingestion
        Start-Sleep -Seconds 30
    }

    $result["end_time"] = (Get-Date).ToString("o")
    $result | ConvertTo-Json -Depth 10 | Set-Content $logFile
    Write-Host "Log written to: $logFile" -ForegroundColor Green
    return $result
}

# Usage
Invoke-AtomicWithLogging -TechniqueId "T1059.001" -TestNumbers @(1, 2, 3)

步骤 4：在 SIEM 中验证检测

查询你的 SIEM，确认原子测试是否生成了告警：

Splunk SPL Queries for Detection Validation:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

-- T1059.001: PowerShell Execution
index=windows sourcetype="WinEventLog:Microsoft-Windows-PowerShell/Operational"
  EventCode=4104
  | eval script_block=ScriptBlockText
  | where len(script_block) > 500
  | stats count by host, script_block
  | sort -count

-- T1003.001: LSASS Memory Credential Dumping
index=windows sourcetype="WinEventLog:Security" EventCode=4663
  ObjectName="*lsass*"
  | stats count by host, SubjectUserName, ProcessName
  | where count > 0

-- T1547.001: Registry Run Key Persistence
index=windows sourcetype="WinEventLog:Microsoft-Windows-Sysmon/Operational"
  EventCode=13
  TargetObject="*\\CurrentVersion\\Run*"
  | stats count by host, Image, TargetObject, Details

-- T1053.005: Scheduled Task Creation
index=windows sourcetype="WinEventLog:Security" EventCode=4698
  | stats count by host, SubjectUserName, TaskName, TaskContent
  | sort -count

-- Generic: Hunt for Atomic Red Team artifacts
index=windows (sourcetype="WinEventLog:Microsoft-Windows-Sysmon/Operational"
  OR sourcetype="WinEventLog:Security")
  | search "*AtomicRedTeam*" OR "*atomic*" OR "*Invoke-AtomicTest*"
  | stats count by sourcetype, EventCode, host

Elastic / KQL Queries:
━━━━━━━━━━━━━━━━━━━━━

-- PowerShell script block logging
event.code: "4104" and powershell.file.script_block_text: *

-- Sysmon process creation from AtomicRedTeam paths
event.code: "1" and process.executable: *AtomicRedTeam*

-- Registry modification (persistence)
event.code: "13" and registry.path: *CurrentVersion\\Run*

-- Credential access indicators
event.code: "10" and winlog.event_data.TargetImage: *lsass.exe*

步骤 5：ATT&CK 覆盖率差距分析

生成覆盖率矩阵，对比已测试与已检测的技术：

#!/usr/bin/env python3
"""ATT&CK coverage gap analysis - compares atomic test results against SIEM detections."""

import json
import os
import yaml
from pathlib import Path
from datetime import datetime


def load_atomics_inventory(atomics_path):
    """Parse all atomic test YAML files to build technique inventory."""
    inventory = {}
    atomics_dir = Path(atomics_path)

    for yaml_file in atomics_dir.glob("T*/T*.yaml"):
        try:
            with open(yaml_file, "r", encoding="utf-8") as f:
                data = yaml.safe_load(f)

            tech_id = data.get("attack_technique", "")
            if not tech_id:
                continue

            tests = data.get("atomic_tests", [])
            inventory[tech_id] = {
                "name": data.get("display_name", "Unknown"),
                "test_count": len(tests),
                "platforms": list(set(
                    p for t in tests
                    for p in t.get("supported_platforms", [])
                )),
                "tests": [
                    {
                        "name": t.get("name", "Unnamed"),
                        "description": t.get("description", ""),
                        "platforms": t.get("supported_platforms", []),
                        "executor": t.get("executor", {}).get("name", "unknown"),
                    }
                    for t in tests
                ],
            }
        except Exception as e:
            print(f"[WARN] Failed to parse {yaml_file}: {e}")

    return inventory


def load_execution_logs(log_dir):
    """Load atomic test execution logs."""
    executed = {}
    log_path = Path(log_dir)

    if not log_path.exists():
        return executed

    for log_file in log_path.glob("T*_*.json"):
        try:
            with open(log_file, "r") as f:
                data = json.load(f)
            tech_id = data.get("technique_id", "")
            if tech_id:
                if tech_id not in executed:
                    executed[tech_id] = {
                        "executions": [],
                        "last_executed": data.get("end_time", ""),
                    }
                executed[tech_id]["executions"].append({
                    "timestamp": data.get("start_time", ""),
                    "results": data.get("results", []),
                })
        except Exception as e:
            print(f"[WARN] Failed to parse {log_file}: {e}")

    return executed


def load_detection_results(detection_file):
    """Load SIEM detection validation results (JSON export from SIEM queries)."""
    if not os.path.exists(detection_file):
        return {}

    with open(detection_file, "r") as f:
        data = json.load(f)

    detections = {}
    for entry in data:
        tech_id = entry.get("technique_id", "")
        if tech_id:
            detections[tech_id] = {
                "detected": entry.get("detected", False),
                "alert_count": entry.get("alert_count", 0),
                "rule_name": entry.get("rule_name", ""),
                "confidence": entry.get("confidence", "unknown"),
                "data_sources": entry.get("data_sources", []),
            }

    return detections


# MITRE ATT&CK tactic ordering for structured output
TACTIC_ORDER = [
    "reconnaissance", "resource-development", "initial-access",
    "execution", "persistence", "privilege-escalation",
    "defense-evasion", "credential-access", "discovery",
    "lateral-movement", "collection", "command-and-control",
    "exfiltration", "impact",
]

# Tactic-to-technique mapping for common techniques (subset for illustration)
TACTIC_TECHNIQUE_MAP = {
    "execution": [
        "T1059", "T1059.001", "T1059.003", "T1059.004", "T1059.005",
        "T1059.006", "T1059.007", "T1047", "T1053", "T1053.005",
        "T1129", "T1203", "T1569", "T1569.002",
    ],
    "persistence": [
        "T1547", "T1547.001", "T1547.004", "T1547.009",
        "T1053.005", "T1136", "T1136.001", "T1543", "T1543.003",
        "T1546", "T1546.001", "T1546.003", "T1574", "T1574.001",
        "T1197", "T1505", "T1505.003",
    ],
    "credential-access": [
        "T1003", "T1003.001", "T1003.002", "T1003.003",
        "T1003.004", "T1003.005", "T1003.006",
        "T1110", "T1110.001", "T1110.003",
        "T1555", "T1555.003", "T1552", "T1552.001",
        "T1558", "T1558.003",
    ],
    "defense-evasion": [
        "T1070", "T1070.001", "T1070.004",
        "T1218", "T1218.001", "T1218.003", "T1218.005",
        "T1218.010", "T1218.011",
        "T1027", "T1140", "T1562", "T1562.001",
        "T1036", "T1036.005",
    ],
    "discovery": [
        "T1082", "T1083", "T1087", "T1087.001", "T1087.002",
        "T1016", "T1049", "T1057", "T1069", "T1069.001",
        "T1069.002", "T1518", "T1518.001",
    ],
    "lateral-movement": [
        "T1021", "T1021.001", "T1021.002", "T1021.003",
        "T1021.004", "T1021.006", "T1570",
    ],
    "command-and-control": [
        "T1071", "T1071.001", "T1071.004",
        "T1105", "T1132", "T1573", "T1573.001",
        "T1219", "T1090",
    ],
    "exfiltration": [
        "T1041", "T1048", "T1048.003", "T1567",
    ],
    "impact": [
        "T1485", "T1486", "T1489", "T1490", "T1491",
    ],
}


def generate_coverage_report(atomics_inventory, execution_logs, detection_results):
    """Generate comprehensive coverage gap analysis."""
    report = {
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "summary": {},
        "tactics": {},
        "gaps": [],
        "recommendations": [],
    }

    total_available = len(atomics_inventory)
    total_executed = len(execution_logs)
    total_detected = sum(1 for d in detection_results.values() if d.get("detected"))

    report["summary"] = {
        "total_techniques_with_atomics": total_available,
        "total_techniques_executed": total_executed,
        "total_techniques_detected": total_detected,
        "execution_coverage_pct": round(
            (total_executed / total_available * 100) if total_available else 0, 1
        ),
        "detection_coverage_pct": round(
            (total_detected / total_executed * 100) if total_executed else 0, 1
        ),
    }

    # Per-tactic analysis
    for tactic, technique_ids in TACTIC_TECHNIQUE_MAP.items():
        tactic_data = {
            "techniques_available": 0,
            "techniques_executed": 0,
            "techniques_detected": 0,
            "gaps": [],
        }

        for tech_id in technique_ids:
            if tech_id in atomics_inventory:
                tactic_data["techniques_available"] += 1

                executed = tech_id in execution_logs
                detected = detection_results.get(tech_id, {}).get("detected", False)

                if executed:
                    tactic_data["techniques_executed"] += 1
                if detected:
                    tactic_data["techniques_detected"] += 1

                if executed and not detected:
                    gap = {
                        "technique_id": tech_id,
                        "technique_name": atomics_inventory[tech_id]["name"],
                        "tactic": tactic,
                        "status": "BLIND_SPOT",
                        "detail": "Test executed but no detection triggered",
                    }
                    tactic_data["gaps"].append(gap)
                    report["gaps"].append(gap)
                elif not executed and tech_id in atomics_inventory:
                    gap = {
                        "technique_id": tech_id,
                        "technique_name": atomics_inventory[tech_id]["name"],
                        "tactic": tactic,
                        "status": "NOT_TESTED",
                        "detail": "Atomic test available but not yet executed",
                    }
                    tactic_data["gaps"].append(gap)

        avail = tactic_data["techniques_available"]
        tactic_data["coverage_pct"] = round(
            (tactic_data["techniques_detected"] / avail * 100) if avail else 0, 1
        )
        report["tactics"][tactic] = tactic_data

    # Generate prioritized recommendations
    blind_spots = [g for g in report["gaps"] if g["status"] == "BLIND_SPOT"]
    if blind_spots:
        report["recommendations"].append({
            "priority": "CRITICAL",
            "action": f"Write detection rules for {len(blind_spots)} blind spot techniques",
            "techniques": [g["technique_id"] for g in blind_spots],
        })

    low_coverage_tactics = [
        t for t, d in report["tactics"].items() if d["coverage_pct"] < 30
    ]
    if low_coverage_tactics:
        report["recommendations"].append({
            "priority": "HIGH",
            "action": f"Expand testing in low-coverage tactics: {', '.join(low_coverage_tactics)}",
            "detail": "These tactics have less than 30% detection coverage",
        })

    return report


def generate_navigator_layer(atomics_inventory, execution_logs, detection_results,
                             layer_name="Purple Team Coverage"):
    """Generate ATT&CK Navigator layer JSON for heatmap visualization."""
    layer = {
        "name": layer_name,
        "versions": {
            "attack": "15",
            "navigator": "5.1",
            "layer": "4.5",
        },
        "domain": "enterprise-attack",
        "description": f"Purple team atomic testing coverage - Generated {datetime.utcnow().isoformat()}Z",
        "filters": {"platforms": ["Windows", "Linux", "macOS"]},
        "sorting": 0,
        "layout": {
            "layout": "side",
            "aggregateFunction": "average",
            "showID": True,
            "showName": True,
        },
        "hideDisabled": False,
        "techniques": [],
        "gradient": {
            "colors": ["#ff6666", "#ffeb3b", "#66bb6a"],
            "minValue": 0,
            "maxValue": 100,
        },
        "legendItems": [
            {"label": "No Coverage (Blind Spot)", "color": "#ff6666"},
            {"label": "Logged Only (Partial)", "color": "#ffeb3b"},
            {"label": "Alert/Detection Active", "color": "#66bb6a"},
            {"label": "Not Tested", "color": "#d3d3d3"},
        ],
        "metadata": [],
        "links": [],
        "showTacticRowBackground": True,
        "tacticRowBackground": "#dddddd",
        "selectTechniquesAcrossTactics": True,
        "selectSubtechniquesWithParent": False,
    }

    for tech_id, tech_data in atomics_inventory.items():
        executed = tech_id in execution_logs
        detection = detection_results.get(tech_id, {})
        detected = detection.get("detected", False)
        confidence = detection.get("confidence", "none")

        if detected and confidence in ("high", "medium"):
            score = 100
            color = "#66bb6a"  # Green - high confidence detection
            comment = f"DETECTED - {detection.get('rule_name', 'Alert active')}"
        elif detected:
            score = 50
            color = "#ffeb3b"  # Yellow - logged/partial
            comment = "PARTIAL - Detection exists but low confidence"
        elif executed:
            score = 0
            color = "#ff6666"  # Red - blind spot
            comment = "BLIND SPOT - Test executed, no detection"
        else:
            score = 0
            color = "#d3d3d3"  # Gray - not tested
            comment = f"NOT TESTED - {tech_data['test_count']} atomic tests available"

        technique_entry = {
            "techniqueID": tech_id,
            "tactic": "",
            "color": color,
            "comment": comment,
            "score": score,
            "enabled": True,
            "metadata": [
                {"name": "tests_available", "value": str(tech_data["test_count"])},
                {"name": "executed", "value": str(executed)},
                {"name": "detected", "value": str(detected)},
            ],
            "links": [],
            "showSubtechniques": False,
        }
        layer["techniques"].append(technique_entry)

    return layer


def print_coverage_report(report):
    """Print formatted coverage report to console."""
    print("=" * 72)
    print("PURPLE TEAM ATOMIC TESTING - COVERAGE GAP ANALYSIS")
    print("=" * 72)
    print(f"Generated: {report['generated_at']}")
    print()

    s = report["summary"]
    print("EXECUTIVE SUMMARY")
    print("-" * 40)
    print(f"  Techniques with atomics:  {s['total_techniques_with_atomics']}")
    print(f"  Techniques executed:      {s['total_techniques_executed']}")
    print(f"  Techniques detected:      {s['total_techniques_detected']}")
    print(f"  Execution coverage:       {s['execution_coverage_pct']}%")
    print(f"  Detection coverage:       {s['detection_coverage_pct']}%")
    print()

    print("PER-TACTIC COVERAGE")
    print("-" * 72)
    print(f"{'Tactic':<25} {'Available':>9} {'Executed':>9} {'Detected':>9} {'Coverage':>9}")
    print("-" * 72)
    for tactic in TACTIC_ORDER:
        if tactic in report["tactics"]:
            t = report["tactics"][tactic]
            bar = "#" * int(t["coverage_pct"] / 5) + "." * (20 - int(t["coverage_pct"] / 5))
            print(
                f"  {tactic:<23} {t['techniques_available']:>9} "
                f"{t['techniques_executed']:>9} {t['techniques_detected']:>9} "
                f"{t['coverage_pct']:>8.1f}%"
            )
    print()

    blind_spots = [g for g in report["gaps"] if g["status"] == "BLIND_SPOT"]
    if blind_spots:
        print("CRITICAL BLIND SPOTS (executed but not detected)")
        print("-" * 72)
        for gap in blind_spots:
            print(f"  [!] {gap['technique_id']} - {gap['technique_name']}")
            print(f"      Tactic: {gap['tactic']}")
        print()

    if report["recommendations"]:
        print("RECOMMENDATIONS")
        print("-" * 72)
        for rec in report["recommendations"]:
            print(f"  [{rec['priority']}] {rec['action']}")
            if "techniques" in rec:
                print(f"         Techniques: {', '.join(rec['techniques'][:10])}")
            if "detail" in rec:
                print(f"         {rec['detail']}")
        print()


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="ATT&CK coverage gap analysis for purple team testing")
    parser.add_argument("--atomics-path", default=r"C:\AtomicRedTeam\atomics",
                        help="Path to Atomic Red Team atomics directory")
    parser.add_argument("--log-dir", default=r"C:\AtomicRedTeam\logs",
                        help="Path to atomic test execution logs")
    parser.add_argument("--detections-file", default="detection_results.json",
                        help="Path to SIEM detection validation export (JSON)")
    parser.add_argument("--output-layer", default="navigator_layer.json",
                        help="Output path for ATT&CK Navigator layer JSON")
    parser.add_argument("--output-report", default="coverage_report.json",
                        help="Output path for coverage report JSON")
    args = parser.parse_args()

    print("[*] Loading atomics inventory...")
    inventory = load_atomics_inventory(args.atomics_path)
    print(f"    Found {len(inventory)} techniques with atomic tests")

    print("[*] Loading execution logs...")
    exec_logs = load_execution_logs(args.log_dir)
    print(f"    Found logs for {len(exec_logs)} techniques")

    print("[*] Loading detection results...")
    det_results = load_detection_results(args.detections_file)
    print(f"    Found detection data for {len(det_results)} techniques")

    print("[*] Generating coverage report...")
    report = generate_coverage_report(inventory, exec_logs, det_results)
    print_coverage_report(report)

    # Save report JSON
    with open(args.output_report, "w") as f:
        json.dump(report, f, indent=2)
    print(f"[+] Report saved to {args.output_report}")

    # Generate and save Navigator layer
    print("[*] Generating ATT&CK Navigator layer...")
    layer = generate_navigator_layer(inventory, exec_logs, det_results)
    with open(args.output_layer, "w") as f:
        json.dump(layer, f, indent=2)
    print(f"[+] Navigator layer saved to {args.output_layer}")
    print(f"    Import at: https://mitre-attack.github.io/attack-navigator/")

步骤 6：运行持续原子测试流水线

针对优先技术安排周期性测试：

# Continuous Atomic Testing - scheduled execution with validation
# Run as a scheduled task or via CI/CD pipeline

$PriorityTechniques = @(
    # Top MITRE ATT&CK techniques by prevalence (Red Canary Threat Detection Report)
    @{ Id = "T1059.001"; Name = "PowerShell" },
    @{ Id = "T1059.003"; Name = "Windows Command Shell" },
    @{ Id = "T1547.001"; Name = "Registry Run Keys" },
    @{ Id = "T1053.005"; Name = "Scheduled Task" },
    @{ Id = "T1003.001"; Name = "LSASS Memory" },
    @{ Id = "T1003.003"; Name = "NTDS" },
    @{ Id = "T1070.004"; Name = "File Deletion" },
    @{ Id = "T1218.011"; Name = "Rundll32" },
    @{ Id = "T1082";     Name = "System Information Discovery" },
    @{ Id = "T1105";     Name = "Ingress Tool Transfer" }
)

$ResultsLog = @()
$RunTimestamp = Get-Date -Format "yyyyMMdd_HHmmss"

foreach ($technique in $PriorityTechniques) {
    Write-Host "`n[*] Testing $($technique.Id) - $($technique.Name)" -ForegroundColor Cyan

    # Check prerequisites
    try {
        $prereqs = Invoke-AtomicTest $technique.Id -TestNumbers 1 -CheckPrereqs 2>&1
        $prereqsMet = $prereqs -notmatch "not met"
    } catch {
        $prereqsMet = $false
    }

    if (-not $prereqsMet) {
        Write-Host "    [!] Prerequisites not met, attempting install..." -ForegroundColor Yellow
        try {
            Invoke-AtomicTest $technique.Id -TestNumbers 1 -GetPrereqs 2>&1 | Out-Null
        } catch {
            Write-Host "    [!] Prereq install failed, skipping" -ForegroundColor Red
            $ResultsLog += [PSCustomObject]@{
                TechniqueId = $technique.Id
                Name        = $technique.Name
                Status      = "PREREQS_FAILED"
                Timestamp   = (Get-Date).ToString("o")
            }
            continue
        }
    }

    # Execute the test
    try {
        $startTime = Get-Date
        Invoke-AtomicTest $technique.Id -TestNumbers 1 -Confirm:$false
        $endTime = Get-Date
        $duration = ($endTime - $startTime).TotalSeconds

        Write-Host "    [+] Executed successfully (${duration}s)" -ForegroundColor Green

        $ResultsLog += [PSCustomObject]@{
            TechniqueId = $technique.Id
            Name        = $technique.Name
            Status      = "EXECUTED"
            Duration    = $duration
            Timestamp   = $startTime.ToString("o")
        }
    } catch {
        Write-Host "    [-] Execution failed: $($_.Exception.Message)" -ForegroundColor Red
        $ResultsLog += [PSCustomObject]@{
            TechniqueId = $technique.Id
            Name        = $technique.Name
            Status      = "FAILED"
            Error       = $_.Exception.Message
            Timestamp   = (Get-Date).ToString("o")
        }
    }

    # Cleanup
    try {
        Invoke-AtomicTest $technique.Id -TestNumbers 1 -Cleanup 2>&1 | Out-Null
        Write-Host "    [+] Cleanup completed" -ForegroundColor DarkGray
    } catch {
        Write-Host "    [!] Cleanup failed" -ForegroundColor Yellow
    }

    # Brief pause between tests for log ingestion
    Start-Sleep -Seconds 10
}

# Export results
$ResultsLog | Export-Csv "C:\AtomicRedTeam\logs\continuous_test_$RunTimestamp.csv" -NoTypeInformation
$ResultsLog | Format-Table -AutoSize

# Summary
$executed = ($ResultsLog | Where-Object Status -eq "EXECUTED").Count
$failed = ($ResultsLog | Where-Object Status -ne "EXECUTED").Count
Write-Host "`n=== CONTINUOUS TEST SUMMARY ===" -ForegroundColor Cyan
Write-Host "  Executed: $executed / $($PriorityTechniques.Count)" -ForegroundColor Green
Write-Host "  Failed:   $failed / $($PriorityTechniques.Count)" -ForegroundColor Red

步骤 7：检测验证闭环

通过将测试与检测关联并迭代，闭合紫队闭环：

Detection Validation Loop Workflow:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  ┌──────────────────┐
  │ 1. SELECT        │  Choose technique from threat intel or gap report
  │    TECHNIQUE     │  Map to Atomic Red Team test ID
  └───────┬──────────┘
          │
  ┌───────▼──────────┐
  │ 2. EXECUTE       │  Run Invoke-AtomicTest with logging
  │    ATOMIC TEST   │  Record timestamp, hostname, test details
  └───────┬──────────┘
          │
  ┌───────▼──────────┐
  │ 3. WAIT FOR      │  Allow 30-60 seconds for log ingestion
  │    INGESTION     │  Ensure Sysmon, WinEventLog, EDR agents forward data
  └───────┬──────────┘
          │
  ┌───────▼──────────┐
  │ 4. QUERY SIEM    │  Search for correlated alerts matching technique
  │    FOR ALERTS    │  Check: Did our detection rules fire?
  └───────┬──────────┘
          │
     ┌────┴────┐
     │DETECTED?│
     └────┬────┘
    YES   │   NO
  ┌───────▼──────────┐   ┌──────────────────┐
  │ 5a. MARK GREEN   │   │ 5b. WRITE NEW    │
  │  Update Navigator│   │  Sigma/SIEM Rule │
  │  layer, log pass │   │  for technique   │
  └───────┬──────────┘   └───────┬──────────┘
          │                       │
          │               ┌───────▼──────────┐
          │               │ 5c. DEPLOY RULE  │
          │               │  Push to SIEM    │
          │               └───────┬──────────┘
          │                       │
          │               ┌───────▼──────────┐
          │               │ 5d. RE-EXECUTE   │
          │               │  Re-run atomic   │
          │               │  test, validate  │
          │               └───────┬──────────┘
          │                       │
  ┌───────▼───────────────────────▼──┐
  │ 6. CLEANUP                       │
  │    Run Invoke-AtomicTest -Cleanup│
  │    Document results              │
  └───────┬──────────────────────────┘
          │
  ┌───────▼──────────┐
  │ 7. NEXT          │  Select next technique
  │    TECHNIQUE     │  Repeat from step 1
  └──────────────────┘

关键概念

术语	定义
Atomic Red Team	由 Red Canary 维护的开源库，包含映射到 MITRE ATT&CK 技术的细粒度测试，用于验证检测能力
Invoke-AtomicRedTeam	PowerShell 执行框架，用于在本地或远程运行原子测试，并支持前置条件检查和清理
ATT&CK Navigator	基于 Web 的工具，用于将 MITRE ATT&CK 矩阵标注并可视化为展示检测覆盖情况的分层热力图
Navigator Layer	定义每种 ATT&CK 技术对应颜色、分值和注释的 JSON 文件；用于生成覆盖热力图
检测验证循环	迭代式紫队流程：执行攻击、检查是否检测到、若未检测到则编写/调优规则、重新执行以确认
覆盖缺口	不存在检测规则的 ATT&CK 技术，或该技术执行时规则未能触发的 ATT&CK 技术
盲点	通过原子执行测试过但未产生任何 SIEM 告警的技术，表明存在严重检测失败
Sigma 规则	厂商无关的检测规则格式，可转换为 Splunk SPL、Elastic KQL、Microsoft KQL 及其他 SIEM 查询
紫队	红队执行已知技术、蓝队实时验证并改进检测的协作式安全演练
持续原子测试	按计划自动执行原子测试，以验证检测规则在环境变化时仍能正常工作

工具与系统

• Invoke-AtomicRedTeam：用于执行 Atomic Red Team 测试的 PowerShell 模块，支持前置条件管理和清理
• Atomic Red Team Atomics：包含 200 多种 MITRE ATT&CK 技术的 700 多个测试的库，按技术 ID 组织
• ATT&CK Navigator：位于 mitre-attack.github.io/attack-navigator 的 Web 工具，用于创建技术覆盖热力图
• VECTR：由 SCYTHE 开发的紫队报告平台，用于跟踪测试活动、结果和长期覆盖情况
• Sigma：面向 SIEM 系统的通用签名格式；可使用 sigmac 或 pySigma 转换为目标 SIEM 平台
• Splunk / Elastic / Microsoft Sentinel：用于在原子测试执行后查询检测结果的 SIEM 平台
• Sysmon：Windows 系统监控驱动，提供详细的进程、网络、注册表和文件遥测
• Caldera：由 MITRE 开发的对手仿真平台，用于超越单一原子测试的自动化多步骤攻击链

常见场景

场景：在 EDR 部署后验证凭据访问检测

背景：组织部署了新的 EDR Agent，需要验证其能够检测凭据转储技术（LSASS 访问、NTDS 提取、SAM 转储）。紫队必须在关闭部署工单前确认检测正常工作。

方法：

1. 在已注册到新 EDR 的测试端点上安装 Invoke-AtomicRedTeam
2. 执行凭据访问原子测试：T1003.001（LSASS 内存）、T1003.002（SAM）、T1003.003（NTDS）
3. 等待 60 秒，然后在 EDR 控制台和 SIEM 中查询测试主机名相关告警
4. 对每种技术：在 Navigator layer 中标记为已检测（绿色）或盲点（红色）
5. 对盲点，创建针对 Sysmon EventCode 10（对 lsass.exe 的 ProcessAccess）和 EventCode 1（已知转储工具）的 Sigma 规则
6. 部署规则，重新运行原子测试，确认检测
7. 导出 Navigator layer 和覆盖报告，用于部署工单

陷阱：

• 在没有本地管理员权限的情况下运行凭据转储原子测试（测试会静默失败）
• 忘记启用 Sysmon ProcessAccess 日志记录（EventCode 10），该功能默认禁用
• 未运行清理，留下测试残留物（创建的账户、注册表项、计划任务）
• 只测试 T1003 的一个子技术，并假设所有凭据访问都已覆盖
• 未考虑 EDR Agent 排除项，这些排除项可能将 PowerShell 或 AtomicRedTeam 目录加入白名单

场景：为管理层构建每月 ATT&CK 覆盖率报告

背景：CISO 要求每月提供映射到 MITRE ATT&CK 的检测覆盖率指标。安全团队需要运行周期性测试、跟踪改进情况，并生成一张可视化热力图，展示覆盖率随时间的增长。

方法：

1. 制定一个测试计划，覆盖 Red Canary Threat Detection Report 中的前 50 个 ATT&CK 技术
2. 通过 Windows Task Scheduler 安排持续原子测试，每周运行 PowerShell 管道
3. 每次运行后将执行日志和 SIEM 检测查询结果导出为 JSON
4. 运行 Python 差距分析脚本，生成 Navigator 层和覆盖率报告
5. 将 Navigator 层导入 ATT&CK Navigator，截图用于高管幻灯片
6. 逐月跟踪：绿色（已检测）与红色（盲区）技术的数量
7. 将检测工程冲刺优先安排在高影响盲区上

陷阱：

• 将"已记录"当作"已检测"（在日志中看到事件并不等于有告警规则）
• 未更新 atomics 仓库（新测试会频繁添加）
• 只关注技术数量而非检测质量（一条 90% 误报的嘈杂规则比没有规则更糟）
• 未让将负责告警分类的 SOC 分析师参与验证过程

输出格式

PURPLE TEAM ATOMIC TESTING REPORT
====================================
Campaign:         Q1 2026 Detection Validation
Date Range:       2026-01-15 to 2026-03-15
Test Environment: YOURLAB-WIN10-01, YOURLAB-SRV-01
SIEM:             Splunk Enterprise 9.x

EXECUTION SUMMARY
Techniques Tested:        47 / 200+ available
Individual Tests Run:     112
Tests Succeeded:          104
Tests Failed (prereqs):   8
Cleanup Completed:        104 / 104

DETECTION COVERAGE
                            Tested    Detected    Coverage
Execution:                     12          10       83.3%
Persistence:                    8           5       62.5%
Credential Access:              6           4       66.7%
Defense Evasion:               10           3       30.0%
Discovery:                      5           5      100.0%
Lateral Movement:               3           1       33.3%
Command and Control:            2           1       50.0%
Exfiltration:                   1           0        0.0%

CRITICAL BLIND SPOTS
[!] T1218.011 - Rundll32          (defense-evasion)
[!] T1218.005 - Mshta             (defense-evasion)
[!] T1070.004 - File Deletion     (defense-evasion)
[!] T1574.001 - DLL Search Order  (persistence)
[!] T1021.002 - SMB/Admin Shares  (lateral-movement)
[!] T1048.003 - Exfil Over HTTP   (exfiltration)

NAVIGATOR LAYER: coverage_layer_20260315.json
  Import at: https://mitre-attack.github.io/attack-navigator/

RECOMMENDATIONS
[CRITICAL] Write Sigma rules for 6 blind spot techniques
[HIGH]     Expand defense-evasion testing (30% coverage)
[HIGH]     Enable exfiltration monitoring data sources
[MEDIUM]   Increase lateral movement test coverage (3 of 7 tested)