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