File size: 16,333 Bytes
9b403c4 8211361 9b403c4 8211361 9b403c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 | """只用官方 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()
|