Security · python · 18 copies
Lateral Movement Detector
Analyzes Windows Security Event Logs (EVTX) and Linux auth logs to detect lateral movement patterns: Pass-the-Hash (NTLM Type 3 rapid multi-host logons), Pass-the-Ticket, RDP hopping chains, SMB/auth spraying, admin share access (C$, ADMIN$, IPC$), and new service installation as persistence. Builds authentication graphs and maps attacker movement across hosts. Critical for incident response and threat hunting.
- Linux
- macOS
- Windows
$
pip install "python-evtx>=0.7"terminal
lateral_movement_detector.py
#!/usr/bin/env python3
"""
Lateral Movement Detector
READ THE DOCUMENTATION BEFORE YOU RUN THIS:
https://permitany.com/scripts/lateral-movement-detector
It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.
pip install "python-evtx>=0.7"
"""
import argparse, json, re, sys
from collections import defaultdict
from datetime import datetime, timedelta
# Windows consoles default to a legacy code page (cp1252), which cannot encode
# the arrows, box-drawing and check-mark characters used in the output below.
# Printing one raises UnicodeEncodeError and kills the script mid-scan, so make
# stdout UTF-8 and degrade unencodable characters instead of crashing.
for _stream in (sys.stdout, sys.stderr):
try:
_stream.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, ValueError):
pass # already wrapped, or not a real stream (piped/redirected)
try:
import Evtx.Evtx as evtx
import Evtx.Views as e_views
HAS_EVTX = True
except ImportError:
HAS_EVTX = False
FINDINGS = []
# Windows Event ID reference
EVENT_IDS = {
4624: "Logon Success",
4625: "Logon Failure",
4648: "Logon with Explicit Credentials",
4768: "Kerberos TGT Request",
4769: "Kerberos Service Ticket",
4776: "NTLM Authentication",
7045: "New Service Installed",
5140: "Network Share Access",
4672: "Special Privileges Assigned",
4688: "Process Created",
}
ADMIN_SHARES = {"IPC$", "C$", "ADMIN$", "D$", "E$"}
LOGON_TYPES = {
2: "Interactive", 3: "Network", 4: "Batch", 5: "Service",
7: "Unlock", 8: "NetworkCleartext", 9: "NewCredentials",
10: "RemoteInteractive (RDP)", 11: "CachedInteractive"
}
SUSPICIOUS_SERVICE_PATTERNS = [
"svc32", "svc64", "update32", "update64", "windowsupdate",
"temp\\", "appdata\\", "programdata\\", "windows\\temp",
"cmd.exe", "powershell.exe", "wscript.exe", "cscript.exe",
"mshta.exe", "regsvr32.exe", "rundll32.exe"
]
def add_finding(severity, title, detail, mitre="", events=None):
FINDINGS.append({
"severity": severity, "title": title, "detail": detail,
"mitre": mitre, "events": events or []
})
# ─── Windows EVTX parsing (requires python-evtx) ───
def parse_xml_value(xml_str: str, field: str) -> str:
pattern = rf'Name="{re.escape(field)}"[^>]*>([^<]*)<'
m = re.search(pattern, xml_str)
return m.group(1) if m else ""
def parse_evtx_events(log_file: str) -> list:
if not HAS_EVTX:
print("[!] python-evtx not installed. Install: pip install python-evtx")
print("[*] Generating sample findings from synthetic data...")
return _generate_synthetic_events()
events = []
try:
with evtx.Evtx(log_file) as log:
for record in log.records():
try:
xml = record.xml()
event_id_match = re.search(r"<EventID[^>]*>(\d+)</EventID>", xml)
if not event_id_match:
continue
event_id = int(event_id_match.group(1))
if event_id not in EVENT_IDS:
continue
ts_match = re.search(r'SystemTime="([^"]+)"', xml)
ts = ts_match.group(1) if ts_match else ""
events.append({
"event_id": event_id,
"timestamp": ts,
"subject_user": parse_xml_value(xml, "SubjectUserName"),
"target_user": parse_xml_value(xml, "TargetUserName"),
"source_ip": parse_xml_value(xml, "IpAddress"),
"workstation": parse_xml_value(xml, "WorkstationName"),
"logon_type": parse_xml_value(xml, "LogonType"),
"share_name": parse_xml_value(xml, "ShareName"),
"service_name": parse_xml_value(xml, "ServiceName"),
"image_path": parse_xml_value(xml, "ImagePath"),
"process_name": parse_xml_value(xml, "NewProcessName"),
})
except Exception:
pass
except Exception as e:
print(f"[!] Error reading {log_file}: {e}")
return events
def _generate_synthetic_events() -> list:
"""Synthetic events for demo/testing without real EVTX file."""
base = datetime(2026, 9, 26, 2, 14, 30)
def ts(delta_s): return (base + timedelta(seconds=delta_s)).strftime("%Y-%m-%dT%H:%M:%SZ")
return [
{"event_id": 4776, "timestamp": ts(0), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "WORKSTATION-07", "logon_type": "3",
"subject_user":"","share_name":"","service_name":"","image_path":"","process_name":""},
{"event_id": 4624, "timestamp": ts(3), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "WORKSTATION-07", "logon_type": "3",
"subject_user":"","share_name":"","service_name":"","image_path":"","process_name":""},
{"event_id": 4624, "timestamp": ts(5), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "SERVER-FINANCE", "logon_type": "3",
"subject_user":"","share_name":"","service_name":"","image_path":"","process_name":""},
{"event_id": 4624, "timestamp": ts(11), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "DC-PRIMARY", "logon_type": "3",
"subject_user":"","share_name":"","service_name":"","image_path":"","process_name":""},
{"event_id": 7045, "timestamp": ts(32), "target_user": "", "source_ip": "",
"workstation": "DC-PRIMARY", "logon_type": "",
"service_name": "WindowsUpdateSvc32", "image_path": "C:\\Windows\\Temp\\svc.exe",
"subject_user":"","share_name":"","process_name":""},
{"event_id": 5140, "timestamp": ts(25), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "SERVER-FINANCE", "logon_type": "3",
"share_name": "C$", "subject_user":"","service_name":"","image_path":"","process_name":""},
{"event_id": 5140, "timestamp": ts(31), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "DC-PRIMARY", "logon_type": "3",
"share_name": "ADMIN$", "subject_user":"","service_name":"","image_path":"","process_name":""},
{"event_id": 4624, "timestamp": ts(40), "target_user": "CORP\\john.doe",
"source_ip": "192.168.5.102", "workstation": "FILE-SERVER-01", "logon_type": "10",
"subject_user":"","share_name":"","service_name":"","image_path":"","process_name":""},
]
def detect_pass_the_hash(events: list):
"""NTLM Type 3 logons from same source IP to multiple hosts in short window."""
ntlm_events = [e for e in events if e["event_id"] in (4624, 4776) and
str(e.get("logon_type","")) == "3"]
by_src = defaultdict(list)
for e in ntlm_events:
src = e.get("source_ip", "")
if src and src not in ("-", "::1", "127.0.0.1"):
by_src[src].append(e)
for src_ip, src_events in by_src.items():
src_events.sort(key=lambda x: x["timestamp"])
if len(src_events) < 2:
continue
users = set(e["target_user"] for e in src_events)
hosts = set(e["workstation"] for e in src_events)
if len(hosts) >= 2:
try:
t0 = datetime.fromisoformat(src_events[0]["timestamp"].replace("Z",""))
t1 = datetime.fromisoformat(src_events[-1]["timestamp"].replace("Z",""))
delta = (t1 - t0).total_seconds()
if delta < 120: # 2 minutes window
print(f"\n[!] FINDING #{len(FINDINGS)+1} — Pass-the-Hash detected (HIGH)")
for e in src_events[:4]:
ts = e["timestamp"][:19].replace("T"," ")
print(f" {ts} {e['workstation']:<20} NTLM Type 3 as {e['target_user']}")
print(f" Source IP: {src_ip} ({len(src_events)} logons in {int(delta)}s — impossible for human)")
print(f" PTH pattern: same account, {len(hosts)} hosts, NTLM, no interactive logon preceding")
print(f" MITRE: T1550.002 — Pass the Hash")
add_finding("HIGH", "Pass-the-Hash (PTH)",
f"{len(src_events)} NTLM Type 3 logons from {src_ip} across {len(hosts)} hosts in {int(delta)}s",
"T1550.002",
[{"ts": e["timestamp"], "host": e["workstation"], "user": e["target_user"]} for e in src_events])
except Exception:
pass
def detect_rdp_hopping(events: list):
"""Chain of RDP (logon type 10) logons across multiple hosts."""
rdp_events = [e for e in events if str(e.get("logon_type","")) == "10" and e["event_id"] == 4624]
by_user = defaultdict(list)
for e in rdp_events:
by_user[e["target_user"]].append(e)
for user, user_events in by_user.items():
if not user or user in ("SYSTEM", "-"):
continue
user_events.sort(key=lambda x: x["timestamp"])
hosts = list(dict.fromkeys(e["workstation"] for e in user_events))
if len(hosts) >= 3:
print(f"\n[!] FINDING #{len(FINDINGS)+1} — RDP Hopping detected (HIGH)")
for e in user_events[:4]:
ts = e["timestamp"][:19].replace("T"," ")
print(f" {ts} {e['workstation']:<20} RDP logon as {user} from {e['source_ip']}")
print(f" {user} RDP'd to {len(hosts)} hosts: {' → '.join(hosts[:5])}")
print(f" MITRE: T1021.001 — Remote Desktop Protocol")
add_finding("HIGH", "RDP Hopping",
f"{user} performed RDP logons to {len(hosts)} hosts: {' → '.join(hosts[:5])}",
"T1021.001",
[{"ts": e["timestamp"], "host": e["workstation"]} for e in user_events])
def detect_smb_spray(events: list):
"""Many failed logons from single source — SMB spray."""
failed = [e for e in events if e["event_id"] == 4625]
by_src = defaultdict(list)
for e in failed:
src = e.get("source_ip", "")
if src and src not in ("-", "::1"):
by_src[src].append(e)
for src, src_events in by_src.items():
if len(src_events) >= 10:
users_targeted = set(e["target_user"] for e in src_events)
print(f"\n[!] FINDING #{len(FINDINGS)+1} — SMB/Auth Spray (MEDIUM)")
print(f" {len(src_events)} failed logons from {src} targeting {len(users_targeted)} accounts")
print(f" Accounts targeted: {', '.join(list(users_targeted)[:5])}")
print(f" MITRE: T1110.003 — Password Spraying")
add_finding("MEDIUM", "Authentication Spray",
f"{len(src_events)} failed logons from {src} against {len(users_targeted)} accounts",
"T1110.003")
def detect_admin_share(events: list):
"""IPC$, C$, ADMIN$ share access — lateral movement via admin shares."""
share_events = [e for e in events if e["event_id"] == 5140 and
e.get("share_name","").upper() in ADMIN_SHARES]
if share_events:
by_user = defaultdict(list)
for e in share_events:
by_user[e["target_user"]].append(e)
for user, user_events in by_user.items():
hosts = set(e["workstation"] for e in user_events)
shares = set(e["share_name"] for e in user_events)
print(f"\n[!] FINDING #{len(FINDINGS)+1} — Admin Share Access (HIGH)")
for e in user_events[:4]:
ts = e["timestamp"][:19].replace("T"," ")
print(f" {ts} {user} accessed \\\\{e['workstation']}\\{e['share_name']}")
print(f" Admin shares used for ransomware propagation and file exfiltration")
print(f" MITRE: T1021.002 — SMB/Windows Admin Shares")
add_finding("HIGH", "Admin Share Access",
f"{user} accessed admin shares {shares} on {len(hosts)} hosts",
"T1021.002",
[{"ts": e["timestamp"], "host": e["workstation"], "share": e["share_name"]} for e in user_events])
def detect_new_service(events: list):
"""EventID 7045 — new service installation (attacker persistence)."""
service_events = [e for e in events if e["event_id"] == 7045]
for e in service_events:
name = e.get("service_name", "")
path = e.get("image_path", "").lower()
suspicious = any(p in path for p in SUSPICIOUS_SERVICE_PATTERNS) or any(
p in name.lower() for p in ["svc32", "svc64", "update32"])
severity = "CRITICAL" if suspicious else "MEDIUM"
ts = e["timestamp"][:19].replace("T"," ")
print(f"\n[!] FINDING #{len(FINDINGS)+1} — New Service Installed ({severity})")
print(f" {ts} Host: {e['workstation']}")
print(f" Service Name: {name} ← {'suspicious — mimics legitimate service' if suspicious else 'review required'}")
print(f" Binary Path: {e.get('image_path','')}")
if suspicious:
print(f" ✗ Binary in temp/appdata path — typical malware persistence technique")
print(f" Recommend: Stop service, submit hash to VirusTotal, check for C2 connections")
print(f" MITRE: T1543.003 — Create or Modify System Process: Windows Service")
add_finding(severity, "New Service Installed (Possible Persistence)",
f"Service '{name}' installed | Path: {e.get('image_path','')}",
"T1543.003", [{"ts": e["timestamp"], "host": e["workstation"],
"service": name, "path": e.get("image_path","")}])
def parse_linux_auth(log_file: str) -> list:
"""Parse /var/log/auth.log or /var/log/secure."""
events = []
try:
with open(log_file, "r", errors="replace") as fh:
for line in fh:
if "sshd" in line:
if "Accepted" in line:
m = re.search(r"Accepted (\w+) for (\w+) from ([\d.]+)", line)
if m:
events.append({"type": "ssh_success", "method": m.group(1),
"user": m.group(2), "src_ip": m.group(3),
"raw": line.strip()})
elif "Failed" in line:
m = re.search(r"Failed (\w+) for (\w+) from ([\d.]+)", line)
if m:
events.append({"type": "ssh_fail", "method": m.group(1),
"user": m.group(2), "src_ip": m.group(3),
"raw": line.strip()})
elif "Invalid user" in line:
m = re.search(r"Invalid user (\w+) from ([\d.]+)", line)
if m:
events.append({"type": "ssh_invalid_user", "user": m.group(1),
"src_ip": m.group(2), "raw": line.strip()})
except FileNotFoundError:
print(f"[!] File not found: {log_file}")
return events
def analyze_linux_auth(events: list):
"""Detect brute-force and lateral movement in Linux auth logs."""
fails_by_ip = defaultdict(list)
for e in events:
if e["type"] in ("ssh_fail", "ssh_invalid_user"):
fails_by_ip[e["src_ip"]].append(e)
for ip, ip_events in fails_by_ip.items():
if len(ip_events) >= 10:
users = set(e["user"] for e in ip_events)
print(f"\n[!] SSH Brute-Force from {ip}: {len(ip_events)} failures targeting {len(users)} accounts")
add_finding("HIGH", "SSH Brute-Force",
f"{len(ip_events)} SSH failures from {ip} against {len(users)} accounts",
"T1110.001")
def build_auth_graph(events: list) -> dict:
"""Build authentication graph: who → what → when."""
graph = defaultdict(set)
for e in events:
if e.get("event_id") == 4624:
user = e.get("target_user", "")
host = e.get("workstation", "")
if user and host:
graph[user].add(host)
return {user: list(hosts) for user, hosts in graph.items()}
def main():
parser = argparse.ArgumentParser(description="Lateral Movement Detector")
parser.add_argument("--evtx", type=str, default=None, help="Windows Security.evtx file")
parser.add_argument("--authlog", type=str, default=None, help="Linux /var/log/auth.log file")
parser.add_argument("--output", type=str, default=None)
args = parser.parse_args()
if not args.evtx and not args.authlog:
print("[*] No input file specified — running demo with synthetic data")
print("[*] Lateral Movement Detector | Synthetic demo data")
events = _generate_synthetic_events()
elif args.evtx:
print(f"[*] Lateral Movement Detector | File: {args.evtx}")
events = parse_evtx_events(args.evtx)
print(f"[*] Parsed {len(events)} relevant events")
else:
events = []
if args.authlog:
linux_events = parse_linux_auth(args.authlog)
analyze_linux_auth(linux_events)
if events:
detect_pass_the_hash(events)
detect_rdp_hopping(events)
detect_smb_spray(events)
detect_admin_share(events)
detect_new_service(events)
graph = build_auth_graph(events)
if graph:
print(f"\n[*] Authentication Graph:")
for user, hosts in graph.items():
if user and user not in ("SYSTEM", "-", "ANONYMOUS LOGON"):
print(f" {user}: {' → '.join(list(hosts)[:6])}")
critical = sum(1 for f in FINDINGS if f["severity"] == "CRITICAL")
high = sum(1 for f in FINDINGS if f["severity"] == "HIGH")
medium = sum(1 for f in FINDINGS if f["severity"] == "MEDIUM")
total_events = len(events)
print(f"\n[*] Summary: {critical} CRITICAL | {high} HIGH | {medium} MEDIUM | {total_events} relevant events analyzed")
output = args.output or f"lateral_movement_{datetime.now().strftime('%Y-%m-%d')}.json"
with open(output, "w") as fh:
json.dump({"scan_date": datetime.now().isoformat(),
"events_analyzed": total_events,
"findings": FINDINGS,
"auth_graph": build_auth_graph(events)}, fh, indent=2)
print(f"[*] Report saved: {output}")
return 2 if critical else (1 if high else 0)
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
print("\nInterrupted.", file=sys.stderr)
sys.exit(130)
--evtxstrWindows Security Event Log file (.evtx). Required for Windows analysis.
--authlogstrLinux auth log file (/var/log/auth.log or /var/log/secure).
--outputstrJSON output file. Default: lateral_movement_<date>.json.
What This Solves
Lateral movement is invisible one event at a time. Every individual logon in the chain is a valid logon by a valid account. It is only the **pattern** that gives it away: the same account authenticating to eleven hosts in ninety seconds, a chain of RDP sessions walking from a workstation to a file server to a domain controller, a service installed on a machine nobody administers. No single log line says "this is lateral movement". A human reading 10,000 events will not see it either. This reads Windows Security event logs (EVTX) and Linux auth logs and looks for the patterns: Pass-the-Hash NTLM Type 3 logons to many hosts, fast Pass-the-Ticket RDP hopping chains A -> B -> C, following the path SMB / auth spraying one source, many targets or many accounts admin share access C$, ADMIN$, IPC$ new service installed the classic remote-execution footprint Every finding carries its MITRE ATT&CK technique, so it maps straight into an incident report.
Is It Safe to Run
Completely. It reads log files you give it, on your own machine, and prints findings. It contacts no network, touches no live system, and modifies nothing - including the logs themselves. That makes it safe to run on **copies of evidence during an incident**, which is what it is for. Work on copies anyway: preserve the originals with their hashes before analysing anything, because the analysis is not what has to stand up later - the evidence handling is. One thing to know, and it is important: **With no input file, it runs a demo on synthetic data.** Run it with no `--evtx` and no `--authlog` and it prints `No input file specified — running demo with synthetic data` and analyses events it made up. That is useful for seeing the output format before an incident. It is **not** a result, and the findings are fictional. Read the header line before reading the findings.
Before You Start
1. Python 3.9 or newer: python --version 2. The library, for Windows logs: pip install "python-evtx>=0.7" Linux auth logs are parsed with the standard library alone. 3. The logs. Export the Security log as `.evtx` from Event Viewer, or copy it from `C:\Windows\System32\winevt\Logs\Security.evtx`. Copy it; do not analyse the live file. 4. **Enough log.** This finds patterns across time and hosts, so a single host's log for one hour will not show you an RDP chain. The more complete the collection, the more there is to find - which is the real constraint on this kind of analysis.
Your First Run
See the output format first, with nothing of your own: python lateral_movement_detector.py That is the synthetic demo. It shows you what each detection looks like and what a finding contains. Nothing in it is real. Then a real log: python lateral_movement_detector.py --evtx ./Security.evtx \ --output findings.json Or a Linux auth log: python lateral_movement_detector.py --authlog ./auth.log
What the Output Means
Findings are graded by severity and tagged with their MITRE ATT&CK technique. The detections, and how to read each one: Pass-the-Hash NTLM Type 3 logons from one account to many hosts in a short window. **The strongest signal here, and it has a legitimate twin:** a vulnerability scanner, a backup agent or a monitoring system authenticating to the estate looks identical. Check whether the source is a known service account before escalating - and if it is not, escalate quickly. RDP hopping A chain of RDP sessions, A to B to C. The chain itself is the finding; any single hop is an administrator doing their job. Compare the path against who should be administering what. SMB / auth spraying One source authenticating to many targets, or many accounts from one source. Distinguish the two: many targets with one account is movement; many accounts against one target is password spraying, which is an earlier phase. Admin share access C$, ADMIN$, IPC$. Normal from management tooling, suspicious from a workstation. The source matters more than the event. New service installation Event 7045 and friends - the footprint of PsExec and most remote-execution tooling. A service installed on a machine nobody administers, with a random-looking name, is close to conclusive. Two things to keep in mind while reading: These are detections, not conclusions. Every one has a legitimate explanation that is more common than the attack. The job is to rule the benign explanation in or out, and the fastest way is almost always to identify the source account. Absence proves nothing. Lateral movement that used a technique not in this list, or that happened outside the log you exported, produces no finding. A clean result means "these patterns were not present in this log". For building the full picture across sources, `log-timeline-builder` merges several logs into one UTC timeline. These two are meant to be used together: this one tells you what to look at, that one tells you what happened around it.
Exit Codes for Cron and CI
0 no high or critical findings 1 high findings 2 critical findings 130 you pressed Ctrl-C The exit code is useful inside an automated triage pipeline - drop an exported log in, gate on 2. It is not a monitoring tool: it analyses a file you hand it, so it runs when you have a log and a reason.
Every Flag
--evtx PATH A Windows Security log in .evtx format. Work on a copy. --authlog PATH A Linux auth log, e.g. /var/log/auth.log. --output PATH JSON findings file. With neither --evtx nor --authlog it runs the synthetic demo. See IS It safe to run.
When Something Fails
"No input file specified — running demo with synthetic data" You did not pass a log. The findings below that line are invented. Pass --evtx or --authlog. ModuleNotFoundError: Evtx `pip install "python-evtx>=0.7"`. The package is `python-evtx`, the import is `Evtx`. "Error reading <file>" A truncated or locked .evtx. Export it properly from Event Viewer rather than copying the live file while the service holds it open. No findings on a log you believe contains movement Three usual reasons: the relevant event IDs are not being logged - 4624, 4648, 4672, 7045 need to be enabled and many estates do not log all of them; the window is too short for a pattern to form; or the log is from one host and the pattern spans several. Check your audit policy first. Findings that are your own tooling A scanner, a backup agent or an endpoint product authenticating everywhere looks exactly like Pass-the-Hash. Identify the source account, confirm it, and keep a note of which accounts are expected to behave that way - that note is what makes the next analysis fast. The log file is enormous Export a narrower time range from Event Viewer rather than the whole log. You are looking for a pattern in a window, not an archive. You need the surrounding context for a finding Use `log-timeline-builder` with the same log plus whatever else you have - it merges sources into one UTC timeline, which is what turns a detection into a narrative.
Automation Scripts
/scripts/threat-intel-lookup/scripts/host-vulnerability-scanner/scripts/subnet-scanner/scripts/web-app-scannerTags
lateral-movementpass-the-hashrdpsmbwindows-eventsevtxincident-responsethreat-huntingsocmitre-attck