tests/native_api/test_supabase_auth.py

"""只用官方 HTTP 形状的 fixture 验证 Supabase 适配器，不创建外部账号或凭据。"""
from __future__ import annotations

import json
import logging
from pathlib import Path
import sys
import traceback
import unittest

import httpx

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "apps/market-api"))
from app.native.auth_supabase import AuthDenied, AuthUnavailable, AuthRateLimited, RECOVERY_CALLBACK, RECOVERY_MESSAGE, SupabaseAuth


URL = "https://fixtureproject.supabase.co"
KEY = "sb_publishable_fixture_not_a_secret"
ORIGIN = "https://message.example"
USER_ID = "123e4567-e89b-42d3-a456-426614174000"
EMAIL = "fixture@example.invalid"
ACCESS = "fixture.access.token"
REFRESH = "fixture-refresh-token"
PASSWORD = "fixture-password-only"
VERIFIER = "v" * 64
CHALLENGE = "c" * 43


def identity(**changes):
    return {"id": USER_ID, "email": EMAIL, "role": "authenticated", "is_anonymous": False, **changes}


def session(**changes):
    return {"access_token": ACCESS, "refresh_token": REFRESH, "expires_in": 3600, "token_type": "bearer", "user": identity(), **changes}


class SupabaseAuthTests(unittest.IsolatedAsyncioTestCase):
    def adapter(self, handler):
        client = httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True,
                                  headers={"Authorization": "Bearer unrelated-client-secret", "X-Unrelated": "private"},
                                  cookies={"unrelated": "private"}, auth=("other-user", "other-password"))
        self.addAsyncCleanup(client.aclose)
        return SupabaseAuth(client, URL, KEY, ORIGIN)

    async def test_login_uses_official_endpoint_and_drops_unrelated_client_credentials(self):
        requests = []
        def handler(request):
            requests.append(request)
            self.assertEqual(request.method, "POST")
            self.assertEqual(str(request.url), URL + "/auth/v1/token?grant_type=password")
            self.assertEqual(json.loads(request.content), {"email": EMAIL, "password": "short"})
            self.assertEqual(request.headers["apikey"], KEY)
            for name in ("authorization", "cookie", "x-unrelated"):
                self.assertNotIn(name, request.headers)
            return httpx.Response(200, json=session(provider_token="must-drop", user=identity(user_metadata={"admin": True})))
        result = await self.adapter(handler).login(EMAIL, "short")
        self.assertEqual(result, {"access_token": ACCESS, "refresh_token": REFRESH, "expires_in": 3600, "user": {"id": USER_ID, "email": EMAIL}})
        self.assertEqual(len(requests), 1)

    async def test_identity_and_admin_membership_are_checked_on_every_request(self):
        requests, checks = [], []
        def handler(request):
            requests.append(request)
            self.assertEqual(request.headers["Authorization"], "Bearer " + ACCESS)
            self.assertEqual(request.headers["apikey"], KEY)
            if request.url.path == "/auth/v1/user":
                return httpx.Response(200, json=identity(user_metadata={"admin": True}, app_metadata={"role": "admin"}))
            self.assertEqual(request.url.path, "/rest/v1/rpc/message_current_admin")
            self.assertEqual(request.method, "POST")
            self.assertEqual(json.loads(request.content), {})
            checks.append(True)
            return httpx.Response(200, json=len(checks) == 1)
        auth = self.adapter(handler)
        self.assertEqual(await auth.user(ACCESS), {"id": USER_ID, "email": EMAIL})
        self.assertIsNone(await auth.user(ACCESS), "撤销 membership 或会话必须在下一次请求拒绝")
        self.assertEqual(len(requests), 4)

    async def test_anonymous_and_service_identities_are_never_admins(self):
        for changed in ({"is_anonymous": True}, {"is_anonymous": None}, {"role": "anon"}, {"role": "service_role"}):
            requests = []
            def handler(request):
                requests.append(request)
                return httpx.Response(200, json=identity(**changed, user_metadata={"admin": True}))
            with self.subTest(changed=changed):
                self.assertIsNone(await self.adapter(handler).user(ACCESS))
                self.assertEqual(len(requests), 1)

    async def test_invalid_or_revoked_session_is_denied_without_leaking_response(self):
        for status in (400, 401, 403, 422):
            with self.subTest(status=status):
                auth = self.adapter(lambda _: httpx.Response(status, json={"message": ACCESS, "error_code": "session_not_found"}))
                self.assertIsNone(await auth.user(ACCESS))

    async def test_false_membership_cannot_be_overridden_by_metadata(self):
        def handler(request):
            if request.url.path == "/auth/v1/user":
                return httpx.Response(200, json=identity(user_metadata={"role": "admin"}, app_metadata={"role": "admin"}))
            return httpx.Response(200, json=False)
        self.assertIsNone(await self.adapter(handler).user(ACCESS))

    async def test_recovery_identity_uses_separate_permission_rpc(self):
        requested = []
        def handler(request):
            requested.append(request.url.path)
            if request.url.path == "/auth/v1/user":
                return httpx.Response(200, json=identity())
            return httpx.Response(200, json=request.url.path == "/rest/v1/rpc/message_recovery_admin")
        auth = self.adapter(handler)
        self.assertIsNone(await auth.user(ACCESS))
        self.assertEqual(await auth.user(ACCESS, recovery=True), {"id": USER_ID, "email": EMAIL})
        self.assertEqual(requested, ["/auth/v1/user", "/rest/v1/rpc/message_current_admin", "/auth/v1/user", "/rest/v1/rpc/message_recovery_admin"])

    async def test_malformed_membership_never_becomes_truthy_authorization(self):
        for value in (None, 1, "true", {}, [{"message_current_admin": True}]):
            def handler(request):
                return httpx.Response(200, json=identity() if request.url.path == "/auth/v1/user" else value)
            with self.subTest(value=value), self.assertRaises(AuthUnavailable):
                await self.adapter(handler).user(ACCESS)

    async def test_missing_rpc_is_an_unavailable_service_not_a_non_admin(self):
        def handler(request):
            return httpx.Response(200, json=identity()) if request.url.path == "/auth/v1/user" else httpx.Response(404, text="function absent")
        with self.assertRaises(AuthUnavailable):
            await self.adapter(handler).user(ACCESS)

    async def test_identity_requires_canonical_nonzero_uuid(self):
        for value in ("first-user", "00000000-0000-0000-0000-000000000000", USER_ID.upper(), None, 3):
            with self.subTest(value=value), self.assertRaises(AuthUnavailable):
                await self.adapter(lambda _: httpx.Response(200, json=identity(id=value))).user(ACCESS)

    async def test_refresh_uses_single_request_and_returns_only_session_fields(self):
        requests = []
        def handler(request):
            requests.append(request)
            self.assertEqual(request.url.params["grant_type"], "refresh_token")
            self.assertEqual(json.loads(request.content), {"refresh_token": REFRESH})
            self.assertNotIn("authorization", request.headers)
            return httpx.Response(200, json=session(refresh_token="fixture-rotated-refresh"))
        result = await self.adapter(handler).refresh(REFRESH)
        self.assertEqual(result["refresh_token"], "fixture-rotated-refresh")
        self.assertEqual(len(requests), 1)

    async def test_invalid_upstream_sessions_are_rejected(self):
        for changed in ({"access_token": ""}, {"access_token": "bad\nheader"}, {"refresh_token": None}, {"expires_in": 0}, {"expires_in": True}, {"expires_in": 86401}, {"token_type": "MAC"}, {"user": {}}):
            with self.subTest(changed=changed), self.assertRaises((AuthUnavailable, AuthDenied)):
                await self.adapter(lambda _: httpx.Response(200, json=session(**changed))).login(EMAIL, PASSWORD)

    async def test_recovery_uses_s256_and_fixed_public_origin_callback(self):
        requests = []
        def handler(request):
            requests.append(request)
            self.assertEqual(request.url.path, "/auth/v1/recover")
            self.assertEqual(request.url.params["redirect_to"], ORIGIN + RECOVERY_CALLBACK)
            self.assertEqual(json.loads(request.content), {"email": EMAIL, "code_challenge": CHALLENGE, "code_challenge_method": "s256"})
            self.assertNotIn(EMAIL, str(request.url))
            self.assertNotIn(VERIFIER, str(request.url))
            return httpx.Response(200, json={})
        self.assertEqual(await self.adapter(handler).recover(EMAIL, CHALLENGE), RECOVERY_MESSAGE)
        self.assertEqual(len(requests), 1)

    async def test_recovery_code_and_verifier_are_only_in_post_body(self):
        code = "fixture-single-use-code"
        def handler(request):
            self.assertEqual(request.method, "POST")
            self.assertEqual(str(request.url), URL + "/auth/v1/token?grant_type=pkce")
            self.assertEqual(json.loads(request.content), {"auth_code": code, "code_verifier": VERIFIER})
            self.assertNotIn(code, str(request.url))
            self.assertNotIn(VERIFIER, str(request.url))
            return httpx.Response(200, json=session())
        self.assertEqual((await self.adapter(handler).exchange_recovery(code, VERIFIER))["user"]["id"], USER_ID)

    async def test_password_update_is_only_a_supabase_user_update(self):
        def handler(request):
            self.assertEqual(request.method, "PUT")
            self.assertEqual(request.url.path, "/auth/v1/user")
            self.assertEqual(json.loads(request.content), {"password": PASSWORD})
            self.assertEqual(request.headers["authorization"], "Bearer " + ACCESS)
            return httpx.Response(200, json=identity())
        self.assertEqual(await self.adapter(handler).update_password(ACCESS, PASSWORD), {"id": USER_ID, "email": EMAIL})

    async def test_logout_is_scoped_to_current_supabase_session(self):
        def handler(request):
            self.assertEqual(request.method, "POST")
            self.assertEqual(str(request.url), URL + "/auth/v1/logout?scope=local")
            self.assertEqual(request.headers["authorization"], "Bearer " + ACCESS)
            return httpx.Response(204)
        self.assertIsNone(await self.adapter(handler).logout(ACCESS))

    async def test_redirects_never_receive_credentials(self):
        requests = []
        def handler(request):
            requests.append(request)
            return httpx.Response(307, headers={"location": "https://attacker.invalid/collect"})
        with self.assertRaises(AuthUnavailable):
            await self.adapter(handler).login(EMAIL, PASSWORD)
        self.assertEqual(len(requests), 1)
        self.assertEqual(requests[0].url.host, "fixtureproject.supabase.co")

    async def test_denial_rate_limit_and_network_errors_never_expose_secrets(self):
        leaked = " ".join((PASSWORD, ACCESS, REFRESH, EMAIL, VERIFIER))
        for status in (400, 401, 403, 422, 429, 500):
            expected = AuthDenied if status in {400, 401, 403, 422} else AuthUnavailable
            with self.subTest(status=status):
                auth = self.adapter(lambda _: httpx.Response(status, json={"error": leaked}))
                try:
                    await auth.login(EMAIL, PASSWORD)
                except expected as error:
                    self.assertNotIn(leaked, "".join(traceback.format_exception(error)))
                else:
                    self.fail("上游失败必须拒绝")
        calls = []
        def timeout(request):
            calls.append(request)
            raise httpx.ReadTimeout(leaked, request=request)
        try:
            await self.adapter(timeout).refresh(REFRESH)
        except AuthUnavailable as error:
            self.assertNotIn(leaked, "".join(traceback.format_exception(error)))
        else:
            self.fail("刷新超时必须失败关闭")
        self.assertEqual(len(calls), 1, "不得自动重复刷新单次令牌")

    async def test_malformed_json_and_logout_failure_have_generic_errors(self):
        with self.assertRaises(AuthUnavailable) as raised:
            await self.adapter(lambda _: httpx.Response(200, text=ACCESS)).login(EMAIL, PASSWORD)
        self.assertNotIn(ACCESS, "".join(traceback.format_exception(raised.exception)))
        with self.assertRaises(AuthDenied) as raised:
            await self.adapter(lambda _: httpx.Response(403, json={"message": ACCESS})).logout(ACCESS)
        self.assertNotIn(ACCESS, str(raised.exception))

    async def test_secret_bodies_do_not_appear_in_http_client_logs(self):
        with self.assertLogs("httpx", level=logging.INFO) as logs:
            await self.adapter(lambda _: httpx.Response(200, json=session())).login(EMAIL, PASSWORD)
        emitted = "\n".join(logs.output)
        for secret in (PASSWORD, EMAIL, ACCESS, REFRESH):
            self.assertNotIn(secret, emitted)

    async def test_invalid_inputs_do_not_make_network_requests(self):
        requests = []
        def handler(request):
            requests.append(request)
            return httpx.Response(200, json=session())
        auth = self.adapter(handler)
        for operation in (auth.login(EMAIL, ""), auth.login(EMAIL, "x" * 257), auth.refresh("bad\r\nheader"), auth.exchange_recovery("code", "short"), auth.recover(EMAIL, "bad"), auth.update_password(ACCESS, "short"), auth.update_password(ACCESS, "x" * 257)):
            with self.assertRaises(AuthDenied):
                await operation
        self.assertIsNone(await auth.user("bad\nheader"))
        self.assertEqual(requests, [])

    async def test_constructor_rejects_credential_destination_confusion(self):
        client = httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200)))
        self.addAsyncCleanup(client.aclose)
        for url in ("http://fixture.supabase.co", "https://supabase.co", "https://fixture.supabase.co.evil", "https://nested.fixture.supabase.co", "https://user@fixture.supabase.co", URL + "/auth/v1", URL + "?key=secret", URL + "#fragment", URL + ":443"):
            with self.subTest(url=url), self.assertRaises(ValueError):
                SupabaseAuth(client, url, KEY, ORIGIN)
        for key in ("sb_secret_fixture", "service_role", "sb_publishable_bad\nheader", ""):
            with self.subTest(key=key), self.assertRaises(ValueError):
                SupabaseAuth(client, URL, key, ORIGIN)
        for origin in ("http://external.example", ORIGIN + "/path", ORIGIN + "?next=bad", "https://user:password@message.example", "https://message.example\\evil", "https://message.example:bad"):
            with self.subTest(origin=origin), self.assertRaises(ValueError):
                SupabaseAuth(client, URL, KEY, origin)
        for origin in (ORIGIN, ORIGIN + "/", "http://127.0.0.1:8000", "http://localhost:8000", "http://[::1]:8000"):
            SupabaseAuth(client, URL + "/", KEY, origin)


    async def test_recovery_rate_limit_has_clear_safe_message_and_retry_header(self):
        adapter = self.adapter(lambda _request: httpx.Response(429, headers={'Retry-After': '120'}, json={'code':'over_email_send_rate_limit', 'msg':'fixture-password-and-secret-must-not-leak'}))
        with self.assertRaises(AuthRateLimited) as caught:
            await adapter.recover(EMAIL, CHALLENGE)
        self.assertEqual(caught.exception.status_code, 429)
        self.assertEqual(caught.exception.retry_after, '120')
        self.assertIn('邮件发送过于频繁', str(caught.exception))
        self.assertNotIn('fixture-password', str(caught.exception))

    async def test_rate_limit_does_not_invent_wait_time(self):
        adapter = self.adapter(lambda _request: httpx.Response(429, json={'code':'over_email_send_rate_limit'}))
        with self.assertRaises(AuthRateLimited) as caught:
            await adapter.recover(EMAIL, CHALLENGE)
        self.assertIsNone(caught.exception.retry_after)
        for header in ('99999999', 'invalid', '-1', '0'):
            self.assertIsNone(AuthRateLimited(retry_after=header).retry_after)

if __name__ == "__main__":
    unittest.main()