276 lines
11 KiB
Python
276 lines
11 KiB
Python
"""oauth 脚手架单测：state、授权 URL、回调、换 token 失败。
|
|
|
|
全部外部调用均为注入的 fake http 客户端，本文件不向真实
|
|
GitHub / Google 发任何请求；出现的 client_id / secret 哨兵值
|
|
均为测试用的假值（形如 test- / fake- 前缀），不是真实凭证。
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
try:
|
|
from cloud.auth import AuthStore
|
|
from cloud.oauth import (
|
|
OAuthError,
|
|
OAuthStateStore,
|
|
build_authorization_url,
|
|
exchange_code_for_token,
|
|
handle_oauth_callback,
|
|
map_external_identity_to_local_user,
|
|
start_oauth_login,
|
|
)
|
|
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.auth import AuthStore
|
|
from cloud.oauth import (
|
|
OAuthError,
|
|
OAuthStateStore,
|
|
build_authorization_url,
|
|
exchange_code_for_token,
|
|
handle_oauth_callback,
|
|
map_external_identity_to_local_user,
|
|
start_oauth_login,
|
|
)
|
|
|
|
PROVIDER = "github"
|
|
CLIENT_ID = "test-client-id-not-real"
|
|
REDIRECT_URI = "https://example.invalid/oauth/callback"
|
|
FAKE_SECRET_SENTINEL = "fake-secret-sentinel-not-real"
|
|
|
|
|
|
def _fake_token_client(token_result=None, error=None):
|
|
"""返回 fake http_client：记录调用，不联网。"""
|
|
calls = []
|
|
|
|
def fake(token_url, payload):
|
|
calls.append((token_url, dict(payload)))
|
|
if error is not None:
|
|
raise RuntimeError(error)
|
|
return dict(token_result or {"access_token": "fake-token-not-real"})
|
|
|
|
fake.calls = calls
|
|
return fake
|
|
|
|
|
|
class TestOAuthState(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "oauth.db")
|
|
self.store = OAuthStateStore(self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_state_create_then_validate_ok(self):
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
self.assertTrue(state)
|
|
record = self.store.validate_state(state, PROVIDER, REDIRECT_URI)
|
|
self.assertIsNotNone(record)
|
|
self.assertEqual(record["provider"], PROVIDER)
|
|
self.assertEqual(record["redirect_uri"], REDIRECT_URI)
|
|
# 一次性：第二次校验必然失败（防重放）
|
|
self.assertIsNone(self.store.validate_state(state, PROVIDER))
|
|
|
|
def test_state_expired_rejected(self):
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
expired_now = time.time() + 601 # 默认 TTL 600 秒
|
|
self.assertIsNone(self.store.validate_state(
|
|
state, PROVIDER, now=expired_now))
|
|
# 过期行已被删除，不会残留
|
|
row = self.store.conn.execute(
|
|
"SELECT state FROM oauth_states WHERE state = ?", (state,)
|
|
).fetchone()
|
|
self.assertIsNone(row)
|
|
|
|
def test_state_mismatch_rejected(self):
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
# 不存在的 state
|
|
self.assertIsNone(self.store.validate_state(
|
|
"no-such-state", PROVIDER))
|
|
# provider 不匹配：拒绝，且不消耗合法 state
|
|
self.assertIsNone(self.store.validate_state(state, "google"))
|
|
# redirect_uri 不匹配：同样拒绝且不消耗
|
|
self.assertIsNone(self.store.validate_state(
|
|
state, PROVIDER, "https://example.invalid/other"))
|
|
# 合法组合仍可通过（前两次不匹配没有烧掉它）
|
|
self.assertIsNotNone(self.store.validate_state(
|
|
state, PROVIDER, REDIRECT_URI))
|
|
|
|
|
|
class TestAuthorizationUrl(unittest.TestCase):
|
|
def test_url_contains_params_and_no_secret(self):
|
|
state = "test-state-abc"
|
|
url = build_authorization_url(
|
|
PROVIDER, CLIENT_ID, REDIRECT_URI,
|
|
scope="read:user", state=state)
|
|
parsed = urlparse(url)
|
|
self.assertEqual(parsed.netloc, "github.com")
|
|
params = {k: v[0] for k, v in parse_qs(parsed.query).items()}
|
|
self.assertEqual(params["client_id"], CLIENT_ID)
|
|
self.assertEqual(params["redirect_uri"], REDIRECT_URI)
|
|
self.assertEqual(params["scope"], "read:user")
|
|
self.assertEqual(params["state"], state)
|
|
self.assertNotIn("secret", url.lower())
|
|
self.assertNotIn(FAKE_SECRET_SENTINEL, url)
|
|
|
|
def test_url_default_scope_and_google(self):
|
|
url = build_authorization_url(
|
|
"github", CLIENT_ID, REDIRECT_URI, state="s1")
|
|
self.assertIn("read%3Auser", url) # 默认 scope read:user
|
|
url = build_authorization_url(
|
|
"google", CLIENT_ID, REDIRECT_URI, state="s2")
|
|
params = {k: v[0] for k, v in parse_qs(urlparse(url).query).items()}
|
|
self.assertEqual(params["scope"], "openid email profile")
|
|
self.assertEqual(params["response_type"], "code")
|
|
self.assertNotIn("secret", url.lower())
|
|
|
|
def test_start_login_stores_state(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
try:
|
|
store = OAuthStateStore(os.path.join(tmp.name, "s.db"))
|
|
state, url = start_oauth_login(
|
|
store, PROVIDER, CLIENT_ID, REDIRECT_URI)
|
|
self.assertIn("state=" + state, url)
|
|
row = store.conn.execute(
|
|
"SELECT provider FROM oauth_states WHERE state = ?",
|
|
(state,),
|
|
).fetchone()
|
|
self.assertEqual(row[0], PROVIDER)
|
|
finally:
|
|
tmp.cleanup()
|
|
|
|
def test_unknown_provider_rejected(self):
|
|
with self.assertRaises(OAuthError):
|
|
build_authorization_url(
|
|
"not-a-provider", CLIENT_ID, REDIRECT_URI)
|
|
|
|
|
|
class TestOAuthCallback(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "oauth.db")
|
|
self.store = OAuthStateStore(self.db_path)
|
|
# 同库上的 AuthStore：用于断言“失败不建用户/不记审计”。
|
|
self.auth = AuthStore(self.db_path)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def _login(self, fake=None, **kwargs):
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
return handle_oauth_callback(
|
|
self.store, PROVIDER, "test-code-not-real", state,
|
|
CLIENT_ID, REDIRECT_URI,
|
|
http_client=fake or _fake_token_client(
|
|
{"access_token": "fake-token-not-real",
|
|
"id": 12345, "login": "fake-user"}),
|
|
**kwargs)
|
|
|
|
def _count(self, table):
|
|
return self.store.conn.execute(
|
|
"SELECT COUNT(*) FROM %s" % table).fetchone()[0]
|
|
|
|
def test_callback_success_returns_identity(self):
|
|
identity = self._login()
|
|
self.assertEqual(identity["provider"], PROVIDER)
|
|
self.assertEqual(identity["external_id"], "12345")
|
|
self.assertEqual(identity["login"], "fake-user")
|
|
# access_token 不返回、不入库：回调只交身份
|
|
self.assertNotIn("access_token", identity)
|
|
|
|
def test_callback_maps_identity_to_local_user(self):
|
|
identity = self._login()
|
|
user_id, created = map_external_identity_to_local_user(
|
|
self.auth, PROVIDER, identity)
|
|
self.assertTrue(created)
|
|
user_id2, created2 = map_external_identity_to_local_user(
|
|
self.auth, PROVIDER, identity)
|
|
self.assertEqual(user_id, user_id2)
|
|
self.assertFalse(created2)
|
|
|
|
def test_callback_state_expired_rejected(self):
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
with self.assertRaises(OAuthError) as ctx:
|
|
handle_oauth_callback(
|
|
self.store, PROVIDER, "code", state, CLIENT_ID, REDIRECT_URI,
|
|
http_client=_fake_token_client(),
|
|
now=time.time() + 601)
|
|
self.assertEqual(ctx.exception.reason, "state_expired")
|
|
|
|
def test_callback_state_mismatch_rejected(self):
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
with self.assertRaises(OAuthError) as ctx:
|
|
handle_oauth_callback(
|
|
self.store, "google", "code", state, CLIENT_ID, REDIRECT_URI,
|
|
http_client=_fake_token_client())
|
|
self.assertEqual(ctx.exception.reason, "state_mismatch")
|
|
with self.assertRaises(OAuthError) as ctx:
|
|
handle_oauth_callback(
|
|
self.store, PROVIDER, "code", "missing-state",
|
|
CLIENT_ID, REDIRECT_URI,
|
|
http_client=_fake_token_client())
|
|
self.assertEqual(ctx.exception.reason, "state_not_found")
|
|
|
|
def test_token_exchange_failure_creates_no_session_or_user(self):
|
|
fake = _fake_token_client(error="provider refused code")
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
with self.assertRaises(OAuthError) as ctx:
|
|
handle_oauth_callback(
|
|
self.store, PROVIDER, "bad-code", state,
|
|
CLIENT_ID, REDIRECT_URI, http_client=fake)
|
|
self.assertEqual(ctx.exception.reason, "token_exchange_failed")
|
|
# 换 token 失败：无用户、无审计（本模块本来就不建会话，
|
|
# 这里断言失败路径也没有任何副作用写入同库的 auth 表）。
|
|
self.assertEqual(self._count("users"), 0)
|
|
self.assertEqual(self._count("audit_log"), 0)
|
|
|
|
def test_token_error_dict_creates_nothing(self):
|
|
fake = _fake_token_client(
|
|
{"error": "bad_verification_code"})
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
with self.assertRaises(OAuthError):
|
|
handle_oauth_callback(
|
|
self.store, PROVIDER, "bad-code", state,
|
|
CLIENT_ID, REDIRECT_URI, http_client=fake)
|
|
self.assertEqual(self._count("users"), 0)
|
|
|
|
def test_real_path_without_secret_refuses_before_network(self):
|
|
# http_client=None 且环境变量无 secret：必须在联网前拒绝。
|
|
# 用 patch 清空两个 secret 环境变量，并断言 urlopen 未被调用。
|
|
with mock.patch.dict(os.environ, {}, clear=False):
|
|
os.environ.pop("GITHUB_OAUTH_CLIENT_SECRET", None)
|
|
os.environ.pop("GOOGLE_OAUTH_CLIENT_SECRET", None)
|
|
with mock.patch("cloud.oauth._urllib_post") as mocked:
|
|
with self.assertRaises(OAuthError) as ctx:
|
|
exchange_code_for_token(
|
|
PROVIDER, "code", CLIENT_ID, REDIRECT_URI,
|
|
http_client=None)
|
|
self.assertEqual(
|
|
ctx.exception.reason, "client_secret_missing")
|
|
mocked.assert_not_called()
|
|
|
|
def test_userinfo_client_used_when_token_has_no_identity(self):
|
|
def userinfo(access_token):
|
|
self.assertEqual(access_token, "fake-token-not-real")
|
|
return {"id": 999, "login": "via-userinfo"}
|
|
|
|
state = self.store.create_state(PROVIDER, REDIRECT_URI)
|
|
identity = handle_oauth_callback(
|
|
self.store, PROVIDER, "code", state, CLIENT_ID, REDIRECT_URI,
|
|
http_client=_fake_token_client(
|
|
{"access_token": "fake-token-not-real"}),
|
|
userinfo_client=userinfo)
|
|
self.assertEqual(identity["external_id"], "999")
|
|
self.assertEqual(identity["login"], "via-userinfo")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|