tests/test_flashlight.py

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
import hashlib
import json
from pathlib import Path
import subprocess
import tarfile
import tempfile
import unittest

import flashlight


class FlashlightContractTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.package_root = flashlight.prepare_runtime()

    def assert_error(self, document: dict, code: str) -> None:
        self.assertEqual(document["document_type"], flashlight.DOCUMENT_TYPE)
        self.assertEqual(document["status"], "error")
        self.assertIs(document["complete"], False)
        self.assertIsNone(document["scope"])
        self.assertEqual(document["findings"], [])
        self.assertEqual(document["summary"], flashlight._zero_summary())
        self.assertEqual(document["error"], {"code": code})

    def public_document(self, source: str, language: str) -> dict:
        encoded = flashlight.scan_code(source, language)
        self.assertIsInstance(encoded, str)
        return json.loads(encoded)

    def test_exact_text_artifact_decodes_to_manifest_identity(self) -> None:
        artifact = flashlight._read_verified_vendor_bytes()
        self.assertEqual(len(artifact), 87_196)
        self.assertEqual(
            hashlib.sha256(artifact).hexdigest(),
            "b7d004947bc3c7619daa38f002d9ddde731e2865644af0d0e609c8dd86528d3c",
        )
        encoded = flashlight.VENDOR_ARTIFACT_B64.read_bytes()
        self.assertTrue(encoded.isascii())
        self.assertNotIn(b"\x00", encoded)

    def test_verified_runtime_contains_only_the_exact_package_identity(self) -> None:
        metadata = json.loads(
            (self.package_root / "package.json").read_text(encoding="utf-8")
        )
        self.assertEqual(metadata["name"], "@agenttool/whitehack-scan")
        self.assertEqual(metadata["version"], "0.9.0")
        self.assertNotIn("dependencies", metadata)
        scripts = metadata.get("scripts", {})
        self.assertTrue(
            {"preinstall", "install", "postinstall"}.isdisjoint(scripts)
        )
        self.assertTrue((self.package_root / "LICENSE").is_file())
        self.assertTrue((self.package_root / "src" / "core.js").is_file())

    def test_safe_extractor_rejects_escape_and_link_entries(self) -> None:
        escape = tarfile.TarInfo("../escape")
        escape.type = tarfile.REGTYPE
        with self.assertRaises(RuntimeError):
            flashlight._safe_member_path(escape)

        absolute = tarfile.TarInfo("/package/core.js")
        absolute.type = tarfile.REGTYPE
        with self.assertRaises(RuntimeError):
            flashlight._safe_member_path(absolute)

        link = tarfile.TarInfo("package/link")
        link.type = tarfile.SYMTYPE
        link.linkname = "../../escape"
        with self.assertRaises(RuntimeError):
            flashlight._safe_member_path(link)

    def test_planted_javascript_finding_has_closed_shape(self) -> None:
        document = self.public_document("eval(userInput)\n", "javascript")
        self.assertEqual(document["status"], "complete")
        self.assertIs(document["complete"], True)
        self.assertEqual(document["scanner"]["version"], "0.9.0")
        self.assertEqual(document["scope"]["rules_considered"], 43)
        self.assertEqual(document["scope"]["utf8_bytes"], 16)
        self.assertEqual(document["scope"]["lines"], 2)
        self.assertTrue(
            any(finding["check"] == "unsafe-eval" for finding in document["findings"])
        )
        for finding in document["findings"]:
            self.assertEqual(
                set(finding),
                {
                    "line",
                    "check",
                    "title",
                    "confidence",
                    "doctrine",
                    "principle",
                },
            )
        serialized = json.dumps(document, sort_keys=True)
        self.assertNotIn("userInput", serialized)
        self.assertNotIn('"snippet":', serialized)
        self.assertNotIn('"message":', serialized)
        self.assertNotIn("[pasted-input]", serialized)

    def test_safe_excerpt_is_complete_but_never_claimed_safe(self) -> None:
        document = self.public_document(
            "export const add = (left, right) => left + right\n",
            "javascript",
        )
        self.assertEqual(document["status"], "complete")
        self.assertEqual(document["findings"], [])
        self.assertEqual(document["summary"]["finding_count"], 0)
        self.assertEqual(
            document["interpretation"]["empty"],
            "not_proof_of_safety",
        )
        serialized = json.dumps(document).lower()
        self.assertNotIn('"safe"', serialized)
        self.assertNotIn('"passed"', serialized)

    def test_same_input_produces_exactly_equal_documents(self) -> None:
        source = "const response = await fetch(url)\nreturn response.json()\n"
        first_raw = flashlight.scan_code(source, "javascript")
        second_raw = flashlight.scan_code(source, "javascript")
        self.assertEqual(first_raw, second_raw)
        first = json.loads(first_raw)
        second = json.loads(second_raw)
        self.assertEqual(
            first_raw,
            json.dumps(first, ensure_ascii=False, sort_keys=True, separators=(",", ":")),
        )
        self.assertEqual(first, second)

    def test_language_rule_scopes_are_explicit(self) -> None:
        cases = {
            "javascript": ("const value = 1\n", 43),
            "python": ("value = 1\n", 21),
            "solidity": ("contract Value {}\n", 10),
        }
        for language, (source, expected_rules) in cases.items():
            with self.subTest(language=language):
                document = self.public_document(source, language)
                self.assertEqual(document["status"], "complete")
                self.assertEqual(
                    document["scope"]["rules_considered"],
                    expected_rules,
                )

    def test_planted_python_secret_is_prompted_but_not_returned(self) -> None:
        # Construct planted test material at runtime so a Whitehack self-scan
        # does not mistake this fixture file for an actual committed secret.
        planted_value = "0x" + ("9a" * 32)
        source = f'private_key = "{planted_value}"\n'
        document = self.public_document(source, "python")
        self.assertEqual(document["status"], "complete")
        self.assertEqual(document["scope"]["rules_considered"], 21)
        self.assertTrue(
            {"hardcoded-secret", "exposed-config"}.issubset(
                {finding["check"] for finding in document["findings"]}
            )
        )
        serialized = json.dumps(document, sort_keys=True)
        self.assertNotIn(planted_value, serialized)
        self.assertNotIn(source, serialized)

    def test_planted_solidity_unchecked_transfer_is_a_review_prompt(self) -> None:
        source = (
            "contract Pay { function pay(IERC20 token, address to, uint256 amount) "
            "external { token.transfer(to, amount); } }\n"
        )
        document = self.public_document(source, "solidity")
        self.assertEqual(document["status"], "complete")
        self.assertEqual(document["scope"]["rules_considered"], 10)
        self.assertTrue(
            any(
                finding["check"] == "unchecked-transfer"
                for finding in document["findings"]
            )
        )
        serialized = json.dumps(document, sort_keys=True)
        self.assertNotIn("token.transfer", serialized)
        self.assertNotIn(source, serialized)

    def test_secret_shaped_text_is_never_returned(self) -> None:
        planted_value = "0x" + ("4f" * 32)
        source = f'const privateKey = "{planted_value}"\n'
        document = self.public_document(source, "javascript")
        serialized = json.dumps(document, sort_keys=True)
        self.assertNotIn(planted_value, serialized)
        self.assertNotIn(source, serialized)
        self.assertTrue(
            any(
                finding["check"] == "hardcoded-secret"
                for finding in document["findings"]
            )
        )

    def test_input_limits_fail_closed_before_scanning(self) -> None:
        cases = [
            ("", "javascript", "input_empty"),
            (" \n\t", "javascript", "input_empty"),
            ("x", "rust", "unsupported_language"),
            (None, "javascript", "invalid_input"),
            ("\ud800", "javascript", "invalid_utf8"),
            ("a" * (flashlight.MAX_UTF8_BYTES + 1), "javascript", "input_byte_limit_exceeded"),
            ("x\n" * flashlight.MAX_LINES, "javascript", "input_line_limit_exceeded"),
        ]
        for source, language, expected in cases:
            with self.subTest(expected=expected):
                self.assert_error(
                    flashlight._scan_code(source, language),
                    expected,
                )

    def test_finding_limit_returns_error_without_partial_findings(self) -> None:
        document = self.public_document(
            "eval(value)\n" * (flashlight.MAX_FINDINGS + 1),
            "javascript",
        )
        self.assert_error(document, "scan_finding_limit_exceeded")

    def test_source_is_stdin_data_never_command_or_shell_source(self) -> None:
        captured: dict = {}

        def recording_runner(command, **kwargs):
            captured["command"] = list(command)
            captured["input"] = kwargs["input"]
            captured["kwargs"] = dict(kwargs)
            return subprocess.run(command, **kwargs)

        with tempfile.TemporaryDirectory() as directory:
            marker = Path(directory) / "must-not-exist"
            source = f'eval("$(touch {marker})")\n'
            document = flashlight._scan_code(
                source,
                "javascript",
                runner=recording_runner,
            )
            self.assertEqual(document["status"], "complete")
            self.assertFalse(marker.exists())

        self.assertEqual(len(captured["command"]), 3)
        self.assertNotIn(source, captured["command"])
        self.assertNotIn("shell", captured["kwargs"])
        self.assertEqual(json.loads(captured["input"])["source"], source)
        self.assertEqual(captured["kwargs"]["cwd"], str(flashlight.BASE_DIR))

    def test_timeout_nonzero_stderr_and_malformed_output_are_closed(self) -> None:
        def timeout_runner(*_args, **_kwargs):
            raise subprocess.TimeoutExpired(["node"], 3)

        def nonzero_runner(*_args, **_kwargs):
            return subprocess.CompletedProcess([], 2, "", "")

        def stderr_runner(*_args, **_kwargs):
            return subprocess.CompletedProcess([], 0, "{}", "unexpected warning")

        def malformed_runner(*_args, **_kwargs):
            return subprocess.CompletedProcess([], 0, "{", "")

        source = "const value = 1\n"
        self.assert_error(
            flashlight._scan_code(source, "javascript", runner=timeout_runner),
            "scanner_timeout",
        )
        self.assert_error(
            flashlight._scan_code(source, "javascript", runner=nonzero_runner),
            "scanner_failed",
        )
        self.assert_error(
            flashlight._scan_code(source, "javascript", runner=stderr_runner),
            "scanner_failed",
        )
        self.assert_error(
            flashlight._scan_code(source, "javascript", runner=malformed_runner),
            "scanner_protocol_error",
        )

    def test_open_or_inconsistent_bridge_response_is_rejected(self) -> None:
        document = self.public_document("const value = 1\n", "javascript")
        opened = dict(document)
        opened["unexpected"] = True
        with self.assertRaises(ValueError):
            flashlight._validate_closed_response(opened)

        inconsistent = json.loads(json.dumps(document))
        inconsistent["summary"]["finding_count"] += 1
        with self.assertRaises(ValueError):
            flashlight._validate_closed_response(inconsistent)

    def test_bridge_scope_must_match_the_exact_submitted_text(self) -> None:
        valid = self.public_document("const value = 1\n", "javascript")
        mismatched = json.loads(json.dumps(valid))
        mismatched["scope"]["utf8_bytes"] += 1

        def mismatched_runner(*_args, **_kwargs):
            return subprocess.CompletedProcess(
                [],
                0,
                json.dumps(mismatched, separators=(",", ":")),
                "",
            )

        self.assert_error(
            flashlight._scan_code(
                "const value = 1\n",
                "javascript",
                runner=mismatched_runner,
            ),
            "scanner_protocol_error",
        )

    def test_parallel_calls_are_isolated_and_deterministic(self) -> None:
        source = "eval(parallelInput)\n"
        with ThreadPoolExecutor(max_workers=4) as executor:
            documents = list(
                executor.map(
                    lambda _: flashlight.scan_code(source, "javascript"),
                    range(8),
                )
            )
        self.assertTrue(all(document == documents[0] for document in documents))
        decoded = [json.loads(document) for document in documents]
        self.assertTrue(all(document == decoded[0] for document in decoded))

    def test_core_import_graph_has_no_active_builtin_capability_imports(self) -> None:
        source_root = self.package_root / "src"
        scanned = [source_root / "core.js"]
        scanned.extend(sorted((source_root / "checks").glob("*.js")))
        scanned.extend(
            source_root / name
            for name in ("lines.js", "redaction.js", "secret-text.js", "source-text.js")
        )
        for path in scanned:
            text = path.read_text(encoding="utf-8")
            with self.subTest(path=path.name):
                self.assertNotRegex(text, r"(?m)^\s*import\s.+\sfrom\s+['\"]node:")
                self.assertNotRegex(text, r"(?m)^\s*import\s+['\"]node:")

    def test_space_surface_keeps_one_public_text_only_mcp_tool(self) -> None:
        app = (flashlight.BASE_DIR / "app.py").read_text(encoding="utf-8")
        readme = (flashlight.BASE_DIR / "README.md").read_text(encoding="utf-8")
        requirements = (
            flashlight.BASE_DIR / "requirements.txt"
        ).read_text(encoding="utf-8")
        packages = (flashlight.BASE_DIR / "packages.txt").read_text(encoding="utf-8")

        self.assertIn('api_name="scan_code"', app)
        self.assertIn("mcp_server=True", app)
        self.assertIn('flagging_mode="never"', app)
        self.assertIn("concurrency_limit=1", app)
        self.assertNotIn("gr.File", app)
        self.assertNotIn("gr.UploadButton", app)
        self.assertNotIn("outputs=gr.JSON", app)
        self.assertIn("outputs=gr.Code", app)
        self.assertIn("sdk: gradio", readme)
        self.assertIn("sdk_version: 6.21.0", readme)
        self.assertEqual(requirements.strip(), "gradio[mcp]==6.21.0")
        self.assertEqual(packages.strip(), "nodejs")


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