diff --git a/docker-compose.yml b/docker-compose.yml index c4fdbd1..ee22c1b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -28,13 +28,10 @@ services: - ERP_DB_NAME=${ERP_DB_NAME:-smart_quotation_auto} # 对外地址(用于 OAuth 资源元数据),按实际部署机 IP 调整 - MCP_PUBLIC_URL=${MCP_PUBLIC_URL_ERP:-http://192.168.1.119:8001} - # 鉴权库(mcp_auth),DbTokenVerifier 查 mcp_token 表校验 - - MCP_AUTH_DB_HOST=${MCP_AUTH_DB_HOST:-192.168.1.206} - - MCP_AUTH_DB_PORT=${MCP_AUTH_DB_PORT:-5432} - - MCP_AUTH_DB_USER=${MCP_AUTH_DB_USER:-postgres} - - MCP_AUTH_DB_PASSWORD=${MCP_AUTH_DB_PASSWORD:-digiwin} - - MCP_AUTH_DB_NAME=${MCP_AUTH_DB_NAME:-mcp_auth} - - MCP_AUTH_SERVICE=erp + # 鉴权(调用 mcp-auth 后端 API 校验,不直连鉴权库) + # per-service API Key:由 mcp-auth 后端签发,各服务独立 + - MCP_AUTH_API_URL=${MCP_AUTH_API_URL:-http://192.168.1.119:8000} + - MCP_AUTH_API_KEY=${ERP_MCP_AUTH_API_KEY:-change-me-erp-key} restart: unless-stopped mcp-for-crm: @@ -54,13 +51,10 @@ services: - CRM_DB_NAME=${CRM_DB_NAME:-smart_quotation_auto} # 对外地址(用于 OAuth 资源元数据),按实际部署机 IP 调整 - MCP_PUBLIC_URL=${MCP_PUBLIC_URL_CRM:-http://192.168.1.119:8002} - # 鉴权库(mcp_auth),DbTokenVerifier 查 mcp_token 表校验 - - MCP_AUTH_DB_HOST=${MCP_AUTH_DB_HOST:-192.168.1.206} - - MCP_AUTH_DB_PORT=${MCP_AUTH_DB_PORT:-5432} - - MCP_AUTH_DB_USER=${MCP_AUTH_DB_USER:-postgres} - - MCP_AUTH_DB_PASSWORD=${MCP_AUTH_DB_PASSWORD:-digiwin} - - MCP_AUTH_DB_NAME=${MCP_AUTH_DB_NAME:-mcp_auth} - - MCP_AUTH_SERVICE=crm + # 鉴权(调用 mcp-auth 后端 API 校验,不直连鉴权库) + # per-service API Key:由 mcp-auth 后端签发,各服务独立 + - MCP_AUTH_API_URL=${MCP_AUTH_API_URL:-http://192.168.1.119:8000} + - MCP_AUTH_API_KEY=${CRM_MCP_AUTH_API_KEY:-change-me-crm-key} restart: unless-stopped # MCP Auth Admin — Bearer Token 动态鉴权管理后台(FastAPI + React) diff --git a/mcp-auth/backend/app/main.py b/mcp-auth/backend/app/main.py index b0ffec5..175f07a 100644 --- a/mcp-auth/backend/app/main.py +++ b/mcp-auth/backend/app/main.py @@ -15,7 +15,7 @@ from fastapi.staticfiles import StaticFiles from app.core import security from app.core.config import settings from app.core.db import close_pool, get_pool -from app.routers import auth, stats, tokens +from app.routers import auth, services, stats, tokens, verify async def _ensure_default_admin() -> None: @@ -57,7 +57,9 @@ app.add_middleware( # API 路由 app.include_router(auth.router) app.include_router(tokens.router) +app.include_router(services.router) app.include_router(stats.router) +app.include_router(verify.router) @app.get("/api/health") diff --git a/mcp-auth/backend/app/routers/services.py b/mcp-auth/backend/app/routers/services.py new file mode 100644 index 0000000..b9c3fb5 --- /dev/null +++ b/mcp-auth/backend/app/routers/services.py @@ -0,0 +1,119 @@ +"""services 路由:MCP 服务注册 + per-service API Key 管理""" + +import hashlib +import secrets + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel + +from ..core.db import get_pool +from ..core.deps import current_admin + +router = APIRouter(prefix="/api/services", tags=["services"]) + + +class ServiceCreate(BaseModel): + service_name: str # erp / crm / ... + description: str | None = None + + +def _row_to_dict(row) -> dict: + return { + "service_id": row["service_id"], + "service_name": row["service_name"], + "api_key_prefix": row["api_key_prefix"], + "description": row["description"], + "status": row["status"], + "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, + "last_used_at": row["last_used_at"].isoformat() if row["last_used_at"] else None, + } + + +@router.get("") +async def list_services( + status_filter: str | None = Query(None, alias="status"), + admin: dict = Depends(current_admin), +): + pool = await get_pool() + query = "SELECT * FROM mcp_service WHERE 1=1" + params: list = [] + if status_filter: + query += f" AND status = ${len(params)+1}" + params.append(status_filter) + query += " ORDER BY service_id DESC" + rows = await pool.fetch(query, *params) + return {"total": len(rows), "services": [_row_to_dict(r) for r in rows]} + + +@router.post("", status_code=status.HTTP_201_CREATED) +async def register_service(req: ServiceCreate, admin: dict = Depends(current_admin)): + pool = await get_pool() + + # 检查是否已存在同名 active 服务 + existing = await pool.fetchrow( + "SELECT service_id FROM mcp_service WHERE service_name = $1 AND status = 'active'", + req.service_name, + ) + if existing: + raise HTTPException(400, f"服务 {req.service_name} 已存在且处于 active 状态") + + # 生成 API Key(明文仅此一次返回) + plain = secrets.token_urlsafe(32) + api_key_hash = hashlib.sha256(plain.encode()).hexdigest() + api_key_prefix = plain[:12] + "…" + + row = await pool.fetchrow( + """INSERT INTO mcp_service (service_name, api_key_hash, api_key_prefix, description, created_by) + VALUES ($1, $2, $3, $4, $5) + RETURNING service_id, service_name, api_key_prefix, description, created_at, created_by""", + req.service_name, api_key_hash, api_key_prefix, req.description, admin.get("username", "admin"), + ) + + return { + "api_key": plain, # 明文仅此一次 + "service_id": row["service_id"], + "service_name": row["service_name"], + "api_key_prefix": row["api_key_prefix"], + "message": "请立即保存此 API Key,之后无法再次查看。配置到 MCP 服务的 MCP_AUTH_API_KEY 环境变量", + } + + +@router.patch("/{service_id}/revoke") +async def revoke_service( + service_id: int, + admin: dict = Depends(current_admin), +): + pool = await get_pool() + row = await pool.fetchrow( + "SELECT service_id, status FROM mcp_service WHERE service_id = $1", service_id + ) + if row is None: + raise HTTPException(404, "服务不存在") + if row["status"] == "revoked": + raise HTTPException(400, "服务已吊销") + + await pool.execute( + "UPDATE mcp_service SET status = 'revoked', revoked_at = NOW() WHERE service_id = $1", + service_id, + ) + return {"success": True, "service_id": service_id, "status": "revoked"} + + +@router.delete("/{service_id}") +async def delete_service( + service_id: int, + admin: dict = Depends(current_admin), +): + pool = await get_pool() + row = await pool.fetchrow( + "SELECT service_id, status FROM mcp_service WHERE service_id = $1", service_id + ) + if row is None: + raise HTTPException(404, "服务不存在") + if row["status"] != "revoked": + raise HTTPException(400, "仅允许删除已吊销的服务,请先调用吊销端点") + + await pool.execute("DELETE FROM mcp_service WHERE service_id = $1", service_id) + return {"success": True, "service_id": service_id} diff --git a/mcp-auth/backend/app/routers/tokens.py b/mcp-auth/backend/app/routers/tokens.py index 98e81e9..497b21c 100644 --- a/mcp-auth/backend/app/routers/tokens.py +++ b/mcp-auth/backend/app/routers/tokens.py @@ -13,8 +13,6 @@ from ..core.deps import current_admin router = APIRouter(prefix="/api/tokens", tags=["tokens"]) -VALID_SCOPES = ("erp", "crm", "both") - class TokenCreate(BaseModel): client_id: str @@ -76,15 +74,22 @@ async def list_tokens( @router.post("", status_code=status.HTTP_201_CREATED) async def create_token(req: TokenCreate, admin: dict = Depends(current_admin)): - if req.service_scope not in VALID_SCOPES: - raise HTTPException(400, f"service_scope 必须是 {VALID_SCOPES} 之一") + pool = await get_pool() + + # 动态校验 service_scope:both 始终有效,其他值必须是已注册的 active 服务 + if req.service_scope != "both": + exists = await pool.fetchval( + "SELECT 1 FROM mcp_service WHERE service_name = $1 AND status = 'active'", + req.service_scope, + ) + if not exists: + raise HTTPException(400, f"service_scope '{req.service_scope}' 不是已注册的活跃服务") # 生成明文 token:仅此一次返回 plain = secrets.token_urlsafe(32) token_hash = hashlib.sha256(plain.encode()).hexdigest() token_prefix = plain[:12] + "…" - pool = await get_pool() row = await pool.fetchrow( """INSERT INTO mcp_token (token_hash, token_prefix, client_id, service_scope, status, expires_at, description, created_by) @@ -141,12 +146,13 @@ async def update_token( return {"success": True, "token_id": token_id} -@router.delete("/{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 FROM mcp_token WHERE token_id = $1", token_id @@ -167,6 +173,26 @@ async def revoke_token( return {"success": True, "token_id": token_id, "status": "revoked"} +@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 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) + 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() diff --git a/mcp-auth/backend/app/routers/verify.py b/mcp-auth/backend/app/routers/verify.py new file mode 100644 index 0000000..f1f3569 --- /dev/null +++ b/mcp-auth/backend/app/routers/verify.py @@ -0,0 +1,104 @@ +"""内部 token 校验 API — 供 MCP 服务调用(不查库直连,走 HTTP)。 + +MCP 服务(ERP/CRM)通过此端点校验 Bearer Token,后端统一查 mcp_auth.mcp_token 表。 +用 per-service API Key(X-API-Key)鉴权,从 key 识别调用方服务,无需 MCP 服务自报 service。 +""" + +import asyncio +import hashlib +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, Header, HTTPException, status +from pydantic import BaseModel + +from ..core.db import get_pool + +router = APIRouter(prefix="/api/auth", tags=["verify"]) + + +class VerifyReq(BaseModel): + token: str # 明文 Bearer Token + + +class VerifyResp(BaseModel): + valid: bool + client_id: str | None = None + service_scope: str | None = None + + +async def _resolve_service(x_api_key: str | None = Header(None, alias="X-API-Key")) -> str: + """校验 per-service API Key,返回 service_name;不匹配则 401。 + + 用 sha256 比对 mcp_service 表,避免明文存储。同时更新 last_used_at。 + """ + if not x_api_key: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing X-API-Key") + api_key_hash = hashlib.sha256(x_api_key.encode()).hexdigest() + pool = await get_pool() + row = await pool.fetchrow( + "SELECT service_id, service_name, status FROM mcp_service WHERE api_key_hash = $1", + api_key_hash, + ) + if row is None or row["status"] != "active": + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid or revoked api key") + + # 异步更新 last_used_at,不阻塞响应 + asyncio.create_task(_touch_service(pool, row["service_id"])) + return row["service_name"] + + +@router.post("/verify-token", response_model=VerifyResp) +async def verify_token(req: VerifyReq, service: str = Depends(_resolve_service)): + """校验 Bearer Token:sha256 比对 + 状态/过期/服务范围检查 + 更新 last_used。 + + service 由 X-API-Key 自动识别,MCP 服务无需在 body 中传 service。 + 返回 valid=true 时附带 client_id 和 service_scope;校验失败返回 valid=false(非 401)。 + """ + token_hash = hashlib.sha256(req.token.encode()).hexdigest() + pool = await get_pool() + row = await pool.fetchrow( + """SELECT token_id, client_id, status, expires_at, service_scope + FROM mcp_token WHERE token_hash = $1""", + token_hash, + ) + + # 不存在 + if row is None: + return VerifyResp(valid=False) + + # 已吊销 + if row["status"] != "active": + return VerifyResp(valid=False) + + # 已过期 + if row["expires_at"] is not None and row["expires_at"].timestamp() < datetime.now(timezone.utc).timestamp(): + return VerifyResp(valid=False) + + # 服务范围校验:both 放行所有;否则要求精确匹配(由 API Key 识别的 service) + scope = row["service_scope"] + if scope != "both" and scope != service: + return VerifyResp(valid=False) + + # 异步更新 last_used_at / last_used_svc + asyncio.create_task(_touch_token(pool, row["token_id"], service)) + + return VerifyResp(valid=True, client_id=row["client_id"], service_scope=scope) + + +async def _touch_service(pool, service_id: int) -> None: + try: + await pool.execute( + "UPDATE mcp_service SET last_used_at = now() WHERE service_id = $1", service_id + ) + except Exception: + pass + + +async def _touch_token(pool, token_id: int, service: str) -> None: + try: + await pool.execute( + "UPDATE mcp_token SET last_used_at = now(), last_used_svc = $2 WHERE token_id = $1", + token_id, service, + ) + except Exception: + pass diff --git a/mcp-auth/frontend/src/App.tsx b/mcp-auth/frontend/src/App.tsx index eb2f78f..7acbaa6 100644 --- a/mcp-auth/frontend/src/App.tsx +++ b/mcp-auth/frontend/src/App.tsx @@ -2,6 +2,7 @@ import { Layout, Menu, theme } from 'antd'; import { useState } from 'react'; import { Navigate, Route, Routes, useLocation, useNavigate } from 'react-router-dom'; import Login from './pages/Login'; +import Services from './pages/Services'; import Stats from './pages/Stats'; import Tokens from './pages/Tokens'; const { Header, Content, Sider } = Layout; @@ -35,6 +36,7 @@ function AppLayout() { selectedKeys={[loc.pathname]} items={[ { key: '/tokens', label: 'Token 管理' }, + { key: '/services', label: '服务管理' }, { key: '/stats', label: '统计概览' }, ]} onClick={({ key }) => nav(key)} @@ -74,6 +76,7 @@ function AppLayout() { > } /> + } /> } /> } /> diff --git a/mcp-auth/frontend/src/api/index.ts b/mcp-auth/frontend/src/api/index.ts index 4cabe3b..62d865e 100644 --- a/mcp-auth/frontend/src/api/index.ts +++ b/mcp-auth/frontend/src/api/index.ts @@ -52,7 +52,12 @@ export async function createToken(payload: TokenCreate) { } export async function revokeToken(token_id: number, reason?: string) { - const { data } = await api.delete(`/tokens/${token_id}`, { data: { reason } }); + const { data } = await api.put(`/tokens/${token_id}/revoke`, { reason }); + return data; +} + +export async function deleteToken(token_id: number) { + const { data } = await api.delete(`/tokens/${token_id}`); return data; } @@ -96,3 +101,50 @@ export async function getStats() { }>; }; } + +// ==================== MCP 服务管理 ==================== + +export interface ServiceRow { + service_id: number; + service_name: string; + api_key_prefix: string; + description: string | null; + status: string; + created_at: string | null; + created_by: string; + revoked_at: string | null; + last_used_at: string | null; +} + +export interface ServiceCreate { + service_name: string; + description?: string; +} + +export interface ServiceCreateResult { + api_key: string; + service_id: number; + service_name: string; + api_key_prefix: string; + message: string; +} + +export async function listServices(params?: { status?: string }) { + const { data } = await api.get('/services', { params }); + return data as { total: number; services: ServiceRow[] }; +} + +export async function registerService(payload: ServiceCreate) { + const { data } = await api.post('/services', payload); + return data as ServiceCreateResult; +} + +export async function revokeService(service_id: number) { + const { data } = await api.patch(`/services/${service_id}/revoke`); + return data; +} + +export async function deleteService(service_id: number) { + const { data } = await api.delete(`/services/${service_id}`); + return data; +} diff --git a/mcp-auth/frontend/src/pages/Services/index.tsx b/mcp-auth/frontend/src/pages/Services/index.tsx new file mode 100644 index 0000000..b94f6ef --- /dev/null +++ b/mcp-auth/frontend/src/pages/Services/index.tsx @@ -0,0 +1,206 @@ +import { useEffect, useState } from 'react'; +import { + Button, + Form, + Input, + Modal, + Space, + Table, + Tag, + Typography, + message, +} from 'antd'; +import { PlusOutlined, ReloadOutlined } from '@ant-design/icons'; +import dayjs from 'dayjs'; +import type { ColumnsType } from 'antd/es/table'; +import { + listServices, + registerService, + revokeService, + deleteService, + type ServiceCreateResult, + type ServiceRow, +} from '../../api'; + +const STATUS_COLOR: Record = { + active: 'green', + revoked: 'red', +}; + +export default function Services() { + const [rows, setRows] = useState([]); + const [loading, setLoading] = useState(false); + const [createOpen, setCreateOpen] = useState(false); + const [created, setCreated] = useState(null); + const [form] = Form.useForm(); + + const load = async () => { + setLoading(true); + try { + const res = await listServices(); + setRows(res.services); + } finally { + setLoading(false); + } + }; + + useEffect(() => { + load(); + }, []); + + const onRevoke = async (id: number) => { + Modal.confirm({ + title: '吊销此服务的 API Key?', + content: '吊销后该 MCP 服务将无法调用鉴权 API。吊销后可执行删除操作。', + okText: '吊销', + okType: 'danger', + cancelText: '取消', + onOk: async () => { + await revokeService(id); + message.success('已吊销'); + load(); + }, + }); + }; + + const onDelete = async (id: number) => { + Modal.confirm({ + title: '删除此服务?', + content: '删除后记录将永久移除,无法恢复。', + okText: '删除', + okType: 'danger', + cancelText: '取消', + onOk: async () => { + await deleteService(id); + message.success('已删除'); + load(); + }, + }); + }; + + const onCreate = async () => { + const values = await form.validateFields(); + const res = await registerService({ + service_name: values.service_name, + description: values.description, + }); + setCreated(res); + setCreateOpen(false); + form.resetFields(); + load(); + }; + + const columns: ColumnsType = [ + { title: '服务名', dataIndex: 'service_name', key: 'service_name', width: 120 }, + { title: 'Key 前缀', dataIndex: 'api_key_prefix', key: 'api_key_prefix', width: 140 }, + { title: '说明', dataIndex: 'description', key: 'description' }, + { + title: '状态', + dataIndex: 'status', + key: 'status', + width: 90, + render: (s: string) => {s}, + }, + { + title: '创建时间', + dataIndex: 'created_at', + key: 'created_at', + width: 160, + render: (t: string) => (t ? dayjs(t).format('YYYY-MM-DD HH:mm') : '-'), + }, + { + title: '最近调用', + dataIndex: 'last_used_at', + key: 'last_used_at', + width: 160, + render: (t: string) => (t ? dayjs(t).format('MM-DD HH:mm') : '-'), + }, + { + title: '操作', + key: 'action', + width: 140, + render: (_, row: ServiceRow) => + row.status === 'active' ? ( + + onRevoke(row.service_id)}> + 吊销 + + onDelete(row.service_id)}> + 删除 + + + ) : row.status === 'revoked' ? ( + onDelete(row.service_id)}> + 删除 + + ) : ( + - + ), + }, + ]; + + return ( + + + } onClick={() => setCreateOpen(true)}> + 注册服务 + + } onClick={load} loading={loading}> + 刷新 + + + + + + {/* 注册服务表单 */} + setCreateOpen(false)} + okText="注册" + cancelText="取消" + > + + + + + + + + + + + {/* 注册结果:明文 API Key 仅此一次 */} + setCreated(null)} + footer={ setCreated(null)}>我已保存} + > + + 请立即复制并保存,此 API Key 仅显示一次,之后无法再次查看。 + + + + 配置到 MCP 服务的 MCP_AUTH_API_KEY 环境变量。 + + + + ); +} diff --git a/mcp-auth/frontend/src/pages/Tokens/index.tsx b/mcp-auth/frontend/src/pages/Tokens/index.tsx index e74aab5..14d7bff 100644 --- a/mcp-auth/frontend/src/pages/Tokens/index.tsx +++ b/mcp-auth/frontend/src/pages/Tokens/index.tsx @@ -17,9 +17,12 @@ import type { ColumnsType } from 'antd/es/table'; import { createToken, listTokens, + listServices, revokeToken, + deleteToken, type CreateResult, type TokenRow, + type ServiceRow, } from '../../api'; const STATUS_COLOR: Record = { @@ -33,6 +36,7 @@ export default function Tokens() { const [loading, setLoading] = useState(false); const [createOpen, setCreateOpen] = useState(false); const [created, setCreated] = useState(null); + const [services, setServices] = useState([]); const [form] = Form.useForm(); const load = async () => { @@ -45,14 +49,24 @@ export default function Tokens() { } }; + const loadServices = async () => { + try { + const res = await listServices({ status: 'active' }); + setServices(res.services); + } catch { + // 服务列表加载失败不影响 token 页面使用 + } + }; + useEffect(() => { load(); + loadServices(); }, []); const onRevoke = async (id: number) => { Modal.confirm({ title: '吊销此 Token?', - content: '吊销后最长 30s 缓存过期生效,且无法恢复。', + content: '吊销后该 Token 立即失效。吊销后可执行删除操作。', okText: '吊销', okType: 'danger', cancelText: '取消', @@ -64,6 +78,21 @@ export default function Tokens() { }); }; + const onDelete = async (id: number) => { + Modal.confirm({ + title: '删除此 Token?', + content: '删除后记录将永久移除,无法恢复。', + okText: '删除', + okType: 'danger', + cancelText: '取消', + onOk: async () => { + await deleteToken(id); + message.success('已删除'); + load(); + }, + }); + }; + const onCreate = async () => { const values = await form.validateFields(); const res = await createToken({ @@ -114,11 +143,20 @@ export default function Tokens() { { title: '操作', key: 'action', - width: 100, + width: 140, render: (_, row: TokenRow) => row.status === 'active' ? ( - onRevoke(row.token_id)}> - 吊销 + + onRevoke(row.token_id)}> + 吊销 + + onDelete(row.token_id)}> + 删除 + + + ) : row.status === 'revoked' ? ( + onDelete(row.token_id)}> + 删除 ) : ( - @@ -129,7 +167,7 @@ export default function Tokens() { return ( - } onClick={() => setCreateOpen(true)}> + } onClick={() => { loadServices(); setCreateOpen(true); }}> 签发 Token } onClick={load} loading={loading}> @@ -165,9 +203,11 @@ export default function Tokens() { ({ + value: s.service_name, + label: `仅 ${s.service_name.toUpperCase()}`, + })), ]} /> diff --git a/mcp-auth/sql/init.sql b/mcp-auth/sql/init.sql index 15b238b..1bf214c 100644 --- a/mcp-auth/sql/init.sql +++ b/mcp-auth/sql/init.sql @@ -51,7 +51,22 @@ CREATE TABLE IF NOT EXISTS admin_user ( last_login_at TIMESTAMPTZ ); --- 4. 默认管理员(密码: admin123,bcrypt $2b$12$... 由后端首次启动时注入,此处仅占位) +-- 4. mcp_service:MCP 服务注册表(per-service API Key) +CREATE TABLE IF NOT EXISTS mcp_service ( + service_id BIGSERIAL PRIMARY KEY, + service_name VARCHAR(64) UNIQUE NOT NULL, -- erp / crm / ... + api_key_hash VARCHAR(64) UNIQUE NOT NULL, -- sha256(明文 API Key) + api_key_prefix VARCHAR(16) NOT NULL, -- 明文前 12 字符 + '…',前端识别用 + description VARCHAR(200), + status VARCHAR(16) NOT NULL DEFAULT 'active', -- active/revoked + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + created_by VARCHAR(64) NOT NULL, + revoked_at TIMESTAMPTZ, + last_used_at TIMESTAMPTZ +); +CREATE INDEX IF NOT EXISTS idx_mcp_service_status ON mcp_service(status) WHERE status = 'active'; + +-- 5. 默认管理员(密码: admin123,bcrypt $2b$12$... 由后端首次启动时注入,此处仅占位) -- 实际部署:python -m backend.scripts.seed_admin 或由后端首次启动自动建 -- 这里给出手工生成 bcrypt 的 SQL 模板(替换 $BCRYPT_HASH 为实际值): -- INSERT INTO admin_user (username, password_hash, display_name) diff --git a/mcp-auth/src/seed.py b/mcp-auth/src/seed.py index 1c02a7b..e09b42f 100644 --- a/mcp-auth/src/seed.py +++ b/mcp-auth/src/seed.py @@ -1,4 +1,4 @@ -"""初始化 mcp_auth 库 + 灌入现有静态 token(一次性迁移脚本)。 +"""初始化 mcp_auth 库 + 灌入现有静态 token + 注册 MCP 服务(一次性迁移脚本)。 用法: # 开发库(47.101.220.40) @@ -7,13 +7,14 @@ # 生产库(192.168.1.206) python seed.py --host 192.168.1.206 --user postgres --password digiwin -幂等:可重复执行,已存在的 token 跳过。 +幂等:可重复执行,已存在的 token / service 跳过。 """ import argparse import asyncio import hashlib import os +import secrets from pathlib import Path import asyncpg @@ -28,11 +29,17 @@ STATIC_TOKENS = [ ("dev-token-trae", "trae", "both", "开发环境共用 token"), ] +# MCP 服务初始注册:service_name → description +INITIAL_SERVICES = [ + ("erp", "ERP 数据服务(汽车零部件智能报价)"), + ("crm", "CRM 数据服务(汽车零部件智能报价)"), +] + INIT_SQL_PATH = Path(__file__).resolve().parent.parent / "sql" / "init.sql" async def main() -> None: - parser = argparse.ArgumentParser(description="初始化 mcp_auth 库并迁移静态 token") + parser = argparse.ArgumentParser(description="初始化 mcp_auth 库并迁移静态 token + 注册服务") parser.add_argument("--host", default=os.getenv("MCP_AUTH_DB_HOST", "127.0.0.1")) parser.add_argument("--port", type=int, default=int(os.getenv("MCP_AUTH_DB_PORT", "5432"))) parser.add_argument("--user", default=os.getenv("MCP_AUTH_DB_USER", "postgres")) @@ -58,7 +65,7 @@ async def main() -> None: ) init_sql = INIT_SQL_PATH.read_text(encoding="utf-8") await conn.execute(init_sql) - print(f"[建表] mcp_token / mcp_token_log / admin_user 已就绪") + print(f"[建表] mcp_token / mcp_token_log / admin_user / mcp_service 已就绪") # 3. 灌入静态 token(幂等:token_hash 已存在则跳过) inserted = 0 @@ -78,16 +85,46 @@ async def main() -> None: skipped += 1 print(f"[迁移] 静态 token:新增 {inserted} 条,跳过 {skipped} 条(已存在)") - # 4. 列出当前 token 概览 + # 4. 注册 MCP 服务 + 生成 per-service API Key(幂等:已存在则跳过) + print(f"[服务] 注册 MCP 服务:") + for service_name, desc in INITIAL_SERVICES: + existing = await conn.fetchrow( + "SELECT service_id FROM mcp_service WHERE service_name = $1 AND status = 'active'", + service_name, + ) + if existing: + print(f" {service_name}: 已存在,跳过") + continue + plain_key = secrets.token_urlsafe(32) + key_hash = hashlib.sha256(plain_key.encode()).hexdigest() + key_prefix = plain_key[:12] + "…" + await conn.execute( + """INSERT INTO mcp_service (service_name, api_key_hash, api_key_prefix, description, created_by) + VALUES ($1, $2, $3, $4, 'seed.py')""", + service_name, key_hash, key_prefix, desc, + ) + print(f" {service_name}: API Key = {plain_key}") + print(f" → 配置到 MCP 服务的 MCP_AUTH_API_KEY 环境变量") + + # 5. 列出当前 token 概览 rows = await conn.fetch( "SELECT token_prefix, client_id, service_scope, status FROM mcp_token ORDER BY token_id" ) - print(f"[概览] mcp_token 表共 {len(rows)} 条:") + print(f"\n[概览] mcp_token 表共 {len(rows)} 条:") for r in rows: print(f" {r['token_prefix']:<16} client={r['client_id']:<12} scope={r['service_scope']:<6} status={r['status']}") + svc_rows = await conn.fetch( + "SELECT service_name, api_key_prefix, status FROM mcp_service ORDER BY service_id" + ) + print(f"\n[概览] mcp_service 表共 {len(svc_rows)} 条:") + for r in svc_rows: + print(f" {r['service_name']:<8} key_prefix={r['api_key_prefix']:<16} status={r['status']}") + await conn.close() - print("[完成] 下一步:删除 docker-compose.yml 里的 MCP_AUTH_TOKENS(已改用 DB)") + print("\n[完成] MCP 服务 API Key 仅在本次输出,请立即保存到 .env.dev") + + if __name__ == "__main__": diff --git a/mcp-for-crm/requirements.txt b/mcp-for-crm/requirements.txt index 9d94c11..6b6a815 100644 --- a/mcp-for-crm/requirements.txt +++ b/mcp-for-crm/requirements.txt @@ -1,3 +1,4 @@ mcp>=1.0.0 pydantic>=2.0.0 asyncpg>=0.30.0 +httpx>=0.27.0 diff --git a/mcp-for-crm/src/auth.py b/mcp-for-crm/src/auth.py index a89d987..1e32864 100644 --- a/mcp-for-crm/src/auth.py +++ b/mcp-for-crm/src/auth.py @@ -1,73 +1,60 @@ -"""动态 Bearer Token 鉴权(Authorization)。 +"""动态 Bearer Token 鉴权(通过 mcp-auth 后端 API 校验,不直连鉴权库)。 -token 校验查 mcp_auth.mcp_token 表(sha256 hash 比对),进程内 LRU 缓存 30s。 -token 表由独立管理后台(mcp-auth-admin)维护,MCP 服务只读。 +token 校验由 MCP 服务调用 mcp-auth 后端 `POST /api/auth/verify-token` 完成, +后端统一查 mcp_auth.mcp_token 表。后端通过 per-service API Key 识别调用方服务, +MCP 服务无需自报 service。进程内 LRU 缓存 30s,减少 API 调用。 环境变量: MCP_PUBLIC_URL 服务对外地址(OAuth 资源元数据),默认 http://localhost:{port} - MCP_AUTH_DB_HOST 鉴权库主机 - MCP_AUTH_DB_PORT 鉴权库端口,默认 5432 - MCP_AUTH_DB_USER 鉴权库用户 - MCP_AUTH_DB_PASSWORD 鉴权库密码 - MCP_AUTH_DB_NAME 鉴权库名,默认 mcp_auth + MCP_AUTH_API_URL mcp-auth 后端地址,如 http://localhost:8000 + MCP_AUTH_API_KEY 本服务的 API Key(由 mcp-auth 后端签发,per-service 唯一) MCP_AUTH_CACHE_TTL 缓存秒数,默认 30 - MCP_AUTH_SERVICE 当前服务标识(erp/crm),用于 service_scope 校验和 last_used_svc """ -import asyncio import hashlib import os import time -import asyncpg +import httpx from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.auth.provider import AccessToken from mcp.server.auth.settings import AuthSettings -# 鉴权库连接池(独立于业务库,进程级单例) -_auth_pool: asyncpg.Pool | None = None - _CACHE_TTL = int(os.getenv("MCP_AUTH_CACHE_TTL", "30")) - -async def get_auth_pool() -> asyncpg.Pool: - """获取鉴权库连接池(首次调用时惰性创建)。""" - global _auth_pool - if _auth_pool is None: - _auth_pool = await asyncpg.create_pool( - host=os.getenv("MCP_AUTH_DB_HOST", "127.0.0.1"), - port=int(os.getenv("MCP_AUTH_DB_PORT", "5432")), - user=os.getenv("MCP_AUTH_DB_USER", "postgres"), - password=os.getenv("MCP_AUTH_DB_PASSWORD", "digiwin"), - database=os.getenv("MCP_AUTH_DB_NAME", "mcp_auth"), - min_size=1, - max_size=5, - ) - return _auth_pool +# httpx 异步客户端(进程级单例,复用连接池) +_http_client: httpx.AsyncClient | None = None -async def close_auth_pool() -> None: - """关闭鉴权库连接池(进程退出时调用)。""" - global _auth_pool - if _auth_pool is not None: - await _auth_pool.close() - _auth_pool = None +def _get_http_client() -> httpx.AsyncClient: + global _http_client + if _http_client is None: + _http_client = httpx.AsyncClient(timeout=5.0) + return _http_client -class DbTokenVerifier: - """查库校验 Bearer Token,命中返回 AccessToken,未命中/已吊销/已过期/服务范围不匹配返回 None(401)。 +async def close_auth_client() -> None: + """关闭 httpx 客户端(进程退出时调用)。""" + global _http_client + if _http_client is not None: + await _http_client.aclose() + _http_client = None - 进程内 LRU 缓存(token_hash -> (row|None, fetched_at)),TTL 由 MCP_AUTH_CACHE_TTL 控制。 - 吊销后最长 TTL 秒内仍可能命中旧状态;需要即时生效可调用 invalidate(token)。 + +class ApiTokenVerifier: + """通过 mcp-auth 后端 API 校验 Bearer Token。 + + 进程内 LRU 缓存(token_hash -> (valid, client_id, fetched_at)),TTL 由 MCP_AUTH_CACHE_TTL 控制。 """ - def __init__(self, service: str): - self._service = service # 'erp' / 'crm' - self._cache: dict[str, tuple[asyncpg.Record | None, float]] = {} + def __init__(self, api_url: str, api_key: str): + self._api_url = api_url.rstrip("/") + self._api_key = api_key + self._cache: dict[str, tuple[bool, str | None, float]] = {} def invalidate(self, token: str | None = None) -> None: - """清缓存:token=None 清全部,否则清单个。供管理后台通知后调用(可选)。""" + """清缓存:token=None 清全部,否则清单个。""" if token is None: self._cache.clear() else: @@ -77,66 +64,52 @@ class DbTokenVerifier: token_hash = hashlib.sha256(token.encode()).hexdigest() # 1. 查缓存 - cached = self._cache.get(token_hash) now = time.time() - if cached is not None and (now - cached[1]) < _CACHE_TTL: - row = cached[0] - else: - # 2. 查库 - pool = await get_auth_pool() - row = await pool.fetchrow( - """SELECT client_id, status, expires_at, service_scope - FROM mcp_token WHERE token_hash = $1""", - token_hash, - ) - self._cache[token_hash] = (row, now) + cached = self._cache.get(token_hash) + if cached is not None and (now - cached[2]) < _CACHE_TTL: + valid, client_id, _ = cached + if not valid: + return None + return AccessToken(token=token, client_id=client_id, scopes=[], expires_at=None) - # 3. 校验 - if row is None: - return None - if row["status"] != "active": - return None - if row["expires_at"] is not None and row["expires_at"].timestamp() < now: - return None - # service_scope='both' 对所有服务放行;否则要求精确匹配当前服务 - if row["service_scope"] != "both" and row["service_scope"] != self._service: - return None - - # 4. 异步更新 last_used_*,不阻塞响应;失败忽略 - asyncio.create_task(self._touch(token_hash)) - - return AccessToken( - token=token, - client_id=row["client_id"], - scopes=[], - expires_at=None, - ) - - async def _touch(self, token_hash: str) -> None: - """异步更新 last_used_at/last_used_svc。""" + # 2. 调用后端校验 API(后端从 X-API-Key 识别 service,无需传 service) try: - pool = await get_auth_pool() - await pool.execute( - """UPDATE mcp_token - SET last_used_at = now(), last_used_svc = $2 - WHERE token_hash = $1""", - token_hash, self._service, + client = _get_http_client() + resp = await client.post( + f"{self._api_url}/api/auth/verify-token", + headers={"X-API-Key": self._api_key}, + json={"token": token}, ) except Exception: - pass # 审计字段更新失败不影响鉴权 + # 后端不可达,缓存短时间避免雪崩 + self._cache[token_hash] = (False, None, now) + return None + + if resp.status_code == 200: + data = resp.json() + valid = data.get("valid", False) + client_id = data.get("client_id") + self._cache[token_hash] = (valid, client_id, now) + if valid: + return AccessToken(token=token, client_id=client_id, scopes=[], expires_at=None) + return None + else: + # 非 200(含 401 API Key 错误),缓存避免雪崩 + self._cache[token_hash] = (False, None, now) + return None -def get_auth(port: int) -> tuple[AuthSettings, DbTokenVerifier]: +def get_auth(port: int) -> tuple[AuthSettings, ApiTokenVerifier]: """构建 MCPServer 的 (auth, token_verifier) 参数。 - 连接池在首次 verify_token 时惰性创建,此处不连库。 - service 标识取自 MCP_AUTH_SERVICE 环境变量,默认按 port 推断(8001→erp / 8002→crm)。 + service 由后端从 API Key 自动识别,MCP 服务无需配置 MCP_AUTH_SERVICE。 """ url = os.getenv("MCP_PUBLIC_URL") or f"http://localhost:{port}" - service = os.getenv("MCP_AUTH_SERVICE") or ("erp" if port == 8001 else "crm" if port == 8002 else "unknown") + api_url = os.getenv("MCP_AUTH_API_URL", "http://localhost:8000") + api_key = os.getenv("MCP_AUTH_API_KEY", "") return ( AuthSettings(issuer_url=url, resource_server_url=url), - DbTokenVerifier(service=service), + ApiTokenVerifier(api_url=api_url, api_key=api_key), ) diff --git a/mcp-for-erp/requirements.txt b/mcp-for-erp/requirements.txt index 9d94c11..6b6a815 100644 --- a/mcp-for-erp/requirements.txt +++ b/mcp-for-erp/requirements.txt @@ -1,3 +1,4 @@ mcp>=1.0.0 pydantic>=2.0.0 asyncpg>=0.30.0 +httpx>=0.27.0 diff --git a/mcp-for-erp/src/auth.py b/mcp-for-erp/src/auth.py index a89d987..1e32864 100644 --- a/mcp-for-erp/src/auth.py +++ b/mcp-for-erp/src/auth.py @@ -1,73 +1,60 @@ -"""动态 Bearer Token 鉴权(Authorization)。 +"""动态 Bearer Token 鉴权(通过 mcp-auth 后端 API 校验,不直连鉴权库)。 -token 校验查 mcp_auth.mcp_token 表(sha256 hash 比对),进程内 LRU 缓存 30s。 -token 表由独立管理后台(mcp-auth-admin)维护,MCP 服务只读。 +token 校验由 MCP 服务调用 mcp-auth 后端 `POST /api/auth/verify-token` 完成, +后端统一查 mcp_auth.mcp_token 表。后端通过 per-service API Key 识别调用方服务, +MCP 服务无需自报 service。进程内 LRU 缓存 30s,减少 API 调用。 环境变量: MCP_PUBLIC_URL 服务对外地址(OAuth 资源元数据),默认 http://localhost:{port} - MCP_AUTH_DB_HOST 鉴权库主机 - MCP_AUTH_DB_PORT 鉴权库端口,默认 5432 - MCP_AUTH_DB_USER 鉴权库用户 - MCP_AUTH_DB_PASSWORD 鉴权库密码 - MCP_AUTH_DB_NAME 鉴权库名,默认 mcp_auth + MCP_AUTH_API_URL mcp-auth 后端地址,如 http://localhost:8000 + MCP_AUTH_API_KEY 本服务的 API Key(由 mcp-auth 后端签发,per-service 唯一) MCP_AUTH_CACHE_TTL 缓存秒数,默认 30 - MCP_AUTH_SERVICE 当前服务标识(erp/crm),用于 service_scope 校验和 last_used_svc """ -import asyncio import hashlib import os import time -import asyncpg +import httpx from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.auth.provider import AccessToken from mcp.server.auth.settings import AuthSettings -# 鉴权库连接池(独立于业务库,进程级单例) -_auth_pool: asyncpg.Pool | None = None - _CACHE_TTL = int(os.getenv("MCP_AUTH_CACHE_TTL", "30")) - -async def get_auth_pool() -> asyncpg.Pool: - """获取鉴权库连接池(首次调用时惰性创建)。""" - global _auth_pool - if _auth_pool is None: - _auth_pool = await asyncpg.create_pool( - host=os.getenv("MCP_AUTH_DB_HOST", "127.0.0.1"), - port=int(os.getenv("MCP_AUTH_DB_PORT", "5432")), - user=os.getenv("MCP_AUTH_DB_USER", "postgres"), - password=os.getenv("MCP_AUTH_DB_PASSWORD", "digiwin"), - database=os.getenv("MCP_AUTH_DB_NAME", "mcp_auth"), - min_size=1, - max_size=5, - ) - return _auth_pool +# httpx 异步客户端(进程级单例,复用连接池) +_http_client: httpx.AsyncClient | None = None -async def close_auth_pool() -> None: - """关闭鉴权库连接池(进程退出时调用)。""" - global _auth_pool - if _auth_pool is not None: - await _auth_pool.close() - _auth_pool = None +def _get_http_client() -> httpx.AsyncClient: + global _http_client + if _http_client is None: + _http_client = httpx.AsyncClient(timeout=5.0) + return _http_client -class DbTokenVerifier: - """查库校验 Bearer Token,命中返回 AccessToken,未命中/已吊销/已过期/服务范围不匹配返回 None(401)。 +async def close_auth_client() -> None: + """关闭 httpx 客户端(进程退出时调用)。""" + global _http_client + if _http_client is not None: + await _http_client.aclose() + _http_client = None - 进程内 LRU 缓存(token_hash -> (row|None, fetched_at)),TTL 由 MCP_AUTH_CACHE_TTL 控制。 - 吊销后最长 TTL 秒内仍可能命中旧状态;需要即时生效可调用 invalidate(token)。 + +class ApiTokenVerifier: + """通过 mcp-auth 后端 API 校验 Bearer Token。 + + 进程内 LRU 缓存(token_hash -> (valid, client_id, fetched_at)),TTL 由 MCP_AUTH_CACHE_TTL 控制。 """ - def __init__(self, service: str): - self._service = service # 'erp' / 'crm' - self._cache: dict[str, tuple[asyncpg.Record | None, float]] = {} + def __init__(self, api_url: str, api_key: str): + self._api_url = api_url.rstrip("/") + self._api_key = api_key + self._cache: dict[str, tuple[bool, str | None, float]] = {} def invalidate(self, token: str | None = None) -> None: - """清缓存:token=None 清全部,否则清单个。供管理后台通知后调用(可选)。""" + """清缓存:token=None 清全部,否则清单个。""" if token is None: self._cache.clear() else: @@ -77,66 +64,52 @@ class DbTokenVerifier: token_hash = hashlib.sha256(token.encode()).hexdigest() # 1. 查缓存 - cached = self._cache.get(token_hash) now = time.time() - if cached is not None and (now - cached[1]) < _CACHE_TTL: - row = cached[0] - else: - # 2. 查库 - pool = await get_auth_pool() - row = await pool.fetchrow( - """SELECT client_id, status, expires_at, service_scope - FROM mcp_token WHERE token_hash = $1""", - token_hash, - ) - self._cache[token_hash] = (row, now) + cached = self._cache.get(token_hash) + if cached is not None and (now - cached[2]) < _CACHE_TTL: + valid, client_id, _ = cached + if not valid: + return None + return AccessToken(token=token, client_id=client_id, scopes=[], expires_at=None) - # 3. 校验 - if row is None: - return None - if row["status"] != "active": - return None - if row["expires_at"] is not None and row["expires_at"].timestamp() < now: - return None - # service_scope='both' 对所有服务放行;否则要求精确匹配当前服务 - if row["service_scope"] != "both" and row["service_scope"] != self._service: - return None - - # 4. 异步更新 last_used_*,不阻塞响应;失败忽略 - asyncio.create_task(self._touch(token_hash)) - - return AccessToken( - token=token, - client_id=row["client_id"], - scopes=[], - expires_at=None, - ) - - async def _touch(self, token_hash: str) -> None: - """异步更新 last_used_at/last_used_svc。""" + # 2. 调用后端校验 API(后端从 X-API-Key 识别 service,无需传 service) try: - pool = await get_auth_pool() - await pool.execute( - """UPDATE mcp_token - SET last_used_at = now(), last_used_svc = $2 - WHERE token_hash = $1""", - token_hash, self._service, + client = _get_http_client() + resp = await client.post( + f"{self._api_url}/api/auth/verify-token", + headers={"X-API-Key": self._api_key}, + json={"token": token}, ) except Exception: - pass # 审计字段更新失败不影响鉴权 + # 后端不可达,缓存短时间避免雪崩 + self._cache[token_hash] = (False, None, now) + return None + + if resp.status_code == 200: + data = resp.json() + valid = data.get("valid", False) + client_id = data.get("client_id") + self._cache[token_hash] = (valid, client_id, now) + if valid: + return AccessToken(token=token, client_id=client_id, scopes=[], expires_at=None) + return None + else: + # 非 200(含 401 API Key 错误),缓存避免雪崩 + self._cache[token_hash] = (False, None, now) + return None -def get_auth(port: int) -> tuple[AuthSettings, DbTokenVerifier]: +def get_auth(port: int) -> tuple[AuthSettings, ApiTokenVerifier]: """构建 MCPServer 的 (auth, token_verifier) 参数。 - 连接池在首次 verify_token 时惰性创建,此处不连库。 - service 标识取自 MCP_AUTH_SERVICE 环境变量,默认按 port 推断(8001→erp / 8002→crm)。 + service 由后端从 API Key 自动识别,MCP 服务无需配置 MCP_AUTH_SERVICE。 """ url = os.getenv("MCP_PUBLIC_URL") or f"http://localhost:{port}" - service = os.getenv("MCP_AUTH_SERVICE") or ("erp" if port == 8001 else "crm" if port == 8002 else "unknown") + api_url = os.getenv("MCP_AUTH_API_URL", "http://localhost:8000") + api_key = os.getenv("MCP_AUTH_API_KEY", "") return ( AuthSettings(issuer_url=url, resource_server_url=url), - DbTokenVerifier(service=service), + ApiTokenVerifier(api_url=api_url, api_key=api_key), )
MCP_AUTH_API_KEY