codex-app-server/cloud/merge_gate.py

"""合并门禁：主分支保护 + 目录隔离（纯函数，无网络、无 GitHub 调用）。

背景与诚实边界（先读，再用）：
- 本模块**不能**替仓库开通主分支保护。保护是 GitHub 仓库设置，
  只能由仓库管理员在 GitHub 侧开通；GitHub App id 5476837 的权限
  也不会自动变化，新权限需安装方重新接受。本模块只能在调用方
  明确告诉它保护状态时做门禁，绝不声称“已保护”。
- 保护状态的唯一来源是调用方传入的值或环境变量
  CLOUD_BRANCH_PROTECTION（protected / unprotected，未设即
  unknown）。unknown 不是“通过”，是“未知”：门禁把它记为风险，
  但不因此阻断——因为本服务没有可信渠道去 GitHub 核实，拿未知
  当未保护去硬堵，会把当前所有合并（当前仓库本身未保护）全停掉。
  想把门禁武装起来，运维要显式设置该环境变量为 unprotected，
  再由调用方在每次合并时传确认，或把模式切到 enforce。
- 目录隔离复用 v0.3 技能的检查口径（见
  cloud/skills/software-engineering/SKILL.md“目录与分支规则”）：
  人 / 宿主跑 ``git diff --name-only $(git merge-base origin/main
  HEAD)...HEAD``，逐项核对每个文件是否落在拥有目录内，契约变更
  经协调员批准的文件除外。本模块不另造一套口径：
  ``check_directory_isolation`` 就是这条规则的可执行版，
  ``directory_diff_command`` 返回的正是 v0.3 文本里的命令。
  v0.3 之前服务端合并端（POST /tasks/{id}/merge）只改任务状态，
  不做 git 合并、不做目录检查、不做保护检查——那是本模块要补的
  缺口，不是重复造检查。

模式（CLOUD_MERGE_GATE_MODE，默认 record）：
- record（默认）：未保护时必须显式确认（explicit_confirm）才放行；
  无确认时阻断，错误码 branch_protection_risk_unconfirmed。
  确认只覆盖“保护风险”，绝不覆盖目录隔离违规——越目录的写
  只能靠 approved_exceptions（契约变更批准）放行，不能靠确认蒙混。
- enforce：未保护时一律阻断，确认也不放行。切到 enforce 之前
  必须先由用户决定并开通保护：升级 Pro、公开仓库、或保持现状
  纪律 + 记风险——三者都是用户侧决定，代码里只留开关，不替选。
- off：门禁显式关闭，只记风险不阻断。off 必须显式设置，
  非法模式值不会悄悄落到 off，而是回落到 record（fail-safe）。
"""

import os

MODES = ("record", "enforce", "off")
PROTECTION_STATES = ("protected", "unprotected", "unknown")

RISK_UNPROTECTED = "branch protection is not enabled for this repo"
RISK_UNKNOWN = "branch protection status is unknown (not declared)"
RISK_GATE_OFF = "merge gate is disabled (mode=off)"


def gate_mode_from_env(env=None):
    """环境变量里的模式；未设或非法时回落 record，绝不回落 off。"""
    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 MODES else "record"


def protection_from_env(env=None):
    """环境变量里的保护状态；未设或非法时为 unknown，不是 protected。"""
    raw = (env if env is not None else os.environ).get(
        "CLOUD_BRANCH_PROTECTION", "")
    state = (raw or "").strip().lower()
    return state if state in PROTECTION_STATES else "unknown"


def directory_diff_command(base="origin/main", head="HEAD"):
    """v0.3 技能文本里的目录检查命令（原样口径，不另造）。"""
    return "git diff --name-only $(git merge-base %s %s)...%s" % (
        base, head, head)


def _normalize_path(path):
    return (path or "").strip().replace("\\", "/")


def _is_escape(path):
    parts = [p for p in _normalize_path(path).split("/") if p not in ("", ".")]
    return (not parts) or parts[0] == ".." or ".." in parts


def check_directory_isolation(files_changed, owned_dir,
                              approved_exceptions=()):
    """v0.3 目录隔离规则的可执行版。

    files_changed：diff 输出的文件清单（相对仓库根）。
    owned_dir：本线程唯一拥有目录，例如 apps/web；为 None/空时
      无法核对，返回 inconclusive（不算通过，也不伪装成通过）。
    approved_exceptions：经协调员批准的契约变更文件（精确路径），
      命中者不算违规，但会记入 exceptions_used 以便合并记录核对。
    返回 dict：passed / violations / exceptions_used / inconclusive。
    """
    owned = _normalize_path(owned_dir)
    exceptions = {_normalize_path(p) for p in approved_exceptions or ()}
    if not owned:
        return {"passed": False, "violations": [],
                "exceptions_used": [], "inconclusive": True,
                "reason": "owned_dir_unknown"}
    prefix = owned.rstrip("/") + "/"
    violations = []
    used = []
    for raw in files_changed or []:
        path = _normalize_path(raw)
        if not path:
            continue
        if path in exceptions:
            used.append(path)
            continue
        if path.startswith("/") or _is_escape(path) \
                or not path.startswith(prefix):
            violations.append(path)
    return {"passed": not violations, "violations": violations,
            "exceptions_used": used, "inconclusive": False,
            "reason": ""}


def evaluate_merge_gate(files_changed=None, owned_dir=None,
                        approved_exceptions=(), protection="unknown",
                        mode="record", explicit_confirm=False):
    """合并门禁纯函数。

    files_changed=None 表示调用方没给 diff 证据：目录检查跳过
    （directory=None），不因“没证据”就判违规，也不伪装成已检查。
    protection="unknown" 同理：记风险、不阻断。
    返回 dict：allowed / mode / protection / directory / risks / blocks。
    blocks 为空即 allowed；每个 block 都是可查的错误码。
    """
    mode = mode if mode in MODES else "record"
    protection = protection if protection in PROTECTION_STATES \
        else "unknown"
    risks = []
    blocks = []
    directory = None
    if files_changed is not None:
        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 protection == "unprotected":
        risks.append(RISK_UNPROTECTED)
        if mode == "enforce":
            blocks.append("branch_protection_enforce")
        elif mode == "record" and not explicit_confirm:
            blocks.append("branch_protection_risk_unconfirmed")
    elif protection == "unknown":
        risks.append(RISK_UNKNOWN)
    if mode == "off" and protection == "unprotected":
        risks.append(RISK_GATE_OFF)
    return {"allowed": not blocks, "mode": mode,
            "protection": protection, "directory": directory,
            "risks": risks, "blocks": blocks}


def gate_from_request(body, env=None):
    """从 /merge 请求体 + 环境变量拼出门禁输入并求值。

    请求体字段（全是可选的；一个都不给时行为与旧接口一致，
    只在响应里附带门禁结果，不阻断）：
    files_changed（str 列表）、owned_dir、approved_exceptions、
    branch_protection、merge_gate_confirm（bool）。
    模式优先取环境变量（运维定），环境变量未设时才取请求体
    merge_gate_mode——请求体不能把运维定的 enforce 降级为 record。
    """
    body = body or {}
    env_map = env if env is not None else os.environ
    mode = gate_mode_from_env(env_map) \
        if env_map.get("CLOUD_MERGE_GATE_MODE") \
        else (body.get("merge_gate_mode") or "record")
    protection = body.get("branch_protection") \
        or protection_from_env(env_map)
    return evaluate_merge_gate(
        files_changed=body.get("files_changed"),
        owned_dir=body.get("owned_dir"),
        approved_exceptions=body.get("approved_exceptions") or (),
        protection=protection, mode=mode,
        explicit_confirm=bool(body.get("merge_gate_confirm")))