From 7b85b4a4fdd6f7619f339affe674b797450ceb42 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 26 Oct 2026 16:27:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=9A=20key=EF=BC=9AOAuth=20=E7=99=BB?= =?UTF-8?q?=E5=BD=95=E7=AD=BE=E5=8F=91=E9=99=84=E5=8A=A0=20key=20=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E8=BD=AE=E6=8D=A2=EF=BC=8Ckeys=20=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cloud/README.md | 21 ++- cloud/auth.py | 165 ++++++++++++++++++- cloud/oauth.py | 15 +- cloud/server.py | 126 ++++++++++++-- cloud/tests/test_api_keys.py | 235 +++++++++++++++++++++++++++ cloud/tests/test_oauth_bind.py | 7 +- cloud/tests/test_oauth_real_login.py | 12 +- 7 files changed, 544 insertions(+), 37 deletions(-) create mode 100644 cloud/tests/test_api_keys.py diff --git a/cloud/README.md b/cloud/README.md index bfd46cf..04d5b5c 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -135,10 +135,23 @@ client_secret（只走环境变量），不是无认证后门。 只绑 `127.0.0.1:8765`，没有公网入口、没有 TLS，这一步是部署 前置——需要在部署侧配反向代理 + TLS 证书并把公网域名指到 server，再把该域名下的回调地址填进 OAuth 应用的注册信息里。 -- 登录后 API key 的保存与轮换规则：新用户拿到创建时生成的 key； - 已存在用户每次 OAuth 登录都会轮换 key（旧 key 立即失效）， - 因此不要把旧 key 存到多处长期复用；key 明文只在回调成功页/ - 响应里出现一次，丢了只能重新走一次 OAuth 登录拿新 key。 +- 登录后 API key 的保存与多 key 规则：新用户拿到创建时生成的 + primary key；已存在用户每次 OAuth 登录签发一个附加 key + （label 为 `oauth:`），旧 key（primary 与之前的附加 + key）全部继续有效，不再轮换——登录不会打断已存得到处都是的 + 旧 key。响应里的 `api_key_rotated` 字段为向后兼容保留，恒为 + False；签发附加 key 时记审计 `api_key_issued`（不再记 + `api_key_rotated`）。key 明文只在回调成功页/响应里出现一次， + 丢了只能重新走一次 OAuth 登录拿新 key（旧 key 不受影响）。 +- API keys 管理（全部只操作认证用户自己的 key，须持本站 key）： + `GET /users/me/keys`（列出 primary（id 为 "legacy"）+ 附加 key， + 含 label、后 4 位提示、创建/最近使用/撤销时间，绝不返回明文 + 与哈希）、`POST /users/me/keys`（body 可含 label，签发附加 key， + 明文只在本次响应里出现一次）、`DELETE /users/me/keys/{id}` + （撤销一个 key，primary 用 id "legacy"，无此 key 或已撤销返回 + 404）、`POST /users/me/keys/revoke-others`（撤销除本次请求 key + 之外的全部自己的 key，返回撤销数）。附加 key 认证成功时更新 + last_used_at，但每小时最多写一次（认证是热路径）。 - 已有本站账号的绑定流程（已实现）：持本站 key 调 `POST /oauth/{provider}/bind/initiate`（body 含 client_id + redirect_uri），返回 `{provider, state, authorization_url}`， diff --git a/cloud/auth.py b/cloud/auth.py index d5d81e8..56aa537 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -85,10 +85,15 @@ class AuthStore: return user_id, api_key_plain def rotate_api_key(self, user_id): - """轮换用户的 API key，返回新明文 key（只返回这一次）。 + """轮换用户的 primary key，返回新明文 key（只返回这一次）。 - 旧 key 的哈希即被替换，轮换后旧 key 立即失效。 - 用户不存在时抛 ValueError（调用方不得用它凭空建用户）。 + 旧 primary key 的哈希即被替换，轮换后旧 primary 立即失效； + 附加 key（api_keys 表）不受影响。用户不存在时抛 ValueError + （调用方不得用它凭空建用户）。 + + 注意：OAuth 登录不再调本方法（登录改为签发附加 key，旧 key + 不失效，见 create_api_key）；本方法只留给“明确要轮换 + primary”的显式操作。 """ row = self.conn.execute( "SELECT id FROM users WHERE id = ?", (user_id,) @@ -104,15 +109,165 @@ class AuthStore: self.conn.commit() return api_key_plain + def _ensure_api_keys_table(self): + """建附加 key 表（幂等）：一行一个 key，只存哈希 + 后 4 位 + 提示，绝不存明文；revoked_at 非空即已撤销，认证时跳过。""" + self.conn.execute( + "CREATE TABLE IF NOT EXISTS api_keys(" + "id TEXT PRIMARY KEY, " + "user_id TEXT, " + "key_hash TEXT UNIQUE, " + "label TEXT, " + "hint TEXT, " + "created_at REAL, " + "last_used_at REAL, " + "revoked_at REAL)" + ) + self.conn.commit() + + def create_api_key(self, user_id, label): + """给用户签发一个附加 key，返回 (key_id, 明文 key)。 + + 明文只返回这一次；库里只存哈希与后 4 位提示。附加 key 与 + primary key（users.api_key_hash）互相独立：撤销、轮换一 + 方不影响另一方。用户不存在时抛 ValueError。 + """ + row = self.conn.execute( + "SELECT id FROM users WHERE id = ?", (user_id,) + ).fetchone() + if row is None: + raise ValueError("unknown user: %r" % (user_id,)) + self._ensure_api_keys_table() + for _ in range(5): + api_key_plain = secrets.token_urlsafe(32) + api_key_hash = hashlib.sha256( + api_key_plain.encode()).hexdigest() + exists = self.conn.execute( + "SELECT id FROM api_keys WHERE key_hash = ?", + (api_key_hash,), + ).fetchone() + if exists is None: + break + else: + raise RuntimeError("failed to generate a unique api key") + key_id = secrets.token_hex(8) + self.conn.execute( + "INSERT INTO api_keys(id, user_id, key_hash, label, hint," + " created_at, last_used_at, revoked_at)" + " VALUES(?, ?, ?, ?, ?, ?, NULL, NULL)", + (key_id, user_id, api_key_hash, label, + api_key_plain[-4:], time.time()), + ) + self.conn.commit() + return key_id, api_key_plain + def authenticate(self, api_key): - """用 API key 认证，成功返回 user_id，失败返回 None。""" + """用 API key 认证，成功返回 user_id，失败返回 None。 + + 先查 primary（users.api_key_hash），再查附加 key（api_keys + 表，未撤销的才算）。附加 key 认证成功时顺手更新 + last_used_at，但每小时最多写一次（认证是热路径，不为 + 每次请求都写库）。 + """ if not api_key: return None api_key_hash = hashlib.sha256(api_key.encode()).hexdigest() row = self.conn.execute( "SELECT id FROM users WHERE api_key_hash = ?", (api_key_hash,) ).fetchone() - return row[0] if row else None + if row: + return row[0] + self._ensure_api_keys_table() + row = self.conn.execute( + "SELECT id, user_id, last_used_at FROM api_keys" + " WHERE key_hash = ? AND revoked_at IS NULL", + (api_key_hash,), + ).fetchone() + if row is None: + return None + now = time.time() + if row[2] is None or now - row[2] >= 3600: + self.conn.execute( + "UPDATE api_keys SET last_used_at = ? WHERE id = ?", + (now, row[0]), + ) + self.conn.commit() + return row[1] + + def list_api_keys(self, user_id): + """列出用户全部 key（只看自己的，调用方须先认证）。 + + 返回 dict 列表：primary（id 为 "legacy"，即建用户时发的 + key，没有 hint）+ 附加 key（按创建时间排序）；已撤销的也 + 列出（revoked_at 非空），调用方按此判断。绝不返回明文、 + 不返回哈希。 + """ + self._ensure_api_keys_table() + keys = [] + row = self.conn.execute( + "SELECT api_key_hash, created_at FROM users WHERE id = ?", + (user_id,), + ).fetchone() + if row is not None and row[0]: + keys.append({"id": "legacy", "label": "primary", "hint": None, + "created_at": row[1], "last_used_at": None, + "revoked_at": None}) + for r in self.conn.execute( + "SELECT id, label, hint, created_at, last_used_at," + " revoked_at FROM api_keys WHERE user_id = ?" + " ORDER BY created_at", + (user_id,)).fetchall(): + keys.append({"id": r[0], "label": r[1], "hint": r[2], + "created_at": r[3], "last_used_at": r[4], + "revoked_at": r[5]}) + return keys + + def revoke_api_key(self, user_id, key_id): + """撤销一个 key，成功返回 True，无此 key 或已撤销返回 False。 + + key_id 为 "legacy" 时清 primary（users.api_key_hash 置空）； + 其它 id 撤销附加 key 行（revoked_at 置当前时间，不删行， + 审计与列表仍可见）。撤销自己的最后一个 key 后须重新登录 + 拿新 key，调用方应把这点写进响应提示。 + """ + self._ensure_api_keys_table() + if key_id == "legacy": + cur = self.conn.execute( + "UPDATE users SET api_key_hash = NULL" + " WHERE id = ? AND api_key_hash IS NOT NULL", + (user_id,), + ) + self.conn.commit() + return cur.rowcount > 0 + cur = self.conn.execute( + "UPDATE api_keys SET revoked_at = ?" + " WHERE id = ? AND user_id = ? AND revoked_at IS NULL", + (time.time(), key_id, user_id), + ) + self.conn.commit() + return cur.rowcount > 0 + + def revoke_other_api_keys(self, user_id, keep_hash): + """撤销除 keep_hash 之外的全部 key（附加 + primary）， + 返回撤销数。用于“这台设备之外的 key 全作废”：调用方传 + 本次请求 key 的哈希（即保留的那一个），其余一律撤销。""" + self._ensure_api_keys_table() + cur = self.conn.execute( + "UPDATE api_keys SET revoked_at = ?" + " WHERE user_id = ? AND key_hash != ?" + " AND revoked_at IS NULL", + (time.time(), user_id, keep_hash), + ) + count = cur.rowcount + cur = self.conn.execute( + "UPDATE users SET api_key_hash = NULL" + " WHERE id = ? AND api_key_hash IS NOT NULL" + " AND api_key_hash != ?", + (user_id, keep_hash), + ) + count += cur.rowcount + self.conn.commit() + return count def create_project(self, project_id, owner_user_id): """创建项目成员记录：指定用户为该项目的 owner。""" diff --git a/cloud/oauth.py b/cloud/oauth.py index 058264e..85de977 100644 --- a/cloud/oauth.py +++ b/cloud/oauth.py @@ -602,12 +602,14 @@ def map_external_identity_to_local_user_with_key(auth_store, provider, 返回 ``(user_id, created, api_key_plain, rotated)``： - 新用户：经 ``AuthStore.create_user`` 创建，返回创建时生成的 - key，``rotated`` 为 False； - - 已存在用户：经 ``AuthStore.rotate_api_key`` 轮换，返回新 key， - 旧 key 即失效，``rotated`` 为 True； + primary key，``rotated`` 为 False； + - 已存在用户：经 ``AuthStore.create_api_key`` 签发一个附加 key + （label 为 ``oauth:``），旧 key（primary 与之前 + 的附加 key）全部继续有效，``rotated`` 恒为 False； 调用方须把明文 key 只在回调成功页/响应里展示一次，之后任何 接口都不再返回它；审计由 server 层记（oauth_login 恒记， - 轮换时另记 api_key_rotated）。 + 签发附加 key 时另记 api_key_issued）。``rotated`` 字段只为 + 向后兼容保留，恒为 False——登录不再轮换任何 key。 """ _provider_config(provider) login = identity.get("login") or identity.get("external_id") \ @@ -625,8 +627,9 @@ def map_external_identity_to_local_user_with_key(auth_store, provider, # 新用户：直接返回创建时生成的 key，不再多轮换一次。 user_id, api_key_plain = auth_store.create_user(username) 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 + _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): diff --git a/cloud/server.py b/cloud/server.py index fc4c992..d7e2bea 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -577,6 +577,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): else: self._send_json(404, {"error": "not found"}) return + if path.startswith("/users/me/keys"): + # API keys 管理：全部只操作认证用户自己的 key。 + if method == "GET" and path == "/users/me/keys": + self._user_keys_list() + elif method == "POST" and path == "/users/me/keys": + self._user_keys_create() + elif method == "POST" and \ + path == "/users/me/keys/revoke-others": + self._user_keys_revoke_others() + elif method == "DELETE" and \ + path.startswith("/users/me/keys/"): + key_id = path[len("/users/me/keys/"):] + if key_id and "/" not in key_id: + self._user_keys_revoke(key_id) + else: + self._send_json(404, {"error": "not found"}) + else: + self._send_json(404, {"error": "not found"}) + return if path.startswith("/github-app/"): # GitHub App 一键安装：setup/state 与 status 须持本站 key # （在处理器内验）；setup 表单页与 callback 靠 setup @@ -1801,9 +1820,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 账号，返回 is_bind=True、api_key_plain=None（绑定成功页 绝不显示任何 key）；冲突抛 BindError； 2. 身份已绑定到某本站账号 -> 登录落到被绑账号，并按 - OAuth 登录规矩轮换 key（created=False、rotated=True）， - 与 oauth 账号登录同一语义，不发明新语义； - 3. 否则走原来的确定性 oauth 账号映射（新建/复用 + 轮换）。 + OAuth 登录规矩签发附加 key（created=False、rotated + 恒为 False），与 oauth 账号登录同一语义，不发明新语义； + 3. 否则走原来的确定性 oauth 账号映射（新建/复用 + 签发 + 附加 key）。 """ store = OAuthStateStore(self.server.db_path) try: @@ -1827,19 +1847,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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) + _key_id, api_key_plain = auth.create_api_key( + bound_user_id, "oauth:%s" % provider) 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, + auth.audit(bound_user_id, "", "api_key_issued", provider) + return (bound_user_id, False, api_key_plain, False, identity, False) local_id, created, api_key_plain, rotated = \ map_external_identity_to_local_user_with_key( auth, provider, identity) auth.audit(local_id, "", "oauth_login", "%s:%s" % (provider, identity.get("external_id"))) - if rotated: - auth.audit(local_id, "", "api_key_rotated", provider) + if not created: + auth.audit(local_id, "", "api_key_issued", provider) return local_id, created, api_key_plain, rotated, identity, False def _oauth_callback(self, path): @@ -1880,7 +1901,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): identity.get("already_bound", False)}) return # api_key 明文只在这里出现一次：调用方须立即保存， - # 之后任何接口都不再返回它；轮换时旧 key 已失效。 + # 之后任何接口都不再返回它；登录签发的是附加 key，旧 key + # 仍可继续使用（多 key；不用的旧 key 可在 keys 管理里撤销）。 self._send_json(200, {"user_id": local_id, "created": created, "provider": provider, "login": identity.get("login"), @@ -1905,10 +1927,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): def _oauth_success_page(self, provider, local_id, created, login, api_key_plain, rotated): """GET 回调的成功页：只显示 login、user_id、created 与新 API key - 一次；绝不显示 access_token、client_secret、旧 API key。""" + 一次；绝不显示 access_token、client_secret、旧 API key。 + + rotated 只为向后兼容保留（登录不再轮换，恒为 False）： + 新 key 是附加 key，旧 key 仍可继续使用，页面不再写“旧 key + 已失效”，而是说明多 key 与撤销入口。 + """ rotation_note = ( "

Your previous API key has been rotated and no longer works.

" - if rotated else "") + if rotated else + "

Previous keys still work (multi-key). Revoke unused keys " + "via the API keys management endpoints.

" + if not created else "") return ( "" "OAuth login succeeded" @@ -1922,7 +1952,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "

%s

" "

Please save it immediately. It will not be shown again " "after you leave or refresh this page." - "%s

" + "

" "

Use it as the X-API-Key header for API calls.

" "" % (html.escape(str(provider)), @@ -1930,8 +1960,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): html.escape(str(local_id)), "yes" if created else "no", rotation_note, - html.escape(str(api_key_plain)), - " Your old key has expired (rotated)." if rotated else "")) + html.escape(str(api_key_plain)))) def _oauth_bind_success_page(self, provider, login): """绑定成功页：只显示 provider + 外部 login + 成功提示。 @@ -2266,6 +2295,75 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): conn.close() self._send_json(200, status) + # -- API keys 管理 --------------------------------------------------- + def _user_keys_list(self): + """列出自己的 key：primary（id "legacy"）+ 附加 key。 + + 只返回 id / label / 后 4 位提示 / 创建 / 最近使用 / 撤销 + 时间，绝不返回明文与哈希；只看得到自己的（认证用户）。 + """ + 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: + keys = auth.list_api_keys(user_id) + self._send_json(200, {"keys": keys}) + + def _user_keys_create(self): + """给自己签发一个附加 key：明文只在本次响应里出现一次。 + + body 可含 label（字符串，缺省 "api-key"）；新 key 立刻 + 可用，与 primary 及已有附加 key 互不影响。 + """ + user_id = self._authenticate() + if user_id is None: + self._send_json(401, {"error": "unauthorized"}) + return + body, err = self._read_json() + if err: + body = {} + label = (body or {}).get("label") + if not isinstance(label, str) or not label: + label = "api-key" + with _auth_store(self.server.db_path) as auth: + key_id, api_key_plain = auth.create_api_key(user_id, label) + self._send_json(201, {"id": key_id, "label": label, + "hint": api_key_plain[-4:], + "api_key": api_key_plain}) + + def _user_keys_revoke(self, key_id): + """撤销自己的一个 key：附加 key 记 revoked_at，primary + （id "legacy"）清哈希；无此 key 或已撤销返回 404。 + + 撤销自己的最后一个 key 后须重新登录拿新 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 = auth.revoke_api_key(user_id, key_id) + if not ok: + self._send_json(404, {"error": "key not found"}) + return + self._send_json(200, {"revoked": True, "id": key_id}) + + def _user_keys_revoke_others(self): + """撤销除本次请求 key 之外的全部自己的 key（附加 + primary）， + 返回撤销数。用于“这台设备之外的 key 全作废”；本次请求 + 的 key 本身不受影响。""" + user_id = self._authenticate() + if user_id is None: + self._send_json(401, {"error": "unauthorized"}) + return + api_key = self.headers.get("X-API-Key") or "" + keep_hash = hashlib.sha256(api_key.encode()).hexdigest() + with _auth_store(self.server.db_path) as auth: + count = auth.revoke_other_api_keys(user_id, keep_hash) + self._send_json(200, {"revoked_count": count}) + # -- 推送订阅与分发 --------------------------------------------------- def _push_create_endpoint(self, user_id, project_id): body, err = self._read_json() diff --git a/cloud/tests/test_api_keys.py b/cloud/tests/test_api_keys.py new file mode 100644 index 0000000..11c2d55 --- /dev/null +++ b/cloud/tests/test_api_keys.py @@ -0,0 +1,235 @@ +"""API keys 多 key 单测：primary + 附加 key、撤销、revoke-others、 +登录签发附加 key（不再轮换）。 + +测试里出现的 key 一律是测试内生成的假 key，绝无真实 secret。 +""" + +import hashlib +import http.client +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.oauth import map_external_identity_to_local_user_with_key + 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.oauth import map_external_identity_to_local_user_with_key + from cloud.server import create_server + + +class ApiKeysFunctionTestCase(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_id, self.primary = self.auth.create_user("user-a") + + def tearDown(self): + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def test_primary_listed_as_legacy(self): + keys = self.auth.list_api_keys(self.user_id) + self.assertEqual(len(keys), 1) + self.assertEqual(keys[0]["id"], "legacy") + self.assertEqual(keys[0]["label"], "primary") + self.assertIsNone(keys[0]["hint"]) + self.assertIsNone(keys[0]["revoked_at"]) + # 列表里绝无明文与哈希。 + for key in keys: + self.assertNotIn("api_key", key) + self.assertNotIn("key_hash", key) + + def test_additional_key_authenticates_and_revokes(self): + key_id, plain = self.auth.create_api_key(self.user_id, "laptop") + self.assertEqual(self.auth.authenticate(plain), self.user_id) + self.assertEqual(plain[-4:], [ + k for k in self.auth.list_api_keys(self.user_id) + if k["id"] == key_id][0]["hint"]) + self.assertTrue(self.auth.revoke_api_key(self.user_id, key_id)) + self.assertIsNone(self.auth.authenticate(plain)) + self.assertEqual(self.auth.authenticate(self.primary), self.user_id) + self.assertFalse(self.auth.revoke_api_key(self.user_id, key_id)) + + def test_revoke_legacy_keeps_additional(self): + _key_id, plain = self.auth.create_api_key(self.user_id, "ci") + self.assertTrue(self.auth.revoke_api_key(self.user_id, "legacy")) + self.assertIsNone(self.auth.authenticate(self.primary)) + self.assertEqual(self.auth.authenticate(plain), self.user_id) + self.assertFalse(self.auth.revoke_api_key(self.user_id, "legacy")) + + def test_revoke_others_keeps_only_current(self): + _id1, plain1 = self.auth.create_api_key(self.user_id, "a") + _id2, plain2 = self.auth.create_api_key(self.user_id, "b") + keep = hashlib.sha256(plain2.encode()).hexdigest() + count = self.auth.revoke_other_api_keys(self.user_id, keep) + self.assertEqual(count, 2) # primary + 第一个附加 key + self.assertIsNone(self.auth.authenticate(self.primary)) + self.assertIsNone(self.auth.authenticate(plain1)) + self.assertEqual(self.auth.authenticate(plain2), self.user_id) + + def test_last_used_throttled(self): + key_id, plain = self.auth.create_api_key(self.user_id, "x") + self.assertEqual(self.auth.authenticate(plain), self.user_id) + first = [k for k in self.auth.list_api_keys(self.user_id) + if k["id"] == key_id][0]["last_used_at"] + self.assertIsNotNone(first) + self.assertEqual(self.auth.authenticate(plain), self.user_id) + second = [k for k in self.auth.list_api_keys(self.user_id) + if k["id"] == key_id][0]["last_used_at"] + self.assertEqual(first, second) # 一小时内不重复写库 + + def test_create_for_unknown_user(self): + with self.assertRaises(ValueError): + self.auth.create_api_key("no-such-user", "x") + + def test_oauth_relogin_issues_additional_not_rotation(self): + # 已存在 oauth 用户再次登录：签发附加 key，旧 primary 不死。 + user_id, primary = self.auth.create_user("oauth:github:gh-user") + _uid, created, plain, rotated = \ + map_external_identity_to_local_user_with_key( + self.auth, "github", {"login": "gh-user", + "external_id": "4242"}) + self.assertEqual(_uid, user_id) + self.assertFalse(created) + self.assertFalse(rotated) + self.assertNotEqual(plain, primary) + self.assertEqual(self.auth.authenticate(primary), user_id) + self.assertEqual(self.auth.authenticate(plain), user_id) + + +class ApiKeysHttpTestCase(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.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() + + 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 test_keys_require_auth(self): + status, _ = self._req("GET", "/users/me/keys") + self.assertEqual(status, 401) + status, _ = self._req("POST", "/users/me/keys", {}) + self.assertEqual(status, 401) + + def test_create_list_revoke_flow(self): + status, body = self._req("GET", "/users/me/keys", key=self.key_a) + self.assertEqual(status, 200, body) + self.assertEqual(len(body["keys"]), 1) + self.assertEqual(body["keys"][0]["id"], "legacy") + # 创建附加 key：明文只出现一次，hint 为后 4 位。 + status, body = self._req( + "POST", "/users/me/keys", {"label": "laptop"}, key=self.key_a) + self.assertEqual(status, 201, body) + new_key = body["api_key"] + self.assertEqual(body["hint"], new_key[-4:]) + key_id = body["id"] + # 新 key 可用；列表里有两条且无明文。 + status, _ = self._req("GET", "/users/me/keys", key=new_key) + self.assertEqual(status, 200) + status, body = self._req("GET", "/users/me/keys", key=self.key_a) + self.assertEqual(len(body["keys"]), 2) + self.assertNotIn(new_key, json.dumps(body)) + # 撤销附加 key：新 key 401，primary 仍可用；再撤 404。 + status, body = self._req( + "DELETE", "/users/me/keys/" + key_id, key=self.key_a) + self.assertEqual(status, 200, body) + status, _ = self._req("GET", "/users/me/keys", key=new_key) + self.assertEqual(status, 401) + status, _ = self._req("GET", "/users/me/keys", key=self.key_a) + self.assertEqual(status, 200) + status, _ = self._req( + "DELETE", "/users/me/keys/" + key_id, key=self.key_a) + self.assertEqual(status, 404) + # 撤销 primary（legacy）：primary 401；再撤需用另一把 key， + # 结果应为 404（已撤销），且另一把 key 不受影响。 + status, body = self._req( + "POST", "/users/me/keys", {"label": "spare"}, key=self.key_a) + self.assertEqual(status, 201, body) + spare_key = body["api_key"] + status, body = self._req( + "DELETE", "/users/me/keys/legacy", key=self.key_a) + self.assertEqual(status, 200, body) + status, _ = self._req("GET", "/users/me/keys", key=self.key_a) + self.assertEqual(status, 401) + status, _ = self._req( + "DELETE", "/users/me/keys/legacy", key=spare_key) + self.assertEqual(status, 404) + status, _ = self._req("GET", "/users/me/keys", key=spare_key) + self.assertEqual(status, 200) + + def test_revoke_legacy_and_revoke_others(self): + _s, created = self._req( + "POST", "/users/me/keys", {"label": "ci"}, key=self.key_a) + ci_key = created["api_key"] + _s, created2 = self._req( + "POST", "/users/me/keys", {"label": "home"}, key=self.key_a) + home_key = created2["api_key"] + # revoke-others（用 home key 调）：primary + ci 被撤销。 + status, body = self._req( + "POST", "/users/me/keys/revoke-others", {}, key=home_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["revoked_count"], 2) + status, _ = self._req("GET", "/users/me/keys", key=self.key_a) + self.assertEqual(status, 401) + status, _ = self._req("GET", "/users/me/keys", key=ci_key) + self.assertEqual(status, 401) + status, _ = self._req("GET", "/users/me/keys", key=home_key) + self.assertEqual(status, 200) + # primary 已被 revoke-others 撤销，再撤 legacy 应为 404； + # home key 本身不受影响。 + status, body = self._req( + "DELETE", "/users/me/keys/legacy", key=home_key) + self.assertEqual(status, 404, body) + status, _ = self._req("GET", "/users/me/keys", key=home_key) + self.assertEqual(status, 200) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_oauth_bind.py b/cloud/tests/test_oauth_bind.py index 362baeb..39dec50 100644 --- a/cloud/tests/test_oauth_bind.py +++ b/cloud/tests/test_oauth_bind.py @@ -232,15 +232,16 @@ class BindHttpTestCase(unittest.TestCase): status, body = self._req( "GET", "/oauth/github/bindings", key=self.key_b) self.assertEqual(body["bindings"], []) - # 登录（无 bind 标记的 state）落到被绑账号并轮换 key。 + # 登录（无 bind 标记的 state）落到被绑账号并签发附加 key， + # 旧 key 仍可用（多 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"]) + self.assertFalse(body["api_key_rotated"]) new_key = body["api_key"] status, _ = self._req( "GET", "/oauth/github/bindings", key=self.key_a) - self.assertEqual(status, 401) + self.assertEqual(status, 200) status, _ = self._req( "GET", "/oauth/github/bindings", key=new_key) self.assertEqual(status, 200) diff --git a/cloud/tests/test_oauth_real_login.py b/cloud/tests/test_oauth_real_login.py index 522c03b..6e6d31b 100644 --- a/cloud/tests/test_oauth_real_login.py +++ b/cloud/tests/test_oauth_real_login.py @@ -107,7 +107,7 @@ class OAuthRealLoginTestCase(unittest.TestCase): self.assertIsNotNone(match, page) return match.group(1) - # -- GET 回调成功：发 key、key 可用、轮换旧 key 失效 ------------- + # -- GET 回调成功：发 key、旧 key 不失效（多 key） --------------- def test_get_callback_success_issues_key_and_rotation(self): state = self._login_state() status, page = self._req( @@ -121,20 +121,22 @@ class OAuthRealLoginTestCase(unittest.TestCase): user_id = self.auth.authenticate(key1) self.assertIsNotNone(user_id) - # 同一身份第二次登录：新 key、旧 key 失效、审计有轮换记录。 + # 同一身份第二次登录：新 key 是附加 key，旧 key 仍可用， + # 审计记 api_key_issued（不再轮换、不再记 api_key_rotated）。 state2 = self._login_state() status, page2 = self._req( "GET", "/oauth/github/callback?code=fake-code&state=" + state2) self.assertEqual(status, 200, page2) self.assertIn("new user: no", page2) - self.assertIn("rotated", page2) + self.assertIn("Previous keys still work", page2) key2 = self._api_key_in_page(page2) self.assertNotEqual(key1, key2) - self.assertIsNone(self.auth.authenticate(key1)) + self.assertEqual(self.auth.authenticate(key1), user_id) self.assertEqual(self.auth.authenticate(key2), user_id) actions = [row["action"] for row in self.auth.list_audit("", limit=50)] self.assertIn("oauth_login", actions) - self.assertIn("api_key_rotated", actions) + self.assertIn("api_key_issued", actions) + self.assertNotIn("api_key_rotated", actions) def test_post_callback_success_returns_key_once(self): state = self._login_state()