admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fa4588a1198161d41a194a862c3927c895890bd9
codex-app-server/cloud/github_app.py
T
host 8fc4e27241 feat: GitHub App manifest 全量获取：仓库/组织/企业/账号四类最高档 
接上轮仓库最大权限：用户明确要求无论什么身份，取该身份
授权范围内的所有权限。manifest 现按四类全拉满（74 项权限、
39 个事件）：仓库补 codespaces_lifecycle_admin、repository_
creation、metadata 只读；组织 members 升 write 并补
organization_administration 等 15 项；企业补 8 项；账号补
blocking、email_addresses、gists、密钥类等 22 项。
身份边界写进注释：GitHub 只授予安装身份能给的那部分，个人
装不上组织/企业、企业装不上仓库/组织；换 token 不传子集，
token 本就是安装全量；企业安装另需 App 公开/内部
（GITHUB_APP_PUBLIC=1）。诚实边界：企业类键名未用真实
GitHub 建 App 实测，新建时表单会校验，某键被拒删键再建，
线上旧 App 不受影响。
2026-10-30 02:53:46 +08:00

821 lines
31 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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/<slug>/
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 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"），两者都支持。
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
Reference in new issue View Git Blame Copy Permalink