#!/usr/bin/env python3 """ sysmon_summary.py - condense exported Sysmon events into a behavior summary. Runs on the host against evidence exported from the analysis VM. Stdlib only. Accepted inputs (auto-detected): * PowerShell Get-WinEvent ... | Select Id,TimeCreated,Message | ConvertTo-Json (array or single object) * PowerShell Get-WinEvent ... | Export-Csv (CSV with Id + Message) * evtx_dump --format jsonl (one {"Event":{"System":..,"EventData":..}} per line) Usage: python3 sysmon_summary.py sysmon.json --process sample.exe python3 sysmon_summary.py sysmon.csv --all python3 sysmon_summary.py --self-test Sections: process tree (EID 1), network (EID 3, beacon interval check), DNS (22), files created (11), registry writes (12/13/14, persistence flagged), remote threads (8), process access (10), non-Windows image loads (7), pipes (17/18), WMI (19/20/21), tampering (25). Pipe output through scripts/ioc_extract.py. """ import argparse import csv import io import json import re import statistics import sys from collections import defaultdict, OrderedDict from datetime import datetime PERSISTENCE_RE = re.compile( r"CurrentVersion\\(Run|RunOnce|RunServices|Explorer\\(Shell Folders|User Shell Folders))" r"|CurrentControlSet\\Services\\|Winlogon\\(Shell|Userinit|Notify)|Image File Execution Options" r"|Schedule\\TaskCache|AppInit_DLLs|Start Menu\\Programs\\Startup", re.IGNORECASE, ) SYSTEM_DIR_RE = re.compile(r"^[A-Z]:\\Windows\\", re.IGNORECASE) def parse_message(msg): d = {} for line in str(msg).replace("\r", "").split("\n"): k, sep, v = line.partition(": ") if sep: d[k.strip()] = v.strip() return d def read_text(path): """Read exported evidence regardless of PowerShell's default encoding (UTF-16LE with BOM).""" raw = open(path, "rb").read() if raw.startswith((b"\xff\xfe", b"\xfe\xff")): return raw.decode("utf-16", errors="replace") return raw.decode("utf-8-sig", errors="replace") def load(path): """Yield (event_id:int, data:dict) regardless of export format.""" text = read_text(path).lstrip("\ufeff") stripped = text.lstrip() if stripped.startswith(("[", "{")): try: obj = json.loads(text) records = obj if isinstance(obj, list) else [obj] except json.JSONDecodeError: # jsonl records = [json.loads(l) for l in text.splitlines() if l.strip()] for r in records: if "Event" in r: # evtx_dump sysm = r["Event"].get("System", {}) eid = sysm.get("EventID") eid = eid.get("#text", eid) if isinstance(eid, dict) else eid data = dict(r["Event"].get("EventData") or {}) yield int(eid), data else: yield int(r.get("Id", 0)), parse_message(r.get("Message", "")) else: for r in csv.DictReader(io.StringIO(text)): yield int(r.get("Id") or 0), parse_message(r.get("Message", "")) def beacon_stats(times): ts = sorted(t for t in times if t) if len(ts) < 5: return "" gaps = [(b - a).total_seconds() for a, b in zip(ts, ts[1:])] mean = statistics.mean(gaps) if mean <= 0: return "" cv = statistics.pstdev(gaps) / mean tag = " <-- BEACON-LIKE" if cv < 0.3 else "" return f" [{len(ts)} conns, mean gap {mean:.0f}s, cv {cv:.2f}]{tag}" def parse_time(s): try: return datetime.strptime(s[:19], "%Y-%m-%d %H:%M:%S") except (ValueError, TypeError): return None def summarize(events, process_names=None): events = list(events) children, info = defaultdict(list), {} for eid, d in events: if eid == 1: pid, ppid = d.get("ProcessId", "?"), d.get("ParentProcessId", "?") info[pid] = (d.get("Image", "?"), d.get("CommandLine", "")) info.setdefault(ppid, (d.get("ParentImage", "?"), d.get("ParentCommandLine", ""))) children[ppid].append(pid) if process_names: wanted = {n.lower() for n in process_names} seeds = [p for p, (img, _) in info.items() if img.split("\\")[-1].lower() in wanted] scope, stack = set(), list(seeds) while stack: p = stack.pop() if p not in scope: scope.add(p) stack.extend(children.get(p, [])) roots = seeds else: scope = None all_children = {c for cs in children.values() for c in cs} roots = [p for p in info if p not in all_children] def in_scope(d): return scope is None or d.get("ProcessId") in scope or d.get("SourceProcessId") in scope net, net_times = OrderedDict(), defaultdict(list) dns, files, reg, threads, access, loads, pipes, wmi, tamper = (OrderedDict() for _ in range(9)) for eid, d in events: if not in_scope(d): continue img = d.get("Image", d.get("SourceImage", "?")).split("\\")[-1] if eid == 3: key = f"{d.get('Protocol', '?')} {d.get('DestinationIp', '?')}:{d.get('DestinationPort', '?')}" + ( f" ({d['DestinationHostname']})" if d.get("DestinationHostname") not in (None, "", "-") else "" ) net.setdefault(key, img) net_times[key].append(parse_time(d.get("UtcTime", ""))) elif eid == 22: dns.setdefault(f"{d.get('QueryName', '?')} -> {d.get('QueryResults', '')}", img) elif eid == 11: files.setdefault(d.get("TargetFilename", "?"), img) elif eid in (12, 13, 14): key = f"{d.get('EventType', '')} {d.get('TargetObject', '?')}" + ( f" = {d['Details']}" if d.get("Details") else "" ) reg.setdefault(key, img) elif eid == 8: threads.setdefault(f"{d.get('SourceImage')} -> {d.get('TargetImage')} @ {d.get('StartAddress')}", img) elif eid == 10: access.setdefault(f"{d.get('SourceImage')} -> {d.get('TargetImage')} access {d.get('GrantedAccess')}", img) elif eid == 7 and not SYSTEM_DIR_RE.match(d.get("ImageLoaded", "")): loads.setdefault(f"{d.get('ImageLoaded')} signed={d.get('Signed', '?')}", img) elif eid in (17, 18): pipes.setdefault(f"{'created' if eid == 17 else 'connected'} {d.get('PipeName')}", img) elif eid in (19, 20, 21): wmi.setdefault(f"EID{eid} {d.get('Name', '')} {d.get('Query', '')} {d.get('Destination', '')}".strip(), img) elif eid == 25: tamper.setdefault(f"{d.get('Image')} {d.get('Type', '')}", img) out = [f"# Sysmon summary ({len(events)} events)", "", "## Process tree (EID 1)"] def tree(pid, depth): img, cmd = info.get(pid, ("?", "")) out.append(f"{' ' * depth}{img} (PID {pid})" + (f" {cmd}" if cmd else "")) for c in children.get(pid, []): tree(c, depth + 1) for r in roots: tree(r, 0) def section(title, items, extra=None): out.extend(["", f"## {title} ({len(items)})"]) for k, who in items.items(): out.append(f"- {k} [{who}]" + (extra(k) if extra else "")) section("Network connections (EID 3)", net, lambda k: beacon_stats(net_times[k])) section("DNS queries (EID 22)", dns) section("Files created (EID 11)", files) section("Registry writes (EID 12/13/14)", reg, lambda k: " <-- PERSISTENCE" if PERSISTENCE_RE.search(k) else "") section("Remote threads (EID 8)", threads) section("Process access (EID 10)", access) section("Image loads outside C:\\Windows (EID 7)", loads) section("Named pipes (EID 17/18)", pipes) section("WMI activity (EID 19/20/21)", wmi) section("Process tampering (EID 25)", tamper) return "\n".join(out) SELF_TEST = [ {"Id": 1, "Message": "Process Create:\nUtcTime: 2026-01-01 10:00:00.000\nProcessId: 200\nImage: C:\\Users\\Public\\sample.exe\nCommandLine: sample.exe /install\nParentProcessId: 100\nParentImage: C:\\Windows\\explorer.exe\nParentCommandLine: explorer.exe"}, {"Id": 1, "Message": "Process Create:\nProcessId: 300\nImage: C:\\Windows\\System32\\cmd.exe\nCommandLine: cmd /c whoami\nParentProcessId: 200\nParentImage: C:\\Users\\Public\\sample.exe"}, {"Id": 13, "Message": "Registry value set:\nEventType: SetValue\nProcessId: 200\nImage: C:\\Users\\Public\\sample.exe\nTargetObject: HKU\\S-1\\Software\\Microsoft\\Windows\\CurrentVersion\\Run\\svc\nDetails: C:\\Users\\Bob\\AppData\\Roaming\\svc.exe"}, {"Id": 11, "Message": "File created:\nProcessId: 999\nImage: C:\\Program Files\\Chrome\\chrome.exe\nTargetFilename: C:\\Users\\Bob\\cache.dat"}, ] + [ {"Id": 3, "Message": f"Network connection detected:\nUtcTime: 2026-01-01 10:0{i}:00.000\nProcessId: 200\nImage: C:\\Users\\Public\\sample.exe\nProtocol: tcp\nDestinationIp: 203.0.113.5\nDestinationPort: 443\nDestinationHostname: -"} for i in range(6) ] def self_test(): import tempfile, os with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f: json.dump(SELF_TEST, f) try: out = summarize(load(f.name), ["sample.exe"]) finally: os.unlink(f.name) assert " C:\\Windows\\System32\\cmd.exe (PID 300)" in out, "child not nested" assert "PERSISTENCE" in out, "run key not flagged" assert "BEACON-LIKE" in out, "regular 60s connections not flagged" assert "cache.dat" not in out, "out-of-scope process leaked" print("self-test OK") def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("path", nargs="?") ap.add_argument("--process", help="comma list of image names to scope on (plus descendants)") ap.add_argument("--all", action="store_true") ap.add_argument("--self-test", action="store_true") a = ap.parse_args() if a.self_test: return self_test() if not a.path: ap.error("path required") names = None if a.all or not a.process else [n.strip() for n in a.process.split(",")] print(summarize(load(a.path), names)) if __name__ == "__main__": main()