Multi-tenant isolation for Docker tasks (per-project mounts + networks + drop caps)
Container-side isolation was project-blind: every task mounted
the shared /app/work (all worktrees + host-state), the shared
codex-home (all thread history), the shared repos dir and the
shared test-repo, and all containers shared the default bridge
network. With --tenant-isolation (Docker only; Local runs on
the host under one user and has no container boundary to
enforce), each project gets its own host dirs under
tenants_root/<project>/{work,codex-home,test-repo,<project>}:
- mounts: container paths stay identical (/app/work,
codex-home, repos/<pid>, test-repo) so host.py config needs
no change; only the host mount sources are per-project.
Static --docker-extra-mounts are replaced per task (not
appended), so a shared mount cannot leak back in.
- codex-home seeds ONLY the template's config.toml (relay
endpoint/provider names, no secret values); thread_history
is never copied (it holds other tasks' prompts/results).
test-repo is a per-project copy of the fixture (worktree
creation needs a writable repo; a shared writable fixture
would let one tenant poison the next task's base code).
- repos: repo_url checkouts go to the tenant dir and task
repo is set to the container path; no-repo_url tasks get
the tenant test-repo copy instead of the shared fixture.
- network: one Docker network per project (codex-<pid>,
created idempotently, already-exists is OK); network
creation failure fails the task instead of silently
falling back to the default network (a silent downgrade
would make isolation fail quietly).
- hardening for all Docker runs: --cap-drop=ALL and
--security-opt=no-new-privileges (non-root user was already
the case); verified live by a good-model task, not assumed.
Honest limit, kept in README: the inner codex sandbox is
still danger-full-access (host kernel blocks user namespaces,
bwrap unusable), so the boundary is container + mounts +
network, not the inner sandbox.
Tests: 7 tenant tests (seeding incl. no-history, per-project
mounts, network sanitization, per-call override, ensure
network ok/fail). Full suites pass.
This commit is contained in:
1 parent
b74deffb26
commit
03a98124db
5 files changed
+452
-21
No files matched your search
@@ -479,6 +479,18 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
|
|||||||
- 清理与取消:`docker run --rm`,正常失败也自动删容器;超时
|
- 清理与取消:`docker run --rm`,正常失败也自动删容器;超时
|
||||||
时先 docker kill 再 docker rm -f;worker 取消时按任务派生的
|
时先 docker kill 再 docker rm -f;worker 取消时按任务派生的
|
||||||
容器名 docker kill,不只改数据库状态。
|
容器名 docker kill,不只改数据库状态。
|
||||||
|
- 多租户隔离(`--tenant-isolation`,仅 Docker 执行器):开启
|
||||||
|
后每个项目用独立的宿主目录 `tenants_root/<项目>/work`、
|
||||||
|
`codex-home`、test-repo 副本与 repo 检出,容器内路径与
|
||||||
|
非租户模式完全一致(`/app/work`、codex-home、repos/<项目>、
|
||||||
|
test-repo),只是宿主挂载源按项目分开——容器内看不到其它
|
||||||
|
项目的任何文件;同时每个项目进独立的 Docker 网络
|
||||||
|
`codex-<项目>`(容器跨项目网络隔离),容器统一加
|
||||||
|
`--cap-drop=ALL --security-opt=no-new-privileges`。codex-home
|
||||||
|
只播种模板的 config.toml(relay 端点与 provider 名,不含
|
||||||
|
密钥值),绝不复制 thread_history。注意:内层 codex 沙箱仍
|
||||||
|
是 danger-full-access(主机内核禁 user namespace,bwrap 不可
|
||||||
|
用),隔离边界是容器 + 挂载 + 网络三层,不是内层沙箱。
|
||||||
- 网络白名单的诚实边界:docker run 默认 bridge 网络,本执行
|
- 网络白名单的诚实边界:docker run 默认 bridge 网络,本执行
|
||||||
器不做 egress 白名单——真正的“只放行模型 API”需要宿主防火
|
器不做 egress 白名单——真正的“只放行模型 API”需要宿主防火
|
||||||
墙或代理规则,那是部署项,不是本轮代码假装做到的,不要宣称
|
墙或代理规则,那是部署项,不是本轮代码假装做到的,不要宣称
|
||||||
|
|||||||
+65
-12
@@ -369,13 +369,24 @@ class DockerExecutor(Executor):
|
|||||||
"Missing environment variable: %s" % self.env_key)
|
"Missing environment variable: %s" % self.env_key)
|
||||||
|
|
||||||
# -- 命令构造 ------------------------------------------------------
|
# -- 命令构造 ------------------------------------------------------
|
||||||
def build_run_command(self, task, inner_cmd):
|
def build_run_command(self, task, inner_cmd, extra_mounts=None,
|
||||||
|
network=None):
|
||||||
"""构造完整的 docker run 命令(纯函数,可单测断言)。
|
"""构造完整的 docker run 命令(纯函数,可单测断言)。
|
||||||
|
|
||||||
顺序固定:docker run --rm --name <容器名> --user <用户>
|
顺序固定:docker run --rm --name <容器名> --user <用户>
|
||||||
--cpus <核数> --memory <内存> -v <worktree>:/work
|
--cpus <核数> --memory <内存> [--cap-drop=ALL
|
||||||
-v <代码目录>:/app:ro -w /work [-e <env_key名>]
|
--security-opt=no-new-privileges] [--network <网络>]
|
||||||
<镜像> <容器内命令...>。-e 只传名字,不带 =值。
|
-v <worktree>:/work -v <代码目录>:/app:ro -w /work
|
||||||
|
[-e <env_key名>] <镜像> <容器内命令...>。-e 只传名字,
|
||||||
|
不带 =值。
|
||||||
|
extra_mounts 给定时替换构造时的静态挂载(多租户隔离
|
||||||
|
用:worker 按项目算出挂载后每次调用传入);不给时用
|
||||||
|
静态挂载。network 给定时容器进该 Docker 网络(多租户
|
||||||
|
用项目独立网络,容器之间默认 bridge 上互相可见的问题
|
||||||
|
由此解决);不给时用 docker 默认网络。
|
||||||
|
降权:容器默认已是非 root 用户,再加 --cap-drop=ALL
|
||||||
|
与 no-new-privileges(提权类工具调用会被挡下,codex
|
||||||
|
的常规文件/git 操作不受影响,见多租户实测)。
|
||||||
"""
|
"""
|
||||||
self._check_docker()
|
self._check_docker()
|
||||||
self._check_key()
|
self._check_key()
|
||||||
@@ -384,13 +395,19 @@ class DockerExecutor(Executor):
|
|||||||
"--user", self.user,
|
"--user", self.user,
|
||||||
"--cpus", self.cpus,
|
"--cpus", self.cpus,
|
||||||
"--memory", self.memory,
|
"--memory", self.memory,
|
||||||
"-v", "%s:/work" % self._worktree_for(task),
|
"--cap-drop=ALL",
|
||||||
|
"--security-opt=no-new-privileges"]
|
||||||
|
if network:
|
||||||
|
cmd += ["--network", network]
|
||||||
|
cmd += ["-v", "%s:/work" % self._worktree_for(task),
|
||||||
"-v", "%s:/app:ro" % self._app_dir(),
|
"-v", "%s:/app:ro" % self._app_dir(),
|
||||||
"-w", "/work"]
|
"-w", "/work"]
|
||||||
# 额外挂载放在基础挂载之后:/app/work 这类子路径挂载会
|
# 额外挂载放在基础挂载之后:/app/work 这类子路径挂载会
|
||||||
# 覆盖 /app:ro 的对应子路径,让容器内 host.py 能写
|
# 覆盖 /app:ro 的对应子路径,让容器内 host.py 能写
|
||||||
# work/ 和 host-state.json,同时代码主体仍只读。
|
# work/ 和 host-state.json,同时代码主体仍只读。
|
||||||
for host_p, cont_p, mode in self.extra_mounts:
|
mounts = self.extra_mounts if extra_mounts is None \
|
||||||
|
else self._parse_extra_mounts(extra_mounts)
|
||||||
|
for host_p, cont_p, mode in mounts:
|
||||||
cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)]
|
cmd += ["-v", "%s:%s:%s" % (host_p, cont_p, mode)]
|
||||||
if self.env_key:
|
if self.env_key:
|
||||||
cmd += ["-e", self.env_key]
|
cmd += ["-e", self.env_key]
|
||||||
@@ -500,18 +517,22 @@ class DockerExecutor(Executor):
|
|||||||
continue
|
continue
|
||||||
return ok
|
return ok
|
||||||
|
|
||||||
def _run_in_container(self, task, inner_cmd, timeout=None):
|
def _run_in_container(self, task, inner_cmd, timeout=None,
|
||||||
|
extra_mounts=None, network=None):
|
||||||
"""在容器里跑一轮内层命令,返回 LocalExecutor 同款结果 dict。
|
"""在容器里跑一轮内层命令,返回 LocalExecutor 同款结果 dict。
|
||||||
|
|
||||||
超时:runner 抛 subprocess.TimeoutExpired(或同名异常)
|
超时:runner 抛 subprocess.TimeoutExpired(或同名异常)
|
||||||
时先 kill+rm 容器,再返回 failed(stdout 截断 + timeout
|
时先 kill+rm 容器,再返回 failed(stdout 截断 + timeout
|
||||||
说明,returncode 为 None);其它 runner 异常也如实返回
|
说明,returncode 为 None);其它 runner 异常也如实返回
|
||||||
failed,不编造成功。timeout 只影响本次调用,不改实例。
|
failed,不编造成功。timeout 只影响本次调用,不改实例。
|
||||||
|
extra_mounts/network 只影响本次调用(多租户按项目传入)。
|
||||||
"""
|
"""
|
||||||
task_id = task.get("id") if isinstance(task, dict) else None
|
task_id = task.get("id") if isinstance(task, dict) else None
|
||||||
if task_id is None:
|
if task_id is None:
|
||||||
raise ValueError("任务缺少 id")
|
raise ValueError("任务缺少 id")
|
||||||
cmd = self.build_run_command(task, inner_cmd)
|
cmd = self.build_run_command(task, inner_cmd,
|
||||||
|
extra_mounts=extra_mounts,
|
||||||
|
network=network)
|
||||||
try:
|
try:
|
||||||
result = self._call_runner(cmd, timeout=timeout)
|
result = self._call_runner(cmd, timeout=timeout)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -536,7 +557,8 @@ class DockerExecutor(Executor):
|
|||||||
parsed["container"] = self.container_name(task)
|
parsed["container"] = self.container_name(task)
|
||||||
return parsed
|
return parsed
|
||||||
|
|
||||||
def dispatch_task(self, task, timeout=None) -> dict:
|
def dispatch_task(self, task, timeout=None, extra_mounts=None,
|
||||||
|
network=None) -> dict:
|
||||||
"""容器内调 host.py dispatch:建 host 任务并跑一轮 turn。"""
|
"""容器内调 host.py dispatch:建 host 任务并跑一轮 turn。"""
|
||||||
if not isinstance(task, dict) or task.get("id") is None:
|
if not isinstance(task, dict) or task.get("id") is None:
|
||||||
raise ValueError("任务缺少 id")
|
raise ValueError("任务缺少 id")
|
||||||
@@ -548,9 +570,12 @@ class DockerExecutor(Executor):
|
|||||||
repo = task.get("repo")
|
repo = task.get("repo")
|
||||||
if repo:
|
if repo:
|
||||||
inner += ["--repo", repo]
|
inner += ["--repo", repo]
|
||||||
return self._run_in_container(task, inner, timeout=timeout)
|
return self._run_in_container(task, inner, timeout=timeout,
|
||||||
|
extra_mounts=extra_mounts,
|
||||||
|
network=network)
|
||||||
|
|
||||||
def run_task(self, task, timeout=None) -> dict:
|
def run_task(self, task, timeout=None, extra_mounts=None,
|
||||||
|
network=None) -> dict:
|
||||||
"""容器内调 host.py run:在已有 host 任务上再跑一轮 turn。"""
|
"""容器内调 host.py run:在已有 host 任务上再跑一轮 turn。"""
|
||||||
if not isinstance(task, dict) or task.get("id") is None:
|
if not isinstance(task, dict) or task.get("id") is None:
|
||||||
raise ValueError("任务缺少 id")
|
raise ValueError("任务缺少 id")
|
||||||
@@ -559,7 +584,35 @@ class DockerExecutor(Executor):
|
|||||||
inner = self._inner_host_command(
|
inner = self._inner_host_command(
|
||||||
"run", "--task", self._host_name(task),
|
"run", "--task", self._host_name(task),
|
||||||
"--prompt", task["prompt"], model=task.get("model"))
|
"--prompt", task["prompt"], model=task.get("model"))
|
||||||
return self._run_in_container(task, inner, timeout=timeout)
|
return self._run_in_container(task, inner, timeout=timeout,
|
||||||
|
extra_mounts=extra_mounts,
|
||||||
|
network=network)
|
||||||
|
|
||||||
|
def ensure_network(self, name):
|
||||||
|
"""确保 Docker 网络存在:不存在就创建,已存在就忽略。
|
||||||
|
|
||||||
|
返回 True 表示网络可用(创建成功或本来就存在),
|
||||||
|
False 表示创建失败(调用方继续派发,docker run 时会
|
||||||
|
因网络不存在而明确失败,不在这里静默降级默认网络——
|
||||||
|
降级会让隔离悄悄失效,比失败更糟,所以这里只返回
|
||||||
|
布尔值,是否继续由调用方按失败处理:worker 在租户
|
||||||
|
模式下网络创建失败时直接让任务 failed,不跑容器)。
|
||||||
|
"""
|
||||||
|
if not name:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
result = self._call_runner(["docker", "network", "create", name],
|
||||||
|
timeout=30)
|
||||||
|
except Exception as exc:
|
||||||
|
return "already exists" in str(exc)
|
||||||
|
try:
|
||||||
|
if getattr(result, "returncode", 1) == 0:
|
||||||
|
return True
|
||||||
|
text = (getattr(result, "stderr", "") or "") + \
|
||||||
|
(getattr(result, "stdout", "") or "")
|
||||||
|
return "already exists" in text
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
def execute(self, task) -> dict:
|
def execute(self, task) -> dict:
|
||||||
"""只读执行:容器内调 host.py status,不直接跑模型回合。
|
"""只读执行:容器内调 host.py status,不直接跑模型回合。
|
||||||
|
|||||||
@@ -0,0 +1,169 @@
|
|||||||
|
"""多租户隔离单测。
|
||||||
|
|
||||||
|
只用临时目录与假 runner,不跑真实 Docker、不联网。
|
||||||
|
覆盖:租户目录播种(codex-home 只取 config.toml,绝不
|
||||||
|
复制 thread_history)、挂载按项目分开、网络名安全化、
|
||||||
|
执行器 per-call 挂载/网络覆盖、网络创建失败不降级。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
from types import SimpleNamespace
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
try:
|
||||||
|
from cloud.sandbox import DockerExecutor
|
||||||
|
from cloud.server import init_db
|
||||||
|
from cloud.worker import Worker
|
||||||
|
except ImportError:
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||||
|
from cloud.sandbox import DockerExecutor
|
||||||
|
from cloud.server import init_db
|
||||||
|
from cloud.worker import Worker
|
||||||
|
|
||||||
|
|
||||||
|
def _ok_runner_factory(seen):
|
||||||
|
def fake(cmd, timeout=None):
|
||||||
|
seen.append(list(cmd))
|
||||||
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||||
|
return fake
|
||||||
|
|
||||||
|
|
||||||
|
class TenantDirsTestCase(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||||
|
init_db(self.db_path)
|
||||||
|
# 模板 home:config.toml + 绝不能复制的 thread_history
|
||||||
|
# 与假密钥文件;模板 repo:sample.py + .git。
|
||||||
|
self.template_home = os.path.join(self.tmp.name, "tpl-home")
|
||||||
|
os.makedirs(self.template_home)
|
||||||
|
open(os.path.join(self.template_home, "config.toml"),
|
||||||
|
"w").write('[model_provider.xcmapi]\nbase_url="https://xcmapi.org/v1"\n')
|
||||||
|
open(os.path.join(self.template_home, "thread_history_1.sqlite"),
|
||||||
|
"w").write("OTHER TENANT HISTORY")
|
||||||
|
open(os.path.join(self.template_home, "auth.json"),
|
||||||
|
"w").write("SECRET")
|
||||||
|
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
|
||||||
|
os.makedirs(os.path.join(self.template_repo, ".git"))
|
||||||
|
open(os.path.join(self.template_repo, "sample.py"),
|
||||||
|
"w").write("def add(a, b):\n return a + b\n")
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
self.tmp.cleanup()
|
||||||
|
|
||||||
|
def _worker(self, **kwargs):
|
||||||
|
kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
|
||||||
|
kwargs.setdefault("tenants_root",
|
||||||
|
os.path.join(self.tmp.name, "tenants"))
|
||||||
|
kwargs.setdefault("tenant_template_home", self.template_home)
|
||||||
|
kwargs.setdefault("tenant_template_repo", self.template_repo)
|
||||||
|
kwargs.setdefault("tenant_isolation", True)
|
||||||
|
return Worker(self.db_path, "/fake/host.py", **kwargs)
|
||||||
|
|
||||||
|
def test_dirs_seed_config_only_not_history(self):
|
||||||
|
worker = self._worker()
|
||||||
|
dirs = worker._tenant_dirs("projA")
|
||||||
|
home = dirs["codex_home"]
|
||||||
|
self.assertTrue(os.path.exists(os.path.join(home, "config.toml")))
|
||||||
|
self.assertFalse(os.path.exists(
|
||||||
|
os.path.join(home, "thread_history_1.sqlite")))
|
||||||
|
self.assertFalse(os.path.exists(os.path.join(home, "auth.json")))
|
||||||
|
# test-repo 副本有 sample.py,且是复制不是链接。
|
||||||
|
self.assertTrue(os.path.exists(
|
||||||
|
os.path.join(dirs["test_repo"], "sample.py")))
|
||||||
|
# 第二次调用不覆盖租户自己的改动。
|
||||||
|
marker = os.path.join(dirs["test_repo"], "mine.txt")
|
||||||
|
open(marker, "w").write("mine")
|
||||||
|
worker._tenant_dirs("projA")
|
||||||
|
self.assertTrue(os.path.exists(marker))
|
||||||
|
|
||||||
|
def test_mounts_are_per_project(self):
|
||||||
|
worker = self._worker()
|
||||||
|
mounts_a = worker._tenant_mounts("projA")
|
||||||
|
mounts_b = worker._tenant_mounts("projB")
|
||||||
|
# 容器内路径相同(/app/work 等),宿主源不同项目不同。
|
||||||
|
cont_a = [m.split(":")[1] for m in mounts_a]
|
||||||
|
cont_b = [m.split(":")[1] for m in mounts_b]
|
||||||
|
self.assertEqual(cont_a, cont_b)
|
||||||
|
self.assertIn("/app/work", cont_a)
|
||||||
|
self.assertIn("/var/lib/codex-app-server/codex-home", cont_a)
|
||||||
|
hosts_a = {m.split(":")[0] for m in mounts_a}
|
||||||
|
hosts_b = {m.split(":")[0] for m in mounts_b}
|
||||||
|
self.assertFalse(hosts_a & hosts_b)
|
||||||
|
# 租户 A 的 work 里写文件,B 的挂载里看不到(宿主侧)。
|
||||||
|
open(os.path.join(worker._tenant_dirs("projA")["work"],
|
||||||
|
"secret-a.txt"), "w").write("SECRET-A")
|
||||||
|
self.assertFalse(os.path.exists(os.path.join(
|
||||||
|
worker._tenant_dirs("projB")["work"], "secret-a.txt")))
|
||||||
|
|
||||||
|
def test_network_name_sanitized(self):
|
||||||
|
worker = self._worker()
|
||||||
|
self.assertEqual(worker._tenant_network("abc123"),
|
||||||
|
"codex-abc123")
|
||||||
|
self.assertEqual(worker._tenant_network("a b/c"),
|
||||||
|
"codex-a-b-c")
|
||||||
|
with self.assertRaises(ValueError):
|
||||||
|
worker._tenant_network("...")
|
||||||
|
|
||||||
|
def test_template_home_without_config_fails_loudly(self):
|
||||||
|
empty_home = os.path.join(self.tmp.name, "empty-home")
|
||||||
|
os.makedirs(empty_home)
|
||||||
|
worker = self._worker(tenant_template_home=empty_home)
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
worker._tenant_dirs("projA")
|
||||||
|
|
||||||
|
|
||||||
|
class DockerTenantCommandTestCase(unittest.TestCase):
|
||||||
|
def test_per_call_mounts_network_and_drop_caps(self):
|
||||||
|
seen = []
|
||||||
|
static_dir = tempfile.mkdtemp()
|
||||||
|
tenant_dir = tempfile.mkdtemp()
|
||||||
|
with mock.patch("cloud.sandbox.shutil.which",
|
||||||
|
return_value="/usr/bin/docker"):
|
||||||
|
ex = DockerExecutor(
|
||||||
|
"/fake/host.py",
|
||||||
|
runner=_ok_runner_factory(seen),
|
||||||
|
env={"XCM_API_KEY": "test-placeholder"},
|
||||||
|
extra_mounts=["%s:/app/work:rw" % static_dir])
|
||||||
|
cmd = ex.build_run_command(
|
||||||
|
{"id": "t1", "worktree": "/tmp/wt"},
|
||||||
|
["echo", "hi"],
|
||||||
|
extra_mounts=["%s:/app/work:rw" % tenant_dir],
|
||||||
|
network="codex-projA")
|
||||||
|
mounts = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-v"]
|
||||||
|
# per-call 挂载替换静态挂载,不是追加。
|
||||||
|
self.assertIn("%s:/app/work:rw" % tenant_dir, mounts)
|
||||||
|
self.assertNotIn("%s:/app/work:rw" % static_dir, mounts)
|
||||||
|
self.assertEqual(cmd[cmd.index("--network") + 1], "codex-projA")
|
||||||
|
self.assertIn("--cap-drop=ALL", cmd)
|
||||||
|
self.assertIn("--security-opt=no-new-privileges", cmd)
|
||||||
|
|
||||||
|
def test_ensure_network_already_exists_is_ok(self):
|
||||||
|
def exists_runner(cmd, timeout=None):
|
||||||
|
return SimpleNamespace(
|
||||||
|
returncode=1, stdout="",
|
||||||
|
stderr="Error response from daemon: network with name "
|
||||||
|
"codex-p already exists")
|
||||||
|
with mock.patch("cloud.sandbox.shutil.which",
|
||||||
|
return_value="/usr/bin/docker"):
|
||||||
|
ex = DockerExecutor("/fake/host.py", runner=exists_runner,
|
||||||
|
env={"XCM_API_KEY": "test-placeholder"})
|
||||||
|
self.assertTrue(ex.ensure_network("codex-p"))
|
||||||
|
|
||||||
|
def test_ensure_network_real_failure_is_false(self):
|
||||||
|
def fail_runner(cmd, timeout=None):
|
||||||
|
return SimpleNamespace(returncode=1, stdout="",
|
||||||
|
stderr="daemon not reachable")
|
||||||
|
with mock.patch("cloud.sandbox.shutil.which",
|
||||||
|
return_value="/usr/bin/docker"):
|
||||||
|
ex = DockerExecutor("/fake/host.py", runner=fail_runner,
|
||||||
|
env={"XCM_API_KEY": "test-placeholder"})
|
||||||
|
self.assertFalse(ex.ensure_network("codex-p"))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
+205
-8
@@ -233,7 +233,9 @@ class Worker:
|
|||||||
task_timeout=1800, max_retries=1,
|
task_timeout=1800, max_retries=1,
|
||||||
retry_backoff_seconds=5.0, repos_dir="./repos",
|
retry_backoff_seconds=5.0, repos_dir="./repos",
|
||||||
repos_runner=None, heartbeat_file=None,
|
repos_runner=None, heartbeat_file=None,
|
||||||
worker_id=None, capacity=1, fallback_models=None):
|
worker_id=None, capacity=1, fallback_models=None,
|
||||||
|
tenant_isolation=False, tenants_root=None,
|
||||||
|
tenant_template_home=None, tenant_template_repo=None):
|
||||||
self.db_path = str(db_path)
|
self.db_path = str(db_path)
|
||||||
self.host_py_path = host_py_path
|
self.host_py_path = host_py_path
|
||||||
self.config_path = config_path
|
self.config_path = config_path
|
||||||
@@ -298,6 +300,22 @@ class Worker:
|
|||||||
# repos_runner 可注入(单测用假 git runner),缺省走真实 git。
|
# repos_runner 可注入(单测用假 git runner),缺省走真实 git。
|
||||||
self.repos_dir = repos_dir
|
self.repos_dir = repos_dir
|
||||||
self.repos_runner = repos_runner
|
self.repos_runner = repos_runner
|
||||||
|
# 多租户隔离(仅 Docker 执行器有效,Local 执行器在本机
|
||||||
|
# 直接跑 host.py,同一用户下本来就没有容器边界,flag
|
||||||
|
# 对它不起作用):开启后每个项目用独立的宿主目录
|
||||||
|
# tenants_root/<project_id>/{work,codex-home,test-repo,
|
||||||
|
# <project_id>},容器内路径保持与原来完全一致(/app/work、
|
||||||
|
# /var/lib/.../codex-home 等),只是宿主侧的挂载源按
|
||||||
|
# 项目分开——容器内看不到其它项目的任何文件。codex-home
|
||||||
|
# 只播种 config.toml(relay 配置,不含任何密钥值),
|
||||||
|
# 绝不复制 thread_history(那是其它任务的 prompt 与结果)。
|
||||||
|
self.tenant_isolation = tenant_isolation
|
||||||
|
self.tenants_root = tenants_root or \
|
||||||
|
"/var/lib/codex-app-server/tenants"
|
||||||
|
self.tenant_template_home = tenant_template_home or \
|
||||||
|
"/var/lib/codex-app-server/codex-home"
|
||||||
|
self.tenant_template_repo = tenant_template_repo or \
|
||||||
|
"/var/lib/codex-app-server/test-repo"
|
||||||
# 运维心跳:路径优先显式参数,其次环境变量,默认当前目录下
|
# 运维心跳:路径优先显式参数,其次环境变量,默认当前目录下
|
||||||
# 的 worker-heartbeat.json;主机路径不在代码里硬编码,主机
|
# 的 worker-heartbeat.json;主机路径不在代码里硬编码,主机
|
||||||
# 部署时用 --heartbeat-file 或环境变量传入(见 deploy/README)。
|
# 部署时用 --heartbeat-file 或环境变量传入(见 deploy/README)。
|
||||||
@@ -856,6 +874,107 @@ class Worker:
|
|||||||
"stdout": out or "", "stderr": err or "",
|
"stdout": out or "", "stderr": err or "",
|
||||||
"returncode": proc.returncode, "host_status": None}
|
"returncode": proc.returncode, "host_status": None}
|
||||||
|
|
||||||
|
# -- 多租户隔离 ----------------------------------------------------
|
||||||
|
def _tenant_safe_name(self, project_id):
|
||||||
|
"""项目 id 的安全目录名:与 repos.checkout_dir 同规则。"""
|
||||||
|
if project_id is None:
|
||||||
|
raise ValueError("project_id 缺失,无法确定租户目录")
|
||||||
|
raw = str(project_id)
|
||||||
|
safe = re.sub(r"[^A-Za-z0-9_.-]", "-", raw).strip(".")
|
||||||
|
if not safe or safe in (".", ".."):
|
||||||
|
raise ValueError("project_id 无法映射为安全的租户目录名: %r"
|
||||||
|
% raw)
|
||||||
|
return safe
|
||||||
|
|
||||||
|
def _tenant_dirs(self, project_id):
|
||||||
|
"""某项目的租户宿主目录(不存在时创建并播种)。
|
||||||
|
|
||||||
|
返回 dict:work / codex_home / repo_checkout / test_repo
|
||||||
|
(均为宿主绝对路径)。播种规则:
|
||||||
|
- codex-home:只复制模板的 config.toml(relay 端点与
|
||||||
|
provider 名,不含密钥值);模板缺 config.toml 时直接
|
||||||
|
报错——空 home 会让 codex 去 api.openai.com 且无 auth,
|
||||||
|
绝不能静默放行(见 PR #18 的教训)。
|
||||||
|
- test-repo:整目录复制模板(它是 sample.py 夹具,含
|
||||||
|
.git 历史也无妨,本来就是公开的起始代码)。
|
||||||
|
- work / repo_checkout:只建目录。
|
||||||
|
并发下两个同项目任务同时播种是安全的:mkdir 用
|
||||||
|
exist_ok,复制前判存在且复制内容相同(模板),即使
|
||||||
|
race 到重复复制也不会写坏租户自己的数据(租户数据
|
||||||
|
在复制完成后才产生)。
|
||||||
|
"""
|
||||||
|
import shutil
|
||||||
|
|
||||||
|
safe = self._tenant_safe_name(project_id)
|
||||||
|
base = os.path.join(self.tenants_root, safe)
|
||||||
|
dirs = {
|
||||||
|
"work": os.path.join(base, "work"),
|
||||||
|
"codex_home": os.path.join(base, "codex-home"),
|
||||||
|
"repo_checkout": os.path.join(base, safe),
|
||||||
|
"test_repo": os.path.join(base, "test-repo"),
|
||||||
|
}
|
||||||
|
# 注意绝不 mkdir repo_checkout:repo_url 任务的检出由
|
||||||
|
# resolve_repo 负责 clone,空目录先占位会导致它误判
|
||||||
|
# “目录已存在但不是 git 仓库”而拒绝 clone;_tenant_mounts
|
||||||
|
# 里按该目录是否存在决定是否挂载它(不存在说明任务
|
||||||
|
# 不用它,走 test-repo)。
|
||||||
|
for key in ("work", "codex_home", "test_repo"):
|
||||||
|
os.makedirs(dirs[key], exist_ok=True)
|
||||||
|
home_config = os.path.join(dirs["codex_home"], "config.toml")
|
||||||
|
if not os.path.exists(home_config):
|
||||||
|
template = os.path.join(self.tenant_template_home,
|
||||||
|
"config.toml")
|
||||||
|
if not os.path.exists(template):
|
||||||
|
raise RuntimeError(
|
||||||
|
"租户 codex-home 模板缺 config.toml:%s(空 home"
|
||||||
|
" 会让 codex 去 api.openai.com 且无 auth,拒绝"
|
||||||
|
" 派发)" % template)
|
||||||
|
shutil.copyfile(template, home_config)
|
||||||
|
if not os.path.exists(os.path.join(dirs["test_repo"], ".git")) \
|
||||||
|
and not os.path.exists(dirs["test_repo"] + "/sample.py"):
|
||||||
|
# 目标为空目录(刚 mkdir 出来的)才播种;已有内容
|
||||||
|
# 时绝不覆盖(那是该项目自己的改动或上次播种结果)。
|
||||||
|
if not os.listdir(dirs["test_repo"]):
|
||||||
|
if not os.path.exists(self.tenant_template_repo):
|
||||||
|
raise RuntimeError(
|
||||||
|
"租户 test-repo 模板不存在:%s"
|
||||||
|
% self.tenant_template_repo)
|
||||||
|
os.rmdir(dirs["test_repo"])
|
||||||
|
shutil.copytree(self.tenant_template_repo,
|
||||||
|
dirs["test_repo"])
|
||||||
|
return dirs
|
||||||
|
|
||||||
|
def _tenant_mounts(self, project_id):
|
||||||
|
"""某项目任务的 Docker 挂载列表(字符串,供执行器解析)。
|
||||||
|
|
||||||
|
容器内路径与非租户模式完全一致,只是宿主源按项目分开:
|
||||||
|
/app/work、codex-home、repos/<pid>、test-repo 全部指向
|
||||||
|
该项目的租户目录;代码目录 /app:ro 由执行器基础挂载
|
||||||
|
提供,不在这里重复。静态 --docker-extra-mounts 在租户
|
||||||
|
模式下被这份列表替换(否则共享挂载会继续泄漏)。
|
||||||
|
repos 挂载只在检出目录已存在时才加(repo_url 任务在
|
||||||
|
派发前已由 _apply_project_repo clone 好;test-repo 任务
|
||||||
|
不用它,不存在时硬挂载会被执行器的存在性检查拒绝)。
|
||||||
|
"""
|
||||||
|
dirs = self._tenant_dirs(project_id)
|
||||||
|
mounts = [
|
||||||
|
"%s:/app/work:rw" % dirs["work"],
|
||||||
|
"%s:/var/lib/codex-app-server/codex-home:rw"
|
||||||
|
% dirs["codex_home"],
|
||||||
|
"%s:/var/lib/codex-app-server/test-repo:rw"
|
||||||
|
% dirs["test_repo"],
|
||||||
|
]
|
||||||
|
if os.path.exists(dirs["repo_checkout"]):
|
||||||
|
mounts.append(
|
||||||
|
"%s:/var/lib/codex-app-server/repos/%s:rw"
|
||||||
|
% (dirs["repo_checkout"],
|
||||||
|
self._tenant_safe_name(project_id)))
|
||||||
|
return mounts
|
||||||
|
|
||||||
|
def _tenant_network(self, project_id):
|
||||||
|
"""某项目的 Docker 网络名(容器跨项目网络隔离用)。"""
|
||||||
|
return "codex-" + self._tenant_safe_name(project_id)
|
||||||
|
|
||||||
# -- 项目仓库 ------------------------------------------------------
|
# -- 项目仓库 ------------------------------------------------------
|
||||||
def _project_repo_url(self, project_id):
|
def _project_repo_url(self, project_id):
|
||||||
"""查任务所属项目的 repo_url;查不到项目、列缺失、表缺失时
|
"""查任务所属项目的 repo_url;查不到项目、列缺失、表缺失时
|
||||||
@@ -882,18 +1001,40 @@ class Worker:
|
|||||||
def _apply_project_repo(self, task):
|
def _apply_project_repo(self, task):
|
||||||
"""派发前按项目 repo_url 解析仓库,返回错误文本或 None。
|
"""派发前按项目 repo_url 解析仓库,返回错误文本或 None。
|
||||||
|
|
||||||
repo_url 为空:不动 task,返回 None(走原有默认 repo)。
|
repo_url 为空:非租户模式不动 task,返回 None(走原有
|
||||||
|
默认 repo);租户模式则把 task["repo"] 指到该项目的
|
||||||
|
test-repo 副本(容器内路径与原来一致,宿主源已按项目
|
||||||
|
分开),否则容器会 fallback 到共享 test-repo。
|
||||||
repo_url 非空:调 cloud.repos.resolve_repo 得到本地检出,
|
repo_url 非空:调 cloud.repos.resolve_repo 得到本地检出,
|
||||||
设 task["repo"] 再交给执行器——LocalExecutor 与
|
设 task["repo"] 再交给执行器——LocalExecutor 与
|
||||||
DockerExecutor 的 dispatch_task 都已读 task["repo"],同一条
|
DockerExecutor 的 dispatch_task 都已读 task["repo"],同一条
|
||||||
路径,不另起分支。resolve 失败返回错误文本,调用方据此把
|
路径,不另起分支。租户模式下检出放在租户目录,task
|
||||||
任务置 failed(仓库解析失败绝不标 completed,也不在此重试,
|
["repo"] 设为容器内路径(挂载目标),而不是宿主检出
|
||||||
避免对不可用的仓库反复 clone 烧时间)。
|
路径(worker 侧不再用 task["repo"] 当宿主路径)。
|
||||||
|
resolve 失败返回错误文本,调用方据此把任务置 failed
|
||||||
|
(仓库解析失败绝不标 completed,也不在此重试,避免对
|
||||||
|
不可用的仓库反复 clone 烧时间)。
|
||||||
"""
|
"""
|
||||||
|
use_tenant = self.tenant_isolation and \
|
||||||
|
isinstance(self.executor, DockerExecutor)
|
||||||
repo_url = self._project_repo_url(task.get("project_id"))
|
repo_url = self._project_repo_url(task.get("project_id"))
|
||||||
if not repo_url:
|
if not repo_url:
|
||||||
|
if use_tenant:
|
||||||
|
try:
|
||||||
|
self._tenant_dirs(task.get("project_id"))
|
||||||
|
except (ValueError, RuntimeError) as exc:
|
||||||
|
return "租户目录准备失败: %s" % exc
|
||||||
|
task["repo"] = "/var/lib/codex-app-server/test-repo"
|
||||||
return None
|
return None
|
||||||
try:
|
try:
|
||||||
|
if use_tenant:
|
||||||
|
safe = self._tenant_safe_name(task.get("project_id"))
|
||||||
|
resolve_repo(
|
||||||
|
repo_url, os.path.join(self.tenants_root, safe),
|
||||||
|
task.get("project_id"), runner=self.repos_runner)
|
||||||
|
task["repo"] = \
|
||||||
|
"/var/lib/codex-app-server/repos/%s" % safe
|
||||||
|
else:
|
||||||
task["repo"] = resolve_repo(
|
task["repo"] = resolve_repo(
|
||||||
repo_url, self.repos_dir, task.get("project_id"),
|
repo_url, self.repos_dir, task.get("project_id"),
|
||||||
runner=self.repos_runner)
|
runner=self.repos_runner)
|
||||||
@@ -958,8 +1099,42 @@ class Worker:
|
|||||||
deadline_seconds=self.task_timeout)
|
deadline_seconds=self.task_timeout)
|
||||||
else:
|
else:
|
||||||
if is_docker:
|
if is_docker:
|
||||||
|
tenant_mounts = None
|
||||||
|
tenant_network = None
|
||||||
|
if self.tenant_isolation:
|
||||||
|
try:
|
||||||
|
tenant_mounts = self._tenant_mounts(
|
||||||
|
task.get("project_id"))
|
||||||
|
tenant_network = self._tenant_network(
|
||||||
|
task.get("project_id"))
|
||||||
|
except (ValueError, RuntimeError) as exc:
|
||||||
|
# 提前返回会跳过下面的 finally(清容器
|
||||||
|
# 标记),这里先手动清掉,否则 _running
|
||||||
|
# 里永远残留该任务,心跳与退出清理
|
||||||
|
# 都会把它当在跑任务。
|
||||||
|
self._running_clear(task_id)
|
||||||
|
return ({"executor": "docker",
|
||||||
|
"task_id": task_id,
|
||||||
|
"status": "failed", "stdout": "",
|
||||||
|
"returncode": None, "host_status": None,
|
||||||
|
"stderr": ""},
|
||||||
|
"租户隔离准备失败: %s" % exc)
|
||||||
|
# 网络建失败绝不降级默认网络跑容器(降级会
|
||||||
|
# 让隔离悄悄失效);直接让本次尝试 failed,
|
||||||
|
# 调用方按普通失败处理(重试/fallback 照常)。
|
||||||
|
if not self.executor.ensure_network(tenant_network):
|
||||||
|
self._running_clear(task_id)
|
||||||
|
return ({"executor": "docker",
|
||||||
|
"task_id": task_id,
|
||||||
|
"status": "failed", "stdout": "",
|
||||||
|
"returncode": None, "host_status": None,
|
||||||
|
"stderr": ""},
|
||||||
|
"租户 Docker 网络创建失败: %s"
|
||||||
|
% tenant_network)
|
||||||
result = self.executor.dispatch_task(
|
result = self.executor.dispatch_task(
|
||||||
task, timeout=effective_docker_timeout)
|
task, timeout=effective_docker_timeout,
|
||||||
|
extra_mounts=tenant_mounts,
|
||||||
|
network=tenant_network)
|
||||||
else:
|
else:
|
||||||
result = self.executor.dispatch_task(task)
|
result = self.executor.dispatch_task(task)
|
||||||
combined = (result.get("stdout", "") or "") + \
|
combined = (result.get("stdout", "") or "") + \
|
||||||
@@ -968,7 +1143,11 @@ class Worker:
|
|||||||
_ALREADY_EXISTS_MARKER in combined:
|
_ALREADY_EXISTS_MARKER in combined:
|
||||||
if is_docker:
|
if is_docker:
|
||||||
result = self.executor.run_task(
|
result = self.executor.run_task(
|
||||||
task, timeout=effective_docker_timeout)
|
task, timeout=effective_docker_timeout,
|
||||||
|
extra_mounts=tenant_mounts
|
||||||
|
if self.tenant_isolation else None,
|
||||||
|
network=tenant_network
|
||||||
|
if self.tenant_isolation else None)
|
||||||
else:
|
else:
|
||||||
result = self.executor.run_task(task)
|
result = self.executor.run_task(task)
|
||||||
except Exception as exc: # runner 异常也如实写回 failed
|
except Exception as exc: # runner 异常也如实写回 failed
|
||||||
@@ -1585,6 +1764,20 @@ def main(argv=None):
|
|||||||
"仓库;Docker 执行器下它必须放在"
|
"仓库;Docker 执行器下它必须放在"
|
||||||
" --docker-extra-mounts 已挂载的目录下,"
|
" --docker-extra-mounts 已挂载的目录下,"
|
||||||
"否则容器内 task repo 路径不存在")
|
"否则容器内 task repo 路径不存在")
|
||||||
|
parser.add_argument("--tenant-isolation", action="store_true",
|
||||||
|
help="多租户隔离(仅 Docker 执行器):每个项目用"
|
||||||
|
"独立的 tenants_root/<项目>/work、codex-home、"
|
||||||
|
"test-repo 副本与项目独立 Docker 网络,容器"
|
||||||
|
"内路径不变,宿主挂载源按项目分开")
|
||||||
|
parser.add_argument("--tenants-root",
|
||||||
|
default="/var/lib/codex-app-server/tenants")
|
||||||
|
parser.add_argument("--tenant-template-home",
|
||||||
|
default="/var/lib/codex-app-server/codex-home",
|
||||||
|
help="租户 codex-home 播种模板:只取其中的"
|
||||||
|
"config.toml,绝不复制 thread_history")
|
||||||
|
parser.add_argument("--tenant-template-repo",
|
||||||
|
default="/var/lib/codex-app-server/test-repo",
|
||||||
|
help="租户 test-repo 播种模板(整目录复制)")
|
||||||
parser.add_argument("--env-key", default=None)
|
parser.add_argument("--env-key", default=None)
|
||||||
parser.add_argument("--codex-config", default=None,
|
parser.add_argument("--codex-config", default=None,
|
||||||
help="含 env_key 行的 codex 配置文件(只读名字)")
|
help="含 env_key 行的 codex 配置文件(只读名字)")
|
||||||
@@ -1665,7 +1858,11 @@ def main(argv=None):
|
|||||||
repos_dir=args.repos_dir,
|
repos_dir=args.repos_dir,
|
||||||
heartbeat_file=args.heartbeat_file,
|
heartbeat_file=args.heartbeat_file,
|
||||||
worker_id=args.worker_id,
|
worker_id=args.worker_id,
|
||||||
capacity=args.worker_capacity)
|
capacity=args.worker_capacity,
|
||||||
|
tenant_isolation=args.tenant_isolation,
|
||||||
|
tenants_root=args.tenants_root,
|
||||||
|
tenant_template_home=args.tenant_template_home,
|
||||||
|
tenant_template_repo=args.tenant_template_repo)
|
||||||
# CLI 启动清理兜底(常驻与单次都做):清掉上一轮残留容器。
|
# CLI 启动清理兜底(常驻与单次都做):清掉上一轮残留容器。
|
||||||
worker.cleanup_stale_containers()
|
worker.cleanup_stale_containers()
|
||||||
# CLI 起 worker 前先建表/迁移:全新库直接起循环不应因缺表而炸。
|
# CLI 起 worker 前先建表/迁移:全新库直接起循环不应因缺表而炸。
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ EnvironmentFile=/etc/codex-app-server/env
|
|||||||
# scaler 会按实例强制追加;--db 只给 scaler,scaler 会补给 worker。
|
# scaler 会按实例强制追加;--db 只给 scaler,scaler 会补给 worker。
|
||||||
# 注意:启用 autoscaler 后必须停掉静态 codex-cloud-worker,
|
# 注意:启用 autoscaler 后必须停掉静态 codex-cloud-worker,
|
||||||
# 否则两边同时派任务(scaler 只管理 autoscale-* 实例)。
|
# 否则两边同时派任务(scaler 只管理 autoscale-* 实例)。
|
||||||
ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos
|
ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos --tenant-isolation --tenants-root /var/lib/codex-app-server/tenants --tenant-template-home /var/lib/codex-app-server/codex-home --tenant-template-repo /var/lib/codex-app-server/test-repo
|
||||||
Restart=on-failure
|
Restart=on-failure
|
||||||
RestartSec=10
|
RestartSec=10
|
||||||
# 停 scaler 时要等它把 worker 优雅排空(drain 60 秒 + 缓冲),
|
# 停 scaler 时要等它把 worker 优雅排空(drain 60 秒 + 缓冲),
|
||||||
|
|||||||
Reference in new issue
Block a user