70 lines
3.3 KiB
Python
70 lines
3.3 KiB
Python
"""Bundled public references work offline without incident-specific configuration."""
|
|
|
|
import ipaddress
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from netscaler_triage.engine import scan
|
|
from netscaler_triage.matching import Matcher, load_config, public_config
|
|
|
|
|
|
class PublicDefaultsTests(unittest.TestCase):
|
|
def test_packaged_references_have_sources_and_only_public_addresses(self) -> None:
|
|
config = public_config()
|
|
sources = {row["id"] for row in config["sources"]}
|
|
self.assertGreaterEqual(len(sources), 37)
|
|
self.assertGreaterEqual(len(config["indicators"]), 35)
|
|
self.assertGreaterEqual(len(config["patterns"]), 9)
|
|
for indicator in config["indicators"]:
|
|
self.assertEqual(indicator["status"], "reference_only")
|
|
self.assertTrue(set(indicator["source_ids"]) <= sources)
|
|
if indicator["type"] == "ipv4":
|
|
self.assertTrue(ipaddress.ip_address(indicator["value"]).is_global)
|
|
|
|
def test_default_public_ioc_and_pattern_matching(self) -> None:
|
|
matcher = Matcher([])
|
|
_, indicators = matcher.search("peer 143.198.7.94:443")
|
|
self.assertTrue(any("mandiant" in row["source_ids"] for row in indicators))
|
|
_, _, details = matcher.search_with_details("HTTP_NSC_CLIENTTYPE")
|
|
self.assertTrue(any(row.get("name") == "ti_shell_headers" for row in details))
|
|
|
|
def test_explicit_duplicate_config_does_not_duplicate_matches(self) -> None:
|
|
text = "peer 143.198.7.94:443 HTTP_NSC_CLIENTTYPE"
|
|
self.assertEqual(Matcher([]).search(text), Matcher([public_config()]).search(text))
|
|
|
|
def test_default_opt_out_keeps_explicit_configuration(self) -> None:
|
|
text = "peer 143.198.7.94:443 HTTP_NSC_CLIENTTYPE"
|
|
self.assertEqual(Matcher([], defaults=False).search(text), ([], []))
|
|
self.assertTrue(Matcher([public_config()], defaults=False).search(text)[1])
|
|
|
|
def test_scan_without_config_retains_reference_provenance(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary).resolve()
|
|
source = root / "logs"
|
|
source.mkdir()
|
|
(source / "network.log").write_text("peer 143.198.7.94:443\n", encoding="utf-8")
|
|
result = scan([("adc", source)], root / "out", [])
|
|
self.assertEqual(result["counts"]["findings"], 1)
|
|
event = json.loads((root / "out/findings.jsonl").read_text())
|
|
hit = next(row for row in event["indicator_matches"] if row["value"] == "143.198.7.94")
|
|
self.assertEqual(hit["status"], "reference_only")
|
|
self.assertIn("mandiant", hit["source_ids"])
|
|
|
|
def test_own_ioc_csv_can_be_reused_as_configuration(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary).resolve()
|
|
source = root / "logs"
|
|
source.mkdir()
|
|
(source / "network.log").write_text("peer 143.198.7.94:443\n", encoding="utf-8")
|
|
scan([("adc", source)], root / "out", [])
|
|
config = load_config(root / "out/IOCS.csv")
|
|
_, hits = Matcher([config], defaults=False).search("peer 143.198.7.94:443")
|
|
hit = next(row for row in hits if row["value"] == "143.198.7.94")
|
|
self.assertEqual(hit["type"], "ipv4")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|