scripts/hidden_character_scanner.py — Hidden Character Scanner

#!/usr/bin/env python3
# By Zero Width Lab (zerowidthlab.com)
# Version: 0.1.1
"""Read-only scanner for concealed Unicode in user-selected text."""

from __future__ import annotations

import argparse
import errno
import json
import os
import stat
import sys
import unicodedata
from pathlib import Path

__author__ = "Zero Width Lab"
DEFAULT_MAX_BYTES = 1024 * 1024

SKIP_DIRS = {".git", "node_modules", ".venv", "dist", "build"}
CODE_SUFFIXES = {
    ".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", ".json",
    ".yaml", ".yml", ".toml", ".html", ".css", ".sh", ".rs",
    ".go", ".java", ".c", ".h", ".cpp", ".cs", ".rb", ".php",
}
ZERO_WIDTH = set(range(0x200B, 0x200E)) | {0x2060, 0xFEFF}
BIDI = {0x200E, 0x200F, 0x061C} | set(range(0x202A, 0x202F)) | set(range(0x2066, 0x206A))
WHITESPACE = {0x00A0, 0x202F, 0x205F, 0x3000} | set(range(0x2000, 0x200B))
EMBED_OPEN = set(range(0x202A, 0x202F)) - {0x202C}
ISOLATE_OPEN = set(range(0x2066, 0x2069))


class CliError(Exception):
    """Expected input or usage failure."""


class SafeArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise CliError(message)


def display_text(value: object) -> str:
    """Render untrusted text on one visible ASCII line."""
    return json.dumps(str(value), ensure_ascii=True)[1:-1].replace(chr(0x7F), "\\u007f")


def codepoint(value: str) -> str:
    return f"U+{ord(value):04X}"


def emoji_base(value: str) -> bool:
    number = ord(value)
    return (value in "0123456789#*"
            or 0x1F000 <= number <= 0x1FAFF or 0x2600 <= number <= 0x27BF
            or number in {0x00A9, 0x00AE, 0x203C, 0x2049, 0x2122, 0x2139}
            or 0x2194 <= number <= 0x21AA or 0x231A <= number <= 0x23FA)


def finding(category: str, severity: str, value: str, file_name: str,
            line: int, column: int) -> dict[str, object]:
    return {
        "category": category,
        "severity": severity,
        "code_point": codepoint(value),
        "unicode_name": unicodedata.name(value, "UNASSIGNED"),
        "file": file_name,
        "line": line,
        "column": column,
    }


def scan_text(text: str, file_name: str, code_file: bool) -> list[dict[str, object]]:
    results: list[dict[str, object]] = []
    offset = 0
    for line_number, line in enumerate(text.splitlines(keepends=True), 1):
        opens: list[tuple[str, int]] = []
        token: list[tuple[str, int]] = []

        def flush_token() -> None:
            if not token:
                return
            scripts = set()
            for char, _ in token:
                name = unicodedata.name(char, "")
                if name.startswith("LATIN "):
                    scripts.add("LATIN")
                elif name.startswith("CYRILLIC "):
                    scripts.add("CYRILLIC")
                elif name.startswith("GREEK "):
                    scripts.add("GREEK")
            if "LATIN" in scripts and ("CYRILLIC" in scripts or "GREEK" in scripts):
                for char, column in token:
                    name = unicodedata.name(char, "")
                    if name.startswith(("CYRILLIC ", "GREEK ")):
                        results.append(finding("mixed-script", "medium", char,
                                               file_name, line_number, column))
            token.clear()

        for index, char in enumerate(line):
            column = index + 1
            number = ord(char)
            if char.isalpha() or (token and unicodedata.category(char).startswith("M")):
                token.append((char, column))
            else:
                flush_token()

            category = None
            if number == 0xFEFF and offset + index == 0:
                continue
            if number in ZERO_WIDTH:
                category = "zero-width"
            elif number in BIDI:
                category = "bidirectional-control"
            elif 0xE0000 <= number <= 0xE007F:
                category = "unicode-tag"
            elif (0xFE00 <= number <= 0xFE0F or 0xE0100 <= number <= 0xE01EF):
                previous = line[index - 1] if index else ""
                if not previous or not emoji_base(previous):
                    category = "variation-selector"
            elif code_file and unicodedata.category(char) == "Cf":
                category = "format-character"
            elif code_file and number in WHITESPACE:
                category = "unusual-whitespace"
            if category:
                results.append(finding(category, "medium", char, file_name,
                                       line_number, column))

            if number in EMBED_OPEN:
                opens.append(("embedding", len(results) - 1))
            elif number in ISOLATE_OPEN:
                opens.append(("isolate", len(results) - 1))
            elif number == 0x202C:
                if opens and opens[-1][0] == "embedding":
                    opens.pop()
            elif number == 0x2069:
                if opens and opens[-1][0] == "isolate":
                    opens.pop()
        flush_token()
        for _, result_index in opens:
            results[result_index]["category"] = "trojan-source"
            results[result_index]["severity"] = "high"
        offset += len(line)
    return results


def skip(result: dict[str, object], label: str, reason: str, incomplete: bool = False) -> None:
    result["skipped"].append({"path": label, "reason": reason})
    if incomplete:
        result["incomplete"] = True


def scan_entry(parent_fd: int, name: str, label: str, args: argparse.Namespace,
               result: dict[str, object], explicit: bool = False) -> None:
    try:
        info = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
        if stat.S_ISLNK(info.st_mode):
            skip(result, label, "symlink", explicit)
            return
        if stat.S_ISDIR(info.st_mode):
            if name in SKIP_DIRS or (name.startswith(".") and not args.include_hidden and name != "."):
                skip(result, label, "excluded directory")
                return
            directory_fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                   dir_fd=parent_fd)
            try:
                with os.scandir(directory_fd) as entries:
                    children = sorted((entry.name for entry in entries))
                for child in children:
                    scan_entry(directory_fd, child, os.path.join(label, child), args, result)
            finally:
                os.close(directory_fd)
            return
        if not stat.S_ISREG(info.st_mode):
            skip(result, label, "not a regular file")
            return
        if name.startswith(".") and not args.include_hidden:
            skip(result, label, "hidden file")
            return
        descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent_fd)
        with os.fdopen(descriptor, "rb") as stream:
            if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
                skip(result, label, "not a regular file")
                return
            data = stream.read(args.max_file_bytes + 1)
        if len(data) > args.max_file_bytes:
            skip(result, label, "size limit", True)
            return
        if b"\x00" in data:
            skip(result, label, "binary NUL")
            return
        try:
            content = data.decode("utf-8")
        except UnicodeError:
            skip(result, label, "invalid UTF-8")
            return
        result["scanned_files"] += 1
        result["findings"].extend(scan_text(content, label, Path(name).suffix.lower() in CODE_SUFFIXES))
    except OSError as error:
        if error.errno == errno.ELOOP:
            skip(result, label, "symlink", explicit)
        else:
            skip(result, label, f"unreadable: {error.strerror or error.__class__.__name__}", True)


def scan_path(path: Path, args: argparse.Namespace, result: dict[str, object]) -> None:
    parts = path.parts
    if not parts:
        parts = (".",)
    root = "/" if path.is_absolute() else "."
    parent_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
    try:
        for component in parts[:-1]:
            if component == "/":
                continue
            try:
                if stat.S_ISLNK(os.stat(component, dir_fd=parent_fd,
                                        follow_symlinks=False).st_mode):
                    skip(result, str(path), "symlink", True)
                    return
                next_fd = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                                  dir_fd=parent_fd)
            except OSError as error:
                skip(result, str(path), "symlink" if error.errno == errno.ELOOP else "unreadable parent", True)
                return
            os.close(parent_fd)
            parent_fd = next_fd
        scan_entry(parent_fd, parts[-1], str(path), args, result, explicit=True)
    finally:
        os.close(parent_fd)


def scan_stdin(args: argparse.Namespace, result: dict[str, object]) -> None:
    data = sys.stdin.buffer.read(args.max_file_bytes + 1)
    if len(data) > args.max_file_bytes:
        result["skipped"].append({"path": "<stdin>", "reason": "size limit"})
        result["incomplete"] = True
        return
    try:
        content = data.decode("utf-8")
    except UnicodeError as error:
        raise CliError(f"stdin is not UTF-8: {error}") from error
    result["scanned_files"] += 1
    result["findings"].extend(scan_text(content, "<stdin>", False))


def scan(args: argparse.Namespace) -> int:
    if any(path != "-" for path in args.paths):
        require_file_scan_support()
    if args.max_file_bytes < 1:
        raise CliError("--max-file-bytes must be positive")
    result: dict[str, object] = {"findings": [], "skipped": [], "scanned_files": 0,
                                 "incomplete": False}
    for raw_path in args.paths:
        if raw_path == "-":
            scan_stdin(args, result)
        else:
            scan_path(Path(raw_path), args, result)
    result["summary"] = {"findings": len(result["findings"]),
                         "scanned_files": result["scanned_files"],
                         "skipped": len(result["skipped"]),
                         "incomplete": result["incomplete"]}
    if args.json:
        print(json.dumps(result, ensure_ascii=True, indent=2).replace(chr(0x7F), "\\u007f"))
    else:
        for item in result["findings"]:
            print(f"{display_text(item['file'])}:{item['line']}:{item['column']} "
                  f"{item['severity']} {item['category']} {item['code_point']} "
                  f"{display_text(item['unicode_name'])}")
        for item in result["skipped"]:
            print(f"Skipped {display_text(item['path'])}: {display_text(item['reason'])}")
        summary = result["summary"]
        print(f"Findings: {summary['findings']}; scanned: {summary['scanned_files']}; "
              f"skipped: {summary['skipped']}; incomplete: {str(summary['incomplete']).lower()}")
    if result["incomplete"] and not args.allow_partial:
        return 2
    return 1 if result["findings"] else 0


def require_file_scan_support() -> None:
    if (os.name != "posix"
            or not getattr(os, "O_NOFOLLOW", 0)
            or not getattr(os, "O_DIRECTORY", 0)
            or not {os.open, os.stat}.issubset(getattr(os, "supports_dir_fd", set()))
            or os.stat not in getattr(os, "supports_follow_symlinks", set())
            or os.scandir not in getattr(os, "supports_fd", set())):
        raise CliError("File scanning requires POSIX directory-descriptor and no-follow support "
                       "(macOS or Linux); Windows is unsupported. Use stdin-only scan or decode-tags.")


def decode_tags(args: argparse.Namespace) -> int:
    if args.text == "-":
        data = sys.stdin.buffer.read(DEFAULT_MAX_BYTES + 1)
        if len(data) > DEFAULT_MAX_BYTES:
            raise CliError("decode-tags stdin exceeds the 1048576-byte (1 MiB) limit")
        try:
            text = data.decode("utf-8")
        except UnicodeError as error:
            raise CliError("decode-tags stdin is not UTF-8") from error
    else:
        text = args.text
    decoded = "".join(chr(ord(char) - 0xE0000) for char in text
                      if 0xE0000 <= ord(char) <= 0xE007F)
    print(display_text(decoded))
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = SafeArgumentParser(
        description="Scan local text for hidden Unicode, offline and read-only.",
        epilog="By Zero Width Lab (zerowidthlab.com)")
    commands = parser.add_subparsers(dest="command", required=True, parser_class=SafeArgumentParser)
    scan_parser = commands.add_parser("scan", help="Scan files, directories or stdin",
                                      epilog="By Zero Width Lab (zerowidthlab.com)")
    scan_parser.add_argument("paths", nargs="+", metavar="PATH")
    scan_parser.add_argument("--json", action="store_true", help="Print JSON")
    scan_parser.add_argument("--max-file-bytes", type=int, default=DEFAULT_MAX_BYTES,
                             help="File and scan stdin byte limit (default: 1048576)")
    scan_parser.add_argument("--include-hidden", action="store_true")
    scan_parser.add_argument("--allow-partial", action="store_true")
    decode_parser = commands.add_parser("decode-tags", help="Reveal ASCII carried by Unicode tag characters",
                                        description="Decode supplied text or UTF-8 stdin (maximum 1048576 bytes).",
                                        epilog="By Zero Width Lab (zerowidthlab.com)")
    decode_parser.add_argument("text", metavar="TEXT|-")
    try:
        args = parser.parse_args(argv)
        return scan(args) if args.command == "scan" else decode_tags(args)
    except (CliError, OSError) as error:
        print(f"Error: {display_text(error)}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

License: MIT (read it).

Back to the listing