Spaces:
Running
Running
File size: 3,952 Bytes
48e7c93 ba9d716 48e7c93 |
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 |
"""
Authentication Module
---------------------
Shared authentication logic for dashboard vs external API access.
"""
from fastapi import HTTPException, Request, Header
from typing import Optional
from db import get_supabase
from config import DEMO_API_KEY
# List of allowed origins/paths that don't need API key (dashboard access)
DASHBOARD_PATHS = ["/", "/docs/public", "/docs", "/static/"]
DASHBOARD_HOSTS = ["localhost", "127.0.0.1"] # Add your domain here when deployed
async def verify_api_key(
request: Request,
authorization: Optional[str] = Header(None),
x_api_key: Optional[str] = Header(None)
):
"""
Verify Bearer Token or X-API-KEY.
Dashboard requests (from same origin) don't need API key.
External API calls (from other origins) require API key.
Web searches require key but don't deduct tokens.
Returns: key_data (dict)
Raises: HTTPException if invalid
"""
token = None
if authorization:
parts = authorization.split()
if len(parts) == 2 and parts[0].lower() == "bearer":
token = parts[1]
if not token and x_api_key:
token = x_api_key
# Check if request is coming from dashboard (same origin)
referer = request.headers.get("referer", "")
origin = request.headers.get("origin", "")
# Check if referer/origin matches dashboard
is_dashboard_request = False
# Check if referer contains dashboard paths
for path in DASHBOARD_PATHS:
if path in referer:
is_dashboard_request = True
break
# Also check if origin is localhost (local development)
for host in DASHBOARD_HOSTS:
if host in origin or host in referer:
is_dashboard_request = True
break
# Check if it's a browser request (has Accept: text/html)
accept_header = request.headers.get("accept", "")
if "text/html" in accept_header and (referer or origin):
is_dashboard_request = True
if not token:
if is_dashboard_request:
# Dashboard access - no key needed
return {"id": "dashboard", "name": "Dashboard User", "limit_tokens": -1, "is_dashboard": True}
else:
# External API call - key required
raise HTTPException(
status_code=401,
detail="Authorization header required. Use 'Authorization: Bearer YOUR_API_KEY'. Get a key from the dashboard."
)
# 1. Check Demo Key
if token == DEMO_API_KEY:
return {"id": "demo", "name": "Demo User", "limit_tokens": -1, "is_dashboard": False}
# 2. Check Database (Non-blocking)
import asyncio
loop = asyncio.get_event_loop()
def _sync_check():
supabase = get_supabase()
if not supabase:
raise HTTPException(status_code=503, detail="Service unavailable")
res = supabase.table("kaiapi_api_keys").select("*").eq("token", token).execute()
if not res.data:
return None
return res.data[0]
try:
key_data = await loop.run_in_executor(None, _sync_check)
if not key_data:
raise HTTPException(status_code=401, detail="Incorrect API key provided")
if not key_data.get("is_active", True):
raise HTTPException(status_code=403, detail="API Key is inactive")
# Check limits
current_usage = key_data.get("usage_tokens", 0)
limit = key_data.get("limit_tokens", 0)
if limit > 0 and current_usage >= limit:
raise HTTPException(status_code=429, detail="You have exceeded your current quota")
key_data["is_dashboard"] = False
return key_data
except HTTPException:
raise
except Exception as e:
print(f"Auth Error: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
|