Merge pull request #96 from openqj/feat-p0-security

feat: P0 安全批（真全局管理员、双人配额门、默认配额、fix-review 收紧、App 配置限管理员、取消真杀任务）

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,),

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"})

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 = (
             "<!doctype html><html><head><meta charset='utf-8'>"
@@ -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):

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 双通道）。

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)
 

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()

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):

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,

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)

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")
 

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。
 

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)

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/<pid>/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*",
 ]