Security Skills Atlas
threat hunting

Detecting Lateral Movement with Splunk

Detect adversary lateral movement across networks using Splunk SPL queries against Windows authentication logs, SMB traffic, and remote service abuse.

lateral-movementmitre-attackproactive-detectionsiemsplunkta0008threat-hunting
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1021 on the official MITRE ATT&CK siteT1021.001 on the official MITRE ATT&CK siteT1021.002 on the official MITRE ATT&CK siteT1021.003 on the official MITRE ATT&CK siteT1021.004 on the official MITRE ATT&CK siteT1021.006 on the official MITRE ATT&CK siteT1046 on the official MITRE ATT&CK siteT1047 on the official MITRE ATT&CK siteT1057 on the official MITRE ATT&CK siteT1082 on the official MITRE ATT&CK siteT1083 on the official MITRE ATT&CK siteT1550.002 on the official MITRE ATT&CK siteT1550.003 on the official MITRE ATT&CK siteT1569.002 on the official MITRE ATT&CK siteT1570 on the official MITRE ATT&CK site

NIST CSF 2.0

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

MITRE D3FEND

Application Protocol Command Analysis on the official MITRE D3FEND siteClient-server Payload Profiling on the official MITRE D3FEND siteNetwork Isolation on the official MITRE D3FEND siteNetwork Traffic Analysis on the official MITRE D3FEND siteNetwork Traffic Community Deviation on the official MITRE D3FEND site

When to Use

  • When hunting for adversary movement between compromised systems
  • After detecting credential theft to trace subsequent lateral activity
  • When investigating unusual authentication patterns across the network
  • During incident response to scope the breadth of compromise
  • When proactively hunting for TA0008 (Lateral Movement) techniques

Prerequisites

  • Splunk Enterprise or Splunk Cloud with Windows event data ingested
  • Windows Security Event Logs forwarded (4624, 4625, 4648, 4672, 4768, 4769)
  • Sysmon deployed for process creation and network connection data
  • Network flow data or firewall logs for SMB/RDP/WinRM correlation
  • Active Directory user and group membership reference data

Workflow

  1. Define Lateral Movement Scope: Identify which lateral movement techniques to hunt (RDP, SMB/Admin Shares, WinRM, PsExec, WMI, DCOM, SSH).
  2. Query Authentication Events: Use SPL to search for Type 3 (Network) and Type 10 (RemoteInteractive) logons across the environment.
  3. Build Authentication Graphs: Map source-to-destination authentication relationships to identify unusual connection patterns.
  4. Detect First-Time Relationships: Identify new source-destination pairs that have not been seen in the historical baseline.
  5. Correlate with Process Activity: Link authentication events to subsequent process creation on destination hosts.
  6. Identify Anomalous Patterns: Flag lateral movement to sensitive servers, unusual hours, service account misuse, or rapid multi-host access.
  7. Report and Contain: Document lateral movement path, affected systems, and coordinate containment response.

Key Concepts

Concept Description
T1021 Remote Services (parent technique)
T1021.001 Remote Desktop Protocol (RDP)
T1021.002 SMB/Windows Admin Shares
T1021.003 Distributed COM (DCOM)
T1021.004 SSH
T1021.006 Windows Remote Management (WinRM)
T1570 Lateral Tool Transfer
T1047 Windows Management Instrumentation
T1569.002 Service Execution (PsExec)
Logon Type 3 Network logon (SMB, WinRM, mapped drives)
Logon Type 10 Remote Interactive (RDP)
Event ID 4624 Successful logon
Event ID 4648 Explicit credential logon (runas, PsExec)

Tools & Systems

Tool Purpose
Splunk Enterprise SIEM for log aggregation and SPL queries
Splunk Enterprise Security Threat detection and notable events
Windows Event Forwarding Centralize Windows logs
Sysmon Detailed process and network telemetry
BloodHound AD attack path analysis
PingCastle AD security assessment

Common Scenarios

  1. PsExec Lateral Movement: Adversary uses PsExec to execute commands on remote systems via SMB, generating Type 3 logon with ADMIN$ share access.
  2. RDP Pivoting: Attacker RDPs to internal systems using stolen credentials, creating Type 10 logon events.
  3. WMI Remote Execution: Adversary uses WMIC process call create to spawn processes on remote hosts.
  4. WinRM PowerShell Remoting: Attacker uses Enter-PSSession or Invoke-Command to execute code on remote systems.
  5. Pass-the-Hash via SMB: Compromised NTLM hashes used to authenticate to remote systems without knowing the plaintext password.

Output Format

Hunt ID: TH-LATMOV-[DATE]-[SEQ]
Movement Type: [RDP/SMB/WinRM/WMI/DCOM/PsExec]
Source Host: [Hostname/IP]
Destination Host: [Hostname/IP]
Account Used: [Username]
Logon Type: [3/10/other]
First Seen: [Timestamp]
Event Count: [Number of events]
Risk Level: [Critical/High/Medium/Low]
Lateral Movement Path: [A -> B -> C -> D]
Source materials

References and resources

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

References 3

api-reference.md1.5 KB

API Reference: Detecting Lateral Movement with Splunk

Key Lateral Movement Techniques

Technique MITRE ID Event Source
Pass-the-Hash T1550.002 Event 4624 Logon_Type=3 NTLM
PSExec T1569.002 Sysmon Event 1 (PSEXESVC.exe)
WMI Remote Exec T1047 Sysmon Event 1 (wmiprvse.exe)
RDP Pivoting T1021.001 Event 4624 Logon_Type=10
SMB/Admin Share T1021.002 Network logs dest_port=445
WinRM T1021.006 Sysmon Event 1 (wsmprovhost.exe)

Splunk SPL Syntax

# Pass-the-Hash detection
index=wineventlog EventCode=4624 Logon_Type=3
| where Authentication_Package="NTLM"
| stats dc(Computer) as targets by Source_Network_Address
| where targets > 3
 
# PSExec detection
index=sysmon EventCode=1
| where ParentImage="*\\services.exe" AND Image="*\\PSEXESVC.exe"

splunklib Python SDK

import splunklib.client as client
import splunklib.results as results
 
service = client.connect(host="splunk", port=8089, token="...")
job = service.jobs.create("search index=wineventlog EventCode=4624")
for result in results.JSONResultsReader(job.results(output_mode="json")):
    print(result)

Windows Logon Types

Type Description
2 Interactive (console)
3 Network (SMB, PSExec)
7 Unlock
10 RemoteInteractive (RDP)

CLI Usage

python agent.py --generate-queries
python agent.py --generate-queries --techniques pass_the_hash psexec_execution
python agent.py --parse-results splunk_output.json
standards.md2.6 KB

Standards and References - Lateral Movement Detection with Splunk

MITRE ATT&CK Lateral Movement (TA0008)

Technique Name Event Indicators
T1021.001 Remote Desktop Protocol Logon Type 10, RDP certificate events
T1021.002 SMB/Windows Admin Shares Logon Type 3, ADMIN$/C$/IPC$ access
T1021.003 Distributed COM Logon Type 3, DCOM process creation
T1021.004 SSH OpenSSH authentication events
T1021.006 Windows Remote Management WinRM/WSMan logon events
T1047 Windows Management Instrumentation WMI remote process creation
T1569.002 Service Execution PsExec service install + Type 3 logon
T1570 Lateral Tool Transfer File copy over SMB/RDP
T1550.002 Pass the Hash Type 3 logon with NTLM authentication
T1550.003 Pass the Ticket Kerberos TGS without preceding TGT

Windows Logon Types Reference

Type Name Description
2 Interactive Local console logon
3 Network SMB, mapped drives, WinRM
4 Batch Scheduled task execution
5 Service Service startup
7 Unlock Workstation unlock
8 NetworkCleartext IIS basic auth
9 NewCredentials RunAs /netonly
10 RemoteInteractive RDP, Terminal Services
11 CachedInteractive Cached domain logon

Key Windows Event IDs for Lateral Movement

Event ID Source Description
4624 Security Successful account logon
4625 Security Failed account logon
4648 Security Logon with explicit credentials
4672 Security Special privileges assigned (admin logon)
4768 Security Kerberos TGT requested
4769 Security Kerberos TGS requested
4776 Security NTLM credential validation
5140 Security Network share accessed
5145 Security Network share object access check
7045 System New service installed
1 Sysmon Process creation
3 Sysmon Network connection

Splunk Data Model References

  • Authentication data model for login events
  • Network_Traffic data model for connection data
  • Endpoint.Processes for process creation events
  • Change.Endpoint_Changes for service installations

Authentication Protocol Indicators

Protocol Lateral Movement Event Indicators
NTLM Pass-the-Hash Event 4776, NtLmSsp package
Kerberos Pass-the-Ticket Event 4768/4769, ticket anomalies
CredSSP RDP Event 4624 Type 10
WSMan WinRM Event 4624 Type 3, WSMan source
workflows.md4.6 KB

Detailed Hunting Workflow - Lateral Movement with Splunk

Phase 1: Network Logon Analysis

Step 1.1 - Type 3 Network Logons (SMB, WinRM)

index=wineventlog EventCode=4624 Logon_Type=3
| where NOT match(Account_Name, "(?i)(SYSTEM|ANONYMOUS|\\$)")
| stats count dc(Computer) as unique_destinations values(Computer) as destinations by Account_Name Source_Network_Address
| where unique_destinations > 3
| sort -unique_destinations

Step 1.2 - Type 10 RDP Logons

index=wineventlog EventCode=4624 Logon_Type=10
| stats count by Account_Name Source_Network_Address Computer
| lookup dnslookup clientip as Source_Network_Address OUTPUT clienthost as src_hostname
| table Account_Name src_hostname Source_Network_Address Computer count
| sort -count

Step 1.3 - Explicit Credential Logons (PsExec, RunAs)

index=wineventlog EventCode=4648
| where NOT match(Target_Server_Name, "(?i)(localhost|\\$)")
| stats count values(Target_Server_Name) as targets by Account_Name Process_Name Computer
| sort -count

Phase 2: Admin Share Access Detection

Step 2.1 - ADMIN$ and C$ Share Access

index=wineventlog EventCode=5140
| where Share_Name IN ("\\\\*\\ADMIN$", "\\\\*\\C$", "\\\\*\\IPC$")
| where NOT match(Account_Name, "(?i)(\\$|SYSTEM)")
| stats count values(Share_Name) as shares by Account_Name Source_Address Computer
| sort -count

Step 2.2 - SMB File Operations on Admin Shares

index=wineventlog EventCode=5145
| where match(Share_Name, "(?i)(ADMIN\\$|C\\$)")
| where match(Relative_Target_Name, "(?i)(\\.exe|\\.dll|\\.ps1|\\.bat|\\.cmd)")
| stats count by Account_Name Source_Address Share_Name Relative_Target_Name Computer

Phase 3: Service-Based Lateral Movement

Step 3.1 - PsExec Service Installation

index=wineventlog EventCode=7045
| where match(Service_File_Name, "(?i)(psexec|PSEXESVC|cmd\.exe|powershell)")
| table _time Computer Service_Name Service_File_Name Service_Account

Step 3.2 - Remote Service Creation Correlation

index=wineventlog EventCode=7045
| eval is_suspicious=if(match(Service_File_Name, "(?i)(temp|appdata|cmd|powershell)"), 1, 0)
| where is_suspicious=1
| join Computer [
    search index=wineventlog EventCode=4624 Logon_Type=3
    | rename Computer as Computer, Source_Network_Address as lateral_src
]
| table _time Computer Service_Name Service_File_Name lateral_src

Phase 4: WMI and DCOM Lateral Movement

Step 4.1 - Remote WMI Execution

index=sysmon EventCode=1
| where match(ParentImage, "(?i)WmiPrvSE\.exe") AND NOT match(Image, "(?i)(WmiApSrv|scrcons)")
| table _time Computer User ParentImage Image CommandLine

Step 4.2 - DCOM Lateral Movement

index=sysmon EventCode=1
| where match(ParentImage, "(?i)(mmc\.exe|excel\.exe|outlook\.exe)")
| where match(Image, "(?i)(cmd\.exe|powershell\.exe|mshta\.exe)")
| table _time Computer User ParentImage Image CommandLine

Phase 5: Authentication Graph Analysis

Step 5.1 - Build Lateral Movement Graph

index=wineventlog EventCode=4624 Logon_Type IN (3, 10)
| where NOT match(Account_Name, "(?i)(\\$|SYSTEM|ANONYMOUS)")
| eval connection=Source_Network_Address."->".Computer
| stats count first(_time) as first_seen last(_time) as last_seen by connection Account_Name
| sort -count

Step 5.2 - First-Time Source-Destination Pairs

index=wineventlog EventCode=4624 Logon_Type IN (3, 10) earliest=-1d
| where NOT match(Account_Name, "(?i)(\\$|SYSTEM)")
| eval pair=Account_Name.":".Source_Network_Address."->".Computer
| search NOT [
    | search index=wineventlog EventCode=4624 Logon_Type IN (3, 10) earliest=-30d latest=-1d
    | eval pair=Account_Name.":".Source_Network_Address."->".Computer
    | dedup pair
    | fields pair
]
| stats count by pair
| sort -count

Phase 6: Anomaly Detection

Step 6.1 - Velocity Anomaly (Rapid Multi-Host Access)

index=wineventlog EventCode=4624 Logon_Type=3
| where NOT match(Account_Name, "(?i)(\\$|SYSTEM)")
| bin _time span=10m
| stats dc(Computer) as hosts_accessed values(Computer) as destinations by _time Account_Name Source_Network_Address
| where hosts_accessed > 5
| sort -hosts_accessed

Step 6.2 - Off-Hours Lateral Movement

index=wineventlog EventCode=4624 Logon_Type IN (3, 10)
| where NOT match(Account_Name, "(?i)(\\$|SYSTEM)")
| eval hour=strftime(_time, "%H")
| where hour < 6 OR hour > 22
| stats count by Account_Name Source_Network_Address Computer hour
| sort -count

Step 6.3 - Service Account Lateral Movement

index=wineventlog EventCode=4624 Logon_Type=10
| where match(Account_Name, "(?i)(svc_|service|admin)")
| stats count by Account_Name Source_Network_Address Computer
| sort -count

Scripts 2

agent.py4.9 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Lateral movement detection agent using Splunk SPL query generation.

Generates and analyzes SPL queries for detecting lateral movement techniques
including pass-the-hash, RDP pivoting, WMI/PSExec execution, and SMB abuse.
"""

import argparse
import json
from datetime import datetime

LATERAL_MOVEMENT_QUERIES = {
    "pass_the_hash": {
        "mitre": "T1550.002",
        "severity": "CRITICAL",
        "spl": """index=wineventlog EventCode=4624 Logon_Type=3
| where Authentication_Package="NTLM" AND Logon_Process="NtLmSsp"
| where NOT match(Source_Network_Address, "^(127\\.0\\.0\\.1|::1|-)")
| stats count dc(Computer) as target_count values(Computer) as targets by Source_Network_Address Account_Name
| where target_count > 3
| sort -target_count"""
    },
    "psexec_execution": {
        "mitre": "T1569.002",
        "severity": "HIGH",
        "spl": """index=sysmon EventCode=1
| where (ParentImage="*\\services.exe" AND Image="*\\PSEXESVC.exe")
   OR (Image="*\\psexec.exe" OR Image="*\\psexec64.exe")
| stats count by Image, ParentImage, CommandLine, Computer, User
| sort -count"""
    },
    "wmi_remote_execution": {
        "mitre": "T1047",
        "severity": "HIGH",
        "spl": """index=sysmon EventCode=1
| where (Image="*\\wmiprvse.exe" AND ParentImage="*\\svchost.exe")
| where CommandLine!=""
| stats count by CommandLine, Computer, User
| sort -count"""
    },
    "rdp_pivoting": {
        "mitre": "T1021.001",
        "severity": "MEDIUM",
        "spl": """index=wineventlog EventCode=4624 Logon_Type=10
| stats count dc(Computer) as rdp_targets values(Computer) as targets by Source_Network_Address Account_Name
| where rdp_targets > 3
| sort -rdp_targets"""
    },
    "smb_lateral": {
        "mitre": "T1021.002",
        "severity": "HIGH",
        "spl": """index=network dest_port=445
| stats count dc(dest_ip) as smb_targets values(dest_ip) as targets by src_ip
| where smb_targets > 5
| sort -smb_targets"""
    },
    "winrm_execution": {
        "mitre": "T1021.006",
        "severity": "HIGH",
        "spl": """index=sysmon EventCode=1
| where Image="*\\wsmprovhost.exe" OR (ParentImage="*\\winrshost.exe")
| stats count by Image, CommandLine, Computer, User
| sort -count"""
    },
    "service_creation": {
        "mitre": "T1543.003",
        "severity": "HIGH",
        "spl": """index=wineventlog EventCode=7045
| where Service_Type="user mode service"
| stats count by Service_Name, Service_File_Name, Computer
| where match(Service_File_Name, "(cmd|powershell|\\\\\\\\|%COMSPEC%)")
| sort -count"""
    },
    "scheduled_task_remote": {
        "mitre": "T1053.005",
        "severity": "HIGH",
        "spl": """index=sysmon EventCode=1 Image="*\\schtasks.exe"
| where match(CommandLine, "/create.*/s\\s")
| stats count by CommandLine, Computer, User
| sort -count"""
    },
}


def generate_queries(techniques=None):
    if techniques:
        selected = {k: v for k, v in LATERAL_MOVEMENT_QUERIES.items() if k in techniques}
    else:
        selected = LATERAL_MOVEMENT_QUERIES

    return [{"technique": name, **details} for name, details in selected.items()]


def parse_splunk_results(filepath):
    findings = []
    with open(filepath, "r") as f:
        try:
            data = json.load(f)
            results = data.get("results", data if isinstance(data, list) else [data])
        except json.JSONDecodeError:
            f.seek(0)
            import csv
            reader = csv.DictReader(f)
            results = list(reader)

    for row in results:
        target_count = int(row.get("target_count", row.get("dc(Computer)", 0)))
        if target_count >= 3:
            findings.append({
                "source": row.get("Source_Network_Address", row.get("src_ip", "")),
                "user": row.get("Account_Name", row.get("User", "")),
                "target_count": target_count,
                "targets": row.get("targets", row.get("Computer", "")),
                "severity": "CRITICAL" if target_count >= 10 else "HIGH",
            })
    return findings


def main():
    parser = argparse.ArgumentParser(description="Lateral Movement Detector (Splunk SPL)")
    parser.add_argument("--generate-queries", action="store_true", help="Generate SPL queries")
    parser.add_argument("--techniques", nargs="+", choices=list(LATERAL_MOVEMENT_QUERIES.keys()),
                        help="Specific techniques to query")
    parser.add_argument("--parse-results", help="Parse Splunk JSON/CSV results file")
    args = parser.parse_args()

    results = {"timestamp": datetime.utcnow().isoformat() + "Z"}

    if args.generate_queries:
        results["queries"] = generate_queries(args.techniques)
        results["total_queries"] = len(results["queries"])

    if args.parse_results:
        findings = parse_splunk_results(args.parse_results)
        results["findings"] = findings
        results["total_findings"] = len(findings)

    print(json.dumps(results, indent=2))


if __name__ == "__main__":
    main()
process.py12.5 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""
Lateral Movement Detection Script
Analyzes Windows authentication logs to detect lateral movement patterns
including RDP, SMB, WinRM, PsExec, and WMI-based movement.
"""

import json
import csv
import argparse
import datetime
import re
from collections import defaultdict
from pathlib import Path

# Lateral movement logon types
LATERAL_LOGON_TYPES = {
    "3": {"name": "Network", "techniques": ["T1021.002", "T1021.006", "T1047"], "risk_base": 20},
    "10": {"name": "RemoteInteractive", "techniques": ["T1021.001"], "risk_base": 25},
}

# Suspicious account patterns
SYSTEM_ACCOUNTS = {"system", "anonymous logon", "anonymous", "local service", "network service", "dwm-1", "umfd-0"}

# Admin share indicators
ADMIN_SHARES = {"admin$", "c$", "ipc$", "d$", "e$"}

# PsExec and service-based indicators
SERVICE_LATERAL_PATTERNS = [
    r"psexec", r"PSEXESVC", r"csexec", r"remcom",
    r"cmd\.exe\s+/c", r"powershell.*-enc",
]


def parse_logs(input_path: str) -> list[dict]:
    """Parse JSON or CSV log files."""
    path = Path(input_path)
    if path.suffix == ".json":
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
            return data if isinstance(data, list) else data.get("events", [])
    elif path.suffix == ".csv":
        with open(path, "r", encoding="utf-8-sig") as f:
            return [dict(row) for row in csv.DictReader(f)]
    return []


def normalize_event(event: dict) -> dict:
    """Normalize authentication event fields."""
    field_map = {
        "event_id": ["EventCode", "EventID", "event_id"],
        "logon_type": ["Logon_Type", "LogonType", "logon_type"],
        "account": ["Account_Name", "TargetUserName", "account_name", "user.name"],
        "source_ip": ["Source_Network_Address", "IpAddress", "source_ip", "source.ip"],
        "source_host": ["Workstation_Name", "WorkstationName", "source_host"],
        "dest_host": ["Computer", "hostname", "DeviceName", "host.name"],
        "logon_process": ["Logon_Process", "LogonProcessName", "logon_process"],
        "auth_package": ["Authentication_Package", "AuthenticationPackageName", "auth_package"],
        "share_name": ["Share_Name", "ShareName", "share_name"],
        "service_name": ["Service_Name", "ServiceName", "service_name"],
        "service_path": ["Service_File_Name", "ServiceFileName", "service_path"],
        "process_name": ["Process_Name", "ProcessName", "process_name"],
        "timestamp": ["_time", "timestamp", "Timestamp", "@timestamp", "UtcTime"],
    }
    normalized = {}
    for target, sources in field_map.items():
        for src in sources:
            if src in event and event[src]:
                normalized[target] = str(event[src]).strip()
                break
        if target not in normalized:
            normalized[target] = ""
    return normalized


def detect_network_logon(event: dict) -> dict | None:
    """Detect lateral movement via network logon events."""
    event_id = event.get("event_id", "")
    logon_type = event.get("logon_type", "")

    if event_id != "4624" or logon_type not in LATERAL_LOGON_TYPES:
        return None

    account = event.get("account", "").lower()
    if account in SYSTEM_ACCOUNTS or account.endswith("$"):
        return None

    source_ip = event.get("source_ip", "")
    if not source_ip or source_ip in ("-", "::1", "127.0.0.1"):
        return None

    lt_info = LATERAL_LOGON_TYPES[logon_type]
    risk = lt_info["risk_base"]
    indicators = [f"Logon Type {logon_type} ({lt_info['name']})"]

    auth_pkg = event.get("auth_package", "").lower()
    if "ntlm" in auth_pkg:
        risk += 10
        indicators.append("NTLM authentication (potential Pass-the-Hash)")
    if "negotiate" in auth_pkg and logon_type == "3":
        indicators.append("Negotiate authentication package")

    return {
        "detection_type": "NETWORK_LOGON",
        "technique": lt_info["techniques"][0],
        "account": event.get("account", ""),
        "source_ip": source_ip,
        "source_host": event.get("source_host", ""),
        "dest_host": event.get("dest_host", ""),
        "logon_type": logon_type,
        "auth_package": event.get("auth_package", ""),
        "timestamp": event.get("timestamp", ""),
        "risk_score": risk,
        "indicators": indicators,
    }


def detect_explicit_creds(event: dict) -> dict | None:
    """Detect explicit credential usage (Event 4648)."""
    if event.get("event_id") != "4648":
        return None

    account = event.get("account", "").lower()
    if account in SYSTEM_ACCOUNTS or account.endswith("$"):
        return None

    return {
        "detection_type": "EXPLICIT_CREDENTIAL",
        "technique": "T1021",
        "account": event.get("account", ""),
        "source_host": event.get("source_host", event.get("dest_host", "")),
        "dest_host": event.get("dest_host", ""),
        "process_name": event.get("process_name", ""),
        "timestamp": event.get("timestamp", ""),
        "risk_score": 35,
        "indicators": ["Explicit credential logon (4648) - possible PsExec/RunAs"],
    }


def detect_share_access(event: dict) -> dict | None:
    """Detect admin share access."""
    if event.get("event_id") != "5140":
        return None

    share = event.get("share_name", "").lower()
    share_name = share.split("\\")[-1] if "\\" in share else share

    if share_name not in ADMIN_SHARES:
        return None

    account = event.get("account", "").lower()
    if account in SYSTEM_ACCOUNTS or account.endswith("$"):
        return None

    risk = 40 if share_name in ("admin$", "c$") else 25

    return {
        "detection_type": "ADMIN_SHARE_ACCESS",
        "technique": "T1021.002",
        "account": event.get("account", ""),
        "source_ip": event.get("source_ip", ""),
        "dest_host": event.get("dest_host", ""),
        "share": share,
        "timestamp": event.get("timestamp", ""),
        "risk_score": risk,
        "indicators": [f"Admin share accessed: {share_name}"],
    }


def detect_service_lateral(event: dict) -> dict | None:
    """Detect service-based lateral movement (PsExec)."""
    if event.get("event_id") not in ("7045", "4697"):
        return None

    service_path = event.get("service_path", "")
    for pattern in SERVICE_LATERAL_PATTERNS:
        if re.search(pattern, service_path, re.IGNORECASE):
            return {
                "detection_type": "SERVICE_LATERAL",
                "technique": "T1569.002",
                "service_name": event.get("service_name", ""),
                "service_path": service_path,
                "dest_host": event.get("dest_host", ""),
                "timestamp": event.get("timestamp", ""),
                "risk_score": 60,
                "indicators": [f"Suspicious service for lateral movement: {pattern}"],
            }
    return None


def build_movement_graph(findings: list[dict]) -> dict:
    """Build a graph of lateral movement paths."""
    graph = defaultdict(lambda: defaultdict(list))
    for finding in findings:
        src = finding.get("source_ip") or finding.get("source_host", "unknown")
        dst = finding.get("dest_host", "unknown")
        if src and dst and src != dst:
            graph[src][dst].append({
                "account": finding.get("account", ""),
                "technique": finding.get("technique", ""),
                "timestamp": finding.get("timestamp", ""),
                "type": finding.get("detection_type", ""),
            })
    return dict(graph)


def analyze_velocity(findings: list[dict], window_minutes: int = 10, threshold: int = 5) -> list[dict]:
    """Detect rapid multi-host access patterns."""
    account_events = defaultdict(list)
    for f in findings:
        if f.get("account") and f.get("timestamp"):
            account_events[f["account"]].append(f)

    velocity_alerts = []
    for account, events in account_events.items():
        events.sort(key=lambda x: x.get("timestamp", ""))
        unique_dests = set()
        window_start = 0

        for i, event in enumerate(events):
            unique_dests.add(event.get("dest_host", ""))
            if len(unique_dests) >= threshold:
                velocity_alerts.append({
                    "detection_type": "VELOCITY_ANOMALY",
                    "account": account,
                    "unique_destinations": len(unique_dests),
                    "destinations": list(unique_dests),
                    "risk_score": 80,
                    "risk_level": "CRITICAL",
                    "indicators": [f"Account accessed {len(unique_dests)} hosts rapidly"],
                })
                break

    return velocity_alerts


def run_hunt(input_path: str, output_dir: str) -> None:
    """Execute lateral movement hunt."""
    print(f"[*] Lateral Movement Hunt - {datetime.datetime.now().isoformat()}")

    events = parse_logs(input_path)
    print(f"[*] Loaded {len(events)} events")

    findings = []
    stats = defaultdict(int)

    detectors = [
        detect_network_logon,
        detect_explicit_creds,
        detect_share_access,
        detect_service_lateral,
    ]

    for raw_event in events:
        event = normalize_event(raw_event)
        for detector in detectors:
            result = detector(event)
            if result:
                risk = result["risk_score"]
                result["risk_level"] = (
                    "CRITICAL" if risk >= 70 else "HIGH" if risk >= 50
                    else "MEDIUM" if risk >= 30 else "LOW"
                )
                findings.append(result)
                stats[result["detection_type"]] += 1

    # Velocity analysis
    velocity_alerts = analyze_velocity(findings)
    findings.extend(velocity_alerts)
    stats["VELOCITY_ANOMALY"] = len(velocity_alerts)

    # Build movement graph
    graph = build_movement_graph(findings)

    # Write output
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    with open(output_path / "lateral_movement_findings.json", "w", encoding="utf-8") as f:
        json.dump({
            "hunt_id": f"TH-LATMOV-{datetime.date.today().isoformat()}",
            "total_events": len(events),
            "total_findings": len(findings),
            "statistics": dict(stats),
            "movement_graph": {src: dict(dsts) for src, dsts in graph.items()},
            "findings": findings,
        }, f, indent=2)

    with open(output_path / "hunt_report.md", "w", encoding="utf-8") as f:
        f.write(f"# Lateral Movement Hunt Report\n\n")
        f.write(f"**Date**: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write(f"**Events**: {len(events)} | **Findings**: {len(findings)}\n\n")
        f.write("## Movement Graph\n\n")
        for src, dests in graph.items():
            for dst, connections in dests.items():
                f.write(f"- `{src}` -> `{dst}` ({len(connections)} connections)\n")
        f.write("\n## Velocity Anomalies\n\n")
        for alert in velocity_alerts:
            f.write(f"- **{alert['account']}**: {alert['unique_destinations']} hosts in short window\n")

    print(f"[+] {len(findings)} findings, {len(graph)} source nodes in movement graph")


def main():
    parser = argparse.ArgumentParser(description="Lateral Movement Detection")
    subparsers = parser.add_subparsers(dest="command")

    hunt_p = subparsers.add_parser("hunt")
    hunt_p.add_argument("--input", "-i", required=True)
    hunt_p.add_argument("--output", "-o", default="./latmov_output")

    subparsers.add_parser("queries", help="Print Splunk SPL queries")

    args = parser.parse_args()

    if args.command == "hunt":
        run_hunt(args.input, args.output)
    elif args.command == "queries":
        print("=== Splunk Lateral Movement Queries ===\n")
        queries = {
            "Network Logons": 'index=wineventlog EventCode=4624 Logon_Type=3\n| where NOT match(Account_Name, "(?i)(SYSTEM|ANONYMOUS|\\\\$)")\n| stats count dc(Computer) by Account_Name Source_Network_Address\n| where count > 3',
            "RDP Sessions": 'index=wineventlog EventCode=4624 Logon_Type=10\n| stats count by Account_Name Source_Network_Address Computer',
            "Admin Shares": 'index=wineventlog EventCode=5140 Share_Name IN ("*ADMIN$","*C$")\n| stats count by Account_Name Source_Address Computer Share_Name',
            "PsExec Services": 'index=wineventlog EventCode=7045\n| where match(Service_File_Name, "(?i)(psexec|PSEXESVC)")\n| table _time Computer Service_Name Service_File_Name',
        }
        for name, query in queries.items():
            print(f"--- {name} ---")
            print(query)
            print()
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

Assets 1

template.mdtext/markdown · 1.2 KB
Keep exploring
Hunting for Living-off-the-Land Binaries (LOLBins)Proactively hunt for adversary abuse of legitimate system binaries (LOLBins) to execute malicious payloads while evading detection.Hunting for Persistence Mechanisms in WindowsSystematically hunt for adversary persistence mechanisms across Windows endpoints including registry, services, startup folders, and WMI subscriptions.Detecting DLL Sideloading AttacksDetect DLL side-loading attacks where adversaries place malicious DLLs alongside legitimate applications to hijack execution flow for defense evasion.Detecting Email Forwarding Rules AttackDetect malicious email forwarding rules created by adversaries to maintain persistent access to email communications for intelligence collection and BEC attacks.Detecting Insider Threat BehaviorsDetect insider threat behavioral indicators including unusual data access, off-hours activity, mass file downloads, privilege abuse, and resignation-correlated data theft.Detecting Kerberoasting AttacksDetect Kerberoasting attacks by monitoring for anomalous Kerberos TGS requests targeting service accounts with SPNs for offline password cracking.