Files
AgentLab/backend/app/core/crypto.py
T
ThearchyHelios 964eeaa829 AgentLab:可视化 Agent 编排实验台
在画布上拖拽编排 agent 工作流,并实时看着它执行。

后端 FastAPI + LangGraph:把画布 JSON 编译成 StateGraph,15 种节点
(模型/Agent/多Agent、工具/沙箱、分支/循环/子图、记忆/检索/整形、
人工介入/结构校验)。选 LangGraph 是因为可视化编排真正难的是暂停、
恢复和回放,而 checkpointer + interrupt() 正好把这几件事做在了底层。

前端 React + React Flow:节点元数据驱动属性面板与连接桩,运行时
节点高亮、边流动画、卡片内流式 token 全部由同一份事件流驱动。

关键实现:
- Agent 内工具审批用 @task 缓存模型调用,interrupt 重放时不重复计费
- 沙箱保持只读根文件系统,代码经 exec+base64 写入(put_archive 会被
  只读容器拒绝),断网、非 root、内存/CPU/超时限额
- 检索为向量与 BM25 混合,本地哈希 embedding 零配置可用
- 分支条件走 AST 白名单求值器;HTTP 工具逐 IP 拦截内网与元数据地址
- Mock provider 让无 API Key 也能跑通完整链路

7 个内置模板全部冒烟通过,含真实模型端到端验证。
2026-09-13 17:27:00 +08:00

61 lines
1.7 KiB
Python

from __future__ import annotations
import base64
import hashlib
import os
from cryptography.fernet import Fernet, InvalidToken
from app.core.config import settings
_PREFIX = "enc::"
def _load_key() -> bytes:
"""取加密密钥:优先用配置里的 secret_key,否则在 data_dir 生成一个持久化的。"""
raw = settings.secret_key
if not raw:
key_file = settings.data_dir / ".secret_key"
if key_file.exists():
raw = key_file.read_text().strip()
else:
raw = base64.urlsafe_b64encode(os.urandom(32)).decode()
key_file.write_text(raw)
key_file.chmod(0o600)
# 任意字符串都归一化成 32 字节 Fernet key
digest = hashlib.sha256(raw.encode()).digest()
return base64.urlsafe_b64encode(digest)
_fernet = Fernet(_load_key())
def encrypt(value: str | None) -> str | None:
"""加密明文。已经是密文的原样返回,方便重复保存同一条记录。"""
if not value:
return value
if value.startswith(_PREFIX):
return value
return _PREFIX + _fernet.encrypt(value.encode()).decode()
def decrypt(value: str | None) -> str | None:
if not value:
return value
if not value.startswith(_PREFIX):
return value # 兼容历史明文
try:
return _fernet.decrypt(value[len(_PREFIX) :].encode()).decode()
except InvalidToken:
return None
def mask(value: str | None) -> str:
"""给前端展示用的掩码,永远不把明文 key 发回浏览器。"""
if not value:
return ""
plain = decrypt(value) or ""
if len(plain) <= 8:
return "••••"
return f"{plain[:4]}••••{plain[-4:]}"