workbuddy_skills/session-mechanism/scripts/deliver-gateway-token.py

# -*- 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] 环境配置外置（由 install.py 生成）；2026-10-11 起落在本包 `env/roots.env`
def _sm_load_roots():
    import os as _os
    _here = _os.path.dirname(_os.path.abspath(__file__))
    # 🔴 2026-10-11：「脚本是脚本 配置是配置」⇒ 配置落本包 `env/`。
    #    首选 `env/roots.env`；并兼容旧的包根 `roots.env`（拷贝未同步时照旧能跑）。
    for _up in range(4):
        _base = _os.path.normpath(_os.path.join(_here, *([".."] * _up)))
        for _rel in (("env", "roots.env"), ("roots.env",)):
            _p = _os.path.join(_base, *_rel)
            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)          # ⛔ 旁路脚本绝不以非零码影响钩子