247 lines
9.6 KiB
Python
247 lines
9.6 KiB
Python
"""OAuth 真实浏览器登录的 HTTP 端点测试。
|
||
|
||
不联网、不向真实 GitHub/Google 发请求：server 的注入点
|
||
（server.oauth_http_client / server.oauth_userinfo_client）挂 fake；
|
||
真实 userinfo 端点本身只由 cloud.oauth.default_userinfo_client
|
||
拼 URL，不在这里调真实网络。测试里出现的 token、key 一律是
|
||
fake 占位，绝无真实 client_secret、真实 token、真实 GitHub 账号。
|
||
"""
|
||
|
||
import http.client
|
||
import os
|
||
import re
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.auth import AuthStore
|
||
from cloud.oauth import OAuthStateStore
|
||
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 OAuthStateStore
|
||
from cloud.server import create_server
|
||
|
||
|
||
def _fake_token_client(result=None, error=None):
|
||
def fake(token_url, payload):
|
||
if error is not None:
|
||
return {"error": error}
|
||
return dict(result if result is not None else
|
||
{"access_token": "fake-token-not-real"})
|
||
return fake
|
||
|
||
|
||
def _fake_userinfo(result=None, exc=None):
|
||
def fake(access_token):
|
||
if exc is not None:
|
||
raise exc
|
||
return dict(result if result is not None else
|
||
{"id": 4242, "login": "gh-user",
|
||
"email": "gh-user@example.invalid"})
|
||
return fake
|
||
|
||
|
||
class OAuthRealLoginTestCase(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.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()
|
||
# 默认挂 fake：token 响应故意不带身份字段，逼走 userinfo 路径。
|
||
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):
|
||
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
||
try:
|
||
headers = {}
|
||
payload = None
|
||
if body is not None:
|
||
import json
|
||
payload = json.dumps(body).encode("utf-8")
|
||
headers["Content-Type"] = "application/json"
|
||
conn.request(method, path, body=payload, headers=headers)
|
||
resp = conn.getresponse()
|
||
data = resp.read().decode("utf-8", "replace")
|
||
finally:
|
||
conn.close()
|
||
return resp.status, data
|
||
|
||
def _login_state(self):
|
||
status, _ = self._req(
|
||
"GET",
|
||
"/oauth/github/login?client_id=test-client"
|
||
"&redirect_uri=http%3A%2F%2F127.0.0.1%2Fcb")
|
||
self.assertEqual(status, 200)
|
||
store = OAuthStateStore(self.db_path)
|
||
try:
|
||
row = store.conn.execute(
|
||
"SELECT state FROM oauth_states ORDER BY created_at DESC"
|
||
" LIMIT 1").fetchone()
|
||
finally:
|
||
store.conn.close()
|
||
return row[0]
|
||
|
||
def _api_key_in_page(self, page):
|
||
match = re.search(r"<code>([^<]+)</code>", page)
|
||
self.assertIsNotNone(match, page)
|
||
return match.group(1)
|
||
|
||
# -- GET 回调成功：发 key、旧 key 不失效（多 key） ---------------
|
||
def test_get_callback_success_issues_key_and_rotation(self):
|
||
state = self._login_state()
|
||
status, page = self._req(
|
||
"GET", "/oauth/github/callback?code=fake-code&state=" + state)
|
||
self.assertEqual(status, 200, page)
|
||
self.assertIn("gh-user", page)
|
||
self.assertIn("new user: yes", page)
|
||
self.assertNotIn("fake-token-not-real", page)
|
||
self.assertNotIn("client_secret", page)
|
||
key1 = self._api_key_in_page(page)
|
||
user_id = self.auth.authenticate(key1)
|
||
self.assertIsNotNone(user_id)
|
||
|
||
# 同一身份第二次登录：新 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("Previous keys still work", page2)
|
||
key2 = self._api_key_in_page(page2)
|
||
self.assertNotEqual(key1, key2)
|
||
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_issued", actions)
|
||
self.assertNotIn("api_key_rotated", actions)
|
||
|
||
def test_post_callback_success_returns_key_once(self):
|
||
state = self._login_state()
|
||
import json
|
||
status, body = self._req(
|
||
"POST", "/oauth/github/callback",
|
||
{"code": "fake-code", "state": state,
|
||
"client_id": "test-client",
|
||
"redirect_uri": "http://127.0.0.1/cb"})
|
||
parsed = json.loads(body)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertIn("api_key", parsed)
|
||
self.assertFalse(parsed["api_key_rotated"])
|
||
self.assertEqual(self.auth.authenticate(parsed["api_key"]),
|
||
parsed["user_id"])
|
||
|
||
# -- GET 回调失败路径 ----------------------------------------------
|
||
def test_get_callback_bad_state_400_no_user(self):
|
||
status, page = self._req(
|
||
"GET", "/oauth/github/callback?code=x&state=no-such-state")
|
||
self.assertEqual(status, 400)
|
||
self.assertNotIn("fake-token-not-real", page)
|
||
count = self.auth.conn.execute(
|
||
"SELECT COUNT(*) FROM users").fetchone()[0]
|
||
self.assertEqual(count, 0)
|
||
|
||
def test_get_callback_expired_state_400(self):
|
||
state = self._login_state()
|
||
store = OAuthStateStore(self.db_path)
|
||
try:
|
||
store.conn.execute(
|
||
"UPDATE oauth_states SET created_at = ? WHERE state = ?",
|
||
(time.time() - 601, state))
|
||
store.conn.commit()
|
||
finally:
|
||
store.conn.close()
|
||
status, _ = self._req(
|
||
"GET", "/oauth/github/callback?code=x&state=" + state)
|
||
self.assertEqual(status, 400)
|
||
|
||
def test_get_callback_state_provider_mismatch_400(self):
|
||
state = self._login_state()
|
||
status, _ = self._req(
|
||
"GET", "/oauth/google/callback?code=x&state=" + state)
|
||
self.assertEqual(status, 400)
|
||
|
||
def test_get_callback_token_exchange_failure_400_no_user(self):
|
||
self.server.oauth_http_client = _fake_token_client(
|
||
error="bad_verification_code")
|
||
state = self._login_state()
|
||
status, _ = self._req(
|
||
"GET", "/oauth/github/callback?code=bad&state=" + state)
|
||
self.assertEqual(status, 400)
|
||
count = self.auth.conn.execute(
|
||
"SELECT COUNT(*) FROM users").fetchone()[0]
|
||
self.assertEqual(count, 0)
|
||
|
||
def test_get_callback_userinfo_failure_400_no_user(self):
|
||
self.server.oauth_userinfo_client = _fake_userinfo(exc=RuntimeError("down"))
|
||
state = self._login_state()
|
||
status, page = self._req(
|
||
"GET", "/oauth/github/callback?code=x&state=" + state)
|
||
self.assertEqual(status, 400, page)
|
||
self.assertIn("userinfo", page)
|
||
count = self.auth.conn.execute(
|
||
"SELECT COUNT(*) FROM users").fetchone()[0]
|
||
self.assertEqual(count, 0)
|
||
|
||
def test_get_callback_userinfo_empty_400_no_user(self):
|
||
self.server.oauth_userinfo_client = _fake_userinfo(result={})
|
||
state = self._login_state()
|
||
status, _ = self._req(
|
||
"GET", "/oauth/github/callback?code=x&state=" + state)
|
||
self.assertEqual(status, 400)
|
||
count = self.auth.conn.execute(
|
||
"SELECT COUNT(*) FROM users").fetchone()[0]
|
||
self.assertEqual(count, 0)
|
||
|
||
def test_get_callback_secret_missing_400_before_network(self):
|
||
# 真实路径（不注入 fake）：无 secret 时换 token 前就拒绝，
|
||
# 不会走到真实网络；fake userinfo 在此路径下不被使用。
|
||
self.server.oauth_http_client = None
|
||
self.server.oauth_userinfo_client = None
|
||
old = os.environ.pop("GITHUB_OAUTH_CLIENT_SECRET", None)
|
||
try:
|
||
state = self._login_state()
|
||
status, page = self._req(
|
||
"GET", "/oauth/github/callback?code=x&state=" + state)
|
||
finally:
|
||
if old is not None:
|
||
os.environ["GITHUB_OAUTH_CLIENT_SECRET"] = old
|
||
self.assertEqual(status, 400, page)
|
||
self.assertIn("client_secret", page)
|
||
|
||
# -- AuthStore 轮换本身 ---------------------------------------------
|
||
def test_rotate_api_key_old_key_dead(self):
|
||
user_id, old_key = self.auth.create_user("someone")
|
||
new_key = self.auth.rotate_api_key(user_id)
|
||
self.assertNotEqual(old_key, new_key)
|
||
self.assertIsNone(self.auth.authenticate(old_key))
|
||
self.assertEqual(self.auth.authenticate(new_key), user_id)
|
||
with self.assertRaises(ValueError):
|
||
self.auth.rotate_api_key("no-such-user")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|