Files
NetScaler_forensik/tests/test_indicator_matching.py

98 lines
4.0 KiB
Python

"""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()