NetScaler_forensik/tests/test_engine.py

import hashlib
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from netscaler_triage.engine import scan
from netscaler_triage.matching import Matcher


class EngineTests(unittest.TestCase):
    def test_hash_iocs_match_file_and_decoded_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            logs = root / "logs"
            logs.mkdir()
            content = b"original file\n"
            (logs / "sample.log").write_bytes(content)
            (logs / "http.log").write_text(
                "192.0.2.1 - - [01/Oct/2026:12:00:00 +0000] "
                '"GET /?x=aWQ%3D HTTP/1.1" 200 0 "-" "ua"\n',
                encoding="utf-8",
            )
            config = {
                "indicators": [
                    {"type": "sha256", "value": hashlib.sha256(value).hexdigest()}
                    for value in (content, b"id")
                ]
            }
            result = scan([("ADC", logs)], root / "out", [config], defaults=False)
            self.assertEqual(result["counts"]["file_hash_hits"], 1)
            file_hits = json.loads((root / "out/file_hash_hits.json").read_text())
            self.assertEqual(file_hits[0]["source"], "sample.log")
            events = [
                json.loads(line) for line in (root / "out/findings.jsonl").read_text().splitlines()
            ]
            self.assertTrue(
                any(
                    hit.get("basis") == "decoded_bytes_sha256"
                    for event in events
                    for hit in event["indicator_matches"]
                )
            )

    def test_interrupt_records_failure_and_keeps_database(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            logs = root / "logs"
            logs.mkdir()
            (logs / "test.log").write_text("one\ntwo\n", encoding="utf-8")
            with (
                patch.object(
                    Matcher,
                    "search_with_details",
                    side_effect=[(["hit"], [], []), KeyboardInterrupt()],
                ),
                self.assertRaises(KeyboardInterrupt),
            ):
                scan([("ADC", logs)], root / "out", [], defaults=False)
            summary = json.loads((root / "out/summary.json").read_text())
            self.assertEqual(summary["error_type"], "KeyboardInterrupt")
            with sqlite3.connect(root / "out/evidence.sqlite") as db:
                self.assertEqual(db.execute("SELECT count(*) FROM events").fetchone()[0], 1)
            db.close()

    def test_match_timeout_preserves_record_and_continues(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            logs = root / "logs"
            logs.mkdir()
            (logs / "test.log").write_text("problem\nfollowing\n", encoding="utf-8")
            with patch.object(
                Matcher, "search_with_details", side_effect=[TimeoutError(), (["hit"], [], [])]
            ):
                result = scan([("ADC", logs)], root / "out", [], defaults=False)
            self.assertEqual(result["status"], "partial")
            self.assertEqual(result["counts"]["records"], 2)
            self.assertEqual(result["counts"]["selected"], 2)
            self.assertEqual(result["counts"]["matching_errors"], 1)
            rows = [
                json.loads(line) for line in (root / "out/findings.jsonl").read_text().splitlines()
            ]
            self.assertEqual(rows[0]["raw"], "problem")
            self.assertTrue(rows[0]["analysis_errors"])
            self.assertEqual(rows[1]["rules"], ["hit"])

    def test_large_decoding_candidates_are_not_cached(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            logs = root / "logs"
            logs.mkdir()
            (logs / "test.log").write_text(("%41" * 2000 + "\n") * 2, encoding="utf-8")
            with patch("netscaler_triage.engine.decode_layers", return_value=[]) as decode:
                scan([("ADC", logs)], root / "out", [], defaults=False)
            self.assertEqual(decode.call_count, 2)

    def test_multi_appliance_and_decoded_match(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            inputs = []
            for name in ("one", "two"):
                directory = root / name
                directory.mkdir()
                (directory / "http.log").write_text(
                    "192.0.2.1 - - [01/Oct/2026:12:00:00 +0000] [42] "
                    '"GET /test?x=aWQ%3D HTTP/1.1" 403 0 "-" "ua"\n',
                    encoding="utf-8",
                )
                inputs.append((name, directory))
            result = scan(
                inputs,
                root / "out",
                [{"patterns": [{"id": "id", "pattern": "^id$"}]}],
                defaults=False,
            )
            self.assertEqual(result["counts"]["findings"], 2)
            db = sqlite3.connect(root / "out/evidence.sqlite")
            self.assertEqual(db.execute("SELECT count(*) FROM events").fetchone()[0], 2)
            event = json.loads(db.execute("SELECT data FROM events LIMIT 1").fetchone()[0])
            self.assertTrue(any(step.get("output") == "id" for step in event["decoding"]))
            db.close()
            with self.assertRaises(ValueError):
                scan(inputs, root / "one/out", [])
            with self.assertRaises(ValueError):
                scan(inputs, root / "out", [])

    def test_literal_ip_boundaries_and_timeout(self) -> None:
        matcher = Matcher(
            [{"indicators": [{"type": "ipv4", "value": "192.0.2.1"}]}], defaults=False
        )
        self.assertFalse(matcher.search("192.0.2.11")[1])
        self.assertTrue(matcher.search("source 192.0.2.1 end")[1])
        pathological = Matcher(
            [{"patterns": [{"pattern": "(a+)+$"}]}], timeout=0.001, defaults=False
        )
        with self.assertRaises(TimeoutError):
            pathological.search("a" * 10000 + "!")


if __name__ == "__main__":
    unittest.main()