#!/usr/bin/env python3
"""安全执行 knowledge.yaml 中已审查的 verificationRegistry 条目。

本入口只接受 registry ID，不接受额外命令或参数。执行前只打开一次项目根目录
fd，知识库、检查目标和子进程 cwd 都固定到该 fd；检查文件逐段以 O_NOFOLLOW
打开后复制到匿名、尽可能 sealed 的稳定快照，再使用结构化 argv 和 shell=False
启动，避免检查与执行之间被替换。

退出码: 0..125 沿用检查结果 / 1 知识或执行失败 / 2 环境、路径或用法错误。
"""

from __future__ import annotations

import argparse
import fcntl
import os
import stat
import subprocess
import sys
import tempfile
from pathlib import Path, PurePosixPath
from typing import Any

from validate_knowledge import (  # type: ignore
    infer_project_root,
    load_yaml_text,
    validate_all,
)

MAX_KNOWLEDGE_BYTES = 16 * 1024 * 1024
MAX_TARGET_BYTES = 16 * 1024 * 1024


def _project_root(
    value: str | None,
    knowledge_path: Path,
) -> tuple[Path | None, str | None]:
    inferred = infer_project_root(knowledge_path)
    if value is not None:
        candidate = Path(value).expanduser()
        if not candidate.is_dir():
            return None, f"项目根目录不存在: {candidate}"
        resolved = candidate.resolve(strict=True)
        if inferred is not None and resolved != inferred:
            return (
                None,
                f"--project-root {resolved} 与 knowledge.yaml 推断的项目根目录 "
                f"{inferred} 不一致",
            )
        return resolved, None
    if inferred is None:
        return None, "无法从 knowledge.yaml 确定现有项目根目录"
    return inferred, None


def _validate_knowledge_location(
    knowledge_path: Path,
    project_root: Path,
) -> str | None:
    expected = project_root / ".pouch" / "ack" / "knowledge.yaml"
    lexical = Path(os.path.abspath(knowledge_path.expanduser()))
    if lexical != expected:
        return (
            "只允许执行项目权威知识库 "
            f"{expected}，当前输入为 {lexical}"
        )
    return None


def _open_regular_beneath(
    project_root: Path | int,
    relative_path: str,
    *,
    require_executable: bool,
) -> tuple[int | None, str | None]:
    """从根目录 fd 逐段打开目标，不允许任一段通过 symlink 跳转。"""
    if not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"):
        return None, "当前平台不支持安全的 O_NOFOLLOW/O_DIRECTORY 路径解析"
    parts = PurePosixPath(relative_path).parts
    if (
        not parts
        or PurePosixPath(relative_path).is_absolute()
        or any(part in {"", ".", ".."} for part in parts)
    ):
        return None, "检查目标必须是规范的项目内相对路径"

    directory_fds: list[int] = []
    target_fd: int | None = None
    try:
        root_fd = (
            os.dup(project_root)
            if isinstance(project_root, int)
            else os.open(
                project_root,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
            )
        )
        directory_fds.append(root_fd)
        current_fd = root_fd
        for segment in parts[:-1]:
            current_fd = os.open(
                segment,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                dir_fd=current_fd,
            )
            directory_fds.append(current_fd)
        target_fd = os.open(
            parts[-1],
            os.O_RDONLY | os.O_NOFOLLOW,
            dir_fd=current_fd,
        )
        metadata = os.fstat(target_fd)
        if not stat.S_ISREG(metadata.st_mode):
            os.close(target_fd)
            return None, "检查目标不是普通文件"
        if require_executable and metadata.st_mode & 0o111 == 0:
            os.close(target_fd)
            return None, "检查目标不可执行"
        return target_fd, None
    except OSError as exc:
        if target_fd is not None:
            os.close(target_fd)
        return None, f"检查目标不存在、不可访问或路径包含软链接: {exc}"
    finally:
        for directory_fd in reversed(directory_fds):
            os.close(directory_fd)


def _stable_metadata(before: os.stat_result, after: os.stat_result) -> bool:
    fields = (
        "st_dev",
        "st_ino",
        "st_mode",
        "st_size",
        "st_mtime_ns",
        "st_ctime_ns",
    )
    return all(getattr(before, field) == getattr(after, field) for field in fields)


def _read_stable_bytes(
    source_fd: int,
    *,
    maximum: int,
) -> tuple[bytes | None, str | None]:
    before = os.fstat(source_fd)
    if before.st_size > maximum:
        return None, f"知识库超过大小上限 {maximum} bytes"
    chunks: list[bytes] = []
    total = 0
    try:
        os.lseek(source_fd, 0, os.SEEK_SET)
        while True:
            chunk = os.read(source_fd, min(1024 * 1024, maximum - total + 1))
            if not chunk:
                break
            chunks.append(chunk)
            total += len(chunk)
            if total > maximum:
                return None, f"知识库超过大小上限 {maximum} bytes"
    except OSError as exc:
        return None, f"无法读取权威知识库快照: {exc}"
    after = os.fstat(source_fd)
    if not _stable_metadata(before, after):
        return None, "权威知识库在读取期间发生变化，拒绝执行"
    return b"".join(chunks), None


def load_authoritative_knowledge(
    project_root: Path | int,
) -> tuple[dict[str, Any] | None, str | None]:
    source_fd, open_error = _open_regular_beneath(
        project_root,
        ".pouch/ack/knowledge.yaml",
        require_executable=False,
    )
    if open_error is not None or source_fd is None:
        return None, open_error or "无法安全打开权威知识库"
    try:
        content, read_error = _read_stable_bytes(
            source_fd,
            maximum=MAX_KNOWLEDGE_BYTES,
        )
    finally:
        os.close(source_fd)
    if read_error is not None or content is None:
        return None, read_error or "无法读取权威知识库"
    try:
        text = content.decode("utf-8")
    except UnicodeDecodeError as exc:
        return None, f"权威知识库不是有效 UTF-8: {exc}"
    return load_yaml_text(text, "知识库"), None


def _write_all(file_descriptor: int, chunk: bytes) -> None:
    remaining = memoryview(chunk)
    while remaining:
        written = os.write(file_descriptor, remaining)
        if written <= 0:
            raise OSError("无法写入检查快照")
        remaining = remaining[written:]


def _snapshot_executable(source_fd: int) -> tuple[int | None, str | None]:
    """复制到匿名快照；Linux 上进一步 seal，冻结本次执行内容。"""
    before = os.fstat(source_fd)
    if before.st_size > MAX_TARGET_BYTES:
        return None, f"检查目标超过大小上限 {MAX_TARGET_BYTES} bytes"
    snapshot_fd: int | None = None
    seal_snapshot = all(
        hasattr(owner, name)
        for owner, name in (
            (os, "memfd_create"),
            (os, "MFD_ALLOW_SEALING"),
            (os, "MFD_CLOEXEC"),
            (fcntl, "F_ADD_SEALS"),
            (fcntl, "F_SEAL_SEAL"),
            (fcntl, "F_SEAL_SHRINK"),
            (fcntl, "F_SEAL_GROW"),
            (fcntl, "F_SEAL_WRITE"),
        )
    )
    try:
        if seal_snapshot:
            snapshot_fd = os.memfd_create(  # type: ignore[attr-defined]
                "ack-verification",
                os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING,  # type: ignore[attr-defined]
            )
        else:
            snapshot_fd, snapshot_path = tempfile.mkstemp(
                prefix="ack-verification-"
            )
            os.unlink(snapshot_path)

        os.lseek(source_fd, 0, os.SEEK_SET)
        total = 0
        while True:
            chunk = os.read(
                source_fd,
                min(1024 * 1024, MAX_TARGET_BYTES - total + 1),
            )
            if not chunk:
                break
            _write_all(snapshot_fd, chunk)
            total += len(chunk)
            if total > MAX_TARGET_BYTES:
                os.close(snapshot_fd)
                return None, f"检查目标超过大小上限 {MAX_TARGET_BYTES} bytes"
        after = os.fstat(source_fd)
        if not _stable_metadata(before, after):
            os.close(snapshot_fd)
            return None, "检查目标在创建执行快照期间发生变化，拒绝执行"

        os.fchmod(snapshot_fd, before.st_mode & 0o777)
        os.lseek(snapshot_fd, 0, os.SEEK_SET)
        if seal_snapshot:
            seals = (
                fcntl.F_SEAL_SEAL
                | fcntl.F_SEAL_SHRINK
                | fcntl.F_SEAL_GROW
                | fcntl.F_SEAL_WRITE
            )
            fcntl.fcntl(snapshot_fd, fcntl.F_ADD_SEALS, seals)
        return snapshot_fd, None
    except OSError as exc:
        if snapshot_fd is not None:
            os.close(snapshot_fd)
        return None, f"无法创建稳定的检查执行快照: {exc}"


def open_target(
    data: dict[str, Any],
    verification_ref: str,
    project_root: Path | int,
) -> tuple[int | None, list[str] | None, str | None]:
    registry = data.get("verificationRegistry")
    if not isinstance(registry, dict) or verification_ref not in registry:
        return None, None, f"verification.ref {verification_ref!r} 未在 registry 注册"
    target = registry.get(verification_ref)
    if not isinstance(target, dict):
        return None, None, f"verificationRegistry.{verification_ref} 不是对象"
    relative_path = target.get("path")
    args = target.get("args")
    if not isinstance(relative_path, str) or not isinstance(args, list):
        return None, None, f"verificationRegistry.{verification_ref} 结构无效"
    if any(not isinstance(arg, str) for arg in args):
        return (
            None,
            None,
            f"verificationRegistry.{verification_ref}.args 必须是字符串数组",
        )
    source_fd, error = _open_regular_beneath(
        project_root,
        relative_path,
        require_executable=True,
    )
    if error is not None or source_fd is None:
        return None, None, error or "无法安全打开检查目标"
    try:
        target_fd, snapshot_error = _snapshot_executable(source_fd)
    finally:
        os.close(source_fd)
    if snapshot_error is not None or target_fd is None:
        return None, None, snapshot_error or "无法创建检查执行快照"
    return target_fd, args, None


def _fd_executable_path(target_fd: int) -> str | None:
    for prefix in ("/proc/self/fd", "/dev/fd"):
        candidate = f"{prefix}/{target_fd}"
        if Path(candidate).exists():
            return candidate
    return None


def _fd_directory_path(directory_fd: int) -> str | None:
    for prefix in ("/proc/self/fd", "/dev/fd"):
        candidate = f"{prefix}/{directory_fd}"
        if Path(candidate).is_dir():
            return candidate
    return None


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="执行 ACK verificationRegistry 中已审查的检查"
    )
    parser.add_argument("knowledge", help="knowledge.yaml 路径")
    parser.add_argument("verification_ref", help="verificationRegistry 中的检查 ID")
    parser.add_argument(
        "--project-root",
        help="项目根目录；默认从 knowledge.yaml 的 .pouch/ack 布局或 Git 推断",
    )
    args = parser.parse_args(argv)

    knowledge_path = Path(args.knowledge)
    project_root, root_error = _project_root(args.project_root, knowledge_path)
    if project_root is None:
        sys.stderr.write(f"{root_error or '无法确定现有项目根目录'}\n")
        return 2
    location_error = _validate_knowledge_location(knowledge_path, project_root)
    if location_error is not None:
        sys.stderr.write(f"{location_error}\n")
        return 2

    try:
        project_root_fd = os.open(
            project_root,
            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
        )
    except OSError as exc:
        sys.stderr.write(f"无法安全打开项目根目录: {exc}\n")
        return 2

    try:
        stable_root_path = _fd_directory_path(project_root_fd)
        if stable_root_path is None:
            sys.stderr.write("当前平台无法固定项目根目录文件描述符\n")
            return 2
        data, knowledge_error = load_authoritative_knowledge(project_root_fd)
        if knowledge_error is not None or data is None:
            sys.stderr.write(f"{knowledge_error or '无法读取权威知识库'}\n")
            return 2
        schema_path = (
            Path(__file__).resolve().parent.parent
            / "templates"
            / "knowledge.schema.json"
        )
        errors, mode = validate_all(
            data,
            schema_path,
            project_root=Path(stable_root_path),
        )
        if errors:
            sys.stderr.write(f"知识库校验失败（{mode}），拒绝执行：\n")
            for error in errors:
                sys.stderr.write(f"  - {error}\n")
            return 1

        target_fd, target_args, error = open_target(
            data,
            args.verification_ref,
            project_root_fd,
        )
        if error is not None or target_fd is None or target_args is None:
            sys.stderr.write(f"{error or '无法解析检查目标'}\n")
            return 2

        try:
            executable_path = _fd_executable_path(target_fd)
            if executable_path is None:
                sys.stderr.write(
                    "当前平台无法从已打开的文件描述符安全执行检查\n"
                )
                return 2
            environment = os.environ.copy()
            environment["ACK_PROJECT_ROOT"] = stable_root_path
            environment["ACK_PROJECT_ROOT_DISPLAY"] = str(project_root)
            environment["ACK_VERIFICATION_REF"] = args.verification_ref
            registry = data.get("verificationRegistry")
            target = (
                registry.get(args.verification_ref)
                if isinstance(registry, dict)
                else None
            )
            if isinstance(target, dict) and isinstance(target.get("path"), str):
                environment["ACK_VERIFICATION_PATH"] = target["path"]
            try:
                completed = subprocess.run(
                    [executable_path, *target_args],
                    cwd=stable_root_path,
                    env=environment,
                    shell=False,
                    check=False,
                    pass_fds=(project_root_fd, target_fd),
                )
            except OSError as exc:
                sys.stderr.write(f"检查启动失败: {exc}\n")
                return 1
        finally:
            os.close(target_fd)
    finally:
        os.close(project_root_fd)
    if 0 <= completed.returncode <= 125:
        return completed.returncode
    return 1


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