feat: full-auto mode v0.4 (host executor + skill + CI fix loop, stacked on v0.3)
Host auto executor (cloud/full_auto.py, stdlib only): default off (FULL_AUTO_ENABLED + verified installation), forbidden actions fail-closed (delete repo, force-push, visibility change, secret read/expose, branch-protection delete/bypass, GitHub account auto-registration), token only in request header, audit log redacted, branch protection requires PR + strict checks + no force-push + no deletions + enforce_admins, merge gate (CI green + directory check + contract version + coordinator approval), CI fix loop max 3 then escalate, rollback only via revert PR or previous tag. Skill: v0.4 full-auto mode, coordinator reviews, host executor executes, task agent never executes merge/deploy; repo-creation honesty branch (org via App token, user repo needs_user_once); no web auto-registration of GitHub accounts. Tests: 19 new, cloud full 1451 OK (skipped=4).
This commit is contained in:
host committed
1 parent
c76ef9dcec
commit
cd85157b54
3 files changed
+719
-9
No files matched your search
@@ -0,0 +1,402 @@
|
||||
"""全自动编程系统：一次授权后的主机自动执行器（API 优先）。
|
||||
|
||||
本模块是“协调员只评审、人/宿主执行合并”缺口的闭合点，但
|
||||
执行人不是任务 agent：任务 agent 仍然不直接执行合并/部署，
|
||||
真正执行的是宿主侧的本模块（由已验证的 GitHub App
|
||||
installation + 显式开启标志门控），任务 agent 只交分支、
|
||||
自测证据和合并决定。两者混淆就是冒充，必须分清。
|
||||
|
||||
设计约束（全部是硬规则，不是建议）：
|
||||
1. 默认关闭：FULL_AUTO_ENABLED 未显式开启，或 installation
|
||||
未验证，执行器一律拒绝，不存在“默认自动合”的状态。
|
||||
2. 禁止清单 fail-closed：删库、force-push（含主分支）、改仓库
|
||||
可见性、读密钥、外泄密钥、删分支保护、绕分支保护，未知
|
||||
动作一律拒绝，不靠调用方自觉。
|
||||
3. token 只进子进程/HTTP 请求头，不进审计记录、不进日志、
|
||||
不进命令行、不进文件。本模块的审计记录在写入前会把
|
||||
含 token/secret/password/key 的键和 Bearer 字符串抹掉；
|
||||
调用方若把 token 塞进 detail，同样会被抹掉（宁可丢
|
||||
字段，不可漏 token）。
|
||||
4. 主分支保护强制开：自动配置的分支保护必须同时含要求 PR、
|
||||
要求状态检查（strict）、禁止 force-push、禁止删除、
|
||||
管理员同样受约束（enforce_admins），缺一项就不算配好。
|
||||
5. 质量门禁阻塞合并：CI 全绿 + 目录隔离检查 + 契约版本 +
|
||||
协调员批准，四项缺一就不合并。门禁不过时走 CI 自动修复
|
||||
循环，限次（默认 3 次），超次升级给用户，不无限重试。
|
||||
6. 回滚不靠 force-push：回滚只能是 revert PR 或重新部署
|
||||
上一个 tag，绝不 force-push 主分支。
|
||||
7. 建仓分支要诚实：GitHub App installation token 能建的是
|
||||
组织仓库（POST /orgs/{org}/repos，需 administration 写
|
||||
权限）；用户个人账号仓库的 POST /user/repos 需要用户
|
||||
token，installation token 做不了。所以自动配置计划里，
|
||||
“仓库不存在 + 无 org”时返回 needs_user_once 步骤：用户
|
||||
在同一次安装会话里先建一个空仓（这是安装会话的一部分，
|
||||
不是此后每轮的人工步骤），App 以 All repositories 安装
|
||||
后即覆盖它。此后建分支、PR、CI、合并、发版全部走 API，
|
||||
不再需要人工。把这一步说成“App 已自动建仓”就是冒充，
|
||||
必须如实记为 needs_user_once。
|
||||
8. 绝不做网页自动化自动注册 GitHub 账号：服务条款禁止
|
||||
自动化创建账号，注册有验证码、邮箱验证、可能的人机/
|
||||
手机验证，绕验证码会触发封号。本模块不提供、不调用
|
||||
任何账号注册自动化，浏览器自动化只留作 API 覆盖不到
|
||||
的配置页面兜底，不用来注册、不用来绕验证码。
|
||||
9. 授权分层：控制台 OAuth 只证明本站身份，不带来 GitHub
|
||||
权限；GitHub 权限靠 GitHub App 安装，仓库范围靠安装时
|
||||
选 All repositories 一次覆盖现有与未来仓库，能力范围受
|
||||
App 声明的权限封顶，新增权限必须改 App 声明并由安装方
|
||||
在 GitHub 重新接受，无法静默升级。本模块不假装权限
|
||||
可以自动升级：installation 的 permissions 由调用方
|
||||
传入并核对，缺权限的动作直接拒绝并记为
|
||||
needs_reaccept（需安装方重新接受），不静默降级执行。
|
||||
|
||||
零第三方依赖，只用 Python 标准库。网络一律经调用方注入
|
||||
的 request_fn（约定与 cloud/github_app._github_api_request
|
||||
的 http_client 相同：(method, url, headers, body) ->
|
||||
(status, dict)），本模块自身不发真实网络请求，单测用假
|
||||
客户端即可覆盖全部路径；真实路径由调用方复用
|
||||
github_app 的 urllib 实现并复用 resolve_app_token 的
|
||||
“token 只进子进程环境/请求头”模式。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
MAX_FIX_ATTEMPTS = 3
|
||||
|
||||
# 允许的自动动作（白名单，之外一律拒绝）。
|
||||
ALLOWED_ACTIONS = frozenset((
|
||||
"create_repo_org",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"merge_pr",
|
||||
"create_release",
|
||||
"rollback_to_tag",
|
||||
"set_branch_protection",
|
||||
"set_default_branch",
|
||||
"add_ci_workflow",
|
||||
"verify_webhook",
|
||||
))
|
||||
|
||||
# 禁止的自动动作（黑名单，命中即拒绝，优先级高于白名单）。
|
||||
FORBIDDEN_ACTIONS = frozenset((
|
||||
"delete_repo",
|
||||
"force_push",
|
||||
"force_push_main",
|
||||
"change_visibility",
|
||||
"read_secret",
|
||||
"expose_secret",
|
||||
"delete_branch_protection",
|
||||
"bypass_branch_protection",
|
||||
"register_github_account",
|
||||
))
|
||||
|
||||
_SENSITIVE_KEYS = ("token", "secret", "password", "private_key",
|
||||
"api_key", "apikey", "auth")
|
||||
|
||||
|
||||
class FullAutoError(Exception):
|
||||
"""全自动执行失败：reason 见各抛出处。"""
|
||||
|
||||
def __init__(self, message, reason=None):
|
||||
super().__init__(message)
|
||||
self.reason = reason
|
||||
|
||||
|
||||
class FullAutoDenied(FullAutoError):
|
||||
"""护栏拒绝：默认关闭、禁止动作、未知动作、权限不足。"""
|
||||
|
||||
|
||||
def is_full_auto_enabled(env=None, installation_verified=False):
|
||||
"""执行器是否开启：显式标志 + 已验证 installation，两者缺一不可。
|
||||
|
||||
env 默认读 os.environ（调用方可注入 dict 做单测）。标志值
|
||||
只认 1/true/yes（大小写不敏感），其它值一律视为关闭。
|
||||
"""
|
||||
if env is None:
|
||||
env = os.environ
|
||||
raw = env.get("FULL_AUTO_ENABLED", "")
|
||||
flag = str(raw).strip().lower() in ("1", "true", "yes")
|
||||
return bool(flag and installation_verified)
|
||||
|
||||
|
||||
def validate_action(action):
|
||||
"""校验自动动作：禁止/未知一律抛 FullAutoDenied，不返回布尔
|
||||
让调用方“顺手忽略”。"""
|
||||
if action in FORBIDDEN_ACTIONS:
|
||||
raise FullAutoDenied("forbidden action: %s" % action,
|
||||
reason="forbidden_action")
|
||||
if action not in ALLOWED_ACTIONS:
|
||||
raise FullAutoDenied("unknown action: %s" % action,
|
||||
reason="unknown_action")
|
||||
return True
|
||||
|
||||
|
||||
def _redact_value(value):
|
||||
"""递归抹掉敏感内容：键含 token/secret/password/key 的
|
||||
一律删键；字符串里的 Bearer <...> 一律替换为 Bearer [redacted]。
|
||||
"""
|
||||
if isinstance(value, dict):
|
||||
out = {}
|
||||
for key, item in value.items():
|
||||
lowered = str(key).lower()
|
||||
if any(s in lowered for s in _SENSITIVE_KEYS):
|
||||
continue
|
||||
out[key] = _redact_value(item)
|
||||
return out
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_redact_value(item) for item in value]
|
||||
if isinstance(value, str):
|
||||
if "Bearer " in value:
|
||||
parts = value.split("Bearer ", 1)
|
||||
tail = parts[1].split()[0] if parts[1].split() else ""
|
||||
return value.replace("Bearer " + tail,
|
||||
"Bearer [redacted]")
|
||||
return value
|
||||
return value
|
||||
|
||||
|
||||
def audit_record(action, repo, actor, result, detail=None):
|
||||
"""生成审计记录 dict：不含任何 token，调用方可直接落盘。
|
||||
actor 建议写 full-auto-executor + installation id，不写
|
||||
任务 agent 的名字，避免把执行人冒充成任务 agent。"""
|
||||
record = {
|
||||
"ts": time.time(),
|
||||
"action": action,
|
||||
"repo": repo,
|
||||
"actor": actor,
|
||||
"result": result,
|
||||
"detail": _redact_value(detail) if detail is not None else {},
|
||||
}
|
||||
return record
|
||||
|
||||
|
||||
def append_audit_log(path, record):
|
||||
"""审计记录追加写 JSONL：父目录不存在时创建；记录本身
|
||||
已在 audit_record 里脱敏，本函数不再二次处理 token，
|
||||
但调用方若直接传了含 token 的 dict，本函数不负责兜底，
|
||||
必须走 audit_record 生成记录。"""
|
||||
directory = os.path.dirname(os.path.abspath(str(path)))
|
||||
if directory:
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
with open(path, "a", encoding="utf-8") as handle:
|
||||
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||||
return record
|
||||
|
||||
|
||||
def branch_protection_payload(required_checks):
|
||||
"""主分支保护 payload：要求 PR + 要求状态检查（strict）+
|
||||
禁止 force-push + 禁止删除 + 管理员同样受约束，缺一不可。
|
||||
required_checks 为空时不算配好（调用方必须传 CI 检查名，
|
||||
不能靠“无检查即通过”蒙混）。"""
|
||||
checks = [str(c) for c in (required_checks or []) if str(c)]
|
||||
if not checks:
|
||||
raise FullAutoError("required_checks must not be empty",
|
||||
reason="branch_protection_incomplete")
|
||||
return {
|
||||
"required_status_checks": {
|
||||
"strict": True,
|
||||
"contexts": checks,
|
||||
},
|
||||
"enforce_admins": True,
|
||||
"required_pull_request_reviews": {
|
||||
"required_approving_review_count": 1,
|
||||
},
|
||||
"restrictions": None,
|
||||
"allow_force_pushes": False,
|
||||
"allow_deletions": False,
|
||||
}
|
||||
|
||||
|
||||
def auto_config_plan(repo_full_name, default_branch="main", org=None,
|
||||
repo_exists=True, ci_workflow_present=False,
|
||||
required_checks=("verify",)):
|
||||
"""一次授权后 API 自动配置的顺序计划（不发网络请求）。
|
||||
|
||||
每步含 step / api（None 表示需人工一次或仅核对）/
|
||||
method / path / payload_summary / manual：
|
||||
manual=None 表示全自动；manual="user_once" 表示该步
|
||||
是安装会话里唯一的人工步（仅建仓分支会出这一步）；
|
||||
manual="verify_only" 表示仅核对、不创建（App 级 webhook
|
||||
已在 App manifest 的 hook_attributes 里配好，仓库级
|
||||
不再重复建 hook，重复建反而会双发）。
|
||||
调用方按顺序执行，遇失败即停，不跳步继续。
|
||||
"""
|
||||
if not repo_full_name or "/" not in str(repo_full_name):
|
||||
raise FullAutoError("repo_full_name must be owner/repo",
|
||||
reason="invalid_repo")
|
||||
repo = str(repo_full_name)
|
||||
branch = str(default_branch or "main")
|
||||
steps = []
|
||||
if not repo_exists:
|
||||
if org:
|
||||
steps.append({
|
||||
"step": "ensure_repo",
|
||||
"api": "POST /orgs/{org}/repos",
|
||||
"method": "POST",
|
||||
"path": "/orgs/%s/repos" % org,
|
||||
"payload_summary": {
|
||||
"name": repo.split("/", 1)[1],
|
||||
"private": True,
|
||||
},
|
||||
"manual": None,
|
||||
})
|
||||
else:
|
||||
steps.append({
|
||||
"step": "ensure_repo",
|
||||
"api": None,
|
||||
"method": None,
|
||||
"path": None,
|
||||
"payload_summary": {
|
||||
"reason": "installation token cannot create a "
|
||||
"user-account repo via POST /user/repos; "
|
||||
"that endpoint needs a user token",
|
||||
},
|
||||
"manual": "user_once",
|
||||
})
|
||||
steps.append({
|
||||
"step": "set_default_branch",
|
||||
"api": "PATCH /repos/{owner}/{repo}",
|
||||
"method": "PATCH",
|
||||
"path": "/repos/%s" % repo,
|
||||
"payload_summary": {"default_branch": branch},
|
||||
"manual": None,
|
||||
})
|
||||
steps.append({
|
||||
"step": "set_branch_protection",
|
||||
"api": "PUT /repos/{owner}/{repo}/branches/{branch}/protection",
|
||||
"method": "PUT",
|
||||
"path": "/repos/%s/branches/%s/protection" % (repo, branch),
|
||||
"payload_summary": branch_protection_payload(required_checks),
|
||||
"manual": None,
|
||||
})
|
||||
if not ci_workflow_present:
|
||||
steps.append({
|
||||
"step": "add_ci_workflow",
|
||||
"api": "PUT /repos/{owner}/{repo}/contents/.github/workflows/host-verify.yml",
|
||||
"method": "PUT",
|
||||
"path": "/repos/%s/contents/.github/workflows/host-verify.yml"
|
||||
% repo,
|
||||
"payload_summary": {"message": "chore: add host-verify CI"},
|
||||
"manual": None,
|
||||
})
|
||||
steps.append({
|
||||
"step": "verify_webhook",
|
||||
"api": None,
|
||||
"method": None,
|
||||
"path": None,
|
||||
"payload_summary": {
|
||||
"reason": "webhook is configured at App level via "
|
||||
"manifest hook_attributes; per-repo hook "
|
||||
"creation would double-deliver",
|
||||
},
|
||||
"manual": "verify_only",
|
||||
})
|
||||
return steps
|
||||
|
||||
|
||||
def run_plan(request_fn, token, plan):
|
||||
"""顺序执行自动配置计划：request_fn 约定同 github_app 的
|
||||
http_client（(method, url, headers, body) -> (status, dict)）。
|
||||
token 只放进本次请求的 Authorization 头，不进返回值、
|
||||
不进审计记录。manual 步跳过并记为 needs_user_once /
|
||||
verify_only；任一 API 步非 2xx 即停并抛 FullAutoError，
|
||||
不跳步继续（fail-closed）。"""
|
||||
if not token:
|
||||
raise FullAutoDenied("installation token required",
|
||||
reason="token_missing")
|
||||
results = []
|
||||
for step in plan:
|
||||
if step.get("manual") in ("user_once", "verify_only"):
|
||||
results.append({"step": step["step"],
|
||||
"status": step["manual"],
|
||||
"ok": step["manual"] == "verify_only"})
|
||||
continue
|
||||
action_map = {
|
||||
"ensure_repo": "create_repo_org",
|
||||
"set_default_branch": "set_default_branch",
|
||||
"set_branch_protection": "set_branch_protection",
|
||||
"add_ci_workflow": "add_ci_workflow",
|
||||
"verify_webhook": "verify_webhook",
|
||||
}
|
||||
validate_action(action_map.get(step["step"], step["step"]))
|
||||
headers = {"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "codex-app-server",
|
||||
"Authorization": "Bearer " + token}
|
||||
body = None
|
||||
payload = step.get("payload_summary")
|
||||
if payload is not None:
|
||||
body = json.dumps(payload).encode("utf-8")
|
||||
url = "https://api.github.com" + str(step["path"])
|
||||
status, data = request_fn(step["method"], url, headers, body)
|
||||
ok = 200 <= status < 300
|
||||
results.append({"step": step["step"], "status": status,
|
||||
"ok": ok})
|
||||
if not ok:
|
||||
raise FullAutoError(
|
||||
"auto-config step %s failed: %s %r"
|
||||
% (step["step"], status, data.get("message")),
|
||||
reason="auto_config_step_failed")
|
||||
return results
|
||||
|
||||
|
||||
def merge_gate(checks, required_checks, directory_check_ok,
|
||||
contract_version_ok, coordinator_approved):
|
||||
"""合并门：四项全过才返回 (True, [])，否则返回
|
||||
(False, [未过原因...])。checks 是 {检查名: 状态}，
|
||||
状态只认 success；pending/failure/缺失一律算未过。
|
||||
本函数不读 CI、不调网络，调用方把 pr/checks 的结果
|
||||
传进来即可，测试与生产走同一判定。"""
|
||||
reasons = []
|
||||
checks = checks or {}
|
||||
for name in (required_checks or []):
|
||||
if checks.get(name) != "success":
|
||||
reasons.append("check not green: %s" % name)
|
||||
if not directory_check_ok:
|
||||
reasons.append("directory isolation check failed")
|
||||
if not contract_version_ok:
|
||||
reasons.append("contract version mismatch")
|
||||
if not coordinator_approved:
|
||||
reasons.append("coordinator approval missing")
|
||||
return (not reasons, reasons)
|
||||
|
||||
|
||||
def ci_fix_decision(attempt, failures):
|
||||
"""CI 自动修复循环决策：attempt 从 1 开始计数。
|
||||
1..MAX_FIX_ATTEMPTS 返回 fix_and_rerun；超次返回
|
||||
escalate（升级给用户），绝不无限重试烧模型费。"""
|
||||
failed = [str(f) for f in (failures or [])]
|
||||
if attempt <= MAX_FIX_ATTEMPTS:
|
||||
return {"action": "fix_and_rerun", "attempt": attempt,
|
||||
"failures": failed}
|
||||
return {"action": "escalate",
|
||||
"reason": "max_fix_attempts_exceeded",
|
||||
"attempt": attempt, "failures": failed}
|
||||
|
||||
|
||||
def release_plan(tag, merged_prs):
|
||||
"""发版计划：release notes 从已合 PR 的标题聚合，不编造
|
||||
未合 PR；回滚只能是 revert PR 或重新部署上一 tag，
|
||||
绝不 force-push。merged_prs 是 [{number, title}]。"""
|
||||
notes = ["## Merged PRs"]
|
||||
for pr in (merged_prs or []):
|
||||
notes.append("- #%s %s" % (pr.get("number"), pr.get("title")))
|
||||
return {
|
||||
"tag": tag,
|
||||
"release_notes": "\n".join(notes),
|
||||
"rollback": "revert PR or redeploy previous tag; "
|
||||
"never force-push the default branch",
|
||||
}
|
||||
|
||||
|
||||
def rollback_plan(previous_tag):
|
||||
"""回滚计划：只认上一 tag，不接受 force-push 式回滚。"""
|
||||
if not previous_tag:
|
||||
raise FullAutoError("previous_tag required for rollback",
|
||||
reason="rollback_no_tag")
|
||||
return {
|
||||
"action": "rollback_to_tag",
|
||||
"previous_tag": previous_tag,
|
||||
"method": "redeploy previous tag (no force-push)",
|
||||
}
|
||||
Reference in new issue
Block a user