208 lines
9.9 KiB
Python
208 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
lock-guard-hook.py — 「无锁不许改库」的强制钩子（WorkBuddy PreToolUse / SessionStart）
|
||
|
||
为什么需要它
|
||
────────────
|
||
2026-09-12 实证：`handoff-guard.sh` 输出「无全局锁」被会话读成「环境干净，可以开工」
|
||
（正确读法是「你快去抢锁」），结果**两个会话同时改了本库**。措辞已在 guard 与
|
||
`交接单/README.md` 里补正，但**约定拦不住不看的人** —— 本脚本是强制层。
|
||
|
||
作用域（**刻意收窄：对其他项目零影响**）
|
||
────────────
|
||
仅当 `Write` / `Edit` 的目标路径落在下列根之内，才做锁判定；其余一律放行：
|
||
· 文档库 <DSH_DOCS_ROOT>（默认 E:\\ProgramData\\AI技能\\aliyun-dsh-server\\dsh-server-docs）
|
||
· 代码库 <DSH_CODE_REPO>（默认 D:\\github\\dsh_shenxian）
|
||
⚠️ 例外：路径中含 `.workbuddy` 目录段的一律放行 —— 会话记忆 / 自动化工作数据属“运行态数据”，不是“仓库内容”。（2026-09-13 实测：「代码仓三方同步（DSH）」自动化写自己的 memory/*.md 时被本钩子 deny，属误伤）
|
||
|
||
判据
|
||
────
|
||
`<文档库根>/交接单/.exec-lock` 存在 = 有人持锁 → 放行（归属由台账 OWNER 承担）；
|
||
不存在 = **deny**，并把可直接粘贴的抢锁命令回给 Agent。
|
||
|
||
为什么**不拦 Bash**
|
||
──────────────────
|
||
① 抢锁命令本身必须能跑（否则把自己锁死 —— 拿不到锁就永远开不了工）；
|
||
② Bash 写文件是少数派、且破坏面可见（git status）；
|
||
③ 宁可留一个显式的安全阀，也不要一个可能导致死锁的强制层。
|
||
|
||
用法（settings.json 的 hooks 段，见同目录 README 或档案 73）
|
||
"PreToolUse": [{ "matcher": "Write|Edit", "hooks": [{ "type": "command", "command": "<python> <此脚本>", "timeout": 10 }] }]
|
||
"SessionStart":[{ "matcher": "startup", "hooks": [{ "type": "command", "command": "<python> <此脚本>", "timeout": 10 }] }]
|
||
|
||
退出码：始终 0（判定通过 JSON 输出表达）；脚本自身异常也放行，绝不误伤。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
DOCS_ROOT = os.environ.get("DSH_DOCS_ROOT", r"D:\github\dsh_shenxian\dsh-server-docs")
|
||
CODE_REPO = os.environ.get("DSH_CODE_REPO", r"D:\github\dsh_shenxian")
|
||
LOCK_DIR = os.path.join(DOCS_ROOT, "交接单", ".exec-lock")
|
||
OWNER_FILE = os.path.join(LOCK_DIR, "OWNER")
|
||
|
||
GUARD_CMD = 'bash scripts/handoff-guard.sh --claim-exec "<你的会话名>"'
|
||
|
||
|
||
def norm(p: str) -> str:
|
||
try:
|
||
return os.path.normcase(os.path.normpath(p))
|
||
except Exception:
|
||
return p
|
||
|
||
|
||
PROTECTED = [norm(DOCS_ROOT), norm(CODE_REPO)]
|
||
|
||
|
||
def is_protected(path: str) -> bool:
|
||
if not path:
|
||
return False
|
||
n = norm(path)
|
||
# 会话记忆 / 自动化工作数据不是「仓库内容」：写它不该被本钩子拦（2026-09-13 新增）
|
||
if ".workbuddy" in n.split(os.sep):
|
||
return False
|
||
for root in PROTECTED:
|
||
if n == root or n.startswith(root + os.sep):
|
||
return True
|
||
return False
|
||
|
||
|
||
def lock_owner() -> str:
|
||
try:
|
||
with open(OWNER_FILE, encoding="utf-8") as f:
|
||
return (f.readline() or "").strip() or "（未写 OWNER）"
|
||
except Exception:
|
||
return "（未写 OWNER）"
|
||
|
||
|
||
def out(obj: dict) -> None:
|
||
# ⚠️ 2026-09-16 修复（**根因**）：必须走 **`sys.stdout.buffer` 写 bytes**，⛔ 不能走文本层
|
||
# （`sys.stdout.write(str)`）。
|
||
# 实证：宿主 spawn 本脚本时 stdout 编码**不保证是 UTF-8**（本机 `PYTHONUTF8=1` 在 hook 环境下
|
||
# 不保证生效），而 deny 文案里含 `⛔` / `🔐` 等字符 ⇒ 文本层写抛 `UnicodeEncodeError` ⇒
|
||
# 异常冒泡、**stdout 为空** ⇒ 宿主**收不到 deny** ⇒ **静默放行**。
|
||
# 而 `hook_log()` 已经把 `PreToolUse-deny` 写进日志 ⇒ 表现成 **「日志里有 DENY，但拦不住」**
|
||
# （2026-09-16 受控自测：无锁时 Write 文档库 ⇒ 文件真被创建）。
|
||
# 对照：`bash-output-guard.py` 用 `sys.stdout.buffer.write(bytes)` ⇒ **实测拦得住**。
|
||
# 📌 教训：本脚本的 `stdin` 早前已按 buffer 加固（见 main 顶部注释），**stdout 漏了** ——
|
||
# 「读写要同时加固」，只修一半等于没修。
|
||
data = json.dumps(obj, ensure_ascii=False).encode('utf-8')
|
||
try:
|
||
sys.stdout.buffer.write(data)
|
||
sys.stdout.buffer.flush()
|
||
except Exception:
|
||
try:
|
||
sys.stdout.write(data.decode('utf-8', 'replace'))
|
||
sys.stdout.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
LOG_FILE = os.environ.get(
|
||
"DSH_LOCK_HOOK_LOG", os.path.join(os.path.dirname(DOCS_ROOT), ".workbuddy", "lock-hook.log")
|
||
)
|
||
|
||
|
||
def hook_log(event: str, detail: str) -> None:
|
||
"""低频自证日志：只在 SessionStart 与 deny 时写一行 —— 用来回答「hook 到底有没有被触发」。
|
||
|
||
为什么需要：hook 配置是**启动时缓存**的，改完必须完全重启才加载；没有日志就只能靠猜。
|
||
写入失败一律静默（hook 绝不能因为自己出问题而干扰工作）。
|
||
"""
|
||
try:
|
||
os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')}\t{event}\t{detail}\n")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def main() -> None:
|
||
try:
|
||
try:
|
||
# ⚠️ 必须走 buffer 显式 UTF-8：本机环境设了 PYTHONUTF8=1，但只要有人给脚本加 `-E`
|
||
# 就会被屏蔽 ⇒ sys.stdin 回退 cp936 ⇒ 含中文路径（`E:\ProgramData\AI技能\…`）的
|
||
# payload 解析即炸。而本函数是 **fail-open**（读不懂就放行）⇒ 会**静默失效**：
|
||
# 锁守卫不再拦人，却没有任何痕迹。2026-09-15 实测（同款坑已在 stop-dialog-guard.py 踩过）。
|
||
raw = sys.stdin.buffer.read().decode("utf-8", "replace")
|
||
except Exception:
|
||
raw = sys.stdin.read()
|
||
payload = json.loads(raw) if raw.strip() else {}
|
||
except Exception as e:
|
||
# 不留痕 = 失效无声（判不出"没被调用"与"被静默放行"）⇒ 必须记一行
|
||
hook_log("payload-unparsable", "%s: %s" % (type(e).__name__, str(e)[:80]))
|
||
return # 读不懂 payload → 放行（fail-open 方向正确，但不能无声）
|
||
|
||
tool = payload.get("tool_name")
|
||
event = payload.get("hook_event_name") or payload.get("hook_event") or ""
|
||
|
||
# ★ 入口即留痕（2026-09-16 加）—— 与 stop-dialog-guard.py / skill-load-guard.py 同款。
|
||
# 为什么不只记 SessionStart 与 deny：2026-09-16 实测 —— 重启后本钩子**一行都没写**，
|
||
# 当时无法区分"没被宿主调用"与"被静默放行"，只能靠猜（最后查明：SessionStart 的
|
||
# `matcher: "startup"` 不匹配"恢复会话"，钩子压根没被执行）。
|
||
# 留痕里带上**实际收到的字段名**（keys）：下次宿主字段一变，日志里立刻能看出来，
|
||
# 而不是等某天发现守卫早已失效。低频：每次调用一行。
|
||
hook_log("entry", "event=%s｜tool=%s｜keys=%s" % (
|
||
event or "?", tool or "-", ",".join(sorted(payload.keys()))[:220]))
|
||
|
||
# ── SessionStart：只提示（该事件的输出只给用户看，不会进 Agent 上下文）──
|
||
if event == "SessionStart" or (tool is None and "source" in payload):
|
||
src = payload.get("source", "?")
|
||
if os.path.isdir(LOCK_DIR):
|
||
who = lock_owner()
|
||
msg = f"🔐 全局执行锁【已被占用】：{who} —— 同一时刻只允许一个执行会话动「文档/代码/服务器」。"
|
||
hook_log("SessionStart", f"已被占用 owner={who} source={src}")
|
||
elif os.path.isdir(DOCS_ROOT):
|
||
msg = "🔐 全局执行锁【空闲】—— 但「空闲」≠「可以开工」：动手前先抢锁 → " + GUARD_CMD
|
||
hook_log("SessionStart", f"空闲 source={src}")
|
||
else:
|
||
msg = ""
|
||
if msg:
|
||
out({"systemMessage": msg, "suppressOutput": True})
|
||
return
|
||
|
||
# ── PreToolUse：真正的强制点 ──
|
||
if tool not in ("Write", "Edit", "MultiEdit", "NotebookEdit"):
|
||
return
|
||
ti = payload.get("tool_input") or {}
|
||
fp = ti.get("file_path") or ti.get("path") or ti.get("notebook_path") or ""
|
||
if not is_protected(fp):
|
||
return
|
||
if os.path.isdir(LOCK_DIR):
|
||
return # 有人持锁 → 放行（归属由 OWNER / 台账承担）
|
||
if not os.path.isdir(DOCS_ROOT):
|
||
return # 库不在这台机器上 → 与本约定无关，放行
|
||
|
||
hook_log("PreToolUse-deny", f"{tool} {fp}")
|
||
|
||
reason = (
|
||
"⛔ 被「锁机制」拦下：本仓库（文档库 / 代码库）当前**无人持全局执行锁**，"
|
||
"不允许直接修改。\n\n"
|
||
"「无锁」的正确读法 = **「你快去抢」**，不是「可以开工」"
|
||
"（2026-09-12 实证：两个会话把「无锁」读成「环境干净」→ 同时改了本库）。\n\n"
|
||
f"请先执行（Bash，不受本钩子限制）：\n cd \"{DOCS_ROOT}\"\n {GUARD_CMD}\n\n"
|
||
"抢到 = 开工许可；**抢不到 = 有会话在跑 → 停手**（输出会告诉你占用者与在做哪单）。\n"
|
||
"若确实只需追加一行、且已确认无人在动，可由用户裁定后临时移除本钩子。"
|
||
)
|
||
out(
|
||
{
|
||
"hookSpecificOutput": {
|
||
"hookEventName": "PreToolUse",
|
||
"permissionDecision": "deny",
|
||
"permissionDecisionReason": reason,
|
||
},
|
||
"systemMessage": "⛔ 已拦下一次无锁写入（本仓库要求先抢锁）",
|
||
"suppressOutput": True,
|
||
}
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception:
|
||
pass # 任何异常都不阻断工作
|
||
sys.exit(0)
|