Security Skills Atlas
threat hunting

Analyzing Ransomware Network Indicators

Identify ransomware network indicators including C2 beaconing patterns, TOR exit node connections, data exfiltration flows, and encryption key exchange via Zeek conn.log and NetFlow analysis

c2-beaconingexfiltrationnetflownetwork-forensicsransomwaretorzeek
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1041 on the official MITRE ATT&CK siteT1048 on the official MITRE ATT&CK siteT1071 on the official MITRE ATT&CK siteT1071.001 on the official MITRE ATT&CK siteT1486 on the official MITRE ATT&CK siteT1567.002 on the official MITRE ATT&CK siteT1573 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 siteCertificate Analysis on the official MITRE D3FEND siteContent Format Conversion on the official MITRE D3FEND siteFile Content Analysis on the official MITRE D3FEND siteFile Metadata Consistency Validation on the official MITRE D3FEND site

Overview

Before and during ransomware execution, adversaries establish C2 channels, exfiltrate data, and download encryption keys. This skill analyzes Zeek conn.log and NetFlow data to detect beaconing patterns (regular-interval callbacks), connections to known TOR exit nodes, large outbound data transfers, and suspicious DNS activity associated with ransomware families.

When to Use

  • When investigating security incidents that require analyzing ransomware network indicators
  • When building detection rules or threat hunting queries for this domain
  • When SOC analysts need structured procedures for this analysis type
  • When validating security monitoring coverage for related attack techniques

Prerequisites

  • Zeek conn.log files or NetFlow CSV/JSON exports
  • Python 3.8+ with standard library
  • TOR exit node list (fetched from Tor Project or threat intel feeds)
  • Optional: Known ransomware C2 IOC list

Steps

  1. Parse Connection Logs — Ingest Zeek conn.log (TSV) or NetFlow records into structured format
  2. Detect Beaconing Patterns — Calculate connection interval statistics (mean, stddev, coefficient of variation) to identify periodic callbacks
  3. Check TOR Exit Node Connections — Cross-reference destination IPs against current TOR exit node list
  4. Identify Data Exfiltration — Flag connections with unusually high outbound byte ratios to external IPs
  5. Analyze DNS Patterns — Detect DGA-like domain queries and high-entropy subdomains
  6. Score and Correlate — Apply composite risk scoring across all indicator types
  7. Generate Report — Produce structured report with timeline and MITRE ATT&CK mapping

Expected Output

  • JSON report with beaconing detections and interval statistics
  • TOR exit node connection alerts
  • Data exfiltration flow analysis
  • Composite ransomware risk score with MITRE mapping (T1071, T1573, T1041)
Source materials

References and resources

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

References 1

api-reference.md2.3 KB

Ransomware Network Indicator Analysis API Reference

Zeek conn.log Fields

Field Description Example
ts Connection timestamp (epoch) 1609459200.123
uid Unique connection ID CYxTKo2zkGkGbfJFi
id.orig_h Source IP 192.168.1.100
id.orig_p Source port 49152
id.resp_h Destination IP 185.220.101.1
id.resp_p Destination port 443
proto Protocol tcp
duration Connection duration (seconds) 0.5
orig_bytes Bytes sent by originator 1024
resp_bytes Bytes sent by responder 512
conn_state Connection state SF

Beaconing Detection Algorithm

1. Group connections by (src_ip, dst_ip, dst_port)
2. Sort timestamps within each group
3. Calculate intervals: t[i+1] - t[i]
4. Compute statistics:
   - mean_interval = mean(intervals)
   - stddev = stdev(intervals)
   - coefficient_of_variation = stddev / mean_interval
5. Flag as beaconing if CV < 0.3 (regular interval pattern)
   - CV < 0.1 = critical (highly regular)
   - CV 0.1-0.3 = high (moderately regular)

TOR Exit Node Detection

# Fetch current TOR exit node list
curl -s https://check.torproject.org/torbulkexitlist > tor_exits.txt
 
# Alternative: Dan.me.uk TOR list
curl -s https://www.dan.me.uk/torlist/?exit > tor_exits_alt.txt
 
# Cross-reference with Zeek conn.log
zeek-cut id.resp_h < conn.log | sort -u | comm -12 - tor_exits_sorted.txt

RITA (Real Intelligence Threat Analytics) for Zeek

# Import Zeek logs into RITA
rita import /opt/zeek/logs/current rita_db
 
# Analyze beaconing
rita show-beacons rita_db
 
# Show long connections
rita show-long-connections rita_db
 
# DNS analysis
rita show-exploded-dns rita_db

Zeek CLI for Live Capture

# Analyze PCAP with Zeek
zeek -r capture.pcap
 
# Live capture on interface
zeek -i eth0 local.zeek
 
# Extract conn.log fields
zeek-cut ts id.orig_h id.resp_h id.resp_p orig_bytes resp_bytes < conn.log

MITRE ATT&CK Mapping

Technique ID Network Indicator
Application Layer Protocol T1071 C2 beaconing patterns
Encrypted Channel T1573 TOR/encrypted C2 traffic
Exfiltration Over C2 Channel T1041 High outbound byte ratio
Data Encrypted for Impact T1486 Ransomware encryption

Scripts 1

agent.py8.7 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Detect ransomware network indicators: C2 beaconing, TOR connections, data exfiltration via Zeek/NetFlow."""

import json
import csv
import argparse
import urllib.request
from datetime import datetime
from collections import defaultdict
from statistics import mean, stdev

TOR_EXIT_LIST_URL = "https://check.torproject.org/torbulkexitlist"


def parse_zeek_conn_log(log_path):
    """Parse Zeek conn.log TSV format into structured records."""
    connections = []
    with open(log_path) as f:
        headers = None
        for line in f:
            if line.startswith("#fields"):
                headers = line.strip().split("\t")[1:]
                continue
            if line.startswith("#"):
                continue
            if headers:
                fields = line.strip().split("\t")
                record = {}
                for i, h in enumerate(headers):
                    record[h] = fields[i] if i < len(fields) else "-"
                connections.append(record)
    return connections


def parse_netflow_csv(log_path):
    """Parse NetFlow CSV export into connection records."""
    connections = []
    with open(log_path) as f:
        reader = csv.DictReader(f)
        for row in reader:
            connections.append({
                "ts": row.get("timestamp", row.get("start_time", "")),
                "id.orig_h": row.get("src_ip", row.get("sa", "")),
                "id.resp_h": row.get("dst_ip", row.get("da", "")),
                "id.resp_p": row.get("dst_port", row.get("dp", "")),
                "proto": row.get("protocol", row.get("pr", "")),
                "orig_bytes": row.get("src_bytes", row.get("ibyt", "0")),
                "resp_bytes": row.get("dst_bytes", row.get("obyt", "0")),
                "duration": row.get("duration", row.get("td", "0")),
            })
    return connections


def fetch_tor_exit_nodes():
    """Fetch current TOR exit node IP list from Tor Project."""
    try:
        req = urllib.request.Request(TOR_EXIT_LIST_URL, headers={"User-Agent": "SecurityAgent/1.0"})
        with urllib.request.urlopen(req, timeout=15) as resp:
            nodes = set()
            for line in resp.read().decode().splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    nodes.add(line)
            return nodes
    except Exception as e:
        return set()


def detect_beaconing(connections, min_connections=10, max_cv=0.3):
    """Detect C2 beaconing by analyzing connection interval regularity."""
    pair_timestamps = defaultdict(list)
    for conn in connections:
        src = conn.get("id.orig_h", "")
        dst = conn.get("id.resp_h", "")
        port = conn.get("id.resp_p", "")
        ts = conn.get("ts", "")
        if src and dst and ts != "-":
            try:
                pair_timestamps[(src, dst, port)].append(float(ts))
            except (ValueError, TypeError):
                continue

    beacons = []
    for (src, dst, port), timestamps in pair_timestamps.items():
        if len(timestamps) < min_connections:
            continue
        timestamps.sort()
        intervals = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)]
        if not intervals:
            continue
        avg_interval = mean(intervals)
        if avg_interval == 0:
            continue
        sd = stdev(intervals) if len(intervals) > 1 else 0
        cv = sd / avg_interval if avg_interval > 0 else 999

        if cv <= max_cv:
            beacons.append({
                "source": src, "destination": dst, "port": port,
                "connection_count": len(timestamps),
                "avg_interval_seconds": round(avg_interval, 2),
                "interval_stddev": round(sd, 2),
                "coefficient_of_variation": round(cv, 4),
                "severity": "critical" if cv < 0.1 else "high",
                "indicator": "Regular beaconing pattern detected",
                "mitre": "T1071",
            })
    return beacons


def detect_tor_connections(connections, tor_nodes):
    """Cross-reference connections against TOR exit node list."""
    tor_hits = []
    for conn in connections:
        dst = conn.get("id.resp_h", "")
        src = conn.get("id.orig_h", "")
        if dst in tor_nodes:
            tor_hits.append({
                "source": src, "destination": dst,
                "port": conn.get("id.resp_p", ""),
                "bytes_sent": conn.get("orig_bytes", "0"),
                "severity": "high",
                "indicator": "Connection to TOR exit node",
                "mitre": "T1573",
            })
    unique_tor = len({h["destination"] for h in tor_hits})
    return tor_hits, unique_tor


def detect_exfiltration(connections, byte_threshold=100_000_000):
    """Detect potential data exfiltration by high outbound byte transfer."""
    pair_bytes = defaultdict(lambda: {"sent": 0, "received": 0, "count": 0})
    for conn in connections:
        src = conn.get("id.orig_h", "")
        dst = conn.get("id.resp_h", "")
        if not src or not dst:
            continue
        if dst.startswith(("10.", "192.168.", "172.16.", "127.")):
            continue
        try:
            sent = int(conn.get("orig_bytes", 0)) if conn.get("orig_bytes", "-") != "-" else 0
            recv = int(conn.get("resp_bytes", 0)) if conn.get("resp_bytes", "-") != "-" else 0
        except ValueError:
            continue
        pair_bytes[(src, dst)]["sent"] += sent
        pair_bytes[(src, dst)]["received"] += recv
        pair_bytes[(src, dst)]["count"] += 1

    exfil_alerts = []
    for (src, dst), stats in pair_bytes.items():
        if stats["sent"] > byte_threshold:
            ratio = stats["sent"] / max(stats["received"], 1)
            exfil_alerts.append({
                "source": src, "destination": dst,
                "bytes_sent": stats["sent"],
                "bytes_received": stats["received"],
                "send_receive_ratio": round(ratio, 2),
                "connection_count": stats["count"],
                "severity": "critical" if stats["sent"] > byte_threshold * 5 else "high",
                "indicator": "High outbound data transfer (potential exfiltration)",
                "mitre": "T1041",
            })
    return exfil_alerts


def generate_report(connections, beacons, tor_hits, tor_unique, exfil_alerts, source):
    """Generate ransomware network indicator report."""
    total_alerts = len(beacons) + len(tor_hits) + len(exfil_alerts)
    risk_score = min(100, len(beacons) * 20 + tor_unique * 15 + len(exfil_alerts) * 25)
    return {
        "report_time": datetime.utcnow().isoformat(),
        "log_source": source,
        "total_connections": len(connections),
        "ransomware_risk_score": risk_score,
        "risk_level": "critical" if risk_score >= 70 else "high" if risk_score >= 40 else "medium",
        "beaconing_detections": beacons,
        "tor_connection_alerts": len(tor_hits),
        "tor_unique_nodes": tor_unique,
        "tor_hits_sample": tor_hits[:10],
        "exfiltration_alerts": exfil_alerts,
        "mitre_techniques": ["T1071", "T1573", "T1041", "T1486"],
        "total_alerts": total_alerts,
    }


def main():
    parser = argparse.ArgumentParser(description="Ransomware Network Indicator Analyzer")
    parser.add_argument("--input", required=True, help="Zeek conn.log or NetFlow CSV file")
    parser.add_argument("--format", choices=["zeek", "netflow"], default="zeek")
    parser.add_argument("--output", default="ransomware_network_report.json")
    parser.add_argument("--beacon-min", type=int, default=10, help="Min connections for beaconing")
    parser.add_argument("--exfil-threshold", type=int, default=100_000_000, help="Exfil byte threshold")
    parser.add_argument("--skip-tor", action="store_true", help="Skip TOR exit node check")
    args = parser.parse_args()

    if args.format == "zeek":
        connections = parse_zeek_conn_log(args.input)
    else:
        connections = parse_netflow_csv(args.input)

    print(f"[*] Parsed {len(connections)} connections from {args.input}")
    beacons = detect_beaconing(connections, min_connections=args.beacon_min)
    tor_nodes = set() if args.skip_tor else fetch_tor_exit_nodes()
    tor_hits, tor_unique = detect_tor_connections(connections, tor_nodes)
    exfil_alerts = detect_exfiltration(connections, args.exfil_threshold)

    report = generate_report(connections, beacons, tor_hits, tor_unique, exfil_alerts, args.input)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Beaconing: {len(beacons)} | TOR: {len(tor_hits)} ({tor_unique} nodes) | Exfil: {len(exfil_alerts)}")
    print(f"[+] Ransomware risk score: {report['ransomware_risk_score']}/100 ({report['risk_level']})")
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring
Hunting for Cobalt Strike BeaconsDetect Cobalt Strike beacon network activity using default TLS certificate signatures (serial 8BB00EE), JA3/JA3S/JARM fingerprints, HTTP C2 profile pattern matching, beacon jitter analysis, and named pipe detection via Zeek, Suricata, and Python PCAP analysis.Hunting for Data Staging Before ExfiltrationDetect data staging activity before exfiltration by monitoring for archive creation with 7-Zip/RAR, unusual temp folder access, large file consolidation, and staging directory patterns via EDR and process telemetryHunting for DNS Tunneling with ZeekDetect DNS tunneling and data exfiltration by analyzing Zeek dns.log for high-entropy subdomain queries, excessive query volume, long query lengths, and unusual DNS record types indicating covert channel communication.Hunting For Shadow Copy DeletionHunt for Volume Shadow Copy deletion activity that indicates ransomware preparation or anti-forensics by monitoring vssadmin, wmic, and PowerShell shadow copy commands.Analyzing Persistence Mechanisms in LinuxDetect and analyze Linux persistence mechanisms including crontab entries, systemd service units, LD_PRELOAD hijacking, bashrc modifications, and authorized_keys backdoors using auditd and file integrity monitoringAnalyzing PowerShell Empire ArtifactsDetect PowerShell Empire framework artifacts in Windows event logs by identifying Base64 encoded launcher patterns, default user agents, staging URL structures, stager IOCs, and known Empire module signatures in Script Block Logging events.