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


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


class HostSandboxError(RuntimeError):
    pass


# 宿主机网络权限档位
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)
    profile = (
        '(version 1) '
        '(deny default) '
        '(allow sysctl-read) '
        '(allow process*) '
        f'{network_policy}'
        '(allow file-read*) '
        '(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)
    return (
        '(version 1) '
        '(deny default) '
        '(allow sysctl-read) '
        '(allow process*) '
        f'{network_policy}'
        '(allow file-read*) '
        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))