"""只用官方 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()