"""GitHub App 一键安装：manifest 建 App + 安装记录 + App webhook。 范围（诚实边界）：本模块只做“安装”这一半——manifest 生成、 建 App 回调换凭证、安装事件记录、App webhook 验签。App 的 pem 虽然存下来了，但本模块不用它签 JWT、不用它换 installation token：推分支、建 PR 目前仍走 gh + PAT（见 cloud/pr.py）， App-token API 动作是下一阶段，不在本模块假装已做。 流程（GitHub manifest 流程，见 GitHub 文档）： POST /github-app/setup/state（持本站 key）-> state + setup_url -> 用户打开 setup 页（表单含 manifest JSON + state）点提交 -> GitHub 建 App 页（用户确认 Create） -> GitHub 跳回 /github-app/callback?code=...&state=... -> server 验 setup state（一次性、1 小时过期）-> 换 code （POST api.github.com/app-manifests/{code}/conversions， 无需认证，1 小时内有效）-> 存 App 配置 -> 成功页给安装链接（https://github.com/apps// installations/new），用户点安装 -> installation 事件 到 /webhooks/github -> 记安装记录。 凭证存放：App 配置存数据库 github_app_config 表（单行）， 不进仓库、不进日志、不出现在任何 API 响应里；webhook_secret 与 pem 只在 server 进程内读（验签用 secret，JWT 阶段用 pem）。 与 env 里 GITHUB_WEBHOOK_SECRET 的关系：手动 webhook 仍用 env secret，App webhook 用库里 secret，验签时两个都试（过渡 期共存）；App 装好后应删掉手动 webhook，避免同一事件送两 遍（/fix-review 另有内容去重兜底，见 claim_fix_review）。 零第三方依赖，只用 Python 标准库；换 code 与 userinfo 均可 注入 fake http 客户端做单测。 """ import base64 as _base64 import hashlib import json import secrets import time import urllib.request from cloud.db import connect as _db_connect # setup state 有效期：与 manifest code 的 1 小时对齐。 SETUP_STATE_TTL_SECONDS = 3600 # manifest 默认值：App 名与主页；hook / 回调地址由部署的公网 # 域名（GITHUB_APP_PUBLIC_BASE）拼出，不硬编码 diante.me。 DEFAULT_APP_NAME = "codex-app-server" DEFAULT_APP_URL = "https://github.com/openqj/codex-app-server" # App 权限与事件：全量获取（用户 2026-10-27 明确要求：无论 # 什么身份，取该身份授权范围内的所有权限）。 # 原则：manifest 里按仓库 / 组织 / 企业 / 账号四类全部拉到最高 # 档，安装时 GitHub 只授予该身份能给的那部分——个人账号装不上 # 组织/企业权限，组织装不上企业权限，企业装不上仓库/组织权限， # 这是 GitHub 的身份边界，不是代码能绕过的。换 token 时 # exchange_installation_token 不传 permissions 子集（payload # 为空），所以 token 本来就是安装权限的全量，不会被代码降级。 # 这是 manifest 的默认值，只影响以后新建的 App；已存在的线上 # App（id 5476837）不会因改代码自动变权限，必须在 GitHub App # 设置里改权限并由安装方接受新权限后才生效。企业安装另需 # App 为公开或内部（GITHUB_APP_PUBLIC=1），私有默认装不上企业。 # 风险明示：administration / organization_administration / # enterprise 类 write 含删库、转移、改组织/企业设置、管账单， # account 类 write 含管用户密钥、Gist、关注、星标，配 All # repositories 安装时影响面为该身份的全部授权范围——只装在 # 自己信任的身份上。 # 诚实边界：仓库/组织/账号键名有官方文档与 manifest 示例佐证； # 企业类键名（enterprise_*）未在本机用真实 GitHub 创建 App # 实测过——新建 App 时 GitHub 建 App 表单会逐项校验，若某键 # 被拒，删该键再建即可，线上旧 App 不受影响。single_file 未 # 列入：它必须配 single_file_paths 才能生效，空列无意义。 APP_PERMISSIONS = { # 仓库：全部拉到最高档，metadata 只读是 GitHub 自动档。 "actions": "write", "actions_variables": "write", "administration": "write", "artifact_metadata": "write", "attestations": "write", "checks": "write", "codespaces": "write", "codespaces_lifecycle_admin": "write", "codespaces_metadata": "read", "codespaces_secrets": "write", "contents": "write", "dependabot_secrets": "write", "deployments": "write", "discussions": "write", "environments": "write", "issues": "write", "merge_queues": "write", "metadata": "read", "packages": "write", "pages": "write", "pull_requests": "write", "repository_advisories": "write", "repository_creation": "write", "repository_custom_properties": "write", "repository_hooks": "write", "secret_scanning_alerts": "write", "secrets": "write", "security_events": "write", "statuses": "write", "vulnerability_alerts": "write", "workflows": "write", # 组织：全部拉到最高档（仅组织安装时生效）。 "members": "write", "organization_administration": "write", "organization_announcement_banners": "write", "organization_copilot_seat_management": "write", "organization_custom_properties": "write", "organization_custom_roles": "write", "organization_dependabot_secrets": "write", "organization_events": "read", "organization_hooks": "write", "organization_packages": "write", "organization_plan": "read", "organization_projects": "write", "organization_secrets": "write", "organization_self_hosted_runners": "write", "organization_user_blocking": "write", # 企业：全部拉到最高档（仅企业安装且 App 公开/内部时生效）。 "enterprise_administration": "write", "enterprise_billing": "write", "enterprise_copilot_seat_management": "write", "enterprise_custom_properties_for_organizations": "write", "enterprise_organization_installation_repositories": "write", "enterprise_organization_installations": "write", "enterprise_policies": "write", "enterprise_rules": "write", # 账号：全部拉到最高档（仅用户另外授权 App 代表自己时生效）。 "blocking": "write", "codespaces_user_secrets": "write", "copilot_editor_context": "read", "copilot_messages": "read", "copilot_requests": "write", "email_addresses": "write", "followers": "write", "gists": "write", "git_signing_ssh_public_keys": "write", "git_ssh_keys": "write", "gpg_keys": "write", "interaction_limits": "write", "knowledge_bases": "write", "plan": "read", "private_repository_invitations": "read", "profile": "write", "starring": "write", "user_events": "read", "user_models": "read", "watching": "write", } APP_EVENTS = [ "push", "create", "delete", "pull_request", "pull_request_review", "pull_request_review_comment", "issue_comment", "issues", "check_run", "check_suite", "workflow_run", "workflow_job", "release", "deployment", "deployment_status", "discussion", "discussion_comment", "commit_comment", "code_scanning_alert", "secret_scanning_alert", "label", "milestone", "member", "membership", "organization", "team", "team_add", "repository", "fork", "watch", "meta", "status", "project", "project_card", "project_column", "gollum", "page_build", "public", "repository_dispatch", ] class AppError(Exception): """App 流程失败：换 code 失败、响应缺字段等。 reason 取值：``exchange_failed`` / ``config_incomplete`` / ``state_invalid`` / ``state_expired``。 """ def __init__(self, message, reason=None): super().__init__(message) self.reason = reason def build_app_manifest(public_base, app_name=DEFAULT_APP_NAME, app_url=DEFAULT_APP_URL, public=False): """拼 manifest dict：GitHub 建 App 表单直接提交这份 JSON。 public=False 是单租户部署的默认（只有 App 所有者能安装）； 多租户 SaaS 形态才需要 public=True（改环境变量翻转）。 """ if not public_base or not isinstance(public_base, str): raise ValueError("public_base required") base = public_base.rstrip("/") return { "name": app_name, "url": app_url, "hook_attributes": {"url": base + "/webhooks/github"}, "redirect_url": base + "/github-app/callback", "callback_urls": [base + "/github-app/callback"], "public": bool(public), "default_permissions": dict(APP_PERMISSIONS), "default_events": list(APP_EVENTS), } class SetupStateStore: """setup state 存储：一次性、1 小时过期，防回调被冒用。 表 github_app_setup_states：state（主键）、created_at、 issued_by（签发人用户 id，P0 安全批新增）。新表建表即 含该列；老库在 __init__ 里用 ALTER 补列（列已存在时 报错直接忽略，老库不炸，见 except 分支）。 state 只由全局管理员调 POST /github-app/setup/state 签发；回调验到即删，同一 state 第二次回调必然失败。 """ def __init__(self, db_path): self.db_path = str(db_path) self.conn = _db_connect(self.db_path) self.conn.execute( "CREATE TABLE IF NOT EXISTS github_app_setup_states(" "state TEXT PRIMARY KEY, created_at REAL," " issued_by TEXT)" ) self.conn.commit() try: self.conn.execute( "ALTER TABLE github_app_setup_states" " ADD COLUMN issued_by TEXT") self.conn.commit() except Exception: try: self.conn.rollback() except Exception: pass def create_state(self, issued_by=None): state = secrets.token_urlsafe(32) self.conn.execute( "INSERT INTO github_app_setup_states" "(state, created_at, issued_by)" " VALUES(?, ?, ?)", (state, time.time(), issued_by), ) self.conn.commit() return state def state_issuer(self, state): """查 state 的签发人（回调审计用，不消耗 state）。""" if not state: return None try: row = self.conn.execute( "SELECT issued_by FROM github_app_setup_states" " WHERE state = ?", (state,), ).fetchone() except Exception: return None if row is None: return None try: return row[0] except Exception: return None def consume_state(self, state, now=None, ttl_seconds=SETUP_STATE_TTL_SECONDS, allowed_issuers=None): """校验并消耗 state：成功返回 True，失败返回 False。 不存在 -> False；过期 -> 删行并返回 False；成功 -> 删行 并返回 True。``now`` 只为单测注入时间。 P0 安全批后：allowed_issuers 非空时，签发人不在该 集合里一律返回 False 且不消耗 state（fail-closed）； 签发人为空（升级前的旧 state）同样不在集合里，等 于升级后旧 state 全部作废，需管理员重新签发。注意： 回调本身是浏览器无认证重定向，服务器无法从回调请求 里认出调用方就是签发人，所以这里核对的是“签发人仍 在管理员清单里”，不是“回调者就是签发人”——后者 在本流程里做不到，文档与 PR 描述里必须如实写清， 不能谎称回调做了身份认证。 """ if not state: return False if now is None: now = time.time() row = self.conn.execute( "SELECT created_at, issued_by FROM github_app_setup_states" " WHERE state = ?", (state,), ).fetchone() if row is None: return False if allowed_issuers is not None: try: issuer = row[1] except Exception: issuer = None if issuer not in allowed_issuers: return False self.conn.execute( "DELETE FROM github_app_setup_states WHERE state = ?", (state,), ) self.conn.commit() if now > row[0] + ttl_seconds: return False return True def _urllib_post_json(url, headers): """默认的换 code 实现：POST 空 body，返回解析后的 dict。""" request = urllib.request.Request(url, data=b"", method="POST", headers=headers) with urllib.request.urlopen(request, timeout=30) as response: raw = response.read().decode("utf-8", "replace") try: parsed = json.loads(raw) except ValueError as exc: raise AppError("manifest conversion response is not JSON: %s" % (exc,), reason="exchange_failed") from exc if not isinstance(parsed, dict): raise AppError("manifest conversion response is not an object", reason="exchange_failed") return parsed def exchange_manifest_code(code, http_client=None): """用 manifest code 换 App 配置：返回 {id, slug, client_id, webhook_secret, pem, html_url}（缺关键字段即 AppError）。 http_client 注入约定（单测用）：``http_client(url, headers) -> dict``。真实路径无需认证（code 本身就是一次性凭证）， 1 小时内必须完成。 """ if not code or not isinstance(code, str): raise AppError("code required", reason="exchange_failed") url = "https://api.github.com/app-manifests/%s/conversions" % code headers = {"Accept": "application/vnd.github+json", "User-Agent": "codex-app-server"} client = http_client if http_client is not None else _urllib_post_json try: result = client(url, headers) except AppError: raise except Exception as exc: raise AppError( "manifest conversion failed: %s" % (exc,), reason="exchange_failed", ) from exc if not isinstance(result, dict) or result.get("error"): raise AppError("manifest conversion failed: %r" % (result,), reason="exchange_failed") missing = [k for k in ("id", "webhook_secret", "pem") if not result.get(k)] if missing: raise AppError( "manifest conversion response missing: %s" % ",".join(missing), reason="config_incomplete") return { "app_id": result.get("id"), "slug": result.get("slug"), "client_id": result.get("client_id"), "webhook_secret": result.get("webhook_secret"), "pem": result.get("pem"), "html_url": result.get("html_url"), } def save_app_config(conn, config): """存 App 配置（单行，id 恒为 1；重复 setup 直接覆盖）。 调用方须保证 config 含 webhook_secret 与 pem；本函数不校验 它们的值，只负责落库。secret/pem 绝不出现在返回值里。 """ conn.execute( "CREATE TABLE IF NOT EXISTS github_app_config(" "id INTEGER PRIMARY KEY CHECK(id = 1), " "app_id TEXT, slug TEXT, client_id TEXT, " "webhook_secret TEXT, pem TEXT, html_url TEXT, created_at REAL)" ) conn.execute( "INSERT OR REPLACE INTO github_app_config(id, app_id, slug," " client_id, webhook_secret, pem, html_url, created_at)" " VALUES(1, ?, ?, ?, ?, ?, ?, ?)", (config.get("app_id"), config.get("slug"), config.get("client_id"), config.get("webhook_secret"), config.get("pem"), config.get("html_url"), time.time()), ) conn.commit() def get_app_config(conn): """取 App 配置 dict，未配置返回 None。 返回里含 webhook_secret 与 pem——调用方只许在 server 进程内 用（验签 / JWT），绝不放进任何 API 响应；公开状态接口须 另外拼不含 secret 的视图。 """ conn.execute( "CREATE TABLE IF NOT EXISTS github_app_config(" "id INTEGER PRIMARY KEY CHECK(id = 1), " "app_id TEXT, slug TEXT, client_id TEXT, " "webhook_secret TEXT, pem TEXT, html_url TEXT, created_at REAL)" ) row = conn.execute( "SELECT app_id, slug, client_id, webhook_secret, pem, html_url," " created_at FROM github_app_config WHERE id = 1", ).fetchone() if row is None: return None return { "app_id": row[0], "slug": row[1], "client_id": row[2], "webhook_secret": row[3], "pem": row[4], "html_url": row[5], "created_at": row[6], } def public_app_status(conn): """公开状态视图：不含 secret/pem，只给配置与否 + 身份信息。""" config = get_app_config(conn) if config is None: return {"configured": False, "app_id": None, "slug": None, "html_url": None} return {"configured": True, "app_id": config["app_id"], "slug": config["slug"], "html_url": config["html_url"]} def _ensure_installations_table(conn): conn.execute( "CREATE TABLE IF NOT EXISTS github_app_installations(" "installation_id TEXT, account_login TEXT, " "repo_full_name TEXT, created_at REAL, " "UNIQUE(installation_id, repo_full_name))" ) conn.commit() def record_installation_repos(conn, installation_id, account_login, repos): """记安装的仓库行（按 (installation_id, repo) 去重，多次 同步不翻倍）。repos 为 full_name 字符串列表。""" _ensure_installations_table(conn) now = time.time() for repo in repos or []: conn.execute( "INSERT OR IGNORE INTO github_app_installations(" "installation_id, account_login, repo_full_name, created_at)" " VALUES(?, ?, ?, ?)", (str(installation_id), account_login, repo, now), ) conn.commit() def remove_installation_repos(conn, installation_id, repos): """删安装的部分仓库行（removed 事件用）。""" _ensure_installations_table(conn) for repo in repos or []: conn.execute( "DELETE FROM github_app_installations" " WHERE installation_id = ? AND repo_full_name = ?", (str(installation_id), repo), ) conn.commit() def remove_installation(conn, installation_id): """删整个安装的全部仓库行（deleted / suspended 后的清理用， 返回删掉的行数）。""" _ensure_installations_table(conn) cur = conn.execute( "DELETE FROM github_app_installations WHERE installation_id = ?", (str(installation_id),), ) conn.commit() return cur.rowcount def list_installations(conn): """列出全部安装行（按 installation_id / repo 排序）。""" _ensure_installations_table(conn) rows = conn.execute( "SELECT installation_id, account_login, repo_full_name, created_at" " FROM github_app_installations" " ORDER BY installation_id, repo_full_name", ).fetchall() return [{"installation_id": r[0], "account_login": r[1], "repo_full_name": r[2], "created_at": r[3]} for r in rows] def parse_installation_event(payload): """解析 installation 事件，返回动作 dict 或 None。 只认 action 在 created / deleted / added / removed / new_permissions_accepted 且带 installation 的 payload；其它 事件返回 None（调用方继续走原有解析）。repositories 取 full_name：created 用 payload repositories，added 用 repositories_added，removed 用 repositories_removed。 """ if not isinstance(payload, dict): return None action = payload.get("action") if action not in ("created", "deleted", "added", "removed", "new_permissions_accepted"): return None installation = payload.get("installation") if not isinstance(installation, dict): return None account = installation.get("account") or {} def names(key): out = [] for repo in payload.get(key) or []: if isinstance(repo, dict) and repo.get("full_name"): out.append(repo["full_name"]) return out if action == "added": repos = names("repositories_added") elif action == "removed": repos = names("repositories_removed") elif action == "deleted": repos = [] else: repos = names("repositories") return { "action": action, "installation_id": installation.get("id"), "account_login": account.get("login"), "repos": repos, } def claim_fix_review(conn, repo, pr_number, comment_id): """ /fix-review 内容去重：同一仓库 + PR + 评论只建一次任务。 手动 webhook 与 App webhook 共存时，同一评论会送两遍， delivery_id 不同、内容相同——delivery 去重拦不住，内容去重 才能拦。返回 (is_new, task_id)：首次认领 True；已认领过 返回 False 与当初建的 task_id（可能为 None：认领方还没来 得及把 task_id 写回，任务正在建）。comment_id 为 None 时 无法去重，直接放行（不阻塞合法任务）。 """ conn.execute( "CREATE TABLE IF NOT EXISTS github_fix_review_dedup(" "repo TEXT, pr_number INTEGER, comment_id TEXT, " "task_id TEXT, created_at REAL, " "UNIQUE(repo, pr_number, comment_id))" ) if comment_id is None: return True, None cur = conn.execute( "INSERT OR IGNORE INTO github_fix_review_dedup(" "repo, pr_number, comment_id, task_id, created_at)" " VALUES(?, ?, ?, NULL, ?)", (repo, pr_number, str(comment_id), time.time()), ) conn.commit() if cur.rowcount > 0: return True, None row = conn.execute( "SELECT task_id FROM github_fix_review_dedup" " WHERE repo = ? AND pr_number = ? AND comment_id = ?", (repo, pr_number, str(comment_id)), ).fetchone() return False, row[0] if row else None def set_fix_review_task_id(conn, repo, pr_number, comment_id, task_id): """认领后把 task_id 写回内容去重行（供重复送达时回读）。""" if comment_id is None: return conn.execute( "UPDATE github_fix_review_dedup SET task_id = ?" " WHERE repo = ? AND pr_number = ? AND comment_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"），两者都支持。 _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