bvandeusenandClaude Opus 5.5
0f5ef0be63
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 56s
CI & Build / integration (push) Successful in 1m4s
CI & Build / Python tests (push) Successful in 1m55s
CI & Build / Build & push image (push) Successful in 44s
Step 2 of the Scribe -> Grimoire rename. - src/scribe -> src/grimoire, and every import, patch target, module path, hypercorn entry point and `python -m grimoire.restore` with it; pyproject, uv.lock and the frontend package are named grimoire. - Image git.fabledsword.com/bvandeusen/fabledgrimoire; CI's Postgres service runs as grimoire / grimoire_test. - Compose files default the DB user, database and (prod) volume to grimoire, each overridable from .env, so an install created as Scribe keeps its data by setting POSTGRES_USER, POSTGRES_DB and PGDATA_VOLUME. - Backups are written as grimoire-<ts>.tar with format grimoire-backup-tar. Archives named scribe-* with the scribe-backup-tar manifest still restore and are still counted at the destination (folder and S3), with tests. - Exports, the MCP auth realm, the ASGI scope key and the persistent logger carry the new name. Kept on purpose: the secret-sealing key label stays "scribe.secrets.v1", because every stored secret was derived under it. Plugin identity (plugin name, hooks, env vars, .scribe marker, skills) is step 3. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from grimoire.models import async_session
|
|
from grimoire.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_token(raw: str) -> str:
|
|
"""The ONE fingerprint for every bearer secret stored by hash — API keys,
|
|
password-reset tokens, invitation tokens. Stored rows hold this, never
|
|
the raw value; a lookup hashes the presented token and compares."""
|
|
return hashlib.sha256(raw.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_token(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_token(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
|