Automated Adversary Emulation: Building a Repeatable Penetration Testing Framework

The Execution Engine

The engine reads plans, enforces safety constraints, executes each phase, and records results:

import yaml
import ipaddress
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

class PhaseStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"
    ABORTED = "aborted"

@dataclass
class EmulationResult:
    phase_id: str
    technique_id: str
    status: PhaseStatus
    started_at: datetime
    completed_at: Optional[datetime] = None
    output: str = ""
    detection_observed: bool = False
    alert_ids: list = field(default_factory=list)

class SafetyController:
    """Enforces scope limits and de-escalation triggers."""

    def __init__(self, safety_config: dict):
        self.allowed_networks = [
            ipaddress.ip_network(n) for n in safety_config["scope"]["networks"]
        ]
        self.excluded_hosts = set(safety_config["scope"].get("exclude_hosts", []))
        self.max_duration = timedelta(
            minutes=safety_config.get("max_duration_minutes", 30)
        )
        self.triggers = safety_config.get("deescalation_triggers", [])
        self.start_time = None
        self.host_failures = 0

    def validate_target(self, target_ip: str) -> bool:
        addr = ipaddress.ip_address(target_ip)
        if target_ip in self.excluded_hosts:
            logging.warning(f"Target {target_ip} is explicitly excluded")
            return False
        return any(addr in net for net in self.allowed_networks)

    def check_triggers(self, context: dict) -> Optional[str]:
        if self.start_time and datetime.now() - self.start_time > self.max_duration:
            return "abort"
        if context.get("host_unreachable_count", 0) > 3:
            return "abort"
        if context.get("privilege_level") == "SYSTEM":
            return "pause_and_notify"
        return None

class EmulationEngine:
    def __init__(self, plan_path: str):
        with open(plan_path) as f:
            self.plan = yaml.safe_load(f)["plan"]
        self.safety = SafetyController(self.plan["safety"])
        self.results: list[EmulationResult] = []
        self.context = {"host_unreachable_count": 0}

    def execute(self):
        self.safety.start_time = datetime.now()
        logging.info(f"Starting emulation: {self.plan['name']}")

        for phase in self.plan["phases"]:
            # Check safety before each phase
            action = self.safety.check_triggers(self.context)
            if action == "abort":
                logging.critical("Safety trigger fired — aborting emulation")
                self._record_abort(phase)
                break
            elif action == "pause_and_notify":
                logging.warning("Elevated privilege detected — pausing")
                if not self._operator_confirms_continue():
                    break

            # Check dependencies
            if not self._dependencies_met(phase):
                self._record_skip(phase)
                continue

            result = self._execute_phase(phase)
            self.results.append(result)

        return self._generate_report()

Safety Controls Are Non-Negotiable

The safety layer deserves its own section because getting it wrong means turning your red team exercise into an actual incident. Every emulation framework must enforce:

Scope locking. The engine must refuse to touch any host outside the defined network ranges. This is not a suggestion — it is a hard block in the execution path. Validate every target IP before any packet leaves.

Time boxing. Set maximum durations per plan and per phase. If an attack chain stalls and an operator walks away, the framework should terminate gracefully after the timeout.

De-escalation triggers. Define conditions that automatically pause or abort execution. Achieving SYSTEM or root privileges unexpectedly, hitting too many unreachable hosts (indicating network issues), or detecting production workload disruption should all halt the run.

Kill switch. Implement an out-of-band kill mechanism — a file on a shared path, an API endpoint, or a message queue topic. If anything goes wrong, any team member should be able to stop all running emulations instantly:

KILL_SWITCH_PATH = "/opt/emulation/EMERGENCY_STOP"

def check_kill_switch() -> bool:
    return os.path.exists(KILL_SWITCH_PATH)

# Check before every network operation
if check_kill_switch():
    raise EmulationAborted("Kill switch activated")

Detection Validation Integration

Running attacks without checking whether your defenses saw them is only half the exercise. Integrate with your SIEM to validate detection coverage:

import requests
from time import sleep

class DetectionValidator:
    """Query SIEM for alerts generated during emulation phases."""

    def __init__(self, siem_url: str, api_key: str):
        self.siem_url = siem_url
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def check_detection(self, technique_id: str,
                         time_start: datetime,
                         time_end: datetime,
                         target_host: str) -> dict:
        # Wait for SIEM ingestion lag
        sleep(30)

        query = {
            "query": f"mitre.technique_id:{technique_id} AND host:{target_host}",
            "time_range": {
                "from": time_start.isoformat(),
                "to": time_end.isoformat()
            }
        }

        resp = requests.post(
            f"{self.siem_url}/api/search",
            json=query,
            headers=self.headers
        )

        alerts = resp.json().get("hits", [])
        return {
            "detected": len(alerts) > 0,
            "alert_count": len(alerts),
            "alert_ids": [a["id"] for a in alerts],
            "mean_time_to_detect": self._calc_mttd(alerts, time_start)
        }

Structured Reporting and Vulnerability Management Integration

Every emulation run produces a structured JSON report that feeds into your vulnerability management pipeline: