diff --git a/cloud/README.md b/cloud/README.md index 04d5b5c..e8bb0d0 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -172,7 +172,7 @@ client_secret（只走环境变量），不是无认证后门。 没有返回 404）后恢复原来的 oauth 账号登录。查看自己的绑定： `GET /oauth/{provider}/bindings`（须持本站 key，只看得到自己的）。 -### GitHub App 一键安装（已实现安装一半；App-token API 动作未做） +### GitHub App 一键安装与 App-token（安装 + token 均已实现） 目标：把手动建 webhook（填 URL、secret、选事件）变成“点一下 安装”。流程用 GitHub manifest 流程（GitHub 文档标准流程）： @@ -207,9 +207,18 @@ client_secret（只走环境变量），不是无认证后门。 /fix-review 也有内容去重（repo + PR + 评论 id， `github_fix_review_dedup` 表），只建一次任务。 -诚实边界：pem 虽然存下来了，但本阶段不用它签 JWT、不用它换 -installation token；推分支、建 PR 仍走 gh + PAT（见“云端派发与 -自动 PR”），App-token API 动作是下一阶段。 +诚实边界（已缩小）：pem 现在用于签 JWT、换 installation +token：`cloud/github_app.py` 的 `mint_app_jwt` 是纯标准库 +RS256（解析 PKCS#1 / PKCS#8 私钥，不依赖 openssl、不落临时 +文件），`exchange_installation_token` 换 1 小时有效的 token +（现换现用、不缓存、不存库），`resolve_app_token` 按仓库名 +查 `github_app_installations` 找 installation id，任一步缺失 +或失败返回 None（调用方回落 PAT，不报错）。推分支、建 PR +走 App 还是 PAT 见“云端派发与自动 PR”；host.py 本地操作 +（pr / checks / merge）仍是操作员自己的 gh + PAT，不走 +App-token。App 安装事件的现场记录路径仍待下一次真实 +安装/增减仓库动作实弹验证（此前一次是数据修复，不是 +webhook 实测）。 ### 推送凭证清单 @@ -830,8 +839,17 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 才先把 worktree 改动提交到分支上（worktree 必须正好在 host/ 上，否则直接失败；nothing to commit 时跳过； 缺 git 身份时回退 cloud/cloud@local 再试），再 push 分支 + - gh pr create，成功后把 pr_url/pr_number 写回任务 - 记录并记审计 pr_created，返回 200 含 pr_url。push 或 pr create + 建 PR，成功后把 pr_url/pr_number 写回任务 + 记录并记审计 pr_created，返回 200 含 pr_url。App-token + 路径：项目 repo_url 能解析出 github.com 仓库名且库里有该 + 仓库的 App 安装记录时，端点现换 installation token 走 App + 路径（响应里 auth_via=app）：push 用 git credential helper， + token 只进子进程环境变量 GIT_TOKEN（不进命令行参数、不进 + 文件，helper 命令里只有变量名），建 PR 走 GitHub REST API + （先读仓库 default_branch 当 base，不猜 main）；没有安装 + 记录或换 token 失败时自动回落原来的 gh + PAT 路径 + （auth_via=pat），不因此报错；App 没装的仓库与 gh 路径 + 的行为与原来完全一致。push 或 pr create 失败时返回 502 且错误文本写清是哪一步（commit、push 还是 pr_create）， 任务状态保持 merged 不变，绝不把任务标 failed。顺序：任务 @@ -846,7 +864,8 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 的路由。gh 的登录凭证由 gh 自己管，控制面不存、不打任何 token。 - 仍未做（不要宣称已做）：厂商专属适配（ECS、Cloud Run 等，需 用户选厂商后另做）；默认自动推（本轮明确不做）；主机上 gh 的安装 - 与登录、远端写凭证（控制面不存，自动 PR 的真实成功路径仍需它们）； + 与登录、远端写凭证（控制面不存；PAT 路径的自动 PR 真实成功 + 仍需它们，App-token 路径不需要 gh）； 多 worker 的并发容量 enforcement（capacity 目前只是注册表展示， run_loop 仍串行）；worker 经 HTTP 发心跳的凭证供给（worker 直接 写库时不需要，外部 worker 走 HTTP 时需自带 API key）。 diff --git a/cloud/github_app.py b/cloud/github_app.py index c3b1c56..085c4a2 100644 --- a/cloud/github_app.py +++ b/cloud/github_app.py @@ -30,6 +30,7 @@ env secret，App webhook 用库里 secret，验签时两个都试（过渡 注入 fake http 客户端做单测。 """ +import hashlib import json import secrets import time @@ -410,3 +411,224 @@ def set_fix_review_task_id(conn, repo, pr_number, comment_id, task_id): (task_id, repo, pr_number, str(comment_id)), ) conn.commit() + + +# -- App-token 阶段：JWT 与 installation token --------------------------- +# 纯标准库 RS256：解析 PEM（PKCS#1 / PKCS#8）取 (n, e, d)， +# PKCS#1 v1.5 + SHA-256 签名。不依赖 openssl CLI、不落临时文件； +# JWT 与 installation token 只在进程内出现，绝不打印、不进日志。 +# GitHub App 的 pem 格式：manifest 换回的通常是 PKCS#1 +#（"BEGIN RSA PRIVATE KEY"），openssl 3 默认生成的是 PKCS#8 +#（"BEGIN PRIVATE KEY"），两者都支持。 + +import base64 as _base64 + +_SHA256_DIGESTINFO_PREFIX = bytes.fromhex( + "3031300d060960864801650304020105000420") + + +def _b64url(data): + if isinstance(data, str): + data = data.encode("utf-8") + return _base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _der_read(data, pos): + """读一个 DER TLV：返回 (tag, value, next_pos)。""" + tag = data[pos] + pos += 1 + length = data[pos] + pos += 1 + if length & 0x80: + nbytes = length & 0x7F + if nbytes == 0 or nbytes > 4: + raise AppError("unsupported DER length", + reason="bad_key") + length = int.from_bytes(data[pos:pos + nbytes], "big") + pos += nbytes + return tag, data[pos:pos + length], pos + length + + +def _der_sequence_items(der): + """SEQUENCE 的子项列表（每项为 (tag, value)）。""" + tag, value, _ = _der_read(der, 0) + if tag != 0x30: + raise AppError("expected DER SEQUENCE", reason="bad_key") + items = [] + pos = 0 + while pos < len(value): + t, v, pos = _der_read(value, pos) + items.append((t, v)) + return items + + +def _parse_pkcs1_der(der): + """PKCS#1 RSAPrivateKey DER -> (n, e, d)。""" + items = _der_sequence_items(der) + if len(items) < 4: + raise AppError("short PKCS#1 key", reason="bad_key") + n = int.from_bytes(items[1][1], "big") + e = int.from_bytes(items[2][1], "big") + d = int.from_bytes(items[3][1], "big") + if n <= 0 or e <= 0 or d <= 0: + raise AppError("invalid RSA parameters", reason="bad_key") + return n, e, d + + +def _parse_rsa_private_key(pem): + """解析 PEM 私钥 -> (n, e, d)，PKCS#1 与 PKCS#8 都支持。""" + if not pem or not isinstance(pem, str): + raise AppError("private key required", reason="bad_key") + body = "".join( + line.strip() for line in pem.splitlines() + if line.strip() and not line.strip().startswith("-----")) + try: + der = _base64.b64decode(body) + except Exception as exc: + raise AppError("private key is not valid base64: %s" % (exc,), + reason="bad_key") from exc + if "RSA PRIVATE KEY" in pem: + return _parse_pkcs1_der(der) + if "PRIVATE KEY" in pem: + items = _der_sequence_items(der) + if len(items) < 3 or items[2][0] != 0x04: + raise AppError("unexpected PKCS#8 shape", reason="bad_key") + return _parse_pkcs1_der(items[2][1]) + raise AppError("unsupported private key format", reason="bad_key") + + +def rsa_sign_sha256(pem, message): + """RS256 签名：返回签名 bytes（长度 = 模长字节数）。""" + n, _e, d = _parse_rsa_private_key(pem) + if isinstance(message, str): + message = message.encode("utf-8") + digest = hashlib.sha256(message).digest() + t = _SHA256_DIGESTINFO_PREFIX + digest + em_len = (n.bit_length() + 7) // 8 + if em_len < len(t) + 11: + raise AppError("RSA key too small for SHA-256", reason="bad_key") + em = b"\x00\x01" + b"\xff" * (em_len - len(t) - 3) + b"\x00" + t + m = int.from_bytes(em, "big") + s = pow(m, d, n) + return s.to_bytes(em_len, "big") + + +def mint_app_jwt(app_id, pem, now=None, ttl_seconds=540): + """签发 App JWT：header alg=RS256，有效期默认 9 分钟 + （GitHub 上限 10 分钟，iat 回拨 60 秒防时钟 skew）。 + + now 只为单测注入时间。pem 非法时抛 AppError（调用方据此 + 回落 PAT，不静默用坏 token 去撞 GitHub）。 + """ + if app_id is None or str(app_id) == "": + raise AppError("app_id required", reason="bad_key") + if now is None: + now = time.time() + header = _b64url(json.dumps({"alg": "RS256", "typ": "JWT"}, + separators=(",", ":"))) + payload = _b64url(json.dumps( + {"iat": int(now) - 60, "exp": int(now) + int(ttl_seconds), + "iss": str(app_id)}, + separators=(",", ":"))) + signing_input = header + "." + payload + signature = rsa_sign_sha256(pem, signing_input.encode("ascii")) + return signing_input + "." + _b64url(signature) + + +def _github_api_request(method, url, token, payload, http_client): + """调 GitHub API：返回 (status, dict)。 + + http_client 注入约定（单测用）：``http_client(method, url, + headers, body_bytes_or_None) -> (status, dict)``。真实路径 + 用 urllib，HTTPError 也解析 body（GitHub 的错误体是 JSON， + 调用方按 status 判断，不靠异常）。 + """ + headers = {"Accept": "application/vnd.github+json", + "User-Agent": "codex-app-server"} + if token: + headers["Authorization"] = "Bearer " + token + body = None + if payload is not None: + body = json.dumps(payload).encode("utf-8") + if http_client is not None: + result = http_client(method, url, headers, body) + if not (isinstance(result, tuple) and len(result) == 2): + raise AppError("http_client must return (status, dict)", + reason="token_exchange_failed") + return result + request = urllib.request.Request(url, data=body, method=method, + headers=headers) + try: + with urllib.request.urlopen(request, timeout=30) as response: + raw = response.read().decode("utf-8", "replace") + status = response.status + except urllib.error.HTTPError as e: + raw = e.read().decode("utf-8", "replace") + status = e.code + try: + parsed = json.loads(raw) + except ValueError: + parsed = {} + if not isinstance(parsed, dict): + parsed = {} + return status, parsed + + +def exchange_installation_token(app_id, pem, installation_id, + http_client=None, now=None): + """换 installation token：返回 {token, expires_at, permissions, + repository_selection}。token 有效期 1 小时，调用方现换现用， + 不缓存、不存库。换失败抛 AppError（调用方回落 PAT）。""" + if not installation_id: + raise AppError("installation_id required", + reason="token_exchange_failed") + jwt = mint_app_jwt(app_id, pem, now=now) + url = ("https://api.github.com/app/installations/%s/access_tokens" + % installation_id) + status, data = _github_api_request("POST", url, jwt, None, http_client) + if status != 201 or not data.get("token"): + raise AppError( + "installation token exchange failed: %s %r" + % (status, data.get("message")), + reason="token_exchange_failed") + return {"token": data["token"], "expires_at": data.get("expires_at"), + "permissions": data.get("permissions"), + "repository_selection": data.get("repository_selection")} + + +def find_installation_for_repo(conn, repo_full_name): + """查仓库对应的 App installation id：无记录返回 None + （调用方回落 PAT，不报错——没装 App 的仓库走老路）。""" + if not repo_full_name: + return None + _ensure_installations_table(conn) + row = conn.execute( + "SELECT installation_id FROM github_app_installations" + " WHERE repo_full_name = ? ORDER BY created_at LIMIT 1", + (repo_full_name,), + ).fetchone() + return row[0] if row else None + + +def resolve_app_token(conn, repo_full_name, http_client=None): + """给仓库解析 App installation token：成功返回 token 字符串， + 任一步缺失或失败返回 None（调用方回落 PAT，不抛错）。 + + 步骤：App 配置（库里）-> 安装记录（仓库名查 installation + id）-> JWT -> 换 token。token 与 JWT 只在进程内出现。 + """ + try: + config = get_app_config(conn) + if config is None or not config.get("pem"): + return None + installation_id = find_installation_for_repo(conn, repo_full_name) + if installation_id is None: + return None + result = exchange_installation_token( + config.get("app_id"), config.get("pem"), installation_id, + http_client=http_client) + return result["token"] + except AppError: + return None + except Exception: + return None diff --git a/cloud/pr.py b/cloud/pr.py index 8b229ed..908415a 100644 --- a/cloud/pr.py +++ b/cloud/pr.py @@ -38,6 +38,15 @@ host/，worktree 在 <仓库根>/work/host-worktrees/ （gh auth login），push/pr 的命令与错误文本里不出现任何 token；错误文本只含 git/gh 的原文摘要，不含远端 URL 里的 userinfo（若 remote URL 含 userinfo，摘要里把它抹掉）。 + + App-token 路径（cloud/github_app.py 的 resolve_app_token 供 + token）：调用方传 app_token 时，跳过 gh 二进制与 gh 登录 + 检查（App 路径不需要 gh）；push 走 git credential helper， + token 只放进子进程环境变量 GIT_TOKEN，不进命令行参数、不 + 进文件，helper 命令里只有变量名；建 PR 走 GitHub REST API + （先读仓库 default_branch 再建，base 不猜 main）。没传 + app_token 时走原来的 gh 路径，两条路不互相干扰；App 没装 + 在该仓库时调用方不传 token、自动回落 PAT。 """ import os @@ -96,12 +105,69 @@ def _scrub_url(text): return re.sub(r"(https?://)[^/@\s]+@", r"\1", text) -def _run(cmd, cwd=None, runner=None): +def _run(cmd, cwd=None, runner=None, env=None): """命令唯一入口：runner 供单测注入，返回带 returncode/stdout/ - stderr 的对象；真实路径走 subprocess。""" + stderr 的对象；真实路径走 subprocess。env 只在真实路径生效 + （与进程环境合并），注入的 runner 不收 env——token 绝不进 + 假 runner 记录的命令，也不进真实路径的命令行参数。""" if runner is not None: return runner(cmd, cwd) - return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True) + full_env = None + if env: + full_env = dict(os.environ) + full_env.update(env) + return subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, + env=full_env) + + +_GITHUB_REMOTE_RE = re.compile( + r"^(?:https?://(?:[^/@\s]+@)?github\.com/" + r"|git@github\.com:|ssh://git@github\.com/)" + r"([^/\s]+)/([^/\s]+?)(?:\.git)?/?$") + + +def parse_git_remote_owner_repo(url): + """解析 git remote URL -> (owner, repo)，非 github.com 仓库 + 或解析失败返回 None（调用方据此回落 PAT，不报错）。""" + if not url or not isinstance(url, str): + return None + m = _GITHUB_REMOTE_RE.match(url.strip()) + if not m: + return None + return m.group(1), m.group(2) + + +def _default_github_api(method, url, token, payload): + """默认 GitHub API 客户端：复用 github_app 的 urllib 实现 + （返回 (status, dict)；懒 import，避免模块导入环）。""" + from cloud.github_app import _github_api_request + return _github_api_request(method, url, token, payload, None) + + +def _create_pr_via_api(owner, repo, branch, title, body, token, + github_api): + """用 installation token 走 REST API 建 PR：先读仓库 + default_branch 当 base（不猜 main），再 POST /pulls。 + 成功返回 {pr_url, pr_number, branch}，失败抛 PrStepError。""" + api = github_api or _default_github_api + status, repo_info = api( + "GET", "https://api.github.com/repos/%s/%s" % (owner, repo), + token, None) + if status != 200: + raise PrStepError( + "pr_create", + "GitHub API 读仓库失败：%s" % (repo_info.get("message"),)) + base = repo_info.get("default_branch") or "main" + status, data = api( + "POST", "https://api.github.com/repos/%s/%s/pulls" % (owner, repo), + token, {"title": title, "head": branch, "base": base, + "body": body}) + if status not in (200, 201) or not data.get("html_url"): + raise PrStepError( + "pr_create", + "GitHub API 建 PR 失败：%s" % (data.get("message"),)) + return {"pr_url": data["html_url"], "pr_number": data.get("number"), + "branch": branch} def _git_remote_url(worktree_dir, git_runner=None): @@ -114,15 +180,18 @@ def _git_remote_url(worktree_dir, git_runner=None): def check_pr_ready(worktree_dir, git_runner=None, gh_which=None, - gh_runner=None): + gh_runner=None, app_token=None): """按顺序做跳过检查，全过返回 (worktree_dir, branch)；任一步 不过抛 PrSkipped。branch 由调用方按 host 任务名算好传进来 的是 worktree 检查——本函数只查 worktree 与 remote/gh/登录， - 分支名由 create_task_pr 组装。""" + 分支名由 create_task_pr 组装。传了 app_token 时跳过 gh + 二进制与 gh 登录检查（App 路径不需要 gh）。""" if not os.path.isdir(worktree_dir): raise PrSkipped(_SKIP_NO_WORKTREE) if _git_remote_url(worktree_dir, git_runner) is None: raise PrSkipped(_SKIP_NO_REMOTE) + if app_token: + return True gh_bin = gh_which if gh_which is not None else shutil.which("gh") if not gh_bin: raise PrSkipped(_SKIP_NO_GH) @@ -195,7 +264,8 @@ def _commit_worktree_changes(worktree_dir, branch, task_id, def create_task_pr(task_id, auto_pr, worktrees_root=None, - git_runner=None, gh_runner=None, gh_which=None): + git_runner=None, gh_runner=None, gh_which=None, + app_token=None, github_api=None): """给一个已 merged 的云端任务建 PR。 auto_pr 不为 1 时直接抛 PrSkipped(auto_pr_disabled)，调用方 @@ -203,6 +273,9 @@ def create_task_pr(task_id, auto_pr, worktrees_root=None, 防止直接调用者漏查。成功返回 {pr_url, pr_number, branch}。 全部跳过检查通过后，先把 worktree 改动提交到分支再 push， 否则未提交的改动进不了 PR。 + 传了 app_token 时走 App 路径：push 用 credential helper + （token 只进子进程环境，不进参数），建 PR 走 GitHub API； + 不传时走原来的 gh 路径。 """ if auto_pr != 1: raise PrSkipped(_SKIP_AUTO_PR) @@ -211,9 +284,35 @@ def create_task_pr(task_id, auto_pr, worktrees_root=None, resolve_worktrees_root(worktrees_root), host_name) branch = "host/" + host_name check_pr_ready(worktree_dir, git_runner=git_runner, - gh_which=gh_which, gh_runner=gh_runner) + gh_which=gh_which, gh_runner=gh_runner, + app_token=app_token) _commit_worktree_changes(worktree_dir, branch, task_id, git_runner=git_runner) + title = "cloud: %s" % task_id + body = "cloud 任务 %s 的自动 PR。" % task_id + if app_token: + remote_url = _git_remote_url(worktree_dir, git_runner) + owner_repo = parse_git_remote_owner_repo(remote_url) + if owner_repo is None: + raise PrStepError( + "push", + "worktree remote 不是 github.com 仓库，App token 无法" + " push：%s" % _scrub_url(remote_url)) + push = _run( + ["git", "-C", worktree_dir, "-c", + "credential.helper=!f() { echo username=x-access-token;" + " echo password=$GIT_TOKEN; }; f", + "push", "origin", branch], + cwd=worktree_dir, runner=git_runner, + env={"GIT_TOKEN": app_token}) + if getattr(push, "returncode", 1) != 0: + raise PrStepError( + "push", "git push origin %s 失败：%s" % ( + branch, _scrub_url( + (getattr(push, "stderr", "") or "") + or (getattr(push, "stdout", "") or "")))) + return _create_pr_via_api(owner_repo[0], owner_repo[1], branch, + title, body, app_token, github_api) push = _run(["git", "-C", worktree_dir, "push", "origin", branch], cwd=worktree_dir, runner=git_runner) if getattr(push, "returncode", 1) != 0: @@ -222,9 +321,8 @@ def create_task_pr(task_id, auto_pr, worktrees_root=None, branch, _scrub_url( (getattr(push, "stderr", "") or "") or (getattr(push, "stdout", "") or "")))) - title = "cloud: %s" % task_id r = _run(["gh", "pr", "create", "--title", title, - "--body", "cloud 任务 %s 的自动 PR。" % task_id, + "--body", body, "--head", branch], cwd=worktree_dir, runner=gh_runner) out = (getattr(r, "stdout", "") or "").strip() diff --git a/cloud/server.py b/cloud/server.py index d7e2bea..e58f5a8 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -59,7 +59,8 @@ from cloud.db import PostgresIntegrityError as _PgIntegrityError from cloud.db import PostgresOperationalError as _PgOperationalError from cloud.kb import KBStore from cloud.memory import MemoryStore -from cloud.pr import PrSkipped, PrStepError, create_task_pr +from cloud.pr import (PrSkipped, PrStepError, create_task_pr, + parse_git_remote_owner_repo) from cloud.workers import ( heartbeat_worker as _registry_heartbeat, list_workers as _registry_list, @@ -100,6 +101,7 @@ from cloud.github_app import ( record_installation_repos, remove_installation, remove_installation_repos, + resolve_app_token, save_app_config, set_fix_review_task_id, ) @@ -1505,6 +1507,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): pr_skipped 与 reason，任务状态保持 merged 不变；push/pr 创建失败时返回 502 并写清哪一步失败，任务同样保持 merged。 成功时把 pr_url/pr_number 写回任务并记审计 pr_created。 + App-token：项目 repo_url 能解析出 github.com 仓库名且 + 库里有该仓库的 App 安装记录时，现换 installation token + 走 App 路径（响应里 auth_via=app），否则走原来的 gh + + PAT 路径（auth_via=pat）；换 token 任一步失败都回落 + PAT，不因此报错。token 只在本次请求内存里出现。 """ row = self._fetch_task(task_id) if row is None: @@ -1525,16 +1532,39 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): conn = self._db() try: prow = conn.execute( - "SELECT auto_pr FROM projects WHERE id = ?", + "SELECT auto_pr, repo_url FROM projects WHERE id = ?", (task["project_id"],)).fetchone() finally: conn.close() auto_pr = 0 + repo_url = None if prow is not None: try: auto_pr = prow["auto_pr"] or 0 except (KeyError, IndexError, TypeError): auto_pr = 0 + try: + repo_url = prow["repo_url"] + except (KeyError, IndexError, TypeError): + repo_url = None + app_token = None + auth_via = "pat" + repo_full_name = None + if repo_url: + parsed = parse_git_remote_owner_repo(repo_url) + if parsed: + repo_full_name = "%s/%s" % parsed + if repo_full_name: + conn = self._db() + try: + app_token = resolve_app_token( + conn, repo_full_name, + http_client=getattr( + self.server, "pr_app_token_client", None)) + finally: + conn.close() + if app_token: + auth_via = "app" try: result = create_task_pr( task_id, auto_pr, @@ -1542,7 +1572,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self.server, "host_worktrees_dir", None), git_runner=getattr(self.server, "pr_git_runner", None), gh_runner=getattr(self.server, "pr_gh_runner", None), - gh_which=getattr(self.server, "pr_gh_which", None)) + gh_which=getattr(self.server, "pr_gh_which", None), + app_token=app_token, + github_api=getattr(self.server, "pr_github_api", None)) except PrSkipped as exc: self._send_json(200, { "task_id": task_id, "status": task["status"], @@ -1575,7 +1607,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): result["pr_url"]) self._send_json(200, dict(task, pr_url=result["pr_url"], pr_number=result["pr_number"], - branch=result["branch"])) + branch=result["branch"], + auth_via=auth_via)) def _heartbeat_task(self, user_id, task_id): row = self._fetch_task(task_id) diff --git a/cloud/tests/test_app_token.py b/cloud/tests/test_app_token.py new file mode 100644 index 0000000..98b61b1 --- /dev/null +++ b/cloud/tests/test_app_token.py @@ -0,0 +1,244 @@ +"""App-token 单测：纯标准库 RS256 签名、JWT、installation token +换取、仓库安装查找、resolve 回落。 + +fixture 是测试专用 RSA key（openssl 现生成、仅供单测，从不用 +于生产；即使公开也无害——它不对应任何真实 GitHub App）。 +不联网：换 token 挂 fake http 客户端。 +""" + +import base64 +import hashlib +import json +import sys +import tempfile +import os +import unittest +from pathlib import Path + +try: + from cloud.db import connect as _db_connect + from cloud.github_app import ( + AppError, + _parse_rsa_private_key, + exchange_installation_token, + find_installation_for_repo, + mint_app_jwt, + record_installation_repos, + resolve_app_token, + rsa_sign_sha256, + save_app_config, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.db import connect as _db_connect + from cloud.github_app import ( + AppError, + _parse_rsa_private_key, + exchange_installation_token, + find_installation_for_repo, + mint_app_jwt, + record_installation_repos, + resolve_app_token, + rsa_sign_sha256, + save_app_config, + ) + +# 测试专用 RSA 私钥（PKCS#1 与 PKCS#8，同一密钥对的两版编码 +# 是两把不同 key——各测各的格式解析，不要求同模）。 +TEST_KEY_PKCS1 = """ +-----BEGIN RSA PRIVATE KEY----- +MIIEogIBAAKCAQEA2M/hYSQQAJxQfcLxcRob7cq4miMLbyWv+Diq4qppENU8XvNS +7p/JSc8XqWRcGX6FusObKjpskakEL1+tf2wBYGW5jbtgiHzg4J46p+iK/OFi9t+8 +O6wjva3+5kWrjvkQIKL/pbGfdvVQjq1j5bg2fbXKbQqb7CbbhxDhRrW3xVt5A+PF +G3+iP0qHcBvWuN1CIUHVvUORSdkY44+8xCs+RjalMqVsOyPzTuMu+ypfPtaHfcGM +DWwkvpXUwq9VrszfrGqi3qWJ6nBQy20eXRUeEnorDS7Rolj+/bILhtbUUT/wFzX7 +F/er1zquAQSntgiMUutB50zpPefNe5Jf/uknZwIDAQABAoIBAEK9izVp8nsdozFz +ZvtX/Czw8G6SRxzJzydr2j64EmuaGGFqKadHDKFp3HDKcgdvCR96wzL2lSEplB45 +J6FH0Ayqg7n7ajXZqye5zXiOfuplPx+bs8fyr3XeL88Wtz7lsG87fdK1oM7vqQKi +oVO/pxZ/dXang2Poej9TlOU9XvqjIoTo4HNckRhEF7MJV+2nQB/d+edlSt9oLrN4 +UxOEBeTEomj+ujJBWFOasOcztRXMl/cQp3PdSZ5Ap4aXElblK5/MocM+tyh6RU++ +hLVBpxxjH2CQbG+HMn0rxn7NbtpBdYU3GTjBdKJUv0MQ9m8s39AFfX4h1YW7PCte +uO0tyfECgYEA+pwlgTWnQL5dlwkw+qsg4mhiG0XcDFsfrh6W/TcsUJnfQIGZCJnC +9NxCJEmfQTxgJWEmTvPj+Mi4iFmoMYlfTo1OqI36pXNnH0fmh4l0l9COAUsarQVJ +DzDh5LaCpOn8uSfgJSzdoBY4aYkxJfl8AVsApCmHBojp9Xjw1gis1csCgYEA3Xmk +qm1BeGkUU/nGDCjz6AuIYEbWuj7OwnV/EsvsGsMTDryoWsk0ZTI0KJdyI/RhjMdM +jnv37EOy8YL1NjD01NinQ7Ac1wY1DSUrA+jJ6ctSkjkPFHKVBmiC2c5TJiTXWxd6 +Ms7fGQesIrdSdbr7+GrkogKCJyKEwskcRNsFIVUCgYAVqKkVIa3yyoEMLJ8vCtss +thjEgqUh7D4wrjQebkZGMQY6bPjs5V0UDHUq7bH10WH92QeWiz2IotBP+G+w2E6E +W3IQ3zRqWC7xZSVnUVniRQQA2JzVlz9ZMMBVI/Dmj9JnriqRC++bhlAZzuGQhp9Q +/qxvg6dHyU1vCaTG2rkVQQKBgGrkviwfhVznYo9RW1wdG/OgNVc6eYL0s3g5m3Zo +A/G6u2T+e0cuG2LIQ1JgHg/qcHaw7mtKvBNFbYWRTma0ZEaitsA9rP2/GjXY/ZTq +cyGrJ5Q5g+S7lN+964HSSmF7oJFvmGQTYqxC/rrWvi7ia6CvfuAO0+CtkWQuCr3U +8/ntAoGAawnmD9G1JVUSo1ONbV2gsg/Lv6SSq0dbWja4B0fhAR4BWtSFQV8ePTwk +Q5bSZc/zdNHcKPIMr29Aa3TTy7jXHkAs2frAwKGRyErt/cVAbah+BLOWxniTNfgD +6CvUI8gTLJ+Dklmer2o6b0JZEeSxz93xx7sy9E5SqHGStOZRutA= +-----END RSA PRIVATE KEY----- +""" + +TEST_KEY_PKCS8 = """ +-----BEGIN PRIVATE KEY----- +MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCgshr1g7WzZ/iF +tXbL+R2tTkzOsKVWeLli7Xe1+5W9jz+mEd/zWl/AD/QuM0ncNNkOQfyCVKgtzX4p +k2yerf4QtdbwvVP36aqs6D0wH1AqkNfC338bbN1eNPNtZ2sG8muAppv4hjUBT7ck +p3hrQ9VU596PFloYlM6M91mkIWGX76oig+pPSL62P82OwYbM3dBKIJsQaAEWbqf6 +V08D4MOUubt8zcS3FIpztLKQ+NPrzWGoPQoaHx0cruDZLWu2RdUTav4S3G0BLJFh +TzekF6KGIUC/CdB6uOKFGobJxr9Pz/ZhiM/3PE3KBZXWeNS/+lerV2JB3MAAL2zc +oPTHbUyXAgMBAAECggEADSAETVmeItKoUlIhgK8kCjkzKHigxdYjcjM/dq40q0sY +IZoj0WO9dARO5zfKMxcKmXxbQaaPe/o6RpljeExxdsTwvYYBOzurHzydbsudR8oV +IrYQKC8f9Odcm0Akq5MWTmeNUA200PAatH7h5NpMAjIpaUTOmWO3a9JyJb086Y9e +Tv6Em0iNjZiyfuTz8q3BRcuZvwdXN3ZJVgTqJTN4iUfKmSbhDieIKHw1NieYbimD +/LNyt6OMcZiU5RKvujY+kGLJBhJ5DO3/yl7+DfHeQjHrnyrddB2O8IwqBZp6wcmS +5zTGX3vqclMsZCv2mKoVdWA3QlbDc36N7wumr9nYPQKBgQDbNsFGAaQ26bEsp37H +ixegd9vhqeO8qFuTDMqNiKVs6t0ZV8PUmI0pB8ZzXxtWRiolUXqxZiqvilzhRz4m +B/1lYKmljpSJdL4fX+t65MNW4pY7S/bqd/5Rt4j+DA77Ez9lcbVH9LwpWrhMXFvU ++qyFVglNq7vrDN82FHkrDVf3dQKBgQC7qXWE2hDMZ7CerNQu06GSLehQ65YBtRg/ +oQ62dcpEv7vxVn24fzwtMBFlmt3WY+N9Vr8O9dxuiX7PmBOIf1P3BW+qLU+bVExc +AQychjKoHRsimkqDaX8XqjJkA7fD56zahLKLShKwmUdWoNvM9Vf8jokO7Ky6/Ckk +8DR1YRI+WwKBgCJ3OWwIVglMNM8Qhp5ZAB36uleAno3K6pr0J+SY12mIg7q4UdHb +KlaJImqBSYsG/Rlg9iluJsfyuSeyS9Mu7mhCCmo9QJ1b/YhxTFTGIvBGL9ii2O+E +bjfq418OBkzzecQ5XIczs2HNIazf2JBj2TwditjrZ2BNZcra7TIXpv0BAoGAFhyP +dknjdNI641dl4OaawQibuP0pPaZC5yxo+WLDpHobAQK56YVEZ+akBh7LSLmYUmpo +NLEPeg7FXFyE4K5wDEkdsSWJnbm/DZAqFj9RtLFJQzh+0yEZEobBTTuK6ZBgs5az +GuptbVWCLFjjzTCTX3TD/Xgxmg6gInQTGkmHxsUCgYB0p7w195Q+/fUhdCSnvwNQ +VlrX5EBjvjkPZFQkXNCFoeIhQuADyQQJ7P9HHlrxSjHsbZOxTTcVQatm0kk8s6Et +LAxPD9wtwVxzZHHVXCK+l8JNfBmiW14n6BQvkTlGgt1HFojrx4ubz86VwgP8FMuW +Qq9ELI2S82qfw0VlsoZWxQ== +-----END PRIVATE KEY----- +""" + + + +def _b64url_decode(s): + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + +def _emsa_sha256(message, em_len): + prefix = bytes.fromhex("3031300d060960864801650304020105000420") + t = prefix + hashlib.sha256(message).digest() + return b"\x00\x01" + b"\xff" * (em_len - len(t) - 3) + b"\x00" + t + + +def _rsa_verify(pem, message, signature): + """纯 Python RSA 公钥验签（与签名实现独立的验算）。""" + n, e, _d = _parse_rsa_private_key(pem) + em_len = (n.bit_length() + 7) // 8 + m = pow(int.from_bytes(signature, "big"), e, n) + return m.to_bytes(em_len, "big") == _emsa_sha256(message, em_len) + + +class Rs256TestCase(unittest.TestCase): + def test_sign_and_verify_pkcs1(self): + message = b"github app jwt signing input" + sig = rsa_sign_sha256(TEST_KEY_PKCS1, message) + n, _e, _d = _parse_rsa_private_key(TEST_KEY_PKCS1) + self.assertEqual(len(sig), (n.bit_length() + 7) // 8) + self.assertTrue(_rsa_verify(TEST_KEY_PKCS1, message, sig)) + + def test_sign_and_verify_pkcs8(self): + message = b"another signing input" + sig = rsa_sign_sha256(TEST_KEY_PKCS8, message) + self.assertTrue(_rsa_verify(TEST_KEY_PKCS8, message, sig)) + self.assertFalse(_rsa_verify(TEST_KEY_PKCS8, b"tampered", sig)) + + def test_bad_key(self): + with self.assertRaises(AppError): + rsa_sign_sha256("not a key", b"x") + with self.assertRaises(AppError): + rsa_sign_sha256("-----BEGIN FOO-----\nAAAA\n-----END FOO-----", + b"x") + + +class JwtTestCase(unittest.TestCase): + def test_mint_structure_and_signature(self): + jwt = mint_app_jwt("5476837", TEST_KEY_PKCS1, now=1000000) + parts = jwt.split(".") + self.assertEqual(len(parts), 3) + header = json.loads(_b64url_decode(parts[0])) + self.assertEqual(header["alg"], "RS256") + payload = json.loads(_b64url_decode(parts[1])) + self.assertEqual(payload["iss"], "5476837") + self.assertEqual(payload["iat"], 1000000 - 60) + self.assertEqual(payload["exp"], 1000000 + 540) + signing_input = (parts[0] + "." + parts[1]).encode("ascii") + self.assertTrue(_rsa_verify( + TEST_KEY_PKCS1, signing_input, _b64url_decode(parts[2]))) + + +def _fake_exchange(expected_installation="175380082"): + def fake(method, url, headers, body): + assert method == "POST", method + assert "/app/installations/%s/access_tokens" % expected_installation in url, url + assert headers["Authorization"].startswith("Bearer "), headers + assert headers["Authorization"].count(".") == 2 + return 201, {"token": "fake-installation-token", + "expires_at": "2030-01-01T00:00:00Z", + "permissions": {"contents": "write"}, + "repository_selection": "selected"} + return fake + + +class TokenExchangeTestCase(unittest.TestCase): + def test_exchange_ok(self): + result = exchange_installation_token( + "5476837", TEST_KEY_PKCS1, "175380082", + http_client=_fake_exchange()) + self.assertEqual(result["token"], "fake-installation-token") + self.assertEqual(result["repository_selection"], "selected") + + def test_exchange_failure_raises(self): + def fake(method, url, headers, body): + return 401, {"message": "Bad credentials"} + with self.assertRaises(AppError) as ctx: + exchange_installation_token( + "5476837", TEST_KEY_PKCS1, "175380082", http_client=fake) + self.assertEqual(ctx.exception.reason, "token_exchange_failed") + + +class ResolveTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.conn = _db_connect(self.db_path) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def _save_config(self): + save_app_config(self.conn, { + "app_id": 5476837, "slug": "fake-app", "client_id": "Iv1.x", + "webhook_secret": "fake-secret", "pem": TEST_KEY_PKCS1, + "html_url": "https://github.com/apps/fake-app"}) + + def test_resolve_full_chain(self): + self._save_config() + record_installation_repos(self.conn, 175380082, "octo", + ["octo/a"]) + token = resolve_app_token(self.conn, "octo/a", + http_client=_fake_exchange()) + self.assertEqual(token, "fake-installation-token") + + def test_resolve_falls_back_to_none(self): + # 无配置、无安装、换 token 失败：一律 None（调用方走 PAT）。 + self.assertIsNone(resolve_app_token( + self.conn, "octo/a", http_client=_fake_exchange())) + self._save_config() + self.assertIsNone(resolve_app_token( + self.conn, "octo/a", http_client=_fake_exchange())) + record_installation_repos(self.conn, 175380082, "octo", + ["octo/a"]) + + def failing(method, url, headers, body): + return 401, {"message": "Bad credentials"} + self.assertIsNone(resolve_app_token( + self.conn, "octo/a", http_client=failing)) + + def test_find_installation(self): + record_installation_repos(self.conn, 11, "octo", ["octo/a"]) + self.assertEqual(find_installation_for_repo(self.conn, "octo/a"), + "11") + self.assertIsNone(find_installation_for_repo(self.conn, "octo/b")) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_pr_app_token.py b/cloud/tests/test_pr_app_token.py new file mode 100644 index 0000000..9e8ac02 --- /dev/null +++ b/cloud/tests/test_pr_app_token.py @@ -0,0 +1,176 @@ +"""pr.py 的 App-token 路径单测：push 走 credential helper、建 PR +走 GitHub API、gh 完全不被调用；remote 解析与失败分支。 + +不联网、不跑真实 git/gh：git 用假 runner，GitHub API 用假 +客户端；token 值绝不出现在记录的命令里（只出现变量名）。 +""" + +import os +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.pr import ( + PrStepError, + check_pr_ready, + create_task_pr, + parse_git_remote_owner_repo, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.pr import ( + PrStepError, + check_pr_ready, + create_task_pr, + parse_git_remote_owner_repo, + ) + + +def _result(returncode=0, stdout="", stderr=""): + return SimpleNamespace(returncode=returncode, stdout=stdout, + stderr=stderr) + + +class _Git: + """假 git runner：remote/branch/add/commit/push 全按脚本回。""" + + def __init__(self, remote="https://github.com/octo/a.git"): + self.calls = [] + self.remote = remote + + def __call__(self, cmd, cwd): + self.calls.append(list(cmd)) + if "remote" in cmd: + return _result(0, self.remote + "\n") + if "branch" in cmd: + return _result(0, "host/t1\n") + if "add" in cmd: + return _result(0, "") + if "commit" in cmd: + return _result(1, "", + "nothing to commit, working tree clean") + if "push" in cmd: + return _result(0, "") + return _result(0, "") + + +def _gh_forbidden(cmd, cwd): + raise AssertionError("App 路径不应调用 gh：%r" % (cmd,)) + + +class _FakeGithubApi: + """假 GitHub API：记录调用，GET 仓库回 default_branch， + POST pulls 回 PR 链接；失败模式可配。""" + + def __init__(self, default_branch="main", pulls_status=201): + self.calls = [] + self.default_branch = default_branch + self.pulls_status = pulls_status + + def __call__(self, method, url, token, payload): + self.calls.append((method, url, token, payload)) + if method == "GET" and url.endswith("/repos/octo/a"): + return 200, {"default_branch": self.default_branch} + if method == "POST" and url.endswith("/repos/octo/a/pulls"): + if self.pulls_status != 201: + return self.pulls_status, {"message": "Validation Failed"} + return 201, {"html_url": "https://github.com/octo/a/pull/42", + "number": 42} + raise AssertionError("unexpected API call: %s %s" % (method, url)) + + +def _worktree(tmp): + d = os.path.join(tmp, "t1") + os.makedirs(d, exist_ok=True) + return d + + +class ParseRemoteTestCase(unittest.TestCase): + def test_variants(self): + self.assertEqual( + parse_git_remote_owner_repo( + "https://github.com/octo/a.git"), ("octo", "a")) + self.assertEqual( + parse_git_remote_owner_repo( + "https://github.com/octo/a"), ("octo", "a")) + self.assertEqual( + parse_git_remote_owner_repo( + "https://user@github.com/octo/a.git"), ("octo", "a")) + self.assertEqual( + parse_git_remote_owner_repo( + "git@github.com:octo/a.git"), ("octo", "a")) + self.assertEqual( + parse_git_remote_owner_repo( + "ssh://git@github.com/octo/a.git"), ("octo", "a")) + + def test_non_github_is_none(self): + self.assertIsNone(parse_git_remote_owner_repo( + "https://example.com/octo/a.git")) + self.assertIsNone(parse_git_remote_owner_repo("")) + self.assertIsNone(parse_git_remote_owner_repo(None)) + + +class AppPathTestCase(unittest.TestCase): + def test_push_uses_credential_helper_and_pr_via_api(self): + with tempfile.TemporaryDirectory() as tmp: + _worktree(tmp) + git = _Git() + api = _FakeGithubApi() + result = create_task_pr( + "t1", 1, worktrees_root=tmp, git_runner=git, + gh_runner=_gh_forbidden, gh_which="/nonexistent/gh", + app_token="fake-app-token", github_api=api) + self.assertEqual( + result["pr_url"], "https://github.com/octo/a/pull/42") + self.assertEqual(result["pr_number"], 42) + self.assertEqual(result["branch"], "host/t1") + pushes = [c for c in git.calls if "push" in c] + self.assertEqual(len(pushes), 1) + helper = [a for a in pushes[0] + if a.startswith("credential.helper=")][0] + self.assertIn("$GIT_TOKEN", helper) + self.assertNotIn("fake-app-token", " ".join(pushes[0])) + posts = [c for c in api.calls if c[0] == "POST"] + self.assertEqual(len(posts), 1) + self.assertEqual(posts[0][2], "fake-app-token") + self.assertEqual(posts[0][3]["head"], "host/t1") + self.assertEqual(posts[0][3]["base"], "main") + self.assertEqual(posts[0][3]["title"], "cloud: t1") + + def test_app_path_skips_gh_checks(self): + # gh_which 指到不存在的路径、gh_runner 会炸：App 路径下 + # check_pr_ready 仍应通过（只查 worktree + remote）。 + with tempfile.TemporaryDirectory() as tmp: + _worktree(tmp) + self.assertTrue(check_pr_ready( + os.path.join(tmp, "t1"), git_runner=_Git(), + gh_which="/nonexistent/gh", gh_runner=_gh_forbidden, + app_token="x")) + + def test_non_github_remote_fails_at_push(self): + with tempfile.TemporaryDirectory() as tmp: + _worktree(tmp) + with self.assertRaises(PrStepError) as ctx: + create_task_pr( + "t1", 1, worktrees_root=tmp, + git_runner=_Git("https://example.com/octo/a.git"), + gh_runner=_gh_forbidden, app_token="x", + github_api=_FakeGithubApi()) + self.assertEqual(ctx.exception.step, "push") + + def test_api_failure_is_pr_create_step(self): + with tempfile.TemporaryDirectory() as tmp: + _worktree(tmp) + with self.assertRaises(PrStepError) as ctx: + create_task_pr( + "t1", 1, worktrees_root=tmp, git_runner=_Git(), + gh_runner=_gh_forbidden, app_token="x", + github_api=_FakeGithubApi(pulls_status=422)) + self.assertEqual(ctx.exception.step, "pr_create") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_server_pr_app_token.py b/cloud/tests/test_server_pr_app_token.py new file mode 100644 index 0000000..fc2cb5a --- /dev/null +++ b/cloud/tests/test_server_pr_app_token.py @@ -0,0 +1,248 @@ +"""POST /tasks/{id}/pr 的 App-token 接线单测（端点级）： + +App 路径：项目 repo_url 解析出仓库名、库里有该仓库的 App +安装记录 -> 现换 installation token（fake 换 token 客户端 + +测试专用 pem，JWT 纯标准库签发、不联网）-> push 走 +credential helper、建 PR 走 GitHub API（fake），响应里 +auth_via=app。 +PAT 回落：没有安装记录 -> 走原来的 gh 路径，auth_via=pat。 + +git/GitHub API 一律 fake，不跑真实命令；token 值绝不出现 +在 git 命令记录里。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.auth import AuthStore + from cloud.db import connect as _db_connect + from cloud.github_app import ( + record_installation_repos, + save_app_config, + ) + from cloud.server import create_server + from cloud.tests.test_app_token import TEST_KEY_PKCS1 +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.db import connect as _db_connect + from cloud.github_app import ( + record_installation_repos, + save_app_config, + ) + from cloud.server import create_server + from cloud.tests.test_app_token import TEST_KEY_PKCS1 + + +def _result(returncode=0, stdout="", stderr=""): + return SimpleNamespace(returncode=returncode, stdout=stdout, + stderr=stderr) + + +class _Git: + """假 git：remote/branch/add/commit/push，记录命令。""" + + def __init__(self, remote, branch): + self.calls = [] + self.remote = remote + self.branch = branch + + def __call__(self, cmd, cwd): + self.calls.append(list(cmd)) + if "remote" in cmd: + return _result(0, self.remote + "\n") + if "branch" in cmd: + return _result(0, self.branch + "\n") + if "add" in cmd: + return _result(0, "") + if "commit" in cmd: + return _result(1, "", + "nothing to commit, working tree clean") + if "push" in cmd: + return _result(0, "") + return _result(0, "") + + +class _Gh: + """假 gh：auth 通过、pr create 回链接。""" + + def __init__(self, pr_url): + self.calls = [] + self.pr_url = pr_url + + def __call__(self, cmd, cwd): + self.calls.append(list(cmd)) + if cmd[:2] == ["gh", "auth"]: + return _result(0, "", "") + if cmd[:2] == ["gh", "pr"]: + return _result(0, self.pr_url + "\n") + return _result(0, "") + + +class _Api: + """假 GitHub API：GET 仓库回 default_branch，POST pulls 回 PR。""" + + def __init__(self): + self.calls = [] + + def __call__(self, method, url, token, payload): + self.calls.append((method, url, token, payload)) + if method == "GET": + return 200, {"default_branch": "main"} + if method == "POST": + return 201, {"html_url": "https://github.com/octo/a/pull/77", + "number": 77} + raise AssertionError("unexpected API call: %s %s" % (method, url)) + + +def _token_client(method, url, headers, body): + """假换 token：JWT 已在前面签好（只验格式），直接回 token。""" + assert method == "POST", method + assert "/access_tokens" in url, url + assert headers["Authorization"].startswith("Bearer "), headers + return 201, {"token": "srv-app-token", + "expires_at": "2030-01-01T00:00:00Z", + "permissions": {"contents": "write"}, + "repository_selection": "selected"} + + +class _Case(unittest.TestCase): + def setUp(self): + 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.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + self.worktrees = os.path.join(self.tmp.name, "worktrees") + os.makedirs(self.worktrees, exist_ok=True) + self.server.host_worktrees_dir = self.worktrees + self.git = None + self.api = _Api() + self.server.pr_github_api = self.api + self.server.pr_app_token_client = _token_client + + 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() + + 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() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _project(self, name, repo_url): + status, project = self._req( + "POST", "/projects", + {"name": name, "repo_url": repo_url, "auto_pr": 1}, + self.owner_key) + self.assertEqual(status, 201, project) + return project["id"] + + def _task(self, task_id, project_id): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at," + " updated_at, last_heartbeat) VALUES(?, ?, ?, 'merged'," + " '[]', 0, 'system', 0, 0, NULL)", + (task_id, project_id, "do it")) + conn.commit() + finally: + conn.close() + os.makedirs(os.path.join(self.worktrees, task_id), exist_ok=True) + + def test_app_path_via_endpoint(self): + project_id = self._project( + "octo/a", "https://github.com/octo/a") + conn = _db_connect(self.db_path) + try: + save_app_config(conn, { + "app_id": 5476837, "slug": "fake-app", + "client_id": "Iv1.x", "webhook_secret": "fake-secret", + "pem": TEST_KEY_PKCS1, + "html_url": "https://github.com/apps/fake-app"}) + record_installation_repos(conn, 175380082, "octo", + ["octo/a"]) + finally: + conn.close() + self.git = _Git("https://github.com/octo/a.git", "host/t-app-1") + self.server.pr_git_runner = self.git + self._task("t-app-1", project_id) + status, body = self._req( + "POST", "/tasks/t-app-1/pr", key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["pr_url"], + "https://github.com/octo/a/pull/77") + self.assertEqual(body["pr_number"], 77) + self.assertEqual(body["auth_via"], "app") + pushes = [c for c in self.git.calls if "push" in c] + self.assertEqual(len(pushes), 1) + self.assertTrue(any(a.startswith("credential.helper=") + for a in pushes[0])) + self.assertNotIn("srv-app-token", + " ".join(" ".join(c) for c in self.git.calls)) + posts = [c for c in self.api.calls if c[0] == "POST"] + self.assertEqual(len(posts), 1) + self.assertEqual(posts[0][2], "srv-app-token") + + def test_pat_fallback_via_endpoint(self): + project_id = self._project( + "octo/b", "https://github.com/octo/b") + self.git = _Git("https://github.com/octo/b.git", "host/t-pat-1") + self.server.pr_git_runner = self.git + gh = _Gh("https://github.com/octo/b/pull/88") + self.server.pr_gh_runner = gh + self.server.pr_gh_which = "/usr/bin/gh" + self._task("t-pat-1", project_id) + status, body = self._req( + "POST", "/tasks/t-pat-1/pr", key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["pr_url"], + "https://github.com/octo/b/pull/88") + self.assertEqual(body["auth_via"], "pat") + pushes = [c for c in self.git.calls if "push" in c] + self.assertEqual(len(pushes), 1) + self.assertFalse(any(a.startswith("credential.helper=") + for a in pushes[0])) + self.assertEqual(self.api.calls, []) + + +if __name__ == "__main__": + unittest.main()