codex-app-server/cloud/tests/test_oauth.py

"""oauth 脚手架单测：state、授权 URL、回调、换 token 失败。

全部外部调用均为注入的 fake http 客户端，本文件不向真实
GitHub / Google 发任何请求；出现的 client_id / secret 哨兵值
均为测试用的假值（形如 test- / fake- 前缀），不是真实凭证。
"""

import os
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()