1057 lines
51 KiB
Python
1057 lines
51 KiB
Python
"""用户敏感信息管理（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_<NAME>`（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 update_secret_meta(sor, *, name: str, label: str = "", remark: str = "",
|
||
org_id: str = "", user_id: str = "", who: str = "") -> Dict:
|
||
"""改标签/备注（**仅本人条目**；值不可改——改值走删除重存，避免指纹/审计歧义）。
|
||
|
||
管理页（2026-09-18）用；与 set_secret_status 同款 own_only + _assert_own 双防线。
|
||
"""
|
||
from .audit import record_audit
|
||
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):
|
||
return {"ok": False, "message": "FAIL: 找不到该敏感信息（或不属于你本人）"}
|
||
await sor.sqlExe(
|
||
"UPDATE " + TABLE + " SET label=${l}$, remark=${r}$, updated_at=NOW() WHERE id=${i}$",
|
||
{"l": label or "", "r": remark or "", "i": row["id"]})
|
||
try:
|
||
await record_audit(org_id or user_id or "", TABLE, row["id"], "secret_meta_update",
|
||
who=who or user_id or "agent", detail="name=" + name)
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "message": 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
|
||
|
||
|
||
# ══════════════════════ 执行边界：占位符 → 环境变量 ══════════════════════
|
||
|
||
#: env 引用形态（$PIPELINE_SEC_NAME / ${PIPELINE_SEC_NAME}）——prompt 承诺的写法
|
||
_ENV_REF_RE = re.compile(r"\$\{?" + ENV_PREFIX + r"([A-Z][A-Z0-9_]*)\}?")
|
||
|
||
|
||
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 引用形态补齐（2026-09-17 缺口修复，e2e 实锤）：SECRET_PROMPT_BLOCK 承诺
|
||
# 「命令里写 $PIPELINE_SEC_X 平台执行瞬间注入真值」，但此前只认占位符形态——
|
||
# agent 照 prompt 写 env 引用时 hits 为空 → env 不注入 → 变量展开为空
|
||
# （实测 stdout `SEC_READY_` 空值）。两种写法从此等价：
|
||
# 已存在名称 → 进 hits（env 注入）；不存在名称 → 进 unknown（拒绝执行，
|
||
# 与陷阱 C「绝不静默空值」同原则——空展开会静默改变命令语义）。
|
||
for m in _ENV_REF_RE.finditer(rewritten):
|
||
n = m.group(1)
|
||
if n in secrets:
|
||
if n not in hits:
|
||
hits.append(n)
|
||
elif n not in unknown:
|
||
unknown.append(n)
|
||
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` 核对名称。
|
||
"""
|