"""只用HTTP fixture验证密钥生命周期，不连接真实Supabase或生成实际账号。""" from __future__ import annotations import hashlib import json import sys import unittest import uuid from pathlib import Path import httpx ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / "apps/market-api")) from app.native.auth_keys import KeyStore, KeyStoreError class KeyStoreTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.requests = [] self.rows = {} self.failure = None self.audit_rows = [] self.client = httpx.AsyncClient(transport=httpx.MockTransport(self.handle), headers={"Authorization": "Bearer inherited-fixture"}, cookies={"session": "inherited-fixture-cookie"}, auth=httpx.BasicAuth("fixture", "fixture-password")) self.store = KeyStore(self.client, "https://fixture.supabase.co", "sb_publishable_fixture") async def asyncTearDown(self): await self.client.aclose() def handle(self, request): self.requests.append(request) if self.failure is not None: return httpx.Response(self.failure, text="fixture-private-upstream-error") payload = json.loads(request.content) name = request.url.path.split("/")[-1] if name == "message_create_api_key": self.rows[payload["p_id"]] = { "id": payload["p_id"], "profile_id": payload["p_profile_id"], "name": payload["p_name"], "prefix": payload["p_prefix"], "key_hash": payload["p_key_hash"], "created_at": 1, "expires_at": 9999999999, "last_used_at": None, "revoked_at": None, "created_by": "fixture-admin", } return httpx.Response(200, json=self.rows[payload["p_id"]]) if name == "message_list_api_keys": return httpx.Response(200, json=[r for r in self.rows.values() if r["profile_id"] == payload["p_profile_id"]]) if name == "message_revoke_api_key": row = self.rows.get(payload["p_id"]) found = bool(row and row["profile_id"] == payload["p_profile_id"] and row["revoked_at"] is None) if found: row["revoked_at"] = 2 return httpx.Response(200, json=found) if name == "message_validate_api_key": found = next((r for r in self.rows.values() if r["profile_id"] == payload["p_profile_id"] and r["key_hash"] == payload["p_key_hash"] and r["revoked_at"] is None), None) return httpx.Response(200, json={"id": found["id"], "profile_id": found["profile_id"]}) if found else httpx.Response(200, text="null") if name == "message_auth_audit": return httpx.Response(200, json=self.audit_rows) raise AssertionError(name) async def test_key_secret_is_returned_once_and_only_hash_crosses_rpc(self): created = await self.store.create("alpha", "编辑器", 30, "fixture-jwt") self.assertRegex(created["key"], r"^msg_key\.[a-f0-9]{32}\.[A-Za-z0-9_-]{43}$") request = self.requests[-1] payload = json.loads(request.content) self.assertNotIn(created["key"], request.content.decode()) self.assertEqual(payload["p_key_hash"], hashlib.sha256(created["key"].encode()).hexdigest()) self.assertEqual(request.headers["apikey"], "sb_publishable_fixture") self.assertEqual(request.headers["Authorization"], "Bearer fixture-jwt") self.assertNotIn("Cookie", request.headers) self.assertNotIn("key_hash", created) listed = await self.store.list("alpha", "fixture-jwt") self.assertEqual(listed, [{k: v for k, v in created.items() if k != "key"}]) self.assertNotIn("key", listed[0]) self.assertNotIn("key_hash", listed[0]) async def test_key_validation_is_profile_bound_and_anonymous(self): created = await self.store.create("alpha", "fixture", 1, "fixture-jwt") identity = await self.store.authenticate(created["key"], "alpha") self.assertEqual(identity, {"id": created["id"], "profile_id": "alpha"}) self.assertNotIn("Authorization", self.requests[-1].headers) self.assertNotIn("Cookie", self.requests[-1].headers) self.assertNotIn(created["key"], self.requests[-1].content.decode()) self.assertIsNone(await self.store.authenticate(created["key"], "beta")) bad = created["key"][:-1] + ("A" if created["key"][-1] != "A" else "B") self.assertIsNone(await self.store.authenticate(bad, "alpha")) async def test_revoke_does_not_cross_profiles_and_immediately_rejects_key(self): created = await self.store.create("alpha", "fixture", 365, "fixture-jwt") self.assertFalse(await self.store.revoke("beta", created["id"], "fixture-jwt")) self.assertTrue(await self.store.revoke("alpha", created["id"], "fixture-jwt")) self.assertIsNone(await self.store.authenticate(created["key"], "alpha")) self.assertFalse(await self.store.revoke("alpha", created["id"], "fixture-jwt")) self.assertIsNotNone((await self.store.list("alpha", "fixture-jwt"))[0]["revoked_at"]) async def test_invalid_creation_inputs_never_reach_network(self): for days in (0, -1, 366, True, 2.5, "30"): with self.subTest(days=days), self.assertRaises(KeyStoreError) as caught: await self.store.create("alpha", "fixture", days, "fixture-jwt") self.assertEqual(caught.exception.status_code, 422) for name in ("", " \n", "x" * 81, "key\x00name"): with self.subTest(name=name), self.assertRaises(KeyStoreError): await self.store.create("alpha", name, 1, "fixture-jwt") self.assertEqual(self.requests, []) async def test_malformed_credentials_never_reach_network(self): for key in ("", "fixture-jwt", "msg_key." + "x" * 2000, None): self.assertIsNone(await self.store.authenticate(key, "alpha")) with self.assertRaises(KeyStoreError): await self.store.revoke("alpha", "not-a-uuid", "fixture-jwt") with self.assertRaises(KeyStoreError): await self.store.list("../other", "fixture-jwt") with self.assertRaises(KeyStoreError): await self.store.list("alpha", "") self.assertEqual(self.requests, []) async def test_upstream_errors_are_safe_and_never_accepted(self): for status, expected in ((401, 401), (403, 403), (400, 422), (409, 409), (500, 503), (302, 503)): self.failure = status with self.subTest(status=status), self.assertRaises(KeyStoreError) as caught: await self.store.list("alpha", "fixture-jwt") self.assertEqual(caught.exception.status_code, expected) self.assertNotIn("fixture-private", str(caught.exception)) async def test_audit_only_returns_known_safe_fields(self): row = {"id": str(uuid.uuid4()), "event": "key.created", "created_at": 1, "actor": "fixture-admin", "profile_id": "alpha", "detail": "已创建客户端密钥"} self.audit_rows = [{**row, "key_hash": "unexpected-private-value"}] self.assertEqual(await self.store.audit("fixture-jwt"), [row]) async def test_malformed_upstream_and_transport_failure_are_unavailable(self): async def invalid(request): return httpx.Response(200, text="not-json") async with httpx.AsyncClient(transport=httpx.MockTransport(invalid)) as client: store = KeyStore(client, "https://fixture.supabase.co", "sb_publishable_fixture") with self.assertRaises(KeyStoreError) as caught: await store.list("alpha", "fixture-jwt") self.assertEqual(caught.exception.status_code, 503) async def failing(request): raise httpx.ConnectError("fixture-network-detail") async with httpx.AsyncClient(transport=httpx.MockTransport(failing)) as client: store = KeyStore(client, "https://fixture.supabase.co", "sb_publishable_fixture") with self.assertRaises(KeyStoreError) as caught: await store.list("alpha", "fixture-jwt") self.assertEqual(caught.exception.status_code, 503) self.assertNotIn("fixture-network-detail", str(caught.exception)) if __name__ == "__main__": unittest.main()