diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 23a14a1..c06534c 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/nettrace-bugfixes.patch b/nettrace-bugfixes.patch new file mode 100644 index 0000000..52e5eee --- /dev/null +++ b/nettrace-bugfixes.patch @@ -0,0 +1,1512 @@ +diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml +index 23a14a1..c06534c 100644 +--- a/.github/workflows/ci.yml ++++ b/.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 +diff --git a/nettrace/analysis/beaconing.py b/nettrace/analysis/beaconing.py +index 519156b..b693dc3 100644 +--- a/nettrace/analysis/beaconing.py ++++ b/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"], + ) +diff --git a/nettrace/analysis/dga_scorer.py b/nettrace/analysis/dga_scorer.py +index 7b7e178..de945d0 100644 +--- a/nettrace/analysis/dga_scorer.py ++++ b/nettrace/analysis/dga_scorer.py +@@ -37,21 +37,27 @@ COMMON_BIGRAMS = { + @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..biz, cdn..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,8 +146,9 @@ 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( +@@ -105,8 +156,10 @@ def score_domains(dns_events: list[DNSEvent], thresholds: dict) -> list[Finding] + 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), +diff --git a/nettrace/analysis/ftp_analyzer.py b/nettrace/analysis/ftp_analyzer.py +index 2907722..7aa63d1 100644 +--- a/nettrace/analysis/ftp_analyzer.py ++++ b/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 +diff --git a/nettrace/analysis/http_analyzer.py b/nettrace/analysis/http_analyzer.py +index 2059099..88ce7eb 100644 +--- a/nettrace/analysis/http_analyzer.py ++++ b/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"], + ) +diff --git a/nettrace/analysis/ioc_extractor.py b/nettrace/analysis/ioc_extractor.py +index c0fadcc..ba10bc3 100644 +--- a/nettrace/analysis/ioc_extractor.py ++++ b/nettrace/analysis/ioc_extractor.py +@@ -24,6 +24,27 @@ SOURCE_PRIORITY = { + "dns_answer_domain": 6, + } + ++# Sources backed by a parsed protocol artifact (a DNS answer, an HTTP host header, ++# a TLS SNI, a request URL). Everything else -- principally raw flow endpoint IPs ++# with a "flow::" source -- is an *observed* network artifact, not a ++# confirmed indicator, and should not be counted or ranked the same way. See bug #11. ++CONFIRMED_SOURCES = { ++ "dns", ++ "dns_answer", ++ "dns_answer_domain", ++ "http_host", ++ "http_url_host", ++ "http_connect_target", ++ "http_request", ++ "http_flow", ++ "tls_sni", ++ "tls_flow", ++} ++ ++ ++def _confidence_for_source(source: str) -> str: ++ return "confirmed" if source in CONFIRMED_SOURCES else "observed" ++ + + def _add(iocs: set[IOC], kind: str, value: str, source: str, packet_number: int = 0) -> None: + if not value: +@@ -35,7 +56,15 @@ def _add(iocs: set[IOC], kind: str, value: str, source: str, packet_number: int + return + else: + normalized = value.lower() if kind == "domain" else value +- iocs.add(IOC(kind=kind, value=normalized, source=source, packet_number=packet_number)) ++ iocs.add( ++ IOC( ++ kind=kind, ++ value=normalized, ++ source=source, ++ packet_number=packet_number, ++ confidence=_confidence_for_source(source), ++ ) ++ ) + + + def _is_public_ioc_ip(ip: str) -> bool: +diff --git a/nettrace/analysis/port_analyzer.py b/nettrace/analysis/port_analyzer.py +index b299959..d13d730 100644 +--- a/nettrace/analysis/port_analyzer.py ++++ b/nettrace/analysis/port_analyzer.py +@@ -34,6 +34,7 @@ def analyze_flows(flows: list[Flow], thresholds: dict) -> list[Finding]: + description="Traffic used a port commonly seen in malware labs, backdoors, or tunneling.", + category="unusual_port", + timestamp=flow.first_seen, ++ confidence="low", + evidence={ + "src_ip": flow.src_ip, + "dst_ip": flow.dst_ip, +@@ -51,6 +52,7 @@ def analyze_flows(flows: list[Flow], thresholds: dict) -> list[Finding]: + description="Flow generated a high volume of packets and may require exfiltration or C2 review.", + category="high_frequency_connections", + timestamp=flow.first_seen, ++ confidence="low", + evidence={ + "src_ip": flow.src_ip, + "dst_ip": flow.dst_ip, +diff --git a/nettrace/cli.py b/nettrace/cli.py +index acf0e62..8f30fbe 100644 +--- a/nettrace/cli.py ++++ b/nettrace/cli.py +@@ -23,8 +23,8 @@ def build_parser() -> argparse.ArgumentParser: + parser.add_argument( + "-c", + "--config", +- default="config.yaml", +- help="Path to config.yaml", ++ default=None, ++ help="Path to config.yaml (defaults to ./config.yaml if present; a value given here must exist)", + ) + parser.add_argument( + "-o", +@@ -53,7 +53,8 @@ def main() -> None: + output_dir = Path(args.output) + + try: +- config = load_config(Path(args.config)) ++ config_path = Path(args.config) if args.config is not None else Path("config.yaml") ++ config = load_config(config_path, explicit=args.config is not None) + except (ConfigError, OSError) as exc: + parser.error(f"configuration error: {exc}") + try: +diff --git a/nettrace/config.py b/nettrace/config.py +index bc8b422..9fedbf4 100644 +--- a/nettrace/config.py ++++ b/nettrace/config.py +@@ -45,6 +45,7 @@ DEFAULT_CONFIG: dict[str, Any] = { + "max_flows": 50_000, + "max_timeline_entries": 100_000, + "max_flow_samples": 256, ++ "max_findings": 20_000, + "max_tcp_streams": 10_000, + "max_tcp_stream_buffer_bytes": 1_048_576, + "max_tcp_pending_segments": 256, +@@ -94,6 +95,29 @@ def _positive_integer(value: Any, name: str, minimum: int = 1) -> None: + raise ConfigError(f"{name} must be an integer of at least {minimum}.") + + ++def _reject_unknown_keys(loaded: dict[str, Any] | None, name: str) -> None: ++ """Bug #13: a misspelled key like 'high_frequncy_connections' previously ++ merged in silently while the real threshold stayed at its default, with ++ no warning. Unknown keys at these two levels are now a hard config error.""" ++ if not loaded: ++ return ++ for section in ("thresholds", "misp", "limits", "protocols", "intel"): ++ section_value = loaded.get(section) ++ if not isinstance(section_value, dict): ++ continue ++ known = set(INTEL_KEYS) if section == "intel" else set(DEFAULT_CONFIG.get(section, {}).keys()) ++ unknown = set(section_value.keys()) - known ++ if unknown: ++ raise ConfigError( ++ f"Unknown key(s) in {name} {section}: {sorted(unknown)}. " ++ f"Known keys: {sorted(known)}." ++ ) ++ known_top = set(DEFAULT_CONFIG.keys()) | {"intel"} ++ unknown_top = set(loaded.keys()) - known_top ++ if unknown_top: ++ raise ConfigError(f"Unknown top-level key(s) in {name}: {sorted(unknown_top)}.") ++ ++ + def validate_config(config: dict[str, Any]) -> None: + thresholds = _mapping(config.get("thresholds"), "thresholds") + _positive_integer(thresholds.get("beacon_min_events"), "thresholds.beacon_min_events", 2) +@@ -152,7 +176,14 @@ def resolve_intel_paths(config: dict[str, Any], base_dir: Path, overridden_keys: + return resolved + + +-def load_config(path: Path) -> dict[str, Any]: ++def load_config(path: Path, explicit: bool = False) -> dict[str, Any]: ++ """Load config. ++ ++ ``explicit`` should be True when the path came from a user-provided ``-c`` ++ flag rather than the packaged default filename. Bug #12: previously a ++ missing file silently fell back to defaults in both cases, so a typo'd ++ ``-c confg.yaml`` looked like it worked while actually using defaults. ++ """ + config = DEFAULT_CONFIG + if DEFAULT_THRESHOLDS_PATH.exists(): + with DEFAULT_THRESHOLDS_PATH.open("r", encoding="utf-8") as handle: +@@ -165,6 +196,8 @@ def load_config(path: Path) -> dict[str, Any]: + _mapping(thresholds, "packaged thresholds.thresholds") + config = deep_merge(config, {"thresholds": thresholds}) + if not path.exists(): ++ if explicit: ++ raise ConfigError(f"Config file not found: {path}") + validate_config(config) + return resolve_intel_paths(config, PROJECT_ROOT) + with path.open("r", encoding="utf-8") as handle: +@@ -175,6 +208,7 @@ def load_config(path: Path) -> dict[str, Any]: + loaded = _mapping(loaded, "config root") + if "intel" in loaded: + _mapping(loaded["intel"], "intel") ++ _reject_unknown_keys(loaded, str(path)) + overridden_keys = set((loaded.get("intel") or {}).keys()) + merged = deep_merge(config, loaded) + validate_config(merged) +diff --git a/nettrace/engine.py b/nettrace/engine.py +index 34e91ee..0ff163b 100644 +--- a/nettrace/engine.py ++++ b/nettrace/engine.py +@@ -137,6 +137,31 @@ def analyze_pcap(pcap_path: Path, config: dict[str, Any]) -> AnalysisReport: + if discarded_tcp_streams: + warnings.add(f"TCP reassembly discarded {discarded_tcp_streams} incomplete or resource-limited streams.") + ++ conflicting_overlaps = sum( ++ extractor.streams.conflicting_overlaps ++ for extractor in (dns_stream_extractor, http_extractor, tls_extractor, ftp_extractor) ++ ) ++ if conflicting_overlaps: ++ warnings.add( ++ f"TCP reassembly observed {conflicting_overlaps} overlapping retransmission(s) whose bytes " ++ "did not match already-buffered data; NetTrace kept the first-seen bytes. This can indicate " ++ "reassembly evasion and should be reviewed manually." ++ ) ++ ++ incomplete_streams = sum( ++ extractor.streams.incomplete_streams ++ for extractor in (dns_stream_extractor, http_extractor, tls_extractor, ftp_extractor) ++ ) ++ incomplete_bytes = sum( ++ extractor.streams.incomplete_buffered_bytes ++ for extractor in (dns_stream_extractor, http_extractor, tls_extractor, ftp_extractor) ++ ) ++ if incomplete_streams: ++ warnings.add( ++ f"TCP reassembly ended with {incomplete_streams} incomplete stream(s) containing " ++ f"{incomplete_bytes} buffered bytes; the capture may have been truncated mid-message." ++ ) ++ + flows = list(flow_state.values()) + + findings = [] +@@ -160,6 +185,17 @@ def analyze_pcap(pcap_path: Path, config: dict[str, Any]) -> AnalysisReport: + + tag_findings(findings) + score_findings(findings) ++ ++ max_findings = config["limits"]["max_findings"] ++ if len(findings) > max_findings: ++ # Bug #10: there was previously no cap at all -- a hostile or extremely ++ # noisy capture could produce hundreds of thousands of findings and an ++ # unusable report. Keep the highest-severity findings rather than an ++ # arbitrary prefix. ++ findings.sort(key=lambda finding: finding.score, reverse=True) ++ findings = findings[:max_findings] ++ warnings.add(f"Findings truncated at {max_findings} entries (kept highest-severity).") ++ + timeline_item_count = ( + len(dns_events) + + len(http_events) +diff --git a/nettrace/mapping/attck_tagger.py b/nettrace/mapping/attck_tagger.py +index f41193a..407acdb 100644 +--- a/nettrace/mapping/attck_tagger.py ++++ b/nettrace/mapping/attck_tagger.py +@@ -65,11 +65,33 @@ def _refine_threat_intel_technique(finding: Finding) -> tuple[str, str] | None: + return None + + ++_HTTP_PORTS = {80, 8080, 8000, 8888} ++_TLS_PORTS = {443, 4443, 8443, 9443} ++ ++ ++def _refine_network_beaconing(finding: Finding) -> tuple[str, str] | None: ++ """Only tag a technique when the destination port confirms the protocol. ++ ++ Bug #1: the previous code mapped every non-DNS beacon straight to ++ T1071.001 (Web Protocols) even when the destination was, say, port 4444. ++ A beacon on an unconfirmed port is reported without an ATT&CK id rather ++ than guessing. ++ """ ++ port = finding.evidence.get("dst_port") ++ if port in _HTTP_PORTS: ++ return "T1071.001", "Application Layer Protocol: Web Protocols" ++ if port in _TLS_PORTS: ++ return "T1573", "Encrypted Channel" ++ return None ++ ++ + def tag_findings(findings: list[Finding]) -> None: + for finding in findings: + attack = ATTACK_MAP.get(finding.category) + if finding.category == "threat_intel_match": + attack = _refine_threat_intel_technique(finding) or attack ++ elif finding.category == "network_beaconing": ++ attack = _refine_network_beaconing(finding) + if attack: + finding.attack_id, finding.attack_name = attack + if finding.attack_id not in finding.tags: +diff --git a/nettrace/mapping/severity.py b/nettrace/mapping/severity.py +index ee8e172..c95e769 100644 +--- a/nettrace/mapping/severity.py ++++ b/nettrace/mapping/severity.py +@@ -9,14 +9,26 @@ BASE_SCORES = { + "network_beaconing": 70, + "dga_domain": 70, + "http_c2": 65, ++ "http_automation_client": 20, + "tls_c2": 55, + "high_frequency_connections": 55, + "unusual_port": 50, + "long_tls_session": 45, +- "ftp_exfiltration": 75, ++ "ftp_exfiltration": 65, ++ "ftp_cleartext_credentials": 55, + "misp_error": 15, + } + ++# Bug #6: severity was previously a flat category lookup, so a 5-event borderline ++# beacon and a 200-event tight-cadence beacon against known-bad infrastructure ++# scored identically. Confidence (set per-finding by the analyzer that produced ++# it -- see beaconing.py, dga_scorer.py, http_analyzer.py) now shifts the score. ++CONFIDENCE_ADJUSTMENT = { ++ "high": 12, ++ "medium": 0, ++ "low": -18, ++} ++ + + def _label(score: int) -> str: + if score >= 85: +@@ -33,9 +45,10 @@ def _label(score: int) -> str: + def score_findings(findings: list[Finding]) -> None: + for finding in findings: + score = BASE_SCORES.get(finding.category, 20) ++ score += CONFIDENCE_ADJUSTMENT.get(finding.confidence, 0) + if "MISP_HIT" in finding.tags: + score += 10 + if "IOC_MATCH" in finding.tags: + score += 5 +- finding.score = min(100, score) ++ finding.score = max(0, min(100, score)) + finding.severity = _label(finding.score) +diff --git a/nettrace/models/events.py b/nettrace/models/events.py +index 83cb50f..ad78714 100644 +--- a/nettrace/models/events.py ++++ b/nettrace/models/events.py +@@ -3,7 +3,52 @@ from __future__ import annotations + from dataclasses import asdict, dataclass, field + import ipaddress + from typing import Any +-from urllib.parse import urlsplit ++from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit ++ ++# Bug #16: HTTP query strings were stored and exported verbatim into JSON/HTML/ ++# Markdown/PDF reports and the IOC list. FTP passwords were already redacted ++# (see ftp_extractor.py) but tokens/session IDs/API keys in URLs were not. ++SENSITIVE_QUERY_KEYS = { ++ "token", ++ "api_key", ++ "apikey", ++ "password", ++ "passwd", ++ "pass", ++ "session", ++ "sessionid", ++ "signature", ++ "auth", ++ "authorization", ++ "access_token", ++ "refresh_token", ++ "secret", ++ "client_secret", ++} ++ ++ ++def redact_sensitive_query_params(uri: str) -> str: ++ """Redact known-sensitive query parameter values in a URI, preserving ++ the path and parameter names (both are useful for analysis) but not the ++ secret values themselves.""" ++ if "?" not in uri: ++ return uri ++ parsed = urlsplit(uri) ++ if not parsed.query: ++ return uri ++ pairs = parse_qsl(parsed.query, keep_blank_values=True) ++ redacted_any = False ++ redacted_pairs = [] ++ for key, value in pairs: ++ if key.lower() in SENSITIVE_QUERY_KEYS: ++ redacted_pairs.append((key, "")) ++ redacted_any = True ++ else: ++ redacted_pairs.append((key, value)) ++ if not redacted_any: ++ return uri ++ new_query = urlencode(redacted_pairs) ++ return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, new_query, parsed.fragment)) + + + @dataclass +@@ -115,6 +160,11 @@ class IOC: + value: str + source: str + packet_number: int = 0 ++ # "confirmed": derived from a parsed protocol artifact (DNS answer, HTTP host, ++ # TLS SNI, request URL). "observed": a raw flow endpoint IP with no protocol ++ # confirmation -- these dominate volume and should not be treated as equal- ++ # weight IOCs. See ioc_extractor.CONFIRMED_SOURCES. ++ confidence: str = "observed" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) +diff --git a/nettrace/models/findings.py b/nettrace/models/findings.py +index f344c60..761691d 100644 +--- a/nettrace/models/findings.py ++++ b/nettrace/models/findings.py +@@ -15,6 +15,7 @@ class Finding: + attack_name: str | None = None + severity: str = "info" + score: int = 0 ++ confidence: str = "medium" + tags: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: +diff --git a/nettrace/models/report.py b/nettrace/models/report.py +index ddda782..d8b1a8a 100644 +--- a/nettrace/models/report.py ++++ b/nettrace/models/report.py +@@ -30,6 +30,8 @@ class AnalysisReport: + "ftp_events": len(self.ftp_events), + "flows": len(self.flows), + "iocs": len(self.iocs), ++ "iocs_confirmed": sum(1 for ioc in self.iocs if ioc.confidence == "confirmed"), ++ "iocs_observed": sum(1 for ioc in self.iocs if ioc.confidence == "observed"), + "findings": len(self.findings), + "critical": sum(1 for finding in self.findings if finding.severity == "critical"), + "high": sum(1 for finding in self.findings if finding.severity == "high"), +diff --git a/nettrace/parsers/http_extractor.py b/nettrace/parsers/http_extractor.py +index 6d84156..002e85e 100644 +--- a/nettrace/parsers/http_extractor.py ++++ b/nettrace/parsers/http_extractor.py +@@ -3,7 +3,7 @@ from __future__ import annotations + from scapy.layers.inet import TCP + from scapy.packet import Raw + +-from nettrace.models.events import HTTPEvent ++from nettrace.models.events import HTTPEvent, redact_sensitive_query_params + from nettrace.parsers.tcp_stream import TCPStreamBuffers, ip_endpoints + + HTTP_METHODS = {"GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH", "CONNECT", "TRACE"} +@@ -81,7 +81,7 @@ def extract_http_event( + dst_ip=endpoints[1], + method=method, + host=host, +- uri=uri, ++ uri=redact_sensitive_query_params(uri), + user_agent=user_agent, + packet_number=packet_number, + ) +@@ -138,7 +138,7 @@ class HTTPStreamExtractor: + dst_ip=state.dst_ip, + method=method, + host=host, +- uri=uri, ++ uri=redact_sensitive_query_params(uri), + user_agent=user_agent, + packet_number=state.first_packet_number, + ) +diff --git a/nettrace/parsers/tcp_stream.py b/nettrace/parsers/tcp_stream.py +index 7eb3664..88ada5b 100644 +--- a/nettrace/parsers/tcp_stream.py ++++ b/nettrace/parsers/tcp_stream.py +@@ -39,6 +39,11 @@ class TCPStreamState: + pending: dict[int, _TCPSegment] = field(default_factory=dict) + first_packet_number: int = 0 + first_timestamp: float = 0.0 ++ # True once consume() has run at least once. Before that, base_seq is just ++ # the earliest segment received so far (out-of-order arrivals below it are ++ # still legitimate and must be prefixed in). After that, base_seq is a ++ # genuine "already parsed and removed" low-water mark -- see bug #8. ++ has_consumed: bool = False + + + def ip_endpoints(packet) -> tuple[str, str] | None: +@@ -64,11 +69,27 @@ class TCPStreamBuffers: + self._streams: OrderedDict[tuple[str, str, int, int], TCPStreamState] = OrderedDict() + self._total_buffered_bytes = 0 + self.discarded_streams = 0 ++ self.conflicting_overlaps = 0 + + @property + def total_buffered_bytes(self) -> int: + return self._total_buffered_bytes + ++ @property ++ def incomplete_streams(self) -> int: ++ """Streams still open (never hit FIN/RST/eviction) when iteration ends. ++ ++ Bug #4: the engine only ever reported streams it actively discarded ++ for resource limits -- a stream that simply never got a chance to ++ finish (truncated capture, missing final segment) looked identical to ++ a clean report with nothing outstanding. ++ """ ++ return sum(1 for state in self._streams.values() if self._state_size(state) > 0) ++ ++ @property ++ def incomplete_buffered_bytes(self) -> int: ++ return sum(self._state_size(state) for state in self._streams.values()) ++ + @staticmethod + def _state_size(state: TCPStreamState) -> int: + return len(state.buffer) + sum(len(item.payload) for item in state.pending.values()) +@@ -83,17 +104,42 @@ class TCPStreamBuffers: + self.discarded_streams += 1 + + @staticmethod +- def _merge_pending(state: TCPStreamState) -> None: ++ def _merge_pending(state: TCPStreamState) -> bool: ++ """Merge ready pending segments into the buffer. ++ ++ Returns True if a byte-level conflict was found between overlapping ++ retransmitted data and data already accepted (bug #3): the old code ++ accepted/trimmed overlapping segments without ever comparing bytes, so ++ a retransmission with *different* content than what was already ++ buffered was silently and arbitrarily resolved. ++ """ + if state.base_seq is None or state.next_seq is None: +- return ++ return False ++ conflict = False + changed = True + while changed: + changed = False + for sequence, segment in sorted(state.pending.items()): + end = sequence + len(segment.payload) ++ # Bug #8: a segment entirely covering already-consumed bytes ++ # must be discarded outright, not left in `pending` forever -- ++ # but only once consume() has actually run. Before that, ++ # base_seq is merely the earliest segment received so far, and ++ # an out-of-order arrival ending exactly at base_seq is a ++ # legitimate missing prefix, not a stale retransmission. ++ if state.has_consumed and end <= state.base_seq: ++ del state.pending[sequence] ++ changed = True ++ break + if end < state.base_seq or sequence > state.next_seq: + continue + if end <= state.next_seq and sequence >= state.base_seq: ++ # Fully inside the already-buffered range. Bug #3: compare ++ # bytes before discarding instead of assuming they match. ++ offset = sequence - state.base_seq ++ existing = bytes(state.buffer[offset : offset + len(segment.payload)]) ++ if existing and existing != segment.payload: ++ conflict = True + del state.pending[sequence] + changed = True + break +@@ -105,11 +151,16 @@ class TCPStreamBuffers: + state.first_timestamp = segment.timestamp + if sequence <= state.next_seq and end > state.next_seq: + overlap = state.next_seq - sequence +- state.buffer.extend(segment.payload[overlap:]) ++ if overlap > 0 and overlap <= len(state.buffer): ++ existing_tail = bytes(state.buffer[-overlap:]) ++ if existing_tail != segment.payload[:overlap]: ++ conflict = True ++ state.buffer.extend(segment.payload[max(overlap, 0) :]) + state.next_seq = end + del state.pending[sequence] + changed = True + break ++ return conflict + + def feed(self, packet, packet_number: int) -> TCPStreamState | None: + endpoints = ip_endpoints(packet) +@@ -170,7 +221,8 @@ class TCPStreamBuffers: + existing = state.pending.get(sequence) + if existing is None or len(payload) > len(existing.payload): + state.pending[sequence] = segment +- self._merge_pending(state) ++ if self._merge_pending(state): ++ self.conflicting_overlaps += 1 + + buffered_bytes = self._state_size(state) + self._total_buffered_bytes += buffered_bytes - previous_size +@@ -188,6 +240,7 @@ class TCPStreamBuffers: + del state.buffer[:length] + if state.base_seq is not None: + state.base_seq += length ++ state.has_consumed = True + state.first_packet_number = next_packet_number + state.first_timestamp = next_timestamp + self._total_buffered_bytes += self._state_size(state) - previous_size +diff --git a/nettrace/report/markdown_report.py b/nettrace/report/markdown_report.py +index c1b5313..d57ed40 100644 +--- a/nettrace/report/markdown_report.py ++++ b/nettrace/report/markdown_report.py +@@ -172,16 +172,43 @@ def build_analyst_paragraph(report: dict[str, Any], victim: str, urls: list[str] + summary = report.get("summary", {}) + url_text = ", ".join(markdown_code(url) for url in urls[:5]) if urls else "no plaintext HTTP staging URLs" + ip_text = ", ".join(markdown_code(ip) for ip in ips[:8]) if ips else "no public C2 candidates" +- return ( +- f"NetTrace analyzed {markdown_code(report.get('pcap_path', 'unknown'))} and identified {markdown_code(victim)} as the primary " +- f"internal host. The capture contains {summary.get('dns_events', 0)} DNS events, " ++ findings = report.get("findings", []) ++ findings_count = summary.get("findings", len(findings)) ++ strong_findings = sum(1 for finding in findings if finding.get("severity") in {"high", "critical"}) ++ ++ host_clause = ( ++ f"identified {markdown_code(victim)} as the internal host with the highest observed external traffic " ++ "volume (a triage heuristic, not confirmation of compromise)" ++ if victim and victim != "unknown" ++ else "did not identify a single dominant internal host" ++ ) ++ base = ( ++ f"NetTrace analyzed {markdown_code(report.get('pcap_path', 'unknown'))} and {host_clause}. " ++ f"The capture contains {summary.get('dns_events', 0)} DNS events, " + f"{summary.get('http_events', 0)} plaintext HTTP requests, {summary.get('tls_events', 0)} TLS SNI events, " +- f"{summary.get('ftp_events', 0)} FTP commands, and {summary.get('flows', 0)} flows. The strongest analyst signal is the combination of HTTP staging " +- f"activity ({url_text}) and high-volume or encrypted traffic involving {ip_text}. These behaviors are " +- "consistent with malware staging and command-and-control triage, while heuristic findings should be " +- "validated with packet context." ++ f"{summary.get('ftp_events', 0)} FTP commands, and {summary.get('flows', 0)} flows." + ) + ++ # Bug #18: the old text asserted "consistent with malware staging and C2 ++ # triage" even on a capture with zero findings and zero staging URLs. The ++ # conclusion now scales with what was actually found. ++ if strong_findings: ++ conclusion = ( ++ f" {strong_findings} high/critical-severity finding(s) were produced, with the strongest signal being " ++ f"HTTP staging activity ({url_text}) and high-volume or encrypted traffic involving {ip_text}. This " ++ "warrants malware staging and command-and-control triage; heuristic findings should still be validated " ++ "with packet context before conclusions are drawn." ++ ) ++ elif findings_count: ++ conclusion = ( ++ f" NetTrace produced {findings_count} lower-confidence heuristic finding(s) ({url_text}; {ip_text}) that " ++ "should be reviewed with packet context. None reached high or critical severity, so this capture does " ++ "not, on its own, establish malware staging or command-and-control activity." ++ ) ++ else: ++ conclusion = " No behavioral findings were produced from this capture." ++ return base + conclusion ++ + + def build_markdown(report: dict[str, Any], source: str = "", source_url: str = "") -> str: + summary = report.get("summary", {}) +@@ -233,7 +260,7 @@ def build_markdown(report: dict[str, Any], source: str = "", source_url: str = " + lines.extend( + [ + "", +- "## ATT&CK Techniques", ++ "## Potential ATT&CK Technique Associations", + "", + bullet_list(techniques), + "", +diff --git a/nettrace/rules/attck_map.yaml b/nettrace/rules/attck_map.yaml +index 0870110..e594fcc 100644 +--- a/nettrace/rules/attck_map.yaml ++++ b/nettrace/rules/attck_map.yaml +@@ -1,9 +1,15 @@ ++# NOTE on "network_beaconing" and "high_frequency_connections": these are ++# intentionally absent from this map. Bug #1: mapping arbitrary periodic TCP/UDP ++# traffic straight to T1071.001 (Web Protocols) or raw packet volume straight to ++# T1020 (Automated Exfiltration) overstates what NetTrace actually observed. ++# network_beaconing gets a technique only when attck_tagger.py's ++# _refine_network_beaconing() can confirm the destination port is actually ++# HTTP/TLS; otherwise it is reported as an unmapped behavioral finding. ++# high_frequency_connections never gets an ATT&CK id by itself -- packet count ++# alone does not establish exfiltration. + dns_beaconing: + id: T1071.004 + name: "Application Layer Protocol: DNS" +-network_beaconing: +- id: T1071.001 +- name: "Application Layer Protocol: Web Protocols" + http_c2: + id: T1071.001 + name: "Application Layer Protocol: Web Protocols" +@@ -14,11 +20,8 @@ dga_domain: + id: T1568.002 + name: "Dynamic Resolution: Domain Generation Algorithms" + long_tls_session: +- id: T1090 +- name: "Proxy" +-high_frequency_connections: +- id: T1020 +- name: "Automated Exfiltration" ++ id: T1573 ++ name: "Encrypted Channel" + unusual_port: + id: T1571 + name: "Non-Standard Port" +diff --git a/nettrace/rules/dga_allowlist.yaml b/nettrace/rules/dga_allowlist.yaml +index 1bc0c64..ab38ba5 100644 +--- a/nettrace/rules/dga_allowlist.yaml ++++ b/nettrace/rules/dga_allowlist.yaml +@@ -1,18 +1,26 @@ ++# Domain-boundary matching only. A domain matches a suffix if it equals it, or ++# ends with "." + suffix -- never via raw substring. See bug #2: ++# "microsoft-login-x8f39q.biz" must NOT match "microsoft". ++domains: ++ - "microsoft.com" ++ - "windowsupdate.com" ++ - "msedge.net" ++ - "azure.com" ++ - "azureedge.net" ++ - "azurefd.us" ++ - "cloudapp.azure.com" ++ - "trafficmanager.net" + suffixes: +- - ".local" +- - ".localdomain" +- - ".microsoft.com" +- - ".windowsupdate.com" +- - ".msedge.net" +- - ".azureedge.net" +- - ".azurefd.us" +- - ".cloudapp.azure.com" +- - ".trafficmanager.net" +-contains: +- - "microsoft" +- - "windowsupdate" +- - "msedge" +- - "azure" ++ - "local" ++ - "localdomain" ++ - "microsoft.com" ++ - "windowsupdate.com" ++ - "msedge.net" ++ - "azure.com" ++ - "azureedge.net" ++ - "azurefd.us" ++ - "cloudapp.azure.com" ++ - "trafficmanager.net" + regexes: + - "^desktop-[a-z0-9-]+(\\..*)?$" + - "^[a-z0-9-]+-dc(\\..*)?$" +diff --git a/pyproject.toml b/pyproject.toml +index a915311..42c04a5 100644 +--- a/pyproject.toml ++++ b/pyproject.toml +@@ -7,7 +7,7 @@ name = "nettrace-malware-analysis" + version = "1.0.0" + description = "Offline malware traffic analysis platform for PCAP triage, IOC extraction, ATT&CK mapping, and analyst reports." + readme = "README.md" +-requires-python = ">=3.11" ++requires-python = ">=3.11,<3.13" + license = { file = "LICENSE" } + authors = [ + { name = "Aryan" } +@@ -36,12 +36,14 @@ dependencies = [ + "scapy~=2.6", + "jinja2~=3.1", + "pyyaml~=6.0", +- "requests~=2.32", +- "pymisp~=2.4", + "reportlab~=4.2" + ] + + [project.optional-dependencies] ++misp = [ ++ "pymisp~=2.4", ++ "requests~=2.32" ++] + dev = [ + "pypdf~=6.0", + "pytest~=8.4", +diff --git a/requirements-dev.txt b/requirements-dev.txt +new file mode 100644 +index 0000000..dd00615 +--- /dev/null ++++ b/requirements-dev.txt +@@ -0,0 +1,4 @@ ++-r requirements.txt ++pypdf~=6.0 ++pytest~=8.4 ++pytest-cov~=7.1 +diff --git a/requirements-misp.txt b/requirements-misp.txt +new file mode 100644 +index 0000000..3048e42 +--- /dev/null ++++ b/requirements-misp.txt +@@ -0,0 +1,3 @@ ++-r requirements.txt ++pymisp~=2.4 ++requests~=2.32 +diff --git a/requirements.txt b/requirements.txt +index 87eb0ab..9b1d4d7 100644 +--- a/requirements.txt ++++ b/requirements.txt +@@ -1,9 +1,4 @@ + scapy~=2.6 + jinja2~=3.1 + pyyaml~=6.0 +-requests~=2.32 +-pymisp~=2.4 + reportlab~=4.2 +-pypdf~=6.0 +-pytest~=8.4 +-pytest-cov~=7.1 +diff --git a/tests/test_cli.py b/tests/test_cli.py +index 37e094a..4907619 100644 +--- a/tests/test_cli.py ++++ b/tests/test_cli.py +@@ -63,7 +63,7 @@ def test_cli_prints_analysis_warnings(tmp_path, monkeypatch, capsys): + timeline=[], + warnings=["HTTP events truncated at 10 entries."], + ) +- monkeypatch.setattr("nettrace.cli.load_config", lambda _path: {}) ++ monkeypatch.setattr("nettrace.cli.load_config", lambda _path, explicit=False: {}) + monkeypatch.setattr("nettrace.cli.analyze_pcap", lambda _path, _config: report) + monkeypatch.setattr( + "sys.argv", +@@ -99,7 +99,7 @@ def test_cli_handles_report_output_errors_without_traceback(tmp_path, monkeypatc + findings=[], + timeline=[], + ) +- monkeypatch.setattr("nettrace.cli.load_config", lambda _path: {}) ++ monkeypatch.setattr("nettrace.cli.load_config", lambda _path, explicit=False: {}) + monkeypatch.setattr("nettrace.cli.analyze_pcap", lambda _path, _config: report) + monkeypatch.setattr("sys.argv", ["nettrace", "sample.pcap", "--output", str(output_file)]) + +diff --git a/tests/test_config.py b/tests/test_config.py +index 6e1187f..fe51187 100644 +--- a/tests/test_config.py ++++ b/tests/test_config.py +@@ -53,3 +53,39 @@ def test_load_config_rejects_invalid_threshold_and_port_types(tmp_path): + bad_port.write_text("protocols:\n http_ports: [80, invalid]\n", encoding="utf-8") + with pytest.raises(ConfigError, match="http_ports"): + load_config(bad_port) ++ ++ ++def test_load_config_errors_on_missing_explicit_path(tmp_path): ++ # Bug #12: a user-specified -c path that doesn't exist must error, not ++ # silently fall back to defaults. ++ missing = tmp_path / "confg.yaml" # typo'd filename, on purpose ++ ++ with pytest.raises(ConfigError, match="Config file not found"): ++ load_config(missing, explicit=True) ++ ++ ++def test_load_config_still_defaults_when_implicit_path_missing(tmp_path): ++ # Unchanged behavior: the packaged default config.yaml is optional. ++ missing = tmp_path / "config.yaml" ++ ++ config = load_config(missing, explicit=False) ++ ++ assert config["thresholds"]["high_frequency_connections"] == 50 ++ ++ ++def test_load_config_rejects_unknown_threshold_key(tmp_path): ++ # Bug #13: a typo'd threshold key previously merged in silently while the ++ # real threshold stayed at its default with no warning. ++ config_path = tmp_path / "config.yaml" ++ config_path.write_text("thresholds:\n high_frequncy_connections: 999\n", encoding="utf-8") ++ ++ with pytest.raises(ConfigError, match="Unknown key"): ++ load_config(config_path) ++ ++ ++def test_load_config_rejects_unknown_top_level_key(tmp_path): ++ config_path = tmp_path / "config.yaml" ++ config_path.write_text("logging:\n level: debug\n", encoding="utf-8") ++ ++ with pytest.raises(ConfigError, match="Unknown top-level key"): ++ load_config(config_path) +diff --git a/tests/test_http_analyzer.py b/tests/test_http_analyzer.py +index 12797bc..27e5a3e 100644 +--- a/tests/test_http_analyzer.py ++++ b/tests/test_http_analyzer.py +@@ -1,5 +1,8 @@ ++from scapy.all import IP, Raw, TCP ++ + from nettrace.analysis.http_analyzer import analyze_http_events +-from nettrace.models.events import HTTPEvent ++from nettrace.models.events import HTTPEvent, redact_sensitive_query_params ++from nettrace.parsers.http_extractor import extract_http_event + + + def test_executable_download_detection_ignores_case_and_query_string(): +@@ -20,3 +23,74 @@ def test_blank_suspicious_user_agent_path_is_ignored(): + findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": ""}}) + + assert findings == [] ++ ++ ++def test_post_to_executable_path_is_not_labeled_a_download(): ++ # Bug #5: POST/PUT/PATCH to a *.exe path is not evidence of a download. ++ event = HTTPEvent(1.0, "10.0.0.5", "203.0.113.66", "POST", "example.com", "/upload/report.exe") ++ ++ findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": ""}}) ++ ++ assert len(findings) == 1 ++ assert "download" not in findings[0].title.lower() ++ assert findings[0].confidence == "low" ++ ++ ++def test_get_executable_path_is_labeled_possible_download(): ++ event = HTTPEvent(1.0, "10.0.0.5", "203.0.113.66", "GET", "example.com", "/payload.exe") ++ ++ findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": ""}}) ++ ++ assert findings[0].title == "Possible executable/script download request" ++ assert findings[0].confidence == "medium" ++ ++ ++def test_suspicious_user_agent_matches_by_tool_name_ignoring_version(tmp_path): ++ # Bug #17: an entry for "curl/7.68.0" should still catch "curl/8.4.0", ++ # and the finding is now low-confidence context, not "suspicious". ++ ua_file = tmp_path / "uas.txt" ++ ua_file.write_text("curl/7.68.0\n", encoding="utf-8") ++ event = HTTPEvent( ++ 1.0, "10.0.0.5", "203.0.113.66", "GET", "example.com", "/", ++ user_agent="curl/8.4.0", ++ ) ++ ++ findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": str(ua_file)}}) ++ ++ assert len(findings) == 1 ++ assert findings[0].category == "http_automation_client" ++ assert findings[0].confidence == "low" ++ ++ ++def test_redact_sensitive_query_params_removes_secrets(): ++ # Bug #16: HTTP query params were stored/exported verbatim, leaking ++ # tokens/API keys/session IDs into every report format. ++ assert redact_sensitive_query_params("/reset-password?token=secret123") == "/reset-password?token=%3Credacted%3E" ++ assert redact_sensitive_query_params("/api?api_key=123456") == "/api?api_key=%3Credacted%3E" ++ ++ ++def test_redact_sensitive_query_params_preserves_non_sensitive_params(): ++ result = redact_sensitive_query_params("/download?signature=xyz&legit_param=keepme") ++ assert "legit_param=keepme" in result ++ assert "signature=%3Credacted%3E" in result ++ ++ ++def test_redact_sensitive_query_params_leaves_benign_query_untouched(): ++ assert redact_sensitive_query_params("/search?q=normal+query") == "/search?q=normal+query" ++ ++ ++def test_redact_sensitive_query_params_no_query_string_untouched(): ++ assert redact_sensitive_query_params("/plain/path") == "/plain/path" ++ ++ ++def test_http_extractor_redacts_query_params_in_constructed_event(): ++ packet = IP(src="10.0.0.5", dst="203.0.113.1") / TCP(sport=50000, dport=80, seq=100, flags="PA") / Raw( ++ load=b"GET /login?session=abcdef123 HTTP/1.1\r\nHost: example.com\r\n\r\n" ++ ) ++ packet.time = 1.0 ++ ++ event = extract_http_event(packet, packet_number=1) ++ ++ assert event is not None ++ assert "abcdef123" not in event.uri ++ assert "session=%3Credacted%3E" in event.uri +diff --git a/tests/test_stream_reassembly.py b/tests/test_stream_reassembly.py +index 1265795..4757fe2 100644 +--- a/tests/test_stream_reassembly.py ++++ b/tests/test_stream_reassembly.py +@@ -175,7 +175,7 @@ def test_engine_uses_stream_reassembly(tmp_path): + report = analyze_pcap(capture, load_config(Path("does-not-exist.yaml"))) + + assert len(report.http_events) == 1 +- assert any(finding.title == "Executable download over HTTP" for finding in report.findings) ++ assert any(finding.title == "Possible executable/script download request" for finding in report.findings) + + + def test_http_stream_resets_when_tcp_tuple_is_reused(): +@@ -299,7 +299,7 @@ def test_engine_reassembles_ipv4_fragments(tmp_path): + + assert len(report.http_events) == 1 + assert report.http_events[0].host == "evil.example" +- assert any(finding.title == "Executable download over HTTP" for finding in report.findings) ++ assert any(finding.title == "Possible executable/script download request" for finding in report.findings) + + + def test_engine_reports_resource_truncation(tmp_path): +@@ -331,4 +331,63 @@ def test_engine_uses_configured_http_ports(tmp_path): + report = analyze_pcap(capture, config) + + assert len(report.http_events) == 1 +- assert any(finding.title == "Executable download over HTTP" for finding in report.findings) ++ assert any(finding.title == "Possible executable/script download request" for finding in report.findings) ++ ++ ++def test_engine_caps_findings_at_configured_max(tmp_path): ++ # Bug #10: there was previously no cap -- a hostile/noisy capture could ++ # produce an unbounded number of findings. ++ packets = [] ++ for index in range(5): ++ packet = ( ++ IP(src="10.0.0.5", dst=f"203.0.113.{index}") ++ / TCP(sport=40000 + index, dport=4444, seq=100, flags="PA") ++ / Raw(load=b"x" * 10) ++ ) ++ packet.time = float(index) ++ packets.append(packet) ++ capture = tmp_path / "many-findings.pcap" ++ wrpcap(str(capture), packets) ++ config = load_config(Path("does-not-exist.yaml")) ++ config["limits"]["max_findings"] = 2 ++ ++ report = analyze_pcap(capture, config) ++ ++ assert len(report.findings) == 2 ++ assert any("Findings truncated at 2 entries" in warning for warning in report.warnings) ++ ++ ++def test_engine_surfaces_conflicting_overlap_warning(tmp_path): ++ # Bug #3: conflicting overlapping retransmissions must be visible to the ++ # analyst, not silently resolved. Uses an incomplete request (no blank ++ # line) so the bytes stay unconsumed in the buffer -- a complete request ++ # would be consumed immediately and hit the bug #8 path instead. ++ packets = [ ++ IP(src="10.0.0.5", dst="45.33.32.156") / TCP(sport=50000, dport=80, seq=100, flags="PA") ++ / Raw(load=b"GET /aaaaaaaaaaaa"), ++ IP(src="10.0.0.5", dst="45.33.32.156") / TCP(sport=50000, dport=80, seq=100, flags="PA") ++ / Raw(load=b"GET /bbbbbbbbbbbb"), ++ ] ++ for number, packet in enumerate(packets, 1): ++ packet.time = float(number) ++ capture = tmp_path / "overlap-conflict.pcap" ++ wrpcap(str(capture), packets) ++ ++ report = analyze_pcap(capture, load_config(Path("does-not-exist.yaml"))) ++ ++ assert any("overlapping retransmission" in warning for warning in report.warnings) ++ ++ ++def test_engine_surfaces_incomplete_stream_warning(tmp_path): ++ # Bug #4: a stream still open (no FIN/RST, never hit a resource limit) ++ # when the capture ends was previously invisible in the report. ++ packet = IP(src="10.0.0.5", dst="45.33.32.156") / TCP(sport=50000, dport=80, seq=100, flags="PA") / Raw( ++ load=b"GET /never-finishes" ++ ) ++ packet.time = 1.0 ++ capture = tmp_path / "truncated.pcap" ++ wrpcap(str(capture), [packet]) ++ ++ report = analyze_pcap(capture, load_config(Path("does-not-exist.yaml"))) ++ ++ assert any("incomplete stream" in warning for warning in report.warnings) +diff --git a/tests/test_tcp_stream.py b/tests/test_tcp_stream.py +new file mode 100644 +index 0000000..dd7899e +--- /dev/null ++++ b/tests/test_tcp_stream.py +@@ -0,0 +1,92 @@ ++from scapy.all import IP, Raw, TCP ++ ++from nettrace.parsers.tcp_stream import TCPStreamBuffers ++ ++ ++def _segment(seq: int, payload: bytes, packet_number: int, timestamp: float = 1.0): ++ packet = IP(src="10.0.0.5", dst="203.0.113.10") / TCP(sport=50000, dport=80, seq=seq, flags="A") / Raw(load=payload) ++ packet.time = timestamp ++ return packet ++ ++ ++def test_stale_retransmission_below_base_seq_is_discarded_not_leaked(): ++ # Bug #8: a segment fully covering bytes already consumed by a protocol ++ # parser must be discarded outright, not left in `pending` forever eating ++ # the pending-segment budget. ++ buffers = TCPStreamBuffers() ++ state = buffers.feed(_segment(100, b"GET / HTTP/1.1\r\n", 1), 1) ++ assert state is not None ++ buffers.consume(state, len(b"GET / HTTP/1.1\r\n"), next_packet_number=2, next_timestamp=1.1) ++ assert state.has_consumed is True ++ assert state.base_seq == 116 ++ ++ # A retransmission of the already-consumed first segment arrives late. ++ state = buffers.feed(_segment(100, b"GET / HTTP/1.1\r\n", 3, timestamp=1.2), 3) ++ assert state is not None ++ assert state.pending == {}, "stale retransmission must not sit in pending forever" ++ assert bytes(state.buffer) == b"" ++ ++ ++def test_retransmission_ending_exactly_at_base_seq_is_not_reprepended(): ++ # Bug #8 (exact case from the review): a segment ending exactly at the new ++ # base_seq must not be prepended back into the live buffer -- that would ++ # resurrect already-consumed, already-reported data. ++ buffers = TCPStreamBuffers() ++ state = buffers.feed(_segment(100, b"AAAA", 1), 1) ++ buffers.consume(state, 4, next_packet_number=2, next_timestamp=1.1) ++ assert state.base_seq == 104 ++ ++ state = buffers.feed(_segment(100, b"AAAA", 3, timestamp=1.2), 3) # end == 104 == base_seq ++ assert bytes(state.buffer) == b"", "must not resurrect consumed bytes" ++ ++ ++def test_out_of_order_prefix_before_any_consumption_still_merges(): ++ # Sanity check: the has_consumed guard must not break the normal, ++ # legitimate out-of-order arrival case (no consume() has happened yet). ++ buffers = TCPStreamBuffers() ++ state = buffers.feed(_segment(102, b"ER analyst\r\n", 2, timestamp=2.0), 2) ++ assert bytes(state.buffer) == b"ER analyst\r\n" ++ ++ state = buffers.feed(_segment(100, b"US", 1, timestamp=1.0), 1) ++ assert bytes(state.buffer) == b"USER analyst\r\n" ++ ++ ++def test_conflicting_overlap_is_detected_not_silently_resolved(): ++ # Bug #3: overlapping retransmitted bytes that *differ* from what's ++ # already buffered were never compared -- silently accepted or dropped ++ # with no signal to the analyst. ++ buffers = TCPStreamBuffers() ++ buffers.feed(_segment(100, b"AAAABBBB", 1), 1) ++ assert buffers.conflicting_overlaps == 0 ++ ++ # Retransmission of the same range with DIFFERENT bytes -- a real conflict. ++ buffers.feed(_segment(100, b"AAAAXXXX", 2, timestamp=1.1), 2) ++ assert buffers.conflicting_overlaps == 1 ++ ++ ++def test_identical_overlap_is_not_flagged_as_conflict(): ++ buffers = TCPStreamBuffers() ++ buffers.feed(_segment(100, b"AAAABBBB", 1), 1) ++ buffers.feed(_segment(100, b"AAAABBBB", 2, timestamp=1.1), 2) ++ ++ assert buffers.conflicting_overlaps == 0 ++ ++ ++def test_incomplete_stream_visible_at_eof(): ++ # Bug #4: a stream that never receives its closing FIN/RST and never hits ++ # a resource limit was previously invisible in the report -- it wasn't ++ # "discarded", it just silently vanished from consideration. ++ buffers = TCPStreamBuffers() ++ buffers.feed(_segment(100, b"incomplete HTTP request, no blank line yet", 1), 1) ++ ++ assert buffers.incomplete_streams == 1 ++ assert buffers.incomplete_buffered_bytes == len(b"incomplete HTTP request, no blank line yet") ++ ++ ++def test_fully_consumed_stream_is_not_counted_as_incomplete(): ++ buffers = TCPStreamBuffers() ++ state = buffers.feed(_segment(100, b"GET / HTTP/1.1\r\n\r\n", 1), 1) ++ buffers.consume(state, len(b"GET / HTTP/1.1\r\n\r\n"), next_packet_number=2, next_timestamp=1.1) ++ ++ assert buffers.incomplete_streams == 0 ++ assert buffers.incomplete_buffered_bytes == 0 diff --git a/nettrace/analysis/beaconing.py b/nettrace/analysis/beaconing.py index 519156b..b693dc3 100644 --- a/nettrace/analysis/beaconing.py +++ b/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"], ) diff --git a/nettrace/analysis/dga_scorer.py b/nettrace/analysis/dga_scorer.py index 7b7e178..de945d0 100644 --- a/nettrace/analysis/dga_scorer.py +++ b/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..biz, cdn..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,8 +146,9 @@ 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( @@ -105,8 +156,10 @@ def score_domains(dns_events: list[DNSEvent], thresholds: dict) -> list[Finding] 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), diff --git a/nettrace/analysis/ftp_analyzer.py b/nettrace/analysis/ftp_analyzer.py index 2907722..7aa63d1 100644 --- a/nettrace/analysis/ftp_analyzer.py +++ b/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 diff --git a/nettrace/analysis/http_analyzer.py b/nettrace/analysis/http_analyzer.py index 2059099..88ce7eb 100644 --- a/nettrace/analysis/http_analyzer.py +++ b/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"], ) diff --git a/nettrace/analysis/ioc_extractor.py b/nettrace/analysis/ioc_extractor.py index c0fadcc..ba10bc3 100644 --- a/nettrace/analysis/ioc_extractor.py +++ b/nettrace/analysis/ioc_extractor.py @@ -24,6 +24,27 @@ "dns_answer_domain": 6, } +# Sources backed by a parsed protocol artifact (a DNS answer, an HTTP host header, +# a TLS SNI, a request URL). Everything else -- principally raw flow endpoint IPs +# with a "flow::" source -- is an *observed* network artifact, not a +# confirmed indicator, and should not be counted or ranked the same way. See bug #11. +CONFIRMED_SOURCES = { + "dns", + "dns_answer", + "dns_answer_domain", + "http_host", + "http_url_host", + "http_connect_target", + "http_request", + "http_flow", + "tls_sni", + "tls_flow", +} + + +def _confidence_for_source(source: str) -> str: + return "confirmed" if source in CONFIRMED_SOURCES else "observed" + def _add(iocs: set[IOC], kind: str, value: str, source: str, packet_number: int = 0) -> None: if not value: @@ -35,7 +56,15 @@ def _add(iocs: set[IOC], kind: str, value: str, source: str, packet_number: int return else: normalized = value.lower() if kind == "domain" else value - iocs.add(IOC(kind=kind, value=normalized, source=source, packet_number=packet_number)) + iocs.add( + IOC( + kind=kind, + value=normalized, + source=source, + packet_number=packet_number, + confidence=_confidence_for_source(source), + ) + ) def _is_public_ioc_ip(ip: str) -> bool: diff --git a/nettrace/analysis/port_analyzer.py b/nettrace/analysis/port_analyzer.py index b299959..d13d730 100644 --- a/nettrace/analysis/port_analyzer.py +++ b/nettrace/analysis/port_analyzer.py @@ -34,6 +34,7 @@ def analyze_flows(flows: list[Flow], thresholds: dict) -> list[Finding]: description="Traffic used a port commonly seen in malware labs, backdoors, or tunneling.", category="unusual_port", timestamp=flow.first_seen, + confidence="low", evidence={ "src_ip": flow.src_ip, "dst_ip": flow.dst_ip, @@ -51,6 +52,7 @@ def analyze_flows(flows: list[Flow], thresholds: dict) -> list[Finding]: description="Flow generated a high volume of packets and may require exfiltration or C2 review.", category="high_frequency_connections", timestamp=flow.first_seen, + confidence="low", evidence={ "src_ip": flow.src_ip, "dst_ip": flow.dst_ip, diff --git a/nettrace/cli.py b/nettrace/cli.py index acf0e62..8f30fbe 100644 --- a/nettrace/cli.py +++ b/nettrace/cli.py @@ -23,8 +23,8 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "-c", "--config", - default="config.yaml", - help="Path to config.yaml", + default=None, + help="Path to config.yaml (defaults to ./config.yaml if present; a value given here must exist)", ) parser.add_argument( "-o", @@ -53,7 +53,8 @@ def main() -> None: output_dir = Path(args.output) try: - config = load_config(Path(args.config)) + config_path = Path(args.config) if args.config is not None else Path("config.yaml") + config = load_config(config_path, explicit=args.config is not None) except (ConfigError, OSError) as exc: parser.error(f"configuration error: {exc}") try: diff --git a/nettrace/config.py b/nettrace/config.py index bc8b422..9fedbf4 100644 --- a/nettrace/config.py +++ b/nettrace/config.py @@ -45,6 +45,7 @@ "max_flows": 50_000, "max_timeline_entries": 100_000, "max_flow_samples": 256, + "max_findings": 20_000, "max_tcp_streams": 10_000, "max_tcp_stream_buffer_bytes": 1_048_576, "max_tcp_pending_segments": 256, @@ -94,6 +95,29 @@ def _positive_integer(value: Any, name: str, minimum: int = 1) -> None: raise ConfigError(f"{name} must be an integer of at least {minimum}.") +def _reject_unknown_keys(loaded: dict[str, Any] | None, name: str) -> None: + """Bug #13: a misspelled key like 'high_frequncy_connections' previously + merged in silently while the real threshold stayed at its default, with + no warning. Unknown keys at these two levels are now a hard config error.""" + if not loaded: + return + for section in ("thresholds", "misp", "limits", "protocols", "intel"): + section_value = loaded.get(section) + if not isinstance(section_value, dict): + continue + known = set(INTEL_KEYS) if section == "intel" else set(DEFAULT_CONFIG.get(section, {}).keys()) + unknown = set(section_value.keys()) - known + if unknown: + raise ConfigError( + f"Unknown key(s) in {name} {section}: {sorted(unknown)}. " + f"Known keys: {sorted(known)}." + ) + known_top = set(DEFAULT_CONFIG.keys()) | {"intel"} + unknown_top = set(loaded.keys()) - known_top + if unknown_top: + raise ConfigError(f"Unknown top-level key(s) in {name}: {sorted(unknown_top)}.") + + def validate_config(config: dict[str, Any]) -> None: thresholds = _mapping(config.get("thresholds"), "thresholds") _positive_integer(thresholds.get("beacon_min_events"), "thresholds.beacon_min_events", 2) @@ -152,7 +176,14 @@ def resolve_intel_paths(config: dict[str, Any], base_dir: Path, overridden_keys: return resolved -def load_config(path: Path) -> dict[str, Any]: +def load_config(path: Path, explicit: bool = False) -> dict[str, Any]: + """Load config. + + ``explicit`` should be True when the path came from a user-provided ``-c`` + flag rather than the packaged default filename. Bug #12: previously a + missing file silently fell back to defaults in both cases, so a typo'd + ``-c confg.yaml`` looked like it worked while actually using defaults. + """ config = DEFAULT_CONFIG if DEFAULT_THRESHOLDS_PATH.exists(): with DEFAULT_THRESHOLDS_PATH.open("r", encoding="utf-8") as handle: @@ -165,6 +196,8 @@ def load_config(path: Path) -> dict[str, Any]: _mapping(thresholds, "packaged thresholds.thresholds") config = deep_merge(config, {"thresholds": thresholds}) if not path.exists(): + if explicit: + raise ConfigError(f"Config file not found: {path}") validate_config(config) return resolve_intel_paths(config, PROJECT_ROOT) with path.open("r", encoding="utf-8") as handle: @@ -175,6 +208,7 @@ def load_config(path: Path) -> dict[str, Any]: loaded = _mapping(loaded, "config root") if "intel" in loaded: _mapping(loaded["intel"], "intel") + _reject_unknown_keys(loaded, str(path)) overridden_keys = set((loaded.get("intel") or {}).keys()) merged = deep_merge(config, loaded) validate_config(merged) diff --git a/nettrace/engine.py b/nettrace/engine.py index 34e91ee..0ff163b 100644 --- a/nettrace/engine.py +++ b/nettrace/engine.py @@ -137,6 +137,31 @@ def analyze_pcap(pcap_path: Path, config: dict[str, Any]) -> AnalysisReport: if discarded_tcp_streams: warnings.add(f"TCP reassembly discarded {discarded_tcp_streams} incomplete or resource-limited streams.") + conflicting_overlaps = sum( + extractor.streams.conflicting_overlaps + for extractor in (dns_stream_extractor, http_extractor, tls_extractor, ftp_extractor) + ) + if conflicting_overlaps: + warnings.add( + f"TCP reassembly observed {conflicting_overlaps} overlapping retransmission(s) whose bytes " + "did not match already-buffered data; NetTrace kept the first-seen bytes. This can indicate " + "reassembly evasion and should be reviewed manually." + ) + + incomplete_streams = sum( + extractor.streams.incomplete_streams + for extractor in (dns_stream_extractor, http_extractor, tls_extractor, ftp_extractor) + ) + incomplete_bytes = sum( + extractor.streams.incomplete_buffered_bytes + for extractor in (dns_stream_extractor, http_extractor, tls_extractor, ftp_extractor) + ) + if incomplete_streams: + warnings.add( + f"TCP reassembly ended with {incomplete_streams} incomplete stream(s) containing " + f"{incomplete_bytes} buffered bytes; the capture may have been truncated mid-message." + ) + flows = list(flow_state.values()) findings = [] @@ -160,6 +185,17 @@ def analyze_pcap(pcap_path: Path, config: dict[str, Any]) -> AnalysisReport: tag_findings(findings) score_findings(findings) + + max_findings = config["limits"]["max_findings"] + if len(findings) > max_findings: + # Bug #10: there was previously no cap at all -- a hostile or extremely + # noisy capture could produce hundreds of thousands of findings and an + # unusable report. Keep the highest-severity findings rather than an + # arbitrary prefix. + findings.sort(key=lambda finding: finding.score, reverse=True) + findings = findings[:max_findings] + warnings.add(f"Findings truncated at {max_findings} entries (kept highest-severity).") + timeline_item_count = ( len(dns_events) + len(http_events) diff --git a/nettrace/mapping/attck_tagger.py b/nettrace/mapping/attck_tagger.py index f41193a..407acdb 100644 --- a/nettrace/mapping/attck_tagger.py +++ b/nettrace/mapping/attck_tagger.py @@ -65,11 +65,33 @@ def _refine_threat_intel_technique(finding: Finding) -> tuple[str, str] | None: return None +_HTTP_PORTS = {80, 8080, 8000, 8888} +_TLS_PORTS = {443, 4443, 8443, 9443} + + +def _refine_network_beaconing(finding: Finding) -> tuple[str, str] | None: + """Only tag a technique when the destination port confirms the protocol. + + Bug #1: the previous code mapped every non-DNS beacon straight to + T1071.001 (Web Protocols) even when the destination was, say, port 4444. + A beacon on an unconfirmed port is reported without an ATT&CK id rather + than guessing. + """ + port = finding.evidence.get("dst_port") + if port in _HTTP_PORTS: + return "T1071.001", "Application Layer Protocol: Web Protocols" + if port in _TLS_PORTS: + return "T1573", "Encrypted Channel" + return None + + def tag_findings(findings: list[Finding]) -> None: for finding in findings: attack = ATTACK_MAP.get(finding.category) if finding.category == "threat_intel_match": attack = _refine_threat_intel_technique(finding) or attack + elif finding.category == "network_beaconing": + attack = _refine_network_beaconing(finding) if attack: finding.attack_id, finding.attack_name = attack if finding.attack_id not in finding.tags: diff --git a/nettrace/mapping/severity.py b/nettrace/mapping/severity.py index ee8e172..c95e769 100644 --- a/nettrace/mapping/severity.py +++ b/nettrace/mapping/severity.py @@ -9,14 +9,26 @@ "network_beaconing": 70, "dga_domain": 70, "http_c2": 65, + "http_automation_client": 20, "tls_c2": 55, "high_frequency_connections": 55, "unusual_port": 50, "long_tls_session": 45, - "ftp_exfiltration": 75, + "ftp_exfiltration": 65, + "ftp_cleartext_credentials": 55, "misp_error": 15, } +# Bug #6: severity was previously a flat category lookup, so a 5-event borderline +# beacon and a 200-event tight-cadence beacon against known-bad infrastructure +# scored identically. Confidence (set per-finding by the analyzer that produced +# it -- see beaconing.py, dga_scorer.py, http_analyzer.py) now shifts the score. +CONFIDENCE_ADJUSTMENT = { + "high": 12, + "medium": 0, + "low": -18, +} + def _label(score: int) -> str: if score >= 85: @@ -33,9 +45,10 @@ def _label(score: int) -> str: def score_findings(findings: list[Finding]) -> None: for finding in findings: score = BASE_SCORES.get(finding.category, 20) + score += CONFIDENCE_ADJUSTMENT.get(finding.confidence, 0) if "MISP_HIT" in finding.tags: score += 10 if "IOC_MATCH" in finding.tags: score += 5 - finding.score = min(100, score) + finding.score = max(0, min(100, score)) finding.severity = _label(finding.score) diff --git a/nettrace/models/events.py b/nettrace/models/events.py index 83cb50f..ad78714 100644 --- a/nettrace/models/events.py +++ b/nettrace/models/events.py @@ -3,7 +3,52 @@ from dataclasses import asdict, dataclass, field import ipaddress from typing import Any -from urllib.parse import urlsplit +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +# Bug #16: HTTP query strings were stored and exported verbatim into JSON/HTML/ +# Markdown/PDF reports and the IOC list. FTP passwords were already redacted +# (see ftp_extractor.py) but tokens/session IDs/API keys in URLs were not. +SENSITIVE_QUERY_KEYS = { + "token", + "api_key", + "apikey", + "password", + "passwd", + "pass", + "session", + "sessionid", + "signature", + "auth", + "authorization", + "access_token", + "refresh_token", + "secret", + "client_secret", +} + + +def redact_sensitive_query_params(uri: str) -> str: + """Redact known-sensitive query parameter values in a URI, preserving + the path and parameter names (both are useful for analysis) but not the + secret values themselves.""" + if "?" not in uri: + return uri + parsed = urlsplit(uri) + if not parsed.query: + return uri + pairs = parse_qsl(parsed.query, keep_blank_values=True) + redacted_any = False + redacted_pairs = [] + for key, value in pairs: + if key.lower() in SENSITIVE_QUERY_KEYS: + redacted_pairs.append((key, "")) + redacted_any = True + else: + redacted_pairs.append((key, value)) + if not redacted_any: + return uri + new_query = urlencode(redacted_pairs) + return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, new_query, parsed.fragment)) @dataclass @@ -115,6 +160,11 @@ class IOC: value: str source: str packet_number: int = 0 + # "confirmed": derived from a parsed protocol artifact (DNS answer, HTTP host, + # TLS SNI, request URL). "observed": a raw flow endpoint IP with no protocol + # confirmation -- these dominate volume and should not be treated as equal- + # weight IOCs. See ioc_extractor.CONFIRMED_SOURCES. + confidence: str = "observed" def to_dict(self) -> dict[str, Any]: return asdict(self) diff --git a/nettrace/models/findings.py b/nettrace/models/findings.py index f344c60..761691d 100644 --- a/nettrace/models/findings.py +++ b/nettrace/models/findings.py @@ -15,6 +15,7 @@ class Finding: attack_name: str | None = None severity: str = "info" score: int = 0 + confidence: str = "medium" tags: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: diff --git a/nettrace/models/report.py b/nettrace/models/report.py index ddda782..d8b1a8a 100644 --- a/nettrace/models/report.py +++ b/nettrace/models/report.py @@ -30,6 +30,8 @@ def summary(self) -> dict[str, int]: "ftp_events": len(self.ftp_events), "flows": len(self.flows), "iocs": len(self.iocs), + "iocs_confirmed": sum(1 for ioc in self.iocs if ioc.confidence == "confirmed"), + "iocs_observed": sum(1 for ioc in self.iocs if ioc.confidence == "observed"), "findings": len(self.findings), "critical": sum(1 for finding in self.findings if finding.severity == "critical"), "high": sum(1 for finding in self.findings if finding.severity == "high"), diff --git a/nettrace/parsers/http_extractor.py b/nettrace/parsers/http_extractor.py index 6d84156..002e85e 100644 --- a/nettrace/parsers/http_extractor.py +++ b/nettrace/parsers/http_extractor.py @@ -3,7 +3,7 @@ from scapy.layers.inet import TCP from scapy.packet import Raw -from nettrace.models.events import HTTPEvent +from nettrace.models.events import HTTPEvent, redact_sensitive_query_params from nettrace.parsers.tcp_stream import TCPStreamBuffers, ip_endpoints HTTP_METHODS = {"GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH", "CONNECT", "TRACE"} @@ -81,7 +81,7 @@ def extract_http_event( dst_ip=endpoints[1], method=method, host=host, - uri=uri, + uri=redact_sensitive_query_params(uri), user_agent=user_agent, packet_number=packet_number, ) @@ -138,7 +138,7 @@ def feed(self, packet, packet_number: int = 0) -> list[HTTPEvent]: dst_ip=state.dst_ip, method=method, host=host, - uri=uri, + uri=redact_sensitive_query_params(uri), user_agent=user_agent, packet_number=state.first_packet_number, ) diff --git a/nettrace/parsers/tcp_stream.py b/nettrace/parsers/tcp_stream.py index 7eb3664..88ada5b 100644 --- a/nettrace/parsers/tcp_stream.py +++ b/nettrace/parsers/tcp_stream.py @@ -39,6 +39,11 @@ class TCPStreamState: pending: dict[int, _TCPSegment] = field(default_factory=dict) first_packet_number: int = 0 first_timestamp: float = 0.0 + # True once consume() has run at least once. Before that, base_seq is just + # the earliest segment received so far (out-of-order arrivals below it are + # still legitimate and must be prefixed in). After that, base_seq is a + # genuine "already parsed and removed" low-water mark -- see bug #8. + has_consumed: bool = False def ip_endpoints(packet) -> tuple[str, str] | None: @@ -64,11 +69,27 @@ def __init__( self._streams: OrderedDict[tuple[str, str, int, int], TCPStreamState] = OrderedDict() self._total_buffered_bytes = 0 self.discarded_streams = 0 + self.conflicting_overlaps = 0 @property def total_buffered_bytes(self) -> int: return self._total_buffered_bytes + @property + def incomplete_streams(self) -> int: + """Streams still open (never hit FIN/RST/eviction) when iteration ends. + + Bug #4: the engine only ever reported streams it actively discarded + for resource limits -- a stream that simply never got a chance to + finish (truncated capture, missing final segment) looked identical to + a clean report with nothing outstanding. + """ + return sum(1 for state in self._streams.values() if self._state_size(state) > 0) + + @property + def incomplete_buffered_bytes(self) -> int: + return sum(self._state_size(state) for state in self._streams.values()) + @staticmethod def _state_size(state: TCPStreamState) -> int: return len(state.buffer) + sum(len(item.payload) for item in state.pending.values()) @@ -83,17 +104,42 @@ def _remove_stream(self, key: tuple[str, str, int, int], *, discarded: bool) -> self.discarded_streams += 1 @staticmethod - def _merge_pending(state: TCPStreamState) -> None: + def _merge_pending(state: TCPStreamState) -> bool: + """Merge ready pending segments into the buffer. + + Returns True if a byte-level conflict was found between overlapping + retransmitted data and data already accepted (bug #3): the old code + accepted/trimmed overlapping segments without ever comparing bytes, so + a retransmission with *different* content than what was already + buffered was silently and arbitrarily resolved. + """ if state.base_seq is None or state.next_seq is None: - return + return False + conflict = False changed = True while changed: changed = False for sequence, segment in sorted(state.pending.items()): end = sequence + len(segment.payload) + # Bug #8: a segment entirely covering already-consumed bytes + # must be discarded outright, not left in `pending` forever -- + # but only once consume() has actually run. Before that, + # base_seq is merely the earliest segment received so far, and + # an out-of-order arrival ending exactly at base_seq is a + # legitimate missing prefix, not a stale retransmission. + if state.has_consumed and end <= state.base_seq: + del state.pending[sequence] + changed = True + break if end < state.base_seq or sequence > state.next_seq: continue if end <= state.next_seq and sequence >= state.base_seq: + # Fully inside the already-buffered range. Bug #3: compare + # bytes before discarding instead of assuming they match. + offset = sequence - state.base_seq + existing = bytes(state.buffer[offset : offset + len(segment.payload)]) + if existing and existing != segment.payload: + conflict = True del state.pending[sequence] changed = True break @@ -105,11 +151,16 @@ def _merge_pending(state: TCPStreamState) -> None: state.first_timestamp = segment.timestamp if sequence <= state.next_seq and end > state.next_seq: overlap = state.next_seq - sequence - state.buffer.extend(segment.payload[overlap:]) + if overlap > 0 and overlap <= len(state.buffer): + existing_tail = bytes(state.buffer[-overlap:]) + if existing_tail != segment.payload[:overlap]: + conflict = True + state.buffer.extend(segment.payload[max(overlap, 0) :]) state.next_seq = end del state.pending[sequence] changed = True break + return conflict def feed(self, packet, packet_number: int) -> TCPStreamState | None: endpoints = ip_endpoints(packet) @@ -170,7 +221,8 @@ def feed(self, packet, packet_number: int) -> TCPStreamState | None: existing = state.pending.get(sequence) if existing is None or len(payload) > len(existing.payload): state.pending[sequence] = segment - self._merge_pending(state) + if self._merge_pending(state): + self.conflicting_overlaps += 1 buffered_bytes = self._state_size(state) self._total_buffered_bytes += buffered_bytes - previous_size @@ -188,6 +240,7 @@ def consume(self, state: TCPStreamState, length: int, next_packet_number: int, n del state.buffer[:length] if state.base_seq is not None: state.base_seq += length + state.has_consumed = True state.first_packet_number = next_packet_number state.first_timestamp = next_timestamp self._total_buffered_bytes += self._state_size(state) - previous_size diff --git a/nettrace/report/markdown_report.py b/nettrace/report/markdown_report.py index c1b5313..d57ed40 100644 --- a/nettrace/report/markdown_report.py +++ b/nettrace/report/markdown_report.py @@ -172,16 +172,43 @@ def build_analyst_paragraph(report: dict[str, Any], victim: str, urls: list[str] summary = report.get("summary", {}) url_text = ", ".join(markdown_code(url) for url in urls[:5]) if urls else "no plaintext HTTP staging URLs" ip_text = ", ".join(markdown_code(ip) for ip in ips[:8]) if ips else "no public C2 candidates" - return ( - f"NetTrace analyzed {markdown_code(report.get('pcap_path', 'unknown'))} and identified {markdown_code(victim)} as the primary " - f"internal host. The capture contains {summary.get('dns_events', 0)} DNS events, " + findings = report.get("findings", []) + findings_count = summary.get("findings", len(findings)) + strong_findings = sum(1 for finding in findings if finding.get("severity") in {"high", "critical"}) + + host_clause = ( + f"identified {markdown_code(victim)} as the internal host with the highest observed external traffic " + "volume (a triage heuristic, not confirmation of compromise)" + if victim and victim != "unknown" + else "did not identify a single dominant internal host" + ) + base = ( + f"NetTrace analyzed {markdown_code(report.get('pcap_path', 'unknown'))} and {host_clause}. " + f"The capture contains {summary.get('dns_events', 0)} DNS events, " f"{summary.get('http_events', 0)} plaintext HTTP requests, {summary.get('tls_events', 0)} TLS SNI events, " - f"{summary.get('ftp_events', 0)} FTP commands, and {summary.get('flows', 0)} flows. The strongest analyst signal is the combination of HTTP staging " - f"activity ({url_text}) and high-volume or encrypted traffic involving {ip_text}. These behaviors are " - "consistent with malware staging and command-and-control triage, while heuristic findings should be " - "validated with packet context." + f"{summary.get('ftp_events', 0)} FTP commands, and {summary.get('flows', 0)} flows." ) + # Bug #18: the old text asserted "consistent with malware staging and C2 + # triage" even on a capture with zero findings and zero staging URLs. The + # conclusion now scales with what was actually found. + if strong_findings: + conclusion = ( + f" {strong_findings} high/critical-severity finding(s) were produced, with the strongest signal being " + f"HTTP staging activity ({url_text}) and high-volume or encrypted traffic involving {ip_text}. This " + "warrants malware staging and command-and-control triage; heuristic findings should still be validated " + "with packet context before conclusions are drawn." + ) + elif findings_count: + conclusion = ( + f" NetTrace produced {findings_count} lower-confidence heuristic finding(s) ({url_text}; {ip_text}) that " + "should be reviewed with packet context. None reached high or critical severity, so this capture does " + "not, on its own, establish malware staging or command-and-control activity." + ) + else: + conclusion = " No behavioral findings were produced from this capture." + return base + conclusion + def build_markdown(report: dict[str, Any], source: str = "", source_url: str = "") -> str: summary = report.get("summary", {}) @@ -233,7 +260,7 @@ def build_markdown(report: dict[str, Any], source: str = "", source_url: str = " lines.extend( [ "", - "## ATT&CK Techniques", + "## Potential ATT&CK Technique Associations", "", bullet_list(techniques), "", diff --git a/nettrace/rules/attck_map.yaml b/nettrace/rules/attck_map.yaml index 0870110..e594fcc 100644 --- a/nettrace/rules/attck_map.yaml +++ b/nettrace/rules/attck_map.yaml @@ -1,9 +1,15 @@ +# NOTE on "network_beaconing" and "high_frequency_connections": these are +# intentionally absent from this map. Bug #1: mapping arbitrary periodic TCP/UDP +# traffic straight to T1071.001 (Web Protocols) or raw packet volume straight to +# T1020 (Automated Exfiltration) overstates what NetTrace actually observed. +# network_beaconing gets a technique only when attck_tagger.py's +# _refine_network_beaconing() can confirm the destination port is actually +# HTTP/TLS; otherwise it is reported as an unmapped behavioral finding. +# high_frequency_connections never gets an ATT&CK id by itself -- packet count +# alone does not establish exfiltration. dns_beaconing: id: T1071.004 name: "Application Layer Protocol: DNS" -network_beaconing: - id: T1071.001 - name: "Application Layer Protocol: Web Protocols" http_c2: id: T1071.001 name: "Application Layer Protocol: Web Protocols" @@ -14,11 +20,8 @@ dga_domain: id: T1568.002 name: "Dynamic Resolution: Domain Generation Algorithms" long_tls_session: - id: T1090 - name: "Proxy" -high_frequency_connections: - id: T1020 - name: "Automated Exfiltration" + id: T1573 + name: "Encrypted Channel" unusual_port: id: T1571 name: "Non-Standard Port" diff --git a/nettrace/rules/dga_allowlist.yaml b/nettrace/rules/dga_allowlist.yaml index 1bc0c64..ab38ba5 100644 --- a/nettrace/rules/dga_allowlist.yaml +++ b/nettrace/rules/dga_allowlist.yaml @@ -1,18 +1,26 @@ +# Domain-boundary matching only. A domain matches a suffix if it equals it, or +# ends with "." + suffix -- never via raw substring. See bug #2: +# "microsoft-login-x8f39q.biz" must NOT match "microsoft". +domains: + - "microsoft.com" + - "windowsupdate.com" + - "msedge.net" + - "azure.com" + - "azureedge.net" + - "azurefd.us" + - "cloudapp.azure.com" + - "trafficmanager.net" suffixes: - - ".local" - - ".localdomain" - - ".microsoft.com" - - ".windowsupdate.com" - - ".msedge.net" - - ".azureedge.net" - - ".azurefd.us" - - ".cloudapp.azure.com" - - ".trafficmanager.net" -contains: - - "microsoft" - - "windowsupdate" - - "msedge" - - "azure" + - "local" + - "localdomain" + - "microsoft.com" + - "windowsupdate.com" + - "msedge.net" + - "azure.com" + - "azureedge.net" + - "azurefd.us" + - "cloudapp.azure.com" + - "trafficmanager.net" regexes: - "^desktop-[a-z0-9-]+(\\..*)?$" - "^[a-z0-9-]+-dc(\\..*)?$" diff --git a/pyproject.toml b/pyproject.toml index a915311..42c04a5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ name = "nettrace-malware-analysis" version = "1.0.0" description = "Offline malware traffic analysis platform for PCAP triage, IOC extraction, ATT&CK mapping, and analyst reports." readme = "README.md" -requires-python = ">=3.11" +requires-python = ">=3.11,<3.13" license = { file = "LICENSE" } authors = [ { name = "Aryan" } @@ -36,12 +36,14 @@ dependencies = [ "scapy~=2.6", "jinja2~=3.1", "pyyaml~=6.0", - "requests~=2.32", - "pymisp~=2.4", "reportlab~=4.2" ] [project.optional-dependencies] +misp = [ + "pymisp~=2.4", + "requests~=2.32" +] dev = [ "pypdf~=6.0", "pytest~=8.4", diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..dd00615 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +-r requirements.txt +pypdf~=6.0 +pytest~=8.4 +pytest-cov~=7.1 diff --git a/requirements-misp.txt b/requirements-misp.txt new file mode 100644 index 0000000..3048e42 --- /dev/null +++ b/requirements-misp.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pymisp~=2.4 +requests~=2.32 diff --git a/requirements.txt b/requirements.txt index 87eb0ab..9b1d4d7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,9 +1,4 @@ scapy~=2.6 jinja2~=3.1 pyyaml~=6.0 -requests~=2.32 -pymisp~=2.4 reportlab~=4.2 -pypdf~=6.0 -pytest~=8.4 -pytest-cov~=7.1 diff --git a/tests/test_cli.py b/tests/test_cli.py index 37e094a..4907619 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -63,7 +63,7 @@ def test_cli_prints_analysis_warnings(tmp_path, monkeypatch, capsys): timeline=[], warnings=["HTTP events truncated at 10 entries."], ) - monkeypatch.setattr("nettrace.cli.load_config", lambda _path: {}) + monkeypatch.setattr("nettrace.cli.load_config", lambda _path, explicit=False: {}) monkeypatch.setattr("nettrace.cli.analyze_pcap", lambda _path, _config: report) monkeypatch.setattr( "sys.argv", @@ -99,7 +99,7 @@ def test_cli_handles_report_output_errors_without_traceback(tmp_path, monkeypatc findings=[], timeline=[], ) - monkeypatch.setattr("nettrace.cli.load_config", lambda _path: {}) + monkeypatch.setattr("nettrace.cli.load_config", lambda _path, explicit=False: {}) monkeypatch.setattr("nettrace.cli.analyze_pcap", lambda _path, _config: report) monkeypatch.setattr("sys.argv", ["nettrace", "sample.pcap", "--output", str(output_file)]) diff --git a/tests/test_config.py b/tests/test_config.py index 6e1187f..fe51187 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -53,3 +53,39 @@ def test_load_config_rejects_invalid_threshold_and_port_types(tmp_path): bad_port.write_text("protocols:\n http_ports: [80, invalid]\n", encoding="utf-8") with pytest.raises(ConfigError, match="http_ports"): load_config(bad_port) + + +def test_load_config_errors_on_missing_explicit_path(tmp_path): + # Bug #12: a user-specified -c path that doesn't exist must error, not + # silently fall back to defaults. + missing = tmp_path / "confg.yaml" # typo'd filename, on purpose + + with pytest.raises(ConfigError, match="Config file not found"): + load_config(missing, explicit=True) + + +def test_load_config_still_defaults_when_implicit_path_missing(tmp_path): + # Unchanged behavior: the packaged default config.yaml is optional. + missing = tmp_path / "config.yaml" + + config = load_config(missing, explicit=False) + + assert config["thresholds"]["high_frequency_connections"] == 50 + + +def test_load_config_rejects_unknown_threshold_key(tmp_path): + # Bug #13: a typo'd threshold key previously merged in silently while the + # real threshold stayed at its default with no warning. + config_path = tmp_path / "config.yaml" + config_path.write_text("thresholds:\n high_frequncy_connections: 999\n", encoding="utf-8") + + with pytest.raises(ConfigError, match="Unknown key"): + load_config(config_path) + + +def test_load_config_rejects_unknown_top_level_key(tmp_path): + config_path = tmp_path / "config.yaml" + config_path.write_text("logging:\n level: debug\n", encoding="utf-8") + + with pytest.raises(ConfigError, match="Unknown top-level key"): + load_config(config_path) diff --git a/tests/test_http_analyzer.py b/tests/test_http_analyzer.py index 12797bc..27e5a3e 100644 --- a/tests/test_http_analyzer.py +++ b/tests/test_http_analyzer.py @@ -1,5 +1,8 @@ +from scapy.all import IP, Raw, TCP + from nettrace.analysis.http_analyzer import analyze_http_events -from nettrace.models.events import HTTPEvent +from nettrace.models.events import HTTPEvent, redact_sensitive_query_params +from nettrace.parsers.http_extractor import extract_http_event def test_executable_download_detection_ignores_case_and_query_string(): @@ -20,3 +23,74 @@ def test_blank_suspicious_user_agent_path_is_ignored(): findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": ""}}) assert findings == [] + + +def test_post_to_executable_path_is_not_labeled_a_download(): + # Bug #5: POST/PUT/PATCH to a *.exe path is not evidence of a download. + event = HTTPEvent(1.0, "10.0.0.5", "203.0.113.66", "POST", "example.com", "/upload/report.exe") + + findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": ""}}) + + assert len(findings) == 1 + assert "download" not in findings[0].title.lower() + assert findings[0].confidence == "low" + + +def test_get_executable_path_is_labeled_possible_download(): + event = HTTPEvent(1.0, "10.0.0.5", "203.0.113.66", "GET", "example.com", "/payload.exe") + + findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": ""}}) + + assert findings[0].title == "Possible executable/script download request" + assert findings[0].confidence == "medium" + + +def test_suspicious_user_agent_matches_by_tool_name_ignoring_version(tmp_path): + # Bug #17: an entry for "curl/7.68.0" should still catch "curl/8.4.0", + # and the finding is now low-confidence context, not "suspicious". + ua_file = tmp_path / "uas.txt" + ua_file.write_text("curl/7.68.0\n", encoding="utf-8") + event = HTTPEvent( + 1.0, "10.0.0.5", "203.0.113.66", "GET", "example.com", "/", + user_agent="curl/8.4.0", + ) + + findings = analyze_http_events([event], {"intel": {"suspicious_user_agents": str(ua_file)}}) + + assert len(findings) == 1 + assert findings[0].category == "http_automation_client" + assert findings[0].confidence == "low" + + +def test_redact_sensitive_query_params_removes_secrets(): + # Bug #16: HTTP query params were stored/exported verbatim, leaking + # tokens/API keys/session IDs into every report format. + assert redact_sensitive_query_params("/reset-password?token=secret123") == "/reset-password?token=%3Credacted%3E" + assert redact_sensitive_query_params("/api?api_key=123456") == "/api?api_key=%3Credacted%3E" + + +def test_redact_sensitive_query_params_preserves_non_sensitive_params(): + result = redact_sensitive_query_params("/download?signature=xyz&legit_param=keepme") + assert "legit_param=keepme" in result + assert "signature=%3Credacted%3E" in result + + +def test_redact_sensitive_query_params_leaves_benign_query_untouched(): + assert redact_sensitive_query_params("/search?q=normal+query") == "/search?q=normal+query" + + +def test_redact_sensitive_query_params_no_query_string_untouched(): + assert redact_sensitive_query_params("/plain/path") == "/plain/path" + + +def test_http_extractor_redacts_query_params_in_constructed_event(): + packet = IP(src="10.0.0.5", dst="203.0.113.1") / TCP(sport=50000, dport=80, seq=100, flags="PA") / Raw( + load=b"GET /login?session=abcdef123 HTTP/1.1\r\nHost: example.com\r\n\r\n" + ) + packet.time = 1.0 + + event = extract_http_event(packet, packet_number=1) + + assert event is not None + assert "abcdef123" not in event.uri + assert "session=%3Credacted%3E" in event.uri diff --git a/tests/test_stream_reassembly.py b/tests/test_stream_reassembly.py index 1265795..4757fe2 100644 --- a/tests/test_stream_reassembly.py +++ b/tests/test_stream_reassembly.py @@ -175,7 +175,7 @@ def test_engine_uses_stream_reassembly(tmp_path): report = analyze_pcap(capture, load_config(Path("does-not-exist.yaml"))) assert len(report.http_events) == 1 - assert any(finding.title == "Executable download over HTTP" for finding in report.findings) + assert any(finding.title == "Possible executable/script download request" for finding in report.findings) def test_http_stream_resets_when_tcp_tuple_is_reused(): @@ -299,7 +299,7 @@ def test_engine_reassembles_ipv4_fragments(tmp_path): assert len(report.http_events) == 1 assert report.http_events[0].host == "evil.example" - assert any(finding.title == "Executable download over HTTP" for finding in report.findings) + assert any(finding.title == "Possible executable/script download request" for finding in report.findings) def test_engine_reports_resource_truncation(tmp_path): @@ -331,4 +331,63 @@ def test_engine_uses_configured_http_ports(tmp_path): report = analyze_pcap(capture, config) assert len(report.http_events) == 1 - assert any(finding.title == "Executable download over HTTP" for finding in report.findings) + assert any(finding.title == "Possible executable/script download request" for finding in report.findings) + + +def test_engine_caps_findings_at_configured_max(tmp_path): + # Bug #10: there was previously no cap -- a hostile/noisy capture could + # produce an unbounded number of findings. + packets = [] + for index in range(5): + packet = ( + IP(src="10.0.0.5", dst=f"203.0.113.{index}") + / TCP(sport=40000 + index, dport=4444, seq=100, flags="PA") + / Raw(load=b"x" * 10) + ) + packet.time = float(index) + packets.append(packet) + capture = tmp_path / "many-findings.pcap" + wrpcap(str(capture), packets) + config = load_config(Path("does-not-exist.yaml")) + config["limits"]["max_findings"] = 2 + + report = analyze_pcap(capture, config) + + assert len(report.findings) == 2 + assert any("Findings truncated at 2 entries" in warning for warning in report.warnings) + + +def test_engine_surfaces_conflicting_overlap_warning(tmp_path): + # Bug #3: conflicting overlapping retransmissions must be visible to the + # analyst, not silently resolved. Uses an incomplete request (no blank + # line) so the bytes stay unconsumed in the buffer -- a complete request + # would be consumed immediately and hit the bug #8 path instead. + packets = [ + IP(src="10.0.0.5", dst="45.33.32.156") / TCP(sport=50000, dport=80, seq=100, flags="PA") + / Raw(load=b"GET /aaaaaaaaaaaa"), + IP(src="10.0.0.5", dst="45.33.32.156") / TCP(sport=50000, dport=80, seq=100, flags="PA") + / Raw(load=b"GET /bbbbbbbbbbbb"), + ] + for number, packet in enumerate(packets, 1): + packet.time = float(number) + capture = tmp_path / "overlap-conflict.pcap" + wrpcap(str(capture), packets) + + report = analyze_pcap(capture, load_config(Path("does-not-exist.yaml"))) + + assert any("overlapping retransmission" in warning for warning in report.warnings) + + +def test_engine_surfaces_incomplete_stream_warning(tmp_path): + # Bug #4: a stream still open (no FIN/RST, never hit a resource limit) + # when the capture ends was previously invisible in the report. + packet = IP(src="10.0.0.5", dst="45.33.32.156") / TCP(sport=50000, dport=80, seq=100, flags="PA") / Raw( + load=b"GET /never-finishes" + ) + packet.time = 1.0 + capture = tmp_path / "truncated.pcap" + wrpcap(str(capture), [packet]) + + report = analyze_pcap(capture, load_config(Path("does-not-exist.yaml"))) + + assert any("incomplete stream" in warning for warning in report.warnings) diff --git a/tests/test_tcp_stream.py b/tests/test_tcp_stream.py new file mode 100644 index 0000000..dd7899e --- /dev/null +++ b/tests/test_tcp_stream.py @@ -0,0 +1,92 @@ +from scapy.all import IP, Raw, TCP + +from nettrace.parsers.tcp_stream import TCPStreamBuffers + + +def _segment(seq: int, payload: bytes, packet_number: int, timestamp: float = 1.0): + packet = IP(src="10.0.0.5", dst="203.0.113.10") / TCP(sport=50000, dport=80, seq=seq, flags="A") / Raw(load=payload) + packet.time = timestamp + return packet + + +def test_stale_retransmission_below_base_seq_is_discarded_not_leaked(): + # Bug #8: a segment fully covering bytes already consumed by a protocol + # parser must be discarded outright, not left in `pending` forever eating + # the pending-segment budget. + buffers = TCPStreamBuffers() + state = buffers.feed(_segment(100, b"GET / HTTP/1.1\r\n", 1), 1) + assert state is not None + buffers.consume(state, len(b"GET / HTTP/1.1\r\n"), next_packet_number=2, next_timestamp=1.1) + assert state.has_consumed is True + assert state.base_seq == 116 + + # A retransmission of the already-consumed first segment arrives late. + state = buffers.feed(_segment(100, b"GET / HTTP/1.1\r\n", 3, timestamp=1.2), 3) + assert state is not None + assert state.pending == {}, "stale retransmission must not sit in pending forever" + assert bytes(state.buffer) == b"" + + +def test_retransmission_ending_exactly_at_base_seq_is_not_reprepended(): + # Bug #8 (exact case from the review): a segment ending exactly at the new + # base_seq must not be prepended back into the live buffer -- that would + # resurrect already-consumed, already-reported data. + buffers = TCPStreamBuffers() + state = buffers.feed(_segment(100, b"AAAA", 1), 1) + buffers.consume(state, 4, next_packet_number=2, next_timestamp=1.1) + assert state.base_seq == 104 + + state = buffers.feed(_segment(100, b"AAAA", 3, timestamp=1.2), 3) # end == 104 == base_seq + assert bytes(state.buffer) == b"", "must not resurrect consumed bytes" + + +def test_out_of_order_prefix_before_any_consumption_still_merges(): + # Sanity check: the has_consumed guard must not break the normal, + # legitimate out-of-order arrival case (no consume() has happened yet). + buffers = TCPStreamBuffers() + state = buffers.feed(_segment(102, b"ER analyst\r\n", 2, timestamp=2.0), 2) + assert bytes(state.buffer) == b"ER analyst\r\n" + + state = buffers.feed(_segment(100, b"US", 1, timestamp=1.0), 1) + assert bytes(state.buffer) == b"USER analyst\r\n" + + +def test_conflicting_overlap_is_detected_not_silently_resolved(): + # Bug #3: overlapping retransmitted bytes that *differ* from what's + # already buffered were never compared -- silently accepted or dropped + # with no signal to the analyst. + buffers = TCPStreamBuffers() + buffers.feed(_segment(100, b"AAAABBBB", 1), 1) + assert buffers.conflicting_overlaps == 0 + + # Retransmission of the same range with DIFFERENT bytes -- a real conflict. + buffers.feed(_segment(100, b"AAAAXXXX", 2, timestamp=1.1), 2) + assert buffers.conflicting_overlaps == 1 + + +def test_identical_overlap_is_not_flagged_as_conflict(): + buffers = TCPStreamBuffers() + buffers.feed(_segment(100, b"AAAABBBB", 1), 1) + buffers.feed(_segment(100, b"AAAABBBB", 2, timestamp=1.1), 2) + + assert buffers.conflicting_overlaps == 0 + + +def test_incomplete_stream_visible_at_eof(): + # Bug #4: a stream that never receives its closing FIN/RST and never hits + # a resource limit was previously invisible in the report -- it wasn't + # "discarded", it just silently vanished from consideration. + buffers = TCPStreamBuffers() + buffers.feed(_segment(100, b"incomplete HTTP request, no blank line yet", 1), 1) + + assert buffers.incomplete_streams == 1 + assert buffers.incomplete_buffered_bytes == len(b"incomplete HTTP request, no blank line yet") + + +def test_fully_consumed_stream_is_not_counted_as_incomplete(): + buffers = TCPStreamBuffers() + state = buffers.feed(_segment(100, b"GET / HTTP/1.1\r\n\r\n", 1), 1) + buffers.consume(state, len(b"GET / HTTP/1.1\r\n\r\n"), next_packet_number=2, next_timestamp=1.1) + + assert buffers.incomplete_streams == 0 + assert buffers.incomplete_buffered_bytes == 0