API Authentication Architecture FAQ & Answers

Use constant-time comparison for API key validation to prevent timing attacks that could enumerate valid keys. Standard pattern: (1) Hash the API key client-side with a cryptographic hash (SHA-256), never store plaintext keys. (2) Query database for hashed key. (3) Use constant-time comparison for final validation. Implementation: import hmac; def validate_api_key(provided_key: str, stored_hash: str) -> bool: provided_hash = hashlib.sha256(provided_key.encode()).hexdigest(); return hmac.compare_digest(provided_hash, stored_hash). Why hmac.compare_digest(): Python's == operator short-circuits (returns False on first mismatched byte), leaking timing information about how many characters matched. hmac.compare_digest() compares entire strings in constant time regardless of differences. Database schema: CREATE TABLE api_keys (id UUID PRIMARY KEY, key_hash TEXT NOT NULL, user_id UUID NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), last_used_at TIMESTAMPTZ, status TEXT DEFAULT 'active', CONSTRAINT unique_key_hash UNIQUE(key_hash)). Query pattern: SELECT user_id, status FROM api_keys WHERE key_hash = $1 AND status = 'active' (indexed lookup on key_hash). Key generation: import secrets; api_key = 'ak_' + secrets.token_urlsafe(32) generates cryptographically secure random key. Store hash: key_hash = hashlib.sha256(api_key.encode()).hexdigest(). Security considerations: (1) Hash function must be consistent (SHA-256 standard). (2) Database query time should be constant (indexed lookup). (3) Return generic error for invalid keys (don't reveal if key exists but inactive). (4) Rate limit validation attempts (5 per minute per IP). (5) Log failed attempts for monitoring. Timing attack example: Attacker measures response time for key_hash='aaa...' (10ms) vs key_hash='zzz...' (12ms). Differences reveal information about database state or comparison logic. Constant-time prevents this. Production example: async def validate_api_key(key: str) -> User | None: key_hash = hashlib.sha256(key.encode()).hexdigest(); row = await db.fetchrow('SELECT user_id, status FROM api_keys WHERE key_hash = $1', key_hash); if row and row['status'] == 'active': return await get_user(row['user_id']); return None. Best practice: Always use hmac.compare_digest() for secret comparisons (passwords, tokens, API keys), index key_hash column for O(1) lookup, rotate keys every 90 days, implement key revocation endpoint. Essential for secure API key authentication.

Standard API key management schema includes users, api_keys, access_tiers, usage_logs, and daily_usage tables for tracking quotas and rate limits. Complete schema: CREATE TABLE users (id UUID PRIMARY KEY DEFAULT gen_random_uuid(), email TEXT NOT NULL UNIQUE, name TEXT, tier_id UUID NOT NULL REFERENCES access_tiers(id), status TEXT DEFAULT 'active', created_at TIMESTAMPTZ DEFAULT NOW()); CREATE TABLE access_tiers (id UUID PRIMARY KEY, name TEXT NOT NULL, daily_quota INTEGER NOT NULL, rate_limit_per_minute INTEGER NOT NULL, features JSONB, price_monthly DECIMAL(10,2)); CREATE TABLE api_keys (id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, key_hash TEXT NOT NULL UNIQUE, name TEXT, scopes TEXT[], created_at TIMESTAMPTZ DEFAULT NOW(), last_used_at TIMESTAMPTZ, expires_at TIMESTAMPTZ, status TEXT DEFAULT 'active', ip_whitelist INET[]); CREATE TABLE usage_logs (id BIGSERIAL PRIMARY KEY, user_id UUID NOT NULL, api_key_id UUID, endpoint TEXT NOT NULL, method TEXT NOT NULL, status_code INTEGER, response_time_ms INTEGER, created_at TIMESTAMPTZ DEFAULT NOW()); CREATE TABLE daily_usage (user_id UUID NOT NULL, date DATE NOT NULL, request_count INTEGER DEFAULT 0, PRIMARY KEY (user_id, date));. Quota enforcement function: CREATE OR REPLACE FUNCTION check_quota_exceeded(p_user_id UUID, p_tier_quota INTEGER) RETURNS BOOLEAN AS $$ DECLARE current_usage INTEGER; BEGIN SELECT COALESCE(request_count, 0) INTO current_usage FROM daily_usage WHERE user_id = p_user_id AND date = CURRENT_DATE; RETURN current_usage >= p_tier_quota; END; $$ LANGUAGE plpgsql;. Usage increment: INSERT INTO daily_usage (user_id, date, request_count) VALUES ($1, CURRENT_DATE, 1) ON CONFLICT (user_id, date) DO UPDATE SET request_count = daily_usage.request_count + 1;. Rate limiting: Stored procedure with sliding window: CREATE OR REPLACE FUNCTION get_rate_limit(p_user_id UUID, p_window_minutes INTEGER) RETURNS INTEGER AS $$ SELECT COUNT(*) FROM usage_logs WHERE user_id = p_user_id AND created_at > NOW() - (p_window_minutes || ' minutes')::INTERVAL; $$ LANGUAGE sql;. Access tier examples: Free (1000/day, 10/min), Pro (100000/day, 100/min), Enterprise (unlimited, 1000/min). Query for validation: SELECT u.id, u.email, t.daily_quota, t.rate_limit_per_minute, ak.scopes FROM users u JOIN access_tiers t ON u.tier_id = t.id JOIN api_keys ak ON ak.user_id = u.id WHERE ak.key_hash = $1 AND ak.status = 'active' AND (ak.expires_at IS NULL OR ak.expires_at > NOW()) AND u.status = 'active';. Best practices: (1) Partition usage_logs by month for performance. (2) Index: daily_usage(user_id, date), usage_logs(user_id, created_at), api_keys(key_hash). (3) Archive old usage_logs after 90 days. (4) Use JSONB for tier features flexibility. (5) Implement soft delete (status='deleted') for api_keys audit trail. (6) Rate limit at application layer with Redis for performance, PostgreSQL for billing/analytics. Essential schema for production API platforms.

Two-tier rate limiting provides different quotas for anonymous users (IP-based) and authenticated users (user ID-based), commonly implemented with Redis for fast in-memory counters. Architecture: Layer 1 (anonymous): Rate limit by IP address with lower quota (50 requests/day). Layer 2 (authenticated): Rate limit by user ID with higher quota (1000-10000/day based on tier). Implementation with Redis and FastAPI: from fastapi import FastAPI, Request, HTTPException; from fastapi.security import HTTPBearer; import redis.asyncio as redis; app = FastAPI(); redis_client = redis.from_url('redis://localhost'); security = HTTPBearer(auto_error=False); async def check_rate_limit(key: str, limit: int, window: int = 86400) -> bool: current = await redis_client.incr(key); if current == 1: await redis_client.expire(key, window); return current <= limit; @app.middleware('http'); async def rate_limit_middleware(request: Request, call_next): # Try to get authenticated user; auth_header = request.headers.get('Authorization'); user_id = None; if auth_header and auth_header.startswith('Bearer '): token = auth_header.split('Bearer ')[1]; user_id = await validate_token(token); # Validated user gets higher quota; if user_id: key = f'rate_limit:user:{user_id}'; limit = 1000; tier = await get_user_tier(user_id); if tier == 'pro': limit = 10000; elif tier == 'enterprise': limit = 100000; else: # Anonymous user rate limited by IP; client_ip = request.client.host; key = f'rate_limit:ip:{client_ip}'; limit = 50; if not await check_rate_limit(key, limit): raise HTTPException(status_code=429, detail='Rate limit exceeded'); response = await call_next(request); return response;. Redis key structure: rate_limit:user:<user_id> for authenticated users, rate_limit:ip:<ip_address> for anonymous. Sliding window (more accurate): Use sorted sets in Redis with timestamp scores: ZADD rate_limit:user:{user_id} {timestamp} {request_id}; ZREMRANGEBYSCORE rate_limit:user:{user_id} 0 {now - window}; ZCARD rate_limit:user:{user_id} returns current count. Return count in response headers: X-RateLimit-Limit: 1000; X-RateLimit-Remaining: 847; X-RateLimit-Reset: 1735689600. Benefits of two-tier: Prevents abuse from anonymous users while providing better UX for authenticated users, encourages authentication/subscription, protects API from DDoS. Edge cases: API key in query param (avoid, use Authorization header), multiple IPs behind NAT (consider X-Forwarded-For with IP whitelist), distributed systems (centralized Redis cluster). Production enhancements: Implement circuit breaker for Redis failures (fail open with warning), use Redis Cluster for high availability, log rate limit violations for abuse detection, implement graduated response (warn at 80%, throttle at 100%). Best practice: Always use Redis/Memcached for rate limiting (PostgreSQL too slow for high-traffic), set appropriate TTL on keys to prevent memory leaks, return 429 Too Many Requests with Retry-After header. Essential for scalable API protection.

Systemd service restart ensures new Python code is loaded by terminating the old process and starting a fresh one with a new Python interpreter. Procedure: (1) Deploy new code: git pull origin main (or scp files to server). (2) Verify files updated: ls -lh /path/to/changed/file.py check timestamp. (3) Test service configuration (if changed): sudo systemctl daemon-reload (only needed if .service file edited). (4) Restart service: sudo systemctl restart myservice. (5) Verify restart: sudo systemctl status myservice shows 'active (running)' with recent 'Started' timestamp. (6) Check logs: sudo journalctl -u myservice -f shows startup messages with new code indicators. (7) Test functionality: curl http://localhost:8000/health or debug endpoint showing version. Why restart necessary: Python loads modules into sys.modules on import, editing .py files does not affect running process, deleting .pyc files has no effect on running process. Only full process restart clears sys.modules and re-imports code. Systemd restart behavior: Sends SIGTERM to process (graceful shutdown), waits for TimeoutStopSec (default 90s) for process to exit, sends SIGKILL if still running after timeout, starts new process with fresh Python interpreter. Service file best practices: [Service]; Type=simple; WorkingDirectory=/home/user/app; ExecStart=/home/user/venv/bin/python main.py; Restart=always; RestartSec=3; TimeoutStopSec=30; User=appuser; Group=appuser; Environment="PYTHONUNBUFFERED=1"; StandardOutput=journal; StandardError=journal;. Common pitfalls: (1) Forgetting to pull latest code before restart (old code still running). (2) Wrong virtual environment activated (service uses different venv). (3) Cached imports in dependencies (rare, but possible - requires dependency restart too). (4) Service file changes without daemon-reload (systemd uses old config). Verification steps: Check process PID changed: systemctl status myservice shows new PID. Check process start time: ps -p <PID> -o lstart shows recent timestamp. Check loaded module file: Add logger.info(f'Loaded module from {__file__}') at module level, verify in journalctl. Alternative reload method (zero-downtime, advanced): Use gunicorn with kill -HUP <gunicorn_master_pid> for graceful worker reload without systemctl restart. Best practice: Always test in staging first, monitor logs during restart for errors, implement health check endpoint returning version/commit hash, use blue-green deployment for critical services. Essential for reliable production deployments.