130 lines
5.5 KiB
Python
130 lines
5.5 KiB
Python
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()
|