from __future__ import annotations import hashlib import hmac import secrets from dataclasses import dataclass from datetime import datetime, timezone from sqlalchemy.orm import Session from app.db.models import ApiKey, User API_KEY_PREFIX_LENGTH = 12 API_KEY_RANDOM_BYTES = 32 @dataclass(slots=True) class CreatedApiKey: model: ApiKey secret: str def hash_api_key(secret: str) -> str: return hashlib.sha256(secret.encode("utf-8")).hexdigest() def verify_api_key(secret: str, expected_hash: str) -> bool: return hmac.compare_digest(hash_api_key(secret), expected_hash) def generate_api_key_secret() -> str: return "mm_" + secrets.token_urlsafe(API_KEY_RANDOM_BYTES) def api_key_prefix(secret: str) -> str: # Prefix is only a lookup helper and must not be enough to authenticate. return secret[:API_KEY_PREFIX_LENGTH] def create_api_key( session: Session, *, user: User, name: str, scopes: list[str], secret: str | None = None, expires_at: datetime | None = None, ) -> CreatedApiKey: secret = secret or generate_api_key_secret() model = ApiKey( tenant_id=user.tenant_id, user_id=user.id, name=name, prefix=api_key_prefix(secret), key_hash=hash_api_key(secret), scopes=scopes, expires_at=expires_at, ) session.add(model) session.flush() return CreatedApiKey(model=model, secret=secret) def authenticate_api_key(session: Session, secret: str) -> ApiKey | None: prefix = api_key_prefix(secret) candidates = session.query(ApiKey).filter(ApiKey.prefix == prefix, ApiKey.revoked_at.is_(None)).all() now = datetime.now(timezone.utc) for candidate in candidates: if candidate.expires_at and candidate.expires_at < now: continue if verify_api_key(secret, candidate.key_hash): candidate.last_used_at = now session.add(candidate) return candidate return None def has_scope(api_key: ApiKey, required_scope: str) -> bool: scopes = set(api_key.scopes or []) return "*" in scopes or required_scope in scopes