"""全自动执行器 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()