diff --git a/backend/app/modules/ai/chat/controller.py b/backend/app/modules/ai/chat/controller.py index cccdefc45..751ff2560 100644 --- a/backend/app/modules/ai/chat/controller.py +++ b/backend/app/modules/ai/chat/controller.py @@ -22,6 +22,7 @@ AiModelProbeSchema, AiModelTestResultSchema, AiModelTestSchema, + AiToolSchema, ChatQuerySchema, ChatSessionCreateSchema, ChatSessionOutSchema, @@ -114,13 +115,14 @@ async def ai_chat_controller( redis: Annotated[Redis, Depends(redis_getter)], data: Annotated[AiChatRequestSchema, Body(description="对话请求")], ) -> JSONResponse: - service = ChatService(auth) + service = ChatService(auth, redis=redis) # 与 WebSocket 流式一致：优先使用用户激活的模型配置 model_config = await get_user_model_config(redis, auth.user.id) result = await service.chat_non_stream( message=data.message, session_id=data.session_id, model_config=model_config, + use_tools=data.use_tools, ) return SuccessResponse( data=AiChatResponseSchema( @@ -214,6 +216,18 @@ async def test_model_connection_controller( return SuccessResponse(data=result, msg="连通性测试通过") +@ChatRouter.get("/tools", summary="列出可用的只读工具（AI 助手能力清单）", response_model=ResponseSchema[list[AiToolSchema]]) +async def list_ai_tools_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ai:chat:query"]))], + redis: Annotated[Redis, Depends(redis_getter)], +) -> JSONResponse: + from app.modules.ai.tools.registry import build_readonly_tools + + tools = build_readonly_tools(auth, redis) + result = [AiToolSchema(name=spec.name, description=spec.description) for spec in tools] + return SuccessResponse(data=result, msg="查询可用工具成功") + + async def _send_error_and_close(websocket: WebSocket, message: str) -> None: """发送错误消息并关闭连接""" try: @@ -266,7 +280,7 @@ async def websocket_chat_controller(websocket: WebSocket) -> None: redis = websocket.app.state.redis logger.info("WebSocket连接已建立: {} - 用户: {}", websocket.client, auth.user.username or "未认证") - chat_service = ChatService(auth) + chat_service = ChatService(auth, redis=redis) # 消息循环 while True: diff --git a/backend/app/modules/ai/chat/schema.py b/backend/app/modules/ai/chat/schema.py index 904d43ba7..212ace94b 100644 --- a/backend/app/modules/ai/chat/schema.py +++ b/backend/app/modules/ai/chat/schema.py @@ -12,6 +12,7 @@ class ChatQuerySchema(BaseModel): session_id: str | None = Field(None, description="会话ID") files: list[dict[str, Any]] | None = Field(None, description="文件信息") action: str | None = Field(None, description="动作类型：stop=停止生成 | None=对话") + use_tools: bool = Field(False, description="是否启用只读工具（服务器状态/存储源/任务/Redis 键名）") class ChatSessionCreateSchema(BaseModel): @@ -70,6 +71,7 @@ class AiChatRequestSchema(BaseModel): message: str = Field(..., min_length=1, description="用户消息内容") session_id: str | None = Field(None, description="会话ID，不传则创建新会话") + use_tools: bool = Field(False, description="是否启用只读工具（服务器状态/存储源/任务/Redis 键名）") @field_validator("message") @classmethod @@ -165,3 +167,10 @@ class AiModelConfigListResponse(BaseModel): items: list[AiModelConfigItemSchema] = Field(default_factory=list, description="配置项列表") active_id: str | None = Field(None, description="当前激活的配置项 ID；为空表示使用系统默认") + + +class AiToolSchema(BaseModel): + """AI 只读工具清单项""" + + name: str = Field(..., description="工具名") + description: str = Field(..., description="工具用途（给模型与用户看）") diff --git a/backend/app/modules/ai/chat/service.py b/backend/app/modules/ai/chat/service.py index c35bcc64b..a5f542c3f 100644 --- a/backend/app/modules/ai/chat/service.py +++ b/backend/app/modules/ai/chat/service.py @@ -153,8 +153,9 @@ async def _ensure_session(crud: ChatSessionCRUD, session_id: str | None) -> str: class ChatService: """聊天会话管理模块服务层""" - def __init__(self, auth: AuthSchema) -> None: + def __init__(self, auth: AuthSchema, redis: Any | None = None) -> None: self.auth = auth + self.redis = redis async def chat_query( self, @@ -168,12 +169,14 @@ async def chat_query( session_id = await _ensure_session(crud, query.session_id) agno_factory = AgnoFactory() + tools = self._chat_tools(bool(query.use_tools)) agent = agno_factory.create_agent( user_id=self.auth.user.username or "user", team_id=TEAM_ID, session_id=session_id, db=crud.db, model_config=model_config, + tools=tools, ) message = (query.message or "").strip() @@ -208,11 +211,20 @@ async def chat_query( logger.info("流式生成结束: session_id={} chunk_count={}", session_id, chunk_count) + def _chat_tools(self, enabled: bool) -> list[Any]: + """按开关构造只读工具（关闭时返回空列表，零行为变化）。""" + if not enabled: + return [] + from app.modules.ai.tools.registry import build_readonly_tools + + return [spec.func for spec in build_readonly_tools(self.auth, self.redis)] + async def chat_non_stream( self, message: str, session_id: str | None, model_config: dict[str, Any] | None = None, + use_tools: bool = False, ) -> dict[str, Any]: """非流式 AI 对话""" crud = ChatSessionCRUD(self.auth) @@ -225,6 +237,7 @@ async def chat_non_stream( session_id=sid, db=crud.db, model_config=model_config, + tools=self._chat_tools(use_tools), ) try: diff --git a/backend/app/modules/ai/tools/__init__.py b/backend/app/modules/ai/tools/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/modules/ai/tools/registry.py b/backend/app/modules/ai/tools/registry.py new file mode 100644 index 000000000..2a0104c12 --- /dev/null +++ b/backend/app/modules/ai/tools/registry.py @@ -0,0 +1,173 @@ +"""AI 只读工具注册表（FastHub ``ai_tools`` 的只读子集）。 + +设计要点： +- **只读**：服务器状态、存储源清单、存储文件列表、定时任务、Redis 键名；不提供写入/删除/执行命令类工具 + （FastHub 的 ``run_host_command`` / 本机终端在多租户 Web 下等于 RCE，明确不移植）； +- **工具签名只允许模型可填的参数**：``auth`` / ``redis`` 通过闭包捕获，绝不放进函数签名， + 否则会被当成「需要模型提供的参数」而诱发无效调用； +- 工具内部异常**不抛给模型**，统一转成 ``{"error": ...}``，避免一次工具失败中断整轮对话； +- 需要数据库的工具自行开短生命周期 session（与传输引擎一致），不依赖请求级 session。 +""" + +import inspect +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import select + +from app.core.base_schema import AuthSchema +from app.core.database import async_db_session +from app.core.logger import logger +from app.modules.monitor.cache.browser import RedisBrowserService +from app.modules.monitor.server.service import ServerService +from app.modules.task.cronjob.job.model import JobModel +from app.modules.task.storage.browse.service import StorageFileService +from app.modules.task.storage.core.base import StoragePage +from app.modules.task.storage.node.service import StorageNodeService + +# 单次返回的条目上限（防止把大结果塞进上下文） +_MAX_ENTRIES = 200 +_MAX_JOBS = 50 +_MAX_KEYS = 200 + + +@dataclass(frozen=True) +class ReadonlyToolSpec: + """一个只读工具：名字 + 描述 + 可调用对象（供 Agno 注入）。""" + + name: str + description: str + func: Callable[..., Awaitable[Any]] + + +def _error(action: str, exc: BaseException) -> dict[str, Any]: + logger.warning(f"AI 只读工具[{action}]失败: {type(exc).__name__}: {exc}") + return {"error": f"{action}失败: {type(exc).__name__}: {exc}"[:300]} + + +def build_readonly_tools(auth: AuthSchema, redis: Any | None = None) -> list[ReadonlyToolSpec]: + """构造当前用户可用的只读工具列表（auth/redis 以闭包捕获）。""" + + async def server_status() -> dict[str, Any]: + """查询服务器资源状态：CPU、内存、磁盘使用率。""" + try: + info = await ServerService.get_server_monitor_info() + return { + "cpu": info.cpu.model_dump(), + "memory": info.mem.model_dump(), + "disks": [disk.model_dump() for disk in (info.disks or [])][:10], + } + except Exception as exc: + return _error("查询服务器状态", exc) + + async def storage_sources() -> list[dict[str, Any]]: + """列出已配置的存储源（名称、协议、状态；不含地址与凭据）。""" + try: + async with async_db_session() as db: + nodes = await StorageNodeService(auth, db).get_list() + return [ + { + "id": node.id, + "name": node.name, + "protocol": node.protocol, + "status": node.status, + "is_default": node.is_default, + } + for node in nodes + ] + except Exception as exc: + return [_error("列出存储源", exc)] + + async def storage_files(source_name: str, path: str = "", bucket: str = "") -> dict[str, Any]: + """列出某个存储源指定目录下的文件与子目录（source_name 必填）。""" + try: + async with async_db_session() as db: + nodes = await StorageNodeService(auth, db).get_list() + node = next((item for item in nodes if item.name == source_name), None) + if node is None: + names = [item.name for item in nodes][:20] + return {"error": f"存储源不存在: {source_name}", "available": names} + result = await StorageFileService(auth, db).list_files( + source_id=node.id, + prefix=path or "", + bucket=bucket or None, + page_size=_MAX_ENTRIES, + ) + if isinstance(result, StoragePage): + return { + "source": source_name, + "path": path, + "entries": [ + {"name": entry.name, "is_dir": entry.is_dir, "size": entry.size} + for entry in result.items[:_MAX_ENTRIES] + ], + "has_next": result.has_next, + } + return { + "source": source_name, + "path": path, + "entries": [ + {"name": entry.name, "is_dir": entry.is_dir, "size": entry.size} + for entry in result[:_MAX_ENTRIES] + ], + "has_next": False, + } + except Exception as exc: + return _error("列出存储文件", exc) + + async def cron_jobs(limit: int = 20) -> list[dict[str, Any]]: + """列出定时任务（名称、触发方式、下次执行时间、状态）。""" + try: + count = max(1, min(int(limit or 20), _MAX_JOBS)) + async with async_db_session() as db: + rows = ( + await db.execute( + select(JobModel) + .where(JobModel.is_deleted.is_(False)) + .order_by(JobModel.id.desc()) + .limit(count) + ) + ).scalars().all() + return [ + { + "job_name": row.job_name, + "trigger_type": row.trigger_type, + "next_run_time": row.next_run_time, + "status": row.status, + } + for row in rows + ] + except Exception as exc: + return [_error("列出定时任务", exc)] + + async def redis_keys(pattern: str = "*", limit: int = 50) -> dict[str, Any]: + """按模式浏览 Redis 键名（只返回键名，不读取值）。""" + if redis is None: + return {"error": "Redis 不可用"} + try: + count = max(1, min(int(limit or 50), _MAX_KEYS)) + page = await RedisBrowserService.scan_keys(redis, db=0, pattern=pattern or "*", count=count) + return {"keys": [item.key for item in page.keys][:count], "cursor": page.cursor} + except Exception as exc: + return _error("浏览 Redis 键", exc) + + tools = [ + ReadonlyToolSpec("server_status", "查询服务器 CPU/内存/磁盘使用率", server_status), + ReadonlyToolSpec("storage_sources", "列出已配置的存储源（名称/协议/状态，不含凭据）", storage_sources), + ReadonlyToolSpec("storage_files", "列出某个存储源指定目录下的文件与子目录", storage_files), + ReadonlyToolSpec("cron_jobs", "列出最近的定时任务及其状态", cron_jobs), + ReadonlyToolSpec("redis_keys", "按模式浏览 Redis 键名（只读，不读取值）", redis_keys), + ] + _assert_no_injected_params(tools) + return tools + + +def _assert_no_injected_params(tools: list[ReadonlyToolSpec]) -> None: + """守卫：工具签名里不允许出现由闭包捕获的参数（否则模型会试图传它们）。""" + forbidden = {"auth", "self", "cls", "redis", "db"} + for spec in tools: + params = set(inspect.signature(spec.func).parameters) + leaked = params & forbidden + if leaked: + raise RuntimeError(f"AI 工具 {spec.name} 的签名包含闭包参数: {sorted(leaked)}") diff --git a/backend/app/modules/monitor/cache/browser.py b/backend/app/modules/monitor/cache/browser.py new file mode 100644 index 000000000..b9439ac26 --- /dev/null +++ b/backend/app/modules/monitor/cache/browser.py @@ -0,0 +1,207 @@ +"""Redis 通用键空间浏览（**只读**）。 + +与 ``CacheService`` 的区别：后者按应用缓存名前缀浏览（``cache_names``），只覆盖本系统自己的键； +本模块提供任意键空间的只读浏览：逻辑库清单 → SCAN 分页 → 单键详情（按类型取值）。 + +安全： +- 只读，不提供写入/删除； +- 键名命中敏感模式（token/session/password 等）时，**值一律脱敏**，只回类型/TTL/大小； +- 值按类型限量返回并标记 ``truncated``，避免把大值整段读进内存与响应体。 + +数据库切换：app 的 Redis 客户端连在默认库上；浏览其它库时按连接参数**新建短生命周期客户端**， +用完即关，避免 ``SELECT`` 污染共享连接池（与 FastHub 的做法一致）。 +""" + +import re +from typing import Any, cast + +from redis.asyncio.client import Redis + +from app.core.exceptions import CustomException + +from .schema import RedisDatabaseSchema, RedisKeyMetaSchema, RedisKeyPageSchema, RedisValueSchema + +# 敏感键模式：命中即对值脱敏（键名本身仍展示，便于运维定位） +_SENSITIVE_KEY_RE = re.compile( + r"(token|secret|password|passwd|session|credential|api[_-]?key|authorization|refresh)", + re.IGNORECASE, +) + +# 单值返回上限（字符/元素数） +_MAX_TEXT = 4000 +_MAX_ELEMENTS = 200 +_MAX_STREAM = 50 + + +def _text(value: Any, limit: int = _MAX_TEXT) -> tuple[str, bool]: + """把值转成文本并截断（bytes 安全）。""" + if isinstance(value, bytes): + value = value.decode("utf-8", errors="backslashreplace") + text = value if isinstance(value, str) else str(value) + if len(text) <= limit: + return text, False + return text[:limit], True + + +def is_sensitive_key(key: str) -> bool: + """键名是否命中敏感模式（命中则值脱敏）。""" + return bool(_SENSITIVE_KEY_RE.search(key or "")) + + +class RedisBrowserService: + """Redis 键空间只读浏览服务。""" + + @staticmethod + def _base_db(redis: Redis) -> int: + """当前客户端所在的逻辑库序号。""" + try: + return int(redis.connection_pool.connection_kwargs.get("db") or 0) + except Exception: # pragma: no cover - 连接参数异常时按默认库处理 + return 0 + + @staticmethod + def _client_for_db(redis: Redis, db: int) -> tuple[Redis, bool]: + """取得指定逻辑库的客户端；返回 (client, 是否需要自行关闭)。""" + if db == RedisBrowserService._base_db(redis): + return redis, False + kwargs = dict(redis.connection_pool.connection_kwargs) + kwargs["db"] = db + return Redis(**kwargs), True + + @staticmethod + async def list_databases(redis: Redis) -> list[RedisDatabaseSchema]: + """列出逻辑库与键数量（解析 ``INFO keyspace``）。""" + raw = await redis.info("keyspace") + result: list[RedisDatabaseSchema] = [] + for name, value in (raw or {}).items(): + if not isinstance(name, str) or not name.startswith("db") or not name[2:].isdigit(): + continue + parsed: dict[str, int] = {} + for part in str(value).split(","): + if "=" in part: + k, v = part.split("=", 1) + try: + parsed[k.strip()] = int(v) + except ValueError: + continue + result.append( + RedisDatabaseSchema( + db=int(name[2:]), + keys=parsed.get("keys", 0), + expires=parsed.get("expires", 0), + ) + ) + result.sort(key=lambda item: item.db) + return result + + @staticmethod + async def scan_keys( + redis: Redis, + *, + db: int = 0, + pattern: str = "*", + cursor: int = 0, + count: int = 100, + ) -> RedisKeyPageSchema: + """SCAN 分页浏览键（附带 type/ttl）。""" + client, owned = RedisBrowserService._client_for_db(redis, db) + try: + next_cursor, keys = await client.scan(cursor=cursor, match=pattern or "*", count=max(1, min(count, 500))) + metas: list[RedisKeyMetaSchema] = [] + if keys: + # pipeline 批量取 type/ttl，避免逐键往返 + async with client.pipeline(transaction=False) as pipe: + for key in keys: + pipe.type(key) + pipe.ttl(key) + results = await pipe.execute() + for index, key in enumerate(keys): + key_type = results[index * 2] + ttl = results[index * 2 + 1] + metas.append( + RedisKeyMetaSchema( + key=_text(key, limit=512)[0], + type=_text(key_type, limit=32)[0] if key_type is not None else "none", + ttl=int(ttl) if isinstance(ttl, int) else -1, + ) + ) + return RedisKeyPageSchema(cursor=int(next_cursor), keys=metas, scanned=len(metas)) + finally: + if owned: + await client.aclose() + + @staticmethod + async def get_value(redis: Redis, *, db: int, key: str) -> RedisValueSchema: + """读取单键详情（只读；敏感键脱敏；值按类型限量）。""" + if not key: + raise CustomException(msg="键名不能为空") + client, owned = RedisBrowserService._client_for_db(redis, db) + try: + key_type = await client.type(key) + key_type = _text(key_type, limit=32)[0] if key_type is not None else "none" + if key_type == "none": + raise CustomException(msg=f"键不存在: {key}") + ttl = await client.ttl(key) + ttl_value = int(ttl) if isinstance(ttl, int) else -1 + size: int | None = None + try: + raw_size = await client.memory_usage(key) + size = int(raw_size) if isinstance(raw_size, int) else None + except Exception: + size = None + + if is_sensitive_key(key): + return RedisValueSchema( + key=key, + type=key_type, + ttl=ttl_value, + size=size, + value="••••••（敏感键已脱敏）", + truncated=False, + masked=True, + ) + + value, truncated = await RedisBrowserService._read_typed(client, key, key_type) + return RedisValueSchema( + key=key, + type=key_type, + ttl=ttl_value, + size=size, + value=value, + truncated=truncated, + ) + finally: + if owned: + await client.aclose() + + @staticmethod + async def _read_typed(client: Redis, key: str, key_type: str) -> tuple[object, bool]: + """按类型读取值（限量）。""" + if key_type == "string": + raw = await client.get(key) + return _text(raw) + if key_type == "hash": + raw_hash = await cast(Any, client.hgetall(key)) + items = list(raw_hash.items())[:_MAX_ELEMENTS] + text_map = {_text(k, 512)[0]: _text(v)[0] for k, v in items} + return text_map, len(raw_hash) > _MAX_ELEMENTS + if key_type == "list": + raw_list = await cast(Any, client.lrange(key, 0, _MAX_ELEMENTS - 1)) + return [_text(item)[0] for item in raw_list], len(raw_list) >= _MAX_ELEMENTS + if key_type == "set": + _, members = await client.sscan(key, count=_MAX_ELEMENTS) + return [_text(item)[0] for item in members[:_MAX_ELEMENTS]], len(members) > _MAX_ELEMENTS + if key_type == "zset": + raw_zset = await client.zrange(key, 0, _MAX_ELEMENTS - 1, withscores=True) + return [ + {"member": _text(member, 512)[0], "score": float(score)} + for member, score in raw_zset[:_MAX_ELEMENTS] + ], len(raw_zset) > _MAX_ELEMENTS + if key_type == "stream": + stream = await client.xrange(key, count=_MAX_STREAM) + return [ + {"id": _text(entry_id, 128)[0], "fields": {_text(k, 128)[0]: _text(v)[0] for k, v in fields.items()}} + for entry_id, fields in stream[:_MAX_STREAM] + ], len(stream) >= _MAX_STREAM + # 未知类型（如 module/自定义）：只回类型提示 + return "（该类型暂不支持详情展示）", False diff --git a/backend/app/modules/monitor/cache/controller.py b/backend/app/modules/monitor/cache/controller.py index c46fd1880..1e4c44f8f 100644 --- a/backend/app/modules/monitor/cache/controller.py +++ b/backend/app/modules/monitor/cache/controller.py @@ -1,13 +1,20 @@ from typing import Annotated -from fastapi import APIRouter, Depends, Path, Security +from fastapi import APIRouter, Depends, Path, Query, Security from fastapi.responses import JSONResponse from redis.asyncio.client import Redis from app.common.response import ResponseSchema, SuccessResponse from app.core.dependencies import AuthPermission, redis_getter from app.core.router_class import OperationLogRoute -from app.modules.monitor.cache.schema import CacheInfoSchema, CacheMonitorSchema +from app.modules.monitor.cache.browser import RedisBrowserService +from app.modules.monitor.cache.schema import ( + CacheInfoSchema, + CacheMonitorSchema, + RedisDatabaseSchema, + RedisKeyPageSchema, + RedisValueSchema, +) from .service import CacheService @@ -62,3 +69,38 @@ async def clear_monitor_cache_all_controller( ) -> JSONResponse: result: bool = await CacheService.clear_monitor_cache_all(redis=redis) return SuccessResponse(msg="所有缓存清除成功", data=result) + + +# ── Redis 通用键空间浏览（只读）─────────────────────────────────────── + + +@CacheRouter.get("/redis/databases", summary="Redis 逻辑库列表（只读）", response_model=ResponseSchema[list[RedisDatabaseSchema]], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))]) +async def list_redis_databases_controller( + redis: Annotated[Redis, Depends(redis_getter)], +) -> JSONResponse: + result: list[RedisDatabaseSchema] = await RedisBrowserService.list_databases(redis=redis) + return SuccessResponse(data=result, msg="查询逻辑库成功") + + +@CacheRouter.get("/redis/keys", summary="SCAN 浏览 Redis 键（只读）", response_model=ResponseSchema[RedisKeyPageSchema], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))]) +async def scan_redis_keys_controller( + redis: Annotated[Redis, Depends(redis_getter)], + db: Annotated[int, Query(ge=0, le=15, description="逻辑库序号")] = 0, + pattern: Annotated[str, Query(max_length=128, description="匹配模式（支持 * ?）")] = "*", + cursor: Annotated[int, Query(ge=0, description="SCAN 游标，首页传 0")] = 0, + count: Annotated[int, Query(ge=1, le=500, description="单页数量提示")] = 100, +) -> JSONResponse: + result: RedisKeyPageSchema = await RedisBrowserService.scan_keys( + redis=redis, db=db, pattern=pattern, cursor=cursor, count=count + ) + return SuccessResponse(data=result, msg="扫描键成功") + + +@CacheRouter.get("/redis/value", summary="读取 Redis 单键详情（只读，敏感键脱敏）", response_model=ResponseSchema[RedisValueSchema], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))]) +async def get_redis_value_controller( + redis: Annotated[Redis, Depends(redis_getter)], + key: Annotated[str, Query(min_length=1, max_length=512, description="键名")], + db: Annotated[int, Query(ge=0, le=15, description="逻辑库序号")] = 0, +) -> JSONResponse: + result: RedisValueSchema = await RedisBrowserService.get_value(redis=redis, db=db, key=key) + return SuccessResponse(data=result, msg="读取键详情成功") diff --git a/backend/app/modules/monitor/cache/schema.py b/backend/app/modules/monitor/cache/schema.py index b73f9f1b6..813d88cef 100644 --- a/backend/app/modules/monitor/cache/schema.py +++ b/backend/app/modules/monitor/cache/schema.py @@ -16,3 +16,42 @@ class CacheInfoSchema(BaseModel): cache_name: str = Field(..., description="缓存名称") cache_value: str | None = Field(default=None, description="缓存值") remark: str | None = Field(default=None, description="备注说明") + + +# ── Redis 通用键空间浏览（只读）─────────────────────────────────────── + + +class RedisDatabaseSchema(BaseModel): + """Redis 逻辑库（INFO keyspace 口径）""" + + db: int = Field(..., description="逻辑库序号") + keys: int = Field(default=0, description="键数量") + expires: int = Field(default=0, description="带过期时间的键数量") + + +class RedisKeyMetaSchema(BaseModel): + """键元信息（扫描列表用）""" + + key: str = Field(..., description="键名") + type: str = Field(default="none", description="Redis 类型：string/hash/list/set/zset/stream/none") + ttl: int = Field(default=-1, description="剩余有效期（秒）：-1 永不过期，-2 键不存在") + + +class RedisKeyPageSchema(BaseModel): + """键扫描分页结果（SCAN 游标）""" + + cursor: int = Field(default=0, description="下一页游标；0 表示已扫完") + keys: list[RedisKeyMetaSchema] = Field(default_factory=list, description="本页键") + scanned: int = Field(default=0, description="本页数量") + + +class RedisValueSchema(BaseModel): + """单键详情（只读）""" + + key: str = Field(..., description="键名") + type: str = Field(..., description="Redis 类型") + ttl: int = Field(default=-1, description="剩余有效期（秒）") + size: int | None = Field(default=None, description="占用字节（MEMORY USAGE，可能不可用）") + value: object = Field(default=None, description="值（按类型给出：字符串/列表/映射）") + truncated: bool = Field(default=False, description="是否因超长被截断") + masked: bool = Field(default=False, description="敏感键的值是否已脱敏") diff --git a/backend/app/utils/ai_factory.py b/backend/app/utils/ai_factory.py index 6fcca94b7..34b93857d 100644 --- a/backend/app/utils/ai_factory.py +++ b/backend/app/utils/ai_factory.py @@ -26,6 +26,7 @@ def create_agent( session_id: str, db: Any | None = None, model_config: dict[str, Any] | None = None, + tools: list[Any] | None = None, ) -> Team: """创建带 Agent 的 Team 实例。 @@ -36,6 +37,7 @@ def create_agent( - db (Any | None): Agno 持久化数据库实例，可选。 - model_config (dict | None): 运行时模型配置，覆盖系统默认。 支持字段：base_url, api_key, model_id, temperature。 + - tools (list[Any] | None): 需要注入给 Agent 的工具（**只读工具**，由调用方按开关决定）。 返回: - Team: 配置好的 Team。 @@ -59,7 +61,7 @@ def create_agent( name="fastapiadmin_agent", role="You are a helpful AI assistant", description=self.AGENT_DESCRIPTION, - tools=[], + tools=list(tools or []), ) # 创建 Team diff --git a/backend/tests/test_ai_readonly_tools.py b/backend/tests/test_ai_readonly_tools.py new file mode 100644 index 000000000..f47f0c52b --- /dev/null +++ b/backend/tests/test_ai_readonly_tools.py @@ -0,0 +1,198 @@ +"""AI 只读工具回归测试。 + +固化： +1. 工具清单只含「查询类」且签名里**没有**闭包参数（auth/redis 不能变成模型要填的参数）； +2. 工具内部出错返回 ``{"error": ...}`` 而不是抛异常； +3. 工厂确实把工具注入 Agent；ChatService 的开关关闭时零行为变化。 +""" + +import inspect +from types import SimpleNamespace +from typing import Any + +import pytest + +from app.core.exceptions import CustomException +from app.modules.ai import tools as ai_tools_pkg +from app.modules.ai.chat.service import ChatService +from app.modules.ai.tools import registry as registry_module +from app.modules.ai.tools.registry import build_readonly_tools + + +class _FakeRedis: + def __init__(self, keys: list[str] | None = None) -> None: + self.keys = keys or [] + self.connection_pool = SimpleNamespace(connection_kwargs={"db": 0}) + + async def scan(self, cursor: int = 0, match: str = "*", count: int = 100) -> tuple[int, list[str]]: + return 0, self.keys + + def pipeline(self, transaction: bool = False) -> Any: + key_count = len(self.keys) + + class _Pipe: + async def __aenter__(self) -> "_Pipe": + return self + + async def __aexit__(self, *args: Any) -> bool: + return False + + def type(self, key: str) -> None: + return None + + def ttl(self, key: str) -> None: + return None + + async def execute(self) -> list[Any]: + return ["string", -1] * key_count + + return _Pipe() + + async def aclose(self) -> None: + return None + + +def test_tool_list_is_readonly_and_has_no_injected_params() -> None: + tools = build_readonly_tools(None, None) # type: ignore[arg-type] + assert [tool.name for tool in tools] == [ + "server_status", + "storage_sources", + "storage_files", + "cron_jobs", + "redis_keys", + ] + forbidden = {"auth", "self", "cls", "redis", "db"} + for tool in tools: + assert tool.description.strip() + params = set(inspect.signature(tool.func).parameters) + assert not (params & forbidden), f"{tool.name} 的签名泄露了闭包参数: {params & forbidden}" + + +async def test_redis_keys_returns_names_only() -> None: + tools = {tool.name: tool for tool in build_readonly_tools(None, _FakeRedis(["a:1", "b:2"]))} # type: ignore[arg-type] + result = await tools["redis_keys"].func(pattern="a*", limit=10) + assert result["keys"] == ["a:1", "b:2"] + + +async def test_redis_keys_without_redis_returns_error() -> None: + tools = {tool.name: tool for tool in build_readonly_tools(None, None)} # type: ignore[arg-type] + result = await tools["redis_keys"].func() + assert "error" in result + + +async def test_server_status_error_is_converted(monkeypatch: pytest.MonkeyPatch) -> None: + async def _boom() -> Any: + raise RuntimeError("psutil broken") + + monkeypatch.setattr(registry_module.ServerService, "get_server_monitor_info", staticmethod(_boom)) + tools = {tool.name: tool for tool in build_readonly_tools(None, None)} # type: ignore[arg-type] + result = await tools["server_status"].func() + assert "error" in result and "psutil broken" in result["error"] + + +async def test_storage_files_unknown_source_returns_available_names(monkeypatch: pytest.MonkeyPatch) -> None: + class _FakeCtx: + async def __aenter__(self) -> object: + return object() + + async def __aexit__(self, *args: Any) -> bool: + return False + + class _FakeNodeService: + def __init__(self, auth: Any, db: Any) -> None: + pass + + async def get_list(self) -> list[SimpleNamespace]: + return [SimpleNamespace(id=1, name="node-a", protocol="sftp", status=0, is_default=False)] + + monkeypatch.setattr(registry_module, "async_db_session", lambda: _FakeCtx()) + monkeypatch.setattr(registry_module, "StorageNodeService", _FakeNodeService) + tools = {tool.name: tool for tool in build_readonly_tools(None, None)} # type: ignore[arg-type] + result = await tools["storage_files"].func(source_name="missing") + assert "error" in result + assert result["available"] == ["node-a"] + + +async def test_cron_jobs_returns_compact_rows(monkeypatch: pytest.MonkeyPatch) -> None: + class _FakeResult: + def scalars(self) -> "_FakeResult": + return self + + def all(self) -> list[SimpleNamespace]: + return [ + SimpleNamespace(job_name="sync", trigger_type="cron", next_run_time="2026-01-01 00:00", status=0), + ] + + class _FakeSession: + async def execute(self, stmt: Any) -> _FakeResult: + return _FakeResult() + + async def __aenter__(self) -> "_FakeSession": + return self + + async def __aexit__(self, *args: Any) -> bool: + return False + + monkeypatch.setattr(registry_module, "async_db_session", lambda: _FakeSession()) + tools = {tool.name: tool for tool in build_readonly_tools(None, None)} # type: ignore[arg-type] + rows = await tools["cron_jobs"].func(limit=5) + assert rows == [{"job_name": "sync", "trigger_type": "cron", "next_run_time": "2026-01-01 00:00", "status": 0}] + + +def test_agno_factory_injects_tools(monkeypatch: pytest.MonkeyPatch) -> None: + from app.utils import ai_factory as factory_module + + captured: dict[str, Any] = {} + + class _FakeAgent: + def __init__(self, **kwargs: Any) -> None: + captured["agent"] = kwargs + + class _FakeTeam: + def __init__(self, **kwargs: Any) -> None: + captured["team"] = kwargs + + class _FakeModel: + def __init__(self, **kwargs: Any) -> None: + captured["model"] = kwargs + + monkeypatch.setattr(factory_module, "Agent", _FakeAgent) + monkeypatch.setattr(factory_module, "Team", _FakeTeam) + monkeypatch.setattr(factory_module, "OpenAILike", _FakeModel) + + def _tool() -> str: + return "ok" + + factory_module.AgnoFactory().create_agent("u1", "t1", "s1", tools=[_tool]) + assert captured["agent"]["tools"] == [_tool] + + +def test_chat_service_tools_switch_off_keeps_behavior() -> None: + service = ChatService.__new__(ChatService) + setattr(service, "auth", None) + setattr(service, "redis", None) + assert service._chat_tools(False) == [] + enabled = service._chat_tools(True) + assert len(enabled) == 5 + assert all(callable(item) for item in enabled) + + +def test_ai_tools_router_exposes_tools_listing() -> None: + from app.modules.ai.chat.controller import ChatRouter + + assert any(str(getattr(route, "path", "")).endswith("/tools") for route in ChatRouter.routes) + assert ai_tools_pkg is not None + + +@pytest.mark.parametrize("missing", [CustomException(msg="x")]) +def test_guard_rejects_hallucinated_signature_param(missing: Exception) -> None: + """guard 自检：若签名里出现闭包参数，构造工具时应直接失败（防止回归）。""" + from app.modules.ai.tools.registry import _assert_no_injected_params + + async def _bad_tool(auth: Any) -> str: # noqa: ANN401 - 故意模拟泄漏 + return "x" + + spec = registry_module.ReadonlyToolSpec("bad", "bad tool", _bad_tool) + with pytest.raises(RuntimeError): + _assert_no_injected_params([spec]) + assert isinstance(missing, Exception) diff --git a/backend/tests/test_redis_browser.py b/backend/tests/test_redis_browser.py new file mode 100644 index 000000000..785dd2616 --- /dev/null +++ b/backend/tests/test_redis_browser.py @@ -0,0 +1,228 @@ +"""Redis 通用键空间浏览（只读）回归测试。 + +用假 Redis 客户端验证：INFO keyspace 解析、SCAN 分页与 type/ttl 批量获取、 +按类型取值与截断、敏感键脱敏、缺失键报错、跨库客户端创建与释放。 +""" + +from types import SimpleNamespace +from typing import Any + +import pytest + +from app.core.exceptions import CustomException +from app.modules.monitor.cache import browser as browser_module +from app.modules.monitor.cache.browser import RedisBrowserService, is_sensitive_key + + +class _FakePipeline: + def __init__(self, redis: "_FakeRedis") -> None: + self._redis = redis + self._ops: list[tuple[str, str]] = [] + + async def __aenter__(self) -> "_FakePipeline": + return self + + async def __aexit__(self, *args: Any) -> bool: + return False + + def type(self, key: str) -> None: + self._ops.append(("type", key)) + + def ttl(self, key: str) -> None: + self._ops.append(("ttl", key)) + + async def execute(self) -> list[Any]: + results: list[Any] = [] + for op, key in self._ops: + results.append(self._redis.types.get(key, "none") if op == "type" else self._redis.ttls.get(key, -1)) + return results + + +class _FakeRedis: + def __init__( + self, + *, + keyspace: dict[str, Any] | None = None, + scan_result: tuple[int, list[str]] = (0, []), + types: dict[str, str] | None = None, + ttls: dict[str, int] | None = None, + values: dict[str, Any] | None = None, + sizes: dict[str, int] | None = None, + db: int = 0, + memory_usage_error: bool = False, + ) -> None: + self.keyspace = keyspace or {} + self.scan_result = scan_result + self.types = types or {} + self.ttls = ttls or {} + self.values = values or {} + self.sizes = sizes or {} + self.memory_usage_error = memory_usage_error + self.connection_pool = SimpleNamespace(connection_kwargs={"db": db}) + self.closed = False + self.ops: list[tuple[str, Any]] = [] + + async def info(self, section: str | None = None) -> dict[str, Any]: + self.ops.append(("info", section)) + return self.keyspace + + async def scan(self, cursor: int = 0, match: str = "*", count: int = 100) -> tuple[int, list[str]]: + self.ops.append(("scan", (cursor, match, count))) + return self.scan_result + + def pipeline(self, transaction: bool = False) -> _FakePipeline: + self.ops.append(("pipeline", transaction)) + return _FakePipeline(self) + + async def type(self, key: str) -> str: + self.ops.append(("type", key)) + return self.types.get(key, "none") + + async def ttl(self, key: str) -> int: + self.ops.append(("ttl", key)) + return self.ttls.get(key, -1) + + async def memory_usage(self, key: str) -> int | None: + self.ops.append(("memory_usage", key)) + if self.memory_usage_error: + raise RuntimeError("MEMORY USAGE disabled") + return self.sizes.get(key) + + async def get(self, key: str) -> Any: + self.ops.append(("get", key)) + return self.values[key] + + async def hgetall(self, key: str) -> dict[str, Any]: + self.ops.append(("hgetall", key)) + return self.values[key] + + async def lrange(self, key: str, start: int, end: int) -> list[Any]: + self.ops.append(("lrange", (key, start, end))) + return self.values[key][start : end + 1] + + async def sscan(self, key: str, count: int = 200) -> tuple[int, list[Any]]: + self.ops.append(("sscan", (key, count))) + return 0, self.values[key] + + async def zrange(self, key: str, start: int, end: int, withscores: bool = False) -> list[Any]: + self.ops.append(("zrange", (key, start, end))) + return self.values[key][start : end + 1] + + async def xrange(self, key: str, count: int | None = None) -> list[Any]: + self.ops.append(("xrange", (key, count))) + return self.values[key] + + async def aclose(self) -> None: + self.closed = True + + +async def test_list_databases_parses_keyspace() -> None: + redis = _FakeRedis(keyspace={"db0": "keys=12,expires=3,avg_ttl=0", "db2": "keys=4,expires=0", "other": "x"}) + result = await RedisBrowserService.list_databases(redis) # type: ignore[arg-type] + assert [(item.db, item.keys, item.expires) for item in result] == [(0, 12, 3), (2, 4, 0)] + + +async def test_scan_keys_returns_meta_and_cursor() -> None: + redis = _FakeRedis( + scan_result=(42, ["a:1", "b:2"]), + types={"a:1": "string", "b:2": "hash"}, + ttls={"a:1": -1, "b:2": 300}, + ) + page = await RedisBrowserService.scan_keys(redis, db=0, pattern="a*", cursor=7, count=50) # type: ignore[arg-type] + assert page.cursor == 42 + assert [(item.key, item.type, item.ttl) for item in page.keys] == [("a:1", "string", -1), ("b:2", "hash", 300)] + assert page.scanned == 2 + assert ("scan", (7, "a*", 50)) in redis.ops + + +async def test_scan_keys_other_db_uses_temp_client_and_closes(monkeypatch: pytest.MonkeyPatch) -> None: + created: list[tuple[dict[str, Any], _FakeRedis]] = [] + + def _fake_redis_ctor(**kwargs: Any) -> _FakeRedis: + client = _FakeRedis(scan_result=(0, ["k"]), types={"k": "string"}, ttls={"k": -1}, db=kwargs.get("db", 0)) + created.append((kwargs, client)) + return client + + monkeypatch.setattr(browser_module, "Redis", _fake_redis_ctor) + base = _FakeRedis(db=0) + page = await RedisBrowserService.scan_keys(base, db=3, pattern="*") # type: ignore[arg-type] + assert page.keys and page.keys[0].key == "k" + assert created and created[0][0]["db"] == 3, "跨库浏览应按 db=3 新建客户端" + assert created[0][1].closed is True, "临时客户端用完后必须关闭" + + +async def test_get_value_string_and_truncation() -> None: + redis = _FakeRedis( + types={"plain": "string", "long": "string"}, + ttls={"plain": -1, "long": 60}, + sizes={"plain": 5}, + values={"plain": "hello", "long": "x" * 5000}, + ) + value = await RedisBrowserService.get_value(redis, db=0, key="plain") # type: ignore[arg-type] + assert value.value == "hello" and value.truncated is False and value.masked is False + assert value.size == 5 + + long_value = await RedisBrowserService.get_value(redis, db=0, key="long") # type: ignore[arg-type] + assert long_value.truncated is True + assert isinstance(long_value.value, str) and len(long_value.value) <= 4000 + + +async def test_get_value_masks_sensitive_keys() -> None: + redis = _FakeRedis( + types={"fastapiadmin:access_token:1": "string"}, + ttls={"fastapiadmin:access_token:1": 120}, + values={"fastapiadmin:access_token:1": "real-token"}, + ) + value = await RedisBrowserService.get_value(redis, db=0, key="fastapiadmin:access_token:1") # type: ignore[arg-type] + assert value.masked is True + assert "real-token" not in str(value.value) + assert value.ttl == 120 + assert ("get", "fastapiadmin:access_token:1") not in redis.ops, "敏感键不应读取值" + + +async def test_get_value_missing_key_raises() -> None: + redis = _FakeRedis(types={}) + with pytest.raises(CustomException): + await RedisBrowserService.get_value(redis, db=0, key="nope") # type: ignore[arg-type] + + +async def test_get_value_hash_list_zset_stream() -> None: + redis = _FakeRedis( + types={"h": "hash", "l": "list", "z": "zset", "s": "stream"}, + ttls={"h": -1, "l": -1, "z": -1, "s": -1}, + values={ + "h": {"name": "alice", "role": "admin"}, + "l": ["a", "b", "c"], + "z": [("m1", 1.5), ("m2", 2)], + "s": [("1-0", {"level": "INFO"})], + }, + ) + hash_value = await RedisBrowserService.get_value(redis, db=0, key="h") # type: ignore[arg-type] + assert hash_value.value == {"name": "alice", "role": "admin"} + list_value = await RedisBrowserService.get_value(redis, db=0, key="l") # type: ignore[arg-type] + assert list_value.value == ["a", "b", "c"] + zset_value = await RedisBrowserService.get_value(redis, db=0, key="z") # type: ignore[arg-type] + assert zset_value.value == [{"member": "m1", "score": 1.5}, {"member": "m2", "score": 2.0}] + stream_value = await RedisBrowserService.get_value(redis, db=0, key="s") # type: ignore[arg-type] + assert stream_value.value == [{"id": "1-0", "fields": {"level": "INFO"}}] + + +async def test_get_value_memory_usage_failure_is_tolerated() -> None: + redis = _FakeRedis(types={"k": "string"}, values={"k": "v"}, memory_usage_error=True) + value = await RedisBrowserService.get_value(redis, db=0, key="k") # type: ignore[arg-type] + assert value.size is None + assert value.value == "v" + + +@pytest.mark.parametrize( + ("key", "expected"), + [ + ("fastapiadmin:access_token:1", True), + ("fastapiadmin:secret:db", True), + ("user:session:9", True), + ("sys:config:site_name", False), + ("task:cronjob:list", False), + ], +) +def test_is_sensitive_key(key: str, expected: bool) -> None: + assert is_sensitive_key(key) is expected diff --git a/frontend/web/src/api/module_monitor/cache.ts b/frontend/web/src/api/module_monitor/cache.ts index 729eda694..8874f5493 100644 --- a/frontend/web/src/api/module_monitor/cache.ts +++ b/frontend/web/src/api/module_monitor/cache.ts @@ -31,6 +31,32 @@ const CacheAPI = { }); }, + /** Redis 逻辑库列表（只读） */ + getRedisDatabases() { + return request>({ + url: `${API_PATH}/redis/databases`, + method: "get", + }); + }, + + /** SCAN 浏览 Redis 键（只读） */ + scanRedisKeys(params: { db?: number; pattern?: string; cursor?: number; count?: number }) { + return request>({ + url: `${API_PATH}/redis/keys`, + method: "get", + params, + }); + }, + + /** 读取 Redis 单键详情（只读；敏感键已脱敏） */ + getRedisValue(params: { db?: number; key: string }) { + return request>({ + url: `${API_PATH}/redis/value`, + method: "get", + params, + }); + }, + deleteCacheName(cacheName: string) { return request({ url: `${API_PATH}/delete/name/${cacheName}`, @@ -93,3 +119,31 @@ export interface CacheMonitor { db_size: number; info: RedisInfo; } + +export interface RedisDatabase { + db: number; + keys: number; + expires: number; +} + +export interface RedisKeyMeta { + key: string; + type: string; + ttl: number; +} + +export interface RedisKeyPage { + cursor: number; + keys: RedisKeyMeta[]; + scanned: number; +} + +export interface RedisValue { + key: string; + type: string; + ttl: number; + size: number | null; + value: unknown; + truncated: boolean; + masked: boolean; +} diff --git a/frontend/web/src/views/module_ai/chat/index.vue b/frontend/web/src/views/module_ai/chat/index.vue index a8d66ae90..61455e3b3 100644 --- a/frontend/web/src/views/module_ai/chat/index.vue +++ b/frontend/web/src/views/module_ai/chat/index.vue @@ -36,6 +36,12 @@ />
+
+ + 启用只读工具（服务器状态 / 存储源 / 定时任务 / Redis 键名） +
([]); const sending = ref(false); +// 是否启用 AI 只读工具（默认关闭：不改变原有对话行为） +const useTools = ref(false); const isConnected = ref(false); const connectionStatus = ref<"connected" | "connecting" | "disconnected">("disconnected"); const error = ref(""); @@ -244,6 +252,7 @@ const handleSendMessage = async (message: string, files?: UploadedFile[]) => { message, session_id: currentSessionId.value, files: files?.map((f) => ({ name: f.name, type: f.type, size: f.size })), + use_tools: useTools.value, }) ); // 注意：sending 状态保持为 true，等待 [DONE] / [STOPPED] 标记清除 diff --git a/frontend/web/src/views/module_monitor/cache/index.vue b/frontend/web/src/views/module_monitor/cache/index.vue index 4624adc85..57ebcf3cb 100644 --- a/frontend/web/src/views/module_monitor/cache/index.vue +++ b/frontend/web/src/views/module_monitor/cache/index.vue @@ -265,6 +265,98 @@
+ + + +
+ 逻辑库 + + + + + 扫描 + 加载更多 + 只读浏览；敏感键（token/session 等）的值自动脱敏 +
+ + + + + + + + + + + + + + + + {{ redisValue?.type ?? "-" }} + + {{ + redisValue?.ttl === -1 + ? "永不过期" + : redisValue?.ttl === -2 + ? "不存在" + : `${redisValue?.ttl ?? "-"}s` + }} + + + {{ redisValue?.size == null ? "-" : `${redisValue.size} B` }} + + + + + + + @@ -276,7 +368,10 @@ import CacheAPI, { type CacheInfo, type CacheForm, type CacheMonitor, + type RedisDatabase, type RedisInfo, + type RedisKeyMeta, + type RedisValue, } from "@/api/module_monitor/cache"; import { echarts } from "@/plugins/echarts"; import { useWindowSize } from "@vueuse/core"; @@ -295,6 +390,66 @@ const loading = ref(true); const subLoading = ref(false); const nowCacheName = ref(""); +// ─── Redis 键空间浏览（只读）──────────────────────────────────────── +const redisDatabases = ref([]); +const redisDb = ref(0); +const redisPattern = ref("*"); +const redisKeys = ref([]); +const redisCursor = ref(0); +const redisLoading = ref(false); +const redisValueVisible = ref(false); +const redisValue = ref(null); +const redisValueText = computed(() => + redisValue.value ? JSON.stringify(redisValue.value.value, null, 2) : "" +); + +async function loadRedisDatabases() { + const res = await CacheAPI.getRedisDatabases(); + redisDatabases.value = res.data.data || []; + if ( + redisDatabases.value.length && + !redisDatabases.value.some((item) => item.db === redisDb.value) + ) { + redisDb.value = redisDatabases.value[0]!.db; + } +} + +async function scanRedisKeys(reset: boolean) { + if (reset) { + redisKeys.value = []; + redisCursor.value = 0; + } + redisLoading.value = true; + try { + const res = await CacheAPI.scanRedisKeys({ + db: redisDb.value, + pattern: redisPattern.value || "*", + cursor: reset ? 0 : redisCursor.value, + count: 100, + }); + const page = res.data.data; + const incoming = page?.keys || []; + redisKeys.value = reset ? incoming : [...redisKeys.value, ...incoming]; + redisCursor.value = page?.cursor ?? 0; + } finally { + redisLoading.value = false; + } +} + +async function openRedisValue(row: unknown) { + const target = row as RedisKeyMeta; + const res = await CacheAPI.getRedisValue({ db: redisDb.value, key: target.key }); + redisValue.value = res.data.data ?? null; + redisValueVisible.value = true; +} + +watch(activeTab, async (tab) => { + if (tab === "2" && !redisDatabases.value.length) { + await loadRedisDatabases(); + await scanRedisKeys(true); + } +}); + const commandstats = ref(null); const usedmemory = ref(null); const cache = ref({