mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan Denis
c5710e88b9
Introduce programmatic API features and observability: add API Key management, API usage analytics, webhook event subscriptions, and OpenAPI docs. New models (ApiKey, ApiUsageLog, ApiUsageSummary, EventSubscription, EventDelivery), services (api_analytics, api_key, event, openapi), and blueprints (/api-keys, /api-analytics, /event-subscriptions, /docs) were added and registered in create_app. Middleware for API key authentication, buffered analytics logging (with a flush thread), and enhanced rate limiting (custom key function and response headers) were implemented; RBAC was updated to accept API key auth via an auth_required decorator. Background threads for hourly aggregation and event retry were added; security CSP updated to allow resources for Swagger UI. Frontend settings UI and styles were added/updated for API keys and webhooks, and requirements were updated.
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""API Key model for programmatic API access."""
|
|
from datetime import datetime
|
|
from app import db
|
|
import hashlib
|
|
import json
|
|
import secrets
|
|
|
|
|
|
class ApiKey(db.Model):
|
|
"""API key for programmatic access to the ServerKit API."""
|
|
__tablename__ = 'api_keys'
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
|
|
name = db.Column(db.String(100), nullable=False)
|
|
key_prefix = db.Column(db.String(8), nullable=False)
|
|
key_hash = db.Column(db.String(256), unique=True, nullable=False)
|
|
scopes = db.Column(db.Text, nullable=True) # JSON array
|
|
tier = db.Column(db.String(20), default='standard') # standard | elevated | unlimited
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
expires_at = db.Column(db.DateTime, nullable=True)
|
|
last_used_at = db.Column(db.DateTime, nullable=True)
|
|
last_used_ip = db.Column(db.String(45), nullable=True)
|
|
usage_count = db.Column(db.Integer, default=0)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
revoked_at = db.Column(db.DateTime, nullable=True)
|
|
|
|
user = db.relationship('User', foreign_keys=[user_id])
|
|
|
|
TIER_STANDARD = 'standard'
|
|
TIER_ELEVATED = 'elevated'
|
|
TIER_UNLIMITED = 'unlimited'
|
|
VALID_TIERS = [TIER_STANDARD, TIER_ELEVATED, TIER_UNLIMITED]
|
|
|
|
@staticmethod
|
|
def generate_key():
|
|
"""Generate a new API key. Returns (raw_key, prefix, hash)."""
|
|
raw = 'sk_' + secrets.token_hex(20)
|
|
prefix = raw[:8]
|
|
key_hash = hashlib.sha256(raw.encode()).hexdigest()
|
|
return raw, prefix, key_hash
|
|
|
|
@staticmethod
|
|
def hash_key(raw_key):
|
|
"""Hash a raw API key."""
|
|
return hashlib.sha256(raw_key.encode()).hexdigest()
|
|
|
|
def get_scopes(self):
|
|
"""Return parsed scopes list."""
|
|
if self.scopes:
|
|
try:
|
|
return json.loads(self.scopes)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return []
|
|
return []
|
|
|
|
def set_scopes(self, scopes_list):
|
|
"""Set scopes from a list."""
|
|
if scopes_list:
|
|
self.scopes = json.dumps(scopes_list)
|
|
else:
|
|
self.scopes = None
|
|
|
|
def is_expired(self):
|
|
"""Check if the key has expired."""
|
|
if self.expires_at is None:
|
|
return False
|
|
return datetime.utcnow() > self.expires_at
|
|
|
|
def is_valid(self):
|
|
"""Check if the key is active and not expired or revoked."""
|
|
return self.is_active and not self.is_expired() and self.revoked_at is None
|
|
|
|
def has_scope(self, required_scope):
|
|
"""Check if this key has the required scope."""
|
|
scopes = self.get_scopes()
|
|
if not scopes or '*' in scopes:
|
|
return True
|
|
if required_scope in scopes:
|
|
return True
|
|
# Check wildcard resource match (e.g. 'apps:*' matches 'apps:read')
|
|
resource = required_scope.split(':')[0] if ':' in required_scope else required_scope
|
|
if f'{resource}:*' in scopes:
|
|
return True
|
|
return False
|
|
|
|
def record_usage(self, ip_address=None):
|
|
"""Record a usage of this key."""
|
|
self.last_used_at = datetime.utcnow()
|
|
self.usage_count = (self.usage_count or 0) + 1
|
|
if ip_address:
|
|
self.last_used_ip = ip_address
|
|
|
|
def to_dict(self):
|
|
"""Serialize key info (never expose hash)."""
|
|
return {
|
|
'id': self.id,
|
|
'user_id': self.user_id,
|
|
'name': self.name,
|
|
'key_prefix': self.key_prefix,
|
|
'scopes': self.get_scopes(),
|
|
'tier': self.tier,
|
|
'is_active': self.is_active,
|
|
'expires_at': self.expires_at.isoformat() if self.expires_at else None,
|
|
'last_used_at': self.last_used_at.isoformat() if self.last_used_at else None,
|
|
'last_used_ip': self.last_used_ip,
|
|
'usage_count': self.usage_count,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'revoked_at': self.revoked_at.isoformat() if self.revoked_at else None,
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<ApiKey {self.key_prefix}... ({self.name})>'
|