305 lines
11 KiB
Python
305 lines
11 KiB
Python
"""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，
|
||
# 旧 key 仍可用（多 key，不再轮换）。
|
||
status, body = self._callback(self._login_state())
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["user_id"], self.user_a)
|
||
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, 200)
|
||
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()
|