96 lines
4.9 KiB
Python
96 lines
4.9 KiB
Python
"""Boundary semantics for the bundled public reference Sigma groups."""
|
|
|
|
import unittest
|
|
from importlib.resources import files
|
|
from typing import ClassVar
|
|
from urllib.parse import urlsplit
|
|
|
|
from netscaler_triage.sigma_rules import SigmaMatcher
|
|
|
|
|
|
class PublicBoundaryTests(unittest.TestCase):
|
|
matcher: ClassVar[SigmaMatcher]
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
content = files("netscaler_triage").joinpath("data", "public_sigma.yml").read_text()
|
|
cls.matcher = SigmaMatcher.from_yaml(content)
|
|
|
|
def matched(self, kind: str, value: str, decoded: bool = False) -> bool:
|
|
hits = self.matcher.search_with_details(
|
|
{"raw": "opaque" if decoded else value}, [value] if decoded else []
|
|
)
|
|
return any(hit["name"] == "public_reference_" + kind for hit in hits)
|
|
|
|
def test_catalog_positive_raw_and_decoded_per_group(self) -> None:
|
|
for row in self.matcher.indicators:
|
|
kind, value = str(row["type"]), str(row["value"])
|
|
if kind not in {"domain", "path", "url", "sha256"}:
|
|
continue
|
|
if value == "/var/python/bin/customsnmpd":
|
|
continue # Legitimate binary intentionally omitted from standalone selectors.
|
|
with self.subTest(kind=kind, value=value):
|
|
self.assertTrue(self.matched(kind, value))
|
|
self.assertTrue(self.matched(kind, value, decoded=True))
|
|
|
|
def test_domains_accept_subdomains_not_lookalike_suffixes(self) -> None:
|
|
domain = next(str(r["value"]) for r in self.matcher.indicators if r["type"] == "domain")
|
|
self.assertTrue(self.matched("domain", "sub." + domain.upper()))
|
|
for value in ("evil" + domain, domain + ".invalid", domain + "-suffix"):
|
|
self.assertFalse(self.matched("domain", value))
|
|
|
|
def test_path_boundaries_case_and_url_embedding(self) -> None:
|
|
value = "/tmp/lula"
|
|
self.assertTrue(self.matched("path", value))
|
|
self.assertTrue(self.matched("path", "https://example.invalid" + value))
|
|
for other in (value + ".backup", value + "-longer", "/other" + value, value.upper()):
|
|
self.assertFalse(self.matched("path", other))
|
|
self.assertTrue(self.matched("path", "/directory/LoginIcon.sig"))
|
|
self.assertFalse(self.matched("path", "/directory/LoginIcon.sig.backup"))
|
|
self.assertFalse(self.matched("path", "/directory/loginicon.sig"))
|
|
|
|
def test_urls_preserve_path_case_and_reject_extension(self) -> None:
|
|
value = next(str(r["value"]) for r in self.matcher.indicators if r["type"] == "url")
|
|
parsed = urlsplit(value)
|
|
authority = parsed.scheme + "://" + parsed.netloc
|
|
self.assertTrue(self.matched("url", authority.upper() + value[len(authority) :]))
|
|
for suffix in ("-longer", ".backup", "/extra"):
|
|
self.assertFalse(self.matched("url", value + suffix))
|
|
if any(char.isalpha() for char in parsed.path):
|
|
self.assertFalse(self.matched("url", authority + value[len(authority) :].upper()))
|
|
|
|
def test_hashes_require_token_boundaries(self) -> None:
|
|
value = next(str(r["value"]) for r in self.matcher.indicators if r["type"] == "sha256")
|
|
self.assertTrue(self.matched("sha256", value.upper()))
|
|
self.assertFalse(self.matched("sha256", "f" + value))
|
|
self.assertFalse(self.matched("sha256", value + "z"))
|
|
|
|
|
|
class PublicLongLineTests(unittest.TestCase):
|
|
def test_encoded_loader_bounds_long_negative_and_positive_lines(self) -> None:
|
|
content = files("netscaler_triage").joinpath("data", "public_sigma.yml").read_text()
|
|
matcher = SigmaMatcher.from_yaml(content)
|
|
loader = next(
|
|
rule for rule in matcher.rules if rule.metadata["name"] == "sygnia-encoded-loader"
|
|
)
|
|
long_encoded = "QUJD" * 200000
|
|
self.assertFalse(loader.predicate({"raw": long_encoded}))
|
|
self.assertTrue(loader.predicate({"raw": long_encoded + " NSPPE base64 httpaccess"}))
|
|
self.assertTrue(loader.predicate({"raw": "first line\nNSPPE base64 httpaccess"}))
|
|
self.assertFalse(loader.predicate({"raw": "NSPPE\nbase64 httpaccess"}))
|
|
|
|
def test_crash_lookahead_long_negatives_and_line_semantics(self) -> None:
|
|
content = files("netscaler_triage").joinpath("data", "public_sigma.yml").read_text()
|
|
matcher = SigmaMatcher.from_yaml(content)
|
|
crash = next(
|
|
rule
|
|
for rule in matcher.rules
|
|
if rule.metadata["name"] == "elastic-inspired-crash-shell"
|
|
)
|
|
for text in ("QUJD" * 200000, "pitboss; " + "packet " * 100000):
|
|
self.assertFalse(crash.predicate({"raw": text}))
|
|
self.assertTrue(crash.predicate({"raw": "QUJD" * 200000 + " pitboss NSPPE;"}))
|
|
self.assertTrue(crash.predicate({"raw": "first\npitboss packet engine;"}))
|
|
self.assertFalse(crash.predicate({"raw": "pitboss packet\nengine;"}))
|
|
self.assertFalse(crash.predicate({"raw": "pitboss engine packet;"}))
|