Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""API Token routes — create, list, revoke Bearer tokens for programmatic access."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.api_token import create_api_token, list_api_tokens, revoke_api_token
|
|
from app.core.db import get_db
|
|
from app.deps import require_permission
|
|
|
|
router = APIRouter(prefix="/api/v1/tokens", tags=["api-tokens"])
|
|
|
|
|
|
class TokenCreateRequest(BaseModel):
|
|
name: str
|
|
scopes: list[str] = []
|
|
expires_in_days: int | None = None
|
|
|
|
|
|
class TokenRevokeRequest(BaseModel):
|
|
token_id: str
|
|
|
|
|
|
@router.post("", status_code=status.HTTP_201_CREATED)
|
|
async def create_token(
|
|
body: TokenCreateRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(require_permission("mcp:write")),
|
|
):
|
|
"""Create a new API token. The plaintext token is returned ONCE."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
|
|
expires_at = None
|
|
if body.expires_in_days is not None:
|
|
expires_at = datetime.now(UTC) + timedelta(days=body.expires_in_days)
|
|
|
|
result = await create_api_token(
|
|
db, tenant_id, user_id, body.name, body.scopes, expires_at,
|
|
)
|
|
await db.commit()
|
|
return result
|
|
|
|
|
|
@router.get("")
|
|
async def list_tokens(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(require_permission("mcp:read")),
|
|
):
|
|
"""List all API tokens for the current user (without token hashes)."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
tokens = await list_api_tokens(db, tenant_id, user_id)
|
|
return {"items": tokens, "total": len(tokens)}
|
|
|
|
|
|
@router.delete("/{token_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def revoke_token(
|
|
token_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(require_permission("mcp:write")),
|
|
):
|
|
"""Revoke an API token."""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
try:
|
|
tid = uuid.UUID(token_id)
|
|
except ValueError:
|
|
raise HTTPException(400, detail={"detail": "Invalid token_id", "code": "invalid_id"}) from None
|
|
revoked = await revoke_api_token(db, tenant_id, tid)
|
|
if not revoked:
|
|
raise HTTPException(404, detail={"detail": "Token not found or already revoked", "code": "not_found"})
|
|
await db.commit()
|