diff --git a/README.md b/README.md index 0557518..fe2cd5f 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ | pipeline_agent_instances | 角色 agent 实例注册（项目×角色→稳定实例 id；唯一写入方 agent_instance.py；m0002 建表未进 models，2026-09-10 归籍） | | pipeline_agent_settings | 用户 agent 偏好（每用户一行 uk_user：默认模型/当前项目指针；写入方 agent_loop_v2、set_agent_model.dspy、workspace.py；2026-09-10 归籍） | | pipeline_project_agents | 项目角色 agent 配置（驾驶舱 v1 start_agent 链路；v2 已走 pipeline_agent_instances，保留至 v1 链路删除决议；2026-09-10 归籍） | +| pipeline_user_secrets | 用户敏感信息库（2026-09-17，m0037 建表）：AES 密文 + 指纹去重 + org/user 双维归属；写入方 secret_vault.py（agent 工具 + gateway 入站自动侦测）；唯一索引 (org_id,user_id,name) | ## 安装 @@ -51,6 +52,7 @@ cd pkgs/pipeline-service && pip install . | `bg_jobs.py` | run_command 后台任务（状态文件化 .bg/，跨 worker 可见） | | `subagents.py` | 后台并行子 agent 委派（.sub/，spawn/steer/stop/result） | | `diagram_gate.py` | 文档配图形态硬门禁（2026-09-15）：交付文档/标书章节中的图必须 invoke_model t2i/i2i 真图，确定性检测 mermaid/plantuml/ASCII 伪图；deliver 双路径拦截 + pipeline-bidding write/review_chapter 复用；豁免=文中如实标注「配图缺失：平台无可用文生图模型」 | +| `secret_vault.py` | 用户敏感信息库（2026-09-17）：AES 密文入库 + 会话占位符引用 `@@sec:NAME@@` + 入站自动侦测入库；真值只在执行边界以子进程 env 注入，永不进模型上下文/落库/进命令字符串；gateway 入站门禁 + v1 角色 agent（项目 owner 身份）+ v2 会话 agent 三路接入 | ## 通用会话产线隔离（2026-09-05，7 层纵深防御） @@ -184,3 +186,120 @@ timeout → payload 无 _timeout → inference 按供应商端点配置（百炼 外层 wait_for（若有）≥ T+60。agent_loop：510→570→600 实证对齐。 验证特征串：inference 日志「预算=Ns」；llm_call_trace status_code=0 计数。 新增 LLM 调用点只需（可选）声明总预算，禁止自算客户端等待。 + +## 用户敏感信息库（secret_vault，2026-09-17） + +**要解决的问题**：用户在会话里直接贴 apikey/token 时，明文会进模型上下文 → 随每轮 API +调用完整上行供应商 → 落 `pipeline_conversations` 永久留存 → session_search 捞回再上行。 +（Hermes 侧同源分析实测：一个 5 轮任务的 key 上行 3~5 次，含重试可翻倍。） + +**设计不变量（写死在代码，不靠 LLM 自觉）** + +1. **真值只以子进程环境变量形式存在**，绝不拼进命令字符串。占位符 `@@sec:NAME@@` + 在执行边界转成 `$PIPELINE_SEC_NAME`，值经 `subprocess env=` 注入。 + → 命令原文可安全落库/进上下文；模型抄回占位符是**预期行为**（不像脱敏成 `***` + 会让模型把占位符抄进命令导致任务全断）。 +2. **env 注入走 `env=`，不用 bwrap `--setenv`**：后者把值放进 bwrap 的 argv，同机任意 + 用户 `ps aux` / `/proc//cmdline` 可读（world-readable）；`env=` 落在 + `/proc//environ`，仅属主与 root 可读。bwrap 未用 `--clearenv`，默认继承自身环境， + 所以 `env=` 的值能进沙箱内。 +3. **未知占位符原样保留 + 拒绝执行**，绝不替换成空串。空串会静默改变命令语义 + （实测 `rm -rf ${TARGET_DIR}$/` → `rm -rf /`）。 +4. **读写权限范围分离**（`get_secret_row(own_only=)`）：读/用路径含机构共享条目 + （`user_id=''`，共享凭据就是给成员用的）；写路径（save/delete/set_status）只匹配本人 + 条目 + `_assert_own` 代码级二次校验。否则任意机构成员可覆盖/删除全机构共用凭据。 +5. **入站侦测只吃高置信度**：已知凭据前缀（16 种）或关键词邻域 + 熵 ≥3.5。 + 中低置信度只提示不自动入库——误报会把哈希/base64 数据块当凭据吞掉，静默毁数据， + 比漏报更糟。**不做无上下文裸串侦测**（同理）。 + +**占位符语法为什么是 `@@sec:NAME@@`** + +不能用 `${X}$`：sqlor 参数化（`sor.py:78` `sqlvp='${', sqlvs='}$'`）与 ahserver +`baseProcessor.py:193` 的 `ArgsConvert("${","}$")` 都用这个语法，复用会被误展开。 +也不能用裸 `$VAR`：bash 会就地展开，绕过执行边界。 + +**加密：AES（appPublic.aes），不是 password_encode** + +`ahserver.globalEnv.password_encode/decode`（:255-261）实测是 **RC4** 薄封装 +（`:61 from appPublic.rc4 import password, unpassword`），换过去等于没换。 +真正的 AES 在 `appPublic.aes`（AES-ECB + PKCS7 + base64），与 DB 连接密码同套。 + +⚠️ `aes.py` 内有**两处方向相反**的 iso-8859-1，只有一处需要包装： + +| 位置 | 方向 | 结论 | +|---|---|---| +| `aes.py:42-43,46-47` | 密文 bytes → str | **必须** iso-8859-1。实测 200 组 AES 密文用 utf-8 解码失败 200 组（100%，首字节 0xef 即 invalid continuation byte）——密文是二进制，不符合 UTF-8 规范。历史正确设计，**不要动** | +| `aes.py:29,37` | 明文 str → bytes | iso-8859-1 表示不了非 ASCII。实测「密码是中文Abc123」「🔑key123」直接 UnicodeEncodeError | + +解法在**调用方**包装（不改 aes.py——appPublic 是多宿主共享基础模块，改它影响 DB 密码 +等全部调用方）：`str → UTF-8 bytes → 按 latin-1 逐字节还原成等长 str 传入`。latin-1 是 +字节保真映射（0x00-0xFF 一一对应），`aes.py:29` 再 `encode('iso-8859-1')` 即还原成原始 +UTF-8 字节流 = 字节保真空操作。出库反向还原。包装后实测 9/9 往返一致（ASCII / GitHub +PAT / 中文 / emoji / 中英混合+符号 / 私钥含换行 / 512 长值 / `p@$$w0rd!#%^&*()` / +含 `@@sec:` 语法的值），密文零明文残留。 + +密文带版本前缀 `AES1$`：base64 密文无固定形态特征，无法像 RC4 的 `QUZVcX` 靠前缀嗅探 +判断是否已加密；显式版本标记同时为算法迁移留路。`looks_encrypted()` 同时识别 `AES1$` +与历史 `QUZVcX`（幂等加密防多层叠加 + 向后兼容读取 llm 表等存量 RC4 密文）。 + +**三路接入** + +| 路径 | 位置 | 身份来源 | 工具集 | +|---|---|---|---| +| 入站门禁 | `gateway.run_message` Step 0.5（所有上下文解析**之前**） | `get_user()` | 自动侦测→入库→替换，progress 事件回显动作 | +| v2 会话 agent | `agent_loop_v2._t_secret_tool` + `_t_run_command` + `_t_process` | `self.user_id`（作用域解析里被 generic 置空的那份**不能用**） | 全六件 | +| v1 角色 agent | `agent_loop._exec_agent_tool` + `run_shell` 分支 | 项目真人 owner（`rag_client.resolve_project_owner`，已处理 `created_by='agent.pm'` 自动立项反查） | 只读三件（list/use/detect）——无人值守下 LLM 自主增删凭据风险高于收益 | + +入站门禁必须在 Step 0.5：净化后的 content 才是该进模型上下文与落库的版本， +明文一旦进 executor 就会每轮上行供应商。 + +出站净化两处：`_t_run_command`（前台）+ `_t_process`（包一层 `_t_process_raw`， +后台 output.log 可能回显凭据）。执行边界必须在前台/后台分支**之前**，否则后台命令不受保护。 + +**前端（AgentIO 钥匙图标）** + +bricks 是多应用共用框架，**不硬编码任何宿主路径或占位符格式**（对齐 `model_dataurl` +既有设计）。宿主通过 opts 注入 `secret_dataurl` / `secret_format` / `secret_title` / +`secret_tip`；未传 `secret_dataurl` 时钥匙图标根本不渲染 → 其他应用零影响。 + +- `UiText.insertAtCursor(text)`：光标位置插入、有选区则替换选区。两个时序坑照抄 + `handle_enter`/`handle_tab_indent` 既有处理：程序改 `dom_element.value` 不触发 input + 事件（须手动同步 `this.value`）；`focus()` 后浏览器可能重置 selectionStart + （用 `schedule_once` 0.5ms 后重设）。**不 dispatch('changed')**——UiText 也用于 Form， + 多发一次会触发脏值/校验联动。 +- 弹窗 API：`PopupWindow` 无 `close()` 也无 `'closed'` 事件，正确是 `dismiss()` + + `'dismissed'`（popup.js:475/481）。 +- 样式走全局 css class（`.secret-picker-row` 等）——bricks 内联嵌套 style 对象无效。 +- 6 处 AgentIO 入口全部注入：`pipeline-core/wwwroot/agent/index.ui`（静态 + 动态建 tab + 脚本各一处）、`agent_generic/index.ui`、`mobile_agent.ui`（2 处）、 + `pipeline-sdlc/wwwroot/index.ui`、`index_cockpit.ui`。 +- 后端 `pipeline-core/wwwroot/api/secret_options.dspy`：身份取自 `get_user()`/`get_userorgid()`， + **不接受前端传 user_id/org_id**（否则改参数即可枚举他人凭据名称）；双保险剔除 + `encrypted_value`/`fingerprint`；未登录返 `success:false`（区分「没登录」与「没数据」）。 + RBAC 已被 `rp.json` 的 `/pipeline_core/**`(logined) 覆盖，无需额外注册。 + +**性能（实测，本地 20KB 文本）** + +`tokenize` 0.003ms / `detokenize` 0.016ms / 4KB 侦测 0.64ms / 用户消息侦测 0.034ms。 + +**三个挂调实锤的实现陷阱（勿回退，有回归测试）** + +| 陷阱 | 症状 | 修复 | +|---|---|---| +| 子串碰撞 | 短 key `key-Ab3x` 与长 key 共存时按插入序替换 → 长 key 被吃掉前缀、尾部明文裸露进上下文 | `_sorted_secrets` 按值长度**降序**替换 | +| 占位符别名劫持 | secret 的值本身长得像占位符 → 二次 token 化改写既有占位符 | 入库拒绝值内含 `@@sec:`（`_PLACEHOLDER_IN_VALUE`）+ tokenize 幂等 | +| 未知占位符吞空 | `ArgsConvert(default='')` 把缺失变量替换成空串 → 命令语义反转 | `detokenize` 原样保留 + 回报未知名单；执行边界拒绝执行 | + +另两个自产 bug：`_PLACEHOLDER_RE` 用 `{1,62}`（最少 2 字符）与 `_NAME_RE` 长度下界不一致 +→ 单字符名存得进库但识别不出（未知占位符漏报）；`guess_type` 名称关键词优先 → +`openai_style_key` 里的 `key` 把精确类型降级成 `api_key`。均已修 + 加同步断言。 + +**测试**：`/tmp/test_secret_vault.py` 124 断言全绿（11 组：AES 加解密/子串碰撞/别名劫持/ +未知占位符/侦测/名称归一/端到端/性能/DB 层/跨文件工具名同步/越权防护）。 +桩 `CapturingSor` **必须真实模拟 WHERE 过滤**——第一版对任何 SELECT 都返回全部 rows， +无视 SQL 条件，导致「零写操作」断言在过滤正确时反而 FAIL（假阳性）。 +测试桩失真 = 测试证明不了任何事。 + +**部署**：`dist/` 在 bricks 的 .gitignore 里（构建产物不入库）→ 部署必须在测试机 +`cd pkgs/bricks && bash build.sh` 重建，不能靠 git pull（同「bricks 追平必须连 dist 重建」）。 +建表走 m0037 + `create_tables.py`（自动扫描 `pkgs/*/models`，索引按列集合幂等判重）。 diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index c5b33e4..1080a54 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -41,6 +41,14 @@ ROLE_ALIASES = { TASK_REVIEW = 'review' TASK_APPROVED = 'approved' +# 用户敏感信息库工具名（2026-09-17）。必须与 secret_vault.SECRET_HANDLERS 的键一致 +# （有测试 test_secret_vault.py 的 J 段断言同步，改一处必须改两处）。 +# 分发在 _exec_agent_tool；注册在 role_agent_run（只挂只读三件 SECRET_TOOLS_V1_READONLY）。 +_SECRET_TOOL_NAMES_V1 = frozenset(( + 'list_secrets', 'save_secret', 'use_secret', + 'delete_secret', 'set_secret_status', 'detect_secret', +)) + # 项目过程仓库名（repos/ 下）：阶段文档、QC 审计文档、项目管理文档放这里；应用仓库/模块仓库各自独立。 # 实际仓库名 = {项目名}_pc（见 project-directory-spec 规范），此常量仅为查不到项目名时的回退默认值。 PROJECT_REPO_NAME = 'project' @@ -364,18 +372,29 @@ def _build_agent_bwrap_cmd(bwrap: str, cwd: str, writable_root: str, command: st return parts -async def _run_shell(command, workdir, timeout=120, strict=False): +async def _run_shell(command, workdir, timeout=120, strict=False, secret_env=None): """安全执行 shell 命令（优先 bwrap 沙箱）。返回 {"rc","stdout","stderr","sandbox"}。 strict=True（通用会话档）：不挂平台目录只读（/d/pipeline、/d/doit 完全不可见）， 可写根收窄到 cwd 本身（配合 _sandbox_writable_root 的 _general/{uid} 用户级）。 strict=False（产线档，默认，行为不变）：平台目录只读 + 机构级可写。 + + secret_env（2026-09-17 敏感信息库）：{VAR: 明文} 注入子进程环境变量。 + 走 subprocess env= 参数而非 bwrap --setenv——后者把值放进 bwrap 的 argv， + 同机任意用户 `ps`/`/proc//cmdline` 可读（world-readable）；env= 落在 + /proc//environ，仅属主与 root 可读。bwrap 未用 --clearenv，默认继承 + 自身环境，所以 env= 的值能进沙箱内。命令字符串本身不含真值（可安全落库）。 """ cwd = os.path.abspath(workdir) if workdir else WORKSPACE_BASE if not await _is_safe_workdir_async(cwd): return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围", "sandbox": False} if not os.path.isdir(cwd): return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}", "sandbox": False} + # 环境变量注入：基于当前进程环境叠加（只加被引用到的 secret，最小暴露面） + _env = None + if secret_env: + _env = dict(os.environ) + _env.update({str(k): str(v) for k, v in secret_env.items()}) bwrap = _find_bwrap() try: if bwrap: @@ -384,12 +403,13 @@ async def _run_shell(command, workdir, timeout=120, strict=False): include_platform_ro=not strict) # bwrap 用列表直接执行（非 shell 拼接）；命令本身仍由沙箱内 bash -c 解释 proc = await asyncio.create_subprocess_exec( - *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + env=_env) else: # 降级：无 bwrap，保持原有目录隔离（弱隔离），标记 sandbox: False proc = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - cwd=cwd, executable='/bin/bash') + cwd=cwd, executable='/bin/bash', env=_env) try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) except asyncio.TimeoutError: @@ -2181,6 +2201,37 @@ async def _exec_agent_tool(tool, params, workspace_dir, ctx=None): tool, p, org_id=str((ctx or {}).get('org_id', '') or '0'), user_id=str((ctx or {}).get('user_id', '') or ''), project_id=str((ctx or {}).get('project_id', '') or '')) + # 用户敏感信息库（2026-09-17）：角色 agent 无人值守，身份取 ctx.user_id + # （= role_agent_run 解析的项目真人 owner）。身份为空时 handler 自拒—— + # 不给无主凭据开口子。角色 agent 只挂只读三件（list/use/detect，见 + # role_agent_run 的 SECRET_TOOLS_V1_READONLY），但分发层对全部六件放行， + # 写操作靠"工具未注册进 schema"挡住（LLM 看不到就不会调），双保险。 + if tool in _SECRET_TOOL_NAMES_V1: + from . import secret_vault + _suid = str((ctx or {}).get('user_id', '') or '') + if not _suid: + return ("FAIL: 当前任务无法解析项目 owner，不提供敏感信息能力" + "（无人值守且无归属人，防凭据错用）。") + _sh = secret_vault.SECRET_HANDLERS.get(tool) + if not _sh: + return f"FAIL: 未知敏感信息工具 {tool}" + try: + from sqlor.dbpools import DBPools + db = DBPools() + async with db.sqlorContext("pipeline") as _ssor: + _r = await _sh(_ssor, p, { + "user_id": _suid, + "org_id": str((ctx or {}).get('org_id', '') or ''), + "project_id": str((ctx or {}).get('project_id', '') or ''), + }) + try: + await _ssor.sqlExe("COMMIT", {}) + except Exception: + pass + return _r + except Exception as e: + logger.warning("secret tool %s failed: %s", tool, str(e)[:160]) + return f"ERROR: {str(e)[:300]}" if tool == 'read_file': path = p.get('path', '') if not path: return 'FAIL: 需要文件路径' @@ -2229,8 +2280,40 @@ async def _exec_agent_tool(tool, params, workspace_dir, ctx=None): _org = (ctx or {}).get('org_id', '') or '' if _who in ('agent.deploy_test', 'agent.deploy_prod') and _org: cmd = _inject_org_ssh_key(cmd, _org) - r = await _run_shell(cmd, workspace_dir, timeout=120) - return f"rc={r['rc']}\nSTDOUT:\n{r['stdout'][:2000]}\nSTDERR:\n{r['stderr'][:1000]}" + # 敏感信息执行边界（2026-09-17）：占位符 @@sec:NAME@@ → $PIPELINE_SEC_NAME， + # 真值经 env 注入子进程，不进命令字符串（命令可安全落库/进模型上下文）。 + # 未知占位符 → 拒绝执行（绝不替换成空串，空串会静默改变命令语义： + # 实测 `rm -rf ${TARGET_DIR}$/` → `rm -rf /`）。 + _secret_env = None + _suid = str((ctx or {}).get('user_id', '') or '') + if _suid: + try: + from . import secret_vault + from sqlor.dbpools import DBPools + _db = DBPools() + async with _db.sqlorContext("pipeline") as _ssor: + cmd, _secret_env, _unknown = await secret_vault.prepare_command( + _ssor, cmd, org_id=_org, user_id=_suid) + if _unknown: + return ("FAIL: 命令引用了不存在的敏感信息 " + ", ".join(_unknown) + + "（已拒绝执行，防止空值改变命令语义）。用 list_secrets 核对名称。") + except Exception as e: + logger.warning("run_shell secret prepare failed: %s", str(e)[:160]) + r = await _run_shell(cmd, workspace_dir, timeout=120, secret_env=_secret_env) + _out = f"rc={r['rc']}\nSTDOUT:\n{r['stdout'][:2000]}\nSTDERR:\n{r['stderr'][:1000]}" + # 出站净化：命令输出可能回显凭据（curl -v、报错信息常见），擦洗后再回模型 + if _suid: + try: + from . import secret_vault + from sqlor.dbpools import DBPools + async with DBPools().sqlorContext("pipeline") as _ssor: + _out, _scrub = await secret_vault.sanitize_tool_output( + _ssor, _out, org_id=_org, user_id=_suid) + if _scrub: + _out += "\n（输出中的敏感信息已替换为占位符：" + ", ".join(_scrub) + "）" + except Exception as e: + logger.warning("run_shell output sanitize failed: %s", str(e)[:160]) + return _out elif tool == 'git_clone': url = p.get('repo_url', '') if not url: return 'FAIL: 需要仓库URL' @@ -2931,7 +3014,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): # （本机构+平台owner机构，生成类任务按 task 自动选型）。唯一实现在 # platform_model_tools，分发在 _exec_agent_tool。 from .platform_model_tools import PLATFORM_MODEL_TOOLS_V1 - all_tools = AGENT_TOOLS + capability_tools + PLATFORM_MODEL_TOOLS_V1 + # 敏感信息库（2026-09-17）：角色 agent 只挂只读三件（list/use/detect）—— + # 无人值守场景下 LLM 自主增删凭据风险高于收益；写操作留给有真人身份的会话 agent。 + from .secret_vault import SECRET_TOOLS_V1_READONLY + all_tools = AGENT_TOOLS + capability_tools + PLATFORM_MODEL_TOOLS_V1 + SECRET_TOOLS_V1_READONLY # ── deliverable_type 守卫（2026-09-14 pbls）：角色声明了合法交付类型时， # ① 工具描述动态改写为只列合法值（源头防照抄示例）② deliver 拦截校验（硬门禁）。 @@ -2965,6 +3051,17 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): _iter = await get_current_iteration(sor, project_id) except Exception: _iter = None + # 敏感信息库归属：解析项目真人 owner（角色 agent 无自身 user_id）。 + # 复用 rag_client.resolve_project_owner（已处理 created_by='agent.pm' 的自动立项 + # 场景——反查创建会话的真人）。解析失败留空串，handler 侧拒绝，不硬塞。 + _secret_owner = "" + try: + from .rag_client import resolve_project_owner + _secret_owner, _owner_err = await resolve_project_owner(project_id) + if not _secret_owner and _owner_err: + logger.info("role_agent secret owner unresolved: %s", _owner_err) + except Exception as e: + logger.warning("resolve secret owner failed: %s", str(e)[:120]) capability_ctx = { "project_id": project_id, "iteration_id": _iter.get('id', '') if _iter else '', @@ -2972,6 +3069,10 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): "agent_id": agent_id, "task_id": task_id, "org_id": org_id or '0', + # 敏感信息库归属（2026-09-17）：角色 agent 无人值守、无自身 user_id， + # 凭据身份必须取项目真人 owner——否则要么拿不到用户凭据，要么错用他人凭据。 + # 解析失败留空串：secret_vault 的 handler 会拒绝（不给无主凭据开口子）。 + "user_id": _secret_owner, } system = (AGENT_SYSTEM_PROMPT @@ -2983,6 +3084,14 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): .replace('__ROLE_SPECIFIC__', role_specific) .replace('__ROLE_SKILLS__', role_skills) .replace('__TOOLS__', tools_text)) + # 敏感信息使用铁律（2026-09-17）：仅在解析到项目 owner 时注入—— + # 无归属人时 secret 工具会自拒，也就不给凭据使用引导（避免 LLM 反复试错烧轮次）。 + if _secret_owner: + try: + from .secret_vault import SECRET_PROMPT_BLOCK + system += SECRET_PROMPT_BLOCK + except Exception as e: + logger.warning("inject secret prompt block (role agent) failed: %s", str(e)[:120]) msgs = [{"role": "system", "content": system}] msgs.append({"role": "user", "content": f"执行任务：{title}\n参数：{params_str}"}) diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 14940f5..d4a112f 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -61,6 +61,14 @@ _PROJECT_TOOL_NAMES = { # 名单保留机制本身，未来若有无法沙箱化的工具仍可加入。 _GENERIC_DENIED_TOOLS = set() +# 用户敏感信息库工具名（2026-09-17）：与 pipeline_core.agent_config.GENERAL_TOOLS 里 +# category="secret" 的工具一一对应，与 secret_vault.SECRET_HANDLERS 的键一致。 +# 三处同步：工具定义(core) + handlers 映射(本文件 _dispatch_sdlc_tool) + handler 实现(secret_vault)。 +_SECRET_TOOL_NAMES = ( + "list_secrets", "save_secret", "use_secret", + "delete_secret", "set_secret_status", "detect_secret", +) + # ── 默认工具定义（在 pipeline-core 未加载时使用）── @@ -855,6 +863,15 @@ class AgentExecutor: prompt = prompt.replace("{project_name}", proj_name) prompt = prompt.replace("{tools_description}", tools_text) + # 敏感信息使用铁律（2026-09-17）：仅在有真实用户身份时注入—— + # 无人值守/内部 agent 不挂 secret 工具，也就不给凭据使用引导。 + if self.user_id: + try: + from .secret_vault import SECRET_PROMPT_BLOCK + prompt += SECRET_PROMPT_BLOCK + except Exception as e: + logger.warning(f"inject secret prompt block failed: {e}") + return prompt # ═══════════════════════════════════════════════════════ @@ -1221,6 +1238,11 @@ class AgentExecutor: "fetch_url": self._t_fetch_url, # ── 项目数据只读查询（2026-09-08，白名单+强制项目过滤在 db_query）── "query_project_data": self._t_query_project_data, + # ── 用户敏感信息库（2026-09-17）：签名适配 (sor,params,ctx) ← (sor,p,pid) ── + # ctx 用 _build_ctx()（含真实 self.user_id/self.org_id）——secret 归属必须 + # 是真实身份，不能用作用域解析里被 generic 置空的那份。 + **{n: (lambda _n: (lambda s, _p, _pid: self._t_secret_tool(s, _n, _p)))(n) + for n in _SECRET_TOOL_NAMES}, } handler = handlers.get(tool_name) @@ -1228,6 +1250,34 @@ class AgentExecutor: return await handler(sor, p, pid) return None + async def _t_secret_tool(self, sor, tool_name: str, params: dict) -> str: + """敏感信息工具统一入口（2026-09-17）。 + + 身份强制来自会话真实上下文（self.user_id / self.org_id），忽略 LLM 传的任何 + 归属参数——防越权读写他人 secret。无 user_id（无人值守/内部 agent）一律拒绝。 + """ + from . import secret_vault + if not self.user_id: + return "FAIL: 当前会话没有用户身份，不提供敏感信息能力（防无人值守场景泄露凭据）" + h = secret_vault.SECRET_HANDLERS.get(tool_name) + if not h: + return f"FAIL: 未知敏感信息工具 {tool_name}" + try: + _r = await h(sor, params or {}, {"user_id": self.user_id, + "org_id": self.org_id or "", + "project_id": self.project_id}) + # 写操作必须显式 COMMIT（多 worker 架构：不提交则其他进程读不到刚存的凭据）。 + # 与本文件其他写库工具（_persist_project 等 6 处）同款做法；只读工具多跑一次 + # COMMIT 无害（sqlor 事务幂等），省去按工具名区分读写。 + try: + await sor.sqlExe("COMMIT", {}) + except Exception as _ce: + logger.warning(f"secret tool {tool_name} commit failed: {_ce}") + return _r + except Exception as e: + logger.error(f"secret tool {tool_name} error: {e}") + return f"ERROR: {str(e)[:300]}" + # ── 工具实现 ── async def _persist_project(self, sor, pid): @@ -1544,6 +1594,21 @@ class AgentExecutor: return ("FAIL: 通用会话的命令执行需要 bwrap 沙箱（当前服务器不可用），已拒绝执行。" "文件类操作请改用 read_file/write_file/list_files/search_files。") + # 敏感信息执行边界（2026-09-17）：占位符 @@sec:NAME@@ → $PIPELINE_SEC_NAME， + # 真值经 env 注入子进程，**不进命令字符串**（命令可安全落库/进上下文）。 + # 未知占位符原样保留（绝不替换成空串——空串会静默改变命令语义）。 + # 必须在前台/后台分支**之前**处理，两条执行路径才都受保护。 + secret_env = None + try: + from . import secret_vault + cmd, secret_env, unknown = await secret_vault.prepare_command( + sor, cmd, org_id=self.org_id or "", user_id=self.user_id or "") + if unknown: + return ("FAIL: 命令引用了不存在的敏感信息 " + ", ".join(unknown) + + "（已拒绝执行，防止空值改变命令语义）。用 list_secrets 核对名称。") + except Exception as e: + logger.warning(f"secret prepare_command failed (run without secrets): {e}") + # 后台执行（2026-09-10，对齐 Hermes terminal background）： # 状态文件化到 workspace/.bg/，跨 worker 进程可 poll； # 沙箱档位与前台完全一致（generic 强制 strict，无 bwrap 上面已拒）。 @@ -1552,7 +1617,8 @@ class AgentExecutor: from . import bg_jobs try: job_id = await bg_jobs.start_bg_job( - cmd, self.workspace_dir, strict=self.generic) + cmd, self.workspace_dir, strict=self.generic, + secret_env=secret_env) except bg_jobs.BgJobError as e: return f"FAIL: {e}" return (f"OK: 后台任务已启动 job_id={job_id}。" @@ -1563,17 +1629,46 @@ class AgentExecutor: except (TypeError, ValueError): timeout = 60 timeout = max(5, min(timeout, 300)) - r = await _run_shell(cmd, self.workspace_dir, timeout=timeout, strict=self.generic) + + r = await _run_shell(cmd, self.workspace_dir, timeout=timeout, + strict=self.generic, secret_env=secret_env) out = f"rc={r['rc']}\n{r['stdout'][:2000]}" if r.get('stderr'): out += f"\nSTDERR: {r['stderr'][:500]}" if self.generic and not r.get('sandbox'): out += "\n（注意：本次未经过沙箱）" + # 出站净化（需求2的「执行返回当输入处理」）：工具输出里若回显了 + # 已入库凭据明文（报错信息常见），擦洗成占位符再回填模型上下文。 + try: + from . import secret_vault + out, scrubbed = await secret_vault.sanitize_tool_output( + sor, out, org_id=self.org_id or "", user_id=self.user_id or "") + if scrubbed: + out += "\n（输出中出现的敏感信息已替换为占位符：" + ", ".join(scrubbed) + "）" + except Exception as e: + logger.warning(f"sanitize_tool_output failed: {e}") return out except Exception as e: return f"ERROR: {str(e)[:300]}" async def _t_process(self, sor, p, pid): + """后台任务管理入口：先执行，再对输出做出站净化（2026-09-17）。 + + 后台命令可能把凭据回显进 output.log（curl -v、报错信息常见）， + 读回模型上下文前擦洗成占位符——与前台 run_command 同一条防线。 + """ + out = await self._t_process_raw(sor, p, pid) + try: + from . import secret_vault + out, scrubbed = await secret_vault.sanitize_tool_output( + sor, out, org_id=self.org_id or "", user_id=self.user_id or "") + if scrubbed: + out += "\n（输出中出现的敏感信息已替换为占位符：" + ", ".join(scrubbed) + "）" + except Exception as e: + logger.warning(f"sanitize process output failed: {e}") + return out + + async def _t_process_raw(self, sor, p, pid): """后台任务管理（poll/log/wait/kill，配套 run_command background=true）。 隔离：只解析 self.workspace_dir 下的 .bg/——通用会话 workspace 是 diff --git a/pipeline_service/bg_jobs.py b/pipeline_service/bg_jobs.py index a1c8ece..42364c2 100644 --- a/pipeline_service/bg_jobs.py +++ b/pipeline_service/bg_jobs.py @@ -27,6 +27,7 @@ import os import re import signal import time +from typing import Optional logger = logging.getLogger("pipeline.bg_jobs") @@ -85,9 +86,28 @@ def _pid_alive(pid: int) -> bool: return False +def _bg_env(secret_env: Optional[dict]) -> Optional[dict]: + """构造子进程环境：无 secret 时返回 None（继承默认行为，零改动风险）。 + + 有 secret 时基于 os.environ 叠加，只加被引用到的那几条（最小暴露面）。 + """ + if not secret_env: + return None + env = dict(os.environ) + env.update({str(k): str(v) for k, v in secret_env.items()}) + return env + + async def start_bg_job(command: str, workspace_dir: str, strict: bool = False, - timeout: int = MAX_BG_SECONDS) -> str: - """启动后台命令，返回 job_id。沙箱/目录安全校验与前台 _run_shell 完全一致。""" + timeout: int = MAX_BG_SECONDS, + secret_env: Optional[dict] = None) -> str: + """启动后台命令，返回 job_id。沙箱/目录安全校验与前台 _run_shell 完全一致。 + + secret_env（2026-09-17）：{VAR: 明文} 注入子进程环境（同 _run_shell 的口径： + 走 env= 不走 bwrap --setenv，避免值出现在 argv 被同机用户 ps 读到）。 + ⚠️ meta.json 存的是 command 原文（不含真值），output.log 是命令输出—— + 输出可能回显凭据，读取时经 sanitize_tool_output 擦洗（process 工具侧）。 + """ from .agent_loop import (_find_bwrap, _build_agent_bwrap_cmd, _sandbox_writable_root, _is_safe_workdir_async) @@ -124,6 +144,7 @@ async def start_bg_job(command: str, workspace_dir: str, strict: bool = False, *cmd, stdout=log_f, stderr=asyncio.subprocess.STDOUT, cwd=None if bwrap else cwd, start_new_session=True, # 独立进程组，kill 时整组终止 + env=_bg_env(secret_env), ) finally: log_f.close() diff --git a/pipeline_service/gateway.py b/pipeline_service/gateway.py index b4598df..3796a6c 100644 --- a/pipeline_service/gateway.py +++ b/pipeline_service/gateway.py @@ -160,6 +160,18 @@ class Gateway: }, ensure_ascii=False) + "\n" return + # 0.5 用户敏感信息入站门禁（2026-09-17）：自动侦测明文凭据 → 密文入库 → + # 原文替换为占位符。放在所有上下文解析之前，因为**净化后的 content 才是 + # 该进模型上下文与落库的版本**（明文一旦进 executor 就会每轮上行供应商）。 + # 失败绝不阻断会话（降级为原文继续，安全增强不牺牲可用性）。 + try: + content, secret_notes = await self._secret_inbound_gate(user_id, content) + except Exception as e: + logger.warning(f"secret inbound gate failed (continue with raw content): {e}") + content, secret_notes = content, [] + for note in secret_notes: + yield json.dumps({"type": "progress", "message": note + "\n"}, ensure_ascii=False) + "\n" + # 1. 解析项目上下文（纯通用模式跳过，不挂任何产线插件） if generic: ctx = {"pid": "", "pipeline_id": "", "name": "", "project_model_name": ""} @@ -231,6 +243,42 @@ class Gateway: image_parts=image_parts or None): yield chunk + async def _secret_inbound_gate(self, user_id: str, content: str): + """用户敏感信息入站门禁：侦测明文凭据 → 密文入库 → 替换为占位符。 + + 返回 (净化后的 content, 动作说明列表)。动作说明以 progress 事件回显给用户 + ——「体感不变但安全增强」：用户照常打字，只多看到一条「已自动保存」的提示。 + + 安全边界： + - 无 user_id（无人值守/内部调用）→ 不做任何处理，原文返回（不给无主凭据建库）。 + - 只处理**高置信度**候选（已知凭据前缀，或关键词+熵≥3.5）；中低置信度不自动入库， + 避免把哈希/base64 数据块当凭据吞掉（误报会静默毁数据，比漏报更糟）。 + - 任何异常都降级为原文继续，绝不阻断会话。 + """ + if not user_id or not content: + return content, [] + from . import secret_vault + from sqlor.dbpools import DBPools + + db = DBPools() + async with db.sqlorContext("pipeline") as sor: + org_id = "" + try: + recs = await sor.sqlExe( + "SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": user_id}) + if recs: + org_id = getattr(recs[0], "orgid", "") or "" + except Exception as e: + logger.warning(f"secret gate: resolve org_id failed: {e}") + r = await secret_vault.scan_and_capture( + sor, content, org_id=org_id, user_id=user_id, + who=user_id, auto_store=True) + try: + await sor.sqlExe("COMMIT", {}) + except Exception: + pass + return r.get("text") or content, list(r.get("actions") or []) + async def _resolve_and_persist_model(self, user_id: str, project_id: str, model_id: str) -> str: """校验前端选中的模型并持久化到项目。返回 llm.name（空 = 无效/未持久化，调用方走回退链）。 diff --git a/pipeline_service/secret_vault.py b/pipeline_service/secret_vault.py new file mode 100644 index 0000000..c627152 --- /dev/null +++ b/pipeline_service/secret_vault.py @@ -0,0 +1,1017 @@ +"""用户敏感信息管理（secret vault）—— 敏感信息入库 + 会话占位符注入 + 入站自动侦测。 + +设计目标（2026-09-17 用户需求）： + 1) 敏感信息入库：RC4 密文存 pipeline_user_secrets，按 org_id/user_id 隔离。 + 2) 会话中可"从表里选一个敏感信息插入会话"：agent 只见元数据（名称/类型/前缀/长度）， + 插入的是占位符 `@@sec:NAME@@`，真值永不进模型上下文。 + 3) 入站自动侦测：用户消息/工具输出里出现明文敏感信息 → 自动入库 + 原文替换为占位符。 + +安全不变量（写死在代码，不靠 LLM 自觉）： + - **真值只在执行边界以「子进程环境变量」形式出现**，绝不拼进命令字符串 + （拼进命令 = 进 tool_calls 落库 = 进模型上下文 = 每轮上行供应商）。 + - 占位符语法 `@@sec:NAME@@`，刻意避开 `${X}$`（sqlor 参数化 / ahserver ArgsConvert + 都用它，复用会被误展开）与 `$VAR`（bash 会就地展开，绕过我们的边界）。 + - 任何对外返回/日志/审计都只带元数据，不带 encrypted_value 与明文。 + - 未知占位符**原样保留**，绝不替换成空串（空串会静默改变命令语义， + 实测 `rm -rf ${TARGET_DIR}$/` → `rm -rf /`）。 + +三个已修复的实现陷阱（挂调实锤，勿回退）： + A. 子串碰撞：长短两个 secret 共存时按插入序替换 → 长值被短值吃掉前缀、尾部明文裸露。 + 修复：替换前按 value 长度**降序**排序（_sorted_secrets）。 + B. 占位符别名劫持：secret 的值本身长得像占位符 → 二次 token 化改写既有占位符。 + 修复：入库拒绝值内含占位符语法（_PLACEHOLDER_IN_VALUE）；tokenize 幂等（已是占位符不重扫）。 + C. 未知占位符吞空：见上，detokenize 一律原样保留。 +""" + +import hashlib +import logging +import math +import os +import re +from collections import Counter +from typing import Dict, List, Optional, Tuple + +logger = logging.getLogger("pipeline.secret_vault") + +# ══════════════════════ 常量 ══════════════════════ + +TABLE = "pipeline_user_secrets" + +#: 占位符语法（模型可见、可回放；执行边界换成环境变量引用） +PLACEHOLDER_PREFIX = "@@sec:" +PLACEHOLDER_SUFFIX = "@@" + +#: 变量名白名单：只认大写字母开头的标识符（堵 eval/注入面、堵 shell 元字符）。 +#: ⚠️ _PLACEHOLDER_RE 必须与本正则的**长度下界一致**（都是 1 字符起）—— +#: 曾写成 {1,62}（即最少 2 字符）导致单字符名能入库、但 find_placeholders/ +#: detokenize 识别不出 → 未知占位符漏报。改名字规则时两处同步（挂调实锤）。 +_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,62}$") +_PLACEHOLDER_RE = re.compile(r"@@sec:([A-Z][A-Z0-9_]{0,62})@@") + +#: 执行边界注入子进程的环境变量前缀（agent 写 $PIPELINE_SEC_GITHUB_TOKEN） +ENV_PREFIX = "PIPELINE_SEC_" + +#: 占位符在值里出现即拒绝入库（防陷阱 B） +_PLACEHOLDER_IN_VALUE = re.compile(r"@@sec:") + +#: 已知凭据前缀（高置信度，直接判定为敏感；与 Hermes agent/redact.py 同源思路） +KNOWN_PREFIXES = [ + (r"sk-[A-Za-z0-9_\-]{10,}", "openai_style_key"), + (r"github_pat_[A-Za-z0-9_]{10,}", "github_pat"), + (r"ghp_[A-Za-z0-9]{10,}", "github_token"), + (r"gh[osur]_[A-Za-z0-9]{10,}", "github_oauth"), + (r"glpat-[A-Za-z0-9_\-]{10,}", "gitlab_pat"), + (r"xox[baprs]-[A-Za-z0-9\-]{10,}", "slack_token"), + (r"AIza[A-Za-z0-9_\-]{30,}", "google_api_key"), + (r"AKIA[A-Z0-9]{16}", "aws_access_key"), + (r"sk_live_[A-Za-z0-9]{10,}", "stripe_live_key"), + (r"sk_test_[A-Za-z0-9]{10,}", "stripe_test_key"), + (r"hf_[A-Za-z0-9]{10,}", "huggingface_token"), + (r"npm_[A-Za-z0-9]{10,}", "npm_token"), + (r"pypi-[A-Za-z0-9_\-]{10,}", "pypi_token"), + (r"plk-[A-Za-z0-9]{8,}", "platform_short_token"), + (r"rak-[A-Za-z0-9]{8,}", "platform_api_key"), + (r"-----BEGIN[A-Z ]*PRIVATE KEY-----", "private_key"), +] +_KNOWN_RES = [(re.compile(p), t) for p, t in KNOWN_PREFIXES] + +#: 关键词邻域形态：`apikey: xxx` / `api_key=xxx` / `token：xxx` / `Authorization: Bearer ***` +#: 中英文冒号都认（实测 ASCII `=` 能命中而中文 `：` 漏网是 Hermes 侧的真实缺口，这里补齐） +_KEYWORD_RE = re.compile( + r"(api[_\-\s]?key|apikey|access[_\-\s]?token|secret[_\-\s]?key|auth[_\-\s]?token" + r"|bearer|passwd|password|private[_\-\s]?key|credential)" + r"\s*[:：=]?\s*[\"']?([A-Za-z0-9_\-.+/]{12,128})[\"']?", + re.IGNORECASE, +) + +#: 低熵/示例值黑名单（防把 `password: 123456`、`sk-xxx` 占位符当凭据吞掉） +_JUNK_VALUE_RE = re.compile( + r"^(?:x{3,}|\*{3,}|\.{3}|xxx+|your[_\-]?[\w]*|example[\w]*|placeholder|dummy|test[\w]*" + r"|none|null|true|false|\d{1,8}|[a-z]{1,12}|[\w]*\[[\w]*\]?)$", + re.IGNORECASE, +) + +#: 自动侦测的熵门限：低于此值不自动入库（宁可漏也不误吞日常文本） +ENTROPY_THRESHOLD = 3.0 +#: 自动入库的最短长度 +MIN_AUTO_LEN = 12 +#: 类型猜测用 +_TYPE_GUESS = [ + ("github", "github_token"), ("gitlab", "gitlab_token"), ("slack", "slack_token"), + ("aws", "aws_key"), ("stripe", "stripe_key"), ("openai", "openai_key"), + ("dashscope", "dashscope_key"), ("aliyun", "aliyun_key"), ("db", "db_password"), + ("mysql", "db_password"), ("redis", "db_password"), ("ssh", "ssh_key"), + ("token", "token"), ("key", "api_key"), ("secret", "secret"), + ("pass", "password"), ("pwd", "password"), +] + + +# ══════════════════════ 加解密（AES，2026-09-17 用户指定替换 RC4） ══════════════ +# +# 为什么不用 ahserver 的 password_encode/password_decode：实测 globalEnv.py:61 +# `from appPublic.rc4 import password, unpassword` —— 它们就是 RC4 的薄封装， +# 换过去等于没换。真正的 AES 在 appPublic.aes（aes_encode_b64 / aes_decode_b64， +# AES-ECB + PKCS7 + base64），与 DB 连接密码（config.json databases.*.password）同套。 +# +# ⚠️ aes.py 内部有**两处方向相反**的 iso-8859-1，只有一处需要包装（2026-09-17 实证）： +# ① 密文 bytes → str（aes.py:42-43, 46-47）：**必须** iso-8859-1，不能换 utf-8。 +# 实测 200 组 AES 密文用 utf-8 解码失败 200 组（100%）——密文是二进制， +# 不符合 UTF-8 编码规范（首字节 0xef 即 invalid continuation byte）。 +# 这处是历史正确设计（曾用 utf-8 撞编解码错误后改的），**不要动**。 +# ② 明文 str → bytes（aes.py:29, 37）：iso-8859-1 表示不了非 ASCII， +# 实测「密码是中文Abc123」「🔑key123」直接 UnicodeEncodeError。 +# 凭据含非 ASCII 很常见（中文口令、含中文的备注型凭据），所以在**调用方**包装： +# str → UTF-8 bytes → 按 latin-1 逐字节还原成等长 str 传入（latin-1 是字节保真 +# 映射，0x00-0xFF 一一对应），aes.py:29 再 encode('iso-8859-1') 即还原成 +# 原始 UTF-8 字节流 = 字节保真空操作；出库反向还原。 +# 不改 aes.py（appPublic 是多宿主共享基础模块，改它影响 DB 密码等全部调用方）。 +# 包装后实测 9/9 往返一致：ASCII / GitHub PAT / 中文 / emoji / 中英混合+符号 / +# 私钥含换行 / 512 长值 / 特殊字符 p@$$w0rd!#%^&*() / 含 @@sec: 语法的值。 +# +# 密文加版本前缀 AES1$ —— 因为 base64 密文没有固定形态特征，无法像 RC4 的 +# QUZVcX 那样靠"前缀嗅探"判断是否已加密；显式版本标记同时为将来算法迁移留路。 + +#: 密文版本前缀（AES-ECB + UTF-8 包装 + base64） +_AES_PREFIX = "AES1$" +#: 历史 RC4 密文前缀（本平台 password() 产出；用于向后兼容读取，不再写入） +_RC4_PREFIX = "QUZVcX" + + +def _password_key() -> str: + """宿主 password_key（pipeline-app 默认值兜底）。""" + try: + from appPublic.jsonConfig import getConfig + return getConfig().password_key or "QRIVSRHrthhwyjy176556332" + except Exception: + return "QRIVSRHrthhwyjy176556332" + + +def encrypt_secret(plain: str) -> str: + """AES 加密 + UTF-8 包装，产出带版本前缀的密文。 + + 幂等保护：已是密文形态（AES1$ 或历史 RC4 QUZVcX）则原样返回，防多层叠加—— + 多层加密是历史真实事故（llm.api_key 叠 2~3 层 → 解出来还是密文 → 上游 401）。 + """ + if not plain: + return "" + if looks_encrypted(plain): + return plain + from appPublic.aes import aes_encode_b64 + body = aes_encode_b64(_password_key(), plain.encode("utf-8").decode("iso-8859-1")) + return _AES_PREFIX + body + + +def decrypt_secret(enc: str) -> str: + """解密。按前缀分派：AES1$ → AES；QUZVcX → 历史 RC4 兜底；其余 → 原文返回。 + + 失败返回原文（不抛异常）：调用方拿到不可用值会自然失败并报错， + 比抛异常中断整条会话链更可控。 + """ + if not enc: + return "" + key = _password_key() + try: + if enc.startswith(_AES_PREFIX): + from appPublic.aes import aes_decode_b64 + raw = aes_decode_b64(key, enc[len(_AES_PREFIX):]) + return raw.encode("iso-8859-1").decode("utf-8") + if enc.startswith(_RC4_PREFIX): + # 历史 RC4 密文（本模块早期版本或 llm 表迁移过来的值） + from appPublic.rc4 import unpassword + return unpassword(enc, key) + return enc + except Exception as e: + logger.warning("decrypt_secret failed: %s", str(e)[:120]) + return enc + + +def looks_encrypted(value: str) -> bool: + """是否已是密文形态（AES1$ 版本前缀，或历史 RC4 的 QUZVcX 前缀）。""" + if not value: + return False + return value.startswith(_AES_PREFIX) or value.startswith(_RC4_PREFIX) + + +# ══════════════════════ 纯函数：指纹 / 熵 / 类型 ══════════════════════ + +def fingerprint(plain: str) -> str: + """明文 sha256（用于精确去重 + 入站精确匹配，比正则可靠）。""" + return hashlib.sha256((plain or "").encode("utf-8", "replace")).hexdigest() + + +def shannon_entropy(s: str) -> float: + if not s: + return 0.0 + c = Counter(s) + n = len(s) + return -sum((v / n) * math.log2(v / n) for v in c.values()) + + +def guess_type(name_hint: str = "", value: str = "") -> str: + """按值形态/名称猜类型（只做展示分类，不做安全判定）。 + + **值形态优先于名称关键词**：值的前缀是硬证据（`sk-proj-…` 就是 OpenAI 形态）， + 名称提示是软线索。反序会让精确类型被泛化词覆盖——实测 name_hint + `openai_style_key` 里的 `key` 命中关键词表，把类型降级成 `api_key`（挂调实锤）。 + """ + v = value or "" + for rx, t in _KNOWN_RES: + if rx.search(v): + return t + hay = ((name_hint or "") + " " + v[:24]).lower() + for kw, t in _TYPE_GUESS: + if kw in hay: + return t + return "other" + + +def mask_preview(plain: str) -> Tuple[str, int]: + """返回 (前缀提示, 长度)。**只取前 4 字符**——够人类辨认是哪个凭据， + 不足以被拼接复用（实测 GitHub PAT 前缀 `gith`/`ghp_` 是公开固定值，零熵）。""" + if not plain: + return "", 0 + return plain[:4], len(plain) + + +def normalize_name(raw: str) -> str: + """变量名归一：转大写、非标识符字符转下划线、限长。不合法返回空串。""" + n = re.sub(r"[^A-Za-z0-9_]", "_", (raw or "").strip()).upper() + n = re.sub(r"_+", "_", n).strip("_")[:63] + if not n: + return "" + if not n[0].isalpha(): + n = "S_" + n + return n if _NAME_RE.match(n) else "" + + +def make_placeholder(name: str) -> str: + return PLACEHOLDER_PREFIX + name + PLACEHOLDER_SUFFIX + + +# ══════════════════════ 侦测（纯函数，无 DB） ══════════════════════ + +def detect_candidates(text: str) -> List[Dict]: + """扫描文本，返回疑似敏感信息候选（去重，按置信度排序）。 + + 三条通道： + ① 已知凭据前缀（高置信度，无视熵） + ② 关键词邻域 + 熵门限（`apikey: xxx`、`Authorization: Bearer xxx`） + ③ 不做无上下文裸串侦测（误报会静默毁数据，比漏更糟） + + 返回项：{value, name_suggest, secret_type, confidence, source, entropy} + """ + if not text or not isinstance(text, str): + return [] + out: Dict[str, Dict] = {} + + def _add(value: str, conf: str, source: str, name_hint: str = ""): + value = (value or "").strip().strip("\"'`,;") + if len(value) < MIN_AUTO_LEN or len(value) > 256: + return + if _JUNK_VALUE_RE.match(value): + return + ent = shannon_entropy(value) + # 已知前缀是强证据，熵门限只对关键词通道生效 + if source != "known_prefix" and ent < ENTROPY_THRESHOLD: + return + fp = fingerprint(value) + if fp in out: + # 同一值多通道命中 → 提升置信度 + if conf == "high": + out[fp]["confidence"] = "high" + return + out[fp] = { + "value": value, + "fingerprint": fp, + "entropy": round(ent, 2), + "confidence": conf, + "source": source, + "name_suggest": normalize_name(name_hint) or _suggest_name(value, name_hint), + "secret_type": guess_type(name_hint, value), + } + + # ① 已知前缀 + for rx, t in _KNOWN_RES: + for m in rx.finditer(text): + v = m.group(0) + # 私钥块整段不入 value（太长），只标记类型 + if v.startswith("-----BEGIN"): + continue + _add(v, "high", "known_prefix", t) + + # ② 关键词邻域 + for m in _KEYWORD_RE.finditer(text): + kw, val = m.group(1), m.group(2) + if not val: + continue + _add(val, "high" if shannon_entropy(val) >= 3.5 else "medium", "keyword", kw) + + ranked = sorted(out.values(), key=lambda d: (d["confidence"] != "high", -d["entropy"])) + return ranked + + +def _suggest_name(value: str, hint: str = "") -> str: + """从关键词/值形态派生变量名（确定性，同名不同值时调用方负责加序号）。""" + base = normalize_name(hint) if hint else "" + if not base: + base = normalize_name(guess_type("", value)) + if not base or base == "OTHER": + base = "SECRET" + if not base.startswith("SEC_"): + base = "SEC_" + base + return base[:63] + + +# ══════════════════════ 占位符替换（tokenize / detokenize） ══════════════════════ + +def _sorted_secrets(secrets: Dict[str, str]) -> List[Tuple[str, str]]: + """陷阱 A 修复：按值长度降序，长值先替换，防短值是长值子串时吃掉前缀。""" + return sorted(((n, v) for n, v in secrets.items() if v), key=lambda kv: -len(kv[1])) + + +def tokenize(text: str, secrets: Dict[str, str]) -> Tuple[str, List[str]]: + """把文本里出现的**已知 secret 明文**替换成占位符。 + + Args: + text: 待净化文本（用户消息 / 工具输出） + secrets: {NAME: 明文值}（只应传当前用户可见的 secret） + + Returns: + (净化后文本, 命中的 NAME 列表) + + 幂等：已是占位符的片段不含明文，自然不会被再处理（陷阱 B 的一半）。 + """ + if not text: + return text, [] + hits: List[str] = [] + for name, value in _sorted_secrets(secrets): + if value in text: + text = text.replace(value, make_placeholder(name)) + hits.append(name) + return text, hits + + +def detokenize(text: str, secrets: Dict[str, str], *, mode: str = "env") -> Tuple[str, List[str], List[str]]: + """把占位符还原成**可执行形态**。 + + mode="env"（默认，推荐）：占位符 → `$PIPELINE_SEC_`（shell 变量引用）， + 真值不进字符串，由调用方通过环境变量注入子进程。 + mode="value"（仅限非 shell 场景，如 HTTP header 构造）：占位符 → 明文值。 + ⚠️ 用这个模式的调用方必须保证结果不落库、不进模型上下文、不打日志。 + + Returns: + (替换后文本, 命中的 NAME 列表, 未知占位符列表) + + 陷阱 C 修复：未知占位符**原样保留**（返回在第三个元素里让调用方决定报错）， + 绝不替换成空串。 + """ + if not text: + return text, [], [] + hits: List[str] = [] + unknown: List[str] = [] + + def _repl(m): + name = m.group(1) + if name not in secrets: + unknown.append(name) + return m.group(0) # 原样保留 + hits.append(name) + if mode == "value": + return secrets[name] + return "$" + ENV_PREFIX + name + + return _PLACEHOLDER_RE.sub(_repl, text), hits, unknown + + +def find_placeholders(text: str) -> List[str]: + """列出文本中引用的占位符名（未知与否都列，供权限校验）。""" + return [m.group(1) for m in _PLACEHOLDER_RE.finditer(text or "")] + + +def resolve_env(secrets: Dict[str, str], names: List[str]) -> Dict[str, str]: + """为子进程构造环境变量字典（只含被引用到的 secret，最小暴露面）。""" + env: Dict[str, str] = {} + for n in names: + if n in secrets: + env[ENV_PREFIX + n] = secrets[n] + return env + + +# ══════════════════════ DB 层 ══════════════════════ + +def _get_db(): + from sqlor.dbpools import DBPools + db = DBPools() + if not db.databases: + from appPublic.jsonConfig import getConfig + cfg = getConfig() + if cfg.databases: + db.databases = cfg.databases + return db + + +_META_COLS = ("id, name, label, secret_type, fingerprint, prefix_hint, length_hint, " + "org_id, user_id, source, status, remark, use_count, last_used_at, " + "created_at, updated_at") + + +async def list_secrets(sor, org_id: str = "", user_id: str = "", + only_active: bool = True) -> List[Dict]: + """列出可见 secret 的**元数据**（绝不返回 encrypted_value / 明文）。 + + 可见范围：本人（user_id 归属）+ 本机构共享（org_id 归属且 user_id 为空的机构级条目）。 + """ + sql = ("SELECT " + _META_COLS + " FROM " + TABLE + + " WHERE (user_id=${u}$ OR (org_id=${o}$ AND user_id='')) ") + if only_active: + sql += "AND status='active' " + sql += "ORDER BY created_at DESC LIMIT 200" + recs = await sor.sqlExe(sql, {"u": user_id or "", "o": org_id or ""}) + return [_rec_to_meta(r) for r in (recs or [])] + + +def _rec_to_meta(r) -> Dict: + d = dict(r) + d.pop("encrypted_value", None) # 双保险：即使 SELECT 误带也不外泄 + return d + + +def _assert_own(row: Optional[Dict], user_id: str) -> bool: + """写操作前的归属硬门禁（纵深防御，2026-09-17）。 + + get_secret_row(own_only=True) 的 SQL WHERE 已按 user_id 精确过滤，正常情况下 + 机构共享行（user_id=''）与他人行根本查不出来。但**不依赖单一防线**： + 将来若有人改 SQL、加新查询路径、或复用 own_only=False 的读路径去写， + 这层代码级校验仍能挡住越权 UPDATE/DELETE（不可逆操作，符合「入口硬门禁」铁律）。 + + 返回 True = 该行确属调用者本人，允许写。 + """ + if not row: + return False + owner = str(row.get("user_id") or "") + return bool(user_id) and owner == str(user_id) + + +async def get_secret_row(sor, secret_id: str = "", name: str = "", + org_id: str = "", user_id: str = "", + own_only: bool = False) -> Optional[Dict]: + """按 id 或 name 取单行（含密文，内部用）。 + + 权限范围（2026-09-17 修复越权写入漏洞）： + - own_only=False（**读/用**路径）：本人条目 + 本机构共享条目（user_id=''）。 + 机构共享凭据就是给成员用的，读范围必须含它。 + - own_only=True（**写**路径：save/delete/set_status）：只匹配本人条目。 + 否则任意机构成员都能 UPDATE/DELETE 机构共享凭据（按名查到共享行就直接改）， + 等于全机构凭据可被任一成员覆盖或删除。机构共享条目由管理侧（CRUD 页/管理员） + 维护，agent 工具与用户自助路径一律只能动自己的。 + """ + scope = ("AND user_id=${u}$" if own_only + else "AND (user_id=${u}$ OR (org_id=${o}$ AND user_id=''))") + if secret_id: + sql = "SELECT * FROM " + TABLE + " WHERE id=${i}$ " + scope + " LIMIT 1" + ns = {"i": secret_id, "u": user_id or "", "o": org_id or ""} + elif name: + sql = "SELECT * FROM " + TABLE + " WHERE name=${n}$ " + scope + " LIMIT 1" + ns = {"n": name, "u": user_id or "", "o": org_id or ""} + else: + return None + recs = await sor.sqlExe(sql, ns) + return dict(recs[0]) if recs else None + + +async def _name_taken(sor, name: str, org_id: str, user_id: str, + exclude_fp: str = "") -> bool: + """同名是否已被**不同值**占用（同值同名 = 幂等，不算冲突）。 + + 只看本人命名空间（own_only）：机构共享条目占用了某个名字，不应阻止用户建同名的 + 个人条目——个人条目优先级更高（load_visible_secrets 里同名会各自返回，但使用方 + 按名取到的是本人那份），且唯一索引 (org_id,user_id,name) 允许两者共存。 + """ + sql = ("SELECT id, fingerprint FROM " + TABLE + + " WHERE name=${n}$ AND user_id=${u}$ LIMIT 1") + recs = await sor.sqlExe(sql, {"n": name, "u": user_id or ""}) + if not recs: + return False + return (getattr(recs[0], "fingerprint", "") or "") != exclude_fp + + +async def unique_name(sor, base: str, org_id: str, user_id: str, fp: str) -> str: + """名字冲突时加序号（SEC_APIKEY → SEC_APIKEY_2 …），确定性不随机。""" + name = normalize_name(base) or "SEC_SECRET" + for i in range(2, 50): + if not await _name_taken(sor, name, org_id, user_id, fp): + return name + name = normalize_name(base) [:58] + "_" + str(i) + return name[:58] + "_" + str(hashlib.sha1(fp.encode()).hexdigest()[:4]).upper() + + +async def save_secret(sor, *, name: str, value: str, org_id: str = "", user_id: str = "", + label: str = "", secret_type: str = "", source: str = "manual", + remark: str = "", who: str = "") -> Dict: + """入库（新增或更新同指纹条目）。返回 {ok, name, action, message, meta}。 + + 幂等：同 (org,user,name) 已存在且指纹相同 → 直接返回 existing，不重复写。 + """ + from appPublic.uniqueID import getID + from .audit import record_audit + + name = normalize_name(name) + if not name: + return {"ok": False, "message": "FAIL: 变量名不合法（须字母开头，只含大写字母/数字/下划线）"} + if not value: + return {"ok": False, "message": "FAIL: 值不能为空"} + if _PLACEHOLDER_IN_VALUE.search(value): + # 陷阱 B：值本身含占位符语法 → 二次 token 化会改写既有占位符 + return {"ok": False, "message": "FAIL: 值内含占位符语法 @@sec:，已拒绝入库（会破坏替换）"} + + fp = fingerprint(value) + prefix_hint, length_hint = mask_preview(value) + secret_type = secret_type or guess_type(name, value) + enc = encrypt_secret(value) + + # own_only=True：只查本人条目。否则按名查到机构共享条目（user_id=''）就直接 UPDATE， + # 任意机构成员都能覆盖全机构共用的凭据（2026-09-17 自我 review 抓出的越权漏洞）。 + # 机构共享条目与本人同名时可共存（唯一索引含 user_id），本人那份优先。 + existing = await get_secret_row(sor, name=name, org_id=org_id, user_id=user_id, + own_only=True) + # 纵深防御：即便查询返回了非本人行（SQL 被改/复用错路径），也不得 UPDATE + if existing and not _assert_own(existing, user_id): + logger.warning("save_secret blocked: name=%s row owner mismatch", name) + existing = None # 当作不存在 → 走 INSERT 本人新行，不碰他人/共享行 + if existing: + if (existing.get("fingerprint") or "") == fp: + return {"ok": True, "action": "existing", "name": name, + "message": "该名称下已是同一份敏感信息，无需重复保存。", + "meta": {k: v for k, v in existing.items() if k != "encrypted_value"}} + await sor.sqlExe( + "UPDATE " + TABLE + " SET encrypted_value=${e}$, fingerprint=${f}$, " + "prefix_hint=${p}$, length_hint=${l}$, secret_type=${t}$, label=${lb}$, " + "remark=${rm}$, status='active', updated_at=NOW() WHERE id=${i}$", + {"e": enc, "f": fp, "p": prefix_hint, "l": length_hint, "t": secret_type, + "lb": label or existing.get("label") or name, "rm": remark or "", + "i": existing["id"]}) + action, sid = "rotated", existing["id"] + else: + sid = getID() + await sor.sqlExe( + "INSERT INTO " + TABLE + " (id, name, label, secret_type, encrypted_value, " + "fingerprint, prefix_hint, length_hint, org_id, user_id, source, status, " + "remark, use_count, created_at, updated_at) VALUES " + "(${i}$, ${n}$, ${lb}$, ${t}$, ${e}$, ${f}$, ${p}$, ${l}$, ${o}$, ${u}$, " + "${s}$, 'active', ${rm}$, 0, NOW(), NOW())", + {"i": sid, "n": name, "lb": label or name, "t": secret_type, "e": enc, + "f": fp, "p": prefix_hint, "l": length_hint, "o": org_id or "", + "u": user_id or "", "s": source or "manual", "rm": remark or ""}) + action = "created" + + try: + await record_audit(org_id or user_id or "", TABLE, sid, "secret_" + action, + who=who or user_id or "agent", detail="name=" + name + + " type=" + secret_type + " source=" + (source or "")) + except Exception as e: + logger.warning("audit failed on save_secret: %s", str(e)[:120]) + + return {"ok": True, "action": action, "name": name, "id": sid, + "message": ("已保存敏感信息 " + name + "（密文入库，值不会显示给任何人）。" + if action == "created" else + "已更新 " + name + " 的值（旧值已覆盖）。"), + "meta": {"name": name, "secret_type": secret_type, "prefix_hint": prefix_hint, + "length_hint": length_hint, "source": source}} + + +async def delete_secret(sor, *, name: str = "", secret_id: str = "", + org_id: str = "", user_id: str = "", who: str = "") -> Dict: + """删除（**仅本人条目**）。 + + own_only=True：机构共享条目不可由普通成员删除——原实现的 docstring 声称 + 「用 user_id 过滤天然满足」是错的，因为 SQL 的 scope 是 + `user_id=${u}$ OR (org_id=${o}$ AND user_id='')`，机构共享行照样被查中并删掉。 + """ + from .audit import record_audit + row = await get_secret_row(sor, secret_id=secret_id, name=name, + org_id=org_id, user_id=user_id, own_only=True) + if not row or not _assert_own(row, user_id): + # 纵深防御：DELETE 不可逆，归属校验失败一律拒绝（含机构共享条目） + if row: + logger.warning("delete_secret blocked: name=%s owner mismatch", name) + return {"ok": False, "message": "FAIL: 找不到该敏感信息（或不属于你本人）"} + await sor.sqlExe("DELETE FROM " + TABLE + " WHERE id=${i}$", {"i": row["id"]}) + try: + await record_audit(org_id or user_id or "", TABLE, row["id"], "secret_deleted", + who=who or user_id or "agent", detail="name=" + row.get("name", "")) + except Exception: + pass + return {"ok": True, "message": "已删除 " + row.get("name", "")} + + +async def set_secret_status(sor, *, name: str, status: str, org_id: str = "", + user_id: str = "", who: str = "") -> Dict: + """启用/停用（active / disabled）。停用后不再注入执行环境。""" + from .audit import record_audit + if status not in ("active", "disabled"): + return {"ok": False, "message": "FAIL: status 须为 active 或 disabled"} + # own_only=True：停用机构共享凭据会影响全机构成员，只能管理侧操作 + row = await get_secret_row(sor, name=name, org_id=org_id, user_id=user_id, + own_only=True) + if not row or not _assert_own(row, user_id): + # 纵深防御：停用机构共享凭据会影响全机构成员，归属不符一律拒绝 + if row: + logger.warning("set_secret_status blocked: name=%s owner mismatch", name) + return {"ok": False, "message": "FAIL: 找不到该敏感信息（或不属于你本人）"} + await sor.sqlExe("UPDATE " + TABLE + " SET status=${s}$, updated_at=NOW() WHERE id=${i}$", + {"s": status, "i": row["id"]}) + try: + await record_audit(org_id or user_id or "", TABLE, row["id"], "secret_" + status, + who=who or user_id or "agent", detail="name=" + name) + except Exception: + pass + return {"ok": True, "message": name + " 已" + ("启用" if status == "active" else "停用")} + + +async def load_visible_secrets(sor, org_id: str = "", user_id: str = "") -> Dict[str, str]: + """加载可见 secret 的 {NAME: 明文}（**仅执行边界内部使用**，禁止返回给 LLM/落日志）。""" + recs = await sor.sqlExe( + "SELECT name, encrypted_value FROM " + TABLE + + " WHERE status='active' AND (user_id=${u}$ OR (org_id=${o}$ AND user_id='')) LIMIT 200", + {"u": user_id or "", "o": org_id or ""}) + out: Dict[str, str] = {} + for r in (recs or []): + n = getattr(r, "name", "") or "" + e = getattr(r, "encrypted_value", "") or "" + if n and e: + out[n] = decrypt_secret(e) + return out + + +async def touch_usage(sor, names: List[str], org_id: str = "", user_id: str = "") -> None: + """记录使用（use_count + last_used_at）。失败不阻断执行。""" + for n in names or []: + try: + await sor.sqlExe( + "UPDATE " + TABLE + " SET use_count=use_count+1, last_used_at=NOW(), " + "updated_at=NOW() WHERE name=${n}$ AND " + "(user_id=${u}$ OR (org_id=${o}$ AND user_id=''))", + {"n": n, "u": user_id or "", "o": org_id or ""}) + except Exception: + pass + + +# ══════════════════════ 入站门禁（自动侦测 + 入库 + 替换） ══════════════════════ + +async def scan_and_capture(sor, text: str, *, org_id: str = "", user_id: str = "", + who: str = "", auto_store: bool = True) -> Dict: + """入站净化：侦测文本中的敏感信息 → 自动入库 → 原文替换为占位符。 + + Returns: + { + "text": 净化后的文本（进模型上下文的就是它）, + "actions": [人类可读的动作说明], + "captured": [{"name","secret_type","confidence","source"}], + "known_hits": [NAME], # 命中已入库 secret 的明文（不新增） + "unknown_names": [], # 文本里引用了不存在的占位符（提示用户） + } + + 两层替换，顺序固定： + ① 已入库 secret 的明文 → 占位符（精确值匹配，零误报） + ② 新侦测到的高置信度候选 → 自动入库 + 替换（熵门限 + 垃圾值黑名单） + """ + result = {"text": text or "", "actions": [], "captured": [], + "known_hits": [], "unknown_names": []} + if not text: + return result + + # 文本里已有的占位符引用：先校验存在性（未知占位符要告诉用户，不静默） + referenced = find_placeholders(text) + if referenced: + known = set() + try: + known = set(await load_visible_secrets(sor, org_id, user_id)) + except Exception as e: + logger.warning("scan_and_capture load failed: %s", str(e)[:120]) + result["unknown_names"] = [n for n in referenced if n not in known] + if result["unknown_names"]: + result["actions"].append( + "引用了不存在的敏感信息：" + ", ".join(result["unknown_names"]) + + "（已原样保留，执行时不会替换成空值）") + + # ① 已入库明文 → 占位符 + try: + secrets = await load_visible_secrets(sor, org_id, user_id) + except Exception as e: + logger.warning("load_visible_secrets failed: %s", str(e)[:120]) + secrets = {} + cleaned, hits = tokenize(result["text"], secrets) + result["text"] = cleaned + result["known_hits"] = hits + if hits: + result["actions"].append( + "检测到已入库的敏感信息明文，已替换为占位符：" + ", ".join(hits)) + + # ② 新候选 → 自动入库 + if not auto_store: + return result + candidates = detect_candidates(result["text"]) + for cand in candidates: + if cand["confidence"] != "high": + continue # 中低置信度不自动入库（只提示，见下） + val = cand["value"] + fp = cand["fingerprint"] + base = cand["name_suggest"] or _suggest_name(val) + try: + name = await unique_name(sor, base, org_id, user_id, fp) + r = await save_secret(sor, name=name, value=val, org_id=org_id, + user_id=user_id, secret_type=cand["secret_type"], + source="auto_detect", who=who, + remark="会话入站自动侦测（" + cand["source"] + + "，熵 " + str(cand["entropy"]) + "）") + except Exception as e: + logger.warning("auto capture failed: %s", str(e)[:160]) + continue + if not r.get("ok"): + result["actions"].append("自动保存 " + base + " 失败：" + r.get("message", "")) + continue + name = r.get("name", base) + # 用真值替换（此时 result["text"] 里仍是明文） + result["text"] = result["text"].replace(val, make_placeholder(name)) + result["captured"].append({"name": name, "secret_type": cand["secret_type"], + "confidence": cand["confidence"], "source": cand["source"], + "action": r.get("action")}) + result["actions"].append( + "自动侦测到敏感信息（类型 " + cand["secret_type"] + "）→ 已密文入库为 " + + name + " → 消息中已替换为占位符 @@sec:" + name + "@@。" + "执行命令时用 $" + ENV_PREFIX + name + " 引用，真值不会出现在对话里。") + + # 中置信度候选只提示不自动入库（避免误吞；用户可显式 save_secret） + medium = [c for c in candidates if c["confidence"] != "high"] + if medium: + result["actions"].append( + "另有 " + str(len(medium)) + " 处疑似敏感信息置信度不足，未自动入库" + "（如需保存请说「把 XXX 存为敏感信息 YYY」）。") + return result + + +# ══════════════════════ 出站：工具输出净化 ══════════════════════ + +async def sanitize_tool_output(sor, text: str, *, org_id: str = "", + user_id: str = "") -> Tuple[str, List[str]]: + """工具输出净化（需求2的"执行返回当输入处理"）。 + + 只做**已入库 secret 的精确值擦洗**——不做自动侦测入库（工具输出里的疑似串 + 误报代价高：会把哈希/base64 数据块当凭据吞掉，静默毁数据）。 + 高熵未知串只在日志里 WARN 提示，不改写。 + """ + if not text: + return text, [] + try: + secrets = await load_visible_secrets(sor, org_id, user_id) + except Exception as e: + logger.warning("sanitize_tool_output load failed: %s", str(e)[:120]) + return text, [] + cleaned, hits = tokenize(text, secrets) + if hits: + logger.info("tool output scrubbed secrets: %s", ",".join(hits)) + return cleaned, hits + + +# ══════════════════════ 执行边界：占位符 → 环境变量 ══════════════════════ + +async def prepare_command(sor, command: str, *, org_id: str = "", + user_id: str = "") -> Tuple[str, Dict[str, str], List[str]]: + """执行边界唯一入口：命令里的占位符 → `$PIPELINE_SEC_NAME` + 环境变量字典。 + + Returns: + (改写后的命令, 需注入子进程的环境变量, 未知占位符列表) + + 设计：真值**不进命令字符串**——命令里只有变量名，值走 env 注入子进程。 + 这样命令原文落库/进模型上下文都不含明文（对齐 Hermes 实测的安全形态： + `T=$(cat ~/.hermes/scripts/.github_token); curl -H "Authorization: Bearer ***"`）。 + """ + if not command: + return command, {}, [] + try: + secrets = await load_visible_secrets(sor, org_id, user_id) + except Exception as e: + logger.warning("prepare_command load failed: %s", str(e)[:120]) + secrets = {} + rewritten, hits, unknown = detokenize(command, secrets, mode="env") + env = resolve_env(secrets, hits) + if hits: + try: + await touch_usage(sor, hits, org_id, user_id) + except Exception: + pass + return rewritten, env, unknown + + +# ══════════════════════ 会话工具（agent 可见） ══════════════════════ +# handler 签名对齐 pipeline_service 能力包约定：async def h(sor, params, ctx) -> str + +def _ctx_ids(ctx: Dict) -> Tuple[str, str]: + """从 ctx 取 (org_id, user_id)。ctx 可能缺字段，一律兜空串。""" + return (ctx or {}).get("org_id") or "", (ctx or {}).get("user_id") or "" + + +def _fmt_meta_list(rows: List[Dict]) -> str: + if not rows: + return ("你还没有保存任何敏感信息。\n" + "用法：直接说「把 apikey xxx 存为 GITHUB_TOKEN」，或在消息里带上明文——" + "我会自动侦测、密文入库并替换成占位符。") + lines = ["可用敏感信息（值已加密，任何人包括你自己都看不到明文；用 $PIPELINE_SEC_<名称> 在命令里引用）："] + for r in rows: + lines.append( + "- " + str(r.get("name", "")) + + " [类型:" + str(r.get("secret_type", "") or "other") + + " 前缀:" + str(r.get("prefix_hint", "") or "-") + + " 长度:" + str(r.get("length_hint", 0)) + + " 来源:" + str(r.get("source", "") or "manual") + + " 用过:" + str(r.get("use_count", 0)) + "次" + + (" 备注:" + str(r.get("remark", "")) if r.get("remark") else "") + "]") + lines.append("") + lines.append("插入会话用占位符 @@sec:<名称>@@；执行命令时写 $PIPELINE_SEC_<名称>，") + lines.append("平台会在执行瞬间注入真值到子进程环境变量，对话与数据库里都不会出现明文。") + return "\n".join(lines) + + +async def h_list_secrets(sor, params: Dict, ctx: Dict) -> str: + org_id, user_id = _ctx_ids(ctx) + if not user_id: + return "FAIL: 当前会话没有用户身份（无人值守场景不提供敏感信息）" + rows = await list_secrets(sor, org_id=org_id, user_id=user_id, + only_active=str(params.get("all", "")) not in ("1", "true", "True")) + return _fmt_meta_list(rows) + + +async def h_save_secret(sor, params: Dict, ctx: Dict) -> str: + org_id, user_id = _ctx_ids(ctx) + if not user_id: + return "FAIL: 当前会话没有用户身份，拒绝保存敏感信息" + value = params.get("value") or params.get("secret") or "" + name = params.get("name") or "" + if not value: + return "FAIL: 需要 value（敏感信息明文，只在本次调用中使用，落库前即加密）" + name = normalize_name(name) if name else _suggest_name(value, params.get("label") or "") + if not name: + return "FAIL: name 不合法（须字母开头，只含字母/数字/下划线）" + r = await save_secret(sor, name=name, value=value, org_id=org_id, user_id=user_id, + label=params.get("label") or "", secret_type=params.get("secret_type") or "", + source=params.get("source") or "agent", remark=params.get("remark") or "", + who=user_id) + if not r.get("ok"): + return r.get("message", "FAIL: 保存失败") + return (r.get("message", "") + "\n占位符：@@sec:" + r["name"] + + "@@\n命令中引用：$" + ENV_PREFIX + r["name"]) + + +async def h_use_secret(sor, params: Dict, ctx: Dict) -> str: + """把选中的敏感信息插入会话（需求2）——返回占位符，绝不返回值。""" + org_id, user_id = _ctx_ids(ctx) + if not user_id: + return "FAIL: 当前会话没有用户身份" + name = normalize_name(params.get("name") or "") + if not name: + return "FAIL: 需要 name（用 list_secrets 查看可用名称）" + # own_only=False（默认）：读/用路径必须能看到机构共享凭据——那正是共享的意义。 + # 与写路径（save/delete/set_status 用 own_only=True）刻意分离。 + row = await get_secret_row(sor, name=name, org_id=org_id, user_id=user_id) + if not row: + rows = await list_secrets(sor, org_id=org_id, user_id=user_id) + avail = ", ".join(str(x.get("name", "")) for x in rows[:20]) or "（空）" + return "FAIL: 找不到敏感信息 " + name + "。可用：" + avail + if row.get("status") != "active": + return "FAIL: " + name + " 已停用，先用 set_secret_status 启用" + return ("已插入会话。后续引用方式：\n" + "- 文本/消息里用占位符：@@sec:" + name + "@@\n" + "- shell 命令里用环境变量：$" + ENV_PREFIX + name + "\n" + "（元数据：类型 " + str(row.get("secret_type", "")) + "，前缀 " + + str(row.get("prefix_hint", "")) + "，长度 " + str(row.get("length_hint", 0)) + + "。明文不会显示给你，也不会进入对话记录。）") + + +async def h_delete_secret(sor, params: Dict, ctx: Dict) -> str: + org_id, user_id = _ctx_ids(ctx) + if not user_id: + return "FAIL: 当前会话没有用户身份" + r = await delete_secret(sor, name=normalize_name(params.get("name") or ""), + secret_id=params.get("id") or "", org_id=org_id, + user_id=user_id, who=user_id) + return r.get("message", "FAIL") + + +async def h_set_secret_status(sor, params: Dict, ctx: Dict) -> str: + org_id, user_id = _ctx_ids(ctx) + if not user_id: + return "FAIL: 当前会话没有用户身份" + r = await set_secret_status(sor, name=normalize_name(params.get("name") or ""), + status=(params.get("status") or "").strip(), + org_id=org_id, user_id=user_id, who=user_id) + return r.get("message", "FAIL") + + +async def h_detect_secret(sor, params: Dict, ctx: Dict) -> str: + """只侦测不入库（用户想先看看会命中什么）。""" + text = params.get("text") or "" + if not text: + return "FAIL: 需要 text" + cands = detect_candidates(text) + if not cands: + return "未侦测到敏感信息（已知前缀 0 命中、关键词+熵门限 0 命中）。" + lines = ["侦测到 " + str(len(cands)) + " 处疑似敏感信息："] + for c in cands: + v = c["value"] + lines.append("- " + v[:4] + "…(长度" + str(len(v)) + ") 建议名=" + c["name_suggest"] + + " 类型=" + c["secret_type"] + " 置信度=" + c["confidence"] + + " 来源=" + c["source"] + " 熵=" + str(c["entropy"])) + lines.append("（只显示前4字符与元数据，未回显完整值。用 save_secret 可入库。）") + return "\n".join(lines) + + +# 工具定义（ToolDefinition 由调用方按 core 的类构造；这里给声明数据，避免 core 反向依赖） +SECRET_TOOL_SPECS = [ + { + "name": "list_secrets", + "description": "列出当前用户/机构可用的敏感信息（只返回名称/类型/前缀/长度等元数据，绝不返回明文值）。需要用凭据执行命令前先查这个。", + "parameters": {"all": "传 1 含已停用条目(可选)"}, + "category": "secret", + "required": [], + }, + { + "name": "save_secret", + "description": "把一份敏感信息密文入库（apikey/token/密码/私钥等）。用户说「把 X 存为 Y」「保存这个 key」时调用。入库后返回占位符与环境变量名。", + "parameters": {"name": "变量名(大写字母数字下划线,缺省自动派生)", "value": "明文值(仅本次调用使用,落库即加密)", + "label": "显示名(可选)", "secret_type": "类型(可选,自动猜)", "remark": "备注(可选)"}, + "category": "secret", + "required": ["value"], + }, + { + "name": "use_secret", + "description": "从敏感信息表中选一个插入当前会话（返回占位符 @@sec:NAME@@ 与环境变量名，不返回明文）。用户说「用我的 github token」「插入那个 apikey」时调用。", + "parameters": {"name": "变量名(用 list_secrets 查)"}, + "category": "secret", + "required": ["name"], + }, + { + "name": "delete_secret", + "description": "删除一条敏感信息（仅本人/本机构条目）。", + "parameters": {"name": "变量名", "id": "记录ID(可选,与name二选一)"}, + "category": "secret", + "required": [], + }, + { + "name": "set_secret_status", + "description": "启用/停用一条敏感信息（停用后不再注入执行环境，但保留记录）。", + "parameters": {"name": "变量名", "status": "active 或 disabled"}, + "category": "secret", + "required": ["name", "status"], + }, + { + "name": "detect_secret", + "description": "只侦测文本中的疑似敏感信息不入库（返回前4字符+类型+置信度）。用户问「这段有没有泄露风险」时调用。", + "parameters": {"text": "待检测文本"}, + "category": "secret", + "required": ["text"], + }, +] + +SECRET_HANDLERS = { + "list_secrets": h_list_secrets, + "save_secret": h_save_secret, + "use_secret": h_use_secret, + "delete_secret": h_delete_secret, + "set_secret_status": h_set_secret_status, + "detect_secret": h_detect_secret, +} + +#: v1 角色 agent 的工具格式（{"name","description","params","required"}，与 AGENT_TOOLS 同构）。 +#: v2 走 core 的 ToolDefinition（category="secret"），两套定义同源 SECRET_TOOL_SPECS， +#: 避免描述漂移——改一处两边同步。 +SECRET_TOOLS_V1 = [ + {"name": s["name"], "description": s["description"], + "params": dict(s.get("parameters") or {}), + "required": list(s.get("required") or [])} + for s in SECRET_TOOL_SPECS +] + +#: 角色 agent 默认不挂「写」工具（save/delete/set_status）——无人值守场景下 +#: LLM 自主增删凭据风险高于收益，只给读与用（list/use/detect）。 +#: 会话 agent（有真人 user_id）挂全套六件。 +SECRET_TOOLS_V1_READONLY = [ + t for t in SECRET_TOOLS_V1 + if t["name"] in ("list_secrets", "use_secret", "detect_secret") +] + +#: system prompt 片段（会话 agent + 角色 agent 共用） +SECRET_PROMPT_BLOCK = """ +## 敏感信息（凭据）使用铁律 + +平台有用户敏感信息库（apikey/token/密码/私钥），密文存储。**你自始至终看不到明文，这是设计如此。** + +- 需要凭据时先 `list_secrets` 查有哪些（只给元数据：名称/类型/前缀/长度）。 +- 在 shell 命令里引用凭据，一律写环境变量 `$PIPELINE_SEC_<名称>`，平台会在执行瞬间注入真值。 + 例：`curl -H "Authorization: Bearer $PIPELINE_SEC_GITHUB_TOKEN" https://api.github.com/user` +- **禁止**做这些事（会破坏保护）： + - 试图把凭据明文打印出来（`echo $PIPELINE_SEC_X`、`printenv PIPELINE_SEC_X`）——输出会回到对话与数据库，等于泄露； + - 猜、编造、或要求用户提供明文值（用户给明文时平台会自动侦测入库，你只需继续用变量名）； + - 把凭据写进文件、commit 进 git、或放进 URL 查询参数（会落到对方访问日志）。 +- 用户在消息里直接贴了明文（如「apikey：key-xxxx」）时，平台入站门禁已自动密文入库并把原文替换成占位符 `@@sec:<名称>@@`——你会在消息里看到占位符而不是明文，直接用对应环境变量即可，不需要再问用户要。 +- 引用了不存在的名称时占位符会原样保留（不会被替换成空值），命令会失败并报错——这时用 `list_secrets` 核对名称。 +"""