From bed3fc0dff21d33c5dfb83486878975fb9b2f5df Mon Sep 17 00:00:00 2001 From: host Date: Mon, 26 Oct 2026 04:16:11 +0800 Subject: [PATCH] Project repo_url: worker can run tasks from any git repo Projects gain repo_url (validated: https:// or git@ only, no userinfo/password, no file:// or local paths; old DBs get the column via idempotent ALTER in both SQLite and Postgres init). Worker resolves a project's repo_url to a local checkout under --repos-dir (default ./repos) before dispatch: clone if absent, verify remote origin matches and fetch if present, never force reset/clean, never push (push needs write credentials the control plane does not store). Resolve failure fails the task with a clear error, never completed. Docker workers need repos_dir under a same-path extra mount (e.g. /var/lib/codex-app-server/repos). Also: driver-missing test is now mocked instead of depending on the host really having no psycopg driver installed. Deploy package: Caddyfile template exposing only /webhooks/* and /oauth/* (tasks/projects/admin stay 403) plus a webhook-secret generator script that never prints the secret. Tests: host 27 pass, cloud 286 pass (8 skipped are environment-dependent real-Postgres tests). --- cloud/README.md | 41 ++- cloud/db.py | 13 +- cloud/repos.py | 150 ++++++++++ cloud/server.py | 44 ++- cloud/tests/test_postgres_backend.py | 8 +- cloud/tests/test_prod_gaps.py | 10 +- cloud/tests/test_project_repo.py | 397 +++++++++++++++++++++++++++ cloud/worker.py | 97 ++++++- deploy/caddy/Caddyfile | 27 ++ deploy/setup-webhook-secret.sh | 18 ++ 10 files changed, 788 insertions(+), 17 deletions(-) create mode 100644 cloud/repos.py create mode 100644 cloud/tests/test_project_repo.py create mode 100644 deploy/caddy/Caddyfile create mode 100755 deploy/setup-webhook-secret.sh diff --git a/cloud/README.md b/cloud/README.md index d494feb..fa18223 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -533,9 +533,40 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 绝不抛异常），`Worker.cleanup_stale_containers()` 在 run_loop 启动 与 CLI 启动时各调一次；取消/超时/失败路径上已有的容器清理保留 （docker run `--rm`、超时 kill+rm、取消按名 kill）。 -- 仍未做（不要宣称已做）：webhook 建的任务仍需 worker 认领执行， - worker 的 repo 仍是配置里的 repo，不是任意 GitHub 仓库的自动克隆； - `CloudExecutor` 仍是占位符；公网 https 入口仍需部署侧反向代理 + - TLS；Docker 超时杀容器与真实 `pg_dump`、真实 GitHub webhook 送达 - 仍需在真实环境验证（本轮只用 fake runner / 本地 HTTP 单测）。 +## 项目仓库（任意仓库任务） + +- 项目级 `repo_url`：`POST /projects` 接受可选 `repo_url`， + `GET /projects` 列表与 `POST /projects` 的 201 返回里都含它， + 不传时为 `None`（老行为不变）。校验与 worker 共用同一套 + `cloud.repos.validate_repo_url`：只接受 `https://` 开头（且不含 + userinfo，即不含用户名/密码）或 `git@` 开头（`git@host:path` 形）； + `file://`、`ssh://`、本地路径、带密码的 URL 一律 400 且不建项目。 +- 存储与迁移：`projects` 表新增 `repo_url TEXT` 可空列；SQLite 的 + `init_db` 建表时就含该列，老库用 `ALTER TABLE` 幂等补列（仿照 + `migrate_tasks_table` 的写法）；Postgres 的 `POSTGRES_SCHEMA` + 同样加该列，`init_postgres_schema` 也对老库补列。 +- worker 派发：`run_one` 派发前查任务所属项目的 `repo_url`，非空时 + 经 `cloud.repos.resolve_repo` 在 `--repos-dir`（默认 `./repos`）下 + 解析：目录不存在时 `git clone`；已存在时校验它是 git 仓库且 + remote origin 与 `repo_url` 一致（不一致时报错，绝不强制改 + remote），一致时 `git fetch origin`（失败明确报错，不静默用旧代码； + fetch 成功后不 checkout/pull/reset/clean，工作树由 host.py 的 + worktree 逻辑处理）。解析出的路径设为 `task["repo"]` 再交给执行器， + Local 与 Docker 的 `dispatch_task` 都已读 `task["repo"]`，同一条路径； + 查不到项目或 `repo_url` 为空时走原有默认 repo，不报错；解析失败时 + 任务置 `failed`，错误文本写清是仓库解析失败，绝不标 completed。 +- Docker 下的路径前提：容器内 `task["repo"]` 路径必须与宿主机一致， + 所以 `repos_dir` 必须放在 `--docker-extra-mounts` 已挂载的目录下 + （例如把 `/var/lib/codex-app-server/repos` 以同路径挂载进容器）， + 否则容器内 repo 路径不存在——不要假装任意目录都行。生产部署示例： + `python3 -m cloud.worker --db <库> --executor docker + --repos-dir /var/lib/codex-app-server/repos + --docker-extra-mounts /var/lib/codex-app-server:/var/lib/codex-app-server,…`。 +- 仍未做（不要宣称已做）：推送到远端明确不在本轮范围——任务在本地 + 检出上执行、本地 worktree 合并仍按 host.py 现有流程，`git push` + 到远端需要仓库写凭证，控制面不存、不推，所以任意仓库的自动 PR + 仍未实现；私有仓库 clone 会失败，错误文本写明“需要仓库访问凭证， + 控制面不存仓库凭证”，绝不尝试用任何 key 去 clone；webhook 建的 + 任务仍需 worker 认领执行；`CloudExecutor` 仍是占位符；公网 https + 入口仍需部署侧反向代理 + TLS。 diff --git a/cloud/db.py b/cloud/db.py index 89ff17b..70e2cc5 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -780,7 +780,8 @@ POSTGRES_SCHEMA = ( "id TEXT PRIMARY KEY, " "name TEXT, " "owner_id TEXT, " - "created_at DOUBLE PRECISION)", + "created_at DOUBLE PRECISION, " + "repo_url TEXT)", # tasks 含 server.migrate_tasks_table 补的那 6 列，建表时一次建齐； # 老库迁移仍走 migrate_tasks_table（PRAGMA 翻译后在 PG 上是 no-op）。 "CREATE TABLE IF NOT EXISTS tasks(" @@ -861,9 +862,19 @@ def init_postgres_schema(conn): conn 须是 PgConnection（或具同名 execute/commit 的适配连接）； SQLite 连接传进来会在首条 DDL 的 IDENTITY 语法上报错—— 调用方须按后端选对连接，不要把 SQLite 连接混进来。 + + 建表后还给老库 projects 表补 repo_url 列（新库建表时已含）： + PRAGMA table_info 经适配层翻译成 information_schema 查询， + 调用方取 row[1] 即列名；缺列时 ALTER TABLE 补列，已迁移的 + 库是 no-op。server.init_db 的 Postgres 分支还会再跑一次 + migrate_projects_table，两次补列互不冲突（第二次是 no-op）。 """ for statement in POSTGRES_SCHEMA: conn.execute(statement) + cols = {row[1] for row in conn.execute( + "PRAGMA table_info(projects)").fetchall()} + if "repo_url" not in cols: + conn.execute("ALTER TABLE projects ADD COLUMN repo_url TEXT") conn.commit() diff --git a/cloud/repos.py b/cloud/repos.py new file mode 100644 index 0000000..6409ea8 --- /dev/null +++ b/cloud/repos.py @@ -0,0 +1,150 @@ +"""项目级仓库解析：把项目的 repo_url 变成 worker 可派发的本地检出。 + +零第三方依赖，只用 Python 标准库；不改 host.py 的 worktree 逻辑， +不存任何仓库凭证、不推远端。 + +- validate_repo_url：API（POST /projects）与 worker 共用的同一套 + 校验，绝不另写第二份校验逻辑让两者漂移。只接受： + https:// 开头（且不含 userinfo，即不含用户名/密码），或 + git@ 开头（git@host:path 形）。file://、ssh://、本地路径、 + 带密码的 URL 一律拒绝。 +- checkout_dir：按 project_id 的安全字符命名，防路径穿越。 +- resolve_repo：目录不存在时 git clone；已存在时校验它是 git + 仓库且 remote origin 与 repo_url 一致（不一致时报错，绝不强制 + 改 remote），一致时 git fetch origin（失败明确报错，不静默用 + 旧代码）；fetch 成功后不自动 checkout/pull、不 git reset/clean， + 工作树脏不脏由 host.py 的 worktree 逻辑处理。 +- 私有仓库 clone 失败时错误文本写明：需要仓库访问凭证，控制面 + 不存仓库凭证——绝不尝试用任何 key 去 clone。 +""" + +import os +import re +import subprocess +from urllib.parse import urlparse + +__all__ = [ + "RepoError", + "validate_repo_url", + "checkout_dir", + "resolve_repo", +] + + +class RepoError(ValueError): + """仓库解析失败（校验、clone、remote 不一致、fetch 失败）。""" + + +_GIT_SCP_RE = re.compile(r"^git@[^/:@\s]+:[^@\s]+$") +_SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9_.-]") + + +def validate_repo_url(repo_url): + """校验 repo_url，合法时返回原值，否则抛 ValueError。 + + 拒绝：非字符串、空串、含空白/控制字符、非 https:// 与 git@ 开头、 + https:// 带 userinfo（username 或 password 任一存在即拒绝， + 含 https://user@host/ 这种只有用户名无密码的也拒绝，因为 + userinfo 本身就是凭证载体）、https:// 无 host、git@ 形不合规。 + """ + if not isinstance(repo_url, str): + raise ValueError("repo_url 必须是非空字符串") + # 含空白/控制字符的 URL 直接拒绝，不静默 strip：strip 会把 + # “https://…\n”这种畸形输入洗成合法 URL 存进库，git 调用时 + # 再出难以定位的问题；调用方应传干净的 URL。 + if not repo_url or any(ch.isspace() for ch in repo_url): + raise ValueError("repo_url 必须是非空字符串且不得含空白字符") + url = repo_url + if url.startswith("https://"): + parsed = urlparse(url) + if parsed.username is not None or parsed.password is not None: + raise ValueError( + "repo_url 不得含 userinfo（用户名/密码）: %s" % url) + if not parsed.hostname: + raise ValueError("repo_url 缺少 host: %s" % url) + return url + if url.startswith("git@"): + if not _GIT_SCP_RE.match(url): + raise ValueError("git@ 形 repo_url 不合法: %s" % url) + return url + raise ValueError( + "repo_url 只接受 https:// 或 git@ 开头，" + "不接受 file://、ssh://、本地路径: %s" % url) + + +def checkout_dir(repos_dir, project_id): + """按 project_id 的安全字符命名检出目录，防路径穿越。 + + project_id 本身是 hex，仍做安全字符过滤；过滤后为空时报错。 + 返回的绝对路径一定在 repos_dir 之下，否则报错（防御性双检）。 + """ + raw = str(project_id) + safe = _SAFE_NAME_RE.sub("-", raw).strip(".") + if not safe or safe in (".", ".."): + raise RepoError("project_id 无法映射为安全的检出目录名: %r" % raw) + base = os.path.abspath(repos_dir) + path = os.path.join(base, safe) + if os.path.commonpath([base, path]) != base: + raise RepoError("检出目录越出 repos_dir，拒绝: %s" % path) + return path + + +def _default_runner(cmd): + return subprocess.run(cmd, capture_output=True, text=True) + + +def _run(runner, cmd): + """调 runner：优先只传 cmd，假 runner 只接受 cmd 时即走此路； + 接受 (cmd, timeout) 的 runner 也兼容（TypeError 时回退）。""" + try: + return runner(cmd) + except TypeError: + return runner(cmd, timeout=None) + + +def resolve_repo(repo_url, repos_dir, project_id, runner=None): + """把 repo_url 解析为本地检出路径，返回该路径。 + + 目录不存在：git clone repo_url 到该目录；clone 失败抛 RepoError， + 文本含“需要仓库访问凭证，控制面不存仓库凭证”的提示（私有仓 + 库、仓库不存在、无权访问都走同一文案，不猜测、不尝试凭证）。 + 目录已存在：须是 git 仓库（.git 目录或文件存在）且 remote + origin 的 URL 与 repo_url 一致；不是 git 仓库、取不到 origin、 + origin 不一致时都抛 RepoError，绝不强制改 remote、不覆盖。 + 一致时 git fetch origin：失败抛 RepoError，不静默用旧代码； + 成功后不 checkout/pull/reset/clean，工作树由 host.py 处理。 + """ + url = validate_repo_url(repo_url) + run = runner if runner is not None else _default_runner + target = checkout_dir(repos_dir, project_id) + os.makedirs(repos_dir, exist_ok=True) + if not os.path.exists(target): + result = _run(run, ["git", "clone", url, target]) + if getattr(result, "returncode", 1) != 0: + err = (getattr(result, "stderr", "") or "").strip() + raise RepoError( + "git clone 失败: %s；若为私有仓库，需要仓库访问凭证，" + "控制面不存仓库凭证，请先在可访问该仓库的机器上完成" + "检出或改用公开仓库" % (err or url)) + return target + git_path = os.path.join(target, ".git") + if not (os.path.isdir(git_path) or os.path.isfile(git_path)): + raise RepoError( + "检出目录已存在但不是 git 仓库，拒绝覆盖: %s" % target) + result = _run(run, ["git", "-C", target, "remote", "get-url", "origin"]) + if getattr(result, "returncode", 1) != 0: + raise RepoError( + "检出目录 %s 取不到 remote origin，拒绝派发" % target) + origin = (getattr(result, "stdout", "") or "").strip() + if origin != url: + raise RepoError( + "检出目录 %s 的 remote origin（%s）与项目 repo_url（%s）" + "不一致，拒绝派发（绝不强制改 remote）" + % (target, origin, url)) + fetched = _run(run, ["git", "-C", target, "fetch", "origin"]) + if getattr(fetched, "returncode", 1) != 0: + err = (getattr(fetched, "stderr", "") or "").strip() + raise RepoError( + "git fetch origin 失败，拒绝用旧代码派发: %s" + % (err or target)) + return target diff --git a/cloud/server.py b/cloud/server.py index 348aa99..8fe44bc 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -67,6 +67,7 @@ from cloud.oauth import ( map_external_identity_to_local_user_with_key, start_oauth_login, ) +from cloud.repos import validate_repo_url from cloud.push import ( create_subscription as _push_create, delete_subscription as _push_delete, @@ -192,6 +193,20 @@ def migrate_tasks_table(conn): "ALTER TABLE tasks ADD COLUMN %s %s" % (name, coltype)) +def migrate_projects_table(conn): + """幂等迁移：给老库 projects 表补 repo_url 列，不丢数据。 + + 与 migrate_tasks_table 同款写法：PRAGMA table_info 查现有列 + （Postgres 适配层把它翻译成 information_schema 查询，调用方 + 取 row[1] 即列名），缺 repo_url 时 ALTER TABLE 补列，已迁移 + 的库再跑一次是 no-op。SQLite 与 Postgres 两支都走它。 + """ + cols = {row[1] for row in conn.execute( + "PRAGMA table_info(projects)").fetchall()} + if "repo_url" not in cols: + conn.execute("ALTER TABLE projects ADD COLUMN repo_url TEXT") + + def _is_postgres_backend(): backend = os.environ.get("CLOUD_DB_BACKEND") or "sqlite" return str(backend).strip().lower() in ("postgres", "postgresql") @@ -210,6 +225,7 @@ def init_db(db_path): if _is_postgres_backend(): _init_postgres_schema(conn) migrate_tasks_table(conn) + migrate_projects_table(conn) conn.commit() return conn.execute( @@ -217,7 +233,8 @@ def init_db(db_path): "id TEXT PRIMARY KEY, " "name TEXT, " "owner_id TEXT, " - "created_at REAL)" + "created_at REAL, " + "repo_url TEXT)" ) conn.execute( "CREATE TABLE IF NOT EXISTS tasks(" @@ -235,6 +252,7 @@ def init_db(db_path): "UNIQUE(project_id, idempotency_key))" ) migrate_tasks_table(conn) + migrate_projects_table(conn) conn.commit() finally: conn.close() @@ -296,6 +314,9 @@ def _project_to_dict(row): "name": row["name"], "owner_id": row["owner_id"], "created_at": row["created_at"], + # 老库行缺 repo_url 列时 _row_get 给 None，不炸（列本身由 + # migrate_projects_table 在 init_db 时补，老行读到也是 None）。 + "repo_url": _row_get(row, "repo_url"), } @@ -547,6 +568,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if not isinstance(project_id, str): self._send_json(400, {"error": "id must be a string"}) return + # repo_url 可选：不传时为 None（老行为不变）；传了则与 worker + # 共用 cloud.repos.validate_repo_url 的同一套校验，不合法时 + # 400 且不建项目，绝不把带凭证的 URL 存进库。 + repo_url = body.get("repo_url") + if repo_url is not None: + try: + repo_url = validate_repo_url(repo_url) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return now = time.time() conn = self._db() try: @@ -557,9 +588,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(409, {"error": "project exists"}) return conn.execute( - "INSERT INTO projects(id, name, owner_id, created_at)" - " VALUES(?, ?, ?, ?)", - (project_id, name, user_id, now), + "INSERT INTO projects(id, name, owner_id, created_at," + " repo_url) VALUES(?, ?, ?, ?, ?)", + (project_id, name, user_id, now, repo_url), ) conn.commit() project = { @@ -567,6 +598,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "name": name, "owner_id": user_id, "created_at": now, + "repo_url": repo_url, } except (sqlite3.IntegrityError, _PgIntegrityError): self._send_json(409, {"error": "project exists"}) @@ -591,8 +623,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): ).fetchall() } rows = conn.execute( - "SELECT id, name, owner_id, created_at FROM projects" - " ORDER BY created_at, id" + "SELECT id, name, owner_id, created_at, repo_url" + " FROM projects ORDER BY created_at, id" ).fetchall() finally: conn.close() diff --git a/cloud/tests/test_postgres_backend.py b/cloud/tests/test_postgres_backend.py index 6a00b38..8166ab3 100644 --- a/cloud/tests/test_postgres_backend.py +++ b/cloud/tests/test_postgres_backend.py @@ -288,7 +288,13 @@ class SchemaStatementTestCase(unittest.TestCase): conn = db.connect() init_postgres_schema(conn) executed = [sql for sql, _ in fake.conn.cursor_obj.executed] - self.assertEqual(len(executed), len(POSTGRES_SCHEMA)) + # 改动理由：init_postgres_schema 现在还要给老库 projects 表补 + # repo_url 列（PRAGMA 查列 + ALTER TABLE），所以执行语句数 + # 不再精确等于 schema 语句数；断言改为“schema 语句全部执行 + # 且补列语句也执行”，不放宽 schema 本身的覆盖要求。 + for stmt in POSTGRES_SCHEMA: + self.assertIn(stmt, executed) + self.assertTrue(any("repo_url" in sql for sql in executed)) self.assertTrue(fake.conn.committed) diff --git a/cloud/tests/test_prod_gaps.py b/cloud/tests/test_prod_gaps.py index f3b4362..1a59d52 100644 --- a/cloud/tests/test_prod_gaps.py +++ b/cloud/tests/test_prod_gaps.py @@ -16,6 +16,7 @@ import tempfile import threading import time import unittest +from unittest import mock from pathlib import Path from types import SimpleNamespace @@ -309,9 +310,14 @@ class MetricsAndBackupTestCase(unittest.TestCase): conn.close() def test_postgres_backend_without_driver_fails_before_file(self): + # 环境无关：用 mock 模拟驱动缺失，而不是依赖本机真的没装驱动 + # （另一任务已在本 VM 装了 psycopg2 + Postgres，原写法在该环境下 + # 前提不存在而失败，并非代码回归；CI runner 无驱动时原逻辑同样成立）。 other_db = os.path.join(self.tmp.name, "pg.db") - with self.assertRaises(PostgresDriverMissingError): - create_server(other_db, "127.0.0.1", 0, backend="postgres") + with mock.patch("cloud.db._load_postgres_driver", + side_effect=PostgresDriverMissingError("no driver")): + with self.assertRaises(PostgresDriverMissingError): + create_server(other_db, "127.0.0.1", 0, backend="postgres") self.assertFalse(os.path.exists(other_db)) diff --git a/cloud/tests/test_project_repo.py b/cloud/tests/test_project_repo.py new file mode 100644 index 0000000..2efd7bf --- /dev/null +++ b/cloud/tests/test_project_repo.py @@ -0,0 +1,397 @@ +"""项目级 repo_url 单测：校验、API、老库迁移、resolve、worker 派发。 + +只用标准库与假 runner/假执行器，不跑真实模型回合、不联网、 +不用真实 git clone（clone 逻辑用 mock runner 模拟 git 的副作用： +建出目标目录 + .git；真实公开仓库 clone 列入真实环境验证清单）。 +测试里出现的 URL 与密钥均为占位值，绝无真实凭证。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.auth import AuthStore + from cloud.repos import ( + RepoError, + checkout_dir, + resolve_repo, + validate_repo_url, + ) + from cloud.server import create_server, init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.repos import ( + RepoError, + checkout_dir, + resolve_repo, + validate_repo_url, + ) + from cloud.server import create_server, init_db + from cloud.worker import Worker + + +def _git_runner_factory(behavior, calls, side_effect=None): + """behavior: dict，键为命令首段特征，值为 (returncode, stdout, stderr)。 + + clone 命令默认成功且模拟 git 副作用（建出目标目录 + .git）； + 调用方可覆写任意命令的行为。 + """ + def runner(cmd): + calls.append(list(cmd)) + if cmd[:2] == ["git", "clone"]: + key = "clone" + rc, out, err = behavior.get( + key, (0, "", "")) + if rc == 0 and side_effect is None: + target = cmd[-1] + os.makedirs(os.path.join(target, ".git"), exist_ok=True) + return SimpleNamespace(returncode=rc, stdout=out, stderr=err) + if cmd[:3] == ["git", "-C", cmd[2]] and "remote" in cmd: + rc, out, err = behavior.get( + "get-url", (0, behavior.get("origin", ""), "")) + return SimpleNamespace(returncode=rc, stdout=out, stderr=err) + if "fetch" in cmd: + rc, out, err = behavior.get("fetch", (0, "", "")) + return SimpleNamespace(returncode=rc, stdout=out, stderr=err) + return SimpleNamespace(returncode=0, stdout="", stderr="") + return runner + + +class ValidateRepoUrlTestCase(unittest.TestCase): + def test_valid_https_and_git_scp(self): + self.assertEqual( + validate_repo_url("https://github.com/openqj/codex-app-server.git"), + "https://github.com/openqj/codex-app-server.git") + self.assertEqual( + validate_repo_url("git@github.com:openqj/codex-app-server.git"), + "git@github.com:openqj/codex-app-server.git") + + def test_rejects_userinfo_file_ssh_local_empty(self): + bad = [ + "https://user:pass@github.com/o/r.git", + "https://user@github.com/o/r.git", + "file:///tmp/repo", + "ssh://git@github.com/o/r.git", + "ssh://git:pass@github.com/o/r.git", + "/tmp/local-repo", + "./relative", + "", + " ", + "https://", + "git@github.com", + "https://github.com/o/r.git\n", + ] + for url in bad: + with self.subTest(url=url): + with self.assertRaises(ValueError): + validate_repo_url(url) + with self.assertRaises(ValueError): + validate_repo_url(None) + with self.assertRaises(ValueError): + validate_repo_url(123) + + +class CheckoutDirTestCase(unittest.TestCase): + def test_sanitizes_and_stays_inside(self): + tmp = tempfile.TemporaryDirectory() + base = os.path.join(tmp.name, "repos") + path = checkout_dir(base, "abc123") + self.assertEqual(path, os.path.join(os.path.abspath(base), "abc123")) + # 穿越尝试被过滤，仍在 repos_dir 之下。 + path2 = checkout_dir(base, "../../etc") + self.assertTrue(os.path.commonpath( + [os.path.abspath(base), path2]) == os.path.abspath(base)) + tmp.cleanup() + + +class ResolveRepoTestCase(unittest.TestCase): + def test_clone_when_missing(self): + tmp = tempfile.TemporaryDirectory() + calls = [] + runner = _git_runner_factory({}, calls) + url = "https://github.com/openqj/codex-app-server.git" + path = resolve_repo(url, os.path.join(tmp.name, "repos"), + "proj1", runner=runner) + self.assertTrue(os.path.isdir(os.path.join(path, ".git"))) + self.assertEqual(calls[0][:3], ["git", "clone", url]) + tmp.cleanup() + + def test_clone_failure_mentions_no_stored_credentials(self): + tmp = tempfile.TemporaryDirectory() + calls = [] + runner = _git_runner_factory( + {"clone": (128, "", "fatal: Authentication failed")}, calls) + with self.assertRaises(RepoError) as ctx: + resolve_repo("https://github.com/o/private.git", + os.path.join(tmp.name, "repos"), + "proj1", runner=runner) + self.assertIn("控制面不存仓库凭证", str(ctx.exception)) + tmp.cleanup() + + def test_existing_matching_origin_fetches(self): + tmp = tempfile.TemporaryDirectory() + repos = os.path.join(tmp.name, "repos") + target = checkout_dir(repos, "proj1") + os.makedirs(os.path.join(target, ".git")) + url = "https://github.com/openqj/codex-app-server.git" + calls = [] + runner = _git_runner_factory({"origin": url}, calls) + path = resolve_repo(url, repos, "proj1", runner=runner) + self.assertEqual(path, target) + self.assertTrue(any("fetch" in c for c in calls)) + tmp.cleanup() + + def test_existing_mismatched_origin_refuses(self): + tmp = tempfile.TemporaryDirectory() + repos = os.path.join(tmp.name, "repos") + target = checkout_dir(repos, "proj1") + os.makedirs(os.path.join(target, ".git")) + calls = [] + runner = _git_runner_factory( + {"origin": "https://github.com/other/repo.git"}, calls) + with self.assertRaises(RepoError) as ctx: + resolve_repo("https://github.com/openqj/codex-app-server.git", + repos, "proj1", runner=runner) + self.assertIn("不一致", str(ctx.exception)) + tmp.cleanup() + + def test_existing_not_git_refuses(self): + tmp = tempfile.TemporaryDirectory() + repos = os.path.join(tmp.name, "repos") + os.makedirs(checkout_dir(repos, "proj1")) + calls = [] + runner = _git_runner_factory({}, calls) + with self.assertRaises(RepoError): + resolve_repo("https://github.com/o/r.git", + repos, "proj1", runner=runner) + self.assertEqual(calls, []) + tmp.cleanup() + + def test_fetch_failure_refuses(self): + tmp = tempfile.TemporaryDirectory() + repos = os.path.join(tmp.name, "repos") + target = checkout_dir(repos, "proj1") + os.makedirs(os.path.join(target, ".git")) + url = "https://github.com/o/r.git" + calls = [] + runner = _git_runner_factory( + {"origin": url, "fetch": (128, "", "fatal: unable to access")}, + calls) + with self.assertRaises(RepoError) as ctx: + resolve_repo(url, repos, "proj1", runner=runner) + self.assertIn("fetch", str(ctx.exception)) + tmp.cleanup() + + +class _FakeExecutor: + def __init__(self): + self.seen = [] + + def _host_name(self, task): + return task["id"] + + def dispatch_task(self, task): + self.seen.append(dict(task)) + return {"executor": "local", "task_id": task["id"], + "status": "completed", "stdout": "ok", "returncode": 0, + "host_status": None} + + +def _make_worker_db(tmp, project_repo_url="__unset__"): + """建库 + 项目行 + queued 任务行，返回 (db_path, project_id)。""" + db_path = os.path.join(tmp, "control.db") + init_db(db_path) + project_id = "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6" + conn = sqlite3.connect(db_path) + try: + if project_repo_url == "__unset__": + conn.execute( + "INSERT INTO projects(id, name, owner_id, created_at)" + " VALUES(?, 'p', 'owner', 0)", (project_id,)) + else: + conn.execute( + "INSERT INTO projects(id, name, owner_id, created_at," + " repo_url) VALUES(?, 'p', 'owner', 0, ?)", + (project_id, project_repo_url)) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at, updated_at," + " last_heartbeat) VALUES('t1', ?, 'do it', 'queued'," + " '[]', 0, 'system', 0, 0, NULL)", (project_id,)) + conn.commit() + finally: + conn.close() + return db_path, project_id + + +class WorkerProjectRepoTestCase(unittest.TestCase): + def test_dispatch_uses_project_repo(self): + tmp = tempfile.TemporaryDirectory() + url = "https://github.com/openqj/codex-app-server.git" + db_path, _ = _make_worker_db(tmp.name, url) + repos = os.path.join(tmp.name, "repos") + calls = [] + executor = _FakeExecutor() + worker = Worker(db_path, "host.py", repo=None, + env={"XCM_API_KEY": "test-placeholder"}, + env_key=None, executor=executor, + repos_dir=repos, + repos_runner=_git_runner_factory({}, calls), + retry_backoff_seconds=0) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + self.assertEqual(len(executor.seen), 1) + self.assertEqual(executor.seen[0].get("repo"), + checkout_dir(repos, + "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")) + conn = sqlite3.connect(db_path) + try: + status = conn.execute( + "SELECT status FROM tasks WHERE id='t1'").fetchone()[0] + finally: + conn.close() + self.assertEqual(status, "completed") + tmp.cleanup() + + def test_no_repo_url_uses_default(self): + tmp = tempfile.TemporaryDirectory() + db_path, _ = _make_worker_db(tmp.name, "__unset__") + executor = _FakeExecutor() + worker = Worker(db_path, "host.py", repo="/default/repo", + env={"XCM_API_KEY": "test-placeholder"}, + env_key=None, executor=executor, + repos_dir=os.path.join(tmp.name, "repos"), + retry_backoff_seconds=0) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + # task 本身无 repo 字段：执行器走自身默认 repo，不被改写。 + self.assertNotIn("repo", executor.seen[0]) + tmp.cleanup() + + def test_repo_resolve_failure_is_failed_not_completed(self): + tmp = tempfile.TemporaryDirectory() + url = "https://github.com/o/private.git" + db_path, _ = _make_worker_db(tmp.name, url) + executor = _FakeExecutor() + worker = Worker(db_path, "host.py", repo=None, + env={"XCM_API_KEY": "test-placeholder"}, + env_key=None, executor=executor, + repos_dir=os.path.join(tmp.name, "repos"), + repos_runner=_git_runner_factory( + {"clone": (128, "", + "fatal: Authentication failed")}, + []), + retry_backoff_seconds=0) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertIn("仓库解析失败", result["error"]) + self.assertEqual(executor.seen, []) + conn = sqlite3.connect(db_path) + try: + status, error = conn.execute( + "SELECT status, error_text FROM tasks WHERE id='t1'" + ).fetchone() + finally: + conn.close() + self.assertEqual(status, "failed") + self.assertIn("仓库解析失败", error) + tmp.cleanup() + + +class ProjectRepoApiTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + # 老库：projects 表不含 repo_url 列，init_db 必须幂等补列。 + conn = sqlite3.connect(self.db_path) + conn.execute( + "CREATE TABLE projects(id TEXT PRIMARY KEY, name TEXT," + " owner_id TEXT, created_at REAL)") + conn.execute( + "CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT," + " prompt TEXT, status TEXT, idempotency_key TEXT," + " depends_on TEXT, priority INTEGER, created_by TEXT," + " created_at REAL, updated_at REAL, last_heartbeat REAL," + " UNIQUE(project_id, idempotency_key))") + conn.commit() + conn.close() + auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = auth.create_user("owner") + auth.conn.close() + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + payload = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Content-Type": "application/json"} + if key is not None: + headers["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=headers) + resp = conn.getresponse() + return resp.status, json.loads(resp.read() or b"{}") + finally: + conn.close() + + def test_old_db_migrated_and_create_with_repo_url(self): + conn = sqlite3.connect(self.db_path) + cols = {r[1] for r in conn.execute( + "PRAGMA table_info(projects)").fetchall()} + conn.close() + self.assertIn("repo_url", cols) + status, project = self._req( + "POST", "/projects", + {"name": "p1", + "repo_url": "https://github.com/openqj/codex-app-server.git"}, + self.owner_key) + self.assertEqual(status, 201, project) + self.assertEqual( + project["repo_url"], + "https://github.com/openqj/codex-app-server.git") + status, listed = self._req("GET", "/projects", key=self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(len(listed["projects"]), 1) + self.assertEqual( + listed["projects"][0]["repo_url"], + "https://github.com/openqj/codex-app-server.git") + + def test_create_without_repo_url_is_none(self): + status, project = self._req( + "POST", "/projects", {"name": "p2"}, self.owner_key) + self.assertEqual(status, 201, project) + self.assertIsNone(project["repo_url"]) + + def test_create_with_bad_repo_url_is_400_and_not_created(self): + for bad in ("https://user:pass@github.com/o/r.git", + "file:///tmp/r", "ssh://git@github.com/o/r.git", + "/tmp/local"): + status, body = self._req( + "POST", "/projects", + {"name": "bad", "repo_url": bad}, self.owner_key) + self.assertEqual(status, 400, (bad, body)) + status, listed = self._req("GET", "/projects", key=self.owner_key) + self.assertEqual(listed["projects"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 23f9f4f..9c01d97 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -82,6 +82,18 @@ interrupted，不用执行结果覆盖它。 按任务派生的容器名 docker kill，不只改数据库状态；本机无 docker 时 DockerExecutor 在调 runner 前就明确报错，任务标 failed 且 error_text 含 docker 缺失，绝不静默降级为本地执行。 + +项目仓库：run_one 派发前查任务所属项目的 repo_url（projects +表），非空时经 cloud.repos.resolve_repo 在 repos_dir（CLI +--repos-dir，默认 ./repos）下 clone/fetch 该项目仓库，设 +task["repo"] 再交给执行器；Local 与 Docker 执行器都已读 +task["repo"]，同一条路径。repo_url 为空时走原有默认 repo； +解析失败时任务直接 failed，错误文本写清是仓库解析失败。 +Docker 下 repos_dir 必须放在 extra-mounts 已挂载的目录下， +否则容器内 repo 路径不存在（见 cloud/README.md）。 +推送到远端不在本轮范围：任务在本地检出上执行、本地 worktree +合并仍按 host.py 现有流程，git push 需要仓库写凭证，控制面 +不存、不推。 """ import argparse @@ -100,6 +112,7 @@ from cloud.db import connect as _db_connect from cloud.logging_config import get_logger, log_event, setup_logging from cloud.models import ModelRouter, is_fallbackable, record_fallback from cloud.queue import budget_exceeded, ready_tasks +from cloud.repos import RepoError, resolve_repo from cloud.sandbox import DockerExecutor, LocalExecutor from cloud.sandbox import _default_runner as _sandbox_default_runner @@ -194,7 +207,8 @@ class Worker: max_turns=None, max_cost_usd=None, executor=None, cost_store=None, process_factory=None, task_timeout=1800, max_retries=1, - retry_backoff_seconds=5.0): + retry_backoff_seconds=5.0, repos_dir="./repos", + repos_runner=None): self.db_path = str(db_path) self.host_py_path = host_py_path self.config_path = config_path @@ -232,6 +246,12 @@ class Worker: self.task_timeout = task_timeout self.max_retries = max_retries self.retry_backoff_seconds = retry_backoff_seconds + # 项目仓库检出根目录：CLI --repos-dir 可调（默认 ./repos）； + # Docker 执行器下它必须放在 extra-mounts 已挂载的目录下， + # 否则容器内 task["repo"] 路径不存在——见 cloud/README.md。 + # repos_runner 可注入（单测用假 git runner），缺省走真实 git。 + self.repos_dir = repos_dir + self.repos_runner = repos_runner # -- 成本存储（惰性建表，避免只认领不执行时也建表）-------------- @property @@ -615,6 +635,53 @@ class Worker: "stdout": out or "", "stderr": err or "", "returncode": proc.returncode, "host_status": None} + # -- 项目仓库 ------------------------------------------------------ + def _project_repo_url(self, project_id): + """查任务所属项目的 repo_url；查不到项目、列缺失、表缺失时 + 一律返回 None（走原有默认 repo，不报错、不炸掉派发）。 + repo_url 本身只在 projects 表里，绝不从任务 prompt 里猜。""" + try: + conn = _db_connect(self.db_path) + try: + row = conn.execute( + "SELECT repo_url FROM projects WHERE id = ?", + (project_id,)).fetchone() + finally: + conn.close() + except Exception: + return None + if row is None: + return None + try: + url = row["repo_url"] + except (KeyError, IndexError, TypeError): + return None + return url or None + + def _apply_project_repo(self, task): + """派发前按项目 repo_url 解析仓库，返回错误文本或 None。 + + repo_url 为空：不动 task，返回 None（走原有默认 repo）。 + repo_url 非空：调 cloud.repos.resolve_repo 得到本地检出， + 设 task["repo"] 再交给执行器——LocalExecutor 与 + DockerExecutor 的 dispatch_task 都已读 task["repo"]，同一条 + 路径，不另起分支。resolve 失败返回错误文本，调用方据此把 + 任务置 failed（仓库解析失败绝不标 completed，也不在此重试， + 避免对不可用的仓库反复 clone 烧时间）。 + """ + repo_url = self._project_repo_url(task.get("project_id")) + if not repo_url: + return None + try: + 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: + return "仓库解析失败: %s" % exc + return None + # -- 执行 ---------------------------------------------------------- def _run_attempt(self, task): """单次执行尝试（不含认领、key 门、重试、成本与最终审计）。 @@ -735,6 +802,25 @@ class Worker: "stdout": "", "error": error, "events": [], "cost_record_id": None} + # 项目仓库解析：认领 + 缺 key 门之后、真正执行之前。 + # repo_url 为空时不动（默认 repo）；解析失败时任务直接 failed， + # 错误文本写清是仓库解析失败，绝不标 completed，也不在此重试。 + repo_error = self._apply_project_repo(task) + if repo_error is not None: + self._set_status(task_id, "failed") + self._write_execution_result( + task_id, error_text=repo_error, actual_model=self.model, + started_at=started_at, finished_at=time.time()) + self._audit(task["project_id"], "worker_repo_failed", repo_error) + log_event(self.logger, "worker_repo_failed", task_id=task_id, + project_id=task["project_id"], status="failed") + return {"dispatched": True, "task_id": task_id, + "host_name": self.executor._host_name(task) + if hasattr(self.executor, "_host_name") else task_id, + "status": "failed", "executor": self._executor_name(), + "stdout": "", "error": repo_error, "events": [], + "cost_record_id": None} + events = [] fallback_count = 0 result = None @@ -939,6 +1025,12 @@ def main(argv=None): help="host.py 路径，缺省为 cloud/ 上一级的 host.py") parser.add_argument("--config", default=None) parser.add_argument("--repo", default=None) + parser.add_argument("--repos-dir", default="./repos", + help="项目仓库检出根目录（默认 ./repos）；项目有" + " repo_url 时 worker 在此 clone/fetch 各项目" + "仓库；Docker 执行器下它必须放在" + " --docker-extra-mounts 已挂载的目录下，" + "否则容器内 task repo 路径不存在") parser.add_argument("--env-key", default=None) parser.add_argument("--codex-config", default=None, help="含 env_key 行的 codex 配置文件（只读名字）") @@ -988,7 +1080,8 @@ def main(argv=None): executor=_build_cli_executor(args, host_py), task_timeout=args.task_timeout, max_retries=args.max_retries, - retry_backoff_seconds=args.retry_backoff_seconds) + retry_backoff_seconds=args.retry_backoff_seconds, + repos_dir=args.repos_dir) # CLI 启动清理兜底（常驻与单次都做）：清掉上一轮残留容器。 worker.cleanup_stale_containers() # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。 diff --git a/deploy/caddy/Caddyfile b/deploy/caddy/Caddyfile new file mode 100644 index 0000000..e4805a5 --- /dev/null +++ b/deploy/caddy/Caddyfile @@ -0,0 +1,27 @@ +# 公网入口模板：只暴露 webhook 与 OAuth 回调，任务/项目/管理 API 不对外。 +# 使用前把 换成你自己的域名，把 换成 Let's Encrypt 联系邮箱。 +# 部署侧在主机上装 Caddy 后，把本文件放到 /etc/caddy/Caddyfile 并 reload。 +# 控制面本身仍只监听 127.0.0.1:8765，不直接对外；Caddy 只做反向代理 + TLS。 +# GitHub webhook 与 OAuth 回调都要求 https 且证书有效，Caddy 的自动证书满足这一点； +# 自签名证书不行，GitHub 会拒绝送达，浏览器也会拦截 OAuth 回调。 + + { + encode gzip + tls + + # GitHub webhook：签名校验在控制面做（GITHUB_WEBHOOK_SECRET），Caddy 只转发。 + @github path /webhooks/github + reverse_proxy @github 127.0.0.1:8765 + + # Slack webhook（目前只有解析，未接执行；先留着入口，执行接通后再用）。 + @slack path /webhooks/slack + reverse_proxy @slack 127.0.0.1:8765 + + # OAuth 浏览器登录/回调：回调 URL 必须与 OAuth App 里填的完全一致， + # 即 https:///oauth//callback。 + @oauth path /oauth/* + reverse_proxy @oauth 127.0.0.1:8765 + + # 其余一切（/tasks、/projects、/admin/*、/metrics 等）一律不对公网开放。 + respond "forbidden" 403 +} diff --git a/deploy/setup-webhook-secret.sh b/deploy/setup-webhook-secret.sh new file mode 100755 index 0000000..5cb3b39 --- /dev/null +++ b/deploy/setup-webhook-secret.sh @@ -0,0 +1,18 @@ +#!/bin/sh +# 在主机上生成 GitHub webhook secret 并写入 /etc/codex-app-server/env，绝不打印 secret。 +# 用法（主机 root 执行）：sh deploy/setup-webhook-secret.sh +# 效果：若 env 文件里已有 GITHUB_WEBHOOK_SECRET 则不动它并退出（避免轮换后 GitHub 端旧 secret 失效）； +# 否则生成 32 字节随机 hex，追加到 env 文件，重启 codex-cloud-server。 +# 生成的 secret 只存在主机 env 文件里（0600 root），不要把它发回聊天、不要写进仓库。 +set -eu +ENV_FILE=/etc/codex-app-server/env +if grep -q '^GITHUB_WEBHOOK_SECRET=' "$ENV_FILE" 2>/dev/null; then + echo "GITHUB_WEBHOOK_SECRET already set, leaving it unchanged." + exit 0 +fi +SECRET=$(openssl rand -hex 32) +printf 'GITHUB_WEBHOOK_SECRET=%s\n' "$SECRET" >> "$ENV_FILE" +chmod 600 "$ENV_FILE" +systemctl restart codex-cloud-server +echo "Webhook secret generated and stored in $ENV_FILE only; server restarted." +echo "Next step: paste the same secret into GitHub repo Settings -> Webhooks -> Secret (the value is on the host only, read it from $ENV_FILE as root if you need to copy it)."