155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
"""
|
|
Authentication Module
|
|
Handles API key authentication and management.
|
|
"""
|
|
|
|
import hmac
|
|
import secrets
|
|
import hashlib
|
|
import os
|
|
from functools import wraps
|
|
from flask import request, abort, current_app, g
|
|
from .db import get_db
|
|
|
|
|
|
def hash_api_key(key: str, salt: str = None) -> tuple[str, str]:
|
|
"""Hash API key with salt for secure storage."""
|
|
if salt is None:
|
|
salt = secrets.token_hex(16)
|
|
|
|
# Use PBKDF2 for key derivation
|
|
key_hash = hashlib.pbkdf2_hmac(
|
|
'sha256',
|
|
key.encode('utf-8'),
|
|
salt.encode('utf-8'),
|
|
100000 # High iteration count for security
|
|
).hex()
|
|
|
|
return key_hash, salt
|
|
|
|
|
|
def verify_api_key(provided_key: str, stored_hash: str, salt: str) -> bool:
|
|
"""Verify provided API key against stored hash."""
|
|
expected_hash, _ = hash_api_key(provided_key, salt)
|
|
return hmac.compare_digest(expected_hash, stored_hash)
|
|
|
|
|
|
def require_api_key(f):
|
|
"""Decorator to require valid API key authentication."""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
# Check for bootstrap admin key (for initial key creation)
|
|
bootstrap_key = os.environ.get('ADMIN_BOOTSTRAP_KEY') or current_app.config.get('ADMIN_BOOTSTRAP_KEY')
|
|
if bootstrap_key:
|
|
provided_bootstrap = request.headers.get('X-Admin-Bootstrap-Key')
|
|
if provided_bootstrap and hmac.compare_digest(provided_bootstrap, bootstrap_key):
|
|
return f(*args, **kwargs)
|
|
|
|
# Check for regular API key
|
|
provided_key = request.headers.get('X-API-Key')
|
|
if not provided_key:
|
|
current_app.logger.warning(f"API key missing from request: {request.method} {request.path}")
|
|
abort(401, "API key required")
|
|
|
|
# Query database for key
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT key_hash, salt, revoked, scopes FROM api_keys WHERE key_id = ?",
|
|
(provided_key,)
|
|
).fetchone()
|
|
|
|
if not row:
|
|
current_app.logger.warning(f"Unknown API key used: {request.method} {request.path}")
|
|
abort(401, "Invalid API key")
|
|
|
|
if row['revoked']:
|
|
current_app.logger.warning(f"Revoked API key used: {request.method} {request.path}")
|
|
abort(401, "API key revoked")
|
|
|
|
# Verify key
|
|
if not verify_api_key(provided_key, row['key_hash'], row['salt']):
|
|
current_app.logger.warning(f"Invalid API key hash: {request.method} {request.path}")
|
|
abort(401, "Invalid API key")
|
|
|
|
# Update last used timestamp
|
|
db.execute(
|
|
"UPDATE api_keys SET last_used_at = datetime('now', '+8 hours') WHERE key_id = ?",
|
|
(provided_key,)
|
|
)
|
|
db.commit()
|
|
|
|
# Store key info in request context
|
|
g.api_key_id = provided_key
|
|
g.api_key_scopes = row['scopes'] or '*'
|
|
|
|
return f(*args, **kwargs)
|
|
|
|
return decorated_function
|
|
|
|
|
|
def create_api_key(description: str = "", scopes: str = "*") -> tuple[str, str]:
|
|
"""Create a new API key."""
|
|
key = secrets.token_urlsafe(32) # Generate secure random key
|
|
key_hash, salt = hash_api_key(key)
|
|
|
|
db = get_db()
|
|
db.execute(
|
|
"INSERT INTO api_keys (key_id, key_hash, salt, scopes, description) VALUES (?, ?, ?, ?, ?)",
|
|
(key, key_hash, salt, scopes, description)
|
|
)
|
|
db.commit()
|
|
|
|
current_app.logger.info(f"Created new API key: {key[:8]}... (description: {description})")
|
|
return key, key # Return both hashed and plain versions (plain only once)
|
|
|
|
|
|
def revoke_api_key(key_id: str) -> bool:
|
|
"""Revoke an API key."""
|
|
db = get_db()
|
|
result = db.execute(
|
|
"UPDATE api_keys SET revoked = 1 WHERE key_id = ?",
|
|
(key_id,)
|
|
)
|
|
db.commit()
|
|
|
|
if result.rowcount > 0:
|
|
current_app.logger.info(f"Revoked API key: {key_id[:8]}...")
|
|
return True
|
|
return False
|
|
|
|
|
|
def list_api_keys() -> list[dict]:
|
|
"""List all API keys (without sensitive data)."""
|
|
db = get_db()
|
|
rows = db.execute(
|
|
"SELECT key_id, scopes, description, created_at, last_used_at, revoked FROM api_keys ORDER BY created_at DESC"
|
|
).fetchall()
|
|
|
|
return [{
|
|
'key_id': row['key_id'],
|
|
'scopes': row['scopes'],
|
|
'description': row['description'],
|
|
'created_at': row['created_at'],
|
|
'last_used_at': row['last_used_at'],
|
|
'revoked': bool(row['revoked'])
|
|
} for row in rows]
|
|
|
|
|
|
def get_api_key_info(key_id: str) -> dict | None:
|
|
"""Get information about a specific API key."""
|
|
db = get_db()
|
|
row = db.execute(
|
|
"SELECT key_id, scopes, description, created_at, last_used_at, revoked FROM api_keys WHERE key_id = ?",
|
|
(key_id,)
|
|
).fetchone()
|
|
|
|
if row:
|
|
return {
|
|
'key_id': row['key_id'],
|
|
'scopes': row['scopes'],
|
|
'description': row['description'],
|
|
'created_at': row['created_at'],
|
|
'last_used_at': row['last_used_at'],
|
|
'revoked': bool(row['revoked'])
|
|
}
|
|
return None |