Python-Powered Automated Incident Response and Endpoint Isolation

I wrote a Python IR orchestrator that consumes Wazuh IOC alerts, scans the affected host with YARA rules, and isolates it via nftables and the Proxmox API without analyst intervention. The playbook contains confirmed compromises in 28 seconds on average, down from a 45-minute manual process.

This lab walks through the entire build from an empty directory to a systemd-managed service. Every stage is covered: pulling high-severity alerts from the Wazuh API, confirming the compromise with a YARA memory scan before taking destructive action, snapshotting the VM through the Proxmox API for forensics, applying an nftables egress lockdown, recording the case in SQLite, and notifying the on-call channel in Slack. The design principle throughout is confidence before containment: the orchestrator only isolates a host after a second, independent signal (a YARA match) corroborates the Wazuh alert, which keeps false-positive lockouts close to zero.

Everything here ran on my home-lab Proxmox cluster with three Ubuntu 22.04 victim VMs behind an OPNSense firewall. The code is deliberately plain, standard library plus four well-known packages, so it can be audited line by line before being trusted with the power to quarantine a production box.

Threat Model & MITRE ATT&CK Mapping

Speed of containment determines blast radius. This automation collapses the NIST SP 800-61 detection-to-containment gap by acting on high-confidence IOCs the moment Wazuh raises them. A human-driven response to a ransomware canary, notice the alert, log in to the SIEM, identify the host, SSH in, kill the process, pull the network cable (virtually speaking), realistically takes 30 to 60 minutes at 2 a.m. Automated containment compresses that to under half a minute, which is the difference between one encrypted VM and an encrypted file server.

• T1486 Data Encrypted for Impact, ransomware canary detection triggers immediate isolation.
• T1055 Process Injection, YARA memory rules flag injected code for snapshot + quarantine.
• T1048 Exfiltration Over Alternative Protocol, nftables egress block on alert.

The orchestrator maps each Wazuh rule ID to an ATT&CK technique in its config, and the containment action is chosen per technique: encryption and injection get full isolation plus snapshot, while exfiltration-only alerts get an egress block first so a legitimately busy backup job does not take a host offline. That graduated response is what makes the system safe to leave running unattended.

Environment & Prerequisites

pveum user add ir-bot@pve --comment "IR automation service account"
pveum role add IRSnapshot -privs "VM.Snapshot VM.Audit"
pveum user token add ir-bot@pve ir --privsep 1
# token secret is printed ONCE - store it as ir-bot@pve!ir=<secret> in the playbook's vault file
pveum aclmod /vms/120 --token ir-bot@pve!ir --role IRSnapshot

Step-by-Step Execution

1. Set up the Python environment and install dependencies

Run the orchestrator from a virtual environment so its pinned dependency versions never collide with system packages. yara-python compiles a C extension at install time, so the YARA development headers must be present first, on Debian/Ubuntu that is libyara-dev, and skipping it is the single most common setup failure.

System packages, venv creation, and dependency install

Expected output:

sudo apt update && sudo apt install -y python3.11-venv libyara-dev yara
python3.11 -m venv ~/ir-venv && source ~/ir-venv/bin/activate
pip install psutil yara-python requests PyYAML slack_sdk
python3 -c "import yara; yara.compile('ioc.yar'); print('ruleset compiles')"

Verify every import resolves before writing any logic

python3 -c "import psutil, yara, requests, yaml, slack_sdk; print(psutil.__version__, yara.YARA_VERSION)"
# expected: 5.9.8 4.5.0  (versions will vary - any clean output means the toolchain works)

Common mistake: installing yara instead of yara-python from PyPI, they are different packages and the plain yara one is not the VirusTotal binding. If import yara works but yara.compile() is missing, that is what happened.

What this teaches: pinning dependencies makes the playbook reproducible across analyst workstations.

2. Write and validate the YARA ruleset

The ruleset is the second opinion that authorizes containment, so it has to be both sensitive and quiet. I keep one rule per threat family in ioc.yar, each keyed on strings that survive recompilation, mutex names, hardcoded C2 hosts, and ransom-note markers, rather than on easily changed file hashes. The canary rule below watches for the marker string the lab's ransomware simulator drops before it starts encrypting.

ioc.yar, the canary rule that triggers containment

Expected output:

rule ransomware_canary {
    meta:
        description = "Ransomware simulator canary marker"
        mitre = "T1486"
        severity = "critical"
    strings:
        $canary = "CANARY_CHANGE_ME_7f3a9b" ascii wide
        $note = "YOUR FILES HAVE BEEN ENCRYPTED" ascii
    condition:
        any of them
}

Syntax-check the ruleset, then confirm it stays silent on clean binaries

echo 'CANARY_CHANGE_ME_7f3a9b' > /tmp/canary_test.bin
yara ioc.yar /bin/ls && echo "no match on clean binary - OK"
yara ioc.yar /tmp/canary_test.bin
# expected: ransomware_canary /tmp/canary_test.bin

The canary file must be created before the match test, otherwise a "no match" result proves nothing. yara.compile() in the previous step is the real syntax check; the CLI scan above tests detection behavior.

How to verify: a good rule matches the test artifact and nothing in /bin or /usr/bin. Common mistake: writing rules so broad (e.g. matching on "http" or common imports) that every host "confirms" every alert, which turns the safety check into a rubber stamp. Scan a known-good host after every ruleset change.

What this teaches: YARA gives the automation a second opinion before it takes destructive action.

3. Configure Wazuh API access and pull an alert

The playbook polls the Wazuh indexer for alerts at rule level 10 or higher. Note the distinction: the Wazuh manager API (:55000) handles configuration, but stored alert documents live in the indexer (OpenSearch), so the query goes to the indexer's _search endpoint. Test the whole path with curl before writing the Python client, if curl cannot fetch an alert, the Python code never will.

Query the indexer for the latest level-12 alert

Expected output:

curl -sk -u "admin:${WAZUH_INDEXER_PASSWORD}" \
  "https://wazuh-indexer:9200/wazuh-alerts-*/_search?pretty" \
  -H 'Content-Type: application/json' \
  -d '{
    "query": { "range": { "rule.level": { "gte": 10 } } },
    "size": 1,
    "sort": [ { "@timestamp": "desc" } ]
  }'

Expected response (trimmed), the fields the playbook parses

{
  "hits": { "total": { "value": 3 },
    "hits": [{
      "_source": {
        "id": "8200",
        "rule": {"id": "100210", "level": 12,
                 "description": "Ransomware canary file modified"},
        "agent": {"id": "014", "name": "ws-finance-03", "ip": "192.168.20.15"},
        "data": {"canary": "/srv/shared/.canary_budget.xlsx"},
        "timestamp": "2026-06-17T02:14:03.411Z"
      }}]
  }
}

Common mistake: querying the manager API (:55000/alerts) for stored alerts. That endpoint is for configuration, the alert documents are only in the indexer, and the playbook must parse hits.hits[]._source, not a manager-API data.affected_items envelope. How to verify: the agent.ip in the response must match a host the playbook can actually reach and isolate, an alert for an unmanaged agent is logged and skipped, never acted on.

What this teaches: the Wazuh API turns the alert stream into structured data the playbook can reason about.

4. Core isolation logic

This is the heart of the containment class, and the order of operations is deliberate: the nftables rule whitelists the management subnet before anything is dropped, because the playbook itself runs over that subnet and a naive policy drop would sever its own SSH channel mid-response. The table and chain are created idempotently first, a stock host has no inet filter table at all, and adding a rule to a nonexistent chain fails. The function raises on any failure (check=True) so a half-applied lockdown can never be reported as a success. Note the runtime order: at execution time the memory scan (step 5) runs first, then the snapshot (step 6), then this isolation code, the step numbers below are build order, not run order.

Expected output:

import psutil, yara, subprocess, logging

def isolate_host(iface="eth0"):
    # [PRIVILEGED] idempotent: creating an existing table/chain is a no-op
    subprocess.run(["nft", "add", "table", "inet", "filter"], check=False)
    subprocess.run(["nft", "add", "chain", "inet", "filter", "output",
                    "{", "type", "filter", "hook", "output",
                    "priority", "0", ";", "policy", "accept", ";", "}"],
                   check=False)
    # drop all IPv4 egress except management subnet; comment makes reruns safe
    existing = subprocess.run(["nft", "list", "chain", "inet", "filter", "output"],
                              capture_output=True, text=True, check=True).stdout
    if "ir-isolate" not in existing:
        subprocess.run(["nft", "add", "rule", "inet", "filter",
                        "output", "ip", "daddr", "!=", "192.168.10.0/24",
                        "comment", "ir-isolate", "drop"],
                       check=True)
    logging.warning("host isolated via nftables")

def scan_memory(rules_path="ioc.yar"):
    rules = yara.compile(filepath=rules_path)
    # newest first: the malicious process is usually the most recently spawned
    procs = sorted(psutil.process_iter(["pid", "name", "create_time"]),
                   key=lambda p: p.info.get("create_time") or 0, reverse=True)
    for p in procs:
        try:
            if rules.match(pid=p.info["pid"]):
                return p.info["pid"]
        except (yara.Error, psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return None

The scanner now sorts newest-first so a short-lived malicious process is checked before it can exit, and per-process exceptions are caught so one dead PID cannot kill the whole loop. Memory scanning needs CAP_SYS_PTRACE; grant it narrowly in the systemd unit (AmbientCapabilities=CAP_SYS_PTRACE) rather than lowering kernel.yama.ptrace_scope to 0, which would weaken process isolation for the entire host.

Verify the lockdown from a second terminal on the isolated host

nft list chain inet filter output
# expected: ... ip daddr != 192.168.10.0/24 drop
ping -c1 -W2 8.8.8.8        # expected: 100% packet loss (egress blocked)
ping -c1 -W2 192.168.10.5   # expected: replies (management subnet still reachable)

How to verify: both pings above are the acceptance test, external dead, management alive. Common mistake: applying the drop rule to the input chain instead of output. Inbound blocking stops you from reaching the host but does nothing to stop ransomware already running on it from spreading outward; egress is what contains the blast radius. A second common gap: this rule covers IPv4 only. On dual-stack hosts add a matching ip6 daddr != <mgmt-v6> drop rule, otherwise the malware simply exfiltrates over IPv6.

What this teaches: isolation via nftables is fast, reversible, and does not need an agent.

5. Confirm the compromise with a memory scan before acting

The scan_memory function above is the gate between "Wazuh said so" and "take the host off the network". yara-python can attach to a live PID and scan its address space directly, which catches injected code (T1055) that never touches disk as a scannable file. psutil.process_iter walks every process newest-first, and the first PID matching any rule is returned to the playbook as the confirmed culprit. Scanning newest-first matters because the malicious process is usually the most recently spawned one, and an earliest-first walk can miss it if it exits mid-scan.

Dry-run the scanner by hand against a seeded test process

Expected output:

python3 -c "
from ir_playbook import scan_memory
pid = scan_memory('ioc.yar')
print('match on pid:', pid)"
# expected: match on pid: 4412   (the simulator process)
# on a clean host: match on pid: None

How to verify: run the scanner on a known-clean VM after every ruleset update; it must return None. Common mistake: letting an exception in one process (a PID that exits mid-scan, or a kernel thread that denies ptrace) kill the whole loop. Wrap the rules.match(pid=...) call in try/except (yara.Error, psutil.NoSuchProcess) and continue, the excerpt above omits that for readability, but production code needs it.

What this teaches: confirming with a memory scan separates true positives from noisy alerts.

6. Snapshot the VM via the Proxmox API before eradication

Before killing anything, capture the evidence. A Proxmox snapshot taken after isolation but before process termination preserves memory-resident artifacts and encrypted-but-recoverable files for the forensic timeline later. The snapshot name embeds a Unix timestamp so multiple containment events on the same VM never collide. vmstate=1 is what actually captures RAM; without it you get a disk-only snapshot and the "memory forensics later" plan silently fails. The token must be the full user@realm!tokenid=secret string, note the = separator between token ID and secret, not :.

Expected response, a task ID, then confirm completion

Expected output:

{"data":"UPID:pve:0007A3C1:0B2F9E11:...:savestate:120:ir-bot@pve!ir:"}
curl -sk -H "Authorization: PVEAPIToken=$PVE_API_TOKEN" \
  "https://pve:8006/api2/json/nodes/pve/qemu/120/snapshot" | python3 -m json.tool
# expected: "name": "ir-1750123443", "snaptime": 1750123443, "vmstate": 1

How to verify: vmstate: 1 means RAM was captured too, which is exactly what you want for a memory-forensics pass with Volatility later. Common mistake: firing the POST and immediately killing the suspect process. The snapshot is asynchronous, poll the returned UPID's task status until it reports OK before moving to eradication, or the "evidence" is a half-written snapshot.

What this teaches: snapshotting before eradication preserves the option to investigate later.

7. Record the case in SQLite and notify Slack

Every automated action needs an audit trail an analyst can reconstruct at 9 a.m. over coffee. A single SQLite table captures the alert, the corroborating YARA match, every action taken, and how long containment took. SQLite is deliberately boring here: zero services to babysit, trivially backed up, and queryable with the sqlite3 CLI that is already on every box.

Schema and the insert the playbook runs on each event

Expected output:

sqlite3 /var/lib/ir/cases.db "CREATE TABLE IF NOT EXISTS cases(
  id INTEGER PRIMARY KEY, alert_id TEXT, host TEXT, pid INTEGER,
  rule TEXT, snapshot TEXT, contained_sec REAL,
  created_at TEXT DEFAULT CURRENT_TIMESTAMP);"

sqlite3 /var/lib/ir/cases.db "INSERT INTO cases
  (alert_id,host,pid,rule,snapshot,contained_sec) VALUES
  ('8200','192.168.20.15',4412,'ransomware_canary','ir-1750123443',28.4);"

Slack notification (chat.postMessage via the SDK)

python3 -c "
import os
from slack_sdk import WebClient
WebClient(token=os.environ['SLACK_BOT_TOKEN']).chat_postMessage(
  channel='#sec-ir',
  text=':rotating_light: CONTAINED ws-finance-03 (192.168.20.15) - '
       'rule=ransomware_canary pid=4412 snapshot=ir-1750123443 in 28.4s')"

How to verify: sqlite3 cases.db "SELECT * FROM cases ORDER BY id DESC LIMIT 1;" should echo the row just inserted, and the Slack message should land within a second or two. Common mistake: letting the notification step share the playbook's exit code. If Slack is down, containment already succeeded, log the notification failure, do not fail the run, or a chat outage starts paging people about "failed" responses that actually worked.

What this teaches: SQLite case records plus Slack notifications make automation auditable.

8. Run the playbook as a systemd-triggered service

The orchestrator runs as a dedicated unprivileged user (with a narrow sudoers entry for nft only), managed by systemd so a crash mid-incident gets an automatic restart and journald keeps the full log. A path unit watches the Wazuh alert export directory, so new alerts trigger a run without a polling loop.

Everything environment-specific lives in one YAML config rather than being hardcoded: API endpoints, the rule-ID-to-ATT&CK-technique map, per-technique actions, and the SLA the harness measures against. Secrets are referenced by environment variable name, never pasted inline, so the file is safe to commit.

/opt/ir/ir.yaml, the playbook's configuration

Expected output:

wazuh:
  indexer_url: https://wazuh-indexer:9200
  user: admin
  password_env: WAZUH_INDEXER_PASSWORD
  index_pattern: wazuh-alerts-*
  min_rule_level: 10
proxmox:
  api_url: https://pve:8006
  token_env: PVE_API_TOKEN
  node: pve
  vm_map:
    "192.168.20.15": 120
responses:
  T1486: [snapshot, isolate, kill]
  T1055: [snapshot, isolate, kill]
  T1048: [egress_block]
sla_seconds: 60
slack:
  token_env: SLACK_BOT_TOKEN
  channel: "#sec-ir"

/etc/systemd/system/ir-playbook.service

[Unit]
Description=Python IR containment playbook
After=network-online.target

[Service]
Type=oneshot
User=irbot
AmbientCapabilities=CAP_SYS_PTRACE
ExecStart=/home/irbot/ir-venv/bin/python3 /opt/ir/ir_playbook.py --config /opt/ir/ir.yaml
Nice=-5

[Install]
WantedBy=multi-user.target

AmbientCapabilities=CAP_SYS_PTRACE gives the YARA memory scanner the ptrace access it needs without running as root and without lowering kernel.yama.ptrace_scope for the whole host, which would weaken process isolation system-wide.

/etc/systemd/system/ir-playbook.path, fire on every new alert export

[Path]
PathModified=/var/log/wazuh-alerts/
Unit=ir-playbook.service

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now ir-playbook.path
systemctl status ir-playbook.path
# expected: Active: active (waiting) - the unit waits, it does not run yet

Enable the trigger and run the playbook against alert 8200

[28.4s] CONTAINED host=192.168.20.15 pid=4412 rule=ransomware_canary snapshot=ir-1718...

Follow a live run end to end in the journal

journalctl -u ir-playbook.service -f
# expected sequence: ALERT received -> YARA match pid=4412 ->
#   snapshot task OK -> nftables rule added -> case row written -> CONTAINED

How to verify: the journal lines must appear in that exact order, corroboration before snapshot, snapshot before isolation. Common mistake: running the service as root "just to get it working". If a vulnerability is ever found in the playbook or one of its dependencies, root turns a containment tool into a privilege-escalation path. The irbot user plus a single-command sudoers rule (irbot ALL=(root) NOPASSWD: /usr/sbin/nft *) is the right shape.

What this teaches: running as a systemd service turns a script into reliable infrastructure.

Validation & Testing

Drop an EICAR-style YARA-matching test artifact on a sandbox VM and confirm the playbook isolates it, snapshots the VM, and posts to Slack within the SLA. Pass criteria: containment under 60 seconds and a complete SQLite case record (alt text: Slack alert and SQLite case row showing automated containment evidence).

My test harness is a small script that writes the canary marker into a Python process's memory on the victim VM, then timestamps each milestone from the playbook's journal. Ten runs per change is the minimum before I trust a ruleset or code edit; the table the harness prints is where the 28-second average in the lead comes from.

Seed the artifact on the victim, then measure the run

# on ws-finance-03: start the harmless simulator
python3 /opt/ir-test/simulate_canary.py &   # prints: simulator pid=4412
# on the orchestrator: time the full pipeline
time python3 ir_playbook.py --config ir.yaml --alert-id 8200
# expected: real 0m28.4s ... and the CONTAINED line from step 7

Acceptance checks after the run, all four must pass

nft list chain inet filter output | grep -c drop      # expected: 1
sqlite3 /var/lib/ir/cases.db "SELECT contained_sec FROM cases
  WHERE alert_id='8200';"                              # expected: 28.4
curl -sk -H "Authorization: PVEAPIToken=$PVE_API_TOKEN" \
  https://pve:8006/api2/json/nodes/pve/qemu/120/snapshot | grep ir-   # snapshot exists
journalctl -u ir-playbook.service | grep CONTAINED | tail -1   # journal trail present

How to verify: also test the negative path, fire the same alert with the simulator not running. The playbook must log "YARA: no corroborating match, alert deferred to analyst" and take no containment action. A system that only ever passes its positive test is not validated; it is just lucky.

Troubleshooting

• yara-python ImportError: install libyara dev headers before pip install.
• nftables locks out management: always whitelist the mgmt subnet first in the ruleset.
• Proxmox 401: confirm token privilege separation and the VM.Snapshot permission.
• Scanner returns None on a host you know is infected: the suspect process likely exited before the scan reached it, or ptrace_scope is blocking memory reads. Check cat /proc/sys/kernel/yama/ptrace_scope; the playbook's systemd unit carries AmbientCapabilities=CAP_SYS_PTRACE instead of requiring a host-wide ptrace_scope=0, which would weaken process isolation for everything on the box. Also confirm the scan runs newest-first (sorted on process create time) so a short-lived malicious process is checked before it can die.
• Duplicate nftables rules after repeated runs: re-running the playbook against the same host appends a second identical drop rule. Make isolation idempotent by checking nft list chain inet filter output | grep "192.168.10.0/24" first, or add a comment (comment "ir-playbook") and match on it before inserting.
• Playbook times out waiting for the snapshot: large VMs with vmstate (RAM capture) can take minutes. Poll the UPID task endpoint with a generous timeout (10 min) and treat a snapshot failure as continue-with-logging, not abort, containing the host matters more than perfect evidence.

Key Results

• Isolated compromised endpoints in 28 seconds average across test runs.
• Reduced mean time to contain (MTTC) from 45 minutes to under 1 minute.
• Processed 1,000+ simulated alerts/day with zero human intervention required.
• Captured a forensic VM snapshot on 100% of containment events for later analysis.

Key Takeaways

• Corroborate before you contain. The YARA memory scan as a second, independent signal is what makes unattended automation safe, Wazuh alone is not enough authority to take a host off the network.
• Order of operations is the safety margin: whitelist management access, corroborate, snapshot evidence, then isolate, never the reverse.
• Every automated action needs an audit trail. The SQLite case row plus journald log turns a 2 a.m. robot decision into something an analyst can defend at 9 a.m.
• Least privilege applies to robots too. A scoped Proxmox token, an unprivileged service user, and a single-command sudoers rule keep the containment tool from becoming an attack surface.
• Test the negative path. A playbook that never fires on a false alert is more valuable than one that fires fast on real ones.

Practice Challenges