flashlight.py

"""Closed, text-only Python boundary for the Whitehack Flashlight Space."""

from __future__ import annotations

import base64
import binascii
import hashlib
import io
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import subprocess
import tarfile
import tempfile
import threading
from typing import Any, Callable, Literal


DOCUMENT_TYPE = "whitehack-flashlight/v0.1"
SCANNER_NAME = "whitehack"
SCANNER_VERSION = "0.9.0"
SCANNER_CHECK_COUNT = 47

MAX_UTF8_BYTES = 65_536
MAX_LINES = 2_000
MAX_FINDINGS = 500
SCANNER_TIMEOUT_SECONDS = 3.0
MAX_BRIDGE_OUTPUT_BYTES = 512 * 1024

BASE_DIR = Path(__file__).resolve().parent
BRIDGE_PATH = BASE_DIR / "bridge.mjs"
VENDOR_ARTIFACT_B64 = (
    BASE_DIR / "vendor" / "agenttool-whitehack-scan-0.9.0.tgz.b64"
)
VENDOR_SHA256 = "b7d004947bc3c7619daa38f002d9ddde731e2865644af0d0e609c8dd86528d3c"
VENDOR_DECODED_SIZE = 87_196
MAX_VENDOR_B64_BYTES = 128 * 1024
MAX_VENDOR_MEMBERS = 256
MAX_VENDOR_EXTRACTED_BYTES = 2 * 1024 * 1024

Language = Literal["javascript", "python", "solidity"]

LANGUAGES: dict[str, dict[str, Any]] = {
    "javascript": {"core": "js", "rules": 43},
    "python": {"core": "py", "rules": 21},
    "solidity": {"core": "sol", "rules": 10},
}

LIMITS = {
    "max_utf8_bytes": MAX_UTF8_BYTES,
    "max_lines": MAX_LINES,
    "max_findings": MAX_FINDINGS,
    "timeout_seconds": SCANNER_TIMEOUT_SECONDS,
}

INTERPRETATION = {
    "finding": "review_prompt_not_vulnerability_verdict",
    "empty": "not_proof_of_safety",
}

PRIVACY = {
    "source_returned": False,
    "snippets_returned": False,
    "application_persistence": "not_written_by_this_app",
    "hosting_platform_retention": "unknown",
}

CONFIDENCES = ("high", "medium-high", "medium", "heuristic")
_CHECK_ID = re.compile(r"^[a-z][a-z0-9-]{0,63}$")
_ERROR_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
_runtime_lock = threading.Lock()
_runtime_package_root: Path | None = None


def _scanner_identity() -> dict[str, Any]:
    return {
        "name": SCANNER_NAME,
        "version": SCANNER_VERSION,
        "check_count": SCANNER_CHECK_COUNT,
    }


def _zero_summary() -> dict[str, Any]:
    return {
        "finding_count": 0,
        "by_confidence": {confidence: 0 for confidence in CONFIDENCES},
    }


def _error_response(code: str) -> dict[str, Any]:
    if not _ERROR_CODE.fullmatch(code):
        code = "scanner_protocol_error"
    return {
        "document_type": DOCUMENT_TYPE,
        "status": "error",
        "complete": False,
        "scanner": _scanner_identity(),
        "limits": dict(LIMITS),
        "scope": None,
        "summary": _zero_summary(),
        "findings": [],
        "interpretation": dict(INTERPRETATION),
        "privacy": dict(PRIVACY),
        "error": {"code": code},
    }


def _validated_input(
    source: object,
    language: object,
) -> tuple[str, str, int, int] | dict[str, Any]:
    if not isinstance(source, str):
        return _error_response("invalid_input")
    if not isinstance(language, str) or language not in LANGUAGES:
        return _error_response("unsupported_language")
    try:
        encoded = source.encode("utf-8", errors="strict")
    except UnicodeEncodeError:
        return _error_response("invalid_utf8")
    if len(encoded) > MAX_UTF8_BYTES:
        return _error_response("input_byte_limit_exceeded")
    line_count = source.count("\n") + 1
    if line_count > MAX_LINES:
        return _error_response("input_line_limit_exceeded")
    if not source.strip():
        return _error_response("input_empty")
    return source, language, len(encoded), line_count


def _read_verified_vendor_bytes() -> bytes:
    info = VENDOR_ARTIFACT_B64.lstat()
    if not info or not VENDOR_ARTIFACT_B64.is_file() or VENDOR_ARTIFACT_B64.is_symlink():
        raise RuntimeError("vendor artifact must be a regular file")
    if info.st_size < 1 or info.st_size > MAX_VENDOR_B64_BYTES:
        raise RuntimeError("vendor artifact text size is outside the fixed boundary")

    encoded = VENDOR_ARTIFACT_B64.read_bytes()
    compact = b"".join(encoded.split())
    try:
        artifact = base64.b64decode(compact, validate=True)
    except (binascii.Error, ValueError) as error:
        raise RuntimeError("vendor artifact is not canonical base64 text") from error
    if len(artifact) != VENDOR_DECODED_SIZE:
        raise RuntimeError("vendor artifact decoded size mismatch")
    if hashlib.sha256(artifact).hexdigest() != VENDOR_SHA256:
        raise RuntimeError("vendor artifact digest mismatch")
    return artifact


def _safe_member_path(member: tarfile.TarInfo) -> PurePosixPath:
    path = PurePosixPath(member.name)
    if (
        path.is_absolute()
        or not path.parts
        or path.parts[0] != "package"
        or any(part in {"", ".", ".."} for part in path.parts)
    ):
        raise RuntimeError("vendor archive contains an unsafe path")
    if not member.isfile() and not member.isdir():
        raise RuntimeError("vendor archive contains a non-regular entry")
    if member.size < 0 or member.size > MAX_VENDOR_EXTRACTED_BYTES:
        raise RuntimeError("vendor archive member is outside the size boundary")
    return path


def _validate_extracted_package(package_root: Path) -> None:
    metadata_path = package_root / "package.json"
    core_path = package_root / "src" / "core.js"
    license_path = package_root / "LICENSE"
    for required in (metadata_path, core_path, license_path):
        if required.is_symlink() or not required.is_file():
            raise RuntimeError("vendor package is missing a required regular file")

    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    if metadata.get("name") != "@agenttool/whitehack-scan":
        raise RuntimeError("vendor package name mismatch")
    if metadata.get("version") != SCANNER_VERSION:
        raise RuntimeError("vendor package version mismatch")
    if metadata.get("dependencies") not in (None, {}):
        raise RuntimeError("vendor package unexpectedly has runtime dependencies")
    scripts = metadata.get("scripts") or {}
    if any(name in scripts for name in ("preinstall", "install", "postinstall")):
        raise RuntimeError("vendor package unexpectedly has an install lifecycle script")


def _extract_verified_runtime() -> Path:
    artifact = _read_verified_vendor_bytes()
    runtime_root = Path(
        tempfile.mkdtemp(prefix=f"whitehack-flashlight-{VENDOR_SHA256[:12]}-")
    )
    extracted_bytes = 0
    member_count = 0

    try:
        with tarfile.open(fileobj=io.BytesIO(artifact), mode="r:gz") as archive:
            members = archive.getmembers()
            if not members or len(members) > MAX_VENDOR_MEMBERS:
                raise RuntimeError("vendor archive member count is outside the boundary")

            seen: set[PurePosixPath] = set()
            for member in members:
                member_count += 1
                path = _safe_member_path(member)
                if path in seen:
                    raise RuntimeError("vendor archive contains a duplicate path")
                seen.add(path)
                extracted_bytes += member.size
                if extracted_bytes > MAX_VENDOR_EXTRACTED_BYTES:
                    raise RuntimeError("vendor archive exceeds the extraction boundary")

                destination = runtime_root.joinpath(*path.parts)
                if member.isdir():
                    destination.mkdir(mode=0o700, parents=True, exist_ok=False)
                    continue

                destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
                source = archive.extractfile(member)
                if source is None:
                    raise RuntimeError("vendor archive member cannot be read")
                with destination.open("xb") as target:
                    shutil.copyfileobj(source, target, length=64 * 1024)
                if destination.stat().st_size != member.size:
                    raise RuntimeError("vendor archive member size mismatch")
                destination.chmod(0o600)

        if member_count != len(members):
            raise RuntimeError("vendor archive member count changed during extraction")
        package_root = runtime_root / "package"
        _validate_extracted_package(package_root)
        return package_root
    except Exception:
        shutil.rmtree(runtime_root, ignore_errors=True)
        raise


def prepare_runtime() -> Path:
    """Verify and safely unpack the exact scanner artifact once per process."""

    global _runtime_package_root
    if _runtime_package_root is not None:
        return _runtime_package_root
    with _runtime_lock:
        if _runtime_package_root is None:
            _runtime_package_root = _extract_verified_runtime()
    return _runtime_package_root


def _node_binary() -> str:
    candidate = shutil.which("node")
    if not candidate:
        raise RuntimeError("node runtime is unavailable")
    resolved = Path(candidate).resolve()
    if not resolved.is_file() or not os.access(resolved, os.X_OK):
        raise RuntimeError("node runtime is not an executable regular file")
    return str(resolved)


def _bridge_request(source: str, language: str) -> str:
    return json.dumps(
        {"source": source, "language": language},
        ensure_ascii=False,
        separators=(",", ":"),
    )


def _validate_summary(summary: object, findings: list[dict[str, Any]]) -> None:
    if not isinstance(summary, dict) or set(summary) != {
        "finding_count",
        "by_confidence",
    }:
        raise ValueError("summary shape mismatch")
    by_confidence = summary["by_confidence"]
    if not isinstance(by_confidence, dict) or tuple(by_confidence) != CONFIDENCES:
        raise ValueError("confidence summary shape mismatch")
    expected = {confidence: 0 for confidence in CONFIDENCES}
    for finding in findings:
        expected[finding["confidence"]] += 1
    if summary["finding_count"] != len(findings) or by_confidence != expected:
        raise ValueError("summary values mismatch")


def _validate_closed_response(document: object) -> dict[str, Any]:
    if not isinstance(document, dict):
        raise ValueError("bridge response must be an object")
    expected_keys = {
        "document_type",
        "status",
        "complete",
        "scanner",
        "limits",
        "scope",
        "summary",
        "findings",
        "interpretation",
        "privacy",
        "error",
    }
    if set(document) != expected_keys:
        raise ValueError("bridge response is open")
    if document["document_type"] != DOCUMENT_TYPE:
        raise ValueError("document type mismatch")
    if document["scanner"] != _scanner_identity():
        raise ValueError("scanner identity mismatch")
    if document["limits"] != LIMITS:
        raise ValueError("limit declaration mismatch")
    if document["interpretation"] != INTERPRETATION:
        raise ValueError("interpretation mismatch")
    if document["privacy"] != PRIVACY:
        raise ValueError("privacy declaration mismatch")

    findings = document["findings"]
    if not isinstance(findings, list) or len(findings) > MAX_FINDINGS:
        raise ValueError("finding collection is invalid")
    previous_key: tuple[int, str] | None = None
    for finding in findings:
        if not isinstance(finding, dict) or set(finding) != {
            "line",
            "check",
            "title",
            "confidence",
            "doctrine",
            "principle",
        }:
            raise ValueError("finding shape mismatch")
        if (
            not isinstance(finding["line"], int)
            or isinstance(finding["line"], bool)
            or finding["line"] < 0
            or finding["line"] > MAX_LINES
        ):
            raise ValueError("finding line is invalid")
        if not isinstance(finding["check"], str) or not _CHECK_ID.fullmatch(
            finding["check"]
        ):
            raise ValueError("finding check is invalid")
        if (
            not isinstance(finding["title"], str)
            or not finding["title"]
            or len(finding["title"]) > 240
        ):
            raise ValueError("finding title is invalid")
        if finding["confidence"] not in CONFIDENCES:
            raise ValueError("finding confidence is invalid")
        if not isinstance(finding["doctrine"], str) or not _CHECK_ID.fullmatch(
            finding["doctrine"]
        ):
            raise ValueError("finding doctrine is invalid")
        if (
            not isinstance(finding["principle"], int)
            or isinstance(finding["principle"], bool)
            or finding["principle"] < 1
            or finding["principle"] > 6
        ):
            raise ValueError("finding principle is invalid")
        order_key = (finding["line"], finding["check"])
        if previous_key is not None and order_key < previous_key:
            raise ValueError("findings are not canonically ordered")
        previous_key = order_key

    _validate_summary(document["summary"], findings)

    if document["complete"] is True:
        if document["status"] != "complete" or document["error"] is not None:
            raise ValueError("complete response state mismatch")
        scope = document["scope"]
        if not isinstance(scope, dict) or set(scope) != {
            "kind",
            "language",
            "utf8_bytes",
            "lines",
            "rules_considered",
        }:
            raise ValueError("scope shape mismatch")
        language = scope["language"]
        if language not in LANGUAGES:
            raise ValueError("scope language mismatch")
        if scope["kind"] != "caller-provided-text":
            raise ValueError("scope kind mismatch")
        if (
            not isinstance(scope["utf8_bytes"], int)
            or isinstance(scope["utf8_bytes"], bool)
            or scope["utf8_bytes"] < 1
            or scope["utf8_bytes"] > MAX_UTF8_BYTES
        ):
            raise ValueError("scope byte count mismatch")
        if (
            not isinstance(scope["lines"], int)
            or isinstance(scope["lines"], bool)
            or scope["lines"] < 1
            or scope["lines"] > MAX_LINES
        ):
            raise ValueError("scope line count mismatch")
        if scope["rules_considered"] != LANGUAGES[language]["rules"]:
            raise ValueError("scope rule count mismatch")
    else:
        if (
            document["status"] != "error"
            or document["scope"] is not None
            or findings
            or document["summary"] != _zero_summary()
            or not isinstance(document["error"], dict)
            or set(document["error"]) != {"code"}
            or not isinstance(document["error"]["code"], str)
            or not _ERROR_CODE.fullmatch(document["error"]["code"])
        ):
            raise ValueError("error response state mismatch")
    return document


Runner = Callable[..., subprocess.CompletedProcess[str]]


def _scan_code(
    source: object,
    language: object,
    *,
    runner: Runner = subprocess.run,
) -> dict[str, Any]:
    validated = _validated_input(source, language)
    if isinstance(validated, dict):
        return validated
    bounded_source, bounded_language, byte_count, line_count = validated

    try:
        package_root = prepare_runtime()
        node = _node_binary()
    except Exception:
        return _error_response("scanner_unavailable")

    request = _bridge_request(bounded_source, bounded_language)
    command = [node, str(BRIDGE_PATH), str(package_root)]
    child_env = {
        "HOME": str(package_root.parent),
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
        "PATH": str(Path(node).parent),
    }
    try:
        completed = runner(
            command,
            input=request,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="strict",
            timeout=SCANNER_TIMEOUT_SECONDS,
            check=False,
            cwd=str(BASE_DIR),
            env=child_env,
        )
    except subprocess.TimeoutExpired:
        return _error_response("scanner_timeout")
    except (OSError, UnicodeError, ValueError):
        return _error_response("scanner_unavailable")

    if (
        completed.returncode != 0
        or completed.stderr
        or len(completed.stdout.encode("utf-8", errors="strict"))
        > MAX_BRIDGE_OUTPUT_BYTES
    ):
        return _error_response("scanner_failed")
    try:
        decoded = json.loads(completed.stdout)
        document = _validate_closed_response(decoded)
        if document["complete"]:
            scope = document["scope"]
            if (
                scope["language"] != bounded_language
                or scope["utf8_bytes"] != byte_count
                or scope["lines"] != line_count
                or any(finding["line"] > line_count for finding in document["findings"])
            ):
                return _error_response("scanner_protocol_error")
        return document
    except (json.JSONDecodeError, UnicodeError, ValueError, TypeError):
        return _error_response("scanner_protocol_error")


def scan_code(source: str, language: Language = "javascript") -> str:
    """Return bounded Whitehack review prompts for caller-provided source text.

    This is heuristic text analysis, not a vulnerability verdict or proof of
    safety. It has no dedicated path, file, archive, repository, URL, wallet,
    or credential capability: strings containing those remain inert text and
    are never opened, fetched, authenticated, signed, broadcast, or executed.
    The app cannot reliably recognize every secret. It does not intentionally
    persist or return source text, but Hugging Face platform retention is
    unknown. Do not submit secrets, private, or proprietary code.

    Args:
        source: Caller-provided UTF-8 source text only; maximum 65,536 encoded bytes and 2,000 lines; never pass secrets, private, proprietary, personal, file, archive, repository, path, or URL data.
        language: Exact rule-pack enum: ``javascript`` for the shared JavaScript/TypeScript pack, ``python``, or ``solidity``.

    Returns:
        Deterministic compact JSON text containing one closed
        ``whitehack-flashlight/v0.1`` object without source, snippets, raw
        messages, or raw errors. ``complete`` describes this bounded scan only;
        it does not certify the source as safe or vulnerable.
    """

    return json.dumps(
        _scan_code(source, language),
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    )