diff --git a/cloud/github_app.py b/cloud/github_app.py
index 085c4a2..a61ade9 100644
--- a/cloud/github_app.py
+++ b/cloud/github_app.py
@@ -92,9 +92,12 @@ def build_app_manifest(public_base, app_name=DEFAULT_APP_NAME,
class SetupStateStore:
"""setup state 存储：一次性、1 小时过期，防回调被冒用。
- 表 github_app_setup_states：state（主键）、created_at。
- state 只由持本站 key 调 POST /github-app/setup/state 的
- 操作方签发；回调验到即删，同一 state 第二次回调必然失败。
+ 表 github_app_setup_states：state（主键）、created_at、
+ issued_by（签发人用户 id，P0 安全批新增）。新表建表即
+ 含该列；老库在 __init__ 里用 ALTER 补列（列已存在时
+ 报错直接忽略，老库不炸，见 except 分支）。
+ state 只由全局管理员调 POST /github-app/setup/state
+ 签发；回调验到即删，同一 state 第二次回调必然失败。
"""
def __init__(self, db_path):
@@ -102,38 +105,86 @@ class SetupStateStore:
self.conn = _db_connect(self.db_path)
self.conn.execute(
"CREATE TABLE IF NOT EXISTS github_app_setup_states("
- "state TEXT PRIMARY KEY, created_at REAL)"
+ "state TEXT PRIMARY KEY, created_at REAL,"
+ " issued_by TEXT)"
)
self.conn.commit()
+ try:
+ self.conn.execute(
+ "ALTER TABLE github_app_setup_states"
+ " ADD COLUMN issued_by TEXT")
+ self.conn.commit()
+ except Exception:
+ try:
+ self.conn.rollback()
+ except Exception:
+ pass
- def create_state(self):
+ def create_state(self, issued_by=None):
state = secrets.token_urlsafe(32)
self.conn.execute(
- "INSERT INTO github_app_setup_states(state, created_at)"
- " VALUES(?, ?)",
- (state, time.time()),
+ "INSERT INTO github_app_setup_states"
+ "(state, created_at, issued_by)"
+ " VALUES(?, ?, ?)",
+ (state, time.time(), issued_by),
)
self.conn.commit()
return state
+ def state_issuer(self, state):
+ """查 state 的签发人（回调审计用，不消耗 state）。"""
+ if not state:
+ return None
+ try:
+ row = self.conn.execute(
+ "SELECT issued_by FROM github_app_setup_states"
+ " WHERE state = ?",
+ (state,),
+ ).fetchone()
+ except Exception:
+ return None
+ if row is None:
+ return None
+ try:
+ return row[0]
+ except Exception:
+ return None
+
def consume_state(self, state, now=None,
- ttl_seconds=SETUP_STATE_TTL_SECONDS):
+ ttl_seconds=SETUP_STATE_TTL_SECONDS,
+ allowed_issuers=None):
"""校验并消耗 state：成功返回 True，失败返回 False。
不存在 -> False；过期 -> 删行并返回 False；成功 -> 删行
并返回 True。``now`` 只为单测注入时间。
+ P0 安全批后：allowed_issuers 非空时，签发人不在该
+ 集合里一律返回 False 且不消耗 state（fail-closed）；
+ 签发人为空（升级前的旧 state）同样不在集合里，等
+ 于升级后旧 state 全部作废，需管理员重新签发。注意：
+ 回调本身是浏览器无认证重定向，服务器无法从回调请求
+ 里认出调用方就是签发人，所以这里核对的是“签发人仍
+ 在管理员清单里”，不是“回调者就是签发人”——后者
+ 在本流程里做不到，文档与 PR 描述里必须如实写清，
+ 不能谎称回调做了身份认证。
"""
if not state:
return False
if now is None:
now = time.time()
row = self.conn.execute(
- "SELECT created_at FROM github_app_setup_states"
+ "SELECT created_at, issued_by FROM github_app_setup_states"
" WHERE state = ?",
(state,),
).fetchone()
if row is None:
return False
+ if allowed_issuers is not None:
+ try:
+ issuer = row[1]
+ except Exception:
+ issuer = None
+ if issuer not in allowed_issuers:
+ return False
self.conn.execute(
"DELETE FROM github_app_setup_states WHERE state = ?",
(state,),
diff --git a/cloud/migrations.py b/cloud/migrations.py
index 49a3864..a6ae120 100644
--- a/cloud/migrations.py
+++ b/cloud/migrations.py
@@ -39,6 +39,10 @@ MIGRATION_TABLES = (
"oauth_states",
"slack_users",
"push_subscriptions",
+ # P0 安全批：配额行必须随库迁移，否则迁后用户变回无行
+ # （本批后无行按默认额度处理，不再是无限额，但显式
+ # 管理员额度仍需原样迁过去，不能丢）。
+ "user_quotas",
)
_SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"})
diff --git a/cloud/server.py b/cloud/server.py
index f761727..c7c2025 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -146,6 +146,7 @@ from cloud.github_app import (
)
from cloud.webhooks import (
WebhookDeliveryStore,
+ is_trusted_fix_review_author,
parse_github_check_run,
parse_github_pr_comment,
parse_slack_command,
@@ -979,11 +980,12 @@ def _default_quota_usd_micros():
def _apply_default_quota(conn, user_id):
- """给新用户写默认配额行（已存在则不动，幂等）。
+ """给用户写默认配额行（已存在则不动，幂等）。
- 只在 OAuth 建号（created=True）时调：INSERT OR IGNORE
- 保证重复回调不覆盖管理员后来改过的额度。返回 True 表示
- 本次写了新行（调用方据此记 quota_default 审计）。
+ INSERT OR IGNORE 保证重复回调不覆盖管理员后来改过的额度。
+ 返回 True 表示本次写了新行（调用方据此记 quota_default
+ 审计）。P0 安全批后：OAuth 新建、绑定、已存在用户登录
+ 三条分支都调它补行，不再只给新建用户补。
"""
_ensure_quotas_table(conn)
cur = conn.execute(
@@ -997,6 +999,43 @@ def _apply_default_quota(conn, user_id):
return cur.rowcount == 1
+def _global_admin_ids():
+ """真全局管理员清单：环境变量 CODEX_GLOBAL_ADMINS。
+
+ 逗号分隔的用户 id 列表，空白项忽略。未配置或为空时返回
+ 空集——此时 admin 端点一律 fail-closed，绝不回落到项目
+ 角色推导（P0 安全批前“任一项目 maintainer 即全局管理”
+ 的推导已被废掉，见 _is_global_admin）。
+ """
+ raw = os.environ.get("CODEX_GLOBAL_ADMINS") or ""
+ return {part.strip() for part in raw.split(",") if part.strip()}
+
+
+def _backfill_default_quotas(conn):
+ """给已有无行用户回填默认配额行（init_db 时调，幂等）。
+
+ 只补“users 里有、user_quotas 里无”的用户，已有行（含
+ 管理员显式设为 NULL 的不限额行）一律不动。返回补写行数。
+ """
+ _ensure_quotas_table(conn)
+ try:
+ cur = conn.execute(
+ "INSERT OR IGNORE INTO user_quotas(user_id,"
+ " quota_usd_micros, updated_at)"
+ " SELECT u.id, ?, ? FROM users u"
+ " LEFT JOIN user_quotas q ON q.user_id = u.id"
+ " WHERE q.user_id IS NULL",
+ (_default_quota_usd_micros(), time.time()))
+ conn.commit()
+ return cur.rowcount or 0
+ except Exception:
+ try:
+ conn.rollback()
+ except Exception:
+ pass
+ return 0
+
+
_ENV_BRANCH_RE = re.compile(r"^[A-Za-z0-9_.\-/]+$")
@@ -1058,6 +1097,7 @@ def init_db(db_path):
_ensure_followups_table(conn)
_ensure_environments_table(conn)
_ensure_quotas_table(conn)
+ _backfill_default_quotas(conn)
_ensure_gateway_keys(conn)
_ensure_uploads_tables(conn)
conn.commit()
@@ -1091,6 +1131,7 @@ def init_db(db_path):
_ensure_followups_table(conn)
_ensure_environments_table(conn)
_ensure_quotas_table(conn)
+ _backfill_default_quotas(conn)
_ensure_uploads_tables(conn)
# worker 每轮 poll 都按 status 取任务（见 load_tasks 的
# WHERE status IN (...)），无索引时是全表扫描；任务表随
@@ -3005,6 +3046,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
user_id, task["project_id"], "create_task"):
self._send_json(403, {"error": "forbidden"})
return
+ if self._reject_if_over_quota_for_task(user_id, task):
+ return
text = body.get("body")
if not text or not isinstance(text, str) or not text.strip():
self._send_json(400, {"error": "body required"})
@@ -3143,6 +3186,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
user_id, sender["project_id"], "read"):
self._send_json(403, {"error": "forbidden"})
return
+ if self._reject_if_over_quota_for_task(user_id, target):
+ return
text = body.get("body")
if not text or not isinstance(text, str) or not text.strip():
self._send_json(400, {"error": "body required"})
@@ -3680,6 +3725,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
user_id, task["project_id"], "create_task"):
self._send_json(403, {"error": "forbidden"})
return
+ if self._reject_if_over_quota_for_task(user_id, task):
+ return
if task["status"] in ("queued", "running"):
self._send_json(409, {
"error": "进行中任务请用后续消息继续",
@@ -3812,8 +3859,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
interrupted / merge-failed（其它状态一律 409：running
本来就在跑，queued 本来就在等，completed/merged 已有
结果，重跑应是新任务而不是覆盖结果）。
- 配额按任务创建者查（费用记在创建者名下）：维护者替
- 他人重试时，操作人超限不应拦、创建者超限才应拦。
+ 配额按双人门查（P0 安全批，见
+ _reject_if_over_quota_for_task）：触发人与原任务创建
+ 人任一超限即拦。费用仍记在创建者名下，所以只查创建
+ 者不够——触发人用自己的额度过门、费用记他人名下，
+ 仍是绕限；只查触发人也不够——原创建者已超限时，
+ 替他人重试仍会把费用记到已超限的账上。
成功后状态回 queued，上一轮的 result_text/error_text/
actual_model 清空（否则新一轮的结果会和旧结果混在一起
无法分辨）；prompt、模型、优先级、依赖原样保留，依赖
@@ -3832,8 +3883,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
user_id, task["project_id"], "approve_merge"):
self._send_json(403, {"error": "forbidden"})
return
- if self._reject_if_over_quota(
- user_id, quota_user_id=task.get("created_by") or user_id):
+ if self._reject_if_over_quota_for_task(user_id, task):
return
if task["status"] not in ("failed", "interrupted", "merge-failed"):
self._send_json(409, {
@@ -3882,9 +3932,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _quota_status(self, user_id):
"""返回 (quota_usd_micros|None, used_usd_micros, month)。
- quota 为 None 表示不限额：老用户与未设配额的用户默认
- 不限额，行为与过去一致，不会因为上了配额功能就突然
- 被挡住。"""
+ P0 安全批后：无配额行不再等于不限额——缺行时按默认
+ 配额处理（与 _default_quota_usd_micros() 同值，只在
+ 返回值里合成，不在这里写行；写行由 _apply_default_quota
+ / _backfill_default_quotas 负责）。只有管理员显式把
+ 行写成 NULL 时，quota 才为 None（不限额）。"""
month = self._quota_month()
conn = self._db()
try:
@@ -3902,7 +3954,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
"SELECT quota_usd_micros FROM user_quotas"
" WHERE user_id = ?",
(user_id,)).fetchone()
- quota = row[0] if row is not None else None
+ if row is None:
+ quota = _default_quota_usd_micros()
+ else:
+ quota = row[0]
used = conn.execute(
"SELECT COALESCE(SUM(c.usd_micros), 0)"
" FROM cost_records c JOIN tasks t ON c.task_id = t.id"
@@ -3959,22 +4014,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _check_unpriced_token_cap(self, user_id):
"""无价 token 上限准入：超限返回 (cap, used, month)，否则 None。
- 只有有配额行的用户才受限（无配额行 = 不限额，与 USD
- 口径一致）；USD 配额本身不计量无价模型，所以这道是
- 补 USD 口径的洞，不是第二套计费。"""
+ P0 安全批后：无配额行用户同样受限（无价 token 上限与
+ 配额行存在与否无关）；USD 配额本身不计量无价模型，所
+ 以这道是补 USD 口径的洞，不是第二套计费。"""
month = self._quota_month()
- conn = self._db()
- try:
- _ensure_quotas_table(conn)
- conn.commit()
- row = conn.execute(
- "SELECT quota_usd_micros FROM user_quotas"
- " WHERE user_id = ?",
- (user_id,)).fetchone()
- if row is None:
- return None
- finally:
- conn.close()
used = self._quota_unpriced_tokens(user_id, month)
cap = unpriced_token_cap()
if used >= cap:
@@ -3987,9 +4030,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
太快，所以不用 429。
准入点：建任务 / 分支 / 重试 / 续跑，均为权限校验
- 通过后、写库之前。后续消息（followup）创建不查：
- 排队本身不花钱，执行时轮次边界的配额门照常生效，
- 不要顺手把 followup 也改成准入校验。
+ 通过后、写库之前。P0 安全批后，后续消息与任务间
+ 消息创建时同样查双人配额门（见
+ _reject_if_over_quota_for_task）：排队本身不花钱，
+ 但续跑/复活必然花钱，只靠执行时轮次边界门拦不住
+ “超限用户靠他人任务续命”，所以创建时先拦一道，
+ 执行时轮次边界门仍照常生效。
跑中掐断（与旧 docstring 不同，旧说法已过时）：
Local popen 路径的 worker 轮询每 30 秒查一次配额，
@@ -4030,19 +4076,43 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
return True
return False
- def _is_global_maintainer(self, user_id):
- """全局管理判定：沿用 _admin_backup 的口径——在任一
- 项目里是 maintainer/owner 即算（没有单设的全局管理
- 员表；没有任何项目角色的用户无权改任何人的配额）。"""
- conn = self._db()
+ def _reject_if_over_quota_for_task(self, caller_id, task):
+ """花钱路径的双人配额门：触发人 + 原任务创建人。
+
+ 任一超限即 402（USD 配额与无价 token cap 都查）。
+ 触发人等于创建人时只查一次。调用方须在权限校验后、
+ 写状态前调它；返回 True 表示已回 402，调用方直接
+ return。"""
+ if self._reject_if_over_quota(caller_id):
+ return True
+ creator = (task or {}).get("created_by")
+ if creator and creator != caller_id:
+ if self._reject_if_over_quota(creator):
+ return True
+ return False
+
+ def _audit_admin(self, user_id, action, detail):
+ """admin 调用审计（成功与被拒都记，失败绝不挡主流程）。"""
try:
- row = conn.execute(
- "SELECT COUNT(*) FROM project_members"
- " WHERE user_id = ? AND role IN ('maintainer', 'owner')",
- (user_id,)).fetchone()
- return (row[0] if row is not None else 0) > 0
- finally:
- conn.close()
+ with _auth_store(self.server.db_path) as store:
+ store.audit(user_id, "", action, detail)
+ except Exception:
+ pass
+
+ def _is_global_admin(self, user_id):
+ """真全局管理员：只认 CODEX_GLOBAL_ADMINS。
+
+ 未配置时返回 False（fail-closed）。P0 安全批前“任一
+ 项目 maintainer 即全局管理”的推导已被废掉：项目角色
+ 只管项目内权限，不管全系统管理权。部署时运维必须先
+ 把管理员用户 id 写进 CODEX_GLOBAL_ADMINS 再重启，
+ 否则 admin 端点对所有人都是 403（数据不丢，改 env
+ 重启即恢复）。"""
+ return user_id is not None and user_id in _global_admin_ids()
+
+ def _is_global_maintainer(self, user_id):
+ """旧名兼容：等价于 _is_global_admin（推导口径已废）。"""
+ return self._is_global_admin(user_id)
def _user_exists(self, target):
conn = self._db()
@@ -4061,7 +4131,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._quota_unpriced_tokens(user_id)))
def _admin_quota_get(self, user_id, query):
- if not self._is_global_maintainer(user_id):
+ if not self._is_global_admin(user_id):
+ self._audit_admin(user_id, "admin_quota_denied",
+ "get quotas")
self._send_json(403, {"error": "forbidden"})
return
target = (query.get("user_id") or [None])[0]
@@ -4078,7 +4150,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._quota_unpriced_tokens(target)))
def _admin_quota_set(self, user_id):
- if not self._is_global_maintainer(user_id):
+ if not self._is_global_admin(user_id):
+ self._audit_admin(user_id, "admin_quota_denied",
+ "set quotas")
self._send_json(403, {"error": "forbidden"})
return
body, err = self._read_json()
@@ -4323,14 +4397,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _admin_sweep(self, user_id):
"""POST /admin/sweep：把超阈值的陈旧 running 任务标 interrupted。
- 权限：与 _admin_backup 同口径——调用方须在任一项目里
- 是 maintainer 以上，否则 403。过去这里没有任何权限检查，
- 任意持 key 用户都能跨用户改任务状态。threshold_seconds
- 下限钳到 60 秒：0 会把所有 running 任务一口气扫掉，
- 只能是管理员的显式紧急操作才需要那么小的值，而 60 秒
- 以下的“陈旧”本就没有意义。扫完记 admin_sweep 审计。
+ 权限：只认 CODEX_GLOBAL_ADMINS（见 _is_global_admin），
+ 未配置时一律 403。threshold_seconds 下限钳到 60 秒：
+ 0 会把所有 running 任务一口气扫掉，只能是管理员的
+ 显式紧急操作才需要那么小的值，而 60 秒以下的“陈旧”
+ 本就没有意义。扫完记 admin_sweep 审计，被拒也记审计。
"""
- if not self._is_global_maintainer(user_id):
+ if not self._is_global_admin(user_id):
+ self._audit_admin(user_id, "admin_sweep_denied", "sweep")
self._send_json(403, {"error": "forbidden"})
return
body, err = self._read_json()
@@ -4354,32 +4428,21 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _admin_backup(self, user_id):
"""POST /admin/backup：备份当前后端数据库，返回路径与规模信息。
- 权限：调用方须在至少一个项目里是 maintainer 以上（全局
- 管理端点没有单个项目上下文，按“任一项目 maintainer+”
- 判定；无此角色返回 403）。返回只含 backend、备份文件
- 路径、文件大小与任务行数，绝不返回数据库密码、备份文件
- 内容或任何凭据。out_dir 可由 body 指定，否则走环境变量
- CLOUD_BACKUP_DIR，未设则用 cloud-backups。
+ 权限：只认 CODEX_GLOBAL_ADMINS（见 _is_global_admin），
+ 未配置时一律 403。返回只含 backend、备份文件路径、
+ 文件大小与任务行数，绝不返回数据库密码、备份文件内容
+ 或任何凭据。out_dir 可由 body 指定，否则走环境变量
+ CLOUD_BACKUP_DIR，未设则用 cloud-backups。成功与被拒
+ 都记审计（审计里只记路径与规模，不记库内容）。
"""
+ if not self._is_global_admin(user_id):
+ self._audit_admin(user_id, "admin_backup_denied", "backup")
+ self._send_json(403, {"error": "forbidden"})
+ return
body, err = self._read_json()
if err:
self._send_json(400, {"error": err})
return
- conn = self._db()
- try:
- member_rows = conn.execute(
- "SELECT role FROM project_members WHERE user_id = ?",
- (user_id,),
- ).fetchall()
- finally:
- conn.close()
- role_levels = {"viewer": 0, "contributor": 1,
- "maintainer": 2, "owner": 3}
- is_maintainer = any(
- role_levels.get(row[0], 0) >= 2 for row in member_rows)
- if not is_maintainer:
- self._send_json(403, {"error": "forbidden"})
- return
out_dir = body.get("out_dir") or os.environ.get(
"CLOUD_BACKUP_DIR") or "cloud-backups"
if not isinstance(out_dir, str) or not out_dir:
@@ -4403,6 +4466,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
except Exception as exc:
self._send_json(500, {"error": "backup failed: %s" % (exc,)})
return
+ self._audit_admin(
+ user_id, "admin_backup",
+ "backend=%s path=%s size=%d tasks=%d"
+ % (backend, dest, size_bytes, task_count))
self._send_json(200, {
"backend": backend, "path": dest,
"size_bytes": size_bytes, "task_count": task_count,
@@ -4571,6 +4638,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
auth, provider, identity, bind_user_id)
auth.audit(bind_user_id, "", "bind_external",
"%s:%s" % (provider, identity.get("external_id")))
+ if _apply_default_quota(auth.conn, bind_user_id):
+ auth.audit(bind_user_id, "", "quota_default",
+ "quota_usd_micros=%d"
+ % _default_quota_usd_micros())
identity["already_bound"] = binding["already"]
return (bind_user_id, False, None, False, identity, True)
bound_user_id = find_bound_user(auth, provider, identity)
@@ -4580,6 +4651,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
auth.audit(bound_user_id, "", "oauth_login",
"%s:%s" % (provider, identity.get("external_id")))
auth.audit(bound_user_id, "", "api_key_issued", provider)
+ if _apply_default_quota(auth.conn, bound_user_id):
+ auth.audit(bound_user_id, "", "quota_default",
+ "quota_usd_micros=%d"
+ % _default_quota_usd_micros())
return (bound_user_id, False, api_key_plain, False, identity,
False)
local_id, created, api_key_plain, rotated = \
@@ -4587,7 +4662,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
auth, provider, identity)
auth.audit(local_id, "", "oauth_login",
"%s:%s" % (provider, identity.get("external_id")))
- if created and _apply_default_quota(auth.conn, local_id):
+ if _apply_default_quota(auth.conn, local_id):
auth.audit(local_id, "", "quota_default",
"quota_usd_micros=%d"
% _default_quota_usd_micros())
@@ -4864,17 +4939,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
html.escape(str(message))))
def _github_app_setup_state(self):
- """签发 setup state：须持本站 key（只有操作方能发起建 App）。
+ """签发 setup state：只限全局管理员（见 _is_global_admin）。
- 要求环境变量 GITHUB_APP_PUBLIC_BASE（公网域名，如
+ P0 安全批前任何持 key 用户都能签发并回调覆盖全局
+ App 配置；现在非管理员直接 403 并记审计。要求环境
+ 变量 GITHUB_APP_PUBLIC_BASE（公网域名，如
https://api.diante.me）：manifest 里的 hook / 回调地址
都由它拼出。返回 {state, setup_url}，setup_url 是用户
- 在浏览器里打开的表单页；state 一次性、1 小时过期。
+ 在浏览器里打开的表单页；state 一次性、1 小时过期，
+ 且绑定签发人。
"""
user_id = self._authenticate()
if user_id is None:
self._send_json(401, {"error": "unauthorized"})
return
+ if not self._is_global_admin(user_id):
+ self._audit_admin(user_id, "github_app_setup_denied",
+ "setup state")
+ self._send_json(403, {"error": "forbidden"})
+ return
base = self._github_app_public_base()
if not base:
self._send_json(400, {
@@ -4886,7 +4969,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
org = (body or {}).get("org")
store = SetupStateStore(self.server.db_path)
try:
- state = store.create_state()
+ state = store.create_state(issued_by=user_id)
finally:
try:
store.conn.close()
@@ -4968,6 +5051,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
报错误页，不存任何东西。成功页只显示 App 身份（slug、
id、主页）与安装链接，绝不显示 webhook_secret 与 pem
（两者只在库里，server 进程内用）。
+ P0 安全批后：回调要求签发人仍在 CODEX_GLOBAL_ADMINS
+ 里，否则 state 不消耗、直接报 invalid_state；覆盖全局
+ 配置时记审计。诚实边界：回调是浏览器无认证重定向，
+ 这里核对的是“签发人仍是管理员”，不是“回调者就是
+ 签发人”（见 SetupStateStore.consume_state 文档）。
"""
query = parse_qs(urlparse(self.path).query)
code = (query.get("code") or [None])[0]
@@ -4978,13 +5066,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
return
store = SetupStateStore(self.server.db_path)
try:
- ok = store.consume_state(state)
+ issuer = store.state_issuer(state)
+ ok = store.consume_state(
+ state, allowed_issuers=_global_admin_ids())
finally:
try:
store.conn.close()
except Exception:
pass
if not ok:
+ self._audit_admin(
+ issuer or "unknown", "github_app_callback_denied",
+ "state issuer not a global admin")
self._send_html(400, self._github_app_error_page(
"invalid_state", "setup state invalid or expired"))
return
@@ -5001,6 +5094,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
save_app_config(conn, config)
finally:
conn.close()
+ self._audit_admin(
+ issuer or "unknown", "github_app_config_replaced",
+ "slug=%s app_id=%s" % (config.get("slug"),
+ config.get("app_id")))
slug = config.get("slug") or ""
page = (
""
@@ -5130,17 +5227,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
if not event_type or not project_id:
self._send_json(400, {"error": "event_type and project_id required"})
return
- # 分发者须是该项目的成员（至少可读），否则不能往别人项目里推。
- with _auth_store(self.server.db_path) as auth:
- if not auth.check_permission(user_id, project_id, "read"):
- self._send_json(403, {"error": "forbidden"})
- return
+ # P0 安全批后：/admin/push/dispatch 是管理端点，只认
+ # CODEX_GLOBAL_ADMINS，不再只凭项目 read 就能让服务端
+ # 向订阅目标（可能含内网地址）真 POST。被拒与成功都记审计。
+ if not self._is_global_admin(user_id):
+ self._audit_admin(user_id, "admin_push_denied",
+ "dispatch %s %s" % (event_type, project_id))
+ self._send_json(403, {"error": "forbidden"})
+ return
# 分发：sender=None 时用 push 的内置真实发送端
# （webhook/slack 真 POST，webpush/apns/fcm 仍报缺凭证）；
# 显式注入的 sender 优先于内置端（见 push.dispatch_event）。
result = _push_dispatch(
self.server.db_path, event_type, project_id,
body.get("payload") or {}, sender=None)
+ self._audit_admin(user_id, "admin_push_dispatch",
+ "dispatch %s %s" % (event_type, project_id))
self._send_json(200, result)
# -- webhooks（签名认证，不走 X-API-Key）---------------------------
@@ -5255,12 +5357,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
GITHUB_WEBHOOK_SECRET（手动 webhook，过渡期共存；App
装好、手动 webhook 删掉后自然不再被命中）。
/fix-review 解析成功后在控制面建一个 queued 任务（prompt
- 含 repo、PR 号、评论原文、作者），由 worker 认领执行——
- webhook 本身不跑模型。任务归属：按项目名 == repo
- full_name 匹配项目；没有匹配项目时不凭空建项目、不建
- 任务，返回 task_id 为 None 并说明原因（worker 的 repo
- 仍是配置里的 repo，不是任意 GitHub 仓库的自动克隆，
- 见 README）。check_run / check_suite 只做通知解析，
+ 是固定模板 + 截断后的评论摘录，不是评论原文原样），
+ 由 worker 认领执行——webhook 本身不跑模型。任务归属：
+ 按项目名 == repo full_name 匹配项目，created_by 归属
+ 项目 owner 并走 owner 的配额门，executor 强制 docker；
+ 没有匹配项目时不凭空建项目、不建任务，返回 task_id
+ 为 None 并说明原因（worker 的 repo 仍是配置里的 repo，
+ 不是任意 GitHub 仓库的自动克隆，见 README）。作者
+ 不可信时同样不建任务（见 is_trusted_fix_review_author）。check_run / check_suite 只做通知解析，
返回解析结果而不建任务、不改云端任务状态（CI 状态回填
由 pr-checks 主动拉取）。installation / installation_repositories
事件记安装记录（created/added 增行、removed 减行、deleted
@@ -5368,12 +5472,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(200, {"result": None})
return
# /fix-review 执行：建 queued 任务（不跑模型，worker 认领）。
+ # P0 安全批后：先验作者、再查配额、最后才建任务。
+ # 作者校验用 GitHub 事件自带的 author_association 与
+ # PR/issue 作者比对（见 is_trusted_fix_review_author）：
+ # 不可信一律不建任务、不占去重位，fail-closed。
conn = self._db()
try:
project_row = None
if repo_name:
project_row = conn.execute(
- "SELECT id FROM projects WHERE name = ?",
+ "SELECT id, owner_id FROM projects WHERE name = ?",
(repo_name,),
).fetchone()
if project_row is None:
@@ -5382,6 +5490,28 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
error="no matching project for repo"))
return
project_id = project_row[0]
+ project_owner = project_row[1]
+ if not is_trusted_fix_review_author(payload):
+ with _auth_store(self.server.db_path) as auth:
+ auth.audit("system", project_id,
+ "webhook_fix_review_rejected",
+ "untrusted author %s"
+ % (result.get("author"),))
+ self._send_json(200, dict(
+ result, task_id=None, project_id=project_id,
+ error="untrusted fix-review author"))
+ return
+ if not project_owner:
+ self._send_json(200, dict(
+ result, task_id=None, project_id=project_id,
+ error="project has no owner for quota attribution"))
+ return
+ if self._reject_if_over_quota(project_owner):
+ with _auth_store(self.server.db_path) as auth:
+ auth.audit("system", project_id,
+ "webhook_fix_review_quota_blocked",
+ project_owner)
+ return
# 内容去重：手动 webhook 与 App webhook 共存时同一评论
# 送两遍（delivery_id 不同），只建一次任务；重复送达
# 回读当初的 task_id，不重复记审计。
@@ -5395,20 +5525,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
return
task_id = uuid.uuid4().hex
now = time.time()
+ # prompt 不再原样贴评论全文：固定模板 + 截断后的
+ # 评论摘录（上限 2000 字），防超长评论撑爆 prompt
+ # 与指令注入面无限放大。
prompt = (
"[webhook /fix-review] repo=%s pr=%s author=%s\n%s"
% (result.get("repo"), result.get("pr_number"),
- result.get("author"), result.get("comment")))
+ result.get("author"),
+ (result.get("comment") or "")[:2000]))
try:
conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" idempotency_key, depends_on, priority, created_by,"
- " created_at, updated_at, last_heartbeat)"
- " VALUES(?, ?, ?, 'queued', ?, '[]', 0, 'system',"
- " ?, ?, NULL)",
+ " created_at, updated_at, last_heartbeat, executor)"
+ " VALUES(?, ?, ?, 'queued', ?, '[]', 0, ?,"
+ " ?, ?, NULL, 'docker')",
(task_id, project_id, prompt,
"webhook:%s" % delivery_id if delivery_id else None,
- now, now),
+ project_owner, now, now),
)
conn.commit()
except (sqlite3.IntegrityError, _PgIntegrityError):
diff --git a/cloud/tests/test_github_app.py b/cloud/tests/test_github_app.py
index 6d49cd1..3eff05e 100644
--- a/cloud/tests/test_github_app.py
+++ b/cloud/tests/test_github_app.py
@@ -289,6 +289,10 @@ class AppHttpTestCase(unittest.TestCase):
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.user_a, self.key_a = self.auth.create_user("user-a")
+ # P0 安全批后：setup state 只限全局管理员，本用例里
+ # user-a 即管理员。
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.user_a
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
@@ -307,6 +311,10 @@ class AppHttpTestCase(unittest.TestCase):
self.tmp.cleanup()
os.environ.pop("GITHUB_APP_PUBLIC_BASE", None)
os.environ.pop("GITHUB_WEBHOOK_SECRET", None)
+ if self._old_admins is None:
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ else:
+ os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
def _req(self, method, path, body=None, key=None, headers=None):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
@@ -488,7 +496,8 @@ class AppHttpTestCase(unittest.TestCase):
payload = {
"action": "created",
"comment": {"id": 777, "body": "please /fix-review this",
- "user": {"login": "octo"}},
+ "user": {"login": "octo",
+ "author_association": "COLLABORATOR"}},
"issue": {"number": 5},
"repository": {"full_name": "octo/a"}}
# 同一评论送两遍，delivery_id 不同（模拟手动 + App 双通道）。
diff --git a/cloud/tests/test_multiuser_gaps.py b/cloud/tests/test_multiuser_gaps.py
index aecf232..3099434 100644
--- a/cloud/tests/test_multiuser_gaps.py
+++ b/cloud/tests/test_multiuser_gaps.py
@@ -72,6 +72,8 @@ class MultiuserGapsTestCase(unittest.TestCase):
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
self.viewer_id, self.viewer_key = self.auth.create_user("viewer")
self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
self.server = create_server(self.db_path, "127.0.0.1", 0)
@@ -95,6 +97,10 @@ class MultiuserGapsTestCase(unittest.TestCase):
except Exception:
pass
self.tmp.cleanup()
+ if getattr(self, "_old_admins", None) is None:
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ else:
+ os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
# -- helpers -------------------------------------------------------
def _create_task(self, key=None, body_extra=None):
@@ -264,13 +270,15 @@ class MultiuserGapsTestCase(unittest.TestCase):
finally:
_setenv("CLOUD_QUOTA_UNPRICED_TOKEN_CAP", old)
- def test_unpriced_tokens_unlimited_without_quota_row(self):
+ def test_unpriced_token_cap_blocks_without_quota_row(self):
+ # P0 安全批后：无配额行不再等于不限额，无价 cap 照样挡。
for i in range(7):
self._insert_unpriced_cost("donor-%d" % i, 100, 50)
old = _setenv("CLOUD_QUOTA_UNPRICED_TOKEN_CAP", "1000")
try:
- s, task = self._create_task()
- self.assertEqual(s, 201, task)
+ s, body = self._create_task()
+ self.assertEqual(s, 402, body)
+ self.assertEqual(body["error"], "quota_exceeded")
finally:
_setenv("CLOUD_QUOTA_UNPRICED_TOKEN_CAP", old)
diff --git a/cloud/tests/test_p0_security.py b/cloud/tests/test_p0_security.py
new file mode 100644
index 0000000..9338656
--- /dev/null
+++ b/cloud/tests/test_p0_security.py
@@ -0,0 +1,466 @@
+"""P0 安全批单测（基线 99efa93 后的硬缺口修法）。
+
+不跑真实模型回合、不联网、不碰生产：server 侧走本地临时
+端口 + 临时 SQLite；用量直接插 cost_records 行；webhook 走
+本地签名；GitHub App 回调走注入的 fake 换 code 客户端。
+测试用的 key、secret、state 全是假值。
+"""
+
+import hashlib
+import hmac
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest import mock
+
+try:
+ from cloud.auth import AuthStore
+ from cloud.github_app import SetupStateStore
+ from cloud.server import create_server
+ from cloud.webhooks import is_trusted_fix_review_author
+except ImportError:
+ sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+ from cloud.auth import AuthStore
+ from cloud.github_app import SetupStateStore
+ from cloud.server import create_server
+ from cloud.webhooks import is_trusted_fix_review_author
+
+
+def _req(port, method, path, body=None, key=None, headers=None,
+ raw_body=None):
+ conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
+ hdrs = dict(headers or {})
+ if raw_body is not None:
+ payload = raw_body
+ elif body is not None:
+ payload = json.dumps(body).encode("utf-8")
+ hdrs.setdefault("Content-Type", "application/json")
+ else:
+ payload = None
+ if key is not None:
+ hdrs["X-API-Key"] = key
+ try:
+ conn.request(method, path, body=payload, headers=hdrs)
+ resp = conn.getresponse()
+ data = resp.read()
+ finally:
+ conn.close()
+ try:
+ parsed = json.loads(data.decode("utf-8")) if data else None
+ except ValueError:
+ parsed = data.decode("utf-8", "replace")
+ return resp.status, parsed
+
+
+def _sig(secret, raw):
+ return "sha256=" + hmac.new(
+ secret.encode(), raw, hashlib.sha256).hexdigest()
+
+
+def _utc_today():
+ return datetime.now(timezone.utc).strftime("%Y-%m-%d")
+
+
+class P0SecurityTestCase(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.db_path = os.path.join(self.tmp.name, "control.db")
+ self.auth = AuthStore(self.db_path)
+ self.owner_id, self.owner_key = self.auth.create_user("owner")
+ self.member_id, self.member_key = self.auth.create_user("member")
+ self.outsider_id, self.outsider_key = self.auth.create_user(
+ "outsider")
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
+ 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()
+ s, proj = _req(self.port, "POST", "/projects",
+ {"name": "owner/repo"}, self.owner_key)
+ self.assertEqual(s, 201, proj)
+ self.project_id = proj["id"]
+ self.auth.set_member(
+ self.project_id, self.owner_id, self.member_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
+ os.environ.pop("GITHUB_APP_PUBLIC_BASE", None)
+ os.environ.pop("GITHUB_WEBHOOK_SECRET", None)
+
+ # -- helpers ---------------------------------------------------
+ def _create_task(self, key=None, prompt="p0 probe"):
+ return _req(self.port, "POST", "/tasks",
+ {"project_id": self.project_id, "prompt": prompt},
+ key if key is not None else self.owner_key)
+
+ def _set_status(self, task_id, status):
+ conn = sqlite3.connect(self.db_path)
+ try:
+ conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+ (status, task_id))
+ conn.commit()
+ finally:
+ conn.close()
+
+ def _insert_cost(self, task_id, usd_micros, day=None):
+ conn = sqlite3.connect(self.db_path)
+ try:
+ conn.execute(
+ "INSERT INTO cost_records(task_id, project_id, model,"
+ " input_tokens, output_tokens, total_tokens, usd_micros,"
+ " source, day, created_at) VALUES(?, ?, ?, ?, ?, ?, ?,"
+ " ?, ?, ?)",
+ (task_id, self.project_id, "gpt-5.6-terra", 100, 50, 150,
+ usd_micros, "actual", day or _utc_today(), 0))
+ conn.commit()
+ finally:
+ conn.close()
+
+ def _insert_unpriced_cost(self, task_id, in_tok, out_tok):
+ conn = sqlite3.connect(self.db_path)
+ try:
+ conn.execute(
+ "INSERT INTO cost_records(task_id, project_id, model,"
+ " input_tokens, output_tokens, total_tokens, usd_micros,"
+ " source, day, created_at) VALUES(?, ?, ?, ?, ?, ?,"
+ " NULL, ?, ?, ?)",
+ (task_id, self.project_id, "gpt-5.6-terra", in_tok,
+ out_tok, in_tok + out_tok, "actual", _utc_today(), 0))
+ conn.commit()
+ finally:
+ conn.close()
+
+ def _set_quota(self, target, quota):
+ return _req(self.port, "POST", "/admin/quotas",
+ {"user_id": target, "quota_usd_micros": quota},
+ self.owner_key)
+
+ def _audit_actions(self):
+ conn = sqlite3.connect(self.db_path)
+ try:
+ return [r[0] for r in conn.execute(
+ "SELECT action FROM audit_log").fetchall()]
+ finally:
+ conn.close()
+
+ def _webhook(self, payload, delivery, secret="test-secret"):
+ raw = json.dumps(payload).encode("utf-8")
+ headers = {
+ "Content-Type": "application/json",
+ "X-Hub-Signature-256": _sig(secret, raw),
+ "X-GitHub-Event": "issue_comment",
+ "X-GitHub-Delivery": delivery,
+ }
+ with mock.patch.dict(
+ os.environ, {"GITHUB_WEBHOOK_SECRET": secret}):
+ return _req(self.port, "POST", "/webhooks/github",
+ headers=headers, raw_body=raw)
+
+ def _fix_payload(self, author="alice", association="COLLABORATOR",
+ comment="请 /fix-review 看一下"):
+ return {
+ "action": "created",
+ "comment": {"id": 999, "body": comment,
+ "user": {"login": author,
+ "author_association": association}},
+ "repository": {"full_name": "owner/repo"},
+ "issue": {"number": 42, "user": {"login": "alice"}},
+ }
+
+ def _tasks(self):
+ conn = sqlite3.connect(self.db_path)
+ try:
+ return conn.execute(
+ "SELECT id, prompt, status, executor, created_by"
+ " FROM tasks").fetchall()
+ finally:
+ conn.close()
+
+ # -- 1. 真全局管理员 -------------------------------------------
+ def test_admin_fail_closed_without_env(self):
+ with mock.patch.dict(os.environ, {}, clear=False):
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ for method, path, body in (
+ ("POST", "/admin/sweep", {}),
+ ("GET", "/admin/quotas?user_id=" + self.owner_id,
+ None),
+ ("POST", "/admin/backup", {"out_dir": "x"}),
+ ("POST", "/admin/push/dispatch",
+ {"event_type": "failed",
+ "project_id": self.project_id})):
+ s, _ = _req(self.port, method, path, body,
+ self.owner_key)
+ self.assertEqual(s, 403, (method, path))
+ # 配了 env 的管理员仍可调 sweep。
+ s, body = _req(self.port, "POST", "/admin/sweep", {},
+ self.owner_key)
+ self.assertEqual(s, 200, body)
+ self.assertIn("admin_sweep", self._audit_actions())
+
+ def test_admin_denied_is_audited(self):
+ with mock.patch.dict(os.environ, {}, clear=False):
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ _req(self.port, "POST", "/admin/sweep", {},
+ self.member_key)
+ self.assertIn("admin_sweep_denied", self._audit_actions())
+
+ # -- 2. 所有花钱路径统一配额门 ----------------------------------
+ def test_continue_blocked_when_creator_over_quota(self):
+ # 任务由 owner 创建，member（maintainer）来续跑：
+ # member 自己没超，但 owner 超了，仍应 402。
+ s, task = self._create_task(key=self.owner_key)
+ self.assertEqual(s, 201, task)
+ self._set_status(task["id"], "completed")
+ self._insert_cost(task["id"], 5000000)
+ s, _ = self._set_quota(self.owner_id, 100)
+ self.assertEqual(s, 200)
+ s, body = _req(self.port, "POST",
+ "/tasks/%s/continue" % task["id"],
+ {"message": "继续"}, self.member_key)
+ self.assertEqual(s, 402, body)
+ self.assertEqual(body["error"], "quota_exceeded")
+
+ def test_followup_blocked_when_creator_over_quota(self):
+ s, task = self._create_task(key=self.owner_key)
+ self.assertEqual(s, 201, task)
+ self._set_status(task["id"], "running")
+ self._insert_cost(task["id"], 5000000)
+ s, _ = self._set_quota(self.owner_id, 100)
+ self.assertEqual(s, 200)
+ s, body = _req(self.port, "POST",
+ "/tasks/%s/followups" % task["id"],
+ {"body": "后续", "mode": "queued"},
+ self.member_key)
+ self.assertEqual(s, 402, body)
+
+ def test_peer_message_blocked_when_target_creator_over_quota(self):
+ s, target = self._create_task(key=self.owner_key)
+ self.assertEqual(s, 201, target)
+ self._set_status(target["id"], "running")
+ s, sender = self._create_task(key=self.member_key)
+ self.assertEqual(s, 201, sender)
+ self._insert_cost(target["id"], 5000000)
+ s, _ = self._set_quota(self.owner_id, 100)
+ self.assertEqual(s, 200)
+ s, body = _req(self.port, "POST",
+ "/tasks/%s/messages" % target["id"],
+ {"from_task_id": sender["id"], "body": "hi"},
+ self.member_key)
+ self.assertEqual(s, 402, body)
+
+ def test_retry_blocked_when_creator_over_quota(self):
+ s, task = self._create_task(key=self.owner_key)
+ self.assertEqual(s, 201, task)
+ self._set_status(task["id"], "failed")
+ self._insert_cost(task["id"], 5000000)
+ s, _ = self._set_quota(self.owner_id, 100)
+ self.assertEqual(s, 200)
+ s, body = _req(self.port, "POST",
+ "/tasks/%s/retry" % task["id"], {},
+ self.member_key)
+ self.assertEqual(s, 402, body)
+
+ # -- 3. 无配额行不再等于不限额 ----------------------------------
+ def test_no_quota_row_uses_default_usd_quota(self):
+ # member 是 maintainer 但配额行是缺的：先建任务、再把
+ # 当月费用推到默认额度之上，下一次建任务应 402。
+ s, task = self._create_task(key=self.member_key)
+ self.assertEqual(s, 201, task)
+ self._insert_cost(task["id"], 11000000)
+ s, body = self._create_task(key=self.member_key)
+ self.assertEqual(s, 402, body)
+ self.assertEqual(body["quota_usd_micros"], 10000000)
+
+ def test_no_quota_row_unpriced_cap_blocks(self):
+ old = os.environ.get("CLOUD_QUOTA_UNPRICED_TOKEN_CAP")
+ os.environ["CLOUD_QUOTA_UNPRICED_TOKEN_CAP"] = "1000"
+ try:
+ s, task = self._create_task(key=self.member_key)
+ self.assertEqual(s, 201, task)
+ self._insert_unpriced_cost(task["id"], 700, 350)
+ s, body = self._create_task(key=self.member_key)
+ self.assertEqual(s, 402, body)
+ self.assertEqual(body["unpriced_tokens_used"], 1050)
+ finally:
+ if old is None:
+ os.environ.pop("CLOUD_QUOTA_UNPRICED_TOKEN_CAP", None)
+ else:
+ os.environ["CLOUD_QUOTA_UNPRICED_TOKEN_CAP"] = old
+
+ # -- 4. webhook /fix-review 收紧 --------------------------------
+ def test_webhook_untrusted_author_rejected(self):
+ s, body = self._webhook(
+ self._fix_payload(author="bob", association="CONTRIBUTOR"),
+ delivery="d-p0-untrusted")
+ self.assertEqual(s, 200, body)
+ self.assertIsNone(body["task_id"])
+ self.assertEqual(body["error"], "untrusted fix-review author")
+ self.assertEqual(self._tasks(), [])
+ self.assertIn("webhook_fix_review_rejected",
+ self._audit_actions())
+
+ def test_webhook_trusted_forces_docker_and_owner_and_truncates(self):
+ long_comment = "请 /fix-review 看一下 " + "X" * 3000 + "TAILMARK"
+ s, body = self._webhook(
+ self._fix_payload(comment=long_comment),
+ delivery="d-p0-trusted")
+ self.assertEqual(s, 200, body)
+ self.assertIsNotNone(body["task_id"])
+ rows = self._tasks()
+ self.assertEqual(len(rows), 1)
+ task_id, prompt, status, executor, created_by = rows[0]
+ self.assertEqual(task_id, body["task_id"])
+ self.assertEqual(status, "queued")
+ self.assertEqual(executor, "docker")
+ self.assertEqual(created_by, self.owner_id)
+ self.assertNotIn("TAILMARK", prompt)
+ self.assertLessEqual(len(prompt), 2200)
+
+ def test_webhook_owner_quota_blocks(self):
+ s, task = self._create_task(key=self.owner_key)
+ self.assertEqual(s, 201, task)
+ self._insert_cost(task["id"], 5000000)
+ s, _ = self._set_quota(self.owner_id, 100)
+ self.assertEqual(s, 200)
+ s, body = self._webhook(self._fix_payload(), delivery="d-p0-quota")
+ self.assertEqual(s, 402, body)
+ # 未建新任务：只有之前手动建的那一个任务。
+ self.assertEqual(len(self._tasks()), 1)
+
+ def test_is_trusted_fix_review_author_unit(self):
+ self.assertTrue(is_trusted_fix_review_author(
+ {"comment": {"user": {"login": "bob",
+ "author_association": "MEMBER"}}}))
+ self.assertTrue(is_trusted_fix_review_author(
+ {"comment": {"user": {"login": "alice",
+ "author_association": "FIRST_TIMER"}},
+ "issue": {"user": {"login": "alice"}}}))
+ self.assertFalse(is_trusted_fix_review_author(
+ {"comment": {"user": {"login": "bob",
+ "author_association": "CONTRIBUTOR"}},
+ "issue": {"user": {"login": "alice"}}}))
+ self.assertFalse(is_trusted_fix_review_author(
+ {"comment": {"user": {"login": "bob"}}}))
+
+ # -- 5. GitHub App 配置覆盖限管理员 -------------------------------
+ def test_github_app_setup_requires_admin(self):
+ os.environ["GITHUB_APP_PUBLIC_BASE"] = "https://example.invalid"
+ s, _ = _req(self.port, "POST", "/github-app/setup/state", {},
+ self.member_key)
+ self.assertEqual(s, 403)
+ s, body = _req(self.port, "POST", "/github-app/setup/state", {},
+ self.owner_key)
+ self.assertEqual(s, 200, body)
+ self.assertIn("state", body)
+
+ def test_setup_state_issuer_mismatch_rejected(self):
+ store = SetupStateStore(self.db_path)
+ try:
+ state = store.create_state(issued_by="someone-else")
+ self.assertFalse(store.consume_state(
+ state, allowed_issuers={self.owner_id}))
+ # 未被消耗：管理员清单里换成签发人后仍可消耗。
+ self.assertTrue(store.consume_state(
+ state, allowed_issuers={"someone-else"}))
+ self.assertFalse(store.consume_state(state))
+ finally:
+ store.conn.close()
+
+ def test_github_app_callback_issuer_mismatch(self):
+ self.server.github_app_exchange_client = lambda url, headers: {
+ "id": 424242, "slug": "fake-codex-app",
+ "client_id": "Iv1.x", "webhook_secret": "fake-app-secret",
+ "pem": "fake-pem",
+ "html_url": "https://github.com/apps/fake-codex-app"}
+ store = SetupStateStore(self.db_path)
+ try:
+ state = store.create_state(issued_by="someone-else")
+ finally:
+ store.conn.close()
+ s, page = _req(self.port, "GET",
+ "/github-app/callback?code=x&state=%s" % state)
+ self.assertEqual(s, 400, page)
+
+
+class P0WorkerCancelTestCase(unittest.TestCase):
+ """取消写回守卫：迟到写回不覆盖新结果（mock DB 级验证）。"""
+
+ def test_write_result_guarded_by_status(self):
+ from cloud.worker import Worker
+ tmp = tempfile.TemporaryDirectory()
+ try:
+ db_path = os.path.join(tmp.name, "control.db")
+ conn = sqlite3.connect(db_path)
+ try:
+ conn.execute(
+ "CREATE TABLE tasks(id TEXT PRIMARY KEY,"
+ " status TEXT, result_text TEXT, error_text TEXT,"
+ " actual_model TEXT, started_at REAL,"
+ " finished_at REAL, turn_id TEXT,"
+ " root_turn_id TEXT, updated_at REAL)")
+ conn.execute(
+ "INSERT INTO tasks(id, status) VALUES('t1', 'queued')")
+ conn.commit()
+ finally:
+ conn.close()
+ worker = Worker.__new__(Worker)
+ worker.db_path = db_path
+ worker._ensure_result_columns = lambda c: None
+ # 状态已是 queued，不是 completed：带守卫的写回应被拦。
+ ok = worker._write_execution_result(
+ "t1", result_text="late", only_if_status="completed")
+ self.assertFalse(ok)
+ conn = sqlite3.connect(db_path)
+ try:
+ row = conn.execute(
+ "SELECT result_text FROM tasks WHERE id = 't1'"
+ ).fetchone()
+ finally:
+ conn.close()
+ self.assertIsNone(row[0])
+ # 状态对上时写回成功。
+ ok = worker._write_execution_result(
+ "t1", result_text="ok", only_if_status="queued")
+ self.assertTrue(ok)
+ finally:
+ tmp.cleanup()
+
+ def test_shell_env_exclude_covers_tokens(self):
+ import sys as _sys
+ _sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+ import host as _host
+ excl = _host.SHELL_ENV_EXCLUDE
+ for pat in ("*_TOKEN", "*_AUTH*", "*_CREDENTIAL*", "*API_KEY*"):
+ self.assertIn(pat, excl)
+ # gh 走文件登录：host.py 里不应有 GH_TOKEN 环境变量依赖。
+ import pathlib
+ text = pathlib.Path(
+ Path(__file__).resolve().parents[2] / "host.py"
+ ).read_text()
+ self.assertNotIn("GH_TOKEN", text)
+ self.assertNotIn("GITHUB_TOKEN", text)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/cloud/tests/test_prod_webhook_exec.py b/cloud/tests/test_prod_webhook_exec.py
index d550066..5540bde 100644
--- a/cloud/tests/test_prod_webhook_exec.py
+++ b/cloud/tests/test_prod_webhook_exec.py
@@ -39,6 +39,8 @@ class WebhookExecTestCase(unittest.TestCase):
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
self.viewer_id, self.viewer_key = self.auth.create_user("viewer")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
@@ -61,6 +63,10 @@ class WebhookExecTestCase(unittest.TestCase):
except Exception:
pass
self.tmp.cleanup()
+ if getattr(self, "_old_admins", None) is None:
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ else:
+ os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
def _req(self, method, path, body=None, key=None, headers=None,
raw_body=None):
@@ -106,9 +112,10 @@ class WebhookExecTestCase(unittest.TestCase):
return {
"action": "created",
"comment": {"body": "请 /fix-review 看一下",
- "user": {"login": "alice"}},
+ "user": {"login": "alice",
+ "author_association": "COLLABORATOR"}},
"repository": {"full_name": "owner/repo"},
- "issue": {"number": 42},
+ "issue": {"number": 42, "user": {"login": "alice"}},
}
def _tasks(self):
diff --git a/cloud/tests/test_server.py b/cloud/tests/test_server.py
index 79aabbb..6e70715 100644
--- a/cloud/tests/test_server.py
+++ b/cloud/tests/test_server.py
@@ -34,6 +34,8 @@ class ServerTestCase(unittest.TestCase):
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
self.viewer_id, self.viewer_key = self.auth.create_user("viewer")
self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
self.server = create_server(self.db_path, "127.0.0.1", 0)
@@ -57,6 +59,10 @@ class ServerTestCase(unittest.TestCase):
except Exception:
pass
self.tmp.cleanup()
+ if getattr(self, "_old_admins", None) is None:
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ else:
+ os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
# -- HTTP 工具 -----------------------------------------------------
def _req(self, method, path, body=None, key=None, headers=None,
diff --git a/cloud/tests/test_server_oauth_push.py b/cloud/tests/test_server_oauth_push.py
index 242d2e1..68581db 100644
--- a/cloud/tests/test_server_oauth_push.py
+++ b/cloud/tests/test_server_oauth_push.py
@@ -30,6 +30,8 @@ class OAuthPushEndpointsTestCase(unittest.TestCase):
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
@@ -49,6 +51,10 @@ class OAuthPushEndpointsTestCase(unittest.TestCase):
except Exception:
pass
self.tmp.cleanup()
+ if getattr(self, "_old_admins", None) is None:
+ os.environ.pop("CODEX_GLOBAL_ADMINS", None)
+ else:
+ os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
def _req(self, method, path, body=None, key=None):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
diff --git a/cloud/tests/test_task_quotas.py b/cloud/tests/test_task_quotas.py
index 0dfd287..fa2ef4a 100644
--- a/cloud/tests/test_task_quotas.py
+++ b/cloud/tests/test_task_quotas.py
@@ -90,6 +90,10 @@ class ServerQuotasTestCase(unittest.TestCase):
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.other_id, self.other_key = self.auth.create_user("other")
+ # P0 安全批后：admin 端点只认 CODEX_GLOBAL_ADMINS，
+ # 本文件里 owner 即管理员，测前写 env、测后还原。
+ self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
+ os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
@@ -104,6 +108,10 @@ class ServerQuotasTestCase(unittest.TestCase):
self.server.shutdown()
self.server.server_close()
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 _create_task(self, key=None):
return _req(self.port, "POST", "/tasks",
@@ -116,13 +124,14 @@ class ServerQuotasTestCase(unittest.TestCase):
{"user_id": target, "quota_usd_micros": quota},
key if key is not None else self.owner_key)
- def test_self_quota_default_unlimited(self):
+ def test_self_quota_default_is_default_quota_not_unlimited(self):
+ # P0 安全批后：无配额行按默认额度处理，不再是 None。
s, q = _req(self.port, "GET", "/users/me/quota",
key=self.owner_key)
self.assertEqual(s, 200, q)
- self.assertIsNone(q["quota_usd_micros"])
+ self.assertEqual(q["quota_usd_micros"], 10000000)
self.assertEqual(q["used_usd_micros"], 0)
- self.assertIsNone(q["remaining_usd_micros"])
+ self.assertEqual(q["remaining_usd_micros"], 10000000)
self.assertEqual(q["month"],
datetime.now(timezone.utc).strftime("%Y-%m"))
@@ -234,9 +243,11 @@ class ServerQuotasTestCase(unittest.TestCase):
self.assertEqual(s, 200, retried)
self.assertEqual(retried["status"], "queued")
- def test_retry_uses_creator_quota_not_operator(self):
- # 重试按任务创建者查配额：费用记创建者名下，操作人
- # 是谁不应影响准入。other 须是 maintainer 才能过
+ def test_retry_blocked_when_operator_or_creator_over_quota(self):
+ # P0 安全批后：重试按双人门查配额（触发人 + 创建人）。
+ # #90 的“只查创建者、操作人超限不拦”已被 P0 有意收紧：
+ # 操作人用自己的额度过门、费用记创建者名下，仍是绕限，
+ # 所以操作人超限也必须拦。other 须是 maintainer 才能过
# approve_merge 权限门，否则 403 会掩盖配额判断。
self.auth.set_member(self.project_id, self.owner_id,
self.other_id, "maintainer")
@@ -252,12 +263,22 @@ class ServerQuotasTestCase(unittest.TestCase):
"/tasks/%s/retry" % task["id"], {},
self.other_key)
self.assertEqual(s, 402, body)
- # 创建者不限额、操作人额度为 0：按旧口径（查操作人）
- # 会拦，按创建者口径应放行。
+ # 创建者不限额、操作人额度为 0：P0 双人门应拦（#90
+ # 只查创建者的旧口径会放行，那正是 P0 要堵的绕限）。
s, _ = self._set_quota(self.owner_id, None)
self.assertEqual(s, 200)
s, _ = self._set_quota(self.other_id, 0)
self.assertEqual(s, 200)
+ s, body = _req(self.port, "POST",
+ "/tasks/%s/retry" % task["id"], {},
+ self.other_key)
+ self.assertEqual(s, 402, body)
+ s, row = _req(self.port, "GET", "/tasks/" + task["id"],
+ key=self.owner_key)
+ self.assertEqual(row["status"], "failed")
+ # 两人都不超限才放行。
+ s, _ = self._set_quota(self.other_id, None)
+ self.assertEqual(s, 200)
s, retried = _req(self.port, "POST",
"/tasks/%s/retry" % task["id"], {},
self.other_key)
@@ -289,7 +310,6 @@ class ServerQuotasTestCase(unittest.TestCase):
{"message": "again"}, self.owner_key)
self.assertEqual(s, 200, continued)
self.assertEqual(continued["status"], "queued")
- # 续跑同样按创建者查：操作人额度为 0、创建者不限额时放行。
self.auth.set_member(self.project_id, self.owner_id,
self.other_id, "maintainer")
s, task2 = self._create_task()
@@ -297,9 +317,21 @@ class ServerQuotasTestCase(unittest.TestCase):
_set_task_status(self.db_path, task2["id"], "completed")
s, _ = self._set_quota(self.other_id, 0)
self.assertEqual(s, 200)
+ # P0 安全批后：续跑按双人门查，操作人额度为 0、
+ # 创建者不限额时也应拦（#90 只查创建者的旧口径会
+ # 放行，那正是 P0 要堵的绕限）；两人都不超限才放行。
+ s, body = _req(self.port, "POST",
+ "/tasks/%s/continue" % task2["id"],
+ {"message": "again"}, self.other_key)
+ self.assertEqual(s, 402, body)
+ s, row = _req(self.port, "GET", "/tasks/" + task2["id"],
+ key=self.owner_key)
+ self.assertEqual(row["status"], "completed")
+ s, _ = self._set_quota(self.other_id, None)
+ self.assertEqual(s, 200)
s, continued2 = _req(self.port, "POST",
- "/tasks/%s/continue" % task2["id"],
- {"message": "again"}, self.other_key)
+ "/tasks/%s/continue" % task2["id"],
+ {"message": "again"}, self.other_key)
self.assertEqual(s, 200, continued2)
self.assertEqual(continued2["status"], "queued")
diff --git a/cloud/webhooks.py b/cloud/webhooks.py
index e860359..b2b1e58 100644
--- a/cloud/webhooks.py
+++ b/cloud/webhooks.py
@@ -25,6 +25,7 @@ from cloud.db import connect as _db_connect
__all__ = [
"verify_github_signature",
+ "is_trusted_fix_review_author",
"parse_github_pr_comment",
"parse_github_check_run",
"WebhookDeliveryStore",
@@ -110,6 +111,48 @@ def parse_github_pr_comment(payload):
}
+# GitHub 在 issue_comment / pull_request_review_comment 事件里
+# 自带 author_association（由 GitHub 设定，不是评论者自填）：
+# OWNER / MEMBER / COLLABORATOR 是仓库可信身份；CONTRIBUTOR
+# 及以下不是。PR/issue 作者本人评论自己的 PR 也可信（作者
+# 只能在自己的 PR 上触发，不构成跨仓库免费入口）。
+_TRUSTED_FIX_REVIEW_ASSOCIATIONS = frozenset(
+ {"OWNER", "MEMBER", "COLLABORATOR"})
+
+
+def is_trusted_fix_review_author(payload):
+ """P0 安全批：/fix-review 评论者是否可信。
+
+ 只认两条，任一成立即 True，其余一律 False（fail-closed）：
+ 1. comment.user.author_association 在可信集里；
+ 2. 评论者 login 等于 issue 作者或 pull_request 作者。
+ association 缺失且作者比对不上时返回 False——此时不要
+ 用 App token 现查协作者（本函数不联网），直接拒单，
+ 绝不因“查不到”就放行。这是 GitHub 事件字段校验，不
+ 是实时协作者 API 查询，部署文档里必须如实写清这一点。
+ """
+ if not isinstance(payload, dict):
+ return False
+ comment = payload.get("comment")
+ if not isinstance(comment, dict):
+ return False
+ user = comment.get("user") or {}
+ author = user.get("login")
+ if not author:
+ return False
+ if user.get("author_association") in _TRUSTED_FIX_REVIEW_ASSOCIATIONS:
+ return True
+ issue = payload.get("issue") or {}
+ issue_user = issue.get("user") or {}
+ if author == issue_user.get("login"):
+ return True
+ pull = payload.get("pull_request") or {}
+ pull_user = pull.get("user") or {}
+ if author == pull_user.get("login"):
+ return True
+ return False
+
+
def parse_github_check_run(payload):
"""解析 GitHub check_run / check_suite 事件，返回结构化结果或 None。
diff --git a/cloud/worker.py b/cloud/worker.py
index 296dc57..d9cf973 100644
--- a/cloud/worker.py
+++ b/cloud/worker.py
@@ -700,45 +700,57 @@ class Worker:
error_text=None, actual_model=None,
started_at=None, finished_at=None,
child_pid=None, turn_id=None,
- root_turn_id=None):
+ root_turn_id=None, only_if_status=None):
"""把执行结果写进 tasks 新列；老库缺列时先迁移再写。
turn_id/root_turn_id 只在新值非空时覆盖：某轮没产出
turn（如提前失败门、解析不到 JSON）时保留上一轮的值，
列表里始终显示最近一个真实回合的编号，而不是被空值
洗掉。
+ P0 安全批后：only_if_status 非空时写回带状态守卫
+ （UPDATE ... WHERE id = ? AND status = ?），状态已
+ 变（例如迟到的旧进程在任务已被取消/新一轮认领后才
+ 写回）时不覆盖新结果，返回 False；不传时保持旧行为
+ 并返回 True。调用方在终局路径必须传守卫，不能只靠
+ 写回前的 _set_status_if_running（两者之间仍有竞态窗）。
"""
conn = _db_connect(self.db_path)
try:
self._ensure_result_columns(conn)
+ status_clause = ""
+ extra = ()
+ if only_if_status is not None:
+ status_clause = " AND status = ?"
+ extra = (only_if_status,)
if child_pid is None:
- conn.execute(
+ cur = conn.execute(
"UPDATE tasks SET result_text = ?, error_text = ?,"
" actual_model = ?, started_at = ?, finished_at = ?,"
" turn_id = COALESCE(?, turn_id),"
" root_turn_id = COALESCE(?, root_turn_id),"
- " updated_at = ? WHERE id = ?",
+ " updated_at = ? WHERE id = ?" + status_clause,
((result_text or "")[:4000] if result_text else None,
(error_text or "")[:4000] if error_text else None,
actual_model, started_at, finished_at,
turn_id, root_turn_id,
- time.time(), task_id),
+ time.time(), task_id) + extra,
)
else:
- conn.execute(
+ cur = conn.execute(
"UPDATE tasks SET result_text = ?, error_text = ?,"
" actual_model = ?, started_at = ?, finished_at = ?,"
" child_pid = ?,"
" turn_id = COALESCE(?, turn_id),"
" root_turn_id = COALESCE(?, root_turn_id),"
- " updated_at = ? WHERE id = ?",
+ " updated_at = ? WHERE id = ?" + status_clause,
((result_text or "")[:4000] if result_text else None,
(error_text or "")[:4000] if error_text else None,
actual_model, started_at, finished_at, child_pid,
turn_id, root_turn_id,
- time.time(), task_id),
+ time.time(), task_id) + extra,
)
conn.commit()
+ return cur.rowcount == 1
finally:
conn.close()
@@ -1079,6 +1091,38 @@ class Worker:
project_id=task["project_id"], status="interrupted")
return claimed
+ def _docker_cancel_watchdog(self, task_id, container_name,
+ stop_event):
+ """Docker 同步路径的取消看门狗（_run_attempt 起的线程）。
+
+ 同步路径阻塞在 docker run 里，worker 的 cancel() 在
+ 另一线程只能改库、清 _running 项，阻塞中的 run 本身
+ 不会醒——P0 安全批前这就是“取消杀不掉容器”的根因。
+ 本线程每 1 秒查一次库：任务变 interrupted 或本进程
+ 收到取消标志时，按容器名真杀容器（kill+rm，尽力而
+ 为），然后退出；调用方在 _run_attempt 结束时置
+ stop_event 并 join。查库失败时当没取消，不误杀。
+ """
+ while not stop_event.wait(1.0):
+ try:
+ if task_id in self._cancel_requested:
+ cancelled = True
+ else:
+ db_task = self.get_task(task_id)
+ cancelled = (db_task is not None and
+ db_task.get("status") == "interrupted")
+ except Exception:
+ cancelled = False
+ if not cancelled:
+ continue
+ try:
+ kill = getattr(self.executor, "kill_container", None)
+ if kill is not None and container_name:
+ kill(container_name)
+ except Exception:
+ pass
+ return
+
def cleanup_stale_containers(self):
"""启动清理兜底：清掉上一轮残留的 codex- 容器。
@@ -1298,12 +1342,29 @@ class Worker:
quota_hit = False
started = time.time()
last_quota_check = 0.0
+ last_db_cancel_check = 0.0
try:
while proc.poll() is None:
if task_id in self._cancel_requested or self._loop_stop:
cancelled = True
self._kill_proc_group(proc)
break
+ # P0 安全批后：popen 轮询也要轮询库状态，不只
+ # 看本进程标志——控制台取消 / sweep 是另一进程
+ # 改库，旧代码只看本进程标志时杀不掉本进程的
+ # 子进程。节流到每 1 秒查一次库，避免 0.05 秒
+ # 轮询把库打爆；查库失败时当没取消，不误杀。
+ if time.time() - last_db_cancel_check >= 1.0:
+ last_db_cancel_check = time.time()
+ try:
+ db_task = self.get_task(task_id)
+ except Exception:
+ db_task = None
+ if db_task is not None and \
+ db_task.get("status") == "interrupted":
+ cancelled = True
+ self._kill_proc_group(proc)
+ break
if deadline_seconds is not None and \
time.time() - started >= deadline_seconds:
timed_out = True
@@ -2003,6 +2064,23 @@ class Worker:
float(self.executor.timeout))
except (TypeError, ValueError):
effective_docker_timeout = self.task_timeout
+ # P0 安全批后：Docker 同步路径起取消看门狗线程
+ # （阻塞在 docker run 里时，库状态变 interrupted 就
+ # 按容器名真杀；非 Docker 路径不启动，避免空线程）。
+ _watchdog_stop = None
+ _watchdog_thread = None
+ if is_docker and hasattr(self.executor, "kill_container"):
+ try:
+ _container_name = self.executor.container_name(task)
+ except Exception:
+ _container_name = None
+ if _container_name:
+ _watchdog_stop = threading.Event()
+ _watchdog_thread = threading.Thread(
+ target=self._docker_cancel_watchdog,
+ args=(task_id, _container_name, _watchdog_stop),
+ daemon=True)
+ _watchdog_thread.start()
try:
gateway_provider, gateway_env = self._ensure_gateway(task)
if self._use_popen():
@@ -2138,6 +2216,10 @@ class Worker:
"host_status": None, "stderr": ""}
error = str(exc)
finally:
+ if _watchdog_stop is not None:
+ _watchdog_stop.set()
+ if _watchdog_thread is not None:
+ _watchdog_thread.join(timeout=2)
# 同步路径执行结束：清掉取消用的容器标记（取消路径里
# 若已清过，这里再清一次无害）。进程模式的 proc 由
# _execute_via_process 自己的 finally 清，这里只清
@@ -2417,10 +2499,12 @@ class Worker:
if not final_written:
final = "interrupted"
finished_at = time.time()
- self._write_execution_result(
- task_id, result_text=report, error_text=error,
- actual_model=None, started_at=started_at,
- finished_at=finished_at)
+ # P0 安全批后：CAS 没改到行时不写回（见终局路径注释）。
+ if final_written:
+ self._write_execution_result(
+ task_id, result_text=report, error_text=error,
+ actual_model=None, started_at=started_at,
+ finished_at=finished_at, only_if_status=final)
self._cleanup_task_seed(task)
if final_written:
self._audit(project_id,
@@ -2489,7 +2573,8 @@ class Worker:
self._set_status(task_id, "failed")
self._write_execution_result(
task_id, error_text=error, actual_model=self.model,
- started_at=started_at, finished_at=time.time())
+ started_at=started_at, finished_at=time.time(),
+ only_if_status="failed")
# 缺 key 门提前返回：同样要清暂存，否则与仓库解析
# 失败一路同款残留（见 _cleanup_task_seed）。
self._cleanup_task_seed(task)
@@ -2516,7 +2601,8 @@ class Worker:
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())
+ started_at=started_at, finished_at=time.time(),
+ only_if_status="failed")
# 仓库解析失败提前返回：够不到正常终局后的清理，
# 这里必须先清暂存，否则 bundle 目录永久残留。
self._cleanup_task_seed(task)
@@ -2544,7 +2630,8 @@ class Worker:
self._set_status(task_id, "interrupted")
self._write_execution_result(
task_id, error_text=quota_error, actual_model=self.model,
- started_at=started_at, finished_at=time.time())
+ started_at=started_at, finished_at=time.time(),
+ only_if_status="interrupted")
self._audit(task["project_id"], "worker_quota_stop",
"%s: %s" % (task_id, quota_error))
log_event(self.logger, "worker_quota_stop", task_id=task_id,
@@ -2594,7 +2681,8 @@ class Worker:
self._set_status(task_id, "failed")
self._write_execution_result(
task_id, error_text=error, actual_model=self.model,
- started_at=started_at, finished_at=time.time())
+ started_at=started_at, finished_at=time.time(),
+ only_if_status="failed")
self._audit(task["project_id"], "worker_gateway_failed",
"%s: %s" % (task_id, error))
log_event(self.logger, "worker_gateway_failed",
@@ -2810,7 +2898,8 @@ class Worker:
error_text="cancelled", actual_model=actual_model,
started_at=started_at, finished_at=time.time(),
turn_id=(result or {}).get("turn_id"),
- root_turn_id=(result or {}).get("root_turn_id"))
+ root_turn_id=(result or {}).get("root_turn_id"),
+ only_if_status="interrupted")
log_event(self.logger, "worker_cancelled", task_id=task_id,
project_id=task["project_id"], status="interrupted")
return {"dispatched": True, "task_id": task_id,
@@ -2874,12 +2963,17 @@ class Worker:
"%s -> %s" % (self.model, nxt))
final_written = self._set_status_if_running(task_id, final)
finished_at = time.time()
- self._write_execution_result(
- task_id, result_text=stdout, error_text=error,
- actual_model=actual_model, started_at=started_at,
- finished_at=finished_at,
- turn_id=(result or {}).get("turn_id"),
- root_turn_id=(result or {}).get("root_turn_id"))
+ # P0 安全批后：CAS 没改到行（任务已被取消/回收）时
+ # 绝不写回结果，否则迟到的旧进程会覆盖新一轮的结果
+ # 文本；CAS 成功时写回也要带状态守卫。
+ if final_written:
+ self._write_execution_result(
+ task_id, result_text=stdout, error_text=error,
+ actual_model=actual_model, started_at=started_at,
+ finished_at=finished_at,
+ turn_id=(result or {}).get("turn_id"),
+ root_turn_id=(result or {}).get("root_turn_id"),
+ only_if_status=final)
# 附件暂存清理：任务终局后删 bundle 目录（重试走已存在
# 续跑，文件已在 worktree 里，不再需要暂存）。
self._cleanup_task_seed(task)
diff --git a/host.py b/host.py
index bedc601..03139a5 100644
--- a/host.py
+++ b/host.py
@@ -211,8 +211,15 @@ TRACE_MAX_LINES = 5000
# XCM_API_KEY 在 local 路可被模型拿到；server 自己的中转调用走
# 进程 env（env_key），不受 shell 策略影响，所以只排 shell 不影响
# 模型调用。inherit 保持默认 all（PATH、代理变量等工作流变量不
-# 动）；gh 用文件登录不用 env，故不排 *_TOKEN；四个精确名是当前
-# worker env 里的全部密钥类变量，三条模式兜底将来新增的同类变量。
+# 动）；gh 用文件登录不用 env，故排 *_TOKEN 不破坏 git 流程
+# （单测验证：gh 登录路径不读 token 环境变量，见 test_shell_env）。
+# P0 安全批后新增 *_TOKEN、*_AUTH*、*_CREDENTIAL*、*API_KEY*
+# 四条模式，兜底将来新增的同类变量。
+# 残留风险（必须如实写清，不谎称修完）：这只是 shell 过滤，
+# 不是安全边界——agent 与 server 进程同 UID 时，模型仍可读
+# server 进程的 /proc//environ 或其它未过滤变量；真正
+# 隔离需要 UID 分离或 hidepid 的部署变更，不在本批代码里，
+# 部署清单里必须单独列为后续项。
SHELL_ENV_EXCLUDE = [
"XCM_API_KEY",
"CLOUD_DB_PASSWORD",
@@ -221,6 +228,10 @@ SHELL_ENV_EXCLUDE = [
"*_KEY",
"*_SECRET",
"*_PASSWORD",
+ "*_TOKEN",
+ "*_AUTH*",
+ "*_CREDENTIAL*",
+ "*API_KEY*",
]