oauth: bind existing accounts to external identities

New external_identities table (UNIQUE(provider, external_id),
UNIQUE(provider, user_id)) + bind/initiate, bindings list/unbind
endpoints. Callback branches on bind_user_id in state: bind success
shows no key and no user id; conflicts are 409
(identity_bound_to_other / user_already_bound); repeat bind to the
same user+identity is idempotent. Login resolution checks bindings
first: a bound identity lands in the bound account and rotates its
key like any OAuth login; unbound identities keep the deterministic
oauth:<provider>:<login> mapping. Unbind restores oauth mapping.
Tests: new cloud/tests/test_oauth_bind.py (11). Full suites: host
29 OK, cloud 389 OK (4 pre-existing skips).

cloud/README.md

@@ -139,8 +139,25 @@ client_secret（只走环境变量），不是无认证后门。
   已存在用户每次 OAuth 登录都会轮换 key（旧 key 立即失效），
   因此不要把旧 key 存到多处长期复用；key 明文只在回调成功页/
   响应里出现一次，丢了只能重新走一次 OAuth 登录拿新 key。
-- 仍缺（非部署前置，但同样未完成）：已有本站账号的绑定流程
-  （目前是确定性用户名新建/复用，不做账号绑定）。
+- 已有本站账号的绑定流程（已实现）：持本站 key 调
+  `POST /oauth/{provider}/bind/initiate`（body 含 client_id +
+  redirect_uri），返回 `{provider, state, authorization_url}`，
+  state 里带 bind_user_id（10 分钟过期、一次性）；用户授权后
+  回调走绑定分支（GET 回调成功页只显示 provider + 外部 login，
+  绝不显示任何 key、也不显示本站 user_id；POST 回调返回
+  `{linked: true, provider, login, external_id, user_id,
+  already_bound}`，同样不含 key）。绑定表
+  `external_identities` 有两个唯一约束：一个外部身份只能绑一个
+  本站账号、一个本站账号每个 provider 只能绑一个外部身份，
+  冲突返回 409（`identity_bound_to_other` /
+  `user_already_bound`），同一用户 + 同一身份重复绑是幂等的。
+  登录解析顺序：先查绑定表——绑过的身份落到被绑账号并按 OAuth
+  登录规矩轮换 key；没绑过才走确定性 oauth 账号映射。以前
+  GitHub 登录建的 oauth 账号不会被绑定删掉：绑完后该身份登录
+  改落到被绑账号，旧 oauth 账号成孤儿（key 仍可用）；解绑
+  （`DELETE /oauth/{provider}/bindings`，须持本站 key，本来就
+  没有返回 404）后恢复原来的 oauth 账号登录。查看自己的绑定：
+  `GET /oauth/{provider}/bindings`（须持本站 key，只看得到自己的）。
 
 ### 推送凭证清单
 

cloud/oauth.py

@@ -16,8 +16,15 @@
 - 本模块不存储 access_token：回调成功后只返回外部身份信息，
   access_token 在提取身份后即丢弃。如后续需要用 token 调 provider
   API，需另设计加密存储，本模块不提供。
-- 本模块不做已有本站账号的绑定：映射仍是确定性用户名
-  ``oauth:<provider>:<login>`` 新建/复用，绑定流程需另加。
+- 已有本站账号的绑定（本模块提供，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 回调地址。
 
@@ -79,6 +86,19 @@ class OAuthError(Exception):
         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)
@@ -95,7 +115,8 @@ class OAuthStateStore:
     """基于 SQLite 的 OAuth state 存储。
 
     表 oauth_states：state（主键，随机串）、provider、redirect_uri、
-    created_at（epoch 秒）。与 AuthStore 的 users / project_members /
+    created_at（epoch 秒）、client_id、bind_user_id（绑定流程才有，
+    登录流程为空）。与 AuthStore 的 users / project_members /
     audit_log 表名不冲突，可共用同一个 db 文件。
     """
 
@@ -108,21 +129,25 @@ class OAuthStateStore:
             "provider TEXT, "
             "redirect_uri TEXT, "
             "client_id TEXT, "
+            "bind_user_id TEXT, "
             "created_at REAL)"
         )
-        # 老库迁移：脚手架版的表没有 client_id 列，缺了就补上。
-        # client_id 随 state 入库，是真实浏览器流程的关键：provider
-        # 重定向只带回 code + state，回调靠 state 记录取回 client_id
-        # 才能换 token；补列是 no-op（已有的库再跑一次不变）。
+        # 老库迁移：脚手架版的表没有 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):
+                     client_id=None, bind_user_id=None):
         """生成并入库一个随机 state，返回 state 字符串。
 
         ``ttl_seconds`` 只影响后续 ``validate_state`` 的过期判定，
@@ -131,6 +156,8 @@ class OAuthStateStore:
         创建时一致，默认两边都是 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):
@@ -138,9 +165,10 @@ class OAuthStateStore:
         state = secrets.token_urlsafe(32)
         self.conn.execute(
             "INSERT INTO oauth_states(state, provider, redirect_uri,"
-            " client_id, created_at)"
-            " VALUES(?, ?, ?, ?, ?)",
-            (state, provider, redirect_uri, client_id, time.time()),
+            " client_id, bind_user_id, created_at)"
+            " VALUES(?, ?, ?, ?, ?, ?)",
+            (state, provider, redirect_uri, client_id, bind_user_id,
+             time.time()),
         )
         self.conn.commit()
         return state
@@ -166,13 +194,14 @@ class OAuthStateStore:
         if now is None:
             now = time.time()
         row = self.conn.execute(
-            "SELECT state, provider, redirect_uri, client_id, created_at"
+            "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, created_at = row
+        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:
@@ -190,6 +219,7 @@ class OAuthStateStore:
             "provider": stored_provider,
             "redirect_uri": stored_redirect_uri,
             "client_id": stored_client_id,
+            "bind_user_id": stored_bind_user_id,
             "created_at": created_at,
         }
 
@@ -445,9 +475,11 @@ def handle_oauth_callback(state_store, provider, code, state, client_id,
                           ttl_seconds=STATE_TTL_SECONDS):
     """处理 provider 回调：验 state -> 换 token -> 返回外部身份。
 
-    成功返回 dict：``{"provider", "external_id", "login", "email"}``
+    成功返回 dict：``{"provider", "external_id", "login", "email",
+    "bind_user_id"}``
     （缺失的字段为 None；external_id 至少与 login/email 其一存在，
-    否则抛 OAuthError）。
+    否则抛 OAuthError；``bind_user_id`` 非空表示这次回调是绑定
+    流程，调用方须走绑定分支、绝不走登录映射）。
 
     失败一律抛 OAuthError，不建任何会话、不写任何用户或审计记录：
     - state 不存在 -> reason=state_not_found；
@@ -484,13 +516,13 @@ def handle_oauth_callback(state_store, provider, code, state, client_id,
     if now is None:
         now = time.time()
     peek = state_store.conn.execute(
-        "SELECT provider, redirect_uri, client_id, created_at"
+        "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_created_at = peek
+    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:
@@ -531,6 +563,7 @@ def handle_oauth_callback(state_store, provider, code, state, client_id,
             and identity["email"] is None:
         raise OAuthError(
             "empty external identity", reason="identity_missing")
+    identity["bind_user_id"] = peek_bind_user_id
     return identity
 
 
@@ -542,9 +575,8 @@ def map_external_identity_to_local_user(auth_store, provider, identity):
     ``AuthStore.create_user`` 创建并返回 ``(user_id, True)``——
     创建时生成的 API key 明文按 auth.py 约定只应返回给调用方一次，
     本函数不返回它（调用方如需发 key 给登录用户，须另行决定
-    发放方式，本模块不代发）。这只是脚手架的默认映射：生产若要
-    把外部身份绑定到已有本站账号（而非每次新建），需另加绑定表
-    与绑定流程，本函数不假装已做绑定。
+    发放方式，本模块不代发）。这是确定性映射（无绑定时用）：
+    已有本站账号的绑定走下面的绑定表与绑定流程，不走本函数。
     """
     _provider_config(provider)
     login = identity.get("login") or identity.get("external_id") \
@@ -595,3 +627,121 @@ def map_external_identity_to_local_user_with_key(auth_store, provider,
         return user_id, True, api_key_plain, False
     api_key_plain = auth_store.rotate_api_key(row[0])
     return row[0], False, api_key_plain, True
+
+
+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

cloud/server.py

@@ -65,12 +65,18 @@ from cloud.workers import (
     list_workers as _registry_list,
 )
 from cloud.oauth import (
+    BindError,
     OAuthError,
     OAuthStateStore,
+    bind_external_identity,
+    build_authorization_url,
     default_userinfo_client,
+    find_bound_user,
     handle_oauth_callback,
+    list_bindings,
     map_external_identity_to_local_user_with_key,
     start_oauth_login,
+    unbind_external_identity,
 )
 from cloud.repos import validate_repo_url
 from cloud.push import (
@@ -528,6 +534,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             # OAuth 登录/回调是 pre-auth 端点（用户还没本站 API key，
             # 这正是 OAuth 要解决的问题），靠 state + client_secret
             # 认证，不走 X-API-Key；不要把它当成无认证后门。
+            # 绑定端点例外：绑定必须知道绑到哪个已有账号，所以在
+            # 处理器内验 X-API-Key（无 key 401），分发仍走这支。
+            bind = self._oauth_bind_route(path)
+            if bind is not None:
+                provider, action = bind
+                if method == "POST" and action == "bind_initiate":
+                    self._oauth_bind_initiate(provider)
+                elif method == "GET" and action == "bindings":
+                    self._oauth_bindings_get(provider)
+                elif method == "DELETE" and action == "bindings":
+                    self._oauth_bindings_delete(provider)
+                else:
+                    self._send_json(404, {"error": "not found"})
+                return
             if method == "GET" and path.endswith("/login"):
                 self._oauth_login(path)
             elif method == "GET" and path.endswith("/callback"):
@@ -1732,13 +1752,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         return http_client, userinfo_client
 
     def _oauth_finish(self, provider, code, state, client_id, redirect_uri):
-        """回调共用后半段：验 state -> 换 token -> 取身份 -> 发本站 key。
+        """回调共用后半段：验 state -> 换 token -> 取身份 -> 分叉。
 
-        成功返回 (local_id, created, api_key_plain, rotated, identity)；
+        成功返回 6 元组
+        ``(local_id, created, api_key_plain, rotated, identity, is_bind)``；
         失败抛 OAuthError（调用方按 JSON/HTML 分别呈现，绝不把
-        access_token、client_secret、旧 API key 写进任何响应）。
+        access_token、client_secret、旧 API key 写进任何响应），
+        绑定冲突抛 BindError（调用方按 409 呈现）。
         client_id / redirect_uri 为 None 时由 handle_oauth_callback
         从 state 记录取回（真实浏览器回调只带 code + state）。
+
+        分叉规则（顺序即优先级）：
+        1. state 带 bind_user_id -> 绑定分支：把身份绑到该本站
+           账号，返回 is_bind=True、api_key_plain=None（绑定成功页
+           绝不显示任何 key）；冲突抛 BindError；
+        2. 身份已绑定到某本站账号 -> 登录落到被绑账号，并按
+           OAuth 登录规矩轮换 key（created=False、rotated=True），
+           与 oauth 账号登录同一语义，不发明新语义；
+        3. 否则走原来的确定性 oauth 账号映射（新建/复用 + 轮换）。
         """
         store = OAuthStateStore(self.server.db_path)
         try:
@@ -1752,6 +1783,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             except Exception:
                 pass
         with _auth_store(self.server.db_path) as auth:
+            bind_user_id = identity.get("bind_user_id")
+            if bind_user_id:
+                binding = bind_external_identity(
+                    auth, provider, identity, bind_user_id)
+                auth.audit(bind_user_id, "", "bind_external",
+                           "%s:%s" % (provider, identity.get("external_id")))
+                identity["already_bound"] = binding["already"]
+                return (bind_user_id, False, None, False, identity, True)
+            bound_user_id = find_bound_user(auth, provider, identity)
+            if bound_user_id is not None:
+                api_key_plain = auth.rotate_api_key(bound_user_id)
+                auth.audit(bound_user_id, "", "oauth_login",
+                           "%s:%s" % (provider, identity.get("external_id")))
+                auth.audit(bound_user_id, "", "api_key_rotated", provider)
+                return (bound_user_id, False, api_key_plain, True, identity,
+                        False)
             local_id, created, api_key_plain, rotated = \
                 map_external_identity_to_local_user_with_key(
                     auth, provider, identity)
@@ -1759,7 +1806,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                        "%s:%s" % (provider, identity.get("external_id")))
             if rotated:
                 auth.audit(local_id, "", "api_key_rotated", provider)
-        return local_id, created, api_key_plain, rotated, identity
+        return local_id, created, api_key_plain, rotated, identity, False
 
     def _oauth_callback(self, path):
         provider = self._oauth_provider(path)
@@ -1778,12 +1825,26 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._send_json(400, {"error": "code and state required"})
             return
         try:
-            local_id, created, api_key_plain, rotated, identity = \
+            local_id, created, api_key_plain, rotated, identity, is_bind = \
                 self._oauth_finish(provider, code, state, client_id,
                                    redirect_uri)
+        except BindError as e:
+            self._send_json(409, {"error": str(e), "reason": e.reason})
+            return
         except OAuthError as e:
             self._send_json(400, {"error": str(e), "reason": e.reason})
             return
+        if is_bind:
+            # 绑定成功：绝不返回任何 key（绑定页的 completer 未必是
+            # 发起人，key 一旦上页就可能落到别人手里）；发起人用
+            # 自己的 key 调 GET /oauth/{provider}/bindings 核对。
+            self._send_json(200, {"linked": True, "provider": provider,
+                                  "login": identity.get("login"),
+                                  "external_id": identity.get("external_id"),
+                                  "user_id": local_id,
+                                  "already_bound":
+                                      identity.get("already_bound", False)})
+            return
         # api_key 明文只在这里出现一次：调用方须立即保存，
         # 之后任何接口都不再返回它；轮换时旧 key 已失效。
         self._send_json(200, {"user_id": local_id, "created": created,
@@ -1838,6 +1899,26 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                html.escape(str(api_key_plain)),
                " Your old key has expired (rotated)." if rotated else ""))
 
+    def _oauth_bind_success_page(self, provider, login):
+        """绑定成功页：只显示 provider + 外部 login + 成功提示。
+
+        绝不显示任何 key，也不显示本站 user_id：回调页的 completer
+        未必是绑定发起人，key 与内部 id 一旦上页就可能落到别人
+        手里；发起人用自己的 key 调 GET /oauth/{provider}/bindings
+        核对绑定。
+        """
+        return (
+            "<!doctype html><html><head><meta charset='utf-8'>"
+            "<title>OAuth binding succeeded</title></head><body>"
+            "<h1>OAuth binding succeeded</h1>"
+            "<p>provider: %s</p>"
+            "<p>login: %s</p>"
+            "<p>Your external identity is now bound to your account.</p>"
+            "<p>You can verify it from your bindings list.</p>"
+            "</body></html>"
+            % (html.escape(str(provider)),
+               html.escape(str(login))))
+
     def _oauth_callback_get(self, path):
         provider = self._oauth_provider(path)
         if provider is None:
@@ -1854,17 +1935,113 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 "missing_params", "code and state required"))
             return
         try:
-            local_id, created, api_key_plain, rotated, identity = \
+            local_id, created, api_key_plain, rotated, identity, is_bind = \
                 self._oauth_finish(provider, code, state, client_id,
                                    redirect_uri)
+        except BindError as e:
+            self._send_html(409, self._oauth_error_page(
+                e.reason or "bind_conflict", str(e)))
+            return
         except OAuthError as e:
             self._send_html(400, self._oauth_error_page(
                 e.reason or "oauth_failed", str(e)))
             return
+        if is_bind:
+            self._send_html(200, self._oauth_bind_success_page(
+                provider, identity.get("login")))
+            return
         self._send_html(200, self._oauth_success_page(
             provider, local_id, created, identity.get("login"),
             api_key_plain, rotated))
 
+    # -- OAuth 绑定（已有账号绑外部身份）------------------------------
+    def _oauth_bind_route(self, path):
+        """绑定相关路径解析：返回 (provider, action) 或 None。
+
+        POST /oauth/{provider}/bind/initiate -> (provider, "bind_initiate")
+        GET|DELETE /oauth/{provider}/bindings -> (provider, "bindings")
+        其它 /oauth/ 路径返回 None，走原来的登录/回调分发。
+        """
+        parts = [p for p in path.split("/") if p]
+        if len(parts) == 4 and parts[0] == "oauth" \
+                and parts[2] == "bind" and parts[3] == "initiate":
+            return parts[1], "bind_initiate"
+        if len(parts) == 3 and parts[0] == "oauth" \
+                and parts[2] == "bindings":
+            return parts[1], "bindings"
+        return None
+
+    def _oauth_bind_initiate(self, provider):
+        """绑定起点：须持本站 key（与登录不同，绑定必须知道绑到谁）。
+
+        body 须含 client_id + redirect_uri（与登录同款）；state 里
+        带 bind_user_id，回调时靠它走绑定分支。只返回授权 URL，
+        不直接跳转；跳转由调用方决定。
+        """
+        user_id = self._authenticate()
+        if user_id is None:
+            self._send_json(401, {"error": "unauthorized"})
+            return
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        client_id = body.get("client_id")
+        redirect_uri = body.get("redirect_uri")
+        scope = body.get("scope")
+        if not client_id or not redirect_uri:
+            self._send_json(400, {"error": "client_id and redirect_uri required"})
+            return
+        store = OAuthStateStore(self.server.db_path)
+        try:
+            state = store.create_state(
+                provider, redirect_uri, client_id=client_id,
+                bind_user_id=user_id)
+            url = build_authorization_url(
+                provider, client_id, redirect_uri, scope=scope, state=state)
+        except OAuthError as e:
+            self._send_json(400, {"error": str(e), "reason": e.reason})
+            return
+        except ValueError as e:
+            self._send_json(400, {"error": str(e)})
+            return
+        finally:
+            try:
+                store.conn.close()
+            except Exception:
+                pass
+        self._send_json(200, {"provider": provider, "state": state,
+                              "authorization_url": url})
+
+    def _oauth_bindings_get(self, provider):
+        """列出调用方自己的绑定（只看得到自己的，别人的看不到）。"""
+        user_id = self._authenticate()
+        if user_id is None:
+            self._send_json(401, {"error": "unauthorized"})
+            return
+        with _auth_store(self.server.db_path) as auth:
+            bindings = list_bindings(auth, user_id)
+        self._send_json(200, {"provider": provider, "bindings": bindings})
+
+    def _oauth_bindings_delete(self, provider):
+        """解绑调用方自己在该 provider 上的绑定：本来就没有返回 404。
+
+        解绑后该身份登录恢复原来的 oauth 账号映射（旧 oauth 账号
+        若存在，key 仍可用，不会被解绑删掉）。
+        """
+        user_id = self._authenticate()
+        if user_id is None:
+            self._send_json(401, {"error": "unauthorized"})
+            return
+        with _auth_store(self.server.db_path) as auth:
+            ok = unbind_external_identity(auth, provider, user_id)
+            if ok:
+                auth.audit(user_id, "", "unbind_external", provider)
+        if not ok:
+            self._send_json(404, {"error": "no binding for provider"})
+            return
+        self._send_json(200, {"unbound": True, "provider": provider})
+
     # -- 推送订阅与分发 ---------------------------------------------------
     def _push_create_endpoint(self, user_id, project_id):
         body, err = self._read_json()

cloud/tests/test_oauth_bind.py

@@ -0,0 +1,303 @@
+"""OAuth 绑定单测：已有本站账号绑外部身份 + 登录解析到绑定账号。
+
+不联网、不向真实 GitHub/Google 发请求：server 的注入点
+（server.oauth_http_client / server.oauth_userinfo_client）挂 fake；
+测试里出现的 token、key 一律是 fake 占位，绝无真实 secret、
+真实 token、真实 GitHub 账号。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.db import connect as _db_connect
+    from cloud.oauth import (
+        BindError,
+        OAuthStateStore,
+        bind_external_identity,
+        find_bound_user,
+        list_bindings,
+        unbind_external_identity,
+    )
+    from cloud.server import create_server
+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.oauth import (
+        BindError,
+        OAuthStateStore,
+        bind_external_identity,
+        find_bound_user,
+        list_bindings,
+        unbind_external_identity,
+    )
+    from cloud.server import create_server
+
+
+def _fake_token_client():
+    def fake(token_url, payload):
+        return {"access_token": "fake-token-not-real"}
+    return fake
+
+
+def _fake_userinfo(user_id=4242, login="gh-user"):
+    def fake(access_token):
+        return {"id": user_id, "login": login,
+                "email": "%s@example.invalid" % login}
+    return fake
+
+
+class BindFunctionTestCase(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.user_a, _ = self.auth.create_user("user-a")
+        self.user_b, _ = self.auth.create_user("user-b")
+
+    def tearDown(self):
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _ident(self, external_id=4242, login="gh-user"):
+        return {"provider": "github", "external_id": str(external_id),
+                "login": login, "email": None}
+
+    def test_bind_happy_and_idempotent(self):
+        first = bind_external_identity(
+            self.auth, "github", self._ident(), self.user_a)
+        self.assertFalse(first["already"])
+        again = bind_external_identity(
+            self.auth, "github", self._ident(), self.user_a)
+        self.assertTrue(again["already"])
+        self.assertEqual(find_bound_user(
+            self.auth, "github", self._ident()), self.user_a)
+        self.assertEqual(len(list_bindings(self.auth, self.user_a)), 1)
+
+    def test_identity_bound_to_other_conflicts(self):
+        bind_external_identity(
+            self.auth, "github", self._ident(), self.user_a)
+        with self.assertRaises(BindError) as ctx:
+            bind_external_identity(
+                self.auth, "github", self._ident(), self.user_b)
+        self.assertEqual(ctx.exception.reason, "identity_bound_to_other")
+
+    def test_user_already_bound_conflicts(self):
+        bind_external_identity(
+            self.auth, "github", self._ident(), self.user_a)
+        with self.assertRaises(BindError) as ctx:
+            bind_external_identity(
+                self.auth, "github", self._ident(9999, "gh-user2"),
+                self.user_a)
+        self.assertEqual(ctx.exception.reason, "user_already_bound")
+
+    def test_unbind_then_rebind_to_other_user(self):
+        bind_external_identity(
+            self.auth, "github", self._ident(), self.user_a)
+        self.assertTrue(unbind_external_identity(
+            self.auth, "github", self.user_a))
+        self.assertFalse(unbind_external_identity(
+            self.auth, "github", self.user_a))
+        self.assertIsNone(find_bound_user(
+            self.auth, "github", self._ident()))
+        bind_external_identity(
+            self.auth, "github", self._ident(), self.user_b)
+        self.assertEqual(find_bound_user(
+            self.auth, "github", self._ident()), self.user_b)
+
+    def test_state_bind_user_id_round_trip_and_migration(self):
+        # 老表（无 bind_user_id 列）上初始化，列须被补上。
+        conn = _db_connect(self.db_path)
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS oauth_states("
+            "state TEXT PRIMARY KEY, provider TEXT, redirect_uri TEXT,"
+            " client_id TEXT, created_at REAL)")
+        conn.commit()
+        conn.close()
+        store = OAuthStateStore(self.db_path)
+        cols = {row[1] for row in store.conn.execute(
+            "PRAGMA table_info(oauth_states)").fetchall()}
+        self.assertIn("bind_user_id", cols)
+        state = store.create_state(
+            "github", "https://example.invalid/cb",
+            client_id="cid", bind_user_id=self.user_a)
+        record = store.validate_state(
+            state, "github", redirect_uri="https://example.invalid/cb")
+        self.assertEqual(record["bind_user_id"], self.user_a)
+        store.conn.close()
+
+
+class BindHttpTestCase(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.user_a, self.key_a = self.auth.create_user("user-a")
+        self.user_b, self.key_b = self.auth.create_user("user-b")
+        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.server.oauth_http_client = _fake_token_client()
+        self.server.oauth_userinfo_client = _fake_userinfo()
+
+    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)
+        try:
+            headers = {}
+            payload = None
+            if body is not None:
+                payload = json.dumps(body)
+                headers["Content-Type"] = "application/json"
+            if key is not None:
+                headers["X-API-Key"] = key
+            conn.request(method, path, body=payload, headers=headers)
+            resp = conn.getresponse()
+            raw = resp.read().decode()
+            try:
+                parsed = json.loads(raw)
+            except ValueError:
+                parsed = raw
+            return resp.status, parsed
+        finally:
+            conn.close()
+
+    def _login_state(self):
+        status, body = self._req(
+            "GET", "/oauth/github/login?client_id=cid"
+            "&redirect_uri=https://example.invalid/cb")
+        self.assertEqual(status, 200, body)
+        return body["state"]
+
+    def _bind_state(self, key):
+        status, body = self._req(
+            "POST", "/oauth/github/bind/initiate",
+            {"client_id": "cid",
+             "redirect_uri": "https://example.invalid/cb"}, key=key)
+        self.assertEqual(status, 200, body)
+        return body["state"]
+
+    def _callback(self, state):
+        return self._req(
+            "POST", "/oauth/github/callback",
+            {"code": "fake-code", "state": state})
+
+    def test_initiate_requires_auth_and_params(self):
+        status, _ = self._req(
+            "POST", "/oauth/github/bind/initiate",
+            {"client_id": "cid",
+             "redirect_uri": "https://example.invalid/cb"})
+        self.assertEqual(status, 401)
+        status, body = self._req(
+            "POST", "/oauth/github/bind/initiate",
+            {"client_id": "cid"}, key=self.key_a)
+        self.assertEqual(status, 400, body)
+
+    def test_bind_then_login_resolves_to_bound_user(self):
+        state = self._bind_state(self.key_a)
+        status, body = self._callback(state)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["linked"])
+        self.assertEqual(body["user_id"], self.user_a)
+        self.assertFalse(body["already_bound"])
+        self.assertNotIn("api_key", body)
+        status, body = self._req(
+            "GET", "/oauth/github/bindings", key=self.key_a)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(len(body["bindings"]), 1)
+        self.assertEqual(body["bindings"][0]["external_login"], "gh-user")
+        status, body = self._req(
+            "GET", "/oauth/github/bindings", key=self.key_b)
+        self.assertEqual(body["bindings"], [])
+        # 登录（无 bind 标记的 state）落到被绑账号并轮换 key。
+        status, body = self._callback(self._login_state())
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["user_id"], self.user_a)
+        self.assertTrue(body["api_key_rotated"])
+        new_key = body["api_key"]
+        status, _ = self._req(
+            "GET", "/oauth/github/bindings", key=self.key_a)
+        self.assertEqual(status, 401)
+        status, _ = self._req(
+            "GET", "/oauth/github/bindings", key=new_key)
+        self.assertEqual(status, 200)
+
+    def test_bind_conflict_identity_bound_to_other(self):
+        self._callback(self._bind_state(self.key_a))
+        status, body = self._callback(self._bind_state(self.key_b))
+        self.assertEqual(status, 409, body)
+        self.assertEqual(body["reason"], "identity_bound_to_other")
+
+    def test_bind_conflict_user_already_bound(self):
+        self._callback(self._bind_state(self.key_a))
+        self.server.oauth_userinfo_client = _fake_userinfo(
+            9999, "gh-user2")
+        status, body = self._callback(self._bind_state(self.key_a))
+        self.assertEqual(status, 409, body)
+        self.assertEqual(body["reason"], "user_already_bound")
+
+    def test_unbind_restores_oauth_mapping(self):
+        self._callback(self._bind_state(self.key_a))
+        status, body = self._req(
+            "DELETE", "/oauth/github/bindings", key=self.key_a)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["unbound"])
+        status, body = self._req(
+            "DELETE", "/oauth/github/bindings", key=self.key_a)
+        self.assertEqual(status, 404, body)
+        status, body = self._callback(self._login_state())
+        self.assertEqual(status, 200, body)
+        self.assertNotEqual(body["user_id"], self.user_a)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            username = conn.execute(
+                "SELECT username FROM users WHERE id = ?",
+                (body["user_id"],)).fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(username, "oauth:github:gh-user")
+
+    def test_get_callback_bind_page_shows_no_key_or_user_id(self):
+        state = self._bind_state(self.key_a)
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        try:
+            conn.request(
+                "GET", "/oauth/github/callback?code=fake-code&state=%s"
+                % state)
+            resp = conn.getresponse()
+            page = resp.read().decode()
+            status = resp.status
+        finally:
+            conn.close()
+        self.assertEqual(status, 200, page)
+        self.assertIn("gh-user", page)
+        self.assertNotIn("API key", page)
+        self.assertNotIn(self.user_a, page)
+        self.assertNotIn("<code>", page)
+
+
+if __name__ == "__main__":
+    unittest.main()