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