feat: add ApiKey service with create/list/revoke/lookup
SHA-256 hashed keys, fmcp_ prefix, soft-delete revocation. Tests cover pure logic (hash, prefix, format, uniqueness, scope validation) and auth middleware bearer token path (pending auth.py update). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,87 @@
|
||||
import hashlib
|
||||
import secrets
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from fabledassistant.models import async_session
|
||||
from fabledassistant.models.api_key import ApiKey
|
||||
|
||||
|
||||
def generate_key() -> str:
|
||||
"""Generate a new full API key. Never stored — caller must hash it."""
|
||||
return "fmcp_" + secrets.token_urlsafe(32)
|
||||
|
||||
|
||||
def _hash_key(key: str) -> str:
|
||||
return hashlib.sha256(key.encode()).hexdigest()
|
||||
|
||||
|
||||
def _key_prefix(key: str) -> str:
|
||||
"""Return first 12 chars of the key for display (e.g. 'fmcp_abcdefg')."""
|
||||
return key[:12]
|
||||
|
||||
|
||||
async def create_api_key(
|
||||
user_id: int, name: str, scope: str
|
||||
) -> tuple[str, dict]:
|
||||
"""Create a new API key. Returns (full_key, key_dict). full_key is never stored."""
|
||||
if scope not in ("read", "write"):
|
||||
raise ValueError("scope must be 'read' or 'write'")
|
||||
full_key = generate_key()
|
||||
key = ApiKey(
|
||||
user_id=user_id,
|
||||
name=name,
|
||||
key_hash=_hash_key(full_key),
|
||||
key_prefix=_key_prefix(full_key),
|
||||
scope=scope,
|
||||
)
|
||||
async with async_session() as session:
|
||||
session.add(key)
|
||||
await session.commit()
|
||||
await session.refresh(key)
|
||||
return full_key, key.to_dict()
|
||||
|
||||
|
||||
async def list_api_keys(user_id: int) -> list[dict]:
|
||||
"""List all non-revoked API keys for the user."""
|
||||
async with async_session() as session:
|
||||
result = await session.execute(
|
||||
select(ApiKey)
|
||||
.where(ApiKey.user_id == user_id, ApiKey.revoked_at.is_(None))
|
||||
.order_by(ApiKey.created_at.desc())
|
||||
)
|
||||
return [k.to_dict() for k in result.scalars().all()]
|
||||
|
||||
|
||||
async def revoke_api_key(user_id: int, key_id: int) -> bool:
|
||||
"""Soft-delete a key by setting revoked_at. Returns True if found."""
|
||||
async with async_session() as session:
|
||||
result = await session.execute(
|
||||
select(ApiKey).where(ApiKey.id == key_id, ApiKey.user_id == user_id)
|
||||
)
|
||||
key = result.scalars().first()
|
||||
if key is None:
|
||||
return False
|
||||
key.revoked_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
return True
|
||||
|
||||
|
||||
async def lookup_key(raw_key: str) -> ApiKey | None:
|
||||
"""Look up a non-revoked ApiKey by raw token value. Updates last_used_at."""
|
||||
key_hash = _hash_key(raw_key)
|
||||
async with async_session() as session:
|
||||
result = await session.execute(
|
||||
select(ApiKey).where(
|
||||
ApiKey.key_hash == key_hash,
|
||||
ApiKey.revoked_at.is_(None),
|
||||
)
|
||||
)
|
||||
key = result.scalars().first()
|
||||
if key is None:
|
||||
return None
|
||||
key.last_used_at = datetime.now(timezone.utc)
|
||||
await session.commit()
|
||||
await session.refresh(key)
|
||||
return key
|
||||
Reference in New Issue
Block a user