103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
import base64
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from netscaler_triage.decoding import has_encoding_candidate
|
|
from netscaler_triage.engine import scan
|
|
|
|
ARTIFACT = "/var/tmp/nsmon.pl"
|
|
RULE = """title: Synthetic decoded artifact
|
|
id: 96ddcb52-8e0b-4f8b-a41d-44e160db7464
|
|
logsource:
|
|
product: netscaler
|
|
detection:
|
|
selection:
|
|
raw|contains: '/var/tmp/nsmon.pl'
|
|
condition: selection
|
|
level: medium
|
|
"""
|
|
|
|
|
|
def encoded(value: str) -> str:
|
|
return base64.b64encode(value.encode()).decode().rstrip("=")
|
|
|
|
|
|
class EncodingGateTests(unittest.TestCase):
|
|
def test_encoded_user_agents_and_raw_blobs_keep_both_match_types(self) -> None:
|
|
once = encoded(ARTIFACT)
|
|
nested = encoded(once)
|
|
url_encoded = "%" + format(ord(once[0]), "02X") + once[1:]
|
|
lines = [
|
|
f'192.0.2.10 - - [01/Oct/2026:12:00:00 +0000] "GET / HTTP/1.1" 404 0 "-" "{agent}"'
|
|
for agent in (once, nested, url_encoded)
|
|
]
|
|
lines.extend([once, "marker:" + once, "log payload " + encoded(ARTIFACT * 3)])
|
|
lines.extend(
|
|
[
|
|
"192.0.2.10 - - [01/Oct/2026:12:00:00 +0000] "
|
|
f'"GET / HTTP/1.1" 404 0 "{once}" "synthetic-agent"',
|
|
"192.0.2.10 - - [01/Oct/2026:12:00:00 +0000] "
|
|
f'"GET / HTTP/1.1" 404 0 "-" "synthetic-agent" Cookie: command={once}',
|
|
]
|
|
)
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary).resolve()
|
|
logs = root / "logs"
|
|
logs.mkdir()
|
|
(logs / "messages").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
config: dict[str, Any] = {
|
|
"sigma_rules": [{"source": "synthetic.yml", "yaml": RULE}],
|
|
"indicators": [{"type": "literal", "value": ARTIFACT, "status": "reference_only"}],
|
|
}
|
|
summary = scan([("ADC", logs)], root / "out", [config], defaults=False)
|
|
self.assertEqual(summary["counts"]["records"], len(lines))
|
|
events = [
|
|
json.loads(row) for row in (root / "out/findings.jsonl").read_text().splitlines()
|
|
]
|
|
self.assertEqual(len(events), len(lines))
|
|
for event, original in zip(events, lines, strict=True):
|
|
self.assertEqual(event["raw"], original)
|
|
self.assertIn("96ddcb52-8e0b-4f8b-a41d-44e160db7464", event["rules"])
|
|
self.assertTrue(any(hit["value"] == ARTIFACT for hit in event["indicator_matches"]))
|
|
self.assertTrue(
|
|
any(ARTIFACT in (layer.get("output") or "") for layer in event["decoding"])
|
|
)
|
|
self.assertTrue(
|
|
any("decoded:" in str(detail["basis"]) for detail in event["rule_details"])
|
|
)
|
|
self.assertEqual(
|
|
sum(layer["transformation"] == "original" for layer in events[0]["decoding"]),
|
|
1,
|
|
"An encoded known field must not cause duplicate raw-line decoding",
|
|
)
|
|
cookie_layers = [
|
|
layer
|
|
for layer in events[-1]["decoding"]
|
|
if layer["transformation"] == "base64"
|
|
and layer["depth"] == 1
|
|
and layer["input"] == once
|
|
]
|
|
self.assertTrue(cookie_layers)
|
|
for layer in cookie_layers:
|
|
self.assertEqual(lines[-1][layer["start"] : layer["end"]], layer["input"])
|
|
|
|
def test_ordinary_words_and_timestamps_are_not_candidates(self) -> None:
|
|
for text in (
|
|
"NetScaler",
|
|
"Mozilla/5.0",
|
|
"ordinary words and numbers 123456789",
|
|
"Oct 01 12:00:00 appliance ready",
|
|
"2026-10-01T12:00:00Z",
|
|
):
|
|
with self.subTest(text=text):
|
|
self.assertFalse(has_encoding_candidate(text))
|
|
self.assertTrue(has_encoding_candidate(encoded("nsmon.pl")))
|
|
self.assertTrue(has_encoding_candidate(encoded(ARTIFACT * 2048)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|