admin
19101acd65
- 按用户指示清空原有 25 技能内容，只提交 session-mechanism（57 文件） - 附 .gitignore（产物 + 本机凭据） - 令牌明文已脱敏（历史 .neodata_token 与 pitfalls 引用均不入库） - 本提交为孤儿提交（父提交为空），历史自此重新开始
136 lines
5.6 KiB
Python
136 lines
5.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""把 WorkBuddy 网关口令**投递**给设备接入垫片（A 方案 · 2026-09-30 用户拍板）。
|
||
|
||
🔴 为什么需要它：垫片原设计「口令只在 env」⇒ 由**脱离会话**的上下文（计划任务／常驻）拉起时
|
||
读不到口令 ⇒ 请求一路 fail-closed 成 503，链路永远差最后一跳。
|
||
本脚本跑在**有口令的那一侧**（宿主进程树：钩子／`--tick`）⇒ 天然读得到 env ⇒ 负责「送」。
|
||
|
||
🔴 安全口径（与本项目红线一致）：
|
||
· ⛔ 口令**不落盘**（本文件不写任何凭据；只在进程内从 env 读到内存，直接进请求体）
|
||
· ⛔ 口令**不进日志**（只写「投了/没投/被拒」＋ HTTP 码）
|
||
· ⛔ 只投回环（127.0.0.1）；垫片侧还有"**探活通过才采纳**"的第二道闸（错口令必被拒）
|
||
|
||
⚠️ 节流：≥ `MIN_GAP` 秒才尝试一次（钩子触发很频繁，⛔ 不能每次工具调用都去探）。
|
||
⛔ 失败一律**静默退出 0** —— 它是旁路，⛔ 不得影响钩子本身、更不得影响 WorkBuddy 运行。
|
||
"""
|
||
from __future__ import annotations
|
||
# [session-mechanism] roots.env 外置（由 install.py 生成；缺失则回落到按位置推导）
|
||
def _sm_load_roots():
|
||
import os as _os
|
||
_here = _os.path.dirname(_os.path.abspath(__file__))
|
||
for _up in range(4):
|
||
_p = _os.path.join(_here, *([".."] * _up), "roots.env")
|
||
_p = _os.path.normpath(_p)
|
||
if _os.path.isfile(_p):
|
||
try:
|
||
with open(_p, encoding="utf-8") as _f:
|
||
for _ln in _f:
|
||
_ln = _ln.strip()
|
||
if _ln and not _ln.startswith("#") and "=" in _ln:
|
||
_k, _v = _ln.split("=", 1)
|
||
_os.environ.setdefault(_k.strip(), _v.strip())
|
||
except Exception:
|
||
pass
|
||
return
|
||
_sm_load_roots()
|
||
# 🔴 2026-10-01 加 · **输出编码兜底**：脚本一旦被重定向（钩子/常驻/后台任务都会这么干），
|
||
# Windows 本地编码（GBK）编不出 ⛔/✅/🔴 这类字符 ⇒ `print` 抛 UnicodeEncodeError
|
||
# ⇒ 被顶层 handler 记成 `fatal`、**整轮失败**（实测：本包里连续 4 次 `fatal 'gbk' codec ...`）。
|
||
# ⇒ 出口一律 UTF-8 ＋ errors="replace"（⛔ 不让"打不出字"升级成"程序死"）。
|
||
try:
|
||
import sys as _sys
|
||
_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
|
||
PORT = 20090
|
||
SELFCHECK = "/__device_access/selfcheck"
|
||
TOKEN_PATH = "/__device_access/token"
|
||
TOKEN_ENV = "CODEBUDDY_GATEWAY_PASSWORD"
|
||
MIN_GAP = 60.0 # 秒
|
||
|
||
WS = os.environ.get("DSH_COLLAB_WS") or os.environ.get("DSH_WS_ROOT") or os.getcwd()
|
||
STAMP = os.path.join(WS, "tmp", "supervise-inbox", "_token-deliver.stamp")
|
||
LOG = os.path.join(WS, "tmp", "supervise-inbox", "_token-deliver.log")
|
||
|
||
|
||
def _log(msg: str) -> None:
|
||
try:
|
||
os.makedirs(os.path.dirname(LOG), exist_ok=True)
|
||
with open(LOG, "a", encoding="utf-8") as f:
|
||
f.write("[%s] %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), msg))
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _http(url: str, data: bytes | None = None, timeout: float = 4.0):
|
||
req = urllib.request.Request(url, data=data, method=("POST" if data is not None else "GET"))
|
||
if data is not None:
|
||
req.add_header("Content-Type", "text/plain")
|
||
with urllib.request.urlopen(req, timeout=timeout) as r: # noqa: S310（只打回环）
|
||
return r.status, r.read().decode("utf-8", "replace")
|
||
|
||
|
||
def main() -> int:
|
||
# ① 节流：⛔ 每次工具调用都探一遍会平白增加负担
|
||
try:
|
||
if os.path.isfile(STAMP) and (time.time() - os.path.getmtime(STAMP)) < MIN_GAP:
|
||
return 0
|
||
os.makedirs(os.path.dirname(STAMP), exist_ok=True)
|
||
with open(STAMP, "w", encoding="utf-8") as f:
|
||
f.write(str(time.time()))
|
||
except Exception:
|
||
pass
|
||
|
||
token = os.environ.get(TOKEN_ENV) or ""
|
||
if token == "":
|
||
# 本进程没有口令 ⇒ 不可能是"送的一方"（钩子/--tick 才有）；⛔ 不报错、不落 NEED-USER
|
||
return 0
|
||
|
||
base = "http://127.0.0.1:%d" % PORT
|
||
# ② 垫片不在 ⇒ 静默走人（⛔ 不起它；起它是另一件事）
|
||
try:
|
||
code, body = _http(base + SELFCHECK)
|
||
except Exception:
|
||
return 0
|
||
if code != 200:
|
||
return 0
|
||
try:
|
||
present = bool((json.loads(body).get("credential") or {}).get("present"))
|
||
except Exception:
|
||
present = False
|
||
if present:
|
||
return 0 # 已就位（env 或已投过）⇒ 不必重复投
|
||
|
||
# ③ 投（⛔ 口令只在请求体里，⛔ 绝不进日志/URL）
|
||
try:
|
||
code, body = _http(base + TOKEN_PATH, data=token.encode("utf-8"))
|
||
except Exception as e:
|
||
_log("投递异常（已静默）: %s" % type(e).__name__)
|
||
return 0
|
||
try:
|
||
j = json.loads(body)
|
||
except Exception:
|
||
j = {}
|
||
if code == 200 and j.get("ok"):
|
||
_log("投递成功 → source=%s" % ((j.get("credential") or {}).get("source") or "?"))
|
||
else:
|
||
# 400 token-rejected-by-gateway = 口令过期/网关重启（正常会自愈，⛔ 不惊动用户）
|
||
_log("投递被拒 http=%s reason=%s" % (code, j.get("reason") or "?"))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except Exception:
|
||
sys.exit(0) # ⛔ 旁路脚本绝不以非零码影响钩子
|