codex-app-server/cloud/repos.py

"""项目级仓库解析：把项目的 repo_url 变成 worker 可派发的本地检出。

零第三方依赖，只用 Python 标准库；不改 host.py 的 worktree 逻辑，
不存任何仓库凭证、不推远端。

- validate_repo_url：API（POST /projects）与 worker 共用的同一套
  校验，绝不另写第二份校验逻辑让两者漂移。只接受：
  https:// 开头（且不含 userinfo，即不含用户名/密码），或
  git@ 开头（git@host:path 形）。file://、ssh://、本地路径、
  带密码的 URL 一律拒绝。
- checkout_dir：按 project_id 的安全字符命名，防路径穿越。
- resolve_repo：目录不存在时 git clone；已存在时校验它是 git
  仓库且 remote origin 与 repo_url 一致（不一致时报错，绝不强制
  改 remote），一致时 git fetch origin（失败明确报错，不静默用
  旧代码）；fetch 成功后不自动 checkout/pull、不 git reset/clean，
  工作树脏不脏由 host.py 的 worktree 逻辑处理。
- 私有仓库 clone 失败时错误文本写明：需要仓库访问凭证，控制面
  不存仓库凭证——绝不尝试用任何 key 去 clone。
"""

import os
import re
import subprocess
from urllib.parse import urlparse

__all__ = [
    "RepoError",
    "validate_repo_url",
    "checkout_dir",
    "resolve_repo",
]


class RepoError(ValueError):
    """仓库解析失败（校验、clone、remote 不一致、fetch 失败）。"""


_GIT_SCP_RE = re.compile(r"^git@[^/:@\s]+:[^@\s]+$")
_SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9_.-]")


def validate_repo_url(repo_url):
    """校验 repo_url，合法时返回原值，否则抛 ValueError。

    拒绝：非字符串、空串、含空白/控制字符、非 https:// 与 git@ 开头、
    https:// 带 userinfo（username 或 password 任一存在即拒绝，
    含 https://user@host/ 这种只有用户名无密码的也拒绝，因为
    userinfo 本身就是凭证载体）、https:// 无 host、git@ 形不合规。
    """
    if not isinstance(repo_url, str):
        raise ValueError("repo_url 必须是非空字符串")
    # 含空白/控制字符的 URL 直接拒绝，不静默 strip：strip 会把
    # “https://…\n”这种畸形输入洗成合法 URL 存进库，git 调用时
    # 再出难以定位的问题；调用方应传干净的 URL。
    if not repo_url or any(ch.isspace() for ch in repo_url):
        raise ValueError("repo_url 必须是非空字符串且不得含空白字符")
    url = repo_url
    if url.startswith("https://"):
        parsed = urlparse(url)
        if parsed.username is not None or parsed.password is not None:
            raise ValueError(
                "repo_url 不得含 userinfo（用户名/密码）: %s" % url)
        if not parsed.hostname:
            raise ValueError("repo_url 缺少 host: %s" % url)
        return url
    if url.startswith("git@"):
        if not _GIT_SCP_RE.match(url):
            raise ValueError("git@ 形 repo_url 不合法: %s" % url)
        return url
    raise ValueError(
        "repo_url 只接受 https:// 或 git@ 开头，"
        "不接受 file://、ssh://、本地路径: %s" % url)


def checkout_dir(repos_dir, project_id):
    """按 project_id 的安全字符命名检出目录，防路径穿越。

    project_id 本身是 hex，仍做安全字符过滤；过滤后为空时报错。
    返回的绝对路径一定在 repos_dir 之下，否则报错（防御性双检）。
    """
    raw = str(project_id)
    safe = _SAFE_NAME_RE.sub("-", raw).strip(".")
    if not safe or safe in (".", ".."):
        raise RepoError("project_id 无法映射为安全的检出目录名: %r" % raw)
    base = os.path.abspath(repos_dir)
    path = os.path.join(base, safe)
    if os.path.commonpath([base, path]) != base:
        raise RepoError("检出目录越出 repos_dir，拒绝: %s" % path)
    return path


def _default_runner(cmd):
    return subprocess.run(cmd, capture_output=True, text=True)


def _run(runner, cmd):
    """调 runner：优先只传 cmd，假 runner 只接受 cmd 时即走此路；
    接受 (cmd, timeout) 的 runner 也兼容（TypeError 时回退）。"""
    try:
        return runner(cmd)
    except TypeError:
        return runner(cmd, timeout=None)


def resolve_repo(repo_url, repos_dir, project_id, runner=None):
    """把 repo_url 解析为本地检出路径，返回该路径。

    目录不存在：git clone repo_url 到该目录；clone 失败抛 RepoError，
    文本含“需要仓库访问凭证，控制面不存仓库凭证”的提示（私有仓
    库、仓库不存在、无权访问都走同一文案，不猜测、不尝试凭证）。
    目录已存在：须是 git 仓库（.git 目录或文件存在）且 remote
    origin 的 URL 与 repo_url 一致；不是 git 仓库、取不到 origin、
    origin 不一致时都抛 RepoError，绝不强制改 remote、不覆盖。
    一致时 git fetch origin：失败抛 RepoError，不静默用旧代码；
    成功后不 checkout/pull/reset/clean，工作树由 host.py 处理。
    """
    url = validate_repo_url(repo_url)
    run = runner if runner is not None else _default_runner
    target = checkout_dir(repos_dir, project_id)
    os.makedirs(repos_dir, exist_ok=True)
    if not os.path.exists(target):
        result = _run(run, ["git", "clone", url, target])
        if getattr(result, "returncode", 1) != 0:
            err = (getattr(result, "stderr", "") or "").strip()
            raise RepoError(
                "git clone 失败: %s；若为私有仓库，需要仓库访问凭证，"
                "控制面不存仓库凭证，请先在可访问该仓库的机器上完成"
                "检出或改用公开仓库" % (err or url))
        return target
    git_path = os.path.join(target, ".git")
    if not (os.path.isdir(git_path) or os.path.isfile(git_path)):
        raise RepoError(
            "检出目录已存在但不是 git 仓库，拒绝覆盖: %s" % target)
    result = _run(run, ["git", "-C", target, "remote", "get-url", "origin"])
    if getattr(result, "returncode", 1) != 0:
        raise RepoError(
            "检出目录 %s 取不到 remote origin，拒绝派发" % target)
    origin = (getattr(result, "stdout", "") or "").strip()
    if origin != url:
        raise RepoError(
            "检出目录 %s 的 remote origin（%s）与项目 repo_url（%s）"
            "不一致，拒绝派发（绝不强制改 remote）"
            % (target, origin, url))
    fetched = _run(run, ["git", "-C", target, "fetch", "origin"])
    if getattr(fetched, "returncode", 1) != 0:
        err = (getattr(fetched, "stderr", "") or "").strip()
        raise RepoError(
            "git fetch origin 失败，拒绝用旧代码派发: %s"
            % (err or target))
    return target