From 9bb9a50e4e61c58b916f80278360859fe6be7543 Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 19:10:49 +0000 Subject: [PATCH] =?UTF-8?q?=E6=AE=8B=E7=95=99=E6=89=B9=E6=AC=A1A=EF=BC=9A?= =?UTF-8?q?=E6=88=90=E5=91=98=E5=88=A0=E9=99=A4=E3=80=81PR=20needs=5Flocal?= =?UTF-8?q?=5Fworker=20=E8=90=BD=E5=BA=93=E3=80=81=E6=8E=A8=E9=80=81=20DNS?= =?UTF-8?q?=20TOCTOU?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 成员删除：DELETE /projects/{id}/members/{user_id}，限 owner 或 全局管理员，禁删唯一 owner，记 remove_member 审计 （AuthStore.remove_member + server 分发，沿用 PATCH 守卫口径） - PR needs_local_worker 落库：非 local 扫到 queued 任务时 CAS 写行状态为 needs_local_worker；claim/find_open 同步认该状态， 标记过的行仍可被 local worker 认领、不被搁置；另修 claim/find_open 对裸 sqlite3 tuple 行的 dict() 兜底 （TypeError/ValueError） - 推送 DNS TOCTOU：发送改走建连时解析+判段的连接类，解析 一次、只拨判过的 IP，Host/SNI 仍为原域名；opener 直接构造 以排除 ProxyHandler 改写（build_opener 默认会塞代理） --- cloud/auth.py | 19 +++ cloud/pr_exec.py | 52 ++++++-- cloud/push.py | 137 ++++++++++++++++++-- cloud/server.py | 52 +++++++- cloud/tests/test_member_delete.py | 161 +++++++++++++++++++++++ cloud/tests/test_p1_security_worker.py | 10 +- cloud/tests/test_pr_jobs.py | 2 +- cloud/tests/test_pr_needs_local.py | 116 +++++++++++++++++ cloud/tests/test_push_toctou.py | 172 +++++++++++++++++++++++++ cloud/worker.py | 42 +++--- 10 files changed, 725 insertions(+), 38 deletions(-) create mode 100644 cloud/tests/test_member_delete.py create mode 100644 cloud/tests/test_pr_needs_local.py create mode 100644 cloud/tests/test_push_toctou.py diff --git a/cloud/auth.py b/cloud/auth.py index 56aa537..0b90cdd 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -307,6 +307,25 @@ class AuthStore: "%s:%s" % (target_user_id, role), ) + def remove_member(self, project_id, actor_user_id, target_user_id): + """删成员：只有 owner 能删，否则 raise PermissionError；同时记审计。""" + if self._member_role(project_id, actor_user_id) != "owner": + raise PermissionError( + "only project owner can change members" + ) + self.conn.execute( + "DELETE FROM project_members" + " WHERE project_id = ? AND user_id = ?", + (project_id, target_user_id), + ) + self.conn.commit() + self.audit( + actor_user_id, + project_id, + "remove_member", + target_user_id, + ) + def check_permission(self, user_id, project_id, action): """检查权限：满足最低角色返回 True，否则 False（含未加入项目）。""" min_role = _ACTION_MIN_ROLE.get(action) diff --git a/cloud/pr_exec.py b/cloud/pr_exec.py index 9715cf7..9d3c22a 100644 --- a/cloud/pr_exec.py +++ b/cloud/pr_exec.py @@ -54,18 +54,23 @@ def ensure_pr_jobs_table(conn): def find_open_pr_job(conn, task_id): - """找该任务未完成的 pr 任务（去重：同一任务同时只排一个）。""" + """找该任务未完成的 pr 任务（去重：同一任务同时只排一个）。 + + needs_local_worker 也算未完成：非 local worker 标记过的任务 + 仍在等 local worker 认领，再调 /pr 必须复用它，不能重排。""" ensure_pr_jobs_table(conn) row = conn.execute( "SELECT * FROM pr_jobs WHERE task_id = ?" - " AND status IN ('queued', 'running')" + " AND status IN ('queued', 'running', 'needs_local_worker')" " ORDER BY created_at ASC LIMIT 1", (task_id,)).fetchone() if row is None: return None try: return dict(row) - except TypeError: + except (TypeError, ValueError): + # 裸 sqlite3 连接（无 Row 工厂）返回 tuple：dict() 对 + # tuple 抛 ValueError（对 Row 抛 TypeError 的旧路径不动）。 return {"id": row[0], "task_id": row[1], "project_id": row[2], "host": row[3], "status": row[4]} @@ -84,34 +89,63 @@ def enqueue_pr_job(conn, task_id, project_id, host): def claim_pr_job(conn, host): - """CAS 认领本主机的一个 queued pr 任务（最老优先）。 + """CAS 认领本主机的一个 pr 任务（最老优先）。 - 两个同主机 worker 抢同一任务时只有一个能拿到 rowcount == 1。 + 认领 queued 与 needs_local_worker 两种状态：后者是非 local + worker 标记的“需 local worker”任务，仍在等 local worker， + 必须能被认领（否则标记等于搁置）。两个同主机 worker 抢 + 同一任务时只有一个能拿到 rowcount == 1。 返回任务 dict 或 None。 """ ensure_pr_jobs_table(conn) row = conn.execute( - "SELECT * FROM pr_jobs WHERE host = ? AND status = 'queued'" + "SELECT * FROM pr_jobs WHERE host = ?" + " AND status IN ('queued', 'needs_local_worker')" " ORDER BY created_at ASC LIMIT 1", (host,)).fetchone() if row is None: return None - job_id = row["id"] if hasattr(row, "__getitem__") else row[0] + try: + job_id = row["id"] + except TypeError: + # 裸 sqlite3 连接（无 Row 工厂）返回 tuple：按位置取。 + # hasattr(__getitem__) 判据对 tuple 也为真，不能用它区分。 + job_id = row[0] now = time.time() cur = conn.execute( "UPDATE pr_jobs SET status = 'running', started_at = ?," - " updated_at = ? WHERE id = ? AND status = 'queued'", + " updated_at = ? WHERE id = ?" + " AND status IN ('queued', 'needs_local_worker')", (now, now, job_id)) conn.commit() if cur.rowcount != 1: return None try: return dict(row) - except TypeError: + except (TypeError, ValueError): + # 同上：tuple 行的 dict() 抛 ValueError。 return {"id": row[0], "task_id": row[1], "project_id": row[2], "host": row[3], "status": "running"} +def mark_pr_job_needs_local_worker(conn, job_id, reason): + """把 queued 的 pr 任务标为 needs_local_worker（CAS）。 + + 只有仍是 queued 时才改写并返回 True；已被 local worker + 认领（running）或已终态时返回 False，绝不覆盖。标记后 + 任务仍可被 local worker 认领（见 claim_pr_job），只是 + 重读时能看到它在等 local worker，而不是“没人处理”。 + """ + ensure_pr_jobs_table(conn) + now = time.time() + cur = conn.execute( + "UPDATE pr_jobs SET status = 'needs_local_worker', error = ?," + " updated_at = ? WHERE id = ? AND status = 'queued'", + (reason, now, job_id)) + conn.commit() + return cur.rowcount == 1 + + def finish_pr_job(conn, job_id, status, pr_url=None, pr_number=None, error=None): """写回 pr 任务终态（completed/failed）。""" diff --git a/cloud/push.py b/cloud/push.py index 3fa2ba3..1835212 100644 --- a/cloud/push.py +++ b/cloud/push.py @@ -46,6 +46,8 @@ 密钥、token、证书一律不由本模块读取、写入或返回。 """ +import errno +import http.client import ipaddress import json import os @@ -285,13 +287,16 @@ def _resolve_channel_sender(sender, channel): _PUSH_SEND_TIMEOUT = 10 -def _post_json(url, payload): +def _post_json(url, payload, resolver=None): """POST JSON 到 url：返回 (ok, detail)，2xx 为 ok。 - HTTP 状态错误与网络异常一律返回 (False, reason)，不抛 - （调用方据此记 errors）；只有调用方传了非法 URL 时 - Request 构造本身可能抛 ValueError，由发送端先校验 scheme - 挡住，不走到这里。 + 建连时解析一次并判段、钉 IP 直连（见 _checked_opener）： + Host 头与 TLS SNI 仍为原域名，只把拨号地址换成判过的 + IP——预检解析与建连解析不一致时，建连侧的判定为准， + 绝不会拨向未验证的 IP。HTTP 状态错误与网络异常一律 + 返回 (False, reason)，不抛（调用方据此记 errors）； + 只有调用方传了非法 URL 时 Request 构造本身可能抛 + ValueError，由发送端先校验 scheme 挡住，不走到这里。 """ data = json.dumps(payload).encode("utf-8") request = urllib.request.Request( @@ -299,7 +304,7 @@ def _post_json(url, payload): headers={"Content-Type": "application/json", "User-Agent": "codex-app-server"}) try: - with urllib.request.urlopen( + with _checked_opener(resolver).open( request, timeout=_PUSH_SEND_TIMEOUT) as response: response.read() return True, "HTTP %s" % response.status @@ -309,6 +314,114 @@ def _post_json(url, payload): return False, "%s: %s" % (type(exc).__name__, exc) +def _pick_connect_ip(host, resolver): + """建连时解析一次并判段，返回要拨的 IP（只抛 OSError）。 + + 字面量 IP 不经过 DNS（只复核段）；域名解析一次： + allow-private 时取首个，否则取首个全局单播 IP；无可用 + IP 时抛 OSError。调用方拨的就是这里判过的 IP，之后 + 不再二次解析——校验与建连之间的 DNS 重绑定窗口就此 + 闭合（旧版 validate 后 urlopen 再解析一次的 TOCTOU）。 + """ + if _allow_private_targets(): + try: + ipaddress.ip_address(host) + return host + except ValueError: + pass + ips = (resolver or _default_resolver)(host) + if not ips: + raise OSError("DNS resolution failed for target host") + return ips[0] + try: + ipaddress.ip_address(host) + except ValueError: + pass + else: + if _ip_is_blocked(host): + raise OSError( + "target resolves to a non-public address: %s" % host) + return host + ips = (resolver or _default_resolver)(host) + if not ips: + raise OSError("DNS resolution failed for target host") + for ip_text in ips: + if not _ip_is_blocked(ip_text): + return ip_text + raise OSError("target host resolves to a non-public address") + + +def _checked_connection_class(base, resolver): + """建连时解析+判段的连接类：Host 头与 TLS SNI 仍为原域名 + （self.host 不动），只把拨号地址换成判过的 IP。 + + resolver 由调用方注入（测试 mock DNS 用）；为 None 时 + 用默认解析。代理隧道场景（_tunnel_host 已置）仍走基类： + 隧道里解析权在代理方，钉 IP 不适用，预检判定为准。 + """ + class CheckedConnection(base): + def connect(self): + if getattr(self, "_tunnel_host", None): + super().connect() + return + picked = _pick_connect_ip(self.host, resolver) + self.sock = socket.create_connection( + (picked, self.port), self.timeout, self.source_address) + try: + self.sock.setsockopt( + socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + except OSError as exc: + if exc.errno != errno.ENOPROTOOPT: + raise + if isinstance(self, http.client.HTTPSConnection): + self.sock = self._context.wrap_socket( + self.sock, server_hostname=self.host) + return CheckedConnection + + +class _CheckedHTTPHandler(urllib.request.HTTPHandler): + def __init__(self, resolver=None): + super().__init__() + self._resolver = resolver + + def http_open(self, req): + return self.do_open( + _checked_connection_class(http.client.HTTPConnection, + self._resolver), req) + + +class _CheckedHTTPSHandler(urllib.request.HTTPSHandler): + def __init__(self, resolver=None, debuglevel=0, context=None, + check_hostname=None): + super().__init__(debuglevel=debuglevel, context=context, + check_hostname=check_hostname) + self._resolver = resolver + + def https_open(self, req): + return self.do_open( + _checked_connection_class(http.client.HTTPSConnection, + self._resolver), req, + context=self._context, check_hostname=self._check_hostname) + + +def _checked_opener(resolver=None): + """发送用 opener：建连时解析+判段的连接类 + 重定向/错误处理。 + + 刻意直接构造 OpenerDirector、不用 build_opener：后者会 + 自动塞入 ProxyHandler，本机 egress 代理会把请求改写成 + 代理端口、钉 IP 失效（实测拨号跑到代理端口）。钉 IP 的 + 前提是直连；代理环境下发送会直连失败并记入 errors + （loud，不静默成功）。重定向仍跟随，但每一跳都重新 + 走建连判定（新目标同样解析+判段）。 + """ + opener = urllib.request.OpenerDirector() + opener.add_handler(_CheckedHTTPHandler(resolver)) + opener.add_handler(_CheckedHTTPSHandler(resolver)) + opener.add_handler(urllib.request.HTTPRedirectHandler()) + opener.add_handler(urllib.request.HTTPErrorProcessor()) + return opener + + def _require_http_url(target, channel): """发送端 target 校验：只接受 http(s) URL，其他一律 ValueError （dispatch 记入 errors，不静默成功、不乱 POST）。""" @@ -358,8 +471,11 @@ def validate_push_target(target, resolver=None, check_dns=True): 顺序：scheme（http/https）-> 主机名存在 -> 元数据主机名 -> IP 字面量直接判段 -> （check_dns 时）DNS 解析后逐 IP 判段， - 解析失败或解析出空结果都拒绝（DNS 重绑定下解析时与发送时 - 仍可能不一致，这是本层能做的上限，见残留说明）。 + 解析失败或解析出空结果都拒绝。本预检只是 fail-fast： + 真正起强制作用的是发送时建连类的解析+判段（解析一次、 + 只拨判过的 IP），预检解析与建连解析不一致时以建连侧 + 的判定为准，不会拨向未验证的 IP（旧版 validate 后 + urlopen 再解析一次的 TOCTOU 已闭合）。 check_dns=False 时只做字面量判定：供直接调 send_* 但不想 做 DNS 的调用方用；dispatch 的真实发送路径一律 check_dns=True，显式 sender 路径不走本预检（见 dispatch）。 @@ -424,7 +540,7 @@ def send_webhook(message): "project_id": message["project_id"], "task_id": payload.get("task_id"), "payload": message["payload"], - }) + }, resolver=message.get("_resolver")) if not ok: raise RuntimeError("webhook delivery failed: %s" % detail) @@ -450,7 +566,8 @@ def send_slack(message): message["project_id"]) if payload.get("error"): text += "\nerror: %s" % str(payload["error"])[:500] - ok, detail = _post_json(target, {"text": text}) + ok, detail = _post_json(target, {"text": text}, + resolver=message.get("_resolver")) if not ok: raise RuntimeError("slack delivery failed: %s" % detail) diff --git a/cloud/server.py b/cloud/server.py index 0c81c4f..94af7a0 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -2471,6 +2471,54 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(200, {"project_id": project_id, "user_id": target, "role": role}) + def _project_member_delete(self, user_id, project_id, target): + """DELETE /projects/{id}/members/{user_id}：限项目 owner + 或全局管理员。删唯一 owner 时 409（否则项目将无 owner + 可管）；目标非成员时 404。成功记 remove_member 审计。""" + if not self._project_exists(project_id): + self._send_json(404, {"error": "project not found"}) + return + if not self._can_manage_members(user_id, project_id): + self._send_json(403, {"error": "forbidden"}) + return + with _auth_store(self.server.db_path) as store: + cur = store.conn.execute( + "SELECT role FROM project_members" + " WHERE project_id = ? AND user_id = ?", + (project_id, target)).fetchone() + if cur is None: + self._send_json(404, {"error": "member not found"}) + return + if cur[0] == "owner": + owners = store.conn.execute( + "SELECT COUNT(*) FROM project_members" + " WHERE project_id = ? AND role = 'owner'", + (project_id,)).fetchone()[0] + if owners <= 1: + self._send_json( + 409, + {"error": "cannot remove the only owner"}) + return + self._remove_member_row(store, project_id, user_id, target) + self._send_json(200, {"project_id": project_id, + "user_id": target}) + + def _remove_member_row(self, store, project_id, actor_id, target_id): + """删成员行并记审计：owner 走 AuthStore.remove_member + （内含 owner 校验与审计）；非成员的全局管理员走直接删 + + 同格式审计（remove_member 的 owner 校验会挡住全局 + 管理员，故分流，与 _set_member_row 同款）。""" + if store.check_permission(actor_id, project_id, + "manage_project"): + store.remove_member(project_id, actor_id, target_id) + return + store.conn.execute( + "DELETE FROM project_members" + " WHERE project_id = ? AND user_id = ?", + (project_id, target_id)) + store.conn.commit() + store.audit(actor_id, project_id, "remove_member", target_id) + # -- 项目子资源：记忆 / 知识库 / 成本 ------------------------------- def _project_exists(self, project_id): conn = self._db() @@ -2510,10 +2558,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): else: self._send_json(404, {"error": "not found"}) return - # /projects/{id}/members/{user_id}：PATCH 改角色 + # /projects/{id}/members/{user_id}：PATCH 改角色 / DELETE 删成员 if len(parts) == 3 and parts[1] == "members": if method == "PATCH": self._project_member_patch(user_id, project_id, parts[2]) + elif method == "DELETE": + self._project_member_delete(user_id, project_id, parts[2]) else: self._send_json(404, {"error": "not found"}) return diff --git a/cloud/tests/test_member_delete.py b/cloud/tests/test_member_delete.py new file mode 100644 index 0000000..47ed8c4 --- /dev/null +++ b/cloud/tests/test_member_delete.py @@ -0,0 +1,161 @@ +"""成员删除端点单测（残留批次 A 任务 1）。 + +DELETE /projects/{id}/members/{user_id}： +- 限项目 owner 或全局管理员，其余 403； +- 目标非成员 404，项目不存在 404； +- 删唯一 owner 时 409（owner 与全局管理员均受此限）； +- 成功后成员列表无此人，且记 remove_member 审计。 + +测试用的 key、用户 id 全是假值；不联网、不碰生产。 +""" + +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.server import create_server +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.server import create_server + + +def _req(port, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) + headers = {"Content-Type": "application/json"} + if key: + headers["X-API-Key"] = key + conn.request(method, path, + json.dumps(body) if body is not None else None, headers) + resp = conn.getresponse() + raw = resp.read() + conn.close() + try: + data = json.loads(raw) + except ValueError: + data = None + return resp.status, data + + +class MemberDeleteTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.trace_dir = os.path.join(self.tmp.name, "host-traces") + os.makedirs(self.trace_dir, exist_ok=True) + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.maint_id, self.maint_key = self.auth.create_user("maint") + self.member_id, self.member_key = self.auth.create_user("member") + self.admin_id, self.admin_key = self.auth.create_user("admin") + self.outsider_id, self.out_key = self.auth.create_user("outsider") + self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") + os.environ["CODEX_GLOBAL_ADMINS"] = self.admin_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.server.trace_dir = self.trace_dir + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, proj = _req(self.port, "POST", "/projects", + {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201, proj) + self.project_id = proj["id"] + self.auth.set_member(self.project_id, self.owner_id, + self.maint_id, "maintainer") + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + if self._old_admins is None: + os.environ.pop("CODEX_GLOBAL_ADMINS", None) + else: + os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins + + def _delete(self, target, key): + return _req(self.port, "DELETE", + "/projects/%s/members/%s" % (self.project_id, target), + key=key) + + def _invite(self, target, role="contributor", key=None): + return _req(self.port, "POST", + "/projects/%s/members" % self.project_id, + {"user_id": target, "role": role}, + key or self.owner_key) + + def _member_ids(self, key=None): + status, data = _req(self.port, "GET", + "/projects/%s/members" % self.project_id, + key=key or self.owner_key) + self.assertEqual(status, 200, data) + return {m["user_id"] for m in data["members"]} + + def _audit_actions(self): + return [(a["action"], a["detail"]) + for a in self.auth.list_audit(self.project_id)] + + def test_owner_delete_member_and_audit(self): + self.assertEqual(self._invite(self.member_id)[0], 201) + status, data = self._delete(self.member_id, self.owner_key) + self.assertEqual(status, 200, data) + self.assertEqual(data["user_id"], self.member_id) + self.assertNotIn(self.member_id, self._member_ids()) + self.assertIn(("remove_member", self.member_id), + self._audit_actions()) + + def test_non_owner_and_outsider_denied(self): + self.assertEqual(self._invite(self.member_id)[0], 201) + status, _ = self._delete(self.member_id, self.maint_key) + self.assertEqual(status, 403) + status, _ = self._delete(self.member_id, self.out_key) + self.assertEqual(status, 403) + # 拒绝后成员仍在。 + self.assertIn(self.member_id, self._member_ids()) + + def test_global_admin_delete_without_membership(self): + self.assertEqual(self._invite(self.member_id)[0], 201) + status, data = self._delete(self.member_id, self.admin_key) + self.assertEqual(status, 200, data) + self.assertNotIn(self.member_id, self._member_ids()) + self.assertIn(("remove_member", self.member_id), + self._audit_actions()) + + def test_delete_non_member_and_missing_project(self): + status, _ = self._delete("ghost", self.owner_key) + self.assertEqual(status, 404) + status, _ = _req(self.port, "DELETE", + "/projects/no-such/members/%s" % self.member_id, + key=self.owner_key) + self.assertEqual(status, 404) + + def test_cannot_remove_only_owner(self): + status, data = self._delete(self.owner_id, self.owner_key) + self.assertEqual(status, 409, data) + status, _ = self._delete(self.owner_id, self.admin_key) + self.assertEqual(status, 409) + # 升第二位 owner 后，删原 owner 放行。 + self.assertEqual(self._invite(self.member_id)[0], 201) + status, _ = _req(self.port, "PATCH", + "/projects/%s/members/%s" + % (self.project_id, self.member_id), + {"role": "owner"}, self.owner_key) + self.assertEqual(status, 200) + status, _ = self._delete(self.owner_id, self.maint_key) + # maint 不是 owner 也不是全局管理员：仍 403（守卫先行）。 + self.assertEqual(status, 403) + status, data = self._delete(self.owner_id, self.member_key) + self.assertEqual(status, 200, data) + self.assertNotIn(self.owner_id, self._member_ids(self.member_key)) diff --git a/cloud/tests/test_p1_security_worker.py b/cloud/tests/test_p1_security_worker.py index 8cf2d4c..3bfa293 100644 --- a/cloud/tests/test_p1_security_worker.py +++ b/cloud/tests/test_p1_security_worker.py @@ -449,7 +449,8 @@ class P1WorkerTestCase(unittest.TestCase): self.assertEqual(_status(self.db_path, "T1"), "failed") self.assertEqual(self.calls, []) - # 15. PR 非 local：显式 needs_local_worker，任务与排队都不动 + # 15. PR 非 local：显式 needs_local_worker，行状态持久化， + # 任务本身不动，标记过的行仍可被 local worker 认领 def test_pr_non_local_needs_local_worker(self): conn = sqlite3.connect(self.db_path) try: @@ -470,9 +471,14 @@ class P1WorkerTestCase(unittest.TestCase): row = conn.execute( "SELECT status FROM pr_jobs WHERE id = ?", (job_id,)).fetchone() + self.assertEqual(row[0], "needs_local_worker") + # 标记过的行仍可被 local worker 认领，不会被搁置。 + from cloud.pr_exec import claim_pr_job + claimed = claim_pr_job(conn, "test-host") + self.assertIsNotNone(claimed) + self.assertEqual(claimed["id"], job_id) finally: conn.close() - self.assertEqual(row[0], "queued") # 16. 多文件签名 + 签名缺失/已变时认领 fail-closed def test_code_signature_multi_file_and_claim_refusal(self): diff --git a/cloud/tests/test_pr_jobs.py b/cloud/tests/test_pr_jobs.py index 5701207..9c6a672 100644 --- a/cloud/tests/test_pr_jobs.py +++ b/cloud/tests/test_pr_jobs.py @@ -308,7 +308,7 @@ class WorkerPrJobsTestCase(unittest.TestCase): finally: conn.close() # P1：Docker worker 不再静默返回空，改为显式 needs_local_worker， - # 任务保持 queued 等 local worker 认领。 + # 行状态持久化为 needs_local_worker、仍可被 local worker 认领。 jobs = worker._run_pr_jobs() self.assertEqual(len(jobs), 1) self.assertEqual(jobs[0]["status"], "needs_local_worker") diff --git a/cloud/tests/test_pr_needs_local.py b/cloud/tests/test_pr_needs_local.py new file mode 100644 index 0000000..d4cd40a --- /dev/null +++ b/cloud/tests/test_pr_needs_local.py @@ -0,0 +1,116 @@ +"""PR needs_local_worker 持久化单测（残留批次 A 任务 2）。 + +- 非 local worker 扫到 queued 的 pr 任务后，行状态持久化为 + needs_local_worker（重读仍在），且返回条目状态一致； +- 标记过的行仍可被 local worker 认领（claim 后 running，可 + 正常 finish），不会被搁置； +- find_open_pr_job 仍把标记过的行算未完成（/pr 不重排）； +- 标记是 CAS：已 running 的行不被覆盖；重复扫不重复记。 + +测试用的 key、主机名、任务 id 全是假值；不联网、不碰生产。 +""" + +import os +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.db import connect as _db_connect + from cloud.pr_exec import ( + claim_pr_job, enqueue_pr_job, find_open_pr_job, + mark_pr_job_needs_local_worker) + from cloud.sandbox import DockerExecutor + 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.db import connect as _db_connect + from cloud.pr_exec import ( + claim_pr_job, enqueue_pr_job, find_open_pr_job, + mark_pr_job_needs_local_worker) + from cloud.sandbox import DockerExecutor + from cloud.worker import Worker + + +class PrNeedsLocalTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + + def _enqueue_t1(self, host="host-a"): + conn = _db_connect(self.db_path) + try: + enqueue_pr_job(conn, "t1", "p1", host) + finally: + conn.close() + + def tearDown(self): + self.tmp.cleanup() + + def _row(self, task_id): + conn = _db_connect(self.db_path) + try: + return conn.execute( + "SELECT id, status, error FROM pr_jobs" + " WHERE task_id = ?", (task_id,)).fetchone() + finally: + conn.close() + + def test_docker_peek_persists_mark(self): + worker = Worker(self.db_path, "/nonexistent/host.py", + env_key="TEST_KEY", + executor=DockerExecutor("/fake/host.py")) + conn = _db_connect(self.db_path) + try: + enqueue_pr_job(conn, "t1", "p1", worker.host_name) + finally: + conn.close() + jobs = worker._run_pr_jobs() + self.assertEqual(len(jobs), 1) + self.assertEqual(jobs[0]["status"], "needs_local_worker") + self.assertEqual(jobs[0]["task_id"], "t1") + # 重读：行状态已持久化，不只是返回值里有。 + row = self._row("t1") + self.assertEqual(row[1], "needs_local_worker") + self.assertIn("local worker", row[2]) + # 再扫一次：已标记的行不再是 queued，不重复返回、不重复记。 + self.assertEqual(worker._run_pr_jobs(), []) + + def test_marked_row_still_claimable_by_local(self): + self._enqueue_t1() + conn = _db_connect(self.db_path) + try: + job_id = conn.execute( + "SELECT id FROM pr_jobs WHERE task_id = 't1'" + ).fetchone()[0] + self.assertTrue(mark_pr_job_needs_local_worker( + conn, job_id, "needs local worker")) + # 标记后仍算未完成：/pr 去重不受影响。 + self.assertIsNotNone(find_open_pr_job(conn, "t1")) + # local worker 认领标记过的行。 + claimed = claim_pr_job(conn, "host-a") + self.assertIsNotNone(claimed) + self.assertEqual(claimed["task_id"], "t1") + row = conn.execute( + "SELECT status FROM pr_jobs WHERE task_id = 't1'" + ).fetchone() + self.assertEqual(row[0], "running") + finally: + conn.close() + + def test_mark_is_cas_on_queued_only(self): + self._enqueue_t1() + conn = _db_connect(self.db_path) + try: + claimed = claim_pr_job(conn, "host-a") + self.assertIsNotNone(claimed) + # 已 running：标记必须失败、不覆盖状态。 + self.assertFalse(mark_pr_job_needs_local_worker( + conn, claimed["id"], "late mark")) + row = conn.execute( + "SELECT status FROM pr_jobs WHERE task_id = 't1'" + ).fetchone() + self.assertEqual(row[0], "running") + finally: + conn.close() diff --git a/cloud/tests/test_push_toctou.py b/cloud/tests/test_push_toctou.py new file mode 100644 index 0000000..b4138dd --- /dev/null +++ b/cloud/tests/test_push_toctou.py @@ -0,0 +1,172 @@ +"""推送 DNS TOCTOU 单测（残留批次 A 任务 3）。 + +旧版：validate_push_target 解析一次，urlopen 建连时又解析 +一次；两次解析结果不一致（DNS 重绑定）时可能拨向未验证 +的内网 IP。新版：建连时解析一次并判段、只拨判过的 IP， +Host 头与 TLS SNI 仍为原域名。 + +- 两次解析不一致（先公网、后内网）时：发送失败，且一次 + 都没有拨向内网 IP； +- 混合解析结果（先内网、后公网）时：只拨公网 IP； +- HTTP 真服务器：域名经 resolver 钉到 127.0.0.1，服务器 + 收到的 Host 仍为原域名； +- HTTPS 连接类：拨号地址为判过的 IP，wrap_socket 的 + server_hostname 仍为原域名（SNI 不变）。 + +测试用的 IP、域名、key 全是假值；除本地回环服务器外不联网。 +""" + +import http.client +import http.server +import os +import sys +import threading +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.push import ( + _checked_connection_class, _post_json, send_webhook) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.push import ( + _checked_connection_class, _post_json, send_webhook) + +PUBLIC_IP = "8.8.8.8" +PRIVATE_IP = "169.254.169.254" + + +class _FlipFlopResolver: + """第一次返回公网 IP，之后返回内网 IP：模拟重绑定。 + + 旧版两次解析恰好踩中这个窗口；新版第二次解析判段后 + 直接拒绝，绝不拨号。 + """ + + def __init__(self): + self.calls = 0 + + def __call__(self, host): + self.calls += 1 + return [PUBLIC_IP] if self.calls == 1 else [PRIVATE_IP] + + +class _DialRecorder: + """记录拨号地址后直接失败：测试只断言拨向，不跑完 HTTP。""" + + def __init__(self): + self.calls = [] + + def __call__(self, address, timeout=None, source_address=None): + self.calls.append(address) + raise OSError("dial stopped by test") + + +class _DummySocket: + def setsockopt(self, *args): + pass + + +class _RecordingHandler(http.server.BaseHTTPRequestHandler): + requests = [] + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length) if length else b"" + _RecordingHandler.requests.append({ + "host": self.headers.get("Host"), + "body": body, + }) + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, *args): + pass + + +class PushToctouTestCase(unittest.TestCase): + def setUp(self): + self._old_allow = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + _RecordingHandler.requests = [] + + def tearDown(self): + if self._old_allow is None: + os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) + else: + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = self._old_allow + + def _message(self, target, resolver=None): + msg = {"event_type": "completed", "project_id": "proj-a", + "payload": {"task_id": "t-1"}, "target": target} + if resolver is not None: + msg["_resolver"] = resolver + return msg + + def test_rebinding_second_resolution_never_dialed(self): + resolver = _FlipFlopResolver() + dials = _DialRecorder() + with mock.patch("socket.create_connection", dials): + with self.assertRaises(RuntimeError): + send_webhook(self._message( + "http://hook.example.invalid/hook", resolver)) + # 两次解析确实都发生了（一次预检、一次建连），但建连侧 + # 判段后拒绝：内网 IP 一次都没被拨。 + self.assertEqual(resolver.calls, 2) + self.assertEqual(dials.calls, []) + + def test_mixed_resolution_dials_only_public(self): + dials = _DialRecorder() + with mock.patch("socket.create_connection", dials): + ok, _detail = _post_json( + "http://hook.example.invalid/hook", {"a": 1}, + resolver=lambda host: [PRIVATE_IP, PUBLIC_IP]) + self.assertFalse(ok) + self.assertEqual(dials.calls, [(PUBLIC_IP, 80)]) + + def test_http_pinned_host_header_preserved(self): + os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" + server = http.server.HTTPServer(("127.0.0.1", 0), + _RecordingHandler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + send_webhook(self._message( + "http://webhook.test.invalid:%d/hook" % port, + resolver=lambda host: ["127.0.0.1"])) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + self.assertEqual(len(_RecordingHandler.requests), 1) + host_header = _RecordingHandler.requests[0]["host"] + # 拨的是 127.0.0.1，但 Host 仍为原域名（虚拟主机依赖它）。 + self.assertEqual(host_header.split(":")[0], + "webhook.test.invalid") + + def test_https_sni_preserved_with_pinned_ip(self): + wrap_calls = [] + + class FakeContext: + def wrap_socket(self, sock, server_hostname=None): + wrap_calls.append(server_hostname) + return sock + + cls = _checked_connection_class(http.client.HTTPSConnection, + lambda host: [PUBLIC_IP]) + conn = cls("example.com", 443) + conn._context = FakeContext() + dials = [] + + def fake_create(address, timeout=None, source_address=None): + dials.append(address) + return _DummySocket() + + with mock.patch("socket.create_connection", fake_create): + conn.connect() + self.assertEqual(dials, [(PUBLIC_IP, 443)]) + self.assertEqual(wrap_calls, ["example.com"]) diff --git a/cloud/worker.py b/cloud/worker.py index e4b28b4..cf1ba8a 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -153,7 +153,8 @@ from cloud.sub2api import gateway_name as _gateway_name from cloud.sub2api import gateway_origin as _gateway_origin from cloud.sub2api import list_gateway_models as _list_gateway_models from cloud.pr import PrSkipped, PrStepError -from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job) +from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job, + mark_pr_job_needs_local_worker) from cloud.queue import budget_exceeded, ready_tasks from cloud.repos import RepoError, checkout_dir, resolve_repo from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor @@ -3547,19 +3548,15 @@ class Worker: P1 非 local 通道：旧版非 local worker 直接返回 []， 排队的 pr 任务看起来像“没人处理”且无任何痕迹—— 静默未完成。现在改为显式 needs_local_worker：只读 - peek 本主机 queued 的 pr 任务，逐条记审计 + 日志并 - 返回 needs_local_worker 条目，任务本身保持 merged、 - pr 任务保持 queued（留给 local worker 认领），绝不 - 把它标 failed（failed 会让调用方以为 PR 已失败，且 - 任务仍可重调 /pr 的语义会被混淆）。为什么不直接把 - pr_jobs 行写成 needs_local_worker 状态：pr_jobs 的认 - 领只认 queued（claim_pr_job 在 pr_exec.py，本路不改 - 该文件），写成新状态后 local worker 也认领不到，任务 - 会被永久搁置；真要落库状态，需 pr_exec.py 的认领同 - 步改成含 needs_local_worker 或 local 重排队——该改动 - 留给协调员（见汇报残留），本路不伪造一个“写了状态 - 却没人能接”的通道。签名无效时同样不碰任务（返回 []， - 只记日志），fail-closed。 + peek 本主机 queued 的 pr 任务，逐条把行状态持久化 + 为 needs_local_worker（CAS，只改仍是 queued 的行， + 已被 local worker 认领的不覆盖），并记审计 + 日志 + 返回 needs_local_worker 条目。任务本身保持 merged， + pr 任务保持可被 local worker 认领（claim_pr_job + 同时认 queued 与 needs_local_worker），绝不把它标 + failed（failed 会让调用方以为 PR 已失败，且任务仍 + 可重调 /pr 的语义会被混淆）。签名无效时同样不碰 + 任务（返回 []，只记日志），fail-closed。 """ if not self._code_signature_valid(): log_event(self.logger, "worker_pr_refused_code_signature", @@ -3588,8 +3585,23 @@ class Worker: continue reason = ("PR 执行需要 local worker（host-worktrees" " 只在主机上可见），本 worker 为 %s，" - "任务保持 queued 等 local worker 认领" + "任务保持可认领等 local worker 认领" % self._executor_name()) + marked = False + try: + conn = _db_connect(self.db_path) + try: + marked = mark_pr_job_needs_local_worker( + conn, job_id, reason) + finally: + conn.close() + except Exception as exc: + log_event(self.logger, "pr_job_mark_failed", + message="pr job mark failed: %s" % exc) + continue + if not marked: + # 已被 local worker 认领或已终态：不覆盖、不重复记。 + continue self._audit(project_id, "pr_job_needs_local_worker", "%s: %s" % (job_id, reason)) log_event(self.logger, "pr_job_needs_local_worker",