Add API key authentication system (Telegram-style)

Auth features: - Token generation with ocv_ prefix - Rate limiting per minute - Monthly minute quotas - Usage tracking - Key revocation - Pricing tiers: free (60min), pro (500min), enterprise (unlimited) API endpoints: - POST /api/keys - Create new API key (requires master key) - GET /api/usage - Check usage stats WebSocket auth: - Pass api_key as query param or x-api-key header - Error codes: 4001 (key required), 4002 (invalid), 4003 (rate limited) Client updates: - Reads API key from URL params or localStorage - Shows auth errors clearly Tests: 25 passing (12 new auth tests)
2026-08-14 09:02:28 +00:00 · 2026-01-30 13:39:19 -05:00
parent 87609ca5e0
commit b2ac1dc0cd
5 changed files with 542 additions and 5 deletions
@@ -163,6 +163,30 @@ Connect to `ws://localhost:8765/ws` and send/receive JSON messages:
 - [ ] Docker GPU support
 - [ ] Kubernetes Helm chart

+## Hosted Service (Coming Soon)
+
+Don't want to self-host? We offer a managed service:
+
+| Tier | Minutes/Month | Price | Features |
+|------|---------------|-------|----------|
+| **Free** | 60 | $0 | Basic voice chat |
+| **Pro** | 500 | $29/mo | + Voice cloning |
+| **Enterprise** | Unlimited | $99/mo | + Priority, SLA |
+
+**API Key Authentication:**
+
+```bash
+# Get an API key
+curl -X POST "https://voice.openclaw.dev/api/keys?name=myapp&tier=pro" \
+     -H "x-master-key: YOUR_MASTER_KEY"
+
+# Connect with API key
+wss://voice.openclaw.dev/ws?api_key=ocv_xxxxx
+
+# Check usage
+curl "https://voice.openclaw.dev/api/usage?api_key=ocv_xxxxx"
+```
+
 ## Cost Comparison

 | Platform | Cost/Minute |
@@ -170,6 +194,7 @@ Connect to `ws://localhost:8765/ws` and send/receive JSON messages:
 | ElevenLabs Conversational AI | $0.08-0.12 |
 | Retell.ai | $0.13-0.31 |
 | Vapi.ai | $0.05-0.15 |
+| **OpenClaw Voice (hosted)** | **~$0.06** |
 | **OpenClaw Voice (self-hosted)** | **~$0.003** |

 ## Contributing
@@ -286,10 +286,19 @@
        let silenceTimer = null;
        let silenceThreshold = 1500; // ms of silence before stopping
        
+        // Get API key from URL params or localStorage
+        function getApiKey() {
+            const urlParams = new URLSearchParams(window.location.search);
+            return urlParams.get('api_key') || localStorage.getItem('openclaw_api_key') || '';
+        }
+        
        // Connect to WebSocket
        function connect() {
            const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
-            const wsUrl = `${protocol}//${window.location.host}/ws`;
+            const apiKey = getApiKey();
+            const wsUrl = apiKey 
+                ? `${protocol}//${window.location.host}/ws?api_key=${apiKey}`
+                : `${protocol}//${window.location.host}/ws`;
            
            ws = new WebSocket(wsUrl);
            
@@ -298,9 +307,20 @@
                errorEl.textContent = '';
            };
            
-            ws.onclose = () => {
-                setStatus('Disconnected - reconnecting...');
-                setTimeout(connect, 2000);
+            ws.onclose = (event) => {
+                if (event.code === 4001) {
+                    setStatus('❌ API key required');
+                    errorEl.textContent = 'This server requires an API key. Add ?api_key=YOUR_KEY to the URL.';
+                } else if (event.code === 4002) {
+                    setStatus('❌ Invalid API key');
+                    errorEl.textContent = 'Your API key is invalid or expired.';
+                } else if (event.code === 4003) {
+                    setStatus('❌ Rate limited');
+                    errorEl.textContent = 'Too many requests. Please wait and try again.';
+                } else {
+                    setStatus('Disconnected - reconnecting...');
+                    setTimeout(connect, 2000);
+                }
            };
            
            ws.onerror = (e) => {
@@ -0,0 +1,243 @@
+"""
+Authentication and API key management.
+
+Token system like Telegram Bot API:
+- Users get an API key to connect their voice widget
+- Keys can be scoped (rate limits, features)
+- Hosted version charges per minute or monthly
+"""
+
+import secrets
+import hashlib
+import time
+from typing import Optional, Dict, Any
+from dataclasses import dataclass, field
+from datetime import datetime
+from loguru import logger
+
+
+@dataclass
+class APIKey:
+    """API key with metadata and limits."""
+    key_id: str
+    key_hash: str  # Store hash, not plaintext
+    name: str
+    created_at: datetime
+    
+    # Limits
+    rate_limit_per_minute: int = 60  # requests per minute
+    monthly_minutes: Optional[int] = None  # None = unlimited
+    
+    # Usage tracking
+    minutes_used: float = 0.0
+    last_request_at: Optional[datetime] = None
+    request_count_this_minute: int = 0
+    
+    # Features
+    features: Dict[str, bool] = field(default_factory=lambda: {
+        "continuous_mode": True,
+        "voice_cloning": False,
+        "priority_queue": False,
+    })
+    
+    # Status
+    active: bool = True
+    tier: str = "free"  # free, pro, enterprise
+
+
+class TokenManager:
+    """
+    Manage API tokens for voice connections.
+    
+    In production, this would be backed by a database.
+    For MVP, we use in-memory storage + env vars.
+    """
+    
+    def __init__(self):
+        self._keys: Dict[str, APIKey] = {}
+        self._key_to_id: Dict[str, str] = {}  # hash -> key_id lookup
+    
+    def generate_key(
+        self,
+        name: str,
+        tier: str = "free",
+        rate_limit: int = 60,
+        monthly_minutes: Optional[int] = None,
+    ) -> tuple[str, APIKey]:
+        """
+        Generate a new API key.
+        
+        Returns:
+            (plaintext_key, APIKey object)
+            
+        Note: Plaintext key is only returned once!
+        """
+        # Generate secure random key
+        key_id = secrets.token_hex(8)
+        plaintext_key = f"ocv_{secrets.token_urlsafe(32)}"
+        key_hash = self._hash_key(plaintext_key)
+        
+        api_key = APIKey(
+            key_id=key_id,
+            key_hash=key_hash,
+            name=name,
+            created_at=datetime.now(tz=None),
+            rate_limit_per_minute=rate_limit,
+            monthly_minutes=monthly_minutes,
+            tier=tier,
+        )
+        
+        self._keys[key_id] = api_key
+        self._key_to_id[key_hash] = key_id
+        
+        logger.info(f"Generated API key: {key_id} ({name}, tier={tier})")
+        
+        return plaintext_key, api_key
+    
+    def validate_key(self, plaintext_key: str) -> Optional[APIKey]:
+        """
+        Validate an API key and return its metadata.
+        
+        Returns None if invalid.
+        """
+        if not plaintext_key or not plaintext_key.startswith("ocv_"):
+            return None
+        
+        key_hash = self._hash_key(plaintext_key)
+        key_id = self._key_to_id.get(key_hash)
+        
+        if not key_id:
+            return None
+        
+        api_key = self._keys.get(key_id)
+        
+        if not api_key or not api_key.active:
+            return None
+        
+        return api_key
+    
+    def check_rate_limit(self, api_key: APIKey) -> bool:
+        """
+        Check if request is within rate limits.
+        
+        Returns True if allowed, False if rate limited.
+        """
+        now = datetime.now(tz=None)
+        
+        # Reset counter if new minute
+        if api_key.last_request_at:
+            elapsed = (now - api_key.last_request_at).total_seconds()
+            if elapsed >= 60:
+                api_key.request_count_this_minute = 0
+        
+        # Check rate limit
+        if api_key.request_count_this_minute >= api_key.rate_limit_per_minute:
+            return False
+        
+        # Update counters
+        api_key.request_count_this_minute += 1
+        api_key.last_request_at = now
+        
+        return True
+    
+    def check_monthly_quota(self, api_key: APIKey, minutes: float = 0) -> bool:
+        """
+        Check if within monthly minute quota.
+        
+        Returns True if allowed, False if quota exceeded.
+        """
+        if api_key.monthly_minutes is None:
+            return True  # Unlimited
+        
+        return (api_key.minutes_used + minutes) <= api_key.monthly_minutes
+    
+    def record_usage(self, api_key: APIKey, minutes: float):
+        """Record minutes used for billing."""
+        api_key.minutes_used += minutes
+        logger.debug(f"Key {api_key.key_id}: used {minutes:.2f} min, total {api_key.minutes_used:.2f}")
+    
+    def get_usage(self, api_key: APIKey) -> Dict[str, Any]:
+        """Get usage stats for an API key."""
+        return {
+            "key_id": api_key.key_id,
+            "name": api_key.name,
+            "tier": api_key.tier,
+            "minutes_used": round(api_key.minutes_used, 2),
+            "monthly_limit": api_key.monthly_minutes,
+            "rate_limit": api_key.rate_limit_per_minute,
+            "features": api_key.features,
+        }
+    
+    def revoke_key(self, key_id: str) -> bool:
+        """Revoke an API key."""
+        if key_id in self._keys:
+            self._keys[key_id].active = False
+            logger.info(f"Revoked API key: {key_id}")
+            return True
+        return False
+    
+    def _hash_key(self, plaintext_key: str) -> str:
+        """Hash an API key for storage."""
+        return hashlib.sha256(plaintext_key.encode()).hexdigest()
+
+
+# Global token manager instance
+token_manager = TokenManager()
+
+
+# Helper to load keys from environment
+def load_keys_from_env():
+    """
+    Load API keys from environment variables.
+    
+    Format: OPENCLAW_API_KEY_<name>=<plaintext_key>
+    
+    For production, use a database instead.
+    """
+    import os
+    
+    # Check for master key (allows all access)
+    master_key = os.getenv("OPENCLAW_MASTER_KEY")
+    if master_key:
+        # Register master key
+        key_hash = token_manager._hash_key(master_key)
+        api_key = APIKey(
+            key_id="master",
+            key_hash=key_hash,
+            name="Master Key",
+            created_at=datetime.now(tz=None),
+            rate_limit_per_minute=1000,
+            monthly_minutes=None,
+            tier="enterprise",
+        )
+        api_key.features = {
+            "continuous_mode": True,
+            "voice_cloning": True,
+            "priority_queue": True,
+        }
+        token_manager._keys["master"] = api_key
+        token_manager._key_to_id[key_hash] = "master"
+        logger.info("Loaded master API key from environment")
+
+
+# Pricing tiers for hosted version
+PRICING_TIERS = {
+    "free": {
+        "monthly_minutes": 60,
+        "rate_limit": 30,
+        "price": 0,
+        "features": ["continuous_mode"],
+    },
+    "pro": {
+        "monthly_minutes": 500,
+        "rate_limit": 120,
+        "price": 29,  # $/month
+        "features": ["continuous_mode", "voice_cloning"],
+    },
+    "enterprise": {
+        "monthly_minutes": None,  # Unlimited
+        "rate_limit": 500,
+        "price": 99,  # $/month
+        "features": ["continuous_mode", "voice_cloning", "priority_queue"],
+    },
+}
@@ -27,6 +27,7 @@ from .stt import WhisperSTT
 from .tts import ChatterboxTTS
 from .backend import AIBackend
 from .vad import VoiceActivityDetector
+from .auth import token_manager, load_keys_from_env, APIKey


 class Settings(BaseSettings):
@@ -36,6 +37,10 @@ class Settings(BaseSettings):
    host: str = "0.0.0.0"
    port: int = 8765
    
+    # Auth
+    require_auth: bool = False  # Set True for production
+    master_key: Optional[str] = None  # Admin key for full access
+    
    # STT
    stt_model: str = "base"  # tiny, base, small, medium, large-v3-turbo
    stt_device: str = "auto"  # auto, cpu, cuda, mps
@@ -75,6 +80,13 @@ async def startup():
    
    logger.info("Initializing OpenClaw Voice server...")
    
+    # Load API keys
+    load_keys_from_env()
+    if settings.require_auth:
+        logger.info("🔐 Authentication ENABLED")
+    else:
+        logger.warning("⚠️ Authentication DISABLED (dev mode)")
+    
    # Initialize STT
    logger.info(f"Loading STT model: {settings.stt_model}")
    stt = WhisperSTT(
@@ -110,14 +122,103 @@ async def index():
    return FileResponse("src/client/index.html")


+@app.post("/api/keys")
+async def create_api_key(
+    name: str,
+    tier: str = "free",
+    master_key: Optional[str] = None,
+):
+    """
+    Create a new API key (requires master key).
+    
+    curl -X POST "http://localhost:8765/api/keys?name=myapp&tier=pro" \
+         -H "x-master-key: YOUR_MASTER_KEY"
+    """
+    # Verify master key
+    if settings.require_auth:
+        if not master_key and not settings.master_key:
+            return {"error": "Master key required"}
+        
+        provided_key = master_key or ""
+        if provided_key != settings.master_key:
+            # Also check if it's a valid master-tier key
+            key = token_manager.validate_key(provided_key)
+            if not key or key.tier != "enterprise":
+                return {"error": "Invalid master key"}
+    
+    from .auth import PRICING_TIERS
+    
+    if tier not in PRICING_TIERS:
+        return {"error": f"Invalid tier. Options: {list(PRICING_TIERS.keys())}"}
+    
+    tier_config = PRICING_TIERS[tier]
+    
+    plaintext_key, api_key = token_manager.generate_key(
+        name=name,
+        tier=tier,
+        rate_limit=tier_config["rate_limit"],
+        monthly_minutes=tier_config["monthly_minutes"],
+    )
+    
+    return {
+        "api_key": plaintext_key,  # Only shown once!
+        "key_id": api_key.key_id,
+        "name": api_key.name,
+        "tier": api_key.tier,
+        "monthly_minutes": api_key.monthly_minutes,
+        "rate_limit": api_key.rate_limit_per_minute,
+    }
+
+
+@app.get("/api/usage")
+async def get_usage(api_key: str):
+    """
+    Get usage stats for an API key.
+    
+    curl "http://localhost:8765/api/usage?api_key=ocv_xxx"
+    """
+    key = token_manager.validate_key(api_key)
+    if not key:
+        return {"error": "Invalid API key"}
+    
+    return token_manager.get_usage(key)
+
+
@app.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket):
    """Handle voice WebSocket connections."""
+    # Check for API key in query params or headers
+    api_key_str = websocket.query_params.get("api_key") or \
+                  websocket.headers.get("x-api-key")
+    
+    api_key: Optional[APIKey] = None
+    
+    if settings.require_auth:
+        if not api_key_str:
+            await websocket.close(code=4001, reason="API key required")
+            return
+        
+        api_key = token_manager.validate_key(api_key_str)
+        if not api_key:
+            await websocket.close(code=4002, reason="Invalid API key")
+            return
+        
+        if not token_manager.check_rate_limit(api_key):
+            await websocket.close(code=4003, reason="Rate limit exceeded")
+            return
+        
+        logger.info(f"Client connected: {api_key.name} (tier={api_key.tier})")
+    else:
+        # Dev mode - allow all
+        if api_key_str:
+            api_key = token_manager.validate_key(api_key_str)
+        logger.info("Client connected (auth disabled)")
+    
    await websocket.accept()
-    logger.info("Client connected")
    
    audio_buffer = []
    is_listening = False
+    session_start = None
    
    try:
        while True:
@@ -0,0 +1,148 @@
+"""
+Tests for authentication module.
+"""
+
+import pytest
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from src.server.auth import TokenManager, APIKey, PRICING_TIERS
+
+
+class TestTokenManager:
+    """Tests for token management."""
+    
+    def test_generate_key(self):
+        """Test API key generation."""
+        tm = TokenManager()
+        plaintext, api_key = tm.generate_key("test-app")
+        
+        assert plaintext.startswith("ocv_")
+        assert len(plaintext) > 40
+        assert api_key.name == "test-app"
+        assert api_key.active
+    
+    def test_validate_key_success(self):
+        """Test validating a valid key."""
+        tm = TokenManager()
+        plaintext, _ = tm.generate_key("test-app")
+        
+        result = tm.validate_key(plaintext)
+        assert result is not None
+        assert result.name == "test-app"
+    
+    def test_validate_key_invalid(self):
+        """Test validating an invalid key."""
+        tm = TokenManager()
+        
+        assert tm.validate_key("invalid") is None
+        assert tm.validate_key("ocv_invalid") is None
+        assert tm.validate_key("") is None
+        assert tm.validate_key(None) is None
+    
+    def test_rate_limit(self):
+        """Test rate limiting."""
+        tm = TokenManager()
+        _, api_key = tm.generate_key("test", rate_limit=5)
+        
+        # Should allow up to rate limit
+        for i in range(5):
+            assert tm.check_rate_limit(api_key) is True
+        
+        # Should block after limit
+        assert tm.check_rate_limit(api_key) is False
+    
+    def test_monthly_quota(self):
+        """Test monthly quota checking."""
+        tm = TokenManager()
+        _, api_key = tm.generate_key("test", monthly_minutes=10)
+        
+        # Should allow within quota
+        assert tm.check_monthly_quota(api_key, 5) is True
+        
+        # Should block over quota
+        assert tm.check_monthly_quota(api_key, 15) is False
+        
+        # Record usage
+        tm.record_usage(api_key, 8)
+        assert api_key.minutes_used == 8
+        
+        # Now over quota
+        assert tm.check_monthly_quota(api_key, 3) is False
+    
+    def test_unlimited_quota(self):
+        """Test unlimited quota (None)."""
+        tm = TokenManager()
+        _, api_key = tm.generate_key("test", monthly_minutes=None)
+        
+        # Should always allow
+        assert tm.check_monthly_quota(api_key, 10000) is True
+    
+    def test_revoke_key(self):
+        """Test key revocation."""
+        tm = TokenManager()
+        plaintext, api_key = tm.generate_key("test")
+        
+        # Key should be valid
+        assert tm.validate_key(plaintext) is not None
+        
+        # Revoke it
+        assert tm.revoke_key(api_key.key_id) is True
+        
+        # Now invalid
+        assert tm.validate_key(plaintext) is None
+    
+    def test_get_usage(self):
+        """Test usage stats retrieval."""
+        tm = TokenManager()
+        _, api_key = tm.generate_key("test", tier="pro")
+        tm.record_usage(api_key, 5.5)
+        
+        usage = tm.get_usage(api_key)
+        
+        assert usage["name"] == "test"
+        assert usage["tier"] == "pro"
+        assert usage["minutes_used"] == 5.5
+    
+    def test_tiers(self):
+        """Test different pricing tiers."""
+        tm = TokenManager()
+        
+        # Free tier
+        _, free_key = tm.generate_key("free-user", tier="free", 
+                                       rate_limit=30, monthly_minutes=60)
+        assert free_key.tier == "free"
+        assert free_key.monthly_minutes == 60
+        
+        # Pro tier
+        _, pro_key = tm.generate_key("pro-user", tier="pro",
+                                      rate_limit=120, monthly_minutes=500)
+        assert pro_key.tier == "pro"
+        assert pro_key.monthly_minutes == 500
+
+
+class TestPricingTiers:
+    """Test pricing tier configuration."""
+    
+    def test_tiers_exist(self):
+        """Test all expected tiers exist."""
+        assert "free" in PRICING_TIERS
+        assert "pro" in PRICING_TIERS
+        assert "enterprise" in PRICING_TIERS
+    
+    def test_free_tier(self):
+        """Test free tier config."""
+        free = PRICING_TIERS["free"]
+        assert free["price"] == 0
+        assert free["monthly_minutes"] == 60
+    
+    def test_enterprise_unlimited(self):
+        """Test enterprise has unlimited minutes."""
+        enterprise = PRICING_TIERS["enterprise"]
+        assert enterprise["monthly_minutes"] is None
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])