Spaces:
Sleeping
Sleeping
| """ | |
| db_api_keys.py — user-supplied Anthropic/OpenAI API keys (user_api_keys table). | |
| Ciphertext only ever leaves this module decrypted via | |
| get_active_decrypted_api_keys() / get_decrypted_api_key(), for runtime | |
| injection into an LLM call (see llm_client.py). Never log a decrypted value — | |
| see the hard rule in crypto_util.py. | |
| """ | |
| from db_client import supabase | |
| from crypto_util import encrypt_secret, decrypt_secret | |
| PROVIDERS = {"anthropic", "openai"} | |
| def get_api_key_status(user_id: str) -> dict: | |
| """{'anthropic': {'has_key': bool, 'active': bool, 'last4': str|None}, 'openai': {...}}. | |
| No decryption — safe for GET /profile/load.""" | |
| status = {p: {"has_key": False, "active": False, "last4": None} for p in PROVIDERS} | |
| try: | |
| resp = supabase.table("user_api_keys").select("provider,active,key_last4").eq("user_id", user_id).execute() | |
| for row in resp.data or []: | |
| provider = row.get("provider") | |
| if provider in status: | |
| status[provider] = { | |
| "has_key": True, | |
| "active": bool(row.get("active")), | |
| "last4": row.get("key_last4"), | |
| } | |
| except Exception as e: | |
| print(f"[get_api_key_status] failed for {user_id}: {e}") | |
| return status | |
| def get_decrypted_api_key(user_id: str, provider: str) -> str | None: | |
| """Decrypted key for runtime use only, or None if no active key is stored.""" | |
| resp = ( | |
| supabase.table("user_api_keys") | |
| .select("key_ciphertext") | |
| .eq("user_id", user_id) | |
| .eq("provider", provider) | |
| .eq("active", True) | |
| .execute() | |
| ) | |
| if not resp.data: | |
| return None | |
| return decrypt_secret(resp.data[0]["key_ciphertext"]) | |
| def get_active_decrypted_api_keys(user_id: str) -> dict: | |
| """{'anthropic': 'sk-...', 'openai': 'sk-...'} — only providers with an active | |
| stored key. Used once per request to populate llm_client's ContextVar.""" | |
| resp = supabase.table("user_api_keys").select("provider,key_ciphertext").eq("user_id", user_id).eq("active", True).execute() | |
| return {row["provider"]: decrypt_secret(row["key_ciphertext"]) for row in (resp.data or [])} | |
| def set_api_key(user_id: str, provider: str, raw_key: str) -> dict: | |
| """Encrypt and store raw_key, marking it active. Caller must have already | |
| validated raw_key with a live provider call — this function does not validate.""" | |
| ciphertext = encrypt_secret(raw_key) | |
| last4 = raw_key[-4:] if len(raw_key) >= 4 else raw_key | |
| supabase.table("user_api_keys").upsert({ | |
| "user_id": user_id, | |
| "provider": provider, | |
| "key_ciphertext": ciphertext, | |
| "key_last4": last4, | |
| "active": True, | |
| }).execute() | |
| return {"has_key": True, "active": True, "last4": last4} | |
| def set_active(user_id: str, provider: str, active: bool) -> dict: | |
| """Flip the use-my-key/use-default toggle without touching the stored ciphertext.""" | |
| supabase.table("user_api_keys").update({"active": active}).eq("user_id", user_id).eq("provider", provider).execute() | |
| status = get_api_key_status(user_id) | |
| return status.get(provider, {"has_key": False, "active": False, "last4": None}) | |
| def delete_api_key(user_id: str, provider: str) -> None: | |
| supabase.table("user_api_keys").delete().eq("user_id", user_id).eq("provider", provider).execute() | |