agent-Specialization/modules/host_sandbox_runner.py

from __future__ import annotations

import os
import platform
import shutil
import shlex
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional
from modules.host_sandbox_policy import (
    get_macos_writable_paths,
    get_macos_deny_read_paths,
    get_macos_deny_read_regexes,
)


@dataclass
class SandboxPlan:
    command: List[str]
    env: Dict[str, str]
    cwd: Optional[str] = None
    seccomp_bpf_path: Optional[str] = None


class HostSandboxError(RuntimeError):
    pass


# macOS 最小可读系统路径集合（Codex 风格 deny-default + allow-list）。
# 当前仅作为常量保留，供后续需要严格 allow-list 的只读沙箱模式使用。
# 当前只读沙箱采用“全局可读 + 敏感路径拒绝”模型，以保证工具兼容性。
MACOS_MINIMAL_READABLE_PATHS = [
    "/bin",
    "/sbin",
    "/usr/bin",
    "/usr/sbin",
    "/usr/libexec",
    "/usr/lib",
    "/usr/share",
    "/usr/local/lib",
    "/opt/homebrew/lib",
    "/lib",
    "/etc",
    "/private/etc",
    "/tmp",
    "/private/tmp",
    "/var/tmp",
    "/private/var/tmp",
    "/var",
    "/private/var",
    "/dev",
    "/System/Library/Frameworks",
    "/System/Library/PrivateFrameworks",
    "/System/Library/SubFrameworks",
    "/System/Library/CoreServices",
    "/System/Library/Extensions",
    "/Library/Apple",
    "/Library/Preferences",
    "/Library/Filesystems/NetFSPlugins",
    "/Applications",
]


def _expand_path(raw: str) -> Optional[str]:
    """展开路径中的 ~ 并返回绝对路径；无法展开时返回 None。"""
    if not raw:
        return None
    try:
        expanded = str(Path(raw).expanduser().resolve())
    except Exception:
        return None
    return expanded


def _build_macos_read_rules(paths: List[str]) -> str:
    """把路径列表转成 (allow file-read* (subpath ...)) 规则段。"""
    rules: List[str] = []
    seen: set[str] = set()
    for raw in paths:
        expanded = _expand_path(raw)
        if expanded and expanded not in seen:
            seen.add(expanded)
            rules.append(f'(allow file-read* (subpath "{expanded}"))')
    return "\n".join(rules)


def _build_macos_deny_rules(paths: List[str]) -> str:
    """把路径列表转成 (deny file-read* (subpath ...)) 规则段。"""
    rules: List[str] = []
    seen: set[str] = set()
    for raw in paths:
        expanded = _expand_path(raw)
        if expanded and expanded not in seen:
            seen.add(expanded)
            rules.append(f'(deny file-read* (subpath "{expanded}"))')
    return "\n".join(rules)


def _build_macos_deny_regex_rules(patterns: List[str]) -> str:
    """把正则列表转成 (deny file-read* (regex #"...")) 规则段。"""
    rules: List[str] = []
    for pattern in patterns:
        rules.append(f'(deny file-read* (regex #"{pattern}"))')
    return "\n".join(rules)


# 宿主机网络权限档位
NETWORK_PERMISSION_RESTRICTED = "restricted"   # macOS: 仅本地回环；Linux/Windows: 暂不隔离
NETWORK_PERMISSION_FULL = "full"               # 完全开放
NETWORK_PERMISSION_NONE = "none"               # 完全禁止网络（后端保留）
_NETWORK_PERMISSION_VALUES = {
    NETWORK_PERMISSION_RESTRICTED,
    NETWORK_PERMISSION_FULL,
    NETWORK_PERMISSION_NONE,
}


def _truthy(name: str, default: str = "1") -> bool:
    return os.environ.get(name, default).strip().lower() not in {"0", "false", "no", "off"}


def host_sandbox_enabled() -> bool:
    return _truthy("HOST_SANDBOX_ENABLED", "1")


def _normalize_network_permission(value: Optional[str]) -> str:
    """归一化网络权限值，非法值回退为 restricted。"""
    normalized = str(value or "").strip().lower()
    if normalized in _NETWORK_PERMISSION_VALUES:
        return normalized
    return NETWORK_PERMISSION_RESTRICTED


def _build_macos_network_policy(network_permission: str) -> str:
    """根据网络权限档位生成 macOS sandbox-exec 网络规则片段。"""
    permission = _normalize_network_permission(network_permission)
    if permission == NETWORK_PERMISSION_NONE:
        return ""
    if permission == NETWORK_PERMISSION_FULL:
        return "(allow network-outbound)\n(allow network-inbound)\n"
    # restricted: 仅允许本地回环出站（涵盖 127.0.0.1 / ::1 的实际效果）
    return '(allow network-outbound (remote ip "localhost:*"))\n'


def build_host_sandbox_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    system = platform.system()
    if system == "Darwin":
        return _build_macos_plan(command, work_path, env, network_permission)
    if system == "Linux":
        return _build_linux_plan(command, work_path, env, network_permission)
    if system == "Windows":
        return _build_windows_plan(command, work_path, env, network_permission)
    raise HostSandboxError(f"不支持的宿主机系统: {system}")


def build_host_sandbox_readonly_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    system = platform.system()
    if system == "Darwin":
        return _build_macos_readonly_plan(command, work_path, env, network_permission)
    if system == "Linux":
        return _build_linux_readonly_plan(command, work_path, env, network_permission)
    if system == "Windows":
        return _build_windows_plan(command, work_path, env, network_permission)
    raise HostSandboxError(f"不支持的宿主机系统: {system}")


def build_host_sandbox_shell_plan(
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    system = platform.system()
    if system == "Darwin":
        return _build_macos_shell_plan(work_path, env, network_permission)
    if system == "Linux":
        return _build_linux_shell_plan(work_path, env, network_permission)
    if system == "Windows":
        return _build_windows_shell_plan(work_path, env, network_permission)
    raise HostSandboxError(f"不支持的宿主机系统: {system}")


def _build_macos_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    sandbox_exec = shutil.which("sandbox-exec")
    if not sandbox_exec:
        raise HostSandboxError("macOS 未找到 sandbox-exec，拒绝执行宿主机命令。")
    profile = _macos_profile_for_workspace(work_path, network_permission)
    cmd = [sandbox_exec, "-p", profile, "/bin/bash", "-lc", command]
    return SandboxPlan(command=cmd, env=env, cwd=str(work_path))


def _build_macos_readonly_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    sandbox_exec = shutil.which("sandbox-exec")
    if not sandbox_exec:
        raise HostSandboxError("macOS 未找到 sandbox-exec，拒绝执行宿主机命令。")
    network_policy = _build_macos_network_policy(network_permission)
    workspace = str(work_path.resolve())

    # 只读沙箱：全局可读 + 敏感路径/文件拒绝，写权限仅 /dev/null。
    # 工作区在 deny 规则之后再显式 allow，保证“工作区内不受沙箱影响”。
    deny_rules = _build_macos_deny_rules(get_macos_deny_read_paths())
    regex_rules = _build_macos_deny_regex_rules(get_macos_deny_read_regexes())
    if regex_rules:
        deny_rules += "\n" + regex_rules

    profile = (
        '(version 1)\n'
        '(deny default)\n'
        '(allow sysctl-read)\n'
        '(allow process*)\n'
        f'{network_policy}'
        '(allow file-read*)\n'
        f'{deny_rules}\n'
        f'(allow file-read* (subpath "{workspace}"))\n'
        '(allow file-write* (literal "/dev/null"))'
    )
    cmd = [sandbox_exec, "-p", profile, "/bin/bash", "-lc", command]
    return SandboxPlan(command=cmd, env=env, cwd=str(work_path))


def _build_macos_shell_plan(
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    sandbox_exec = shutil.which("sandbox-exec")
    if not sandbox_exec:
        raise HostSandboxError("macOS 未找到 sandbox-exec，拒绝启动宿主机沙箱终端。")
    profile = _macos_profile_for_workspace(work_path, network_permission)
    cmd = [sandbox_exec, "-p", profile, "/bin/bash", "-i"]
    return SandboxPlan(command=cmd, env=env, cwd=str(work_path))


def _macos_profile_for_workspace(
    work_path: Path,
    network_permission: Optional[str] = None,
) -> str:
    workspace = str(work_path.resolve())
    writable_paths = [workspace, "/tmp", "/private/tmp", "/dev/null"]
    for raw in get_macos_writable_paths():
        try:
            expanded = str(Path(raw).expanduser().resolve())
        except Exception:
            continue
        if expanded not in writable_paths:
            writable_paths.append(expanded)
    write_rules: list[str] = []
    for entry in writable_paths:
        if entry == "/dev/null":
            write_rules.append('(literal "/dev/null")')
        else:
            write_rules.append(f'(subpath "{entry}")')
    write_expr = " ".join(write_rules)
    network_policy = _build_macos_network_policy(network_permission)
    workspace = str(work_path.resolve())
    deny_rules = _build_macos_deny_rules(get_macos_deny_read_paths())
    regex_rules = _build_macos_deny_regex_rules(get_macos_deny_read_regexes())
    if regex_rules:
        deny_rules += "\n" + regex_rules
    return (
        '(version 1)\n'
        '(deny default)\n'
        '(allow sysctl-read)\n'
        '(allow process*)\n'
        f'{network_policy}'
        '(allow file-read*)\n'
        f'{deny_rules}\n'
        f'(allow file-read* (subpath "{workspace}"))\n'
        f'(allow file-write* {write_expr})'
    )


def _build_linux_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    bwrap = shutil.which("bwrap")
    if not bwrap:
        raise HostSandboxError("Linux 未找到 bubblewrap(bwrap)，拒绝执行宿主机命令。")

    seccomp_bpf = os.environ.get("HOST_SANDBOX_LINUX_SECCOMP_BPF", "").strip()
    if not seccomp_bpf:
        raise HostSandboxError("Linux 缺少 HOST_SANDBOX_LINUX_SECCOMP_BPF，拒绝执行宿主机命令。")
    seccomp_path = Path(seccomp_bpf).expanduser().resolve()
    if not seccomp_path.exists():
        raise HostSandboxError(f"seccomp BPF 文件不存在: {seccomp_path}")
    shell_cmd = ["/bin/bash", "-lc", command]
    # network_permission 暂不参与 Linux 构建，保持现有 --share-net 行为
    return _build_linux_common_plan(work_path, env, shell_cmd, seccomp_path)


def _build_linux_readonly_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    bwrap = shutil.which("bwrap")
    if not bwrap:
        raise HostSandboxError("Linux 未找到 bubblewrap(bwrap)，拒绝执行宿主机命令。")
    seccomp_bpf = os.environ.get("HOST_SANDBOX_LINUX_SECCOMP_BPF", "").strip()
    if not seccomp_bpf:
        raise HostSandboxError("Linux 缺少 HOST_SANDBOX_LINUX_SECCOMP_BPF，拒绝执行宿主机命令。")
    seccomp_path = Path(seccomp_bpf).expanduser().resolve()
    if not seccomp_path.exists():
        raise HostSandboxError(f"seccomp BPF 文件不存在: {seccomp_path}")
    shell_cmd = ["/bin/bash", "-lc", command]
    return _build_linux_common_plan(work_path, env, shell_cmd, seccomp_path, readonly=True)


def _build_linux_shell_plan(
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    bwrap = shutil.which("bwrap")
    if not bwrap:
        raise HostSandboxError("Linux 未找到 bubblewrap(bwrap)，拒绝启动宿主机沙箱终端。")
    seccomp_bpf = os.environ.get("HOST_SANDBOX_LINUX_SECCOMP_BPF", "").strip()
    if not seccomp_bpf:
        raise HostSandboxError("Linux 缺少 HOST_SANDBOX_LINUX_SECCOMP_BPF，拒绝启动宿主机沙箱终端。")
    seccomp_path = Path(seccomp_bpf).expanduser().resolve()
    if not seccomp_path.exists():
        raise HostSandboxError(f"seccomp BPF 文件不存在: {seccomp_path}")
    shell_cmd = ["/bin/bash", "-i"]
    return _build_linux_common_plan(work_path, env, shell_cmd, seccomp_path)


def _build_linux_common_plan(
    work_path: Path,
    env: Dict[str, str],
    shell_cmd: List[str],
    seccomp_path: Path,
    readonly: bool = False,
) -> SandboxPlan:
    bwrap = shutil.which("bwrap")
    if not bwrap:
        raise HostSandboxError("Linux 未找到 bubblewrap(bwrap)。")
    sandbox_root = str(work_path.resolve())
    cmd: List[str] = [
        bwrap,
        "--die-with-parent",
        "--new-session",
        "--unshare-all",
        "--share-net",
        "--ro-bind",
        "/",
        "/",
    ]
    if readonly:
        cmd.extend(["--ro-bind", sandbox_root, sandbox_root])
    else:
        cmd.extend(["--bind", sandbox_root, sandbox_root])
    cmd.extend([
        "--chdir",
        sandbox_root,
        "--proc",
        "/proc",
        "--dev",
        "/dev",
        "--tmpfs",
        "/tmp",
        "--seccomp",
        "__SECCOMP_FD__",
        *shell_cmd,
    ])
    return SandboxPlan(command=cmd, env=env, cwd=sandbox_root, seccomp_bpf_path=str(seccomp_path))


def _build_windows_plan(
    command: str,
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    wsl = shutil.which("wsl.exe")
    if not wsl:
        raise HostSandboxError("Windows 未找到 wsl.exe（WSL2），拒绝执行宿主机命令。")
    work = shlex.quote(str(work_path))
    shell = f"cd {work} && {command}"
    cmd = [wsl, "bash", "-lc", shell]
    return SandboxPlan(command=cmd, env=env, cwd=str(work_path))


def _build_windows_shell_plan(
    work_path: Path,
    env: Dict[str, str],
    network_permission: Optional[str] = None,
) -> SandboxPlan:
    wsl = shutil.which("wsl.exe")
    if not wsl:
        raise HostSandboxError("Windows 未找到 wsl.exe（WSL2），拒绝启动宿主机沙箱终端。")
    work = shlex.quote(str(work_path))
    shell = f"cd {work} && exec bash -i"
    cmd = [wsl, "bash", "-lc", shell]
    return SandboxPlan(command=cmd, env=env, cwd=str(work_path))