Agent Zero
8ad0a19f25
Check Cross-Plugin Imports / check (push) Has been cancelled
5.1 Delegationstoken (app/core/delegation_token.py): - HMAC-SHA256 signiert mit SECRET_KEY, max 60s Lifetime - Payload: user_id, tenant_id, agent_id, audience, expires_at, token_id - Statelose Verifikation, Audience-Check, Expiry-Check 5.2 MCP Bearer-Auth: - app/core/api_token.py: Token Service (create, verify, revoke, list) - app/deps.py: get_current_user_bearer + get_current_user_or_bearer - app/routes/api_tokens.py: Token CRUD Routes (create, list, revoke) - MCP Server Routes: get_current_user_or_bearer akzeptiert Session + Bearer 5.3 Methodenrechte: - MCP nutzt bereits mcp:read/mcp:write basierend auf tool_def.required_permission 5.5 Audit: - MCP Tool-Ausfuehrung wird protokolliert (log_audit mit correlation_id) Tests: 13/13 bestanden (7 API Token + 6 Delegation Token)
79 lines
2.5 KiB
Python
79 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 datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
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 get_current_user, 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(timezone.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"})
|
|
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()
|