"""User API Key management for the Live Platform (Track E.3). Provides bcrypt-hashed API keys for authenticating MCP client connections to per-user endpoints (``/u/{user_id}/{alias}/mcp``). Key format: ``mhu_<32 random urlsafe chars>`` Lookup: ``key_prefix`` column (first 8 chars after ``mhu_``) for indexed DB lookup, then bcrypt verification on the matching row. Usage: from core.user_keys import initialize_user_key_manager, get_user_key_manager initialize_user_key_manager() mgr = get_user_key_manager() result = await mgr.create_key(user_id, "Claude Desktop") # result["key"] is shown once to the user info = await mgr.validate_key("mhu_...") """ import logging import secrets import time from datetime import UTC, datetime, timedelta from typing import Any import bcrypt logger = logging.getLogger(__name__) # Key format constants KEY_PREFIX_TAG = "mhu_" KEY_RANDOM_BYTES = 32 # urlsafe_b64 of 32 bytes = 43 chars KEY_PREFIX_LEN = 8 # chars after "mhu_" stored for fast DB lookup # Validation cache TTL _CACHE_TTL_SECONDS = 300 # 5 minutes class UserKeyManager: """Manages per-user API keys with bcrypt hashing. Keys are stored in the ``user_api_keys`` SQLite table via :class:`core.database.Database`. Each key is bcrypt-hashed; a plaintext ``key_prefix`` column enables indexed lookup without scanning all rows. An in-memory cache avoids repeated bcrypt verification for the same key within a 5-minute window. """ def __init__(self) -> None: # Cache: raw_key -> (key_id, user_id, scopes, site_id, cached_at) self._cache: dict[str, tuple[str, str, str, str | None, float]] = {} async def create_key( self, user_id: str, name: str, scopes: str = "read editor settings install write admin", expires_in_days: int | None = None, site_id: str | None = None, ) -> dict[str, Any]: """Create a new API key for a user. Args: user_id: Owner's UUID. name: Human label (e.g. "Claude Desktop"). scopes: Access scopes. Default enumerates every tier explicitly (read / editor / settings / install / write / admin) so the key matches the per-site Tool Access dropdown 1:1, even if ``UNIVERSAL_SCOPE_TIERS`` is reorganised in a later phase. The binding access filter for any tool call is the **intersection** of these scopes with the per-site ``tool_scope`` preset — narrowing happens per-site, not per-key, so a single key works for every site the user owns. expires_in_days: Optional expiry in days from now. None = never. site_id: Optional site UUID to scope key to a single site. Returns: Dict with ``key`` (plaintext, shown once), ``key_id``, ``name``, ``scopes``, ``created_at``, ``expires_at``, ``site_id``. """ from core.database import get_database raw_key = KEY_PREFIX_TAG + secrets.token_urlsafe(KEY_RANDOM_BYTES) key_prefix = raw_key[len(KEY_PREFIX_TAG) : len(KEY_PREFIX_TAG) + KEY_PREFIX_LEN] key_hash = bcrypt.hashpw(raw_key.encode(), bcrypt.gensalt()).decode() expires_at = None if expires_in_days is not None: expires_at = (datetime.now(UTC) + timedelta(days=expires_in_days)).isoformat() db = get_database() row = await db.create_api_key( user_id=user_id, key_hash=key_hash, key_prefix=key_prefix, name=name, scopes=scopes, expires_at=expires_at, site_id=site_id, ) logger.info("Created user API key %s for user %s (site=%s)", row["id"], user_id, site_id) return { "key": raw_key, # shown once "key_id": row["id"], "name": row["name"], "scopes": row["scopes"], "created_at": row["created_at"], "expires_at": row["expires_at"], "site_id": row.get("site_id"), } async def validate_key(self, api_key: str) -> dict[str, Any] | None: """Validate an API key and return its metadata. Uses an in-memory cache to avoid repeated bcrypt verification. Args: api_key: The raw API key string (e.g. ``mhu_...``). Returns: Dict with ``key_id``, ``user_id``, ``scopes`` if valid, else None. """ if not api_key or not api_key.startswith(KEY_PREFIX_TAG): return None # Check cache first cached = self._cache.get(api_key) if cached is not None: key_id, user_id, scopes, site_id, cached_at = cached if time.time() - cached_at < _CACHE_TTL_SECONDS: # Update usage in background (fire-and-forget via DB) try: import asyncio from core.database import get_database db = get_database() asyncio.create_task(db.update_api_key_usage(key_id)) except Exception: pass # Non-critical return {"key_id": key_id, "user_id": user_id, "scopes": scopes, "site_id": site_id} else: del self._cache[api_key] # Extract prefix for DB lookup key_prefix = api_key[len(KEY_PREFIX_TAG) : len(KEY_PREFIX_TAG) + KEY_PREFIX_LEN] from core.database import get_database db = get_database() row = await db.get_api_key_by_prefix(key_prefix) if row is None: return None # Verify bcrypt hash if not bcrypt.checkpw(api_key.encode(), row["key_hash"].encode()): return None # Check expiry if row["expires_at"] is not None: expires = datetime.fromisoformat(row["expires_at"]) if expires < datetime.now(UTC): return None # Update usage await db.update_api_key_usage(row["id"]) # Cache the result site_id = row.get("site_id") self._cache[api_key] = ( row["id"], row["user_id"], row["scopes"], site_id, time.time(), ) return { "key_id": row["id"], "user_id": row["user_id"], "scopes": row["scopes"], "site_id": site_id, } async def list_keys(self, user_id: str) -> list[dict[str, Any]]: """List all API keys for a user (without hashes). Args: user_id: Owner's UUID. Returns: List of key metadata dicts. """ from core.database import get_database db = get_database() return await db.get_api_keys_by_user(user_id) async def delete_key(self, key_id: str, user_id: str) -> bool: """Delete an API key. Also invalidates the validation cache for any cached key matching this key_id. Args: key_id: API key UUID. user_id: Owner's UUID. Returns: True if deleted, False if not found. """ from core.database import get_database db = get_database() deleted = await db.delete_api_key(key_id, user_id) if deleted: # Purge from cache to_remove = [k for k, v in self._cache.items() if v[0] == key_id] for k in to_remove: del self._cache[k] logger.info("Deleted user API key %s for user %s", key_id, user_id) return deleted def clear_cache(self) -> None: """Clear the entire validation cache.""" self._cache.clear() # Singleton _manager: UserKeyManager | None = None def initialize_user_key_manager() -> UserKeyManager: """Create and store the singleton UserKeyManager.""" global _manager _manager = UserKeyManager() logger.info("UserKeyManager initialized") return _manager def get_user_key_manager() -> UserKeyManager: """Get the singleton UserKeyManager. Returns: The UserKeyManager singleton. Raises: RuntimeError: If not initialized. """ if _manager is None: raise RuntimeError( "UserKeyManager not initialized. Call initialize_user_key_manager() first." ) return _manager