285 lines
13 KiB
Python
285 lines
13 KiB
Python
import gzip
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from netscaler_triage.filesystem import BLOCK, read_selected_members, scan_filesystem
|
|
|
|
|
|
class FilesystemTests(unittest.TestCase):
|
|
def make_archive(self, path: Path, entries: dict[str, bytes]) -> None:
|
|
with tarfile.open(path, "w:gz") as archive:
|
|
for name, data in entries.items():
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(data)
|
|
info.mode = 0o6555 if name == "bin/sh" else 0o644
|
|
archive.addfile(info, io.BytesIO(data))
|
|
link = tarfile.TarInfo("escape-link")
|
|
link.type, link.linkname = tarfile.SYMTYPE, "/etc/passwd"
|
|
archive.addfile(link)
|
|
|
|
def test_streaming_evidence_hashes_links_gzip_and_baseline(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "snapshot.tgz"
|
|
inert = b"<?php /* HTTP_X_UX $_POST shell_exec( inert */"
|
|
gzip_data = gzip.compress(b"nsmon.pl\n") + gzip.compress(b"NSC_TASS\n")
|
|
data = {
|
|
"./var/netscaler/hidden.sig": inert,
|
|
"bin/sh": b"inert",
|
|
"var/log/ns.log.0.gz": gzip_data,
|
|
"../outside": b"inert",
|
|
"etc/crontab": b"# nsmon.pl fixture\n",
|
|
"var/tmp/boundary": b"\n" * (BLOCK - 4) + b"nsmon.pl\n",
|
|
}
|
|
self.make_archive(source, data)
|
|
baseline = root / "baseline"
|
|
baseline.mkdir()
|
|
(baseline / "etc").mkdir()
|
|
(baseline / "etc/crontab").write_text("# earlier\n")
|
|
before = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
result = scan_filesystem(source, root / "result", baseline=baseline)
|
|
self.assertTrue(result["complete_outer_archive"])
|
|
self.assertTrue(result["source_unchanged"])
|
|
self.assertEqual(result["archive_sha256"], before)
|
|
self.assertEqual(result["issues"], [])
|
|
self.assertEqual(result["counts"]["baseline_changed"], 1)
|
|
findings = [
|
|
json.loads(line)
|
|
for line in (root / "result/findings.jsonl").read_text().splitlines()
|
|
]
|
|
rules = {x["rule"] for x in findings}
|
|
self.assertTrue(
|
|
{
|
|
"suid_sgid",
|
|
"unsafe_archive_path",
|
|
"php_in_disguised_file",
|
|
"php_input_and_execution",
|
|
}
|
|
<= rules
|
|
)
|
|
self.assertTrue(
|
|
any(x.get("view") == "gzip" and x.get("marker") == "nsc_tass" for x in findings)
|
|
)
|
|
cross = [
|
|
x
|
|
for x in findings
|
|
if x["path"] == "var/tmp/boundary" and x["rule"] == "campaign_marker"
|
|
]
|
|
self.assertEqual(len(cross), 1)
|
|
self.assertEqual(cross[0]["offset"], BLOCK - 4)
|
|
self.assertFalse((root / "outside").exists())
|
|
self.assertFalse((root / "result/escape-link").exists())
|
|
|
|
def test_corrupt_outer_gzip_is_not_reported_complete(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "input.tgz"
|
|
self.make_archive(source, {"a": b"inert"})
|
|
data = bytearray(source.read_bytes())
|
|
data[-8] ^= 255
|
|
source.write_bytes(data)
|
|
result = scan_filesystem(source, root / "result")
|
|
self.assertFalse(result["complete_outer_archive"])
|
|
self.assertTrue(result["issues"])
|
|
|
|
def test_nested_gzip_error_recorded(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "input.tgz"
|
|
self.make_archive(source, {"bad.gz": gzip.compress(b"inert")[:-5]})
|
|
result = scan_filesystem(source, root / "result")
|
|
self.assertTrue(result["complete_outer_archive"])
|
|
self.assertEqual(result["issues"][0]["error"], "truncated_nested_gzip")
|
|
|
|
def test_byte_limit_and_existing_output(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "input.tgz"
|
|
self.make_archive(source, {"a": b"x" * 100})
|
|
result = scan_filesystem(source, root / "result", max_expanded_bytes=10)
|
|
self.assertFalse(result["complete_outer_archive"])
|
|
with self.assertRaises(ValueError):
|
|
scan_filesystem(source, root / "result")
|
|
|
|
def test_targeted_read_is_bounded_and_does_not_follow_links(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "input.tgz"
|
|
self.make_archive(source, {"./etc/test": b"inert evidence"})
|
|
rows = read_selected_members(source, {"etc/test", "escape-link"})
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual(rows[0][1], b"inert evidence")
|
|
self.assertEqual(rows[1][1], b"")
|
|
self.assertEqual(rows[1][0]["linkname"], "/etc/passwd")
|
|
with self.assertRaises(ValueError):
|
|
read_selected_members(source, {"etc/test"}, max_total_bytes=2)
|
|
|
|
def test_cli_streaming_route(self) -> None:
|
|
from contextlib import redirect_stdout
|
|
|
|
from netscaler_triage.cli import main
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "input.tgz"
|
|
self.make_archive(source, {"a": b"inert"})
|
|
with redirect_stdout(io.StringIO()):
|
|
status = main(["filesystem", str(source), "--output", str(root / "result")])
|
|
self.assertEqual(status, 0)
|
|
self.assertTrue((root / "result" / "summary.json").is_file())
|
|
|
|
def test_log_review_cleans_temporary_copies_on_error(self) -> None:
|
|
from unittest.mock import patch
|
|
|
|
from netscaler_triage.filesystem import scan_snapshot_logs
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "input.tgz"
|
|
self.make_archive(source, {"var/log/ns.log": b"inert log\n"})
|
|
scan_filesystem(source, root / "result")
|
|
seen: list[Path] = []
|
|
|
|
def fail(*args: object, **kwargs: object) -> dict[str, object]:
|
|
inputs = args[0]
|
|
assert isinstance(inputs, list)
|
|
temporary = inputs[0][1]
|
|
assert isinstance(temporary, Path)
|
|
seen.append(temporary)
|
|
self.assertEqual((temporary / "var/log/ns.log").read_bytes(), b"inert log\n")
|
|
raise ValueError("simulated engine failure")
|
|
|
|
with (
|
|
patch("netscaler_triage.engine.scan", side_effect=fail),
|
|
self.assertRaises(ValueError),
|
|
):
|
|
scan_snapshot_logs(source, root / "result")
|
|
self.assertEqual(len(seen), 1)
|
|
self.assertFalse(seen[0].exists())
|
|
|
|
def test_directory_matches_archive_and_never_follows_links(self) -> None:
|
|
import os
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "export"
|
|
(source / "var/tmp").mkdir(parents=True)
|
|
sample = b"HTTP_X_UX nsmon.pl 149.104.78.141\n"
|
|
(source / "var/tmp/sample").write_bytes(sample)
|
|
(source / "outside").symlink_to(root / "private")
|
|
(root / "private").write_text("secret never read")
|
|
os.mkfifo(source / "pipe")
|
|
result = scan_filesystem(source, root / "results")
|
|
self.assertTrue(result["complete_scan"])
|
|
self.assertIsNone(result["source_unchanged"])
|
|
self.assertIsNone(result["archive_sha256"])
|
|
self.assertEqual(result["counts"]["regular_files"], 1)
|
|
self.assertEqual(result["issues"], [])
|
|
rows = [
|
|
json.loads(line)
|
|
for line in (root / "results/inventory.jsonl").read_text().splitlines()
|
|
]
|
|
self.assertEqual(next(r for r in rows if r["path"] == "outside")["type"], "2")
|
|
findings = [
|
|
json.loads(line)
|
|
for line in (root / "results/findings.jsonl").read_text().splitlines()
|
|
]
|
|
self.assertTrue(any(r["rule"] == "known_network_indicator" for r in findings))
|
|
selected = read_selected_members(source, {"var/tmp/sample", "outside"})
|
|
self.assertEqual(
|
|
next(data for meta, data in selected if meta["path"] == "var/tmp/sample"), sample
|
|
)
|
|
self.assertEqual(
|
|
next(data for meta, data in selected if meta["path"] == "outside"), b""
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
scan_filesystem(source, source / "results")
|
|
|
|
def test_network_indicator_token_boundaries(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "export"
|
|
source.mkdir()
|
|
(source / "a").write_text("149.104.78.1419 1149.104.78.141\n149.104.78.141:443\n")
|
|
scan_filesystem(source, root / "results")
|
|
hits = [
|
|
json.loads(line)
|
|
for line in (root / "results/findings.jsonl").read_text().splitlines()
|
|
if json.loads(line)["rule"] == "known_network_indicator"
|
|
]
|
|
self.assertEqual(len(hits), 1)
|
|
self.assertEqual(hits[0]["line"], 2)
|
|
|
|
def test_published_persistence_rules_and_config_paths(self) -> None:
|
|
from netscaler_triage.filesystem import critical
|
|
|
|
samples = {
|
|
"php_nonstandard_handler": "AddType application/x-httpd-php .rpm",
|
|
"php_engine_enabled": "php_flag engine on",
|
|
"web_alias_routing": "AliasMatch /vpn/media/example.ico /vendor/example.sig",
|
|
"css_webshell_alias": "LogonUISimple.html.style.min.ab12.css",
|
|
"nsmon_cron": "*/5 * * * * root perl /var/tmp/.nsmon/nsmon.pl",
|
|
"privileged_monitor_account": "bind system user sec_monitor superuser",
|
|
"command_channel_header": "X-UX-12",
|
|
"cron_scrubbing_reference": "# sed /vpn/scripts/linux /etc/crontab",
|
|
"shell_privilege_change": "# chmod 6555 /bin/sh",
|
|
"reverse_shell_reference": "# /dev/tcp/",
|
|
"configuration_staging_reference": (
|
|
"# cat /flash/nsconfig/ns.conf > /var/netscaler/example"
|
|
),
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "export"
|
|
source.mkdir()
|
|
for rule, sample in samples.items():
|
|
(source / rule).write_text(sample + "\n")
|
|
scan_filesystem(source, root / "results")
|
|
hits = [
|
|
json.loads(line)
|
|
for line in (root / "results/findings.jsonl").read_text().splitlines()
|
|
]
|
|
for rule in samples:
|
|
self.assertTrue(any(h["path"] == rule and h["rule"] == rule for h in hits), rule)
|
|
self.assertTrue(all(h["classification"] == "review_candidate" for h in hits))
|
|
for path in [
|
|
"nsconfig/httpd.conf",
|
|
"flash/nsconfig/httpd.conf",
|
|
"nsconfig/crontab",
|
|
"var/spool/cron/root",
|
|
"etc/cron.d/custom",
|
|
"etc/master.passwd",
|
|
]:
|
|
self.assertTrue(critical(path), path)
|
|
|
|
def test_case_ioc_csv_and_artifact_matrix(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
source = root / "export"
|
|
source.mkdir()
|
|
(source / "evidence").write_text("192.0.2.55 192.0.2.56\n")
|
|
iocs = root / "case.csv"
|
|
iocs.write_text(
|
|
"type,value,status\nip,192.0.2.55,observed_attempt\n"
|
|
"ip,192.0.2.56,own_test\npath,/missing-artifact,reference_only\n"
|
|
)
|
|
result = scan_filesystem(source, root / "results", ioc_file=iocs)
|
|
self.assertEqual(
|
|
result["custom_ioc_sha256"], hashlib.sha256(iocs.read_bytes()).hexdigest()
|
|
)
|
|
hits = [
|
|
json.loads(line)
|
|
for line in (root / "results/findings.jsonl").read_text().splitlines()
|
|
]
|
|
markers = {h.get("marker") for h in hits}
|
|
self.assertIn("192.0.2.55", markers)
|
|
self.assertNotIn("192.0.2.56", markers)
|
|
checks = json.loads((root / "results/artifact_checks.json").read_text())
|
|
self.assertTrue(checks["scan_complete"])
|
|
self.assertIn({"path": "missing-artifact", "present": False}, checks["paths"])
|