NetScaler_forensik/tests/test_ingest.py

import gzip
import io
import tempfile
import unittest
import zipfile
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import IO, Any, cast
from unittest.mock import patch

from netscaler_triage.ingest import InputError, extract_archive, inventory, iter_records


class IngestTests(unittest.TestCase):
    def test_unreadable_file_does_not_hide_readable_evidence(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / "bad").write_text("private")
            (root / "good").write_text("evidence\n")
            original = Path.open

            def guarded(path: Path, *args: Any, **kwargs: Any) -> IO[Any]:
                if path == root / "bad":
                    raise PermissionError(13, "denied", str(path))
                return cast(IO[Any], original(path, *args, **kwargs))

            with patch.object(Path, "open", guarded):
                listing = inventory(root)
                records = list(iter_records(root, "a"))
            self.assertEqual(listing[0]["error"], "PermissionError")
            self.assertIsNone(listing[0]["sha256"])
            self.assertEqual(records[0]["error"], "PermissionError")
            self.assertEqual(records[1]["raw"], "evidence")

    def test_directory_traversal_errors_are_visible(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()

            def failed_walk(
                path: Path, *, followlinks: bool, onerror: Callable[[OSError], None]
            ) -> Iterator[tuple[str, list[str], list[str]]]:
                onerror(PermissionError(13, "denied", str(root / "unreadable")))
                return iter(())

            with patch("netscaler_triage.ingest.os.walk", failed_walk):
                rows = inventory(root)
                records = list(iter_records(root, "a"))
            self.assertEqual(rows[0]["source"], "unreadable")
            self.assertEqual(records[0]["error"], "PermissionError")

    def test_windows_reserved_archive_names(self) -> None:
        from netscaler_triage.ingest import _member

        for name in ("CON", "logs/NUL.txt", "COM1.log", "logs/name.", "logs/name "):
            with self.assertRaises(InputError):
                _member(name)

    def test_binary_is_explicitly_incomplete(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / "binary").write_bytes(b"hello\x00binary")
            self.assertIn("error", next(iter_records(root, "a")))

    def test_gzip_magic_and_limits(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            (root / "messages.1").write_bytes(gzip.compress(b"hello\nworld\n"))
            before = inventory(root)
            rows = list(iter_records(root, "a"))
            self.assertEqual([r["raw"] for r in rows], ["hello", "world"])
            self.assertEqual(rows[1]["line"], 2)
            self.assertIn("error", next(iter_records(root, "a", max_line_bytes=3)))
            self.assertEqual(before, inventory(root))

    def test_extract_and_fail_closed(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            src = root / "a.zip"
            with zipfile.ZipFile(src, "w") as z:
                z.writestr("logs/a.gz", gzip.compress(b"abc"))
            manifest = extract_archive(src, root / "good")
            self.assertEqual(manifest[0]["source"], "logs/a.gz")
            with self.assertRaises(FileExistsError):
                extract_archive(src, root / "good")
            with self.assertRaises(InputError):
                extract_archive(src, root / "limit", max_bytes=2)
            self.assertFalse((root / "limit").exists())
            with zipfile.ZipFile(src, "w") as z:
                z.writestr("../escape", "bad")
            with self.assertRaises(InputError):
                extract_archive(src, root / "bad")
            self.assertFalse((root / "bad").exists())
            self.assertFalse((root / "escape").exists())

    def test_nested_quota_and_symlink(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            nested = io.BytesIO()
            with zipfile.ZipFile(nested, "w") as z:
                z.writestr("a", "contents")
            src = root / "a.zip"
            with zipfile.ZipFile(src, "w") as z:
                z.writestr("nested.zip", nested.getvalue())
            with self.assertRaises(InputError):
                extract_archive(src, root / "bad", max_depth=0)
            info = zipfile.ZipInfo("link")
            info.external_attr = 0o120777 << 16
            with zipfile.ZipFile(src, "w") as z:
                z.writestr(info, "/tmp/escape")
            with self.assertRaises(InputError):
                extract_archive(src, root / "bad")

    def test_seven_zip(self) -> None:
        import py7zr

        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp).resolve()
            src = root / "a.7z"
            with py7zr.SevenZipFile(src, "w") as z:
                z.writestr(b"hello\n", "logs/messages")
            result = extract_archive(src, root / "good")
            self.assertEqual(result[0]["source"], "logs/messages")
            self.assertEqual((root / "good/logs/messages").read_bytes(), b"hello\n")
            with self.assertRaises(InputError):
                extract_archive(src, root / "limit", max_bytes=2)


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