codex-app-server/cloud/oauth.py

"""OAuth 接线（GitHub / Google）。

真实浏览器登录已接通：server 层提供 ``GET /oauth/{provider}/callback``
（provider 重定向带回 code + state），回调验 state、换 token、
调真实 userinfo 取身份、映射到本站用户并签发本站 API key。
单测一律用注入的 fake http 客户端，不向真实 provider 发请求。

诚实边界（仍未做的部分，见 cloud/README.md 的 OAuth 一节）：
- 本模块不硬编码任何真实 client_id / client_secret：client_id 由
  登录请求方传入并随 state 一起入库（回调时从 state 记录取回，
  这样浏览器重定向只带 code + state 也能完成换 token），
  client_secret 只从环境变量读
  （GitHub：``GITHUB_OAUTH_CLIENT_SECRET``，
  Google：``GOOGLE_OAUTH_CLIENT_SECRET``），绝不写文件、绝不
  硬编码、绝不出现在授权 URL 或成功页里。
- 本模块不存储 access_token：回调成功后只返回外部身份信息，
  access_token 在提取身份后即丢弃。如后续需要用 token 调 provider
  API，需另设计加密存储，本模块不提供。
- 已有本站账号的绑定（本模块提供，server 层接线）：绑定表
  ``external_identities`` + ``bind_external_identity`` 等函数。
  绑定流程：已登录用户（持本站 key）调
  ``POST /oauth/{provider}/bind/initiate`` 拿授权 URL，state 里带
  ``bind_user_id``；回调时 state 带 bind 标记即走绑定分支，不走
  登录映射。登录解析顺序：先查绑定表（绑过的身份落到被绑账号，
  并按 OAuth 登录规矩轮换 key），没绑过才走确定性用户名
  ``oauth:<provider>:<login>`` 新建/复用。绑定成功页绝不显示任何
  key；解绑后恢复原来的 oauth 账号登录。
- 本模块不提供公网 https 入口：server 默认只绑 127.0.0.1，
  真实浏览器登录还需要在部署侧配公网可达的 https 回调地址。

流程（标准 OAuth2 授权码流程的接线点）：
    state 生成（OAuthStateStore.create_state，存库，默认 10 分钟过期）
      -> 授权 URL（build_authorization_url / start_oauth_login）
      -> 用户在 provider 处授权，provider 回调本站并带回 code + state
      -> 回调（handle_oauth_callback）：验 state -> 换 token
         -> 取外部身份 -> 返回身份字典（server 层再映射 + 记审计）

零第三方依赖，只用 Python 标准库。
"""

import json
import os
import secrets
from cloud.db import connect as _db_connect
import time
import urllib.parse
import urllib.request
from urllib.parse import parse_qs

# state 默认有效期：10 分钟。
STATE_TTL_SECONDS = 600

# 各 provider 的公开端点与默认 scope、client_secret 环境变量名。
# 这些是 OAuth 协议的公开端点地址，不是本站的任何凭证；真实应用
# 的 client_id / client_secret 仍需另行注册并经环境变量供给。
PROVIDERS = {
    "github": {
        "authorize_url": "https://github.com/login/oauth/authorize",
        "token_url": "https://github.com/login/oauth/access_token",
        "userinfo_url": "https://api.github.com/user",
        "default_scope": "read:user",
        "client_secret_env": "GITHUB_OAUTH_CLIENT_SECRET",
    },
    "google": {
        "authorize_url": "https://accounts.google.com/o/oauth2/v2/auth",
        "token_url": "https://oauth2.googleapis.com/token",
        "userinfo_url": "https://openidconnect.googleapis.com/v1/userinfo",
        "default_scope": "openid email profile",
        "client_secret_env": "GOOGLE_OAUTH_CLIENT_SECRET",
    },
}


class OAuthError(Exception):
    """OAuth 流程失败：state 无效/过期、换 token 失败、身份取不到等。

    ``reason`` 供调用方区分（例如记日志或返回不同的错误提示），
    取值：``unknown_provider`` / ``state_not_found`` / ``state_expired`` /
    ``state_mismatch`` / ``client_secret_missing`` / ``token_exchange_failed``
    / ``identity_missing`` / ``userinfo_failed`` / ``client_id_missing``。
    """

    def __init__(self, message, reason=None):
        super().__init__(message)
        self.reason = reason


class BindError(Exception):
    """绑定冲突：调用方按 409 呈现，绝不静默覆盖别人的绑定。

    reason 取值：``identity_bound_to_other``（该外部身份已被另
    一个本站账号绑定）/ ``user_already_bound``（该本站账号已绑定
    了同一 provider 的另一个外部身份，须先解绑）。
    """

    def __init__(self, message, reason=None):
        super().__init__(message)
        self.reason = reason


def _provider_config(provider):
    """取 provider 配置，未知 provider 抛 OAuthError（不猜端点）。"""
    config = PROVIDERS.get(provider)
    if config is None:
        raise OAuthError(
            "unknown provider: %r (supported: %s)"
            % (provider, ", ".join(sorted(PROVIDERS))),
            reason="unknown_provider",
        )
    return config


class OAuthStateStore:
    """基于 SQLite 的 OAuth state 存储。

    表 oauth_states：state（主键，随机串）、provider、redirect_uri、
    created_at（epoch 秒）、client_id、bind_user_id（绑定流程才有，
    登录流程为空）。与 AuthStore 的 users / project_members /
    audit_log 表名不冲突，可共用同一个 db 文件。
    """

    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 oauth_states("
            "state TEXT PRIMARY KEY, "
            "provider TEXT, "
            "redirect_uri TEXT, "
            "client_id TEXT, "
            "bind_user_id TEXT, "
            "created_at REAL)"
        )
        # 老库迁移：脚手架版的表没有 client_id / bind_user_id 列，
        # 缺了就补上。client_id 随 state 入库，是真实浏览器流程的
        # 关键；bind_user_id 随 state 入库，是绑定流程的关键
        # （回调时靠它知道这次回调是绑定到哪个本站账号）。
        cols = {row[1] for row in self.conn.execute(
            "PRAGMA table_info(oauth_states)").fetchall()}
        if "client_id" not in cols:
            self.conn.execute(
                "ALTER TABLE oauth_states ADD COLUMN client_id TEXT")
        if "bind_user_id" not in cols:
            self.conn.execute(
                "ALTER TABLE oauth_states ADD COLUMN bind_user_id TEXT")
        self.conn.commit()

    def create_state(self, provider, redirect_uri, ttl_seconds=STATE_TTL_SECONDS,
                     client_id=None, bind_user_id=None):
        """生成并入库一个随机 state，返回 state 字符串。

        ``ttl_seconds`` 只影响后续 ``validate_state`` 的过期判定，
        本函数不把过期时间单独存一列：判定统一用
        ``created_at + ttl_seconds``（validate 时传入的 ttl 须与
        创建时一致，默认两边都是 600 秒）。
        ``client_id`` 随 state 一起入库：真实浏览器回调只带
        code + state，回调方靠它取回登录时传入的 client_id。
        ``bind_user_id`` 非空时这个 state 是绑定流程的：回调方靠
        它知道绑定到哪个本站账号，绝不走登录映射。
        """
        _provider_config(provider)
        if not redirect_uri or not isinstance(redirect_uri, str):
            raise ValueError("redirect_uri required")
        state = secrets.token_urlsafe(32)
        self.conn.execute(
            "INSERT INTO oauth_states(state, provider, redirect_uri,"
            " client_id, bind_user_id, created_at)"
            " VALUES(?, ?, ?, ?, ?, ?)",
            (state, provider, redirect_uri, client_id, bind_user_id,
             time.time()),
        )
        self.conn.commit()
        return state

    def validate_state(self, state, provider, redirect_uri=None,
                       ttl_seconds=STATE_TTL_SECONDS, now=None):
        """校验 state，成功返回记录 dict，失败返回 None（绝不抛错）。

        校验规则（任一不满足即拒绝）：
        - state 在库里存在，否则 None（不存在）；
        - 记录的 provider 与传入一致，否则 None（不匹配，且不消耗
          该 state，避免一次错 provider 的尝试烧掉合法 state）；
        - redirect_uri 若传入，须与记录一致，否则 None（不匹配）；
        - ``now``（默认 time.time()）未超过
          ``created_at + ttl_seconds``，否则删除该行并返回 None（过期）。

        成功是“一次性”的：返回前删除该行，同一 state 第二次校验
        必然失败，防止重放。``now`` 参数只为单测注入时间，
        生产调用传默认即可。
        """
        if not state:
            return None
        if now is None:
            now = time.time()
        row = self.conn.execute(
            "SELECT state, provider, redirect_uri, client_id, bind_user_id,"
            " created_at"
            " FROM oauth_states WHERE state = ?",
            (state,),
        ).fetchone()
        if row is None:
            return None
        stored_state, stored_provider, stored_redirect_uri, stored_client_id, stored_bind_user_id, created_at = row
        if stored_provider != provider:
            return None
        if redirect_uri is not None and stored_redirect_uri != redirect_uri:
            return None
        if now > created_at + ttl_seconds:
            self.conn.execute(
                "DELETE FROM oauth_states WHERE state = ?", (state,))
            self.conn.commit()
            return None
        self.conn.execute(
            "DELETE FROM oauth_states WHERE state = ?", (state,))
        self.conn.commit()
        return {
            "state": stored_state,
            "provider": stored_provider,
            "redirect_uri": stored_redirect_uri,
            "client_id": stored_client_id,
            "bind_user_id": stored_bind_user_id,
            "created_at": created_at,
        }


def build_authorization_url(provider, client_id, redirect_uri, scope=None,
                            state=None):
    """拼授权 URL（纯字符串拼接，不发任何请求、不含任何 secret）。

    参数：
    - provider：``"github"`` 或 ``"google"``，未知抛 OAuthError；
    - client_id：调用方传入的真实应用 id，本模块不硬编码、不校验
      其真实性（注册见模块文档“真实 OAuth 仍缺什么”）；
    - redirect_uri：须与在 provider 处注册的回调地址一致；
    - scope：不传则用该 provider 的默认 scope（GitHub ``read:user``，
      Google ``openid email profile``）；
    - state：不传则现场随机生成一个（调用方若要校验，必须自己
      用 OAuthStateStore 存下它；推荐直接用 ``start_oauth_login``，
      它把生成 + 入库 + 拼 URL 一次做完）。

    返回的 URL 只含 client_id / redirect_uri / scope / state（Google
    另含 response_type=code），绝不含 client_secret。
    """
    config = _provider_config(provider)
    if not client_id or not isinstance(client_id, str):
        raise ValueError("client_id required")
    if not redirect_uri or not isinstance(redirect_uri, str):
        raise ValueError("redirect_uri required")
    if state is None:
        state = secrets.token_urlsafe(32)
    params = {
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "scope": scope or config["default_scope"],
        "state": state,
    }
    if provider == "google":
        params["response_type"] = "code"
    return config["authorize_url"] + "?" + urllib.parse.urlencode(params)


def start_oauth_login(state_store, provider, client_id, redirect_uri,
                      scope=None):
    """登录起点：生成并入库 state，再拼授权 URL。

    返回 ``(state, authorization_url)``：调用方把用户重定向到该 URL，
    provider 回调时带回的 state 交给 ``handle_oauth_callback`` 校验。
    client_id 随 state 一起入库，回调时从 state 记录取回。
    """
    state = state_store.create_state(provider, redirect_uri,
                                     client_id=client_id)
    url = build_authorization_url(
        provider, client_id, redirect_uri, scope=scope, state=state)
    return state, url


def _urllib_post(token_url, payload):
    """默认的真实换 token 实现（urllib POST，返回解析后的 dict）。

    只在调用方未注入 http_client 且配好 client_secret 时走到；
    GitHub 的 token 端点默认回表单编码，本函数先试 JSON、
    失败再按表单解析，两种都解析不出则返回空 dict（由调用方
    判为换 token 失败）。
    """
    data = urllib.parse.urlencode(payload).encode("utf-8")
    request = urllib.request.Request(
        token_url, data=data, method="POST",
        headers={
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded",
        },
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        raw = response.read().decode("utf-8", "replace")
    try:
        parsed = json.loads(raw)
        if isinstance(parsed, dict):
            return parsed
    except ValueError:
        pass
    form = parse_qs(raw)
    return {key: values[0] for key, values in form.items() if values}


def exchange_code_for_token(provider, code, client_id, redirect_uri,
                            http_client=None, client_secret=None):
    """用授权码换 token，返回 provider 的 token 响应 dict。

    http_client 注入约定（单测与未来真实接线共用同一签名）：
    ``http_client(token_url, payload_dict) -> dict``，其中 payload 含
    client_id / code / redirect_uri，Google 另含
    grant_type=authorization_code；client_secret 仅在有值时才放进
    payload。返回的 dict 须含 ``access_token``，否则视为换 token 失败；
    http_client 抛异常、或返回含 ``error`` 的 dict，同样视为失败。

    client_secret 供给：参数优先，未传则读该 provider 对应的环境变量
    （见 PROVIDERS 的 client_secret_env）。真实路径（http_client
    为 None）下仍无 secret 时，在发任何请求前就抛 OAuthError，
    绝不以空 secret 去撞真实端点。注入了 http_client 的测试路径
    允许 secret 缺省（fake 不需要真实 secret），生产调用不得依赖
    这一宽松分支。
    """
    config = _provider_config(provider)
    if not code or not isinstance(code, str):
        raise OAuthError("code required", reason="token_exchange_failed")
    if client_secret is None:
        client_secret = os.environ.get(config["client_secret_env"])
    if http_client is None and not client_secret:
        raise OAuthError(
            "client_secret not configured: set %s (never hard-code it)"
            % config["client_secret_env"],
            reason="client_secret_missing",
        )
    payload = {
        "client_id": client_id,
        "code": code,
        "redirect_uri": redirect_uri,
    }
    if provider == "google":
        payload["grant_type"] = "authorization_code"
    if client_secret is not None:
        payload["client_secret"] = client_secret
    client = http_client if http_client is not None else _urllib_post
    try:
        result = client(config["token_url"], payload)
    except Exception as exc:
        raise OAuthError(
            "token exchange failed: %s" % (exc,), reason="token_exchange_failed"
        ) from exc
    if not isinstance(result, dict) or result.get("error"):
        raise OAuthError(
            "token exchange failed: %r" % (result,),
            reason="token_exchange_failed",
        )
    if not result.get("access_token"):
        raise OAuthError(
            "token exchange returned no access_token",
            reason="token_exchange_failed",
        )
    return result


def _urllib_get_json(url, headers):
    """默认的真实 userinfo 实现（urllib GET，返回解析后的 dict）。

    非 2xx、非法 JSON、非 dict 一律抛 OAuthError（reason=userinfo_failed），
    调用方不得把失败静默当成匿名登录。
    """
    request = urllib.request.Request(url, method="GET", headers=headers)
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            raw = response.read().decode("utf-8", "replace")
            status = response.status
    except Exception as exc:
        raise OAuthError(
            "userinfo request failed: %s" % (exc,),
            reason="userinfo_failed",
        ) from exc
    if not (200 <= status < 300):
        raise OAuthError(
            "userinfo request returned HTTP %s" % status,
            reason="userinfo_failed",
        )
    try:
        parsed = json.loads(raw)
    except ValueError as exc:
        raise OAuthError(
            "userinfo response is not JSON", reason="userinfo_failed",
        ) from exc
    if not isinstance(parsed, dict):
        raise OAuthError(
            "userinfo response is not an object", reason="userinfo_failed")
    return parsed


def default_userinfo_client(provider):
    """返回该 provider 的真实 userinfo 调用（access_token -> dict）。

    GitHub 调 https://api.github.com/user（Bearer + User-Agent）取
    login/id；Google 调 https://openidconnect.googleapis.com/v1/userinfo
    取 sub/email/name。失败一律抛 OAuthError（reason=userinfo_failed）。
    """

    config = _provider_config(provider)
    url = config["userinfo_url"]

    def fetch(access_token):
        headers = {"Authorization": "Bearer %s" % access_token,
                   "Accept": "application/json"}
        if provider == "github":
            # GitHub 要求 User-Agent，否则可能直接 403。
            headers["User-Agent"] = "codex-app-server"
        return _urllib_get_json(url, headers)

    return fetch


def _identity_from_token_or_userinfo(provider, token_result, userinfo_client):
    """从 token 响应或 userinfo 回调里抽外部身份。

    顺序：先看 token 响应自带的身份字段（fake / 某些 provider 会带
    id、login、email），没有则调 userinfo_client(access_token)。
    userinfo_client 已传入但调用失败、返回非 dict、或无可用身份字段时，
    抛 OAuthError（reason=userinfo_failed），绝不静默当成匿名登录；
    只有 userinfo_client 本就没传时才返回 None（调用方再判为
    identity_missing）。真实 GitHub 的 token 响应不带身份，生产
    调用必须传 userinfo_client；真实 Google 的 id_token 解析同样
    走 userinfo_client 这一注入点，本模块不解析 JWT。
    """
    for source in (token_result,):
        external_id = source.get("id") or source.get("sub")
        login = source.get("login") or source.get("name")
        email = source.get("email")
        if external_id is not None or login is not None or email is not None:
            return {
                "provider": provider,
                "external_id": str(external_id) if external_id is not None else None,
                "login": login,
                "email": email,
            }
    if userinfo_client is None:
        return None
    try:
        info = userinfo_client(token_result["access_token"])
    except OAuthError:
        raise
    except Exception as exc:
        raise OAuthError(
            "userinfo request failed: %s" % (exc,),
            reason="userinfo_failed",
        ) from exc
    if not isinstance(info, dict):
        raise OAuthError(
            "userinfo response is not an object", reason="userinfo_failed")
    external_id = info.get("id") or info.get("sub")
    login = info.get("login") or info.get("name")
    email = info.get("email")
    if external_id is None and login is None and email is None:
        raise OAuthError(
            "userinfo returned no usable identity fields",
            reason="userinfo_failed",
        )
    return {
        "provider": provider,
        "external_id": str(external_id) if external_id is not None else None,
        "login": login,
        "email": email,
    }


def handle_oauth_callback(state_store, provider, code, state, client_id,
                          redirect_uri, http_client=None, userinfo_client=None,
                          client_secret=None, now=None,
                          ttl_seconds=STATE_TTL_SECONDS):
    """处理 provider 回调：验 state -> 换 token -> 返回外部身份。

    成功返回 dict：``{"provider", "external_id", "login", "email",
    "bind_user_id"}``
    （缺失的字段为 None；external_id 至少与 login/email 其一存在，
    否则抛 OAuthError；``bind_user_id`` 非空表示这次回调是绑定
    流程，调用方须走绑定分支、绝不走登录映射）。

    失败一律抛 OAuthError，不建任何会话、不写任何用户或审计记录：
    - state 不存在 -> reason=state_not_found；
    - state 过期 -> reason=state_expired（过期行在拒绝时被删除）；
    - state 与 provider / redirect_uri 不匹配 ->
      reason=state_mismatch；
    - 换 token 失败 -> reason=token_exchange_failed；
    - userinfo 已传但失败/无可用身份 -> reason=userinfo_failed；
    - 未传 userinfo_client 且 token 本身无身份 -> reason=identity_missing；
    - state 里没有 client_id 且调用方也没传 -> reason=client_id_missing。

    client_id / redirect_uri 未传时从 state 记录取回（见上文），
    因此真实浏览器回调只带 code + state 也能完成。

    本函数 deliberately 不做以下三件事，全部留给 server 层（签名
    与 auth.py 完全兼容，不发明新格式）：
    1. 映射到本站用户：可用 ``map_external_identity_to_local_user``
       （复用 AuthStore 的 users 表）或 server 层自己的映射；
    2. 建会话：本站首版会话就是 AuthStore 的 API key，OAuth 登录后
       是否发 key、发给谁，由 server 层决定，本模块不发；
    3. 记审计：server 层在映射出本站 user_id 后调用
       ``AuthStore.audit(user_id, project_id, "oauth_login",
       "%s:%s" % (provider, external_id))``——与 auth.py 现有
       audit(action, detail) 写法一致，project_id 为登录发生的
       项目上下文、无项目上下文时 server 层自定（例如空串），
       本模块不替 server 层选 project_id。

    access_token 在本函数内只用于取身份（经 userinfo_client），
    返回前即丢弃，本模块不存储、不返回它。
    """
    # 先区分拒绝原因（validate_state 本身只返回记录或 None）：
    # 不存在 / provider 或 redirect_uri 不匹配 / 过期各有 reason；
    # 过期行在此处删除（validate_state 的过期分支同样会删，幂等）。
    if now is None:
        now = time.time()
    peek = state_store.conn.execute(
        "SELECT provider, redirect_uri, client_id, bind_user_id, created_at"
        " FROM oauth_states WHERE state = ?",
        (state,),
    ).fetchone() if state else None
    if peek is None:
        raise OAuthError("unknown state", reason="state_not_found")
    peek_provider, peek_redirect_uri, peek_client_id, peek_bind_user_id, peek_created_at = peek
    # 真实浏览器回调只带 code + state：client_id / redirect_uri 未传
    # 时从 state 记录取回（登录时已随 state 入库）；显式传入的优先。
    if client_id is None:
        client_id = peek_client_id
    if redirect_uri is None:
        redirect_uri = peek_redirect_uri
    if peek_provider != provider or (
            redirect_uri is not None and peek_redirect_uri != redirect_uri):
        raise OAuthError("state does not match provider/redirect_uri",
                         reason="state_mismatch")
    if not client_id:
        raise OAuthError(
            "client_id not recorded for this state; please restart login",
            reason="client_id_missing")
    if now > peek_created_at + ttl_seconds:
        state_store.conn.execute(
            "DELETE FROM oauth_states WHERE state = ?", (state,))
        state_store.conn.commit()
        raise OAuthError("state expired", reason="state_expired")
    record = state_store.validate_state(
        state, provider, redirect_uri=redirect_uri, now=now,
        ttl_seconds=ttl_seconds)
    if record is None:
        raise OAuthError(
            "invalid state", reason="state_not_found")
    token_result = exchange_code_for_token(
        provider, code, client_id, redirect_uri,
        http_client=http_client, client_secret=client_secret)
    identity = _identity_from_token_or_userinfo(
        provider, token_result, userinfo_client)
    if identity is None:
        raise OAuthError(
            "cannot determine external identity: token response carries"
            " no identity fields and no userinfo_client was provided",
            reason="identity_missing",
        )
    if identity["external_id"] is None and identity["login"] is None \
            and identity["email"] is None:
        raise OAuthError(
            "empty external identity", reason="identity_missing")
    identity["bind_user_id"] = peek_bind_user_id
    return identity


def map_external_identity_to_local_user(auth_store, provider, identity):
    """把外部身份映射到本站用户（复用 AuthStore 的 users 表）。

    确定性用户名：``oauth:<provider>:<login 或 external_id>``。
    已存在则返回 ``(user_id, False)``，不存在则经
    ``AuthStore.create_user`` 创建并返回 ``(user_id, True)``——
    创建时生成的 API key 明文按 auth.py 约定只应返回给调用方一次，
    本函数不返回它（调用方如需发 key 给登录用户，须另行决定
    发放方式，本模块不代发）。这是确定性映射（无绑定时用）：
    已有本站账号的绑定走下面的绑定表与绑定流程，不走本函数。
    """
    _provider_config(provider)
    login = identity.get("login") or identity.get("external_id") \
        or identity.get("email")
    if login is None:
        raise OAuthError(
            "identity has no login/external_id/email to map from",
            reason="identity_missing",
        )
    username = "oauth:%s:%s" % (provider, login)
    row = auth_store.conn.execute(
        "SELECT id FROM users WHERE username = ?", (username,)
    ).fetchone()
    if row is not None:
        return row[0], False
    user_id, _api_key_plain = auth_store.create_user(username)
    return user_id, True


def map_external_identity_to_local_user_with_key(auth_store, provider,
                                                 identity):
    """映射到本站用户并签发本站 API key（明文只返回这一次）。

    返回 ``(user_id, created, api_key_plain, rotated)``：
    - 新用户：经 ``AuthStore.create_user`` 创建，返回创建时生成的
      primary key，``rotated`` 为 False；
    - 已存在用户：经 ``AuthStore.create_api_key`` 签发一个附加 key
      （label 为 ``oauth:<provider>``），旧 key（primary 与之前
      的附加 key）全部继续有效，``rotated`` 恒为 False；
    调用方须把明文 key 只在回调成功页/响应里展示一次，之后任何
    接口都不再返回它；审计由 server 层记（oauth_login 恒记，
    签发附加 key 时另记 api_key_issued）。``rotated`` 字段只为
    向后兼容保留，恒为 False——登录不再轮换任何 key。
    """
    _provider_config(provider)
    login = identity.get("login") or identity.get("external_id") \
        or identity.get("email")
    if login is None:
        raise OAuthError(
            "identity has no login/external_id/email to map from",
            reason="identity_missing",
        )
    username = "oauth:%s:%s" % (provider, login)
    row = auth_store.conn.execute(
        "SELECT id FROM users WHERE username = ?", (username,)
    ).fetchone()
    if row is None:
        # 新用户：直接返回创建时生成的 key，不再多轮换一次。
        user_id, api_key_plain = auth_store.create_user(username)
        return user_id, True, api_key_plain, False
    _key_id, api_key_plain = auth_store.create_api_key(
        row[0], "oauth:%s" % provider)
    return row[0], False, api_key_plain, False


def _binding_key(identity):
    """绑定键：external_id 优先（GitHub id / Google sub），没有才
    用 login；登录解析查绑定时用同一派生，键必须两边一致。"""
    return identity.get("external_id") or identity.get("login")


def ensure_external_identities_table(conn):
    """建绑定表（幂等）：provider + 外部键 + 外部 login + 本站
    user_id + 创建时间；两个唯一约束防一对多乱绑。"""
    conn.execute(
        "CREATE TABLE IF NOT EXISTS external_identities("
        "provider TEXT, "
        "external_id TEXT, "
        "external_login TEXT, "
        "user_id TEXT, "
        "created_at REAL, "
        "UNIQUE(provider, external_id), "
        "UNIQUE(provider, user_id))"
    )
    conn.commit()


def bind_external_identity(auth_store, provider, identity, user_id):
    """把外部身份绑到已有本站账号。

    冲突一律抛 BindError（调用方按 409 呈现），绝不静默覆盖：
    - 该外部身份已被另一个本站账号绑定 ->
      reason=identity_bound_to_other；
    - 该本站账号已绑定了同一 provider 的另一个外部身份 ->
      reason=user_already_bound（须先解绑）。
    同一用户 + 同一身份重复绑是幂等的：返回 already=True，不报错、
    不改行。成功返回绑定 dict（含 already=False）。
    """
    _provider_config(provider)
    key = _binding_key(identity)
    if key is None:
        raise BindError("identity has no external_id/login to bind",
                        reason="identity_missing")
    ensure_external_identities_table(auth_store.conn)
    row = auth_store.conn.execute(
        "SELECT user_id FROM external_identities"
        " WHERE provider = ? AND external_id = ?",
        (provider, key),
    ).fetchone()
    if row is not None:
        if row[0] == user_id:
            return {"provider": provider, "external_id": key,
                    "external_login": identity.get("login"),
                    "user_id": user_id, "already": True}
        raise BindError(
            "external identity already bound to another user",
            reason="identity_bound_to_other")
    row = auth_store.conn.execute(
        "SELECT external_id FROM external_identities"
        " WHERE provider = ? AND user_id = ?",
        (provider, user_id),
    ).fetchone()
    if row is not None:
        raise BindError(
            "user already bound to another external identity",
            reason="user_already_bound")
    auth_store.conn.execute(
        "INSERT INTO external_identities(provider, external_id,"
        " external_login, user_id, created_at)"
        " VALUES(?, ?, ?, ?, ?)",
        (provider, key, identity.get("login"), user_id, time.time()),
    )
    auth_store.conn.commit()
    return {"provider": provider, "external_id": key,
            "external_login": identity.get("login"),
            "user_id": user_id, "already": False}


def find_bound_user(auth_store, provider, identity):
    """登录解析用：该外部身份绑到了哪个本站账号，没有返回 None。

    键派生与绑定时同一规则（external_id 优先，否则 login）。
    """
    _provider_config(provider)
    key = _binding_key(identity)
    if key is None:
        return None
    ensure_external_identities_table(auth_store.conn)
    row = auth_store.conn.execute(
        "SELECT user_id FROM external_identities"
        " WHERE provider = ? AND external_id = ?",
        (provider, key),
    ).fetchone()
    return row[0] if row is not None else None


def list_bindings(auth_store, user_id):
    """列出某本站账号的全部绑定（按 provider 排序）。"""
    ensure_external_identities_table(auth_store.conn)
    rows = auth_store.conn.execute(
        "SELECT provider, external_id, external_login, created_at"
        " FROM external_identities WHERE user_id = ? ORDER BY provider",
        (user_id,),
    ).fetchall()
    return [{"provider": r[0], "external_id": r[1],
             "external_login": r[2], "created_at": r[3]} for r in rows]


def unbind_external_identity(auth_store, provider, user_id):
    """解绑某本站账号在某 provider 上的绑定：删掉返回 True，
    本来就没有返回 False。解绑后该身份登录恢复原来的 oauth
    账号映射（旧 oauth 账号若存在，key 仍可用）。"""
    _provider_config(provider)
    ensure_external_identities_table(auth_store.conn)
    cur = auth_store.conn.execute(
        "DELETE FROM external_identities"
        " WHERE provider = ? AND user_id = ?",
        (provider, user_id),
    )
    auth_store.conn.commit()
    return cur.rowcount > 0