人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载本篇技术指南深入剖析 openJiuwen agent-coreopenjiuwen/core/common/security/crypt_utils.py中的加密/解密抽象层设计从统一接口BaseCrypt、内置 AES-256-GCM 单例实现AesGcmCrypt到按名管理的线程安全注册表CryptUtils再到其在记忆引擎中的落地消费方式BaseMemoryManager的记忆加解密与AesStorageCodec存储编解码。读完本文你将掌握如何基于该模块加密 Agent 的敏感配置与记忆数据、如何注册自定义加密算法如 SM4以及如何在加密失败时优雅降级。模块定位与设计意图crypt_utils位于 openJiuwen agent-core 的 common 公共能力层是面向全项目的加密/解密抽象门面。它解决的核心问题是不同业务模块记忆存储、配置持久化、敏感数据落盘对加密算法的诉求不同且密钥生命周期管理复杂因此不适合让每个模块各自硬编码加密逻辑。该模块采用三层架构接口契约层BaseCrypt抽象基类统一encrypt/decrypt两个方法签名算法实现层AesGcmCrypt内置的 AES-256-GCM 实现采用单例模式全局唯一门面注册表层CryptUtils以 名称 → 实例 的 Key-Value 方式管理多个加密实现线程安全。从源码结构看该模块同时服务于多个上层消费者记忆管理基类 base_memory_manager.py 和存储编解码器 aes_storage_codec.py它们均通过CryptUtils.get_crypt(aes_gcm)间接获取加密实例实现上层不感知具体算法的解耦。BaseCrypt统一加解密接口契约BaseCrypt是一个抽象基类继承自abc.ABC定义了两条必须实现的抽象方法是所有自定义加密实现的统一契约。encryptabstractmethod def encrypt(self, key: bytes, origin: str) - strkeybytes加密密钥长度由具体实现决定。AES-GCM 实现要求 32 字节originstr待加密的明文字符串返回str加密后的密文字符串。decryptabstractmethod def decrypt(self, key: bytes, encrypt_str: str) - strkeybytes解密密钥必须与加密时使用的密钥一致encrypt_strstr待解密的密文字符串返回str解密后的明文字符串。源码中该基类位于 crypt_utils.py实现为空方法体...仅承担接口约束职责。任何新的加密算法例如国产 SM4、国密算法或业务自定义方案只需继承BaseCrypt并实现这两个方法即可被CryptUtils注册表统一管理。AesGcmCrypt内置 AES-256-GCM 单例实现AesGcmCrypt是框架自带的加密实现基于pycryptodome库的Crypto.Cipher.AES采用 AES-256-GCM 认证加密模式AEAD密钥 32 字节、Nonce 12 字节、Tag 16 字节。其声明为class AesGcmCrypt(BaseCrypt, metaclassSingleton)crypt_utils.py并使用模块底部AesGcmCrypt()这一顶层语句在导入时自动完成单例创建与注册crypt_utils.py。关键设计点设计点说明单例模式使用 singleton.py 中线程安全的Singleton元类进程内全局唯一实例自动注册__init__中调用CryptUtils.register_crypt(CryptUtils.AES_GCM_CRYPT_NAME, self)注册名固定为aes_gcm不存储密钥密钥不保存在实例内由调用方在每次encrypt/decrypt时显式传入随机 Nonce每次加密使用secrets.token_bytes(NONCE_LENGTH)生成随机 Nonce同一明文同一密钥每次产生不同密文输出格式拼接十六进制串nonce_hex tag_hex ciphertext_hex24 位 32 位 可变长度encrypt 实现原理def encrypt(self, key: bytes, origin: str) - str: AesGcmCrypt._validate_key(key) nonce secrets.token_bytes(NONCE_LENGTH) cipher AES.new(keykey, modeAES.MODE_GCM, noncenonce, mac_lenTAG_LENGTH) cipher_text, tag cipher.encrypt_and_digest(origin.encode(encodingutf-8)) return f{nonce.hex()}{tag.hex()}{cipher_text.hex()}crypt_utils.py关键点_validate_key强制密钥长度为 32 字节否则抛出BaseError错误码COMMON_ENCRYPTION_ERROR见 codes.pyGCM 模式是认证加密encrypt_and_digest一次性完成加密与 Tag 计算Tag 用于解密时的完整性校验可同时保证机密性与完整性能检测密文篡改明文以 UTF-8 编码天然支持中文、emoji 等多字节字符Nonce、Tag、密文三者各自hex()后拼接不引入 Base64 等额外编码输出是纯十六进制字符可直接存入数据库字段或 JSON 配置。decrypt 实现原理def decrypt(self, key: bytes, encrypt_str: str) - str: AesGcmCrypt._validate_key(key) nonce_len NONCE_LENGTH * 2 tag_len TAG_LENGTH * 2 min_len nonce_len tag_len if len(encrypt_str) min_len: raise build_error(...) # COMMON_DECRYPTION_ERROR ... cipher AES.new(keykey, modeAES.MODE_GCM, noncenonce_bytes) plaintext_bytes cipher.decrypt_and_verify( ciphertextciphertext_bytes, received_mac_tagtag_bytes ) return plaintext_bytes.decode(encodingutf-8)crypt_utils.py解密侧的防御性校验层层递进密钥校验非 32 字节抛出COMMON_ENCRYPTION_ERROR最短长度校验密文必须至少12×2 16×2 56个十六进制字符Nonce 段 Tag 段不足则抛出COMMON_DECRYPTION_ERROR分段解析按nonce_hex[:24]、tag_hex[24:56]、ciphertext_hex[56:]切分并转回字节段长度校验Nonce 非 12 字节、Tag 非 16 字节均抛出COMMON_DECRYPTION_ERROR认证解密decrypt_and_verify校验 Tag若密文被篡改、密钥不匹配或数据损坏认证失败抛出异常——这是 GCM 模式防篡改能力的落地体现。对应的错误码定义在 codes.pyCOMMON_ENCRYPTION_ERROR (188006, ...)、COMMON_DECRYPTION_ERROR (188007, ...)异常统一通过build_error构造为框架标准的BaseError。CryptUtils线程安全的加解密注册表CryptUtils是加解密门面类负责管理全局加密实例注册表采用 Key-Value 模式注册实例而非工厂因为加密实例通常是有状态的。源码见 crypt_utils.py。类属性属性类型说明AES_GCM_CRYPT_NAMEstr内置 AES-GCM 的注册名值为aes_gcm_CRYPT_REGISTRYdict[str, BaseCrypt]全局注册表_registry_lockthreading.Lock保护注册表的互斥锁三个静态方法register_crypt(name, crypt)注册前强制校验crypt必须是BaseCrypt实例否则抛出BaseErrorCOMMON_ENCRYPTION_ERROR在锁内写入注册表同名覆盖更新后注册者生效。unregister_crypt(name)在锁内执行pop(name, None)对不存在的名称静默忽略不会抛错。get_crypt(name)在锁内读取并返回对应实例未注册时返回NoneOptional[BaseCrypt]。源码中使用threading.Lock()保护注册与查询操作crypt_utils.py配合单例元类自身的锁保证了并发环境下注册表读写与单例创建的一致性。上手使用内置 AES-GCM 与自定义算法注册方式一使用内置 AES-256-GCM自动注册from openjiuwen.core.common.security.crypt_utils import AesGcmCrypt, CryptUtils # 创建单例首次构造时自动注册到 CryptUtils注册名 aes_gcm crypt AesGcmCrypt() # 加解密 key b0123456789abcdef0123456789abcdef # 32 字节密钥 ciphertext crypt.encrypt(key, Hello, World!) plaintext crypt.decrypt(key, ciphertext) print(plaintext) # Hello, World! # 通过 CryptUtils 注册表获取同一实例 same_crypt CryptUtils.get_crypt(aes_gcm) print(same_crypt is crypt) # True注意AesGcmCrypt()在模块导入时已自动初始化一次因此直接CryptUtils.get_crypt(aes_gcm)通常即可拿到实例显式调用AesGcmCrypt()只是再次确认单例。方式二注册自定义加密实现from openjiuwen.core.common.security.crypt_utils import CryptUtils, AesGcmCrypt, BaseCrypt # 内置 AES-GCM自动注册 aes_crypt AesGcmCrypt() # 自定义加密实现只需继承 BaseCrypt 并实现两个方法 class Sm4Crypt(BaseCrypt): def encrypt(self, key: bytes, origin: str) - str: # SM4 加密实现 ... def decrypt(self, key: bytes, encrypt_str: str) - str: # SM4 解密实现 ... sm4 Sm4Crypt() CryptUtils.register_crypt(sm4, sm4) # 按名获取并使用 crypt CryptUtils.get_crypt(sm4) if crypt: ciphertext crypt.encrypt(key, sensitive data) # 注销某个注册项 CryptUtils.unregister_crypt(sm4)这种按名取用的注册表模式让上层业务完全与算法解耦未来更换或增加算法时无需改动记忆引擎等调用方代码只需注册新实例即可。上层集成一BaseMemoryManager 记忆加解密BaseMemoryManager的encrypt_memory_if_needed与decrypt_memory_if_needed两个静态方法是CryptUtils的直接消费方在记忆存储/读取链路中自动完成加解密。实现位于 base_memory_manager.py。加密流程encrypt_memory_if_needed(key, plaintext) - plaintext 为空 - 直接返回 plaintext key 为空 - 直接返回 plaintext CryptUtils.get_crypt(aes_gcm) 为 None - 直接返回 plaintext crypt.encrypt(key, plaintext) - 返回密文字符串 加密异常 - 记录 warning 日志 - 返回 plaintext解密流程decrypt_memory_if_needed(key, ciphertext) - ciphertext 为空 - 直接返回 ciphertext key 为空 - 直接返回 ciphertext CryptUtils.get_crypt(aes_gcm) 为 None - 直接返回 ciphertext crypt.decrypt(key, ciphertext) - 返回明文字符串 解密异常 - 记录 warning 日志 - 返回 ciphertext这套四级降级设计非常务实未配置密钥、明文/密文为空、注册表不可用、加解密抛异常四种情况下均不中断业务而是回退为透传并记录memory_loggerwarning事件类型MEMORY_PROCESS。这意味着记忆加密是一种尽力而为的安全增强不会因为加密组件故障导致 Agent 记忆功能瘫痪。单元测试 test_base_memory_manager_crypto.py 完整覆盖了这些分支包括有密钥时加解密往返成功且密文为十六进制字符无密钥、空字符串、未注册 crypt 时均原样返回加密/解密抛出异常时返回原文并触发 warning 日志密钥不匹配时解密失败认证校验生效支持中文与 emoji中文测试 émojis以及历史遗留密文的兼容解密。上层集成二AesStorageCodec 存储编解码除记忆管理基类外aes_storage_codec.py 中的AesStorageCodec实现了StorageCodec协议通过BaseMemoryIndex.set_storage_codec注入透明地对记忆文档的text字段进行加解密。其encode/decode逻辑与记忆加密一致key 为空、文本为空或 crypt 缺失时透传异常时降级返回原文并记录 warning。这印证了CryptUtils抽象层的复用价值——同一加密实现被管理级和存储级两条链路共同消费。测试验证行为契约的源码级证据crypt_utils模块自身的单元测试位于 test_crypt_utils.py可作为使用该模块时的行为契约参考加解密往返中文等多字节文本加密后可无损还原语义安全同一明文两次加密产生不同密文随机 Nonce 生效参数校验非 32 字节密钥加密/解密均报错Key must be 32 bytes过短密文报错Ciphertext too short垃圾字符串解密失败跨密钥失败用不同密钥解密会认证失败注册表行为注册后按名取到同一实例注册非BaseCrypt对象抛must be a BaseCrypt注销后查询返回None同名覆盖后取到新实例并发安全多线程并发注册/注销无异常10 个线程并发创建单例得到同一个实例。这些测试同时验证了单例元类 singleton.py 的线程安全设计与注册表锁的正确性。安全使用建议基于该模块的源码设计与 AES-GCM 算法特性实践中有几点值得注意密钥管理密钥需为 32 字节AES_KEY_LENGTH 32且不应硬编码在源码中建议从环境变量、密钥管理系统或配置文件注入加密和解密必须使用同一密钥密钥变更影响由于 GCM 认证机制使用旧密钥加密的历史数据无法用新密钥解密密钥轮换需考虑存量数据迁移或双密钥支持密文特征加密输出为纯十六进制字符串无{...}等 JSON 包裹可直接存储若需入库索引注意密文不可搜索、不可排序应建立密文与原文 ID 的映射降级语义encrypt_memory_if_needed等在异常时透传原文这是可用性优先的设计决策若你的场景要求强制加密如合规审计应在调用层补充策略校验不能依赖此处的静默降级依赖前提模块基于pycryptodome的Crypto.Cipher.AES使用前需确保环境已安装该依赖可参考仓库 pyproject.toml 的依赖声明。小结crypt_utils是 openJiuwen agent-core 中小而精的安全基础组件BaseCrypt定义契约、AesGcmCrypt提供经过认证加密的 AES-256-GCM 默认实现、CryptUtils以线程安全注册表统一编排多算法并通过BaseMemoryManager与AesStorageCodec渗透到记忆与存储链路。对于需要保护 Agent 敏感数据的开发者而言直接复用AesGcmCrypt或注册自有算法即可在不侵入业务代码的前提下获得一致的加密能力与统一的异常降级语义。赞分享人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载相关推荐openJiuwen agent-core 通用模块安全加密实践基于 CryptUtils 的 AES-256-GCM 加解密体系解析openJiuwen agent core 通用模块安全加密实践基于 CryptUtils 的 AES 256 GCM 加解密体系解析 导读 本文基于 ope人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习Apache Pulsar 端到端加密PIP-4实战指南基于 ECDSA/RSA 与 AES-256-GCM 的消息加解密实现Apache Pulsar 端到端加密PIP 4实战指南基于 ECDSA/RSA 与 AES 256 GCM 的消息加解密实现 本文以 Apache Pu消息队列流处理后端微服务消息路由Password Managerpv基于 Argon2id 与 AES-256-GCM 的加密命令行密码管理器实战指南Password Managerpv基于 Argon2id 与 AES 256 GCM 的加密命令行密码管理器实战指南 本篇技术指南围绕 Cybersec上一篇告别AI嘴瓢FaceFusion双引擎唇形同步技术让虚拟人开口更自然下一篇GLM-4-9B-Chat与GLM-4V-9B多模态模型对比如何选择适合的AI方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考