feat: API key list, revoke, and specific 401 errors (OPE-164)

backend/middleware/auth.py

@@ -133,11 +133,21 @@ def _validate_api_key(token: str) -> Optional[AuthContext]:
         except Exception:
             pass  # Non-critical; don't fail auth over timestamp update
 
+        # Reject keys with no linked user at the auth layer
+        # so routes get a clear 401 instead of silently passing with user_id=None
+        if not key_data.get("user_id"):
+            raise HTTPException(
+                status_code=401,
+                detail="API key has no linked user. Contact admin."
+            )
+
         return AuthContext(
             api_key_name=key_data.get("name"),
             user_id=key_data.get("user_id"),
             tier=key_data.get("tier", "free")
         )
+    except HTTPException:
+        raise
     except Exception:
         return None
 
@@ -160,10 +170,29 @@ def _authenticate(token: str) -> AuthContext:
         set_user_context(user_id=ctx.user_id or ctx.api_key_name)
         return ctx
 
-    # Neither worked
+    # Provide specific error for ci_ keys so users can debug
+    if token.startswith("ci_"):
+        try:
+            from services.supabase_service import get_supabase_service
+            db = get_supabase_service().client
+            key_hash = hashlib.sha256(token.encode()).hexdigest()
+            result = db.table("api_keys").select(
+                "active, user_id"
+            ).eq("key_hash", key_hash).execute()
+            if not result.data:
+                detail = "API key not found. Check that the key is correct."
+            elif not result.data[0].get("active"):
+                detail = "API key has been revoked."
+            else:
+                detail = "API key validation failed."
+        except Exception:
+            detail = "API key validation failed."
+    else:
+        detail = "Invalid token or API key"
+
     raise HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Invalid token or API key",
+        detail=detail,
         headers={"WWW-Authenticate": "Bearer"}
     )
 

backend/routes/api_keys.py

@@ -1,4 +1,7 @@
 """API key management and metrics routes."""
+from typing import Any, Dict
+from uuid import UUID
+
 from fastapi import APIRouter, HTTPException, Depends
 from pydantic import BaseModel
 
@@ -15,9 +18,11 @@
 router = APIRouter(prefix="", tags=["API Keys"])
 
 
+MAX_KEYS_PER_USER = 5
+
+
 class CreateAPIKeyRequest(BaseModel):
     name: str
-    tier: str = "free"
 
 
 @router.get("/metrics")
@@ -46,37 +51,54 @@ async def generate_api_key(
     auth: AuthContext = Depends(require_auth)
 ):
     """Generate a new API key."""
-    set_operation_context("generate_api_key", user_id=auth.user_id, tier=request.tier)
-    add_breadcrumb("API key generation requested", category="api_keys", tier=request.tier)
+    if not auth.user_id:
+        raise HTTPException(status_code=401, detail="User ID required")
+
+    set_operation_context("generate_api_key", user_id=auth.user_id, tier=auth.tier)
+    add_breadcrumb("API key generation requested", category="api_keys", tier=auth.tier)
 
     logger.info(
         "API key generation requested",
         user_id=auth.user_id,
         key_name=request.name,
-        tier=request.tier
+        tier=auth.tier
     )
 
+    # Tier is locked to the user's auth tier (no self-escalation)
+    tier = auth.tier
+
+    # Enforce key limit per user
+    key_count = api_key_manager.count_keys(auth.user_id)
+    if key_count >= MAX_KEYS_PER_USER:
+        raise HTTPException(
+            status_code=403,
+            detail=f"Maximum {MAX_KEYS_PER_USER} active API keys allowed. Revoke an existing key first."
+        )
+
     try:
-        with track_time("generate_api_key", tier=request.tier):
-            new_key = api_key_manager.generate_key(
+        with track_time("generate_api_key", tier=tier):
+            result = api_key_manager.generate_key(
                 name=request.name,
-                tier=request.tier,
+                tier=tier,
                 user_id=auth.user_id
             )
 
         logger.info(
             "API key generated successfully",
             user_id=auth.user_id,
             key_name=request.name,
-            tier=request.tier
+            tier=tier
         )
 
         return {
-            "api_key": new_key,
-            "tier": request.tier,
+            "api_key": result["key"],
+            "id": result["id"],
+            "tier": tier,
             "name": request.name,
             "message": "Save this key securely - it won't be shown again"
         }
+    except HTTPException:
+        raise
     except Exception as e:
         logger.error(
             "API key generation failed",
@@ -93,6 +115,49 @@ async def generate_api_key(
         raise HTTPException(status_code=500, detail="Failed to generate API key")
 
 
+@router.get("/keys")
+async def list_api_keys(
+    auth: AuthContext = Depends(require_auth)
+) -> Dict[str, Any]:
+    """List all API keys for the authenticated user."""
+    if not auth.user_id:
+        raise HTTPException(status_code=401, detail="User ID required")
+
+    try:
+        keys = api_key_manager.list_keys(auth.user_id)
+        return {"keys": keys}
+    except Exception as e:
+        logger.error("Failed to list API keys", user_id=auth.user_id, error=str(e))
+        capture_exception(e, operation="list_api_keys", user_id=auth.user_id)
+        raise HTTPException(status_code=500, detail="Failed to list API keys")
+
+
+@router.delete("/keys/{key_id}")
+async def revoke_api_key(
+    key_id: UUID,
+    auth: AuthContext = Depends(require_auth)
+) -> Dict[str, Any]:
+    """Revoke an API key by ID. Soft-deletes (sets active=false)."""
+    if not auth.user_id:
+        raise HTTPException(status_code=401, detail="User ID required")
+
+    try:
+        success = api_key_manager.revoke_key_by_id(str(key_id), auth.user_id)
+        if not success:
+            raise HTTPException(
+                status_code=404,
+                detail="API key not found or not owned by you"
+            )
+        logger.info("API key revoked", user_id=auth.user_id, key_id=key_id)
+        return {"message": "API key revoked", "key_id": key_id}
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to revoke API key", user_id=auth.user_id, error=str(e))
+        capture_exception(e, operation="revoke_api_key", user_id=auth.user_id)
+        raise HTTPException(status_code=500, detail="Failed to revoke API key")
+
+
 @router.get("/keys/usage")
 async def get_api_usage(
     auth: AuthContext = Depends(require_auth)

backend/services/rate_limiter.py

@@ -177,21 +177,35 @@ class APIKeyManager:
     def __init__(self, supabase_client):
         self.db = supabase_client
 
-    def generate_key(self, name: str, tier: str = 'free', user_id: Optional[str] = None) -> str:
-        """Generate a new API key"""
-        # Generate secure random key
+    def generate_key(self, name: str, tier: str = 'free', user_id: Optional[str] = None) -> Dict:
+        """Generate a new API key. Returns dict with raw key + metadata."""
         key = f"ci_{secrets.token_urlsafe(32)}"
-
-        # Store in database
-        self.db.table("api_keys").insert({
+        # Persist last 8 chars of raw key for display (ci_...xYz12345)
+        suffix = key[-8:]
+
+        result = self.db.table("api_keys").insert({
             "key_hash": hashlib.sha256(key.encode()).hexdigest(),
+            "key_suffix": suffix,
             "name": name,
             "tier": tier,
             "user_id": user_id,
             "created_at": datetime.utcnow().isoformat()
         }).execute()
-
-        return key
+
+        row = result.data[0] if result.data else {}
+        return {
+            "key": key,
+            "id": row.get("id"),
+            "name": name,
+            "tier": tier,
+        }
+
+    def count_keys(self, user_id: str) -> int:
+        """Count active keys for a user."""
+        result = self.db.table("api_keys").select(
+            "id", count="exact"
+        ).eq("user_id", user_id).eq("active", True).execute()
+        return result.count or 0
 
     def verify_key(self, api_key: str) -> Optional[Dict]:
         """Verify API key and return metadata"""
@@ -204,7 +218,33 @@ def verify_key(self, api_key: str) -> Optional[Dict]:
         return result.data[0] if result.data else None
 
     def revoke_key(self, api_key: str) -> bool:
-        """Revoke an API key"""
+        """Revoke an API key by raw key value"""
         key_hash = hashlib.sha256(api_key.encode()).hexdigest()
         self.db.table("api_keys").update({"active": False}).eq("key_hash", key_hash).execute()
         return True
+
+    def revoke_key_by_id(self, key_id: str, user_id: str) -> bool:
+        """Revoke an API key by UUID. Verifies ownership via user_id."""
+        result = self.db.table("api_keys").update(
+            {"active": False}
+        ).eq("id", key_id).eq("user_id", user_id).execute()
+        return len(result.data) > 0
+
+    def list_keys(self, user_id: str) -> list:
+        """List all API keys for a user. Returns masked keys (no raw values)."""
+        result = self.db.table("api_keys").select(
+            "id, name, tier, active, created_at, last_used_at, key_suffix"
+        ).eq("user_id", user_id).order("created_at", desc=True).execute()
+        keys = []
+        for row in result.data:
+            suffix = row.get("key_suffix", "")
+            keys.append({
+                "id": row["id"],
+                "name": row["name"],
+                "tier": row["tier"],
+                "active": row["active"],
+                "created_at": row["created_at"],
+                "last_used_at": row.get("last_used_at"),
+                "key_preview": f"ci_...{suffix}" if suffix else "ci_...",
+            })
+        return keys

backend/tests/test_observability_routes.py

@@ -355,7 +355,13 @@ def mock_dependencies(self):
              patch("routes.api_keys.rate_limiter") as mock_limiter, \
              patch("routes.api_keys.metrics") as mock_metrics:
 
-            mock_manager.generate_key = MagicMock(return_value="sk-test-key-123")
+            mock_manager.generate_key = MagicMock(return_value={
+                "key": "ci_test-key-123",
+                "id": "test-uuid",
+                "name": "test",
+                "tier": "free",
+            })
+            mock_manager.count_keys = MagicMock(return_value=0)
             mock_limiter.get_usage = MagicMock(return_value={"minute": 5, "hour": 50})
             mock_metrics.get_metrics = MagicMock(return_value={"searches": 100})
 
@@ -371,7 +377,7 @@ async def test_generate_key_logs_request(self, mock_observability, mock_dependen
         from routes.api_keys import generate_api_key, CreateAPIKeyRequest
         from middleware.auth import AuthContext
 
-        request = CreateAPIKeyRequest(name="Production Key", tier="pro")
+        request = CreateAPIKeyRequest(name="Production Key")
         auth = AuthContext(user_id="user-123", email="test@test.com", tier="pro")
 
         await generate_api_key(request, auth)
@@ -387,7 +393,7 @@ async def test_generate_key_sets_context_with_tier(self, mock_observability, moc
         from routes.api_keys import generate_api_key, CreateAPIKeyRequest
         from middleware.auth import AuthContext
 
-        request = CreateAPIKeyRequest(name="Key", tier="enterprise")
+        request = CreateAPIKeyRequest(name="Key")
         auth = AuthContext(user_id="user", email="test@test.com", tier="enterprise")
 
         await generate_api_key(request, auth)

supabase/migrations/004_api_keys.sql

@@ -13,6 +13,7 @@ CREATE TABLE IF NOT EXISTS api_keys (
   user_id uuid REFERENCES auth.users(id) ON DELETE CASCADE,
   name text NOT NULL,
   key_hash text NOT NULL UNIQUE,
+  key_suffix text,  -- last 8 chars of raw key for masked display (ci_...xYz12345)
   tier text DEFAULT 'free' CHECK (tier IN ('free', 'pro', 'enterprise')),
   active boolean DEFAULT true,
   created_at timestamptz DEFAULT now(),