3a5f9e9001
- Disable SQLAdmin basic auth (comment out mount_admin, statics, redirect) - Add /api-management page (Keycloak admin role required) - Add admin_api_keys.py: REST endpoints for list/create clients and keys - Add api_management.html: manage API clients, keys, permissions with copy-once key display - Update index.html: API Management link -> /api-management - Update auth middleware: add /api-management and /admin/users to PROTECTED_PATHS - Add CHANGES-2026-06-04.md dev notes
142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
"""
|
|
API Client and API Key management endpoints (Admin only)
|
|
Uses Keycloak admin role authentication — same pattern as admin_users.py
|
|
"""
|
|
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy.orm import Session
|
|
from pydantic import BaseModel
|
|
from datetime import datetime
|
|
|
|
from app.db.session import get_db
|
|
from app.db.models import ApiClient, ApiKey
|
|
from app.security.permissions import require_role, Roles
|
|
from app.security.api_key import generate_api_key, hash_api_key, encrypt_api_key, get_prefix
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/admin/api-keys", tags=["admin-api-keys"])
|
|
|
|
|
|
class ApiKeySchema(BaseModel):
|
|
id: int
|
|
name: str | None = None
|
|
key_prefix: str
|
|
permissions: list
|
|
is_active: bool
|
|
created_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class ApiClientSchema(BaseModel):
|
|
id: int
|
|
name: str
|
|
is_active: bool
|
|
api_keys: List[ApiKeySchema] = []
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class ApiClientCreateSchema(BaseModel):
|
|
name: str
|
|
|
|
|
|
class ApiKeyCreateSchema(BaseModel):
|
|
client_id: int
|
|
name: str | None = None
|
|
permissions: list[str] = []
|
|
|
|
|
|
@router.get("/clients", response_model=List[ApiClientSchema])
|
|
async def list_clients(
|
|
db: Session = Depends(get_db),
|
|
current_user: dict = Depends(require_role(Roles.ADMIN)),
|
|
):
|
|
"""List all API clients with their keys (Admin only)"""
|
|
return db.query(ApiClient).order_by(ApiClient.id).all()
|
|
|
|
|
|
@router.post("/clients", response_model=ApiClientSchema)
|
|
async def create_client(
|
|
data: ApiClientCreateSchema,
|
|
db: Session = Depends(get_db),
|
|
current_user: dict = Depends(require_role(Roles.ADMIN)),
|
|
):
|
|
"""Create a new API client (Admin only)"""
|
|
existing = db.query(ApiClient).filter(ApiClient.name == data.name).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Client name already exists")
|
|
client = ApiClient(name=data.name, is_active=True)
|
|
db.add(client)
|
|
db.commit()
|
|
db.refresh(client)
|
|
return client
|
|
|
|
|
|
@router.post("/generate")
|
|
async def generate_key(
|
|
data: ApiKeyCreateSchema,
|
|
db: Session = Depends(get_db),
|
|
current_user: dict = Depends(require_role(Roles.ADMIN)),
|
|
):
|
|
"""Generate a new API key for a client (Admin only). Returns plaintext key once."""
|
|
client = db.get(ApiClient, data.client_id)
|
|
if not client:
|
|
raise HTTPException(status_code=404, detail="Client not found")
|
|
|
|
plain_key = generate_api_key()
|
|
api_key = ApiKey(
|
|
client_id=data.client_id,
|
|
name=data.name,
|
|
key_prefix=get_prefix(plain_key),
|
|
key_hash=hash_api_key(plain_key),
|
|
encrypted_key=encrypt_api_key(plain_key),
|
|
permissions=data.permissions,
|
|
is_active=True,
|
|
)
|
|
db.add(api_key)
|
|
db.commit()
|
|
db.refresh(api_key)
|
|
logger.info(f"Admin {current_user.get('username')} created API key {api_key.id} for client {client.name}")
|
|
return {"key_id": api_key.id, "api_key": plain_key, "key_prefix": api_key.key_prefix, "permissions": data.permissions}
|
|
|
|
|
|
@router.post("/{key_id}/regenerate")
|
|
async def regenerate_key(
|
|
key_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: dict = Depends(require_role(Roles.ADMIN)),
|
|
):
|
|
"""Regenerate an API key — preserves permissions, returns new plaintext once (Admin only)"""
|
|
api_key = db.get(ApiKey, key_id)
|
|
if not api_key:
|
|
raise HTTPException(status_code=404, detail="API Key not found")
|
|
|
|
plain_key = generate_api_key()
|
|
api_key.key_prefix = get_prefix(plain_key)
|
|
api_key.key_hash = hash_api_key(plain_key)
|
|
api_key.encrypted_key = encrypt_api_key(plain_key)
|
|
db.commit()
|
|
db.refresh(api_key)
|
|
logger.info(f"Admin {current_user.get('username')} regenerated API key {key_id}")
|
|
return {"key_id": api_key.id, "api_key": plain_key, "key_prefix": api_key.key_prefix, "permissions": api_key.permissions}
|
|
|
|
|
|
@router.patch("/{key_id}/toggle")
|
|
async def toggle_key(
|
|
key_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: dict = Depends(require_role(Roles.ADMIN)),
|
|
):
|
|
"""Toggle API key active/inactive (Admin only)"""
|
|
api_key = db.get(ApiKey, key_id)
|
|
if not api_key:
|
|
raise HTTPException(status_code=404, detail="API Key not found")
|
|
api_key.is_active = not api_key.is_active
|
|
db.commit()
|
|
return {"key_id": key_id, "is_active": api_key.is_active}
|