Process-to-network attribution for sandbox analyses

Tie every network artifact CAPE captures (suricata alerts/tls/http/files,
network.tcp/udp/dns/hosts) back to the originating Windows process so
analysts and downstream signatures can answer "which process did this?"
without manual correlation work.

Sources, in confidence order:

  1. Sysmon Event ID 3 (NetworkConnect) from evtx.zip — full image path,
     destination hostname, src/dst 5-tuple
  2. Microsoft-Windows-Kernel-Network ETW captured live by a new analyzer
     auxiliary, periodically uploaded so attribution survives crashes
  3. Sigma EID 3 matched events — tertiary catch for late-fire flows
  4. DNS-Client ETW (originating-process DNS) cross-referenced with
     suricata.dns/network.dns/network.hosts/sigma EID 22 QueryResults
     to attribute IPs we never saw a direct connect for. Avoids the
     "everything routes to svchost (dnscache)" failure mode.
  5. Sysmon EID 22 (DnsQuery) — covers queries that fired before the
     DNS-ETW auxiliary subscribed (early CDN resolutions, etc.)
  6. Sysmon EID 1 (ProcessCreate) — names processes that aren't
     monitored by capemon but show up in connection/DNS data

A single AttributionIndex consumes all sources; per-target enrichment
(suricata.alerts/tls/http/files, network.tcp/udp/dns/hosts, sigma
detections) goes through one of four query methods. 5-tuple matching
with src_port disambiguates multi-process flows to the same destination.

New analyzer aux:
  analyzer/windows/modules/auxiliary/network_etw.py
    Captures TCP/UDP connect events from Microsoft-Windows-Kernel-Network
    with periodic full-snapshot uploads. Off by default in
    auxiliary.conf.default — opt-in with [Network_ETW] enabled = yes.

Stop-priority on Network_ETW + Evtx:
  Set stop_priority = -20 so they shut down AFTER the capemon-related
  auxiliaries. Late-fire C2 callbacks that fire between the analysis-
  stop signal and VM teardown still get captured + attributed.

Result-server transport:
  resultserver.py refactor (allowlisted RESULT_UPLOADABLE +
  RESULT_DIRECTORIES) tightens path-traversal protection and adds an
  is_replaceable_result_upload() helper so periodic re-uploads from
  the auxiliaries (tlsdump.log, dns_etw.json, network_etw.json,
  wmi_etw.json, sslkeylogfile/sslkeys.log) truncate-write instead of
  silently failing with EEXIST after the first upload.

Decryption pipeline:
  decryptpcap.py wraps gogorobocap to produce dump_decrypted.pcap +
  dump_mixed.pcap. network.py + suricata.py honour a new pcapsrc
  config knob (auto/original/mixed/decrypted) so users explicitly
  choose which pcap variant to analyse.

UI:
  _suricata_http.html, _suricata_files.html, _hosts.html render
  process attribution columns/rows, gated on the existing
  NETWORK_PROC_MAP setting. _hosts.html shows multi-process attribution
  as badges with hover tooltips that explain the attribution chain
  (direct connect / DNS-resolved / etc). Pre-existing "asn cell skipped
  when empty causes column shift" bug fixed in the same edit.

Tests:
  tests/test_network_capture_integration.py covers AttributionIndex
  build + query, pcapsrc resolution, and replaceable-upload behaviour.

Author: wmetcalf

analyzer/windows/modules/auxiliary/evtx.py

@@ -12,6 +12,15 @@
 
 
 class Evtx(Thread, Auxiliary):
+    # Stop AFTER capemon-related auxiliaries so the final EVTX snapshot
+    # captures sysmon events from late-fire callbacks that fire between
+    # the analysis-stopping signal and the VM teardown (e.g. C2 callbacks
+    # the malware schedules after a delay). Without this priority bump,
+    # those events happen after the last EVTX snapshot and never reach
+    # the host-side processing modules.
+    start_priority = 0
+    stop_priority = -20
+
     evtx_dump = "evtx.zip"
 
     # Event log channels to collect

analyzer/windows/modules/auxiliary/network_etw.py

@@ -0,0 +1,242 @@
+import json
+import logging
+import os
+import shutil
+import socket
+import time
+from threading import Thread
+
+from lib.common.results import upload_to_host
+from lib.common.rand import random_string
+from lib.core.config import Config
+from lib.common.etw_utils import (
+    ETWAuxiliaryWrapper,
+    ETWProviderWrapper,
+    HAVE_ETW,
+    ProviderInfo,
+    GUID,
+    et,
+    encode,
+)
+
+log = logging.getLogger(__name__)
+
+__author__ = "DNS-GEE-O (@wmetcalf)"
+
+KERNEL_NETWORK_GUID = "{7DD42A49-5329-4832-8DFD-43D979153A88}"
+
+CONNECT_EVENT_IDS = [12, 15, 28, 31, 42, 58]
+
+EVENT_NAMES = {
+    12: "tcp_connect_v4",
+    15: "tcp_accept_v4",
+    28: "tcp_connect_v6",
+    31: "tcp_accept_v6",
+    42: "udp_send_v4",
+    58: "udp_send_v6",
+}
+
+# Periodic upload interval in seconds
+UPLOAD_INTERVAL = 15
+
+
+if HAVE_ETW:
+
+    class NetworkETWProvider(ETWProviderWrapper):
+        def __init__(
+            self,
+            level=et.TRACE_LEVEL_INFORMATION,
+            logfile=None,
+            no_conout=False,
+            any_keywords=None,
+            all_keywords=None,
+            filter_ips=None,
+            filter_ports=None,
+        ):
+            self._filter_ips = filter_ips or set()
+            self._filter_ports = filter_ports or set()
+
+            providers = [
+                ProviderInfo(
+                    "Microsoft-Windows-Kernel-Network",
+                    GUID(KERNEL_NETWORK_GUID),
+                    level,
+                    any_keywords or 0x30,
+                    all_keywords,
+                )
+            ]
+            super().__init__(
+                session_name="ETW_KernelNetwork",
+                providers=providers,
+                event_id_filters=CONNECT_EVENT_IDS,
+                logfile=logfile,
+                no_conout=no_conout,
+            )
+
+        def _should_filter(self, event, event_id):
+            src_ip = str(event.get("saddr", ""))
+            dst_ip = str(event.get("daddr", ""))
+            src_port = event.get("sport", 0)
+            dst_port = event.get("dport", 0)
+
+            # Try int conversion for port comparison
+            try:
+                src_port = int(src_port)
+            except (ValueError, TypeError):
+                pass
+            try:
+                dst_port = int(dst_port)
+            except (ValueError, TypeError):
+                pass
+
+            if dst_ip in self._filter_ips:
+                return True
+            if event_id in (15, 31) and src_ip in self._filter_ips:
+                return True
+            if dst_port in self._filter_ports or src_port in self._filter_ports:
+                return True
+            if dst_ip in ("127.0.0.1", "::1", "0.0.0.0", ""):
+                return True
+            return False
+
+        def on_event(self, event_tufo):
+            event_id, event = event_tufo
+            if event_id not in self.event_id_filters:
+                return
+            if self._should_filter(event, event_id):
+                return
+            if self.logfile:
+                self.write_to_log(self.logfile, event_id, event)
+
+        def write_to_log(self, file_handle, event_id, event):
+            header = event.get("EventHeader", {})
+            pid = event.get("PID") or header.get("ProcessId", 0)
+            proto = "TCP" if event_id in (12, 15, 28, 31) else "UDP"
+            direction = "outbound" if event_id in (12, 28, 42, 58) else "inbound"
+
+            entry = {
+                "event_type": EVENT_NAMES.get(event_id, "unknown"),
+                "event_id": event_id,
+                "pid": pid,
+                "protocol": proto,
+                "direction": direction,
+                "src_ip": str(event.get("saddr", "")),
+                "src_port": event.get("sport", 0),
+                "dst_ip": str(event.get("daddr", "")),
+                "dst_port": event.get("dport", 0),
+                "timestamp": str(header.get("TimeStamp", "")),
+            }
+            connid = event.get("connid")
+            if connid:
+                entry["connid"] = connid
+
+            json.dump(entry, file_handle)
+            file_handle.write("\n")
+
+
+class Network_ETW(ETWAuxiliaryWrapper):
+    """Captures TCP/UDP connection events via Microsoft-Windows-Kernel-Network ETW.
+
+    Provides process-to-network 5-tuple mapping.
+    Periodically uploads captured data to ensure availability if analysis
+    terminates unexpectedly.
+
+    Output: aux/network_etw.json (NDJSON)
+    """
+
+    # Stop AFTER capemon-related modules so late-firing network calls get attributed
+    start_priority = 0
+    stop_priority = -20
+
+    def __init__(self, options, config):
+        super().__init__(options, config, "network_etw")
+
+        self.output_dir = os.path.join("C:\\", random_string(5, 10))
+        try:
+            os.mkdir(self.output_dir)
+        except FileExistsError:
+            pass
+
+        self.log_file_path = os.path.join(self.output_dir, "%s.log" % random_string(5, 10))
+        self.log_file = None
+        self._do_periodic = False
+        self._periodic_thread = None
+
+        if HAVE_ETW and self.enabled:
+            filter_ips = set()
+            filter_ports = set()
+
+            try:
+                analysis_cfg = Config(cfg="analysis.conf")
+                host_ip = getattr(analysis_cfg, "ip", "")
+                if host_ip:
+                    filter_ips.add(host_ip)
+                rs_port = getattr(analysis_cfg, "port", 0)
+                if rs_port:
+                    filter_ports.add(int(rs_port))
+            except Exception as e:
+                log.debug("Could not read analysis config for filters: %s", e)
+
+            filter_ports.add(8000)
+            filter_ports.add(53)
+
+            log.info("NetworkETW filters: ips=%s ports=%s", filter_ips, filter_ports)
+
+            try:
+                self.log_file = open(self.log_file_path, "w", encoding="utf-8")
+                self.capture = NetworkETWProvider(
+                    logfile=self.log_file,
+                    level=255,
+                    no_conout=True,
+                    filter_ips=filter_ips,
+                    filter_ports=filter_ports,
+                )
+            except Exception as e:
+                log.error("Failed to open Network ETW log file: %s", e)
+
+    def start(self):
+        result = super().start()
+        # Start periodic upload thread
+        if self.enabled and self.log_file:
+            self._do_periodic = True
+            self._periodic_thread = Thread(target=self._periodic_upload, daemon=True)
+            self._periodic_thread.start()
+        return result
+
+    def _periodic_upload(self):
+        """Periodically flush and upload current data."""
+        while self._do_periodic:
+            for _ in range(UPLOAD_INTERVAL):
+                if not self._do_periodic:
+                    break
+                time.sleep(1)
+            if self._do_periodic and self.log_file:
+                try:
+                    self.log_file.flush()
+                    # Copy the file so we don't interfere with ongoing writes
+                    snap_path = self.log_file_path + ".snap"
+                    shutil.copy2(self.log_file_path, snap_path)
+                    upload_to_host(snap_path, os.path.join("aux", "network_etw.json"))
+                    log.debug("Periodic network_etw upload: %d bytes", os.path.getsize(snap_path))
+                    os.remove(snap_path)
+                except Exception as e:
+                    log.debug("Periodic network_etw upload failed: %s", e)
+
+    def upload_results(self):
+        """Final upload on stop."""
+        self._do_periodic = False
+        if self._periodic_thread:
+            self._periodic_thread.join(timeout=5)
+
+        if self.log_file:
+            try:
+                self.log_file.close()
+            except Exception:
+                pass
+            self.log_file = None
+
+        if os.path.isfile(self.log_file_path) and os.path.getsize(self.log_file_path) > 0:
+            try:
+                upload_to_host(self.log_file_path, os.path.join("aux", "network_etw.json"))
+            except Exception as e:
+                log.error("Final network_etw upload failed: %s", e)

conf/default/auxiliary.conf.default

@@ -32,6 +32,7 @@ browsermonitor = no
 wmi_etw = no
 dns_etw = no
 amsi_etw = no
+network_etw = no
 watchdownloads = no
 
 [AzSniffer]

conf/default/processing.conf.default

@@ -101,6 +101,9 @@ enabled = no
 [network]
 enabled = yes
 sort_pcap = no
+# Which capture to analyze when decryptpcap produced additional outputs:
+# auto | original | mixed | decrypted
+pcapsrc = auto
 # Enable mapping of network events to specific processes using behavioral analysis data
 process_map = no
 # Adds network connections seen in behavior but not in PCAP. Requires process_map = yes
@@ -181,6 +184,9 @@ urlscrub = (^http:\/\/serw\.clicksor\.com\/redir\.php\?url=|&InjectedParam=.+$)
 [suricata]
 enabled = no
 runmode = cli
+# Which capture to analyze when decryptpcap produced additional outputs:
+# auto | original | mixed | decrypted
+pcapsrc = auto
 # Outputfiles
 # if evelog is specified, it will be used instead of the per-protocol log files
 evelog = eve.json
@@ -248,6 +254,16 @@ file_cache = no
 # Store pefile objects for later usage? useful if you doing something in signatures/reporting
 pefile_store = no
 
+[decryptpcap]
+enabled = no
+gogorobocap = data/gogorobocap/gogorobocap-linux-amd64
+# Select how decrypted captures are generated:
+# auto | pcap_with_keylog | sslproxy_synth_pcap
+pcapsrc = auto
+
+[network_etw]
+enabled = no
+
 # Deduplicate screenshots - You need to install dependency ImageHash>=4.3.1
 [deduplication]
 #
@@ -348,4 +364,3 @@ enabled = no
 # Enable when using the PolarProxy option during analysis. This will merge the tls.pcap containing
 # plain-text TLS streams into the task PCAP.
 enabled = no
-

lib/cuckoo/core/resultserver.py

@@ -81,6 +81,14 @@
 
 RESULT_DIRECTORIES = RESULT_UPLOADABLE + (b"reports", b"logs")
 
+REPLACEABLE_RESULT_UPLOADS = (
+    b"tlsdump/",
+    b"aux/dns_etw.json",
+    b"aux/network_etw.json",
+    b"aux/wmi_etw.json",
+    b"aux/sslkeylogfile/sslkeys.log",
+)
+
 
 def netlog_sanitize_fname(path):
     """Validate agent-provided path for result files"""
@@ -90,11 +98,17 @@ def netlog_sanitize_fname(path):
         raise CuckooOperationalError(f"Netlog client requested banned path: {path}")
     if any(c in BANNED_PATH_CHARS for c in name):
         for c in BANNED_PATH_CHARS:
-            path.replace(bytes([c]), b"X")
+            path = path.replace(bytes([c]), b"X")
 
     return path
 
 
+def is_replaceable_result_upload(path):
+    """Return True for result uploads that are expected to overwrite prior
+    content with a full snapshot rather than append a distinct artifact."""
+    return path.startswith(REPLACEABLE_RESULT_UPLOADS)
+
+
 class Disconnect(Exception):
     pass
 
@@ -254,7 +268,15 @@ def handle(self):
             try:
                 if file_path.endswith("_script.log"):
                     self.fd = open_inclusive(file_path)
-                elif not path_exists(file_path):
+                elif is_replaceable_result_upload(dump_path) and path_exists(file_path):
+                    # Auxiliary modules (tlsdump, network_etw, sslkeylogfile…)
+                    # upload the SAME dump_path periodically so accumulated
+                    # key / connection data survives an unexpected analysis
+                    # termination. Each upload is a full replacement of the
+                    # prior content — truncate and rewrite rather than failing
+                    # silently with EEXIST.
+                    self.fd = open(file_path, "wb")
+                else:
                     # open_exclusive will fail if file_path already exists
                     self.fd = open_exclusive(file_path)
             except OSError as e: