host
d973e63b11
GET /full-auto/status, GET /full-auto/config-plan, POST /full-auto/validate-action|merge-gate|ci-fix|release-plan| rollback-plan, POST /full-auto/execute (global admin only, FULL_AUTO_ENABLED + verified installation + branch protection verified via GitHub API, else 403 + audit), GET /full-auto/audit. Guards reused from cloud/full_auto.py; token only in Authorization header; audit via audit_record redaction into AuthStore audit_log.
415 lines
17 KiB
Python
415 lines
17 KiB
Python
"""全自动执行器 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()
|