diff --git a/cloud/server.py b/cloud/server.py index bf9ae32..0f6a256 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -56,6 +56,8 @@ from contextlib import contextmanager from decimal import Decimal from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, quote, urlparse +import urllib.error +import urllib.request from cloud.auth import ROLES as _MEMBER_ROLES from cloud.auth import AuthStore @@ -196,6 +198,7 @@ from cloud.github_app import ( build_app_manifest, claim_fix_review, exchange_manifest_code, + find_installation_for_repo, get_app_config, list_installations, parse_installation_event, @@ -215,6 +218,21 @@ from cloud.webhooks import ( verify_github_signature_any, verify_slack_signature, ) +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 # PR 事件推送失败时记日志（systemd 下进 journal）；handler 的 # log_message 被重写为静默（测试输出干净），所以不用它。 @@ -1615,6 +1633,130 @@ def _global_admin_ids(): return {part.strip() for part in raw.split(",") if part.strip()} +# -- 全自动执行器 HTTP 接线（cloud/full_auto.py 的薄端点层） -------- +# 只做接线，不发明新护栏：开关、动作白名单、分支保护、脱敏、 +# 限次修复全部复用 full_auto.py；本层只负责认证、管理员门、 +# installation 验证、token 解析与审计落盘。token 只进请求头， +# 绝不进命令行、文件、审计记录与 HTTP 响应。 +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 + + +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 _backfill_default_quotas(conn): """给已有无行用户回填默认配额行（init_db 时调，幂等）。 @@ -2828,7 +2970,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(401, {"error": "unauthorized"}) return - if path == "/projects" and method == "POST": + if path == "/full-auto/status" and method == "GET": + self._full_auto_status(user_id, query) + elif path == "/full-auto/config-plan" and method == "GET": + self._full_auto_config_plan(user_id, query) + elif path == "/full-auto/validate-action" and method == "POST": + self._full_auto_validate_action(user_id) + elif path == "/full-auto/merge-gate" and method == "POST": + self._full_auto_merge_gate(user_id) + elif path == "/full-auto/ci-fix" and method == "POST": + self._full_auto_ci_fix(user_id) + elif path == "/full-auto/release-plan" and method == "POST": + self._full_auto_release_plan(user_id) + elif path == "/full-auto/rollback-plan" and method == "POST": + self._full_auto_rollback_plan(user_id) + elif path == "/full-auto/execute" and method == "POST": + self._full_auto_execute(user_id) + elif path == "/full-auto/audit" and method == "GET": + self._full_auto_audit_list(user_id, query) + elif path == "/projects" and method == "POST": self._create_project(user_id) elif path == "/projects" and method == "GET": self._list_projects(user_id) @@ -9830,6 +9990,275 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): conn.close() self._send_json(200, status) + # -- 全自动执行器 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}) + # -- API keys 管理 --------------------------------------------------- def _user_keys_list(self): """列出自己的 key：primary（id "legacy"）+ 附加 key。 @@ -10866,6 +11295,11 @@ def create_server(db_path, host="127.0.0.1", port=8765, backend=None): server.pr_gh_runner = None server.pr_gh_which = None server.host_worktrees_dir = None + # 全自动执行器注入点：生产为 None（走 urllib + 真实 + # installation token 解析）；测试可注入 fake http_client + # 与 token 解析器；绝不注入真实 token、真实 secret。 + server.full_auto_http_client = None + server.full_auto_token_resolver = None return server diff --git a/cloud/tests/test_fullauto_api.py b/cloud/tests/test_fullauto_api.py new file mode 100644 index 0000000..8bdff6c --- /dev/null +++ b/cloud/tests/test_fullauto_api.py @@ -0,0 +1,414 @@ +"""全自动执行器 HTTP 接线单测（cloud/server.py 的 /full-auto/* 端点）。 + +不跑真实模型、不联网、不碰生产与主机：GitHub API 一律走 +server.full_auto_http_client 的 fake 客户端，installation token +一律走 server.full_auto_token_resolver 的 fake 解析（绕过 JWT +签发，生产路径仍走 github_app.resolve_app_token）。测试里 +出现的 token/secret 一律是 fake 占位标记，断言目标正是： +这些标记绝不出现在 HTTP 响应与审计记录里。 + +覆盖：默认关闭、未验证安装拒绝、非管理员拒绝、护栏拒绝 +（禁止/未知动作）、分支保护未验证拒绝、执行成功与审计 +脱敏、配置计划顺序、合并门/修复限次/发版回滚、审计查询权限。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.github_app import record_installation_repos, save_app_config + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.github_app import record_installation_repos, save_app_config + from cloud.server import create_server + +import tempfile + +REPO = "owner/repo" +FAKE_TOKEN = "fake-test-token" +MARKER_SECRET = "marker-secret-value-12345" +MARKER_BEARER = "marker-bearer-value-67890" + +GOOD_PROTECTION = { + "required_status_checks": {"strict": True, "contexts": ["verify"]}, + "enforce_admins": True, + "required_pull_request_reviews": {"required_approving_review_count": 1}, + "allow_force_pushes": False, + "allow_deletions": False, +} + + +def _fake_client_factory(calls, protection=None): + def fake(method, url, headers, body): + calls.append({"method": method, "url": url, + "headers": dict(headers), + "body": body}) + if url.endswith("/branches/main/protection") and method == "GET": + return 200, dict(protection if protection is not None + else GOOD_PROTECTION) + if method in ("PATCH", "PUT"): + return 200, {} + return 404, {"message": "not found"} + return fake + + +class FullAutoApiTestCase(unittest.TestCase): + def setUp(self): + self._old_flag = os.environ.get("FULL_AUTO_ENABLED") + self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") + os.environ.pop("FULL_AUTO_ENABLED", None) + 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.outsider_id, self.outsider_key = self.auth.create_user( + "outsider") + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.server.rate_limit_per_minute = 10000 + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + conn = sqlite3.connect(self.db_path) + try: + save_app_config(conn, { + "app_id": "424242", "slug": "fake-codex-app", + "client_id": "Iv1.fake", + "webhook_secret": "fake-app-secret", + "pem": "-----BEGIN RSA PRIVATE KEY-----\nfake-pem\n" + "-----END RSA PRIVATE KEY-----\n", + "html_url": "https://github.com/apps/fake-codex-app"}) + record_installation_repos(conn, "123", "owner", [REPO]) + finally: + conn.close() + + 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_flag is None: + os.environ.pop("FULL_AUTO_ENABLED", None) + else: + os.environ["FULL_AUTO_ENABLED"] = self._old_flag + 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): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + raw = resp.read() + finally: + conn.close() + try: + parsed = json.loads(raw.decode("utf-8")) if raw else None + except ValueError: + parsed = None + return resp.status, parsed, raw.decode("utf-8", "replace") + + def _audit_details(self): + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute( + "SELECT action, detail FROM audit_log" + " WHERE action LIKE 'full_auto_%'").fetchall() + finally: + conn.close() + return rows + + def _enable(self): + os.environ["FULL_AUTO_ENABLED"] = "1" + + def _arm_execute(self, protection=None): + calls = [] + self.server.full_auto_http_client = _fake_client_factory( + calls, protection) + self.server.full_auto_token_resolver = ( + lambda db_path, repo: FAKE_TOKEN) + return calls + + # -- 状态：默认关闭 ---------------------------------------------- + def test_status_default_off_despite_verified_installation(self): + status, body, _ = self._req( + "GET", "/full-auto/status?repo=" + REPO, key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertFalse(body["enabled"]) + self.assertFalse(body["flag_enabled"]) + self.assertTrue(body["installation_verified"]) + self.assertTrue(body["app_configured"]) + + def test_status_enabled_only_with_flag_and_verified(self): + self._enable() + status, body, _ = self._req( + "GET", "/full-auto/status?repo=" + REPO, key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertTrue(body["enabled"]) + + def test_status_unverified_repo_never_enabled(self): + self._enable() + status, body, _ = self._req( + "GET", "/full-auto/status?repo=other/repo", + key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertFalse(body["installation_verified"]) + self.assertFalse(body["enabled"]) + + def test_status_requires_auth_and_leaks_no_secret(self): + status, body, raw = self._req( + "GET", "/full-auto/status?repo=" + REPO) + self.assertEqual(status, 401) + status, body, raw = self._req( + "GET", "/full-auto/status?repo=" + REPO, key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertNotIn("fake-app-secret", raw) + self.assertNotIn("fake-pem", raw) + + # -- 配置计划顺序 -------------------------------------------------- + def test_config_plan_order_and_branch_protection(self): + status, body, _ = self._req( + "GET", "/full-auto/config-plan?repo=" + REPO + + "&repo_exists=false&org=myorg&ci_workflow_present=false", + key=self.owner_key) + self.assertEqual(status, 200, body) + steps = [s["step"] for s in body["steps"]] + self.assertEqual(steps, ["ensure_repo", "set_default_branch", + "set_branch_protection", + "add_ci_workflow", "verify_webhook"]) + protection = body["steps"][2]["payload_summary"] + self.assertTrue(protection["enforce_admins"]) + self.assertTrue(protection["required_status_checks"]["strict"]) + self.assertFalse(protection["allow_force_pushes"]) + self.assertFalse(protection["allow_deletions"]) + self.assertIsNotNone( + protection["required_pull_request_reviews"]) + + def test_config_plan_user_repo_without_org_is_user_once(self): + status, body, _ = self._req( + "GET", "/full-auto/config-plan?repo=" + REPO + + "&repo_exists=false", + key=self.owner_key) + self.assertEqual(status, 200, body) + first = body["steps"][0] + self.assertEqual(first["manual"], "user_once") + self.assertIsNone(first["api"]) + + def test_config_plan_existing_repo_skips_ensure(self): + status, body, _ = self._req( + "GET", "/full-auto/config-plan?repo=" + REPO, + key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertNotIn("ensure_repo", + [s["step"] for s in body["steps"]]) + + def test_config_plan_bad_repo_rejected(self): + status, body, _ = self._req( + "GET", "/full-auto/config-plan?repo=norepo", + key=self.owner_key) + self.assertEqual(status, 400) + + # -- 护栏：动作校验 ------------------------------------------------ + def test_validate_action_forbidden_and_unknown_denied(self): + for action, reason in (("delete_repo", "forbidden_action"), + ("force_push_main", "forbidden_action"), + ("change_visibility", "forbidden_action"), + ("read_secret", "forbidden_action"), + ("nonsense_action", "unknown_action")): + status, body, _ = self._req( + "POST", "/full-auto/validate-action", + {"action": action}, self.owner_key) + self.assertEqual(status, 403, (action, body)) + self.assertEqual(body["reason"], reason) + status, body, _ = self._req( + "POST", "/full-auto/validate-action", + {"action": "create_pr"}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertTrue(body["allowed"]) + + # -- 执行门禁 ------------------------------------------------------ + def _execute_body(self, **extra): + body = {"repo": REPO, "repo_exists": True, + "ci_workflow_present": True} + body.update(extra) + return body + + def test_execute_default_off_denied_without_github_call(self): + calls = self._arm_execute() + status, body, _ = self._req( + "POST", "/full-auto/execute", self._execute_body(), + self.owner_key) + self.assertEqual(status, 403, body) + self.assertEqual(body["reason"], "full_auto_disabled") + self.assertEqual(calls, []) + + def test_execute_unverified_denied_without_github_call(self): + self._enable() + calls = self._arm_execute() + status, body, _ = self._req( + "POST", "/full-auto/execute", + self._execute_body(repo="other/repo"), self.owner_key) + self.assertEqual(status, 403, body) + self.assertEqual(body["reason"], "full_auto_disabled") + self.assertEqual(calls, []) + + def test_execute_non_admin_denied_without_github_call(self): + self._enable() + calls = self._arm_execute() + status, body, _ = self._req( + "POST", "/full-auto/execute", self._execute_body(), + self.outsider_key) + self.assertEqual(status, 403, body) + self.assertEqual(body["reason"], "not_global_admin") + self.assertEqual(calls, []) + + def test_execute_forbidden_action_denied_without_github_call(self): + self._enable() + calls = self._arm_execute() + status, body, _ = self._req( + "POST", "/full-auto/execute", + self._execute_body(action="delete_repo"), self.owner_key) + self.assertEqual(status, 403, body) + self.assertEqual(body["reason"], "forbidden_action") + self.assertEqual(calls, []) + + def test_execute_branch_protection_unverified_denied(self): + self._enable() + weak = dict(GOOD_PROTECTION) + weak["allow_force_pushes"] = True + calls = self._arm_execute(protection=weak) + status, body, _ = self._req( + "POST", "/full-auto/execute", self._execute_body(), + self.owner_key) + self.assertEqual(status, 403, body) + self.assertEqual(body["reason"], "branch_protection_unverified") + # 只查了保护、没执行任何计划步骤（无 PATCH/PUT）。 + self.assertTrue(all(c["method"] == "GET" for c in calls)) + self.assertTrue(calls) + + def test_execute_success_token_only_in_header_and_audit_redacted(self): + self._enable() + calls = self._arm_execute() + status, body, raw = self._req( + "POST", "/full-auto/execute", + self._execute_body(detail={ + "api_key": MARKER_SECRET, + "note": "Bearer " + MARKER_BEARER}), + self.owner_key) + self.assertEqual(status, 200, body) + self.assertTrue(body["ok"]) + self.assertEqual([s["step"] for s in body["steps"]], + ["set_default_branch", "set_branch_protection", + "verify_webhook"]) + self.assertTrue(all(s["ok"] for s in body["steps"])) + # token 只进请求头：全部调用的 Authorization 头正确， + # 且任何请求体里都没有 token；响应里也没有标记与 token。 + for call in calls: + self.assertEqual(call["headers"].get("Authorization"), + "Bearer " + FAKE_TOKEN) + if call["body"]: + self.assertNotIn(FAKE_TOKEN.encode(), call["body"]) + self.assertNotIn(FAKE_TOKEN, raw) + self.assertNotIn(MARKER_SECRET, raw) + self.assertNotIn(MARKER_BEARER, raw) + # 审计脱敏：api_key 键被删、Bearer 尾被抹，标记全不在。 + rows = self._audit_details() + self.assertTrue(any(a == "full_auto_auto_config" + for a, _ in rows)) + blob = json.dumps(rows, ensure_ascii=False) + self.assertNotIn(MARKER_SECRET, blob) + self.assertNotIn(MARKER_BEARER, blob) + self.assertNotIn(FAKE_TOKEN, blob) + + # -- 纯判定端点：合并门 / 修复限次 / 发版回滚 ---------------------- + def test_merge_gate_endpoint(self): + status, body, _ = self._req( + "POST", "/full-auto/merge-gate", + {"checks": {"verify": "success"}, "required_checks": ["verify"], + "directory_check_ok": True, "contract_version_ok": True, + "coordinator_approved": False}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertFalse(body["passed"]) + self.assertTrue(body["reasons"]) + status, body, _ = self._req( + "POST", "/full-auto/merge-gate", + {"checks": {"verify": "success"}, "required_checks": ["verify"], + "directory_check_ok": True, "contract_version_ok": True, + "coordinator_approved": True}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertTrue(body["passed"]) + self.assertEqual(body["reasons"], []) + + def test_ci_fix_endpoint_limited_to_three(self): + status, body, _ = self._req( + "POST", "/full-auto/ci-fix", + {"attempt": 3, "failures": ["verify"]}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["action"], "fix_and_rerun") + status, body, _ = self._req( + "POST", "/full-auto/ci-fix", + {"attempt": 4, "failures": ["verify"]}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["action"], "escalate") + + def test_release_and_rollback_plan_endpoints(self): + status, body, _ = self._req( + "POST", "/full-auto/release-plan", + {"tag": "v1", "merged_prs": [{"number": 1, "title": "x"}]}, + self.owner_key) + self.assertEqual(status, 200, body) + self.assertIn("never force-push", body["rollback"]) + status, body, _ = self._req( + "POST", "/full-auto/rollback-plan", + {"previous_tag": "v0"}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["previous_tag"], "v0") + self.assertIn("no force-push", body["method"]) + status, body, _ = self._req( + "POST", "/full-auto/rollback-plan", {}, self.owner_key) + self.assertEqual(status, 400) + + # -- 审计查询 ------------------------------------------------------ + def test_audit_list_admin_only(self): + self._enable() + self._arm_execute() + status, _, _ = self._req( + "POST", "/full-auto/execute", self._execute_body(), + self.owner_key) + self.assertEqual(status, 200) + status, body, _ = self._req( + "GET", "/full-auto/audit", key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertTrue(any(a["action"] == "full_auto_auto_config" + for a in body["audit"])) + status, body, _ = self._req( + "GET", "/full-auto/audit", key=self.outsider_key) + self.assertEqual(status, 403) + + +if __name__ == "__main__": + unittest.main()