pipeline_core/wwwroot/api/secret_options.dspy
yumoqing a243a16941 feat(secret): 敏感信息库前端接入——AgentIO钥匙图标+选择窗数据源(2026-09-17)
- agent_config.py: GENERAL_TOOLS += 6 个 ToolDefinition(category=secret,
  required 显式声明防 native FC 把可选参数标必填逼 LLM 编值)。
- wwwroot/api/secret_options.dspy: AgentIO 选择窗数据源。身份取自 get_user()/
  get_userorgid(),**不接受前端传 user_id/org_id**(否则改参数即可枚举他人凭据名称);
  双保险剔除 encrypted_value/fingerprint;未登录返 success:false(区分没登录与没数据)。
  RBAC 已被 rp.json 的 /pipeline_core/**(logined) 覆盖,无需额外注册。
- 3 个 .ui 的 AgentIO options 注入 secret_dataurl/secret_format/secret_title/
  secret_tip(共4处,含 agent/index.ui 动态建 tab 的 script 字符串)。
  未传 secret_dataurl 时钥匙图标不渲染 → 其他宿主应用零影响。

bricks 侧配套(UiText.insertAtCursor / AgentInput 弹窗 / AgentIO 透传)在 bricks 仓库。
2026-09-17 15:45:17 +08:00

40 lines
1.9 KiB
Plaintext
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# secret_options.dspy — AgentIO「插入敏感信息」选择窗的数据源2026-09-17
#
# 前端AgentInput 的 opts.secret_dataurl 指向此处bricks 通用框架不硬编码宿主路径,
# 由 pipeline-core 的 agent/index.ui 注入)。
# 返回:{"success":true, "secrets":[{name,label,secret_type,prefix_hint,length_hint,
# use_count,remark,status}, ...]}
#
# 安全铁律(与 secret_vault.list_secrets 同源):
# - **绝不返回 encrypted_value 或明文**。只给元数据:名称/类型/前缀4字符/长度/使用次数。
# 前缀4字符够人类辨认是哪个凭据拼不出可用值GitHub PAT 前缀是公开固定值,零熵)。
# - 身份取自登录态get_user/get_userorgid**不接受前端传 user_id/org_id**——
# 否则任意用户改参数即可枚举他人凭据名称(元数据本身也是信息)。
# - 未登录返回 success:false不返回空列表区分「没登录」与「没数据」便于前端提示
import json
uid = await get_user()
if not uid:
return json.dumps({"success": False, "error": "未登录"}, ensure_ascii=False)
org_id = (await get_userorgid()) or ''
show_all = str((params_kw or {}).get('all', '') or '').strip() in ('1', 'true', 'True')
from sqlor.dbpools import DBPools
from pipeline_service import secret_vault
async with DBPools().sqlorContext('pipeline') as sor:
rows = await secret_vault.list_secrets(sor, org_id=org_id, user_id=uid,
only_active=not show_all)
await sor.sqlExe("COMMIT", {})
# 双保险:即使上游误带出密文字段,这里再剔一次(返回体是安全边界)
_safe = []
for r in rows:
item = {k: v for k, v in r.items()
if k not in ('encrypted_value', 'fingerprint')}
_safe.append(item)
return json.dumps({"success": True, "secrets": _safe}, ensure_ascii=False, default=str)