Initial release: offline NetScaler triage with Sigma rules and MIT license

tests/test_easy_cli.py

@@ -0,0 +1,214 @@
+"""User-facing automatic preparation with synthetic logs and disposable folders."""
+
+import hashlib
+import io
+import json
+import os
+import tempfile
+import unittest
+import zipfile
+from contextlib import redirect_stderr, redirect_stdout
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import py7zr
+
+from netscaler_triage.cli import main
+from netscaler_triage.preparation import prepare_inputs
+
+LOG = (
+    '192.0.2.10 - - [01/Oct/2026:12:00:00 +0000] "GET /etc/passwd HTTP/1.1" 403 0 "-" "ordinary"\n'
+)
+
+
+class EasyCliTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name).resolve()
+        self.previous = Path.cwd()
+        self.case = self.root / "case"
+        self.case.mkdir()
+        os.chdir(self.case)
+
+    def tearDown(self) -> None:
+        os.chdir(self.previous)
+        self.temporary.cleanup()
+
+    def invoke(self, arguments: list[str]) -> str:
+        stdout, stderr = io.StringIO(), io.StringIO()
+        with redirect_stdout(stdout), redirect_stderr(stderr):
+            result = main(arguments)
+        self.assertEqual(result, 0, stdout.getvalue() + stderr.getvalue())
+        return stdout.getvalue() + stderr.getvalue()
+
+    def events(self) -> list[dict[str, Any]]:
+        files = list(self.root.rglob("findings.jsonl"))
+        self.assertEqual(len(files), 1, files)
+        return [json.loads(line) for line in files[0].read_text().splitlines()]
+
+    def archive(self, name: str, log: str = LOG) -> Path:
+        archive = self.case / name
+        with zipfile.ZipFile(archive, "w") as handle:
+            handle.writestr("var/log/httpaccess.log", log)
+        return archive
+
+    def encrypted_archive(self, password: str) -> Path:
+        original = self.root / "fixture.log"
+        original.write_text(LOG, encoding="utf-8")
+        archive = self.case / "encrypted.7z"
+        with py7zr.SevenZipFile(archive, "w", password=password, header_encryption=True) as handle:
+            handle.write(original, "var/log/httpaccess.log")
+        return archive
+
+    def assert_password_absent(self, password: str, output: str) -> None:
+        self.assertNotIn(password, output)
+        for path in self.root.rglob("*"):
+            if path.is_file() and path.suffix in {".json", ".jsonl", ".txt", ".csv"}:
+                self.assertNotIn(password, path.read_text(encoding="utf-8"), str(path))
+
+    def test_no_arguments_uses_current_folder_and_skips_tool_directories(self) -> None:
+        (self.case / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        (self.case / "pyproject.toml").write_text(
+            '[project]\nname = "netscaler-triage"\n', encoding="utf-8"
+        )
+        for name in ("src", ".runtime", ".venv", "results"):
+            folder = self.case / name
+            folder.mkdir()
+            (folder / "unrelated.log").write_text(LOG * 3, encoding="utf-8")
+        self.invoke([])
+        events = self.events()
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["kind"], "http_access")
+
+    def test_folder_without_subcommand(self) -> None:
+        folder = self.case / "logs"
+        folder.mkdir()
+        (folder / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        self.invoke([str(folder)])
+        self.assertEqual(len(self.events()), 1)
+
+    def test_explicit_scan_folder(self) -> None:
+        folder = self.case / "logs"
+        folder.mkdir()
+        (folder / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        self.invoke(["scan", str(folder)])
+        self.assertEqual(len(self.events()), 1)
+
+    def test_missing_common_drive_uses_current_output_parent(self) -> None:
+        folder = self.case / "logs"
+        folder.mkdir()
+        (folder / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        with patch("netscaler_triage.cli.os.path.commonpath", side_effect=ValueError("drives")):
+            self.invoke([str(folder)])
+        self.assertEqual(len(self.events()), 1)
+
+    def test_advanced_preparation_limits_are_forwarded(self) -> None:
+        folder = self.case / "logs"
+        folder.mkdir()
+        (folder / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        with patch("netscaler_triage.cli.prepare_inputs", wraps=prepare_inputs) as prepare:
+            self.invoke(
+                [str(folder), "--max-bytes", "10000", "--max-files", "9", "--max-depth", "0"]
+            )
+        self.assertEqual(prepare.call_args.kwargs["max_bytes"], 10000)
+        self.assertEqual(prepare.call_args.kwargs["max_files"], 9)
+        self.assertEqual(prepare.call_args.kwargs["max_depth"], 0)
+        self.assertEqual(len(self.events()), 1)
+
+    def test_archive_is_prepared_automatically_without_changing_original(self) -> None:
+        archive = self.archive("adc.zip")
+        before = hashlib.sha256(archive.read_bytes()).digest()
+        self.invoke([str(archive)])
+        self.assertEqual(hashlib.sha256(archive.read_bytes()).digest(), before)
+        events = self.events()
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["kind"], "http_access")
+
+    def test_two_archives_keep_separate_appliance_identity(self) -> None:
+        first = self.archive("adc-one.zip")
+        second = self.archive("adc-two.zip")
+        self.invoke([str(first), str(second)])
+        events = self.events()
+        self.assertEqual(len(events), 2)
+        self.assertEqual(len({event["appliance"] for event in events}), 2)
+
+    def test_current_folder_discovers_multiple_archives(self) -> None:
+        self.archive("adc-one.zip")
+        self.archive("adc-two.zip")
+        self.invoke([])
+        events = self.events()
+        self.assertEqual(len(events), 2)
+        self.assertEqual(len({event["appliance"] for event in events}), 2)
+
+    def test_nested_archive_is_not_counted_twice(self) -> None:
+        nested = io.BytesIO()
+        with zipfile.ZipFile(nested, "w") as handle:
+            handle.writestr("httpaccess.log", LOG)
+        archive = self.case / "outer.zip"
+        with zipfile.ZipFile(archive, "w") as handle:
+            handle.writestr("inner.zip", nested.getvalue())
+        self.invoke([str(archive)])
+        self.assertEqual(len(self.events()), 1)
+
+    def test_archive_password_is_not_written_to_output(self) -> None:
+        password = "SYNTHETIC-password-SECRET-39217"
+        archive = self.encrypted_archive(password)
+        output = self.invoke([str(archive), "--password", password])
+        self.assertEqual(len(self.events()), 1)
+        self.assert_password_absent(password, output)
+
+    def test_password_flag_without_value_prompts_privately(self) -> None:
+        password = "SYNTHETIC-prompt-SECRET-39218"
+        archive = self.encrypted_archive(password)
+        with patch("netscaler_triage.cli.getpass.getpass", return_value=password) as prompt:
+            output = self.invoke([str(archive), "--password"])
+        prompt.assert_called_once()
+        self.assertEqual(len(self.events()), 1)
+        self.assert_password_absent(password, output)
+
+    def test_password_option_alone_uses_current_folder(self) -> None:
+        password = "SYNTHETIC-default-SECRET-39219"
+        self.encrypted_archive(password)
+        output = self.invoke(["--password", password])
+        self.assertEqual(len(self.events()), 1)
+        self.assert_password_absent(password, output)
+
+    def test_output_inside_original_input_is_rejected_before_changes(self) -> None:
+        (self.case / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        before = {
+            p.relative_to(self.root).as_posix(): p.read_bytes() if p.is_file() else None
+            for p in self.root.rglob("*")
+        }
+        stdout, stderr = io.StringIO(), io.StringIO()
+        with redirect_stdout(stdout), redirect_stderr(stderr):
+            result = main([str(self.case), "--output", str(self.case / "generated" / "out")])
+        self.assertNotEqual(result, 0)
+        after = {
+            p.relative_to(self.root).as_posix(): p.read_bytes() if p.is_file() else None
+            for p in self.root.rglob("*")
+        }
+        self.assertEqual(before, after)
+
+    def test_corrupt_archive_keeps_valid_results_and_reports_partial(self) -> None:
+        (self.case / "httpaccess.log").write_text(LOG, encoding="utf-8")
+        (self.case / "broken.zip").write_bytes(b"not a zip archive")
+        stdout, stderr = io.StringIO(), io.StringIO()
+        with redirect_stdout(stdout), redirect_stderr(stderr):
+            result = main([])
+        self.assertEqual(result, 2)
+        self.assertEqual(len(self.events()), 1)
+        summaries = list(self.root.rglob("summary.json"))
+        self.assertEqual(len(summaries), 1)
+        summary = json.loads(summaries[0].read_text())
+        self.assertEqual(summary["status"], "partial")
+        self.assertEqual(summary["counts"]["preparation_errors"], 1)
+        self.assertEqual(summary["preparation_errors"][0]["source"], "broken.zip")
+        manifest = json.loads((summaries[0].parent / "preparation_manifest.json").read_text())
+        failures = [entry for entry in manifest["entries"] if entry["action"] == "error"]
+        self.assertEqual(len(failures), 1)
+        self.assertIn("Preparation incomplete", stderr.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()