75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""GitHub OAuth 提供方对接（授权码模式）。
|
|
|
|
HTTP 调用独立成模块函数，便于测试替身（monkeypatch）注入。
|
|
"""
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger("plugins.auth_oauth.github")
|
|
|
|
_TOKEN_URL = "https://github.com/login/oauth/access_token"
|
|
_USER_API = "https://api.github.com/user"
|
|
_USER_EMAILS_API = "https://api.github.com/user/emails"
|
|
|
|
|
|
async def exchange_code(client_id: str, client_secret: str, code: str,
|
|
redirect_uri: str) -> str:
|
|
"""授权码换 access_token，失败抛 RuntimeError（业务语义，调用方转 fail 响应）。"""
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
resp = await client.post(
|
|
_TOKEN_URL,
|
|
json={
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
data = resp.json()
|
|
if "access_token" not in data:
|
|
detail = data.get("error_description") or data.get("error") or data
|
|
raise RuntimeError(f"GitHub 授权码换取失败: {detail}")
|
|
return data["access_token"]
|
|
|
|
|
|
async def fetch_user(access_token: str) -> dict:
|
|
"""拉取 GitHub 用户并归一化为建号 profile。
|
|
|
|
Returns:
|
|
{uid, username, email, full_name, avatar_url}
|
|
Raises:
|
|
RuntimeError: API 失败或拿不到邮箱（users.email 非空唯一，必须可得）。
|
|
"""
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Accept": "application/vnd.github+json",
|
|
}
|
|
resp = await client.get(_USER_API, headers=headers)
|
|
if resp.status_code != 200:
|
|
raise RuntimeError(f"GitHub 用户信息获取失败: HTTP {resp.status_code}")
|
|
u = resp.json()
|
|
|
|
email = u.get("email")
|
|
if not email:
|
|
# 私密邮箱：走 /user/emails 取主邮箱
|
|
er = await client.get(_USER_EMAILS_API, headers=headers)
|
|
if er.status_code == 200 and isinstance(er.json(), list):
|
|
emails = er.json()
|
|
primary = next(
|
|
(e for e in emails if e.get("primary")), emails[0] if emails else None
|
|
)
|
|
email = primary.get("email") if primary else None
|
|
if not email:
|
|
raise RuntimeError("GitHub 账号未提供可用邮箱，无法创建账号")
|
|
|
|
return {
|
|
"uid": str(u["id"]),
|
|
"username": u.get("login") or f"github_{u['id']}",
|
|
"email": email,
|
|
"full_name": u.get("name"),
|
|
"avatar_url": u.get("avatar_url"),
|
|
}
|