106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from netscaler_triage.engine import scan
|
|
from netscaler_triage.matching import Matcher
|
|
|
|
|
|
class RuleDetailsTests(unittest.TestCase):
|
|
def test_shared_rule_keeps_both_source_descriptions_without_duplicate_regex(self) -> None:
|
|
matcher = Matcher(
|
|
[
|
|
{
|
|
"patterns": [
|
|
{
|
|
"id": "shared",
|
|
"pattern": "signal",
|
|
"description": "Source one",
|
|
"source": "REF-1",
|
|
},
|
|
{
|
|
"id": "shared",
|
|
"pattern": "signal",
|
|
"description": "Source two",
|
|
"source": "REF-2",
|
|
},
|
|
]
|
|
}
|
|
],
|
|
defaults=False,
|
|
)
|
|
self.assertEqual(len(matcher.rules), 1)
|
|
_, _, details = matcher.search_with_details("signal")
|
|
self.assertEqual({row["source"] for row in details}, {"REF-1", "REF-2"})
|
|
|
|
def test_default_descriptions_and_compatible_search(self) -> None:
|
|
matcher = Matcher([], defaults=True)
|
|
ids, hits = matcher.search("payload ;hostname")
|
|
detailed_ids, detailed_hits, details = matcher.search_with_details("payload ;hostname")
|
|
self.assertEqual((ids, hits), (detailed_ids, detailed_hits))
|
|
self.assertTrue(
|
|
any(row.get("name") == "command_injection" and row["description"] for row in details)
|
|
)
|
|
|
|
def test_same_id_does_not_attribute_unmatched_description(self) -> None:
|
|
matcher = Matcher(
|
|
[
|
|
{
|
|
"patterns": [
|
|
{
|
|
"id": "custom",
|
|
"pattern": "MATCH_ME",
|
|
"description": "Matched explanation",
|
|
"source": "https://example.org/source",
|
|
},
|
|
{
|
|
"id": "custom",
|
|
"pattern": "NEVER_MATCH",
|
|
"description": "Must not be attributed",
|
|
"source": "https://example.org/other",
|
|
},
|
|
]
|
|
}
|
|
],
|
|
defaults=False,
|
|
)
|
|
ids, _, details = matcher.search_with_details("MATCH_ME")
|
|
self.assertEqual(ids, ["custom"])
|
|
self.assertEqual(len(details), 1)
|
|
self.assertEqual(details[0]["description"], "Matched explanation")
|
|
self.assertEqual(details[0]["source"], "https://example.org/source")
|
|
|
|
def test_decoded_rule_metadata_reaches_findings(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
logs = root / "logs"
|
|
logs.mkdir()
|
|
(logs / "http.log").write_text(
|
|
'192.0.2.1 - - [01/Oct/2026:12:00:00 +0000] "GET /?x=aWQ%3D HTTP/1.1" '
|
|
'404 0 "-" "ua"\n'
|
|
)
|
|
config = {
|
|
"patterns": [
|
|
{
|
|
"id": "decoded-id",
|
|
"pattern": "^id$",
|
|
"description": "The decoded value is the id command; execution is not "
|
|
"established.",
|
|
"source_id": "REF-1",
|
|
"source": "https://example.org/advisory",
|
|
}
|
|
]
|
|
}
|
|
scan([("ADC", logs)], root / "out", [config], defaults=False)
|
|
events = [
|
|
json.loads(line) for line in (root / "out/findings.jsonl").read_text().splitlines()
|
|
]
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["rules"], ["decoded-id"])
|
|
self.assertEqual(events[0]["rule_details"], config["patterns"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|