Fix bugs in NetTrace

.github/workflows/ci.yml

@@ -14,11 +14,12 @@ concurrency:
 
 jobs:
   test:
-    name: Python ${{ matrix.python-version }}
-    runs-on: ubuntu-latest
+    name: Python ${{ matrix.python-version }} (${{ matrix.os }})
+    runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
       matrix:
+        os: [ubuntu-latest, windows-latest]
         python-version: ["3.11", "3.12"]
 
     steps:
@@ -38,7 +39,7 @@ jobs:
         run: python -m pytest --cov=nettrace --cov-report=term --cov-report=xml --cov-fail-under=80
 
       - name: Upload coverage report
-        if: matrix.python-version == '3.12'
+        if: matrix.python-version == '3.12' && matrix.os == 'ubuntu-latest'
         uses: actions/upload-artifact@v7
         with:
           name: coverage-xml

nettrace/analysis/beaconing.py

@@ -37,9 +37,11 @@ def detect_beaconing(flows: list[Flow], thresholds: dict) -> list[Finding]:
         mean_interval = 0.0
         interval_m2 = 0.0
         previous_timestamp: float | None = None
+        contributing_indices: set[int] = set()
         while heap and processed_events < max_group_events:
             timestamp, activity_index, timestamp_index = heapq.heappop(heap)
             processed_events += 1
+            contributing_indices.add(activity_index)
             if previous_timestamp is not None:
                 interval = timestamp - previous_timestamp
                 if interval > 0:
@@ -61,12 +63,41 @@ def detect_beaconing(flows: list[Flow], thresholds: dict) -> list[Finding]:
         cv = stdev / mean_interval if mean_interval else 999.0
         if cv <= max_cv:
             flow = activities[0][0]
+            # Bug #7: aggregate evidence across every flow that fed the beacon
+            # calculation, not just the first one -- a rotating-source-port
+            # beacon spans multiple flows, and the old code silently attributed
+            # all 30+ events to a single connection's packet numbers.
+            contributing_flows = [activities[index][0] for index in sorted(contributing_indices)]
+            packet_numbers_sample: list[int] = []
+            first_packet_number = 0
+            wireshark_numbers: list[int] = []
+            for contributing_flow in contributing_flows:
+                flow_evidence = flow_packet_evidence(contributing_flow, limit=4)
+                sample = flow_evidence.get("packet_numbers_sample", [])
+                wireshark_numbers.extend(sample)
+                if sample and not packet_numbers_sample:
+                    packet_numbers_sample = sample
+                if not first_packet_number and flow_evidence.get("first_packet_number"):
+                    first_packet_number = flow_evidence["first_packet_number"]
+            observation_window = max((f.last_seen for f in contributing_flows), default=flow.last_seen) - min(
+                (f.first_seen for f in contributing_flows), default=flow.first_seen
+            )
+            # Bug #6: confidence reflects how strong the signal actually is --
+            # more events and a tighter coefficient of variation is stronger
+            # evidence than a borderline 5-event, cv=0.24 group.
+            if processed_events >= 50 and cv <= 0.05:
+                confidence = "high"
+            elif processed_events >= 10 and cv <= 0.15:
+                confidence = "medium"
+            else:
+                confidence = "low"
             findings.append(
                 Finding(
                     title="Possible beaconing behavior",
                     description="Regular connection timing suggests command-and-control beaconing.",
                     category="dns_beaconing" if flow.dst_port == 53 else "network_beaconing",
                     timestamp=flow.first_seen,
+                    confidence=confidence,
                     evidence={
                         "src_ip": flow.src_ip,
                         "dst_ip": flow.dst_ip,
@@ -76,7 +107,15 @@ def detect_beaconing(flows: list[Flow], thresholds: dict) -> list[Finding]:
                         "timing_events_truncated": bool(heap),
                         "mean_interval_seconds": round(mean_interval, 3),
                         "coefficient_of_variation": round(cv, 3),
-                        **flow_packet_evidence(flow),
+                        "connection_count": len(contributing_flows),
+                        "observation_window_seconds": round(observation_window, 3),
+                        "first_packet_number": first_packet_number or flow.first_packet_number,
+                        "packet_numbers_sample": wireshark_numbers[:8],
+                        "wireshark_filter": (
+                            "frame.number in {" + " ".join(str(n) for n in wireshark_numbers[:8]) + "}"
+                            if wireshark_numbers
+                            else ""
+                        ),
                     },
                     tags=["beaconing"],
                 )

nettrace/analysis/dga_scorer.py

@@ -37,21 +37,27 @@
 @lru_cache(maxsize=8)
 def _load_dga_allowlist(path: Path = DEFAULT_ALLOWLIST_PATH) -> dict[str, list[str]]:
     if not path.exists():
-        return {"suffixes": [], "contains": [], "regexes": []}
+        return {"domains": [], "suffixes": [], "regexes": []}
     data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
     return {
-        "suffixes": [item.lower() for item in data.get("suffixes", [])],
-        "contains": [item.lower() for item in data.get("contains", [])],
+        "domains": [item.lower().lstrip(".") for item in data.get("domains", [])],
+        "suffixes": [item.lower().lstrip(".") for item in data.get("suffixes", [])],
         "regexes": data.get("regexes", []),
     }
 
 
+def _matches_suffix(normalized: str, suffix: str) -> bool:
+    # Boundary-safe: "attacker-microsoft.com".endswith("microsoft.com") would be a
+    # false allow, so require an exact match or a "." boundary before the suffix.
+    return normalized == suffix or normalized.endswith("." + suffix)
+
+
 def is_allowlisted_domain(domain: str, allowlist: dict[str, list[str]] | None = None) -> bool:
     rules = _load_dga_allowlist() if allowlist is None else allowlist
     normalized = domain.lower().rstrip(".")
-    if any(normalized.endswith(suffix) for suffix in rules.get("suffixes", [])):
+    if any(normalized == exact for exact in rules.get("domains", [])):
         return True
-    if any(token in normalized for token in rules.get("contains", [])):
+    if any(_matches_suffix(normalized, suffix) for suffix in rules.get("suffixes", [])):
         return True
     return any(re.search(pattern, normalized, re.IGNORECASE) for pattern in rules.get("regexes", []))
 
@@ -64,25 +70,69 @@ def shannon_entropy(value: str) -> float:
     return -sum((count / length) * math.log2(count / length) for count in counts.values())
 
 
+# Labels that never carry a DGA signal on their own -- generic subdomain/hostname
+# prefixes seen in front of a registered domain (www.<random>.biz, cdn.<random>.biz).
+_GENERIC_PREFIX_LABELS = {
+    "www", "cdn", "api", "mail", "smtp", "ftp", "ns1", "ns2", "vpn", "mx",
+    "static", "img", "img1", "img2", "cdn1", "cdn2", "app", "m", "web",
+}
+
+
 def domain_label(domain: str) -> str:
+    """Backward-compatible single-label accessor: still the leftmost label.
+
+    Kept for callers/tests that only care about one label; scoring itself uses
+    candidate_labels() below so it isn't fooled by a benign www/cdn prefix.
+    """
     return domain.split(".")[0].lower()
 
 
-def dga_score(domain: str) -> float:
-    label = re.sub(r"[^a-z0-9]", "", domain_label(domain))
-    if len(label) < 8:
+def candidate_labels(domain: str) -> list[str]:
+    """Labels worth DGA-scoring, skipping a leading generic hostname prefix.
+
+    This is a lightweight approximation of eTLD+1 (no public-suffix-list
+    dependency): for "www.xj3k9q2z7m1p0a8c.biz" it returns
+    ["xj3k9q2z7m1p0a8c"], not ["www"]. It does not handle multi-part public
+    suffixes like ".co.uk" correctly -- that needs a real PSL lookup and is
+    tracked as a follow-up, not silently claimed as solved here.
+    """
+    labels = [label.lower() for label in domain.rstrip(".").split(".") if label]
+    if not labels:
+        return []
+    if len(labels) == 1:
+        return labels
+    if labels[0] in _GENERIC_PREFIX_LABELS and len(labels) > 2:
+        return labels[1:-1] or labels[:1]
+    return labels[:-1]
+
+
+def _label_score(label: str) -> float:
+    cleaned = re.sub(r"[^a-z0-9]", "", label)
+    if len(cleaned) < 8:
         return 0.0
-    entropy = shannon_entropy(label)
-    bigrams = [label[index : index + 2] for index in range(len(label) - 1)]
+    entropy = shannon_entropy(cleaned)
+    bigrams = [cleaned[index : index + 2] for index in range(len(cleaned) - 1)]
     common = sum(1 for bigram in bigrams if bigram in COMMON_BIGRAMS)
     common_ratio = common / max(1, len(bigrams))
-    digit_ratio = sum(1 for char in label if char.isdigit()) / len(label)
+    digit_ratio = sum(1 for char in cleaned if char.isdigit()) / len(cleaned)
     entropy_component = min(1.0, entropy / 4.0)
     language_component = 1.0 - common_ratio
     digit_component = min(1.0, digit_ratio * 2.0)
     return round((entropy_component * 0.5) + (language_component * 0.35) + (digit_component * 0.15), 3)
 
 
+def dga_score(domain: str) -> float:
+    """Highest DGA score across candidate labels. See scored_label() for which
+    label produced it."""
+    labels = candidate_labels(domain) or [domain_label(domain)]
+    return max((_label_score(label) for label in labels), default=0.0)
+
+
+def scored_label(domain: str) -> str:
+    labels = candidate_labels(domain) or [domain_label(domain)]
+    return max(labels, key=_label_score, default=domain_label(domain))
+
+
 def score_domains(dns_events: list[DNSEvent], thresholds: dict) -> list[Finding]:
     findings: list[Finding] = []
     score_threshold = float(thresholds.get("dga_score_threshold", 0.6))
@@ -96,17 +146,20 @@ def score_domains(dns_events: list[DNSEvent], thresholds: dict) -> list[Finding]
         seen.add(normalized_query)
         if is_allowlisted_domain(event.query, allowlist):
             continue
+        label = scored_label(event.query)
         score = dga_score(event.query)
-        entropy = shannon_entropy(domain_label(event.query))
+        entropy = shannon_entropy(re.sub(r"[^a-z0-9]", "", label))
         if score >= score_threshold and entropy >= entropy_threshold:
             findings.append(
                 Finding(
                     title="Possible DGA domain",
                     description="Domain structure has high entropy and weak language-like character patterns.",
                     category="dga_domain",
                     timestamp=event.timestamp,
+                    confidence="high" if score >= 0.8 else "medium",
                     evidence={
                         "domain": event.query,
+                        "scored_label": label,
                         "entropy": round(entropy, 3),
                         "dga_score": score,
                         **packet_evidence(event.packet_number),

nettrace/analysis/ftp_analyzer.py

@@ -11,26 +11,56 @@ def analyze_ftp_events(events: list[FTPEvent]) -> list[Finding]:
         if event.command not in {"PASS", "STOR", "APPE"}:
             continue
         if event.command == "PASS":
-            title = "Cleartext FTP credentials"
-            description = "An FTP password command was observed over an unencrypted control channel."
+            # Bug #1: exposed credentials are not, by themselves, exfiltration --
+            # there's no clean single ATT&CK technique for "cleartext auth
+            # observed," so this stays untagged rather than forced into
+            # T1048.003. attck_tagger.py has no "ftp_cleartext_credentials"
+            # entry, so it is intentionally left without an ATT&CK mapping.
+            findings.append(
+                Finding(
+                    title="Cleartext FTP credentials",
+                    description="An FTP password command was observed over an unencrypted control channel.",
+                    category="ftp_cleartext_credentials",
+                    timestamp=event.timestamp,
+                    confidence="high",
+                    evidence={
+                        "src_ip": event.src_ip,
+                        "dst_ip": event.dst_ip,
+                        "dst_port": event.dst_port,
+                        "command": event.command,
+                        "argument": event.argument,
+                        **packet_evidence(event.packet_number),
+                    },
+                    tags=["ftp", "cleartext", "credentials"],
+                )
+            )
         else:
-            title = "File upload over FTP"
-            description = "An FTP upload command may indicate unencrypted data exfiltration."
-        findings.append(
-            Finding(
-                title=title,
-                description=description,
-                category="ftp_exfiltration",
-                timestamp=event.timestamp,
-                evidence={
-                    "src_ip": event.src_ip,
-                    "dst_ip": event.dst_ip,
-                    "dst_port": event.dst_port,
-                    "command": event.command,
-                    "argument": event.argument,
-                    **packet_evidence(event.packet_number),
-                },
-                tags=["ftp", "cleartext"],
+            # STOR/APPE is a confirmed upload direction, but "exfiltration"
+            # additionally implies the destination is external/untrusted --
+            # this analyzer has no destination-reputation context, so it
+            # reports the observed upload at medium confidence and leaves the
+            # exfiltration judgment for local-intel/MISP correlation or an
+            # analyst to confirm, rather than asserting it outright.
+            findings.append(
+                Finding(
+                    title="File upload over FTP",
+                    description=(
+                        "An FTP upload command was observed over an unencrypted control "
+                        "channel. Confirm the destination is external/untrusted before "
+                        "treating this as exfiltration."
+                    ),
+                    category="ftp_exfiltration",
+                    timestamp=event.timestamp,
+                    confidence="medium",
+                    evidence={
+                        "src_ip": event.src_ip,
+                        "dst_ip": event.dst_ip,
+                        "dst_port": event.dst_port,
+                        "command": event.command,
+                        "argument": event.argument,
+                        **packet_evidence(event.packet_number),
+                    },
+                    tags=["ftp", "cleartext", "upload"],
+                )
             )
-        )
     return findings

nettrace/analysis/http_analyzer.py

@@ -26,19 +26,62 @@ def is_executable_uri(uri: str) -> bool:
     return path.endswith((".exe", ".dll", ".ps1", ".bat", ".vbs", ".scr"))
 
 
+# GET/HEAD on an executable path plausibly means "downloaded"; POST/PUT/PATCH more
+# plausibly means "uploaded" (or is a report/telemetry beacon named *.exe by the
+# attacker to blend in). Bug #5: the old code called both "download" regardless.
+_DOWNLOAD_METHODS = {"GET", "HEAD"}
+_UPLOAD_METHODS = {"POST", "PUT", "PATCH"}
+
+
+def classify_http_executable(method: str) -> tuple[str, str, str]:
+    """Return (title, description, confidence) for an executable-path HTTP request."""
+    method_upper = (method or "").upper()
+    if method_upper in _DOWNLOAD_METHODS:
+        return (
+            "Possible executable/script download request",
+            "HTTP GET/HEAD requested an executable or script path.",
+            "medium",
+        )
+    if method_upper in _UPLOAD_METHODS:
+        return (
+            "Executable/script path observed in HTTP request (upload direction)",
+            "HTTP POST/PUT/PATCH targeted an executable/script path; this looks like "
+            "an upload or beacon, not a download, and should not be labeled as one.",
+            "low",
+        )
+    return (
+        "Executable/script path observed in HTTP request",
+        "HTTP request method does not confirm transfer direction for this "
+        "executable/script path.",
+        "low",
+    )
+
+
+def _tool_name(user_agent: str) -> str:
+    """Tool identity without the version, e.g. 'curl/7.68.0' -> 'curl'."""
+    return user_agent.split("/", 1)[0].strip()
+
+
 def analyze_http_events(events: list[HTTPEvent], config: dict) -> list[Finding]:
     findings: list[Finding] = []
     ua_path = config.get("intel", {}).get("suspicious_user_agents", "")
     suspicious_user_agents = _load_lines(ua_path)
+    suspicious_tool_names = {_tool_name(entry) for entry in suspicious_user_agents if entry}
     for event in events:
         user_agent = event.user_agent.lower()
-        if user_agent and user_agent in suspicious_user_agents:
+        if user_agent and _tool_name(user_agent) in suspicious_tool_names:
             findings.append(
                 Finding(
-                    title="Suspicious HTTP user agent",
-                    description="HTTP request used a user agent that appears in the local suspicious list.",
-                    category="http_c2",
+                    title="Command-line or automation HTTP client observed",
+                    description=(
+                        "HTTP request used a scripting/automation client (e.g. curl, wget, "
+                        "python-requests). This is common for legitimate developer, CI, and "
+                        "monitoring traffic -- treat as low-confidence context, not a standalone "
+                        "indicator, unless combined with other suspicious signals."
+                    ),
+                    category="http_automation_client",
                     timestamp=event.timestamp,
+                    confidence="low",
                     evidence={
                         "host": event.host,
                         "uri": event.uri,
@@ -51,12 +94,14 @@ def analyze_http_events(events: list[HTTPEvent], config: dict) -> list[Finding]:
                 )
             )
         if is_executable_uri(event.uri):
+            title, description, confidence = classify_http_executable(event.method)
             findings.append(
                 Finding(
-                    title="Executable download over HTTP",
-                    description="HTTP URI suggests retrieval of an executable or script payload.",
+                    title=title,
+                    description=description,
                     category="http_c2",
                     timestamp=event.timestamp,
+                    confidence=confidence,
                     evidence={"url": event_url(event), "method": event.method, **packet_evidence(event.packet_number)},
                     tags=["http", "payload"],
                 )