diff --git a/cloud/README.md b/cloud/README.md index 6752683..96b8ade 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -479,6 +479,18 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 - 清理与取消:`docker run --rm`,正常失败也自动删容器;超时 时先 docker kill 再 docker rm -f;worker 取消时按任务派生的 容器名 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 网络,本执行 器不做 egress 白名单——真正的“只放行模型 API”需要宿主防火 墙或代理规则,那是部署项,不是本轮代码假装做到的,不要宣称 diff --git a/cloud/sandbox.py b/cloud/sandbox.py index fe6b41f..6962eac 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -369,13 +369,24 @@ class DockerExecutor(Executor): "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 --rm --name <容器名> --user <用户> - --cpus <核数> --memory <内存> -v :/work - -v <代码目录>:/app:ro -w /work [-e ] - <镜像> <容器内命令...>。-e 只传名字,不带 =值。 + --cpus <核数> --memory <内存> [--cap-drop=ALL + --security-opt=no-new-privileges] [--network <网络>] + -v :/work -v <代码目录>:/app:ro -w /work + [-e ] <镜像> <容器内命令...>。-e 只传名字, + 不带 =值。 + extra_mounts 给定时替换构造时的静态挂载(多租户隔离 + 用:worker 按项目算出挂载后每次调用传入);不给时用 + 静态挂载。network 给定时容器进该 Docker 网络(多租户 + 用项目独立网络,容器之间默认 bridge 上互相可见的问题 + 由此解决);不给时用 docker 默认网络。 + 降权:容器默认已是非 root 用户,再加 --cap-drop=ALL + 与 no-new-privileges(提权类工具调用会被挡下,codex + 的常规文件/git 操作不受影响,见多租户实测)。 """ self._check_docker() self._check_key() @@ -384,13 +395,19 @@ class DockerExecutor(Executor): "--user", self.user, "--cpus", self.cpus, "--memory", self.memory, - "-v", "%s:/work" % self._worktree_for(task), - "-v", "%s:/app:ro" % self._app_dir(), - "-w", "/work"] + "--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(), + "-w", "/work"] # 额外挂载放在基础挂载之后:/app/work 这类子路径挂载会 # 覆盖 /app:ro 的对应子路径,让容器内 host.py 能写 # 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)] if self.env_key: cmd += ["-e", self.env_key] @@ -500,18 +517,22 @@ class DockerExecutor(Executor): continue 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。 超时:runner 抛 subprocess.TimeoutExpired(或同名异常) 时先 kill+rm 容器,再返回 failed(stdout 截断 + timeout 说明,returncode 为 None);其它 runner 异常也如实返回 failed,不编造成功。timeout 只影响本次调用,不改实例。 + extra_mounts/network 只影响本次调用(多租户按项目传入)。 """ task_id = task.get("id") if isinstance(task, dict) else None if task_id is None: 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: result = self._call_runner(cmd, timeout=timeout) except Exception as exc: @@ -536,7 +557,8 @@ class DockerExecutor(Executor): parsed["container"] = self.container_name(task) 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。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -548,9 +570,12 @@ class DockerExecutor(Executor): repo = task.get("repo") if 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。""" if not isinstance(task, dict) or task.get("id") is None: raise ValueError("任务缺少 id") @@ -559,7 +584,35 @@ class DockerExecutor(Executor): inner = self._inner_host_command( "run", "--task", self._host_name(task), "--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: """只读执行:容器内调 host.py status,不直接跑模型回合。 diff --git a/cloud/tests/test_tenant_isolation.py b/cloud/tests/test_tenant_isolation.py new file mode 100644 index 0000000..3fc0409 --- /dev/null +++ b/cloud/tests/test_tenant_isolation.py @@ -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() diff --git a/cloud/worker.py b/cloud/worker.py index 7dd3760..d613c39 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -233,7 +233,9 @@ class Worker: task_timeout=1800, max_retries=1, retry_backoff_seconds=5.0, repos_dir="./repos", 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.host_py_path = host_py_path self.config_path = config_path @@ -298,6 +300,22 @@ class Worker: # repos_runner 可注入(单测用假 git runner),缺省走真实 git。 self.repos_dir = repos_dir self.repos_runner = repos_runner + # 多租户隔离(仅 Docker 执行器有效,Local 执行器在本机 + # 直接跑 host.py,同一用户下本来就没有容器边界,flag + # 对它不起作用):开启后每个项目用独立的宿主目录 + # tenants_root//{work,codex-home,test-repo, + # },容器内路径保持与原来完全一致(/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;主机路径不在代码里硬编码,主机 # 部署时用 --heartbeat-file 或环境变量传入(见 deploy/README)。 @@ -856,6 +874,107 @@ class Worker: "stdout": out or "", "stderr": err or "", "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/、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): """查任务所属项目的 repo_url;查不到项目、列缺失、表缺失时 @@ -882,21 +1001,43 @@ class Worker: def _apply_project_repo(self, task): """派发前按项目 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 得到本地检出, 设 task["repo"] 再交给执行器——LocalExecutor 与 DockerExecutor 的 dispatch_task 都已读 task["repo"],同一条 - 路径,不另起分支。resolve 失败返回错误文本,调用方据此把 - 任务置 failed(仓库解析失败绝不标 completed,也不在此重试, - 避免对不可用的仓库反复 clone 烧时间)。 + 路径,不另起分支。租户模式下检出放在租户目录,task + ["repo"] 设为容器内路径(挂载目标),而不是宿主检出 + 路径(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")) 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 try: - task["repo"] = resolve_repo( - repo_url, self.repos_dir, task.get("project_id"), - runner=self.repos_runner) + 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( + repo_url, self.repos_dir, task.get("project_id"), + runner=self.repos_runner) except (RepoError, ValueError) as exc: return "仓库解析失败: %s" % exc except Exception as exc: @@ -958,8 +1099,42 @@ class Worker: deadline_seconds=self.task_timeout) else: 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( - task, timeout=effective_docker_timeout) + task, timeout=effective_docker_timeout, + extra_mounts=tenant_mounts, + network=tenant_network) else: result = self.executor.dispatch_task(task) combined = (result.get("stdout", "") or "") + \ @@ -968,7 +1143,11 @@ class Worker: _ALREADY_EXISTS_MARKER in combined: if is_docker: 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: result = self.executor.run_task(task) except Exception as exc: # runner 异常也如实写回 failed @@ -1585,6 +1764,20 @@ def main(argv=None): "仓库;Docker 执行器下它必须放在" " --docker-extra-mounts 已挂载的目录下," "否则容器内 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("--codex-config", default=None, help="含 env_key 行的 codex 配置文件(只读名字)") @@ -1665,7 +1858,11 @@ def main(argv=None): repos_dir=args.repos_dir, heartbeat_file=args.heartbeat_file, 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 启动清理兜底(常驻与单次都做):清掉上一轮残留容器。 worker.cleanup_stale_containers() # CLI 起 worker 前先建表/迁移:全新库直接起循环不应因缺表而炸。 diff --git a/deploy/autoscaler/codex-autoscaler.service b/deploy/autoscaler/codex-autoscaler.service index 996e721..d4c9d06 100644 --- a/deploy/autoscaler/codex-autoscaler.service +++ b/deploy/autoscaler/codex-autoscaler.service @@ -14,7 +14,7 @@ EnvironmentFile=/etc/codex-app-server/env # scaler 会按实例强制追加;--db 只给 scaler,scaler 会补给 worker。 # 注意:启用 autoscaler 后必须停掉静态 codex-cloud-worker, # 否则两边同时派任务(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 RestartSec=10 # 停 scaler 时要等它把 worker 优雅排空(drain 60 秒 + 缓冲),