File size: 8,500 Bytes
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 | """只用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()
|