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"])