215 lines
9.1 KiB
Python
215 lines
9.1 KiB
Python
"""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()
|