"""Public intelligence provenance and representative matching regressions.""" import base64 import json import tempfile import unittest from pathlib import Path from typing import Any, ClassVar from urllib.parse import quote from netscaler_triage.decoding import decode_layers, has_encoding_candidate from netscaler_triage.engine import scan from netscaler_triage.matching import Matcher, public_config class IntelInventoryTests(unittest.TestCase): catalog: ClassVar[dict[str, Any]] matcher: ClassVar[Matcher] @classmethod def setUpClass(cls) -> None: cls.catalog = public_config() cls.matcher = Matcher([]) def test_unique_catalog_and_resolvable_public_provenance(self) -> None: sources = {row["id"]: row for row in self.catalog["sources"]} self.assertGreaterEqual(len(sources), 37) rows = self.catalog["indicators"] self.assertEqual(len(rows), len({(row["type"], row["value"]) for row in rows})) self.assertEqual(len(rows), len({row["id"] for row in rows})) for row in rows: with self.subTest(indicator=row["id"]): self.assertTrue(row["roles"]) self.assertTrue(row["observations"]) self.assertTrue(row["source_ids"]) for source_id in row["source_ids"]: self.assertIn(source_id, sources) self.assertIn(sources[source_id]["url"], row["source_urls"]) for observation in row["observations"]: self.assertTrue(observation["source_dates"]) self.assertIn("original_status", observation) for date in observation["source_dates"]: self.assertIn(date["id"], sources) self.assertIn("published", date) self.assertTrue(date["retrieved"]) urls = {source["url"] for source in sources.values()} self.assertIn( "https://misp.ifin.network/feed/63670851-a19e-4fc4-8fe5-c89bebc8e921.json", urls ) self.assertTrue(any("elastic" in url for url in urls)) self.assertTrue(any("sygnia.co" in url for url in urls)) def test_secondary_direction_conflicts_and_shared_proxy_context_survive(self) -> None: rows = self.catalog["indicators"] observations = [ob for row in rows for ob in row["observations"]] conflicts = [ ob for ob in observations if ob.get("misp_type") == "ip-dst" and "direction conflict" in ob.get("description", "") ] self.assertTrue(conflicts) shared = next(row for row in rows if row["value"] == "104.28.215.136") context = [ob for ob in shared["observations"] if ob.get("misp_to_ids") is False] self.assertTrue(context) self.assertTrue(any(ob["original_status"] == "context_only" for ob in context)) self.assertIn("shared", json.dumps(shared).lower()) def test_every_public_indicator_matches_raw_and_decoded_views(self) -> None: for row in self.catalog["indicators"]: value = row["value"] with self.subTest(indicator=row["id"]): _, hits, _ = self.matcher.search_with_details(value) self.assertTrue(any(hit["id"] == row["id"] for hit in hits)) raw = self.matcher.search_sigma({"raw": value}) if value == "/var/python/bin/customsnmpd": self.assertFalse( raw ) # Legitimate binary: standalone Sigma intentionally omitted. else: self.assertTrue(any("raw" in hit["basis"] for hit in raw)) encoded = base64.b64encode(value.encode()).decode() layers = decode_layers("base64:" + encoded) decoded = [ layer["output"] for layer in layers if layer.get("output") and layer.get("output_encoding") != "hex" ] self.assertIn(value, decoded) sigma = self.matcher.search_sigma({"raw": "encoded event"}, decoded) if value == "/var/python/bin/customsnmpd": self.assertFalse(sigma) else: self.assertTrue( any(any(b.startswith("decoded:") for b in hit["basis"]) for hit in sigma) ) escaped = quote(value, safe="") if escaped != value: url_layers = decode_layers(escaped) self.assertTrue(any(layer.get("output") == value for layer in url_layers)) def test_encoding_gate_accepts_canonical_public_indicator_fields(self) -> None: for row in self.catalog["indicators"]: encoded = base64.b64encode(row["value"].encode()).decode() for value in (encoded, encoded.rstrip("=")): with self.subTest(indicator=row["id"], encoded=value): self.assertTrue(has_encoding_candidate(value)) self.assertTrue(has_encoding_candidate("payload=" + value)) for payload in ("HTTP_X_UX", "HTTP_X_UX" * 8): encoded = base64.b64encode(payload.encode()).decode().rstrip("=") self.assertTrue(has_encoding_candidate(encoded)) for ordinary in ("heartbeat", "INFO routine system heartbeat", "GET / HTTP/1.1"): self.assertFalse(has_encoding_candidate(ordinary)) # A valid but noncanonical spelling must not pass the whole-field gate. self.assertFalse(has_encoding_candidate("SFRUUF9YX1VZ==")) def test_ipv4_literals_do_not_match_address_prefixes(self) -> None: for row in self.catalog["indicators"]: if row["type"] != "ipv4": continue for text in ("9" + row["value"], row["value"] + "9", row["value"] + ".1"): with self.subTest(value=row["value"], text=text): _, hits, _ = self.matcher.search_with_details(text) self.assertFalse(any(hit["id"] == row["id"] for hit in hits)) self.assertFalse(self.matcher.search_sigma({"raw": text})) def test_behavior_patterns_match_real_syntax_not_escaped_backslashes(self) -> None: samples = [ "SSL_HANDSHAKE_FAILURE DTLSv1.0 Handshake failure-Internal Error", "NSPPE-42 exit with orphan rings", "pitboss PPE unexpectedly died NSPPE;id;#", "HTTP_X_UX_0", "*/5 * * * * root perl /var/tmp/.nsmon/nsmon.pl", ] for sample in samples: with self.subTest(sample=sample): self.assertTrue(self.matcher.search_sigma({"raw": sample})) benign = self.matcher.search_sigma({"raw": "pitboss PPE missed too many heartbeats"}) self.assertFalse(any(hit.get("level") in {"high", "critical"} for hit in benign)) def test_http_user_agent_encodings_keep_original_and_source_metadata(self) -> None: payload = "/var/tmp/.nsmon/nsmon.pl 149.104.78.208" b64 = base64.b64encode(payload.encode()).decode() variants = [ quote(payload, safe=""), b64, b64.rstrip("="), quote(base64.b64encode(quote(payload, safe="").encode()).decode(), safe=""), ] originals = [ f'192.0.2.10 - - [01/Oct/2026:12:00:00 +0000] "GET / HTTP/1.1" 200 0 "-" "{variant}"' for variant in variants ] with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() logs = root / "logs" logs.mkdir() (logs / "httpaccess.log").write_text("\n".join(originals) + "\n") output = root / "output" scan([("ADC", logs)], output, []) events = [ json.loads(line) for line in (output / "findings.jsonl").read_text().splitlines() ] self.assertEqual(len(events), len(originals)) for event, original in zip(events, originals, strict=True): self.assertEqual(event["raw"], original) hits = [ hit for hit in event["indicator_matches"] if hit["value"] == "149.104.78.208" ] self.assertTrue(hits) self.assertTrue(all(hit["source_urls"] and hit["source_ids"] for hit in hits)) self.assertTrue(any(layer.get("output") == payload for layer in event["decoding"]))