API Security Best Practices: Authentication, Rate Limiting, and Key Management (2026)

August 13, 2026 · 16 min read

AI APIs handle sensitive data and expensive compute. A single exposed API key or missing rate limit can lead to data breaches, cost explosions, or service abuse. This guide covers production-tested security patterns for AI API gateways, with specific examples for TokenEase and general best practices.

Security Threat Model for AI APIs

1. API Key Security

Never Expose Keys in Client-Side Code

// ❌ WRONG: Key visible in browser
const apiKey = "sk-live-abc123...";
fetch("https://tokenease.io/v1/chat/completions", {
    headers: { "Authorization": `Bearer ${apiKey}` }
});

// ✅ CORRECT: Proxy through your backend
// Frontend calls YOUR backend
fetch("/api/chat", {
    method: "POST",
    body: JSON.stringify({ message: userInput })
});

// Backend holds the key and calls TokenEase
@app.post("/api/chat")
async def chat_proxy(request: ChatRequest):
    response = requests.post(
        "https://tokenease.io/v1/chat/completions",
        headers={"Authorization": f"Bearer {BACKEND_API_KEY}"},
        json={"model": "deepseek-v4", "messages": request.messages}
    )
    return response.json()

Key Rotation Strategy

import secrets
import hashlib
from datetime import datetime, timedelta

class APIKeyManager:
    def __init__(self, db):
        self.db = db
    
    def generate_key(self, user_id, tier="standard"):
        """Generate a new API key with metadata"""
        raw_key = f"te_{secrets.token_urlsafe(32)}"
        key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
        
        self.db.execute("""
            INSERT INTO api_keys (user_id, key_hash, tier, created_at, expires_at)
            VALUES (?, ?, ?, ?, ?)
        """, (user_id, key_hash, tier, datetime.now(), 
              datetime.now() + timedelta(days=90)))
        
        return raw_key  # Show once, never stored in plain text
    
    def rotate_key(self, old_key_hash):
        """Rotate an existing key"""
        # Get user from old key
        user = self.db.execute(
            "SELECT user_id, tier FROM api_keys WHERE key_hash = ?",
            (old_key_hash,)
        ).fetchone()
        
        if not user:
            raise ValueError("Invalid key")
        
        # Generate new key
        new_key = self.generate_key(user["user_id"], user["tier"])
        
        # Schedule old key deprecation (grace period)
        self.db.execute("""
            UPDATE api_keys 
            SET deprecated_at = ?, grace_period_days = 7
            WHERE key_hash = ?
        """, (datetime.now(), old_key_hash))
        
        return new_key

# Auto-rotation every 90 days
@app.get("/admin/rotate-expiring-keys")
async def auto_rotate():
    expiring = db.execute("""
        SELECT key_hash FROM api_keys 
        WHERE expires_at < ? AND deprecated_at IS NULL
    """, (datetime.now() + timedelta(days=7),)).fetchall()
    
    for key in expiring:
        manager.rotate_key(key["key_hash"])
    
    return {"rotated": len(expiring)}

2. Rate Limiting Architecture

Multi-Tier Rate Limiting

import redis
from functools import wraps
import time

redis_client = redis.Redis(host="localhost", port=6379, db=0)

class RateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def check_limit(self, key, limit, window):
        """Token bucket rate limiting"""
        pipe = self.redis.pipeline()
        now = time.time()
        
        # Remove expired tokens
        pipe.zremrangebyscore(key, 0, now - window)
        
        # Count current tokens
        pipe.zcard(key)
        
        # Add current request
        pipe.zadd(key, {str(now): now})
        
        # Set expiry on the key
        pipe.expire(key, window)
        
        results = pipe.execute()
        current_count = results[1]
        
        return current_count <= limit
    
    def limit(self, key_prefix, limits):
        """Decorator for multi-tier rate limiting"""
        def decorator(func):
            @wraps(func)
            async def wrapper(request, *args, **kwargs):
                # Tier 1: Per-user limit (e.g., 100 req/min)
                user_key = f"{key_prefix}:user:{request.user_id}"
                if not self.check_limit(user_key, limits["per_user"], 60):
                    raise HTTPException(429, "User rate limit exceeded")
                
                # Tier 2: Per-IP limit (e.g., 20 req/min)
                ip_key = f"{key_prefix}:ip:{request.client.host}"
                if not self.check_limit(ip_key, limits["per_ip"], 60):
                    raise HTTPException(429, "IP rate limit exceeded")
                
                # Tier 3: Global limit (e.g., 10000 req/min)
                global_key = f"{key_prefix}:global"
                if not self.check_limit(global_key, limits["global"], 60):
                    raise HTTPException(429, "Global rate limit exceeded")
                
                return await func(request, *args, **kwargs)
            return wrapper
        return decorator

# Apply to endpoints
limiter = RateLimiter(redis_client)

@app.post("/v1/chat/completions")
@limiter.limit("chat", {"per_user": 100, "per_ip": 20, "global": 10000})
async def chat_completion(request: Request):
    # Process request
    pass

Cost-Based Rate Limiting

def estimate_cost(request_body):
    """Estimate token cost for rate limiting by dollar value"""
    model = request_body.get("model", "deepseek-v4")
    messages = request_body.get("messages", [])
    max_tokens = request_body.get("max_tokens", 1024)
    
    # Estimate input tokens
    input_text = " ".join([m["content"] for m in messages])
    input_tokens = len(input_text.split()) * 1.3  # Rough estimate
    
    # Pricing per 1K tokens
    pricing = {
        "deepseek-v4": {"input": 0.0005, "output": 0.002},
        "glm-4": {"input": 0.0007, "output": 0.0021},
        "kimi-k2": {"input": 0.0003, "output": 0.0012}
    }
    
    p = pricing.get(model, pricing["deepseek-v4"])
    estimated_cost = (input_tokens / 1000 * p["input"] + 
                     max_tokens / 1000 * p["output"])
    
    return estimated_cost

# Rate limit by dollar value instead of request count
@app.post("/v1/chat/completions")
async def chat_with_cost_limit(request: Request):
    body = await request.json()
    cost = estimate_cost(body)
    
    user_id = get_user_id(request)
    daily_spend = get_daily_spend(user_id)
    
    if daily_spend + cost > DAILY_BUDGET:
        raise HTTPException(429, f"Daily budget exceeded. Limit: ${DAILY_BUDGET}")
    
    # Process and track actual cost
    response = await process_chat(body)
    actual_cost = calculate_actual_cost(response)
    track_spend(user_id, actual_cost)
    
    return response

3. Request Authentication Patterns

HMAC Request Signing

import hmac
import hashlib
import base64
from datetime import datetime

def sign_request(method, path, body, timestamp, api_secret):
    """Create HMAC-SHA256 signature for request authentication"""
    
    # Canonical request string
    body_hash = hashlib.sha256(body.encode()).hexdigest()
    canonical = f"{method}\n{path}\n{timestamp}\n{body_hash}"
    
    # Sign with secret
    signature = hmac.new(
        api_secret.encode(),
        canonical.encode(),
        hashlib.sha256
    ).hexdigest()
    
    return signature

# Client side
import requests
import time

API_KEY = "your_key"
API_SECRET = "your_secret"  # Never transmit, only used for signing

def make_signed_request(method, path, body):
    timestamp = str(int(time.time()))
    signature = sign_request(method, path, body, timestamp, API_SECRET)
    
    headers = {
        "X-API-Key": API_KEY,
        "X-Timestamp": timestamp,
        "X-Signature": signature,
        "Content-Type": "application/json"
    }
    
    return requests.request(method, f"https://api.example.com{path}",
                          headers=headers, data=body)

# Server side verification
def verify_signature(request):
    api_key = request.headers.get("X-API-Key")
    timestamp = request.headers.get("X-Timestamp")
    signature = request.headers.get("X-Signature")
    
    # Reject old requests (prevent replay)
    if abs(int(time.time()) - int(timestamp)) > 300:  # 5 min window
        return False
    
    # Look up secret
    secret = get_secret_for_key(api_key)
    if not secret:
        return False
    
    # Reconstruct and verify
    body = request.body()
    body_hash = hashlib.sha256(body.encode()).hexdigest()
    canonical = f"{request.method}\n{request.path}\n{timestamp}\n{body_hash}"
    
    expected = hmac.new(secret.encode(), canonical.encode(), 
                       hashlib.sha256).hexdigest()
    
    return hmac.compare_digest(signature, expected)

4. Input Validation and Sanitization

Prompt Injection Prevention

import re

class InputValidator:
    FORBIDDEN_PATTERNS = [
        r"ignore previous instructions",
        r"disregard (all|previous) (instructions|prompts)",
        r"system prompt",
        r"you are now",
        r"DAN mode",
        r"jailbreak",
        r"\[\s*SYSTEM\s*\]"
    ]
    
    MAX_PROMPT_LENGTH = 10000
    MAX_MESSAGES = 50
    
    @classmethod
    def validate_chat_request(cls, request_body):
        """Validate and sanitize chat completion requests"""
        errors = []
        
        # Check message count
        messages = request_body.get("messages", [])
        if len(messages) > cls.MAX_MESSAGES:
            errors.append(f"Too many messages: {len(messages)} > {cls.MAX_MESSAGES}")
        
        # Check message content
        for i, msg in enumerate(messages):
            content = msg.get("content", "")
            
            # Length check
            if len(content) > cls.MAX_PROMPT_LENGTH:
                errors.append(f"Message {i} too long: {len(content)} chars")
            
            # Injection pattern check
            lower_content = content.lower()
            for pattern in cls.FORBIDDEN_PATTERNS:
                if re.search(pattern, lower_content):
                    errors.append(f"Message {i} contains forbidden pattern")
        
        # Validate model
        allowed_models = ["deepseek-v4", "glm-4", "kimi-k2", "qwen-max", "doubao-pro"]
        if request_body.get("model") not in allowed_models:
            errors.append(f"Invalid model: {request_body.get('model')}")
        
        # Validate max_tokens
        max_tokens = request_body.get("max_tokens", 1024)
        if max_tokens > 8192:
            errors.append("max_tokens exceeds limit of 8192")
        
        if errors:
            raise ValueError("; ".join(errors))
        
        return True

# Middleware
@app.post("/v1/chat/completions")
async def secure_chat(request: Request):
    body = await request.json()
    
    try:
        InputValidator.validate_chat_request(body)
    except ValueError as e:
        raise HTTPException(400, str(e))
    
    # Proceed with validated request
    return await process_chat(body)

5. Monitoring and Alerting

from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class SecurityEvent:
    timestamp: datetime
    event_type: str
    severity: str  # low, medium, high, critical
    source_ip: str
    user_id: str
    details: dict

class SecurityMonitor:
    ALERT_THRESHOLDS = {
        "failed_auth": 10,      # per minute
        "rate_limit_hit": 50,   # per minute
        "cost_spike": 5.0,      # 5x normal spend
        "error_rate": 0.25      # 25% error rate
    }
    
    def log_event(self, event: SecurityEvent):
        """Log security event for analysis"""
        # Write to security log
        with open("/var/log/security_events.jsonl", "a") as f:
            f.write(json.dumps({
                "timestamp": event.timestamp.isoformat(),
                "type": event.event_type,
                "severity": event.severity,
                "ip": event.source_ip,
                "user": event.user_id,
                "details": event.details
            }) + "\n")
        
        # Alert on critical events
        if event.severity == "critical":
            self.send_alert(event)
    
    def check_anomalies(self, window_minutes=5):
        """Detect anomalous patterns"""
        # Check for brute force
        recent_failed = count_failed_auth(window_minutes)
        if recent_failed > self.ALERT_THRESHOLDS["failed_auth"]:
            self.send_alert(SecurityEvent(
                timestamp=datetime.now(),
                event_type="potential_brute_force",
                severity="high",
                source_ip="multiple",
                user_id="unknown",
                details={"failed_attempts": recent_failed}
            ))
        
        # Check for cost spikes
        current_spend = get_current_spend(window_minutes)
        baseline = get_baseline_spend(window_minutes)
        if baseline > 0 and current_spend / baseline > self.ALERT_THRESHOLDS["cost_spike"]:
            self.send_alert(SecurityEvent(
                timestamp=datetime.now(),
                event_type="cost_spike",
                severity="high",
                source_ip="N/A",
                user_id="N/A",
                details={"current": current_spend, "baseline": baseline}
            ))

# Usage in request handlers
@app.exception_handler(401)
async def auth_failure(request, exc):
    monitor.log_event(SecurityEvent(
        timestamp=datetime.now(),
        event_type="auth_failure",
        severity="medium",
        source_ip=request.client.host,
        user_id=get_user_id(request) or "unknown",
        details={"path": request.url.path}
    ))
    return JSONResponse({"error": "Unauthorized"}, 401)

Security Checklist for Production

• Transport — TLS 1.3 only, disable HTTP
• Authentication — API keys + optional HMAC signing
• Authorization — Role-based access control (RBAC)
• Rate Limiting — Per-user, per-IP, global, and cost-based
• Input Validation — Schema validation, prompt injection filters
• Output Sanitization — PII detection, content filtering
• Logging — Request/response logging without sensitive data
• Monitoring — Anomaly detection, cost alerts, error rate tracking
• Key Management — Rotation, hashing, never log in plain text
• Incident Response — Automated key revocation, IP blocking

TokenEase Security Features

TokenEase implements these security patterns out of the box:

• API key hashing — Keys are stored as SHA-256 hashes, never in plain text
• Automatic rate limiting — Per-key request limits enforced at the gateway
• Request signing — Optional HMAC-SHA256 for webhook verification
• IP-based restrictions — Configure allowed IP ranges per key
• Usage analytics — Real-time spend tracking and anomaly detection

Next Steps

Secure your AI API integration:

1. Audit your current API key storage — ensure no keys in client-side code
2. Implement rate limiting with Redis or your cloud provider
3. Add input validation for all LLM prompts
4. Set up monitoring for cost spikes and authentication failures
5. Get a TokenEase API key with built-in security controls

For more on production patterns, see our guides on failover and load balancing and A/B testing AI models.

Last updated: August 2026. Security best practices evolve with threat landscapes.