Chapter 38: Advanced Threat Hunting¶

import pandas as pd
import numpy as np
from collections import Counter

def long_tail_analysis(df: pd.DataFrame, field: str,
                       percentile_threshold: float = 0.01) -> pd.DataFrame:
    """
    Identify values in the bottom percentile of frequency.
    These are hunting leads — statistically unusual activity.

    Args:
        df: DataFrame with log data
        field: Column to analyze
        percentile_threshold: Bottom X% of frequency (default 1%)
    """
    counts = df[field].value_counts()
    total = len(df)

    threshold = counts.quantile(percentile_threshold)
    rare_values = counts[counts <= threshold]

    result = pd.DataFrame({
        'value': rare_values.index,
        'count': rare_values.values,
        'frequency_pct': (rare_values.values / total * 100).round(4)
    })

    return result.sort_values('count')

# Example: hunt for rare UserAgent strings in proxy logs
# Common: Chrome, Firefox, Edge = thousands of occurrences
# Rare: curl/7.68.0, python-requests/2.28 = 1-5 occurrences
# Curl/python in enterprise = likely attacker tool or misconfigured app

import numpy as np
from scipy import stats

def detect_beaconing(timestamps: list, threshold_jitter: float = 0.1) -> dict:
    """
    Detect C2 beaconing by analyzing inter-arrival time regularity.
    Legitimate traffic is irregular; C2 beacons are periodic.

    Args:
        timestamps: List of Unix timestamps for connections to a destination
        threshold_jitter: Coefficient of variation below which beaconing suspected
    """
    if len(timestamps) < 10:
        return {"beaconing": False, "reason": "Insufficient data points"}

    timestamps_sorted = sorted(timestamps)
    intervals = np.diff(timestamps_sorted)

    if len(intervals) == 0:
        return {"beaconing": False, "reason": "Single connection"}

    mean_interval = np.mean(intervals)
    std_interval = np.std(intervals)
    cv = std_interval / mean_interval if mean_interval > 0 else float('inf')

    # Low coefficient of variation = regular timing = beaconing
    is_beaconing = cv < threshold_jitter

    return {
        "beaconing": is_beaconing,
        "mean_interval_seconds": round(mean_interval, 1),
        "std_interval_seconds": round(std_interval, 1),
        "coefficient_of_variation": round(cv, 3),
        "connection_count": len(timestamps),
        "confidence": "HIGH" if cv < 0.05 else "MEDIUM" if cv < 0.1 else "LOW",
        "likely_period": f"~{int(mean_interval)}s ({int(mean_interval/60)}min)" if mean_interval > 60 else f"~{int(mean_interval)}s"
    }

# KQL equivalent — beaconing detection in Sentinel
BEACONING_QUERY = """
// Detect regular connection intervals to external IPs (C2 beaconing)
let timeframe = 24h;
let min_connections = 10;
let max_cv = 0.1;  // Coefficient of variation threshold

CommonSecurityLog
| where TimeGenerated > ago(timeframe)
| where DeviceAction != "block"
| summarize
    timestamps = make_list(TimeGenerated),
    connection_count = count()
  by SourceIP, DestinationIP, DestinationPort
| where connection_count >= min_connections
| mv-expand timestamps to typeof(datetime)
| sort by SourceIP, DestinationIP, timestamps asc
| serialize
| extend interval = datetime_diff('second', timestamps, prev(timestamps))
| where prev(SourceIP) == SourceIP and prev(DestinationIP) == DestinationIP
| summarize
    mean_interval = avg(interval),
    std_interval = stdev(interval),
    count = count()
  by SourceIP, DestinationIP, DestinationPort
| extend cv = std_interval / mean_interval
| where cv < max_cv and mean_interval between (30 .. 3600)
| sort by cv asc
| project SourceIP, DestinationIP, DestinationPort,
    mean_interval_secs = round(mean_interval, 0),
    cv = round(cv, 3),
    connection_count = count,
    threat_level = iff(cv < 0.05, "HIGH", "MEDIUM")
"""

import networkx as nx
import pandas as pd
from collections import defaultdict

def build_lateral_movement_graph(auth_logs: pd.DataFrame) -> nx.DiGraph:
    """
    Build directed graph of authentication events.
    Nodes: hosts/users. Edges: authentication events with metadata.
    """
    G = nx.DiGraph()

    for _, row in auth_logs.iterrows():
        source = row['source_host']
        target = row['target_host']
        user = row['username']
        logon_type = row['logon_type']
        timestamp = row['timestamp']

        # Add/update edge with authentication count
        if G.has_edge(source, target):
            G[source][target]['count'] += 1
            G[source][target]['users'].add(user)
        else:
            G.add_edge(source, target,
                      count=1,
                      users={user},
                      first_seen=timestamp,
                      last_seen=timestamp,
                      logon_types={logon_type})

    return G

def find_unusual_paths(G: nx.DiGraph, normal_graph: nx.DiGraph) -> list:
    """
    Find paths present in current window but absent from baseline.
    New edges = potential lateral movement.
    """
    suspicious = []

    for u, v, data in G.edges(data=True):
        if not normal_graph.has_edge(u, v):
            # New authentication path — not seen in baseline
            suspicious.append({
                'source': u,
                'target': v,
                'count': data['count'],
                'users': list(data['users']),
                'first_seen': data['first_seen'],
                'risk': 'HIGH' if len(data['users']) == 1 else 'MEDIUM'
            })

    return sorted(suspicious, key=lambda x: x['count'])

def find_high_degree_nodes(G: nx.DiGraph, baseline_avg_degree: float) -> list:
    """
    Identify hosts with unusually high connectivity (scanning/spreading).
    """
    suspicious = []
    for node in G.nodes():
        out_degree = G.out_degree(node)
        if out_degree > baseline_avg_degree * 3:
            suspicious.append({
                'host': node,
                'connections_to': out_degree,
                'target_hosts': list(G.successors(node)),
                'indicator': 'Possible C2 spreading or port scanning'
            })
    return suspicious

def detect_pass_the_hash_pattern(auth_logs: pd.DataFrame) -> pd.DataFrame:
    """
    PtH pattern: same user, NTLM logon type 3, multiple target hosts,
    short time window, no interactive logon preceding.
    """
    ntlm_logons = auth_logs[
        (auth_logs['logon_type'] == 3) &
        (auth_logs['auth_package'] == 'NTLM')
    ].copy()

    # Group by user and 5-minute windows
    ntlm_logons['window'] = ntlm_logons['timestamp'].dt.floor('5min')

    spread = ntlm_logons.groupby(['username', 'window']).agg(
        unique_targets=('target_host', 'nunique'),
        connection_count=('target_host', 'count'),
        target_list=('target_host', list)
    ).reset_index()

    # Flag: same user authenticating to many hosts in 5 minutes
    return spread[spread['unique_targets'] >= 5].sort_values(
        'unique_targets', ascending=False
    )

class HuntAutomationPipeline:
    """
    Pipeline converting successful hunts into scheduled detections.
    Integrates with SIEM API for rule deployment.
    """

    def __init__(self, siem_client, sigma_converter):
        self.siem = siem_client
        self.converter = sigma_converter
        self.hunt_registry = {}  # Track hunt history

    def register_hunt(self, hunt_id: str, hypothesis: str,
                      query: str, data_sources: list, mitre_techniques: list):
        """Register a new hunt hypothesis."""
        self.hunt_registry[hunt_id] = {
            'hypothesis': hypothesis,
            'query': query,
            'data_sources': data_sources,
            'mitre_techniques': mitre_techniques,
            'runs': 0,
            'findings': 0,
            'false_positives': 0,
            'status': 'active'
        }

    def run_hunt(self, hunt_id: str, lookback_hours: int = 24) -> dict:
        """Execute hunt and analyze results."""
        hunt = self.hunt_registry[hunt_id]
        results = self.siem.query(hunt['query'], lookback_hours=lookback_hours)

        hunt['runs'] += 1
        hunt['last_run'] = 'now'
        hunt['last_result_count'] = len(results)

        return {
            'hunt_id': hunt_id,
            'result_count': len(results),
            'results': results,
            'auto_promote': self._should_promote(hunt, results)
        }

    def _should_promote(self, hunt: dict, results: list) -> bool:
        """Determine if hunt should be promoted to detection."""
        # Promote if: consistent findings, low FP rate, confirmed malicious
        if hunt['runs'] < 3:
            return False
        fp_rate = hunt['false_positives'] / max(hunt['findings'], 1)
        return hunt['findings'] > 0 and fp_rate < 0.1

    def promote_to_detection(self, hunt_id: str) -> str:
        """Convert successful hunt to Sigma rule and deploy."""
        hunt = self.hunt_registry[hunt_id]
        sigma_rule = self.converter.query_to_sigma(
            query=hunt['query'],
            title=f"Automated: {hunt['hypothesis'][:50]}",
            techniques=hunt['mitre_techniques']
        )
        rule_id = self.siem.deploy_rule(sigma_rule)
        hunt['status'] = 'promoted'
        return rule_id

HUNT_PROGRAM_METRICS = {
    # Volume metrics
    "hunts_per_month": {
        "target": 8,
        "description": "Minimum hunts executed per month",
        "maturity": {"L1": 1, "L2": 4, "L3": 8, "L4": 15, "L5": "continuous"}
    },
    "techniques_covered_pct": {
        "target": 0.30,
        "description": "ATT&CK techniques with at least one hunt completed",
        "maturity": {"L1": 0.05, "L2": 0.15, "L3": 0.30, "L4": 0.50, "L5": 0.75}
    },

    # Quality metrics
    "hunt_to_detection_rate": {
        "target": 0.25,
        "description": "Hunts that result in a new production detection rule",
        "note": "Higher is better — shows hunts are finding real gaps"
    },
    "finding_rate": {
        "target": 0.15,
        "description": "Hunts that find genuine suspicious activity",
        "note": "15% finding rate is healthy — too high = noisy hunts"
    },
    "time_to_hypothesis": {
        "target_hours": 4,
        "description": "Time from CTI report to first hunt execution",
        "note": "Measures intelligence-to-action pipeline speed"
    },

    # Impact metrics
    "mtth": {
        "name": "Mean Time to Hunt (MTTH)",
        "description": "Average time for new threat report to generate a hunt",
        "target_days": 2
    },
    "detection_coverage_delta": {
        "description": "Change in ATT&CK coverage from hunts vs. other means",
        "note": "Hunting should account for >30% of new coverage"
    }
}

def calculate_hunt_roi(hunts_this_quarter: list, incidents_found: int,
                       avg_incident_cost: float) -> dict:
    """
    Calculate ROI of hunting program.
    Incidents found early = lower remediation cost.
    """
    total_hunt_hours = sum(h['analyst_hours'] for h in hunts_this_quarter)
    analyst_cost_per_hour = 75  # Fully-loaded cost

    program_cost = total_hunt_hours * analyst_cost_per_hour
    # Early detection savings: 60% cost reduction vs. post-detection
    incident_savings = incidents_found * avg_incident_cost * 0.60

    return {
        "program_cost_usd": program_cost,
        "incident_savings_usd": incident_savings,
        "roi_multiple": round(incident_savings / program_cost, 1),
        "cost_per_hunt_hour": analyst_cost_per_hour,
        "breakeven_incidents": round(program_cost / (avg_incident_cost * 0.60), 1)
    }