"""docker 容器内只读执行：非特权执行角色（uid/gid）。 设计（2026-08-30 云端实测验证，详见项目记忆 sandbox_readonly_reform_research）： - 容器主进程与可写执行保持 root；sandbox_write_access=False 的执行通道 （run_command 前台/后台、只读语境下创建的持久终端）统一改用非特权 uid （默认 10001:10001，可用环境变量覆盖）。 - 强制力来自内核 DAC：工作区 bind 挂载的文件属主是宿主机 root，非属主且 无 o+w → 写/删/chmod/umount 一律 EACCES/EPERM；600 权限的 .env 等敏感 文件天然不可读。此前的「命令文本特征识别」降级为审批决策的启发式， 不再是 docker 只读的安全边界。 - 不依赖镜像中存在该用户（数字 uid 直接生效），旧镜像直接受益； docker/terminal.Dockerfile 同步创建 agent 用户与 git safe.directory， 供未来构建的镜像使用。 - 逃逸门槛：需要先提权（setuid 漏洞/内核漏洞），远高于嵌套 namespace 方案的 umount 逃逸；云端默认 seccomp 会拦 unshare(CLONE_NEWUSER)， 嵌套方案已被本方案取代。 - 已知边界：macOS Docker Desktop 的 virtiofs（fakeowner）不按 uid 执行 权限检查，本机制仅在 Linux 宿主机（云端/ Linux 桌面）生效，属预期差异； 持久终端在 approval/auto_approval 模式下同为只读身份，写入命令请走 run_command（审批通过后以 root 重跑）。 Landlock 加固（2026-09 起）： - 纯 DAC 的残留漏洞：工作区内历史遗留的 world-writable（777/o+w）路径 对只读 uid 仍开放写权限。为此在只读执行时额外用 Landlock 给进程套上 「工作区写类操作全拒」的内核域（最终权限 = DAC ∩ Landlock），封死该洞。 - launcher 为本目录 landlock_launcher.py，首次只读执行时 docker cp 进容器 并以只读身份自检（本进程+子进程写工作区都必须被拒），通过后才启用； 任何一步失败都静默降级为纯 DAC，并以 warning 日志标注 enforcement level。 - 可用环境变量 DOCKER_READONLY_LANDLOCK=0 整体停用（运维逃生门）。 - 语义刻意对齐纯 DAC 现状：仅工作区写类操作被拒，/tmp 等其余路径写、 全部读/执行行为不变（读保护仍由 DAC 承担）。 """ from __future__ import annotations import logging import os import shutil import subprocess import threading from pathlib import Path from typing import Dict, List, Optional, Tuple logger = logging.getLogger("docker_readonly_exec") LANDLOCK_LAUNCHER_SRC = Path(__file__).resolve().with_name("landlock_launcher.py") LANDLOCK_LAUNCHER_CONTAINER_PATH = "/opt/astrion-landlock/launcher.py" # 容器名 -> "available" / "unavailable:"；None 表示尚未探测 _landlock_states: Dict[str, str] = {} _landlock_states_lock = threading.Lock() def docker_readonly_uid_gid() -> Tuple[str, str]: """只读执行身份的 uid/gid（默认 10001:10001，环境变量可覆盖）。 选取原则：避开宿主机工作区文件属主 uid（否则会被 DAC 认成主人而放行）， 10001 为冷门值；如部署环境文件属主恰为该值，用环境变量改。 """ uid = os.environ.get("DOCKER_READONLY_EXEC_UID", "10001").strip() or "10001" gid = os.environ.get("DOCKER_READONLY_EXEC_GID", uid).strip() or uid return uid, gid def docker_readonly_exec_args() -> List[str]: """构造只读执行的 docker exec 前缀参数（紧跟在 "exec" 之后）。 含： - ``-u uid:gid``：非特权执行身份（内核 DAC 强制只读的核心）； - ``HOME=/tmp``：uid 在 /etc/passwd 无条目时提供可写 HOME； - git safe.directory 修复：工作区属主是 root，非属主 git 会报 "detected dubious ownership"，用 env 方式注入，不依赖镜像内 gitconfig。 """ uid, gid = docker_readonly_uid_gid() return [ "-u", f"{uid}:{gid}", "-e", "HOME=/tmp", "-e", "GIT_CONFIG_COUNT=1", "-e", "GIT_CONFIG_KEY_0=safe.directory", "-e", "GIT_CONFIG_VALUE_0=*", ] def _landlock_enabled() -> bool: return os.environ.get("DOCKER_READONLY_LANDLOCK", "1").strip().lower() not in ( "0", "false", "off", "no", ) def _docker_run(docker_bin: str, args: List[str], timeout: int = 20) -> subprocess.CompletedProcess: return subprocess.run( [docker_bin] + args, capture_output=True, timeout=timeout, ) def _deploy_and_selftest(container_name: str, mount_path: str, docker_bin: str) -> Tuple[bool, str]: """把 launcher 部署进容器并以只读身份自检。返回 (是否可用, 失败原因)。""" mount_path = (mount_path or "/workspace").rstrip("/") or "/" try: # 1. 容器内 python3 可用性 r = _docker_run(docker_bin, ["exec", container_name, "sh", "-c", "command -v python3"], timeout=10) if r.returncode != 0: return False, "no-python3-in-container" # 2. 部署 launcher 文件 r = _docker_run(docker_bin, ["exec", container_name, "mkdir", "-p", os.path.dirname(LANDLOCK_LAUNCHER_CONTAINER_PATH)], timeout=10) if r.returncode != 0: return False, f"mkdir-failed:{r.stderr.decode(errors='replace')[:200]}" r = _docker_run(docker_bin, ["cp", str(LANDLOCK_LAUNCHER_SRC), f"{container_name}:{LANDLOCK_LAUNCHER_CONTAINER_PATH}"], timeout=30) if r.returncode != 0: return False, f"docker-cp-failed:{r.stderr.decode(errors='replace')[:200]}" # 3. 以只读身份自检（与真实只读执行同一 uid/环境） r = _docker_run( docker_bin, ["exec", *docker_readonly_exec_args(), container_name, "python3", LANDLOCK_LAUNCHER_CONTAINER_PATH, "--selftest", "--ro", mount_path], timeout=30, ) out = (r.stdout + r.stderr).decode(errors="replace") # 4. 清理自检意外成功时的残留（此时说明 Landlock 未生效，但仍要扫尾） try: _docker_run(docker_bin, ["exec", container_name, "rm", "-f", f"{mount_path}/.landlock_selftest_probe"], timeout=10) except Exception: pass if r.returncode == 0 and "LANDLOCK_SELFTEST_OK" in out: return True, "" return False, f"selftest-failed:{out.strip()[:300]}" except subprocess.TimeoutExpired: return False, "probe-timeout" except Exception as e: # noqa: BLE001 - 部署探测必须兜底为降级而非异常 return False, f"{type(e).__name__}:{e}" def ensure_landlock_ready(container_name: str, mount_path: str, docker_bin: Optional[str] = None) -> bool: """确保容器内 Landlock 只读域可用（部署+自检，按容器缓存结果）。 任何失败都返回 False（调用方回退纯 DAC），不会抛出。 """ if not _landlock_enabled(): return False with _landlock_states_lock: state = _landlock_states.get(container_name) if state is not None: return state == "available" docker_bin = docker_bin or shutil.which("docker") or "docker" ok, reason = _deploy_and_selftest(container_name, mount_path, docker_bin) with _landlock_states_lock: _landlock_states[container_name] = "available" if ok else f"unavailable:{reason}" if ok: logger.info("landlock readonly ready: container=%s mount=%s", container_name, mount_path) else: logger.warning( "landlock unavailable: container=%s reason=%s — " "readonly enforcement falls back to DAC-only (world-writable " "paths remain writable by the readonly uid)", container_name, reason, ) return ok def docker_readonly_wrap_inner(container_name: str, mount_path: str, inner_cmd: List[str], docker_bin: Optional[str] = None) -> List[str]: """只读执行时包装容器内命令：Landlock 可用则经 launcher 进入只读域。 可用： [python3, launcher, --ro, mount, --] + inner_cmd 不可用：原样返回 inner_cmd（纯 DAC 降级）。 """ if ensure_landlock_ready(container_name, mount_path, docker_bin): mount = (mount_path or "/workspace").rstrip("/") or "/" return ["python3", LANDLOCK_LAUNCHER_CONTAINER_PATH, "--ro", mount, "--", *inner_cmd] return list(inner_cmd)