tests/test_hidden_character_scanner.py — Hidden Character Scanner

# By Zero Width Lab (zerowidthlab.com)
"""Exercise the read-only scanner with synthetic local input.

By Zero Width Lab (zerowidthlab.com)
"""

from __future__ import annotations

import argparse
import contextlib
import importlib.util
import io
import json
import os
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
from pathlib import Path

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
ROOT = Path(__file__).resolve().parent.parent
SCRIPT = ROOT / "scripts" / "hidden_character_scanner.py"
MODULE_SPEC = importlib.util.spec_from_file_location("scanner", SCRIPT)
scanner = importlib.util.module_from_spec(MODULE_SPEC)
MODULE_SPEC.loader.exec_module(scanner)
REAL_TEMP_DIR = Path(tempfile.gettempdir()).resolve()


class HiddenCharacterScannerTests(unittest.TestCase):
    def run_cli(self, *args: str, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
        environment = os.environ.copy()
        return subprocess.run(
            [sys.executable, str(SCRIPT), *args],
            cwd=ROOT,
            env=environment,
            input=input_text,
            capture_output=True,
            text=True,
            check=False,
        )

    def write_text(self, directory: Path, name: str, contents: str) -> Path:
        path = directory / name
        path.write_text(contents, encoding="utf-8")
        return path

    def scan_json(self, *args: str) -> tuple[subprocess.CompletedProcess[str], dict]:
        result = self.run_cli("scan", *args, "--json")
        return result, json.loads(result.stdout)

    def test_file_scan_requires_each_platform_capability_before_reading(self) -> None:
        cases = (
            ("name", "nt"), ("O_NOFOLLOW", None), ("O_DIRECTORY", None),
            ("supports_dir_fd", os.supports_dir_fd - {os.open}),
            ("supports_dir_fd", os.supports_dir_fd - {os.stat}),
            ("supports_follow_symlinks", os.supports_follow_symlinks - {os.stat}),
            ("supports_fd", os.supports_fd - {os.scandir}),
        )
        for attribute, value in cases:
            with self.subTest(attribute=attribute, value=value), contextlib.ExitStack() as stack:
                if value is None:
                    stack.enter_context(mock.patch.object(scanner.os, attribute))
                    delattr(scanner.os, attribute)
                else:
                    stack.enter_context(mock.patch.object(scanner.os, attribute, value))
                file_read = stack.enter_context(mock.patch.object(scanner, "scan_path"))
                stdin_read = stack.enter_context(mock.patch.object(scanner, "scan_stdin"))
                stdout, stderr = io.StringIO(), io.StringIO()
                stack.enter_context(contextlib.redirect_stdout(stdout))
                stack.enter_context(contextlib.redirect_stderr(stderr))
                status = scanner.main(["scan", "-", "selected.txt", "--allow-partial"])
                self.assertEqual(status, 1)
                self.assertEqual(stdout.getvalue(), "")
                self.assertEqual(len(stderr.getvalue().splitlines()), 1)
                self.assertIn("File scanning requires POSIX", stderr.getvalue())
                self.assertIn("Windows is unsupported", stderr.getvalue())
                file_read.assert_not_called()
                stdin_read.assert_not_called()

    def test_text_only_commands_work_without_file_scan_support(self) -> None:
        for args in (["scan", "-"], ["decode-tags", "-"], ["decode-tags", "ordinary"]):
            with self.subTest(args=args), mock.patch.object(scanner.os, "name", "nt"), \
                    mock.patch.object(scanner, "require_file_scan_support", side_effect=AssertionError), \
                    mock.patch.object(scanner.sys, "stdin", io.TextIOWrapper(io.BytesIO(b"plain"))), \
                    contextlib.redirect_stdout(io.StringIO()):
                self.assertEqual(scanner.main(args), 0)

    def test_decode_stdin_byte_limit_and_utf8(self) -> None:
        limit = 1024 * 1024
        tag = chr(0xE0041).encode("utf-8")
        cases = ((b"x" * (limit - 4) + tag, 0, "A"),
                 (b"x" * (limit - 3) + tag, 1, "exceeds"),
                 (b"\xff", 1, "not UTF-8"))
        for data, expected_status, expected_text in cases:
            with self.subTest(length=len(data)):
                stream = mock.Mock(wraps=io.BytesIO(data))
                stdin = mock.Mock(buffer=stream)
                stdout, stderr = io.StringIO(), io.StringIO()
                with mock.patch.object(scanner.sys, "stdin", stdin), \
                        contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
                    self.assertEqual(scanner.main(["decode-tags", "-"]), expected_status)
                stream.read.assert_called_once_with(limit + 1)
                stdin.read.assert_not_called()
                if expected_status:
                    self.assertEqual(stdout.getvalue(), "")
                    self.assertEqual(len(stderr.getvalue().splitlines()), 1)
                    self.assertIn(expected_text, stderr.getvalue())
                else:
                    self.assertEqual(stdout.getvalue(), expected_text + "\n")

    def test_clean_file_and_stdin(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "clean.py", "value = 'visible'\n")
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(data["findings"], [])
        self.assertFalse(data["summary"]["incomplete"])
        stdin_result = self.run_cli("scan", "-", "--json", input_text="plain prompt\n")
        self.assertEqual(stdin_result.returncode, 0, stdin_result.stderr)
        self.assertEqual(json.loads(stdin_result.stdout)["findings"], [])

    def test_zero_width_code_points_and_bom_rule(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            text = chr(0xFEFF) + "a" + chr(0x200B) + chr(0x200C) + chr(0x200D) + chr(0x2060) + chr(0xFEFF)
            path = self.write_text(Path(temporary), "sample.py", text)
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 1, result.stderr)
        points = [item["code_point"] for item in data["findings"]]
        self.assertEqual(points, ["U+200B", "U+200C", "U+200D", "U+2060", "U+FEFF"])
        self.assertEqual(data["findings"][0]["line"], 1)
        self.assertEqual(data["findings"][0]["column"], 3)
        for item in data["findings"]:
            self.assertTrue(item["category"])
            self.assertTrue(item["unicode_name"])
            self.assertTrue(item["file"])
        self.assertNotIn(chr(0x200B), result.stdout)

    def test_bidi_controls_and_unterminated_line(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "source.py", "a" + chr(0x202E) + "b\n")
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertTrue(any(item["code_point"] == "U+202E" for item in data["findings"]))
        self.assertTrue(any(item["severity"] == "high" and item["category"] == "trojan-source" for item in data["findings"]))
        self.assertNotIn(chr(0x202E), result.stdout)

    def test_unterminated_isolates_are_high_severity(self) -> None:
        for point in (0x2066, 0x2067, 0x2068):
            with self.subTest(point=point), tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
                path = self.write_text(Path(temporary), "isolate.txt", "a" + chr(point) + "b\n")
                result, data = self.scan_json(str(path))
                self.assertEqual(result.returncode, 1, result.stderr)
                self.assertEqual(len(data["findings"]), 1)
                self.assertEqual(data["findings"][0]["code_point"], f"U+{point:04X}")
                self.assertEqual(data["findings"][0]["category"], "trojan-source")
                self.assertEqual(data["findings"][0]["severity"], "high")

    def test_tag_decoder_reveals_ascii_without_raw_tag_characters(self) -> None:
        tagged = "".join(chr(0xE0000 + ord(character)) for character in "Hi!")
        result = self.run_cli("decode-tags", "-", input_text=tagged)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("Hi!", result.stdout)
        for character in tagged:
            self.assertNotIn(character, result.stdout)
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "prompt.txt", tagged)
            scan, data = self.scan_json(str(path))
        self.assertEqual(scan.returncode, 1, scan.stderr)
        self.assertEqual([item["code_point"] for item in data["findings"]], ["U+E0048", "U+E0069", "U+E0021"])

    def test_variation_selectors_and_code_whitespace(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            directory = Path(temporary)
            code = self.write_text(directory, "sample.py", "x" + chr(0xFE0F) + " " + chr(0x00A0) + chr(0x2003) + "\n")
            result, data = self.scan_json(str(code))
            prose = self.write_text(directory, "sample.txt", "x" + chr(0x00A0) + chr(0x2003))
            prose_result, prose_data = self.scan_json(str(prose))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertTrue({"U+FE0F", "U+00A0", "U+2003"}.issubset({item["code_point"] for item in data["findings"]}))
        self.assertEqual(prose_result.returncode, 0, prose_result.stderr)
        self.assertEqual(prose_data["findings"], [])

    def test_variation_selector_after_emoji_base_is_not_flagged(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "emoji.txt", chr(0x2764) + chr(0xFE0F))
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(data["findings"], [])

    def test_balanced_bidi_control_stays_in_control_category(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "balanced.txt", chr(0x202A) + "text" + chr(0x202C))
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertEqual([item["category"] for item in data["findings"]],
                         ["bidirectional-control", "bidirectional-control"])

    def test_other_format_character_in_code(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            directory = Path(temporary)
            path = self.write_text(directory, "format.py", "x" + chr(0x00AD))
            result, data = self.scan_json(str(path))
            prose = self.write_text(directory, "format.txt", "x" + chr(0x00AD))
            prose_result, prose_data = self.scan_json(str(prose))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertEqual(data["findings"][0]["category"], "format-character")
        self.assertEqual(prose_result.returncode, 0, prose_result.stderr)
        self.assertEqual(prose_data["findings"], [])

    def test_global_character_categories_in_prose(self) -> None:
        text = (chr(0x200B) + chr(0x202A) + "a" + chr(0x202C)
                + chr(0xE0041) + "x" + chr(0xFE0F))
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "prose.txt", text)
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertEqual([item["category"] for item in data["findings"]],
                         ["zero-width", "bidirectional-control", "bidirectional-control",
                          "unicode-tag", "variation-selector"])

    def test_mixed_script_word_reports_non_latin_letters(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "mixed.py", "ca" + chr(0x0440) + "e g" + chr(0x03BF) + "al\n")
            result, data = self.scan_json(str(path))
        self.assertEqual(result.returncode, 1, result.stderr)
        points = {item["code_point"] for item in data["findings"]}
        self.assertTrue({"U+0440", "U+03BF"}.issubset(points))
        self.assertNotIn(chr(0x0440), result.stdout)

    def test_directory_skips_binary_symlinks_and_standard_exclusions(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            root = Path(temporary)
            self.write_text(root, "visible.py", "x" + chr(0x200B))
            (root / "binary.bin").write_bytes(b"\x00\xff")
            for name in (".git", "node_modules", ".venv", "dist", "build"):
                subdir = root / name
                subdir.mkdir()
                self.write_text(subdir, "ignored.py", "x" + chr(0x200B))
            (root / "link.py").symlink_to(root / "visible.py")
            result, data = self.scan_json(str(root))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertEqual(len(data["findings"]), 1)
        self.assertFalse(data["summary"]["incomplete"])
        self.assertTrue(data["skipped"])

    def test_explicit_path_through_symlinked_parent_is_not_read(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            root = Path(temporary)
            actual = root / "actual"
            actual.mkdir()
            self.write_text(actual, "concealed.py", "x" + chr(0x200B))
            alias = root / "alias"
            alias.symlink_to(actual, target_is_directory=True)
            result, data = self.scan_json(str(alias / "concealed.py"))
        self.assertEqual(result.returncode, 2, result.stderr)
        self.assertEqual(data["findings"], [])
        self.assertEqual(data["scanned_files"], 0)
        self.assertTrue(data["incomplete"])
        self.assertEqual(data["skipped"][0]["reason"], "symlink")

    def test_hidden_files_require_opt_in(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), ".hidden.py", "x" + chr(0x200B))
            ordinary, ordinary_data = self.scan_json(str(path.parent))
            included, included_data = self.scan_json(str(path.parent), "--include-hidden")
        self.assertEqual(ordinary.returncode, 0, ordinary.stderr)
        self.assertEqual(ordinary_data["findings"], [])
        self.assertEqual(included.returncode, 1, included.stderr)
        self.assertEqual(len(included_data["findings"]), 1)

    def test_size_limit_marks_partial_and_lists_file(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "large.py", "abcdefghij")
            result, data = self.scan_json(str(path), "--max-file-bytes", "3")
        self.assertEqual(result.returncode, 2, result.stderr)
        self.assertTrue(data["summary"]["incomplete"])
        self.assertEqual(data["skipped"][0]["reason"], "size limit")

    def test_missing_path_is_partial_unless_opted_in(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            missing = str(Path(temporary) / "missing.py")
            strict, strict_data = self.scan_json(missing)
            allowed, allowed_data = self.scan_json(missing, "--allow-partial")
        self.assertEqual(strict.returncode, 2, strict.stderr)
        self.assertTrue(strict_data["summary"]["incomplete"])
        self.assertTrue(strict_data["skipped"])
        self.assertEqual(allowed.returncode, 0, allowed.stderr)
        self.assertTrue(allowed_data["summary"]["incomplete"])

    def test_invalid_limit_is_usage_error(self) -> None:
        result = self.run_cli("scan", "-", "--max-file-bytes", "0", input_text="a")
        self.assertEqual(result.returncode, 1)
        self.assertTrue(result.stderr)

    def test_human_output_escapes_malicious_path(self) -> None:
        with tempfile.TemporaryDirectory(dir=REAL_TEMP_DIR) as temporary:
            path = self.write_text(Path(temporary), "line\nbreak.py", "x" + chr(0x200B))
            result = self.run_cli("scan", str(path))
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertNotIn("line\nbreak.py", result.stdout)
        self.assertNotIn(chr(0x200B), result.stdout)
        self.assertIn("U+200B", result.stdout)


def main() -> None:
    parser = argparse.ArgumentParser(description="Run synthetic Hidden Character Scanner tests.", epilog=SIGNATURE)
    _, remaining = parser.parse_known_args()
    unittest.main(argv=[sys.argv[0], *remaining])


if __name__ == "__main__":
    main()

License: MIT (read it).

Back to the listing