"""服务端合并硬门禁：五项全查、缺一 409、fail-closed（纯函数）。 本模块堆叠在 #209（cloud/merge_gate.py）之上，不改 #209 的语义： - #209 的 ``evaluate_merge_gate`` / ``gate_from_request`` 原样保留： record 默认、无证据不阻断、unknown 只记风险不阻断。那些是 #209 的过渡语义，本模块不碰它们，所以 #209 的测试一字不用改。 - 本模块只新增一个模式值 ``hard``，以及服务端用它时的模式解析： ``hard_mode_from_env``。服务端（POST /tasks/{id}/merge）以本模块 的解析为准，#209 的 ``gate_from_request`` 不再是服务端的模式来源 （它仍可作纯函数使用，但服务端不再经由它决定模式——否则调用方 可用请求体 ``merge_gate_mode=record`` 把硬门禁降级，那是 fail 但照样合并的路径，本模块堵的就是它）。 模式（环境变量 CLOUD_MERGE_GATE_MODE，运维定、调用方改不了）： - hard（未设或非法值时都落到 hard，fail-closed，绝不回落 record/off）： 下面五项必须全部通过，缺一就阻断。调用方请求体里的 ``merge_gate_mode`` 在服务端被忽略——模式只由环境变量决定。 - record / enforce / off（必须显式设置）：原样沿用 #209 的语义， 属过渡/显式降级。record 下无证据调用仍按旧接口放行（只附带门禁 信息），enforce 下未保护一律阻断，off 下只记风险不阻断。切到 这三者是运维的显式决定，不是调用方的默认权利。 硬门禁五项（全部是调用方在 /merge 请求体里提交的证据； 本服务没有可信渠道直连 GitHub 核实，所以门禁强制的是“证据齐全 且全绿”，不是“服务端已向 GitHub 求证”——这条诚实边界必须写清， 调用方谎报证据不在本模块的验证范围内，由审计（第 3 轨）与 PR/Checks 的只读核对去覆盖）： ① CI 全绿：``ci_status`` 必须为 ``"success"``，且 ``ci_checks`` 非空、每项 status 均为 ``"success"``，且 ``required_checks`` 非空、每项都在 ``ci_checks`` 里且为 success。空 ``required_checks``（``[]``）= 未配置，按分支保护 payload 的同一口径处理（见 docs/merge-gate-branch-protection.md）， 不是“没有必须项所以通过”。 ② 目录隔离：复用 #209 的 ``check_directory_isolation``（v0.3 口径： merge-base 范围 ``git diff --name-only``，拥有目录前缀检查， 含 ``apps/web2`` 非子目录边界与 ``../``、绝对路径逃逸）。 ``files_changed`` 缺失 = 无证据，直接阻断（#209 里是跳过， 硬模式里是阻断，这是本轨改动的核心之一）；越目录只能靠 ``approved_exceptions``（协调员批准的契约变更文件）放行， 显式确认绝不覆盖目录违规。 ③ 合同版本：``contract_version`` 非空，且 ``contract_frozen`` 为 true；若同时给了 ``contract_expected_version``（冻结版）， 两者必须相等，否则阻断。服务端没有独立的合同版本存储， 冻结版由调用方（协调员侧）声明， mismatch 时阻断。 ④ 协调员批准：``coordinator_approved`` 必须为 true；缺失与 显式 false 用不同的 block 码区分（缺证据 vs 已否决）。 ⑤ 无未解决评审意见：``unresolved_review_comments`` 必须为 0 （整数）或空列表；缺失 = 无证据，阻断；大于 0 / 非空 = 阻断。 分支保护（硬模式下的第六条隐含项，来自 #209 的保护定义）： ``branch_protection`` 必须为 ``"protected"``。``"unprotected"`` 即使带 ``merge_gate_confirm`` 也阻断——硬模式没有“确认即放行” 的逃生口，确认只在 record 模式下覆盖保护风险。``"unknown"`` （未声明或非法值）同样阻断：unknown 不等于 protected，这是 #209 里“unknown 只记风险”的过渡语义在硬模式下的收紧。 与第 3 轨（独立审计 + 自动推回）的接口约定（本模块不造推回状态机）： - 阻塞依据 = 409 响应里的 ``merge_gate`` 对象：``mode``、 ``blocks``（稳定的机器可读码，下面的 _BLOCK_DESCRIPTIONS 有 逐条中文说明）、以及 ``ci`` / ``directory`` / ``contract`` / ``coordinator`` / ``reviews`` 五个分项。推回包逐条取 blocks， 不要解析审计 detail 文本。 - 审计依据 = 每次门禁阻断写一条 ``merge_gate_blocked`` 审计 （action 名固定，detail 为 ``" mode= blocks=<逗号分隔的 blocks>"``），任务状态不变（UPDATE 不执行）。 - 门禁结论是单次调用的判定，不持久化为任务状态：一次 200 不 代表之后另一次合并自动通过；修复后必须重新提交完整证据、 重新走全部五项（完整重审，非只审改动点）。 - 本模块不建推回任务、不发通知、不改任务状态机：推回执行、 轮次上限、升级路径全部归第 3 轨。 """ from cloud.merge_gate import ( PROTECTION_STATES, RISK_UNKNOWN, RISK_UNPROTECTED, check_directory_isolation, ) HARD_MODES = ("hard", "record", "enforce", "off") _BLOCK_DESCRIPTIONS = { "ci_evidence_missing": "CI 证据缺失（ci_status / ci_checks / required_checks 有一项未给）", "ci_not_green": "CI 状态不是 success", "ci_check_failed": "CI 检查项中有非 success 项", "ci_required_checks_unconfigured": "required_checks 为空 = 未配置", "ci_required_check_missing": "required_checks 中有项不在 CI 检查里或不是 success", "directory_evidence_missing": "目录证据缺失（files_changed 未给）", "directory_owned_dir_unknown": "拥有目录未知，无法核对目录隔离", "directory_isolation_violation": "目录隔离违规（越拥有目录）", "contract_version_missing": "合同版本缺失", "contract_not_frozen": "合同未冻结（contract_frozen 不是 true）", "contract_version_mismatch": "合同版本与冻结版不一致", "coordinator_approval_missing": "协调员批准证据缺失", "coordinator_not_approved": "协调员未批准", "review_evidence_missing": "评审意见证据缺失", "unresolved_review_comments": "存在未解决评审意见", "branch_protection_unknown": "分支保护状态未知（unknown 不等于 protected）", "branch_protection_unprotected": "分支未保护（硬模式下显式确认也不放行）", } def hard_mode_from_env(env=None): """服务端有效的门禁模式：环境变量定，调用方改不了。 未设、空值或非法值一律回落 hard（fail-closed）。注意这与 #209 的 gate_mode_from_env 不同：后者非法值回落 record，那是 #209 纯函数的过渡语义；服务端模式解析必须 fail-closed，所以 这里绝不回落到 record/off。 """ import os raw = (env if env is not None else os.environ).get( "CLOUD_MERGE_GATE_MODE", "") mode = (raw or "").strip().lower() return mode if mode in HARD_MODES else "hard" def _ci_checks_index(ci_checks): """ci_checks 列表 -> {name: status}；条目不合规时返回 None。""" index = {} for entry in ci_checks: if not isinstance(entry, dict): return None name = entry.get("name") status = entry.get("status") if not isinstance(name, str) or not name \ or not isinstance(status, str): return None index[name] = status return index def evaluate_hard_merge_gate(files_changed=None, owned_dir=None, approved_exceptions=(), protection="unknown", ci_status=None, ci_checks=None, required_checks=None, contract_version=None, contract_frozen=False, contract_expected_version=None, coordinator_approved=None, unresolved_review_comments=None): """硬门禁纯函数：五项 + 保护，缺一即 allowed=False。 files_changed=None、ci_checks=None、required_checks=None、 unresolved_review_comments=None 全部是“无证据”，不是“跳过”。 protection 非法值归一化为 unknown，随后按 unknown 阻断。 返回 dict：allowed / mode / protection / directory / ci / contract / coordinator / reviews / risks / blocks。 """ protection = protection if protection in PROTECTION_STATES \ else "unknown" risks = [] blocks = [] # ① CI 全绿。 if ci_status is None or ci_status == "" \ or ci_checks is None or required_checks is None: blocks.append("ci_evidence_missing") else: if ci_status != "success": blocks.append("ci_not_green") if not isinstance(ci_checks, list) or not ci_checks: blocks.append("ci_evidence_missing") index = None else: index = _ci_checks_index(ci_checks) if index is None: blocks.append("ci_check_failed") elif any(status != "success" for status in index.values()): blocks.append("ci_check_failed") if isinstance(required_checks, list) and not required_checks: blocks.append("ci_required_checks_unconfigured") elif isinstance(required_checks, list) and index is not None: missing = [name for name in required_checks if index.get(name) != "success"] if missing: blocks.append("ci_required_check_missing") elif not isinstance(required_checks, list): blocks.append("ci_evidence_missing") ci = {"status": ci_status, "checks": ci_checks, "required_checks": required_checks, "all_green": False} # 末尾按 CI 分项 blocks 重算，见下。 # ② 目录隔离（复用 #209，不另造口径）。 directory = None if files_changed is None: blocks.append("directory_evidence_missing") else: directory = check_directory_isolation( files_changed, owned_dir, approved_exceptions) if directory["inconclusive"]: blocks.append("directory_owned_dir_unknown") elif not directory["passed"]: blocks.append("directory_isolation_violation") # ③ 合同版本。 if contract_version is None or ( isinstance(contract_version, str) and not contract_version.strip()): blocks.append("contract_version_missing") if contract_frozen is not True: blocks.append("contract_not_frozen") if contract_expected_version is not None \ and contract_version is not None \ and contract_version != contract_expected_version: blocks.append("contract_version_mismatch") contract = {"version": contract_version, "frozen": contract_frozen, "expected": contract_expected_version} # ④ 协调员批准。 if coordinator_approved is None: blocks.append("coordinator_approval_missing") elif coordinator_approved is not True: blocks.append("coordinator_not_approved") coordinator = {"approved": coordinator_approved} # ⑤ 无未解决评审意见。 if unresolved_review_comments is None: blocks.append("review_evidence_missing") reviews = {"unresolved": None} elif isinstance(unresolved_review_comments, bool): blocks.append("review_evidence_missing") reviews = {"unresolved": None} elif isinstance(unresolved_review_comments, list): reviews = {"unresolved": unresolved_review_comments} if unresolved_review_comments: blocks.append("unresolved_review_comments") elif isinstance(unresolved_review_comments, int): reviews = {"unresolved": unresolved_review_comments} if unresolved_review_comments > 0: blocks.append("unresolved_review_comments") elif unresolved_review_comments < 0: blocks.append("review_evidence_missing") else: blocks.append("review_evidence_missing") reviews = {"unresolved": None} # 分支保护：硬模式下 unknown 与 unprotected 一律阻断，确认无效。 if protection == "unprotected": risks.append(RISK_UNPROTECTED) blocks.append("branch_protection_unprotected") elif protection == "unknown": risks.append(RISK_UNKNOWN) blocks.append("branch_protection_unknown") # ci.all_green 只反映 CI 分项本身，不被其它项的 blocks 污染。 ci_block_codes = ("ci_evidence_missing", "ci_not_green", "ci_check_failed", "ci_required_checks_unconfigured", "ci_required_check_missing") ci["all_green"] = not any(code in blocks for code in ci_block_codes) return {"allowed": not blocks, "mode": "hard", "protection": protection, "directory": directory, "ci": ci, "contract": contract, "coordinator": coordinator, "reviews": reviews, "risks": risks, "blocks": blocks} def block_descriptions(blocks): """blocks 码 -> 中文说明（供第 3 轨推回包逐条引用）。""" return [_BLOCK_DESCRIPTIONS.get(code, code) for code in blocks or []] def blocked_audit_detail(task_id, gate): """merge_gate_blocked 审计的 detail：只含任务 id、模式与 blocks 码。 推回包不要解析这段文本，取 409 响应的 merge_gate 对象为准。 """ return "%s mode=%s blocks=%s" % ( task_id, gate.get("mode", "hard"), ",".join(gate.get("blocks") or []))