"""tokens 路由:token CRUD + 签发 + 吊销 + 日志""" import hashlib import json import secrets from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel from ..core.db import get_pool from ..core.deps import current_admin from ..core.redis import cache_delete router = APIRouter(prefix="/api/tokens", tags=["tokens"]) class TokenCreate(BaseModel): client_id: str service_scope: str # MCP 服务地址(如 http://10.100.154.100:8001/mcp),后端反查服务名入库 description: str | None = None expires_at: datetime | None = None # null = 永不过期 class TokenUpdate(BaseModel): description: str | None = None expires_at: datetime | None = None class RevokeReq(BaseModel): reason: str | None = None def _row_to_dict(row) -> dict: return { "token_id": row["token_id"], "token_prefix": row["token_prefix"], "token_plain": row["token_plain"], "client_id": row["client_id"], "service_scope": row["service_scope"], "service_url": row["service_url"], "status": row["status"], "expires_at": row["expires_at"].isoformat() if row["expires_at"] else None, "description": row["description"], "created_at": row["created_at"].isoformat() if row["created_at"] else None, "created_by": row["created_by"], "revoked_at": row["revoked_at"].isoformat() if row["revoked_at"] else None, "revoke_reason": row["revoke_reason"], "last_used_at": row["last_used_at"].isoformat() if row["last_used_at"] else None, "last_used_svc": row["last_used_svc"], } @router.get("") async def list_tokens( client_id: str | None = Query(None), status: str | None = Query(None), service_scope: str | None = Query(None), admin: dict = Depends(current_admin), ): pool = await get_pool() query = "SELECT * FROM mcp_token WHERE 1=1" params: list = [] if client_id: query += f" AND client_id = ${len(params)+1}" params.append(client_id) if status: query += f" AND status = ${len(params)+1}" params.append(status) if service_scope: query += f" AND service_scope = ${len(params)+1}" params.append(service_scope) query += " ORDER BY token_id DESC" rows = await pool.fetch(query, *params) return {"total": len(rows), "tokens": [_row_to_dict(r) for r in rows]} @router.post("", status_code=status.HTTP_201_CREATED) async def create_token(req: TokenCreate, admin: dict = Depends(current_admin)): pool = await get_pool() # 动态校验 service_scope:传入 MCP 服务地址,反查服务名(必须是已注册的 active 服务) svc_row = await pool.fetchrow( "SELECT service_name, service_url FROM mcp_service WHERE service_url = $1 AND status = 'active'", req.service_scope.strip(), ) if svc_row is None: raise HTTPException(400, f"service_scope '{req.service_scope}' 不是已注册的活跃服务") service_name = svc_row["service_name"] # client_id + service_scope 唯一校验:同一客户端同一服务不允许重复签发 existing = await pool.fetchval( "SELECT 1 FROM mcp_token WHERE client_id = $1 AND service_scope = $2 AND status = 'active'", req.client_id, service_name, ) if existing: raise HTTPException(400, f"client_id '{req.client_id}' 已存在服务于 '{service_name}' 的活跃 Token,请先吊销旧 Token") # 生成明文 token:仅此一次返回 plain = secrets.token_urlsafe(32) token_hash = hashlib.sha256(plain.encode()).hexdigest() token_prefix = plain[:12] + "…" row = await pool.fetchrow( """INSERT INTO mcp_token (token_hash, token_plain, token_prefix, client_id, service_scope, service_url, status, expires_at, description, created_by) VALUES ($1, $2, $3, $4, $5, $6, 'active', $7, $8, $9) RETURNING token_id, token_prefix, client_id, service_scope, service_url, status, expires_at, description, created_at, created_by""", token_hash, plain, token_prefix, req.client_id, service_name, svc_row["service_url"], req.expires_at, req.description, admin.get("username", "admin"), ) # 审计日志 await pool.execute( "INSERT INTO mcp_token_log (token_id, event, detail) VALUES ($1, 'issued', $2)", row["token_id"], json.dumps({"client_id": req.client_id, "service_scope": service_name, "service_url": svc_row["service_url"]}), ) return { "token": plain, # 明文仅此一次 "token_id": row["token_id"], "token_prefix": row["token_prefix"], "client_id": row["client_id"], "service_scope": row["service_scope"], "service_url": row["service_url"], "message": "请立即保存此 token,之后无法再次查看", } @router.patch("/{token_id}") async def update_token( token_id: int, req: TokenUpdate, admin: dict = Depends(current_admin), ): pool = await get_pool() existing = await pool.fetchrow("SELECT token_id FROM mcp_token WHERE token_id = $1", token_id) if existing is None: raise HTTPException(404, "token 不存在") updates, params = [], [] if req.description is not None: updates.append(f"description = ${len(params)+1}") params.append(req.description) if req.expires_at is not None: updates.append(f"expires_at = ${len(params)+1}") params.append(req.expires_at) if not updates: raise HTTPException(400, "无更新字段") params.append(token_id) await pool.execute( f"UPDATE mcp_token SET {', '.join(updates)} WHERE token_id = ${len(params)}", *params, ) return {"success": True, "token_id": token_id} @router.put("/{token_id}/revoke") async def revoke_token( token_id: int, req: RevokeReq = RevokeReq(), admin: dict = Depends(current_admin), ): """吊销 token:软删除,status 改为 revoked。吊销后才可删除。""" pool = await get_pool() row = await pool.fetchrow( "SELECT token_id, status, token_hash FROM mcp_token WHERE token_id = $1", token_id ) if row is None: raise HTTPException(404, "token 不存在") if row["status"] == "revoked": raise HTTPException(400, "token 已吊销") await pool.execute( "UPDATE mcp_token SET status = 'revoked', revoked_at = now(), revoke_reason = $2 WHERE token_id = $1", token_id, req.reason, ) await pool.execute( "INSERT INTO mcp_token_log (token_id, event, detail) VALUES ($1, 'revoked', $2)", token_id, json.dumps({"reason": req.reason, "by": admin.get("username")}), ) await cache_delete(f"mcp:tok:{row['token_hash']}", f"mcp:tok:miss:{row['token_hash']}") return {"success": True, "token_id": token_id, "status": "revoked"} @router.put("/{token_id}/enable") async def enable_token( token_id: int, admin: dict = Depends(current_admin), ): """启用 token:将已吊销的 token 恢复为 active。""" pool = await get_pool() row = await pool.fetchrow( "SELECT token_id, status, token_hash FROM mcp_token WHERE token_id = $1", token_id ) if row is None: raise HTTPException(404, "token 不存在") if row["status"] == "active": raise HTTPException(400, "token 已是启用状态") await pool.execute( "UPDATE mcp_token SET status = 'active', revoked_at = NULL, revoke_reason = NULL WHERE token_id = $1", token_id, ) await pool.execute( "INSERT INTO mcp_token_log (token_id, event, detail) VALUES ($1, 'enabled', $2)", token_id, json.dumps({"by": admin.get("username")}), ) await cache_delete(f"mcp:tok:{row['token_hash']}", f"mcp:tok:miss:{row['token_hash']}") return {"success": True, "token_id": token_id, "status": "active"} @router.delete("/{token_id}") async def delete_token( token_id: int, admin: dict = Depends(current_admin), ): """删除 token:物理删除,仅允许删除已吊销的 token。""" pool = await get_pool() row = await pool.fetchrow( "SELECT token_id, status, token_hash FROM mcp_token WHERE token_id = $1", token_id ) if row is None: raise HTTPException(404, "token 不存在") if row["status"] != "revoked": raise HTTPException(400, "仅可删除已吊销的 token,请先吊销") await pool.execute("DELETE FROM mcp_token_log WHERE token_id = $1", token_id) await pool.execute("DELETE FROM mcp_token WHERE token_id = $1", token_id) await cache_delete(f"mcp:tok:{row['token_hash']}", f"mcp:tok:miss:{row['token_hash']}") return {"success": True, "token_id": token_id} @router.get("/{token_id}/logs") async def token_logs(token_id: int, limit: int = 50, admin: dict = Depends(current_admin)): pool = await get_pool() rows = await pool.fetch( """SELECT log_id, event, service, client_ip, occurred_at, detail FROM mcp_token_log WHERE token_id = $1 ORDER BY occurred_at DESC LIMIT $2""", token_id, limit, ) return { "total": len(rows), "logs": [ { "log_id": r["log_id"], "event": r["event"], "service": r["service"], "client_ip": r["client_ip"], "occurred_at": r["occurred_at"].isoformat() if r["occurred_at"] else None, "detail": r["detail"], } for r in rows ], }