88 lines
3.7 KiB
Python
88 lines
3.7 KiB
Python
|
import unittest
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from netscaler_triage.parsing import parse_record
|
||
|
|
|
||
|
|
|
||
|
|
class ParsingTests(unittest.TestCase):
|
||
|
|
def test_native_correlation_fields(self) -> None:
|
||
|
|
row = self.parse(
|
||
|
|
'Sep 28 09:10:11 host bash[1]: return_code="0" trans id 12345 Context [email protected]'
|
||
|
|
)
|
||
|
|
self.assertEqual(row["fields"]["return_code"], "0")
|
||
|
|
self.assertEqual(row["fields"]["transaction_id"], "12345")
|
||
|
|
self.assertEqual(row["fields"]["context"], "[email protected]")
|
||
|
|
row = self.parse('Sep 28 09:10:11 host CMD_EXECUTED 12 - Status "Success"')
|
||
|
|
self.assertEqual(row["fields"]["command_status"], "Success")
|
||
|
|
|
||
|
|
def test_attacker_username_is_not_update(self) -> None:
|
||
|
|
row = self.parse('Sep 28 09:10:11 host aaad[42]: User "installns upgrade rebooting" failed')
|
||
|
|
self.assertEqual(row["categories"], [])
|
||
|
|
|
||
|
|
def test_proxy_source(self) -> None:
|
||
|
|
row = self.parse(
|
||
|
|
'192.0.2.40 -> 198.51.100.10 - - [01/Oct/2026:12:02:28 +0000] "GET / HTTP/1.1" 200 5'
|
||
|
|
)
|
||
|
|
self.assertEqual(row["fields"]["dest"], "198.51.100.10")
|
||
|
|
|
||
|
|
def parse(self, raw: str, **kwargs: Any) -> dict[str, Any]:
|
||
|
|
return parse_record({"raw": raw}, **kwargs)
|
||
|
|
|
||
|
|
def test_access_payload_not_update(self) -> None:
|
||
|
|
row = self.parse(
|
||
|
|
'192.0.2.40 - - [01/Oct/2026:12:02:28 +0200] [23] "GET '
|
||
|
|
'/installns?date=2020-01-01T00:00:00Z HTTP/1.1" 404 23 "-" "x"'
|
||
|
|
)
|
||
|
|
self.assertEqual(row["timestamp"], "2026-10-01T10:02:28Z")
|
||
|
|
self.assertEqual(row["fields"]["uri"], "/installns?date=2020-01-01T00:00:00Z")
|
||
|
|
self.assertEqual(row["categories"], [])
|
||
|
|
|
||
|
|
def test_missing_year_and_gmt(self) -> None:
|
||
|
|
row = self.parse(
|
||
|
|
"Sep 30 16:00:54 <local0.warn> 198.51.100.10 09/30/2026:14:00:54 GMT ns "
|
||
|
|
"0-PPE-0 : message"
|
||
|
|
)
|
||
|
|
self.assertEqual(row["timestamp"], "2026-09-30T14:00:54Z")
|
||
|
|
self.assertIsNone(self.parse("Sep 30 16:00:54 host bash[1]: installns")["timestamp"])
|
||
|
|
row = self.parse(
|
||
|
|
"Sep 30 16:00:54 host bash[1]: installns", year=2026, timezone="Europe/Berlin"
|
||
|
|
)
|
||
|
|
self.assertEqual(row["timestamp"], "2026-09-30T14:00:54Z")
|
||
|
|
self.assertEqual(row["fields"]["update_evidence"], "installer_invocation")
|
||
|
|
|
||
|
|
def test_historical_explicit_year(self) -> None:
|
||
|
|
row = self.parse(
|
||
|
|
"COLLECTOR::INFOA::Sun Sep 28 09:11:12 2025:Successfully loaded "
|
||
|
|
"configuration(NS13.1-60.20)",
|
||
|
|
year=2026,
|
||
|
|
timezone="UTC",
|
||
|
|
)
|
||
|
|
self.assertEqual(row["timestamp"], "2025-09-28T09:11:12Z")
|
||
|
|
self.assertEqual(row["fields"]["versions"], ["13.1-60.20"])
|
||
|
|
self.assertIn("version", row["categories"])
|
||
|
|
|
||
|
|
def test_dst_ambiguity(self) -> None:
|
||
|
|
self.assertIsNone(
|
||
|
|
self.parse("2026-10-25 02:30:00 message", timezone="Europe/Berlin")["timestamp"]
|
||
|
|
)
|
||
|
|
self.assertIsNone(
|
||
|
|
self.parse("2026-03-29 02:30:00 message", timezone="Europe/Berlin")["timestamp"]
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_nssetup_version_transition(self) -> None:
|
||
|
|
row = self.parse(
|
||
|
|
"Sep 28 09:10:11 host nssetup[386]: ns_conf_check(): This is an upgrade "
|
||
|
|
"from NetScaler NS13.1 build 50.10 to build 60.20",
|
||
|
|
year=2026,
|
||
|
|
timezone="UTC",
|
||
|
|
)
|
||
|
|
self.assertEqual(row["fields"]["builds"], ["50.10", "60.20"])
|
||
|
|
self.assertEqual(row["fields"]["update_relation"], "explicit_upgrade_from_to")
|
||
|
|
row = self.parse(
|
||
|
|
"Sep 28 09:10:11 host nssetup[386]: This is NetScaler version NS13.1, build 60.20"
|
||
|
|
)
|
||
|
|
self.assertEqual(row["fields"]["versions"], ["13.1-60.20"])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|