"""Synthetic indicator boundaries, overlapping values and contextual packs.""" import unittest from netscaler_triage.matching import Matcher class IndicatorMatchingTests(unittest.TestCase): def matcher(self, kind: str, value: str) -> Matcher: return Matcher([{"indicators": [{"type": kind, "value": value}]}], defaults=False) def test_typed_boundaries_and_case(self) -> None: cases = [ ("ipv4", "192.0.2.1", "peer 192.0.2.1:443", "peer 192.0.2.10"), ("domain", "example.test", "https://sub.example.test/a", "example.test.evil"), ("domain", "example.test", "EXAMPLE.TEST", "notexample.test"), ("path", "/tmp/evil", "curl https://host.test/tmp/evil", "/tmp/evil.backup"), ("path", "/tmp/evil", "cat /tmp/evil", "/safe/tmp/evil"), ("path", "/tmp/evil", "cat /tmp/evil", "/tmp/EVIL"), ("path", "/tmp/evil", "cat /tmp/evil", "/tmp/evil%2Ebackup"), ( "url", "http://example.test/a", "http://example.test/a", "http://example.test/a%2Flonger", ), ("path", "evil.php", "GET /vpn/evil.php?x=1", "/vpn/not-evil.php"), ("url", "http://example.test/a", "HTTP://EXAMPLE.TEST/a", "http://example.test/ab"), ("url", "http://example.test/A", "http://example.test/A", "http://example.test/a"), ("sha256", "a" * 64, "sha256=" + "A" * 64, "g" + "a" * 64), ] for kind, value, positive, negative in cases: with self.subTest(kind=kind, value=value): matcher = self.matcher(kind, value) self.assertTrue(matcher.search(positive)[1]) self.assertFalse(matcher.search(negative)[1]) def test_unicode_case_mapping_does_not_crash_or_invent_ascii_indicator(self) -> None: matcher = self.matcher("literal", "id") self.assertFalse(matcher.search("İD")[1]) self.assertTrue(matcher.search("ID")[1]) def test_prefix_indicators_are_not_lost_to_longest_alternative(self) -> None: matcher = Matcher( [{"indicators": [{"type": "literal", "value": v} for v in ["abc", "abcdef"]]}], defaults=False, ) self.assertEqual({x["value"] for x in matcher.search("abcdef")[1]}, {"abc", "abcdef"}) matcher = Matcher( [{"indicators": [{"type": "path", "value": v} for v in ["/tmp/", "/tmp/evil"]]}], defaults=False, ) self.assertEqual( {x["value"] for x in matcher.search("/tmp/evil")[1]}, {"/tmp/", "/tmp/evil"} ) def test_invalid_ip_and_hash_fail_before_scanning(self) -> None: for kind, value in [("ipv4", "999.1.2.3"), ("ipv4", "2001:db8::1"), ("sha256", "abc")]: with self.subTest(kind=kind), self.assertRaises(ValueError): self.matcher(kind, value) def test_sigma_metadata_cannot_bypass_context(self) -> None: yaml = """title: Contextual metadata example id: d7518e0b-4717-45f2-babb-970ab3d546dd logsource: product: netscaler service: http_access indicators: - type: ipv4 value: 192.0.2.42 status: reference_only detection: selection: uri: /suspicious raw|contains: 192.0.2.42 condition: selection """ config = {"sigma_rules": [{"source": "example.yml", "yaml": yaml}]} matcher = Matcher([config], defaults=False) self.assertEqual(matcher.search("peer 192.0.2.42"), ([], [])) self.assertFalse(matcher.indicators) self.assertTrue( matcher.search_sigma( {"kind": "http_access", "raw": "peer 192.0.2.42", "uri": "/suspicious"} ) ) self.assertFalse( matcher.search_sigma( {"kind": "http_access", "raw": "peer 192.0.2.42", "uri": "/health"} ) ) with self.assertRaisesRegex(ValueError, "Duplicate rule identifier"): Matcher([config, config], defaults=False) if __name__ == "__main__": unittest.main()