File size: 4,924 Bytes
37b5223 4b8e879 37b5223 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """
Ukweli — API Key Management
GET /api-keys -> list caller's keys
POST /api-keys -> generate a new key (secret shown ONCE)
DELETE /api-keys/{id} -> revoke a key
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.auth import (
AuthContext,
generate_api_key_pair,
hash_api_secret,
require_auth,
)
from app.db.session import get_db_session
from app.models.auth import ApiKey, AuthAuditLog
from app.models.enums import APIKeyStatus, AuthEventType, AuthTier
from app.models.schemas import (
ApiKeyCreateRequest,
ApiKeyCreateResponse,
ApiKeyListResponse,
ApiKeyOut,
)
logger = logging.getLogger("ukweli.api.api_keys")
router = APIRouter(prefix="/api-keys", tags=["API Keys"])
def _to_out(rec: ApiKey) -> ApiKeyOut:
return ApiKeyOut(
id=rec.id,
key_id=rec.key_id,
name=rec.name,
description=rec.description,
tier=rec.tier,
quota_per_day=rec.quota_per_day,
requests_today=rec.requests_today,
last_used_at=rec.last_used_at,
status=rec.status,
created_at=rec.created_at,
expires_at=rec.expires_at,
)
@router.get("", response_model=ApiKeyListResponse)
async def list_api_keys(
db: AsyncSession = Depends(get_db_session),
auth: AuthContext = Depends(require_auth),
) -> ApiKeyListResponse:
"""List the caller's API keys. Requires registered tier."""
if auth.user is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="API key management requires a registered account.",
)
result = await db.execute(
select(ApiKey)
.where(ApiKey.user_id == auth.user.id)
.order_by(ApiKey.created_at.desc())
)
keys = [_to_out(k) for k in result.scalars().all()]
return ApiKeyListResponse(keys=keys, total=len(keys))
@router.post("", response_model=ApiKeyCreateResponse, status_code=status.HTTP_201_CREATED)
async def create_api_key(
payload: ApiKeyCreateRequest,
db: AsyncSession = Depends(get_db_session),
auth: AuthContext = Depends(require_auth),
) -> ApiKeyCreateResponse:
"""
Issue a new API key. The **secret** is returned in plaintext exactly
once — we only store its bcrypt hash.
"""
if auth.user is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="API key management requires a registered account.",
)
key_id, secret = generate_api_key_pair()
rec = ApiKey(
user_id=auth.user.id,
key_id=key_id,
secret_hash=hash_api_secret(secret),
name=payload.name,
description=payload.description,
tier=AuthTier.API.value,
quota_per_day=10000,
status=APIKeyStatus.ACTIVE.value,
requests_today_reset_at=datetime.now(timezone.utc),
)
db.add(rec)
await db.flush()
# Audit
db.add(
AuthAuditLog(
event_type=AuthEventType.API_KEY_CREATED.value,
user_id=auth.user.id,
api_key_id=rec.id,
fingerprint=auth.fingerprint,
success=True,
)
)
logger.info("API key created: user=%s key_id=%s", auth.user.id, key_id)
return ApiKeyCreateResponse(
id=rec.id,
key_id=key_id,
secret=secret,
name=rec.name,
description=rec.description,
tier=rec.tier,
quota_per_day=rec.quota_per_day,
created_at=rec.created_at,
)
@router.delete("/{key_id}", status_code=status.HTTP_204_NO_CONTENT)
async def revoke_api_key(
key_id: str,
db: AsyncSession = Depends(get_db_session),
auth: AuthContext = Depends(require_auth),
):
"""Revoke an API key. Idempotent."""
if auth.user is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="API key management requires a registered account.",
)
result = await db.execute(
select(ApiKey).where(
ApiKey.key_id == key_id, ApiKey.user_id == auth.user.id
)
)
rec = result.scalar_one_or_none()
if rec is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="API key not found.",
)
if rec.status != APIKeyStatus.REVOKED.value:
rec.status = APIKeyStatus.REVOKED.value
db.add(
AuthAuditLog(
event_type=AuthEventType.API_KEY_REVOKED.value,
user_id=auth.user.id,
api_key_id=rec.id,
fingerprint=auth.fingerprint,
success=True,
)
)
logger.info("API key revoked: id=%s", rec.id)
return None
|