feat(ops)+docs: 密钥轮换工具 + 两份文档目录审计收口（D1.1 §23 / D2.1 v6.26）
一、密钥轮换（新增工具 + 操作手册） - 新增 tools/rotate_api_keys.py：--check 体检 + 交互式轮换；getpass 不回显、 自动备份 .env.bak-<时间戳>（已被 ignore 命中）、校验不过整体不写入、 三个 Qwen 变量写同一值 / 两个 DeepSeek 变量写同一值。 实测 --check：Qwen 三变量同值且非空、DeepSeek 两变量同值且非空。 - 新增 开发文档/D3.8-模型密钥轮换与凭据安全操作手册-2026-09-20.md（CS-OPS-2026-023）： .env 5 个变量与读取方取证、五步流程、3 个坑、复核清单、回退方式、能力边界。 - 口径确认：model_endpoint_config.secret_ref 存变量名 ⇒ 轮换只改 .env，不动 DB； 但必须重启 API + Worker。 二、门禁修复：docs/ 编号撞车 - tools/check_authoritative_docs.py（D3.4 N-14 登记的验收命令集之一）实测 FAIL： 我方 docs/46 docs/47（2026-09-20 建）与投顾组 docs/46-投顾Agent需求文档.md docs/47-投顾Agent功能架构文档.md（2026-09-16 建）同号。 - 按「后到者让位」改名：docs/48-可改文件白名单.md / docs/49-底座会签申请单-2026-09-19.md， 同步 8 处引用。修复后：checked 54 documents, no number collision，exit 0。 三、前端品牌残留（W12 合并静默回退） - employee-advisor/dashboard/index.html 与 customer/advisor-plans/index.html 的 <title> 仍是 南方财富（投顾组分支带回）→ 按 DEC-27 改为 南方基金。 - HTTP 实测两页标题已正确；全仓 app/ 复查 南方财富 = 0。 四、文档口径校准（12 处事实漂移） - D1.1：D2.1 版本 v5.3 → v6.26（§4.0 / §4.1 / §1 / §2 四处长期错误）； §8 四行遗留项闭合（D-5 / D-6 / D-7 / 仓库副本同步）+ 新增 §22 §23 留痕； §10.2「本区不在任何 git 仓库内」更正为已入库；新增两编号 ⇒ 计数 56 → 58 全量同步。 - D2.2：顶栏徽标 v2.4 与元数据 v2.5 自相矛盾 → 统一；「投顾已清除」→ 状态更新 （模块 2026-09-20 已恢复，但客服范围裁定 §1.7 / RK-10 不变）。 - D2.3：徽标 v1.0 · 7 批次 51 项 → v1.1 · 8 批次 57 项；投顾清除后果 + §7.1 头号风险 + 风险表 + 不触碰行全部加恢复口径。 - D2.4：v1.3 变更说明 ⑦ / §1.4 Out of scope / Q-09 加投顾恢复口径。 - D2.5：advisor_t 自相矛盾口径改写为账号表一行 + 口径更正；五项自检首选改为 一键脚本 启动演示.bat / demo.ps1；补 D3.8 与未发布 advisor:* 白名单登记。 - D2.6：门禁数字 1856/2 → 1909/3 skipped、ruff 19 → 20、补 portal_api_check 行； §10 两项已闭环（密钥轮换已工具化、A-10 组 3/4 已补签）；头部加 W12/W13 状态更新。 - D4.5：顶部状态更新补指向 D4.7。 - 新增 开发文档/D4.7-投顾模块恢复记录-2026-09-20.md（CS-PURGE-2026-014）： 时间线、8 项恢复动作、客服线不变的结论、DEC-19 理由更正、遗留 1 项、失误登记。 - _consistency.py（维护侧）：§三 改为「投顾状态口径检查」，合法语境扩为 清除史 / 恢复史 / 不属本 Agent 范围。 五、回归实测（全绿） - pytest -q：1909 passed / 3 skipped / 0 failed - ruff check app tools tests：20（与 W12 持平，未引入新债） - mypy app：2（= 既有基线） - tools/check_authoritative_docs.py：54 文档无编号冲突（exit 0） - tools/e2e_smoke_test.py --read-only：31/31 - tools/portal_api_check.py：40 项 通过 35 / 失败 0 / 跳过 5 - _eval_harness/http_probe.py：11/11 succeeded - _consistency.py：GATE PASS - demo.ps1 -SkipStart -NoBrowser：五项自检全过、退出码 0 - 权威副本 ↔ 仓库：逐字节一致（客服agent 24 / 开发文档 52） 六、未做（如实登记） - 投顾 config_release 工具白名单（advisor:*）仍未发布 ⇒ 投顾 Agent 工具调用 fail closed （实测 active_agent_tools 仅 customer_service:* 4 项 + risk:* 4 项）。与客服线无关； 要演投顾线先跑 tools/publish_advisor_demo_config.py --apply。 - 两把 key 的实际轮换需你在控制台建新 key（无法代做），流程见 D3.8。
This commit is contained in:
张胜宇
@@ -0,0 +1,172 @@
|
||||
"""轮换 `.env` 里的模型密钥（DashScope/Qwen 与 DeepSeek）。
|
||||
|
||||
## 为什么需要它
|
||||
|
||||
这两把 key **在聊天/文档里出现过明文**，必须轮换。手工改 `.env` 有三个坑：
|
||||
|
||||
1. **三个 Qwen 变量必须同一个值** —— `QWEN_API_KEY` / `QWEN_EMBEDDING_API_KEY` /
|
||||
`DASHSCOPE_API_KEY`。漏改一个，入库与检索就用了两把不同的 key，
|
||||
症状是「检索能跑、入库失败」或反过来（很难一眼看出是 key 的问题）。
|
||||
2. **两个 DeepSeek 变量也必须同一个值** —— `DEEPSEEK_API_KEY` 与
|
||||
`OFFSITE_DEEPSEEK_API_KEY`（场外文档识别复用同一把）。
|
||||
3. 写文件时若**带上 BOM**或**改变编码/换行**，`.env` 第一项就可能读不出来。
|
||||
|
||||
## 用法
|
||||
|
||||
.venv\\Scripts\\python.exe tools\\rotate_api_keys.py # 交互输入（不回显）
|
||||
.venv\\Scripts\\python.exe tools\\rotate_api_keys.py --check # 只体检当前 5 个变量的取值关系
|
||||
|
||||
## 安全约定
|
||||
|
||||
- 输入**不回显**（`getpass`），**不写入任何日志**，**不回显 key 值**（只打印掩码与长度）。
|
||||
- 改前自动备份为 `.env.bak-<时间戳>`（该模式已被 `.gitignore` 覆盖）。
|
||||
- 任一校验不通过就**整体不写入**（先全量校验、后一次性落盘）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import shutil
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
ENV_PATH = Path(__file__).resolve().parent.parent / ".env"
|
||||
|
||||
#: 变量名 -> 归属的 key（同一个 key 必须写成同一个值）
|
||||
QWEN_VARS = ("QWEN_API_KEY", "QWEN_EMBEDDING_API_KEY", "DASHSCOPE_API_KEY")
|
||||
DEEPSEEK_VARS = ("DEEPSEEK_API_KEY", "OFFSITE_DEEPSEEK_API_KEY")
|
||||
TARGETS = {name: "qwen" for name in QWEN_VARS} | {name: "deepseek" for name in DEEPSEEK_VARS}
|
||||
|
||||
MIN_LEN = 20
|
||||
|
||||
|
||||
def mask(value: str) -> str:
|
||||
"""只暴露前 6 位，其余按长度打码 —— 够判断"换没换"，不够还原。"""
|
||||
if not value:
|
||||
return "(空)"
|
||||
return f"{value[:6]}…({len(value)} 位)"
|
||||
|
||||
|
||||
def parse_env(text: str) -> list[tuple[str, str]]:
|
||||
rows: list[tuple[str, str]] = []
|
||||
for line in text.split("\n"):
|
||||
if "=" in line and not line.lstrip().startswith("#"):
|
||||
key, _, value = line.partition("=")
|
||||
rows.append((key.strip(), value))
|
||||
return rows
|
||||
|
||||
|
||||
def check(rows: list[tuple[str, str]]) -> int:
|
||||
"""体检：三个 Qwen 是否同值、两个 DeepSeek 是否同值、是否为空/像占位符。"""
|
||||
values = dict(rows)
|
||||
problems = 0
|
||||
for group, names in (("Qwen（DashScope）", QWEN_VARS), ("DeepSeek", DEEPSEEK_VARS)):
|
||||
present = [values.get(n, "") for n in names]
|
||||
print(f"· {group}：")
|
||||
for name in names:
|
||||
print(f" {name:<26} {mask(values.get(name, ''))}")
|
||||
uniq = {v for v in present if v}
|
||||
if not uniq:
|
||||
print(" ❌ 全部为空")
|
||||
problems += 1
|
||||
elif len(uniq) > 1:
|
||||
print(" ❌ **取值不一致** —— 这会让「入库/检索」或「主链路/场外链路」用到不同的 key")
|
||||
problems += 1
|
||||
elif len(present) != len([v for v in present if v]):
|
||||
print(" ❌ 有变量为空（必须同一个值）")
|
||||
problems += 1
|
||||
else:
|
||||
print(" ✅ 同值且非空")
|
||||
return problems
|
||||
|
||||
|
||||
def rewrite(text: str, values: dict[str, str]) -> tuple[str, int]:
|
||||
"""只替换目标行的值，其它行（含缩进、注释、顺序）原样保留。"""
|
||||
out: list[str] = []
|
||||
replaced = 0
|
||||
for line in text.split("\n"):
|
||||
stripped = line.lstrip()
|
||||
if stripped.startswith("#") or "=" not in line:
|
||||
out.append(line)
|
||||
continue
|
||||
key, sep, _old = line.partition("=")
|
||||
name = key.strip()
|
||||
if name in values and name in TARGETS:
|
||||
out.append(f"{name}{sep}{values[name]}")
|
||||
replaced += 1
|
||||
else:
|
||||
out.append(line)
|
||||
return "\n".join(out), replaced
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="轮换 .env 里的 Qwen / DeepSeek 密钥")
|
||||
parser.add_argument("--check", action="store_true", help="只体检，不修改")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not ENV_PATH.exists():
|
||||
print(f"❌ 找不到 {ENV_PATH}")
|
||||
return 2
|
||||
raw = ENV_PATH.read_bytes()
|
||||
if raw.startswith(b"\xef\xbb\xbf"):
|
||||
print("⚠️ .env 带 UTF-8 BOM：本次会去掉它（BOM 会让第一个变量名读不出来）")
|
||||
newline = "\r\n" if b"\r\n" in raw else "\n"
|
||||
text = raw.decode("utf-8-sig").replace("\r\n", "\n")
|
||||
rows = parse_env(text)
|
||||
|
||||
print("=== .env 密钥体检（只显示掩码）===")
|
||||
problems = check(rows)
|
||||
if args.check:
|
||||
print()
|
||||
print("体检完成。" + ("存在不一致，建议先修。" if problems else "一致。"))
|
||||
return 1 if problems else 0
|
||||
|
||||
missing = [n for n in TARGETS if n not in dict(rows)]
|
||||
if missing:
|
||||
print(f"❌ .env 里缺少这些变量：{missing}")
|
||||
return 2
|
||||
|
||||
print()
|
||||
print("=== 开始轮换 ===")
|
||||
print("直接回车 = 跳过该项（保留旧值）。输入不会回显，也不会被记录。")
|
||||
print()
|
||||
new_values: dict[str, str] = {}
|
||||
for group, names in (("DashScope / Qwen", QWEN_VARS), ("DeepSeek", DEEPSEEK_VARS)):
|
||||
entered = getpass.getpass(f"{group} 的新 key（回车跳过）：").strip()
|
||||
if not entered:
|
||||
print(f" · {group}：跳过，保留旧值")
|
||||
continue
|
||||
if not entered.startswith("sk-") or len(entered) < MIN_LEN:
|
||||
print(
|
||||
f" ❌ {group}：格式不像 key（应以 'sk-' 开头且长度 ≥ {MIN_LEN}）—— **本次不做任何写入**"
|
||||
)
|
||||
return 3
|
||||
for name in names:
|
||||
new_values[name] = entered
|
||||
print(f" · {group}：将更新 {len(names)} 个变量（{mask(entered)}）")
|
||||
|
||||
if not new_values:
|
||||
print()
|
||||
print("没有输入任何 key，未做修改。")
|
||||
return 0
|
||||
|
||||
backup = ENV_PATH.with_name(f".env.bak-{datetime.now():%Y%m%d-%H%M%S}")
|
||||
shutil.copy2(ENV_PATH, backup)
|
||||
|
||||
updated, replaced = rewrite(text, new_values)
|
||||
if replaced != len(new_values):
|
||||
print(f"❌ 只匹配到 {replaced}/{len(new_values)} 行，已放弃写入（备份在 {backup.name}）")
|
||||
return 3
|
||||
ENV_PATH.write_bytes(updated.replace("\n", newline).encode("utf-8"))
|
||||
print()
|
||||
print(f"✅ 已更新 {replaced} 行；备份：{backup.name}")
|
||||
print(
|
||||
" 下一步：重启 API 与 Worker 让新 key 生效，然后跑一次链路复核（见仓库根 demo.ps1 的自检）。"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user