Cheshire Cat框架StrayCat会话对象隔离与内存管理机制分析

漏洞关键信息总结 漏洞概述 漏洞类型：会话对象（Session Object）在每次 HTTP 请求和 WebSocket 连接时创建，并暴露给插件。 核心机制：框架为每个请求/连接创建 Session 实例，插件可通过 访问该对象。 主要功能： - 提供 （工作记忆）用于存储和检索会话状态。 - 提供 记录模型交互。 - 提供 方法用于通过 WebSocket 发送消息。 影响范围 受影响组件： 中的 类。 影响对象：所有通过 访问会话状态的插件。 关键数据结构： - ：字典结构，用于存储会话状态。 - ：记录模型交互历史。 - 和 ：用于生成报告。 修复方案 会话隔离：确保每个 HTTP 请求和 WebSocket 连接使用独立的 Session 实例，避免状态污染。 内存管理：通过 的 和 方法管理会话状态，确保数据一致性。 消息发送：使用 方法发送消息，确保消息通过 WebSocket 正确传递。 POC/利用代码 ```python class StrayCat: """ The Stray cat goes around tools, hooks and endpoints, making troubles """ def __init__(self): self.user_data = AuthUser() self.user_id = user_data.name + "0000" self.user_data = user_data get working memory from cache or create a new one self.working_memory = self.load_working_memory_from_cache() def _repr_(self): return f"StrayCat(user_id={self.user_id}, user_name={self.user_name})" def _send_ws_json(self, data: Any): Run the coroutine in the main event loop in the main thread and wait for the result app = CheshireCat().fastapi_app ws_manager = app.state.websocket_manager ws_connection = ws_manager.get_connection(self.user_id) if not ws_connection: log.debug(f"No websocket connection is open for user {self.user_id}") return asyncio.run_coroutine_threadsafe( ws_connection.send_json(data), app.state.event_loop, ).result() def _build_ai(self) -> MessageBag: """Build data structure for output (response and why and memories) e.g. TODO: these 3 lines are a mess, simplify episodic_report = [ dict([("type", "Episodic"), ("id", d[0]) for d in self.working_memory.episodic_memories ] declarative_report = [ dict([("type", "Declarative"), ("id", d[0]) for d in self.working_memory.declarative_memories ] procedural_report = [ dict([("type", "Procedural"), ("id", d[0]) for d in self.working_memory.procedural_memories ] Why this response? why = MessageBag( episodic_report=episodic_report, declarative_report=declarative_report, procedural_report=procedural_report, memory={ "episodic": episodic_report, "declarative": declarative_report, "procedural": procedural_report, }, model_interactions=self.working_memory.model_interactions, ) return why def load_working_memory_from_cache(self): """Load the working memory from the cache.""" self.working_memory = ( self.cache.get_value(f"{self.user_id}.working_memory") or WorkingMemory() ) self.update_working_memory_cache(self) """Update the working memory in the cache.""" cache_item = CacheItem(f"{self.user_id}.working_memory", self.working_memory, -1) self.cache.insert(updated_cache_item) def send_ws_message(self, content: str """Send a message via websocket. This method is useful for sending a message via websocket directly without passing through the LLM. In case there is no connection the message is skipped and a warning is logged. Parameters ---------- content : str The content of the message. msg_type : str The type of the message. Should be either "notification" (default), "chat", "chat_token" or "error" Examples -------- Send a notification via websocket >>> cat.send_ws_message("Hello, I'm a notification!") Send a chat message via websocket """

目标达成感谢每一位支持者 — 我们达成了 100% 目标！

Cheshire Cat框架StrayCat会话对象隔离与内存管理机制分析

同源更多文章

目标达成 感谢每一位支持者 — 我们达成了 100% 目标！

Cheshire Cat框架StrayCat会话对象隔离与内存管理机制分析

同源更多文章

目标达成感谢每一位支持者 — 我们达成了 100% 目标！