split: handler mixins for oauth/identity/project/automation/fullauto (94 methods) on integration base 02835f6

- CloudControlPlaneHandler mixes in OAuthMixin, IdentityMemoryMixin,
  ProjectMixin, AutomationMixin, FullAutoMixin; shared helpers move to
  cloud/handler_support.py with server.py re-exports (old import paths
  keep working, incl. _validate_env_branch for tests).
- Method bodies moved verbatim; routers, base plumbing, task/lifecycle,
  webhooks stay in server.py. No >100-line method body split.
- test_handler_split_conformance.py (6 tests) pins resolution, layer
  bans and method accounting (238 total, 144 + 94, no duplicates).
- test_p1_security_server.py: monkeypatch targets follow the defining
  module (handler_project MAX_PROJECTS_PER_USER, handler_identity
  relay helpers); assertions unchanged.

(cherry picked from commit 92165a2b097126470e4827327d96d592a253adf1)

cloud/handler_fullauto.py

@@ -0,0 +1,420 @@
+"""Handler mixin：FullAutoMixin。
+
+由 CloudControlPlaneHandler 拆出，方法体逐字搬移；
+self.* 调用（含其它 mixin 的方法与留守的 _send_json / _db 等）
+在混入后的同一类上解析，行为不变。本模块不 import cloud.server。
+"""
+
+from cloud.full_auto import (
+    FullAutoDenied,
+    FullAutoError,
+    audit_record,
+    auto_config_plan,
+    branch_protection_payload,
+    ci_fix_decision,
+    is_full_auto_enabled,
+    merge_gate,
+    release_plan,
+    rollback_plan,
+    run_plan,
+    validate_action,
+)
+from cloud.github_app import resolve_app_token as _resolve_app_token
+from cloud.github_app import (
+    find_installation_for_repo,
+    get_app_config,
+    list_installations,
+)
+import json
+import os
+import urllib.error
+import urllib.request
+
+
+class FullAutoMixin:
+    """FullAutoMixin 领域方法（见模块 docstring）。"""
+
+    # -- 全自动执行器 HTTP 端点（cloud/full_auto.py 的接线层） ------
+    # 读类端点（status/config-plan）只查计划与状态、不执行、不
+    # 碰 token；写类执行（execute）只限全局管理员，且须同时
+    # 满足：FULL_AUTO_ENABLED 显式开启 + 该仓库 installation 已
+    # 验证 + 主分支保护经 GitHub API 实测达标，三者缺一即
+    # 403 拒绝并记审计，绝不静默放行。审计一律经
+    # full_auto.audit_record 脱敏后落 AuthStore audit_log（action
+    # 前缀 full_auto_），响应与审计里绝不出现 token 明文、
+    # Bearer 明文或调用方塞进 detail 的敏感键。
+    def _full_auto_status(self, user_id, query):
+        repo = (query.get("repo") or [None])[0]
+        conn = self._db()
+        try:
+            configured = get_app_config(conn) is not None
+            verified = _full_auto_installation_verified(conn, repo)
+            installations = list_installations(conn)
+        finally:
+            conn.close()
+        flag = str(os.environ.get("FULL_AUTO_ENABLED", "")
+                   ).strip().lower() in ("1", "true", "yes")
+        self._send_json(200, {
+            "enabled": is_full_auto_enabled(os.environ, verified),
+            "flag_enabled": flag,
+            "installation_verified": verified,
+            "app_configured": configured,
+            "repo": repo,
+            "installations": installations,
+        })
+
+    def _full_auto_config_plan(self, user_id, query):
+        params, err = _full_auto_parse_plan_params({}, query)
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        try:
+            steps = auto_config_plan(**params)
+        except FullAutoError as exc:
+            self._send_json(400, {"error": str(exc),
+                                  "reason": exc.reason})
+            return
+        self._send_json(200, {"repo": params["repo_full_name"],
+                              "steps": steps})
+
+    def _full_auto_validate_action(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        action = (body or {}).get("action")
+        if not isinstance(action, str) or not action:
+            self._send_json(400, {"error": "action required"})
+            return
+        try:
+            validate_action(action)
+        except FullAutoDenied as exc:
+            self._send_json(403, {"error": "action denied",
+                                  "reason": exc.reason,
+                                  "action": action})
+            return
+        self._send_json(200, {"action": action, "allowed": True})
+
+    def _full_auto_merge_gate(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        body = body or {}
+        passed, reasons = merge_gate(
+            body.get("checks") or {},
+            body.get("required_checks") or (),
+            bool(body.get("directory_check_ok")),
+            bool(body.get("contract_version_ok")),
+            bool(body.get("coordinator_approved")))
+        self._send_json(200, {"passed": passed, "reasons": reasons})
+
+    def _full_auto_ci_fix(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        body = body or {}
+        try:
+            attempt = int(body.get("attempt"))
+        except (TypeError, ValueError):
+            self._send_json(400, {"error": "attempt must be an integer"})
+            return
+        self._send_json(200, ci_fix_decision(
+            attempt, body.get("failures") or []))
+
+    def _full_auto_release_plan(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        body = body or {}
+        if not isinstance(body.get("tag"), str) or not body.get("tag"):
+            self._send_json(400, {"error": "tag required"})
+            return
+        self._send_json(200, release_plan(
+            body["tag"], body.get("merged_prs") or []))
+
+    def _full_auto_rollback_plan(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        body = body or {}
+        try:
+            plan = rollback_plan(body.get("previous_tag"))
+        except FullAutoError as exc:
+            self._send_json(400, {"error": str(exc),
+                                  "reason": exc.reason})
+            return
+        self._send_json(200, plan)
+
+    def _full_auto_audit_list(self, user_id, query):
+        """审计查询：只限全局管理员，只返回 full_auto_ 前缀行，
+        且只返回已脱敏落盘的 detail（写时已脱敏，读时不再
+        含 token——调用方若直接写库绕过本层，不在本端点
+        的保证内，写路径见 _full_auto_execute）。"""
+        if not self._is_global_admin(user_id):
+            self._audit_admin(user_id, "full_auto_audit_denied",
+                              "audit list")
+            self._send_json(403, {"error": "forbidden"})
+            return
+        try:
+            limit = int((query.get("limit") or ["100"])[0])
+        except ValueError:
+            limit = 100
+        limit = max(1, min(limit, 100))
+        conn = self._db()
+        try:
+            rows = conn.execute(
+                "SELECT user_id, action, detail, created_at FROM audit_log"
+                " WHERE action LIKE 'full_auto_%'"
+                " ORDER BY created_at DESC, id DESC LIMIT ?",
+                (limit,)).fetchall()
+        except Exception:
+            rows = []
+        finally:
+            conn.close()
+        self._send_json(200, {"audit": [
+            {"user_id": r[0], "action": r[1], "detail": r[2],
+             "created_at": r[3]} for r in rows]})
+
+    def _full_auto_execute(self, user_id):
+        """执行自动配置计划：按 full_auto.auto_config_plan 生成
+        顺序计划，按 full_auto.run_plan 顺序执行，不跳步。
+
+        门禁顺序（任一步拒绝即停、记审计、不执行后续）：
+        1. 全局管理员；2. body.action 若给出须过 validate_action；
+        3. FULL_AUTO_ENABLED 显式开 + 该仓库 installation 已验证；
+        4. installation token 现解析现用（测试注入
+        server.full_auto_token_resolver 仅供单测绕过 JWT 签发，
+        生产为 None 时走 github_app.resolve_app_token）；
+        5. 主分支保护经 GitHub API 实测达标（带 token 查，
+        私有仓无 token 查保护必 404，故 token 先于保护核查）。
+        token 只进 Authorization 请求头；request_fn 注入点
+        server.full_auto_http_client 仅供单测，生产走 urllib。"""
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        body = body or {}
+        params, perr = _full_auto_parse_plan_params(body, {})
+        if perr:
+            self._send_json(400, {"error": perr})
+            return
+        repo = params["repo_full_name"]
+        branch = params["default_branch"]
+        checks = params["required_checks"]
+
+        def _deny(reason, detail):
+            record = audit_record("execute_denied", repo,
+                                  "full-auto-executor", "denied",
+                                  {"reason": reason, "detail": detail})
+            self._audit_admin(user_id, "full_auto_execute_denied",
+                              json.dumps(record, ensure_ascii=False))
+            self._send_json(403, {"error": "full-auto denied",
+                                  "reason": reason})
+        if not self._is_global_admin(user_id):
+            _deny("not_global_admin", "execute requires global admin")
+            return
+        action = body.get("action")
+        if action is not None:
+            if not isinstance(action, str) or not action:
+                self._send_json(400, {"error": "action must be a string"})
+                return
+            try:
+                validate_action(action)
+            except FullAutoDenied as exc:
+                _deny(exc.reason, "action=%s" % action)
+                return
+        conn = self._db()
+        try:
+            verified = _full_auto_installation_verified(conn, repo)
+        finally:
+            conn.close()
+        if not is_full_auto_enabled(os.environ, verified):
+            _deny("full_auto_disabled",
+                  "FULL_AUTO_ENABLED not set or installation unverified")
+            return
+        http_client = getattr(self.server, "full_auto_http_client", None)
+        request_fn = lambda m, u, h, b: _full_auto_github_request(  # noqa: E731
+            http_client, m, u, h, b)
+        resolver = getattr(self.server, "full_auto_token_resolver", None)
+        if resolver is not None:
+            token = resolver(self.server.db_path, repo)
+        else:
+            conn = self._db()
+            try:
+                token = _resolve_app_token(conn, repo)
+            finally:
+                conn.close()
+        if not token:
+            _deny("token_missing",
+                  "installation token unavailable for repo")
+            return
+        protection_url = ("https://api.github.com/repos/%s/branches/%s"
+                          "/protection" % (repo, branch))
+        try:
+            pstatus, protection = request_fn(
+                "GET", protection_url,
+                {"Accept": "application/vnd.github+json",
+                 "User-Agent": "codex-app-server",
+                 "Authorization": "Bearer " + token}, None)
+        except FullAutoError:
+            _deny("branch_protection_unverified",
+                  "protection check request failed")
+            return
+        if pstatus != 200 or not _full_auto_branch_protection_ok(
+                protection, checks):
+            _deny("branch_protection_unverified",
+                  "branch protection incomplete or unverified")
+            return
+        try:
+            steps = auto_config_plan(**params)
+        except FullAutoError as exc:
+            self._send_json(400, {"error": str(exc),
+                                  "reason": exc.reason})
+            return
+        results = []
+        try:
+            results = run_plan(request_fn, token, steps)
+        except FullAutoDenied as exc:
+            _deny(exc.reason, "plan step denied")
+            return
+        except FullAutoError as exc:
+            record = audit_record("auto_config", repo,
+                                  "full-auto-executor", "failed",
+                                  {"reason": exc.reason,
+                                   "detail": body.get("detail")})
+            self._audit_admin(user_id, "full_auto_auto_config",
+                              json.dumps(record, ensure_ascii=False))
+            self._send_json(502, {"error": str(exc),
+                                  "reason": exc.reason,
+                                  "steps": results})
+            return
+        record = audit_record("auto_config", repo,
+                              "full-auto-executor", "ok",
+                              {"detail": body.get("detail"),
+                               "steps": [r["step"] for r in results]})
+        self._audit_admin(user_id, "full_auto_auto_config",
+                          json.dumps(record, ensure_ascii=False))
+        # 响应只含步骤名/状态码/ok，绝不回显请求体 detail、
+        # headers 或 token——results 本身就是 run_plan 的
+        # (step/status/ok) 三元组，无凭据字段。
+        self._send_json(200, {"ok": True, "repo": repo, "steps": results})
+
+
+def _full_auto_parse_plan_params(body, query):
+    """从 POST body 或 GET query 里解析 auto_config_plan 参数。
+
+    repo 必填且须含 /；required_checks 可为列表或逗号字符串；
+    布尔字段接受 true/false/1/0 字符串与布尔值。非法返回
+    (None, error_string)。"""
+    def _get(name):
+        if isinstance(body, dict) and name in body:
+            return body[name]
+        values = (query or {}).get(name)
+        return values[0] if values else None
+
+    def _as_bool(value, default):
+        if value is None:
+            return default
+        if isinstance(value, bool):
+            return value
+        return str(value).strip().lower() in ("1", "true", "yes")
+
+    repo = _get("repo") or _get("repo_full_name")
+    if not isinstance(repo, str) or "/" not in repo:
+        return None, "repo must be owner/repo"
+    checks = _get("required_checks")
+    if checks is None:
+        checks = ("verify",)
+    elif isinstance(checks, str):
+        checks = tuple(c.strip() for c in checks.split(",") if c.strip())
+    elif isinstance(checks, (list, tuple)):
+        checks = tuple(str(c) for c in checks if str(c))
+    else:
+        return None, "required_checks must be a list or string"
+    params = {
+        "repo_full_name": repo,
+        "default_branch": _get("default_branch") or "main",
+        "org": _get("org"),
+        "repo_exists": _as_bool(_get("repo_exists"), True),
+        "ci_workflow_present": _as_bool(_get("ci_workflow_present"), False),
+        "required_checks": checks,
+    }
+    if params["org"] is not None and not isinstance(params["org"], str):
+        return None, "org must be a string"
+    return params, None
+
+def _full_auto_installation_verified(conn, repo_full_name=None):
+    """installation 已验证 = App 配置存在 + 安装记录存在。
+
+    repo 给出时查该仓库的安装记录（用户在 GitHub 侧安装 App
+    后 webhook 落库的记录）；repo 未给出时任一安装记录即算
+    已验证（仅供 /full-auto/status 总览用，执行仍须按仓库查）。
+    真实安装/重新接受只能由用户在 GitHub 侧完成，本层绝不
+    伪造安装记录、绝不把“无记录”说成已验证。"""
+    try:
+        if get_app_config(conn) is None:
+            return False
+        if repo_full_name:
+            return find_installation_for_repo(conn, repo_full_name) is not None
+        rows = list_installations(conn)
+        return bool(rows)
+    except Exception:
+        return False
+
+def _full_auto_branch_protection_ok(protection, required_checks):
+    """分支保护是否达到 full_auto 的强制标准：与
+    branch_protection_payload 的要求逐项核对，缺一项即不算
+    配好；required_checks 须全部在 contexts 里（可多不可少）。
+    protection 为非 dict 或 GET 失败时调用方已先行拒绝。"""
+    if not isinstance(protection, dict):
+        return False
+    try:
+        expected = branch_protection_payload(required_checks)
+    except FullAutoError:
+        return False
+    checks = protection.get("required_status_checks") or {}
+    if not isinstance(checks, dict) or not checks.get("strict"):
+        return False
+    contexts = checks.get("contexts") or []
+    if not set(expected["required_status_checks"]["contexts"]) <= set(contexts):
+        return False
+    if not protection.get("enforce_admins"):
+        return False
+    if not isinstance(protection.get("required_pull_request_reviews"), dict):
+        return False
+    if protection.get("allow_force_pushes") is not False:
+        return False
+    if protection.get("allow_deletions") is not False:
+        return False
+    return True
+
+def _full_auto_github_request(http_client, method, url, headers, body):
+    """full_auto.run_plan 用的 request_fn：注入的 fake 客户端
+    优先（单测）；生产走 urllib，token 只在 headers 里，响应
+    只返回 (status, dict)，绝不把 headers 写进返回值。"""
+    if http_client is not None:
+        result = http_client(method, url, headers, body)
+        if not (isinstance(result, tuple) and len(result) == 2):
+            raise FullAutoError("http_client must return (status, dict)",
+                                reason="auto_config_step_failed")
+        return result
+    request = urllib.request.Request(url, data=body, method=method,
+                                     headers=headers)
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            raw = response.read().decode("utf-8", "replace")
+            status = response.status
+    except urllib.error.HTTPError as exc:
+        raw = exc.read().decode("utf-8", "replace")
+        status = exc.code
+    try:
+        parsed = json.loads(raw)
+    except ValueError:
+        parsed = {}
+    if not isinstance(parsed, dict):
+        parsed = {}
+    return status, parsed