# -*- coding: utf-8 -*- """wake-session.py —— 本机 WorkBuddy 网关「唤醒」投递器（纯本机 · 无代理 · 无垫片） 用途 ---- 把「唤醒某个 WorkBuddy 会话」做成**一次程序调用**： 程序 → 本机 WorkBuddy gateway（127.0.0.1:<动态口>）→ POST /api/v1/sessions/{id}/reply ⛔ 不经过覆盖网络 443、⛔ 不经过设备垫片 20090、⛔ 不经过任何代理 —— 全程本机回环。 设计约束（对应架构定稿 v3 §0.5 T1–T7） -------------------------------------- · 凭据**只**从环境变量 `CODEBUDDY_GATEWAY_PASSWORD` 读：⛔ 命令行明文、⛔ 落盘、⛔ 回显 · 端口**每次现查**（网关每次重启都换口）⇒ **fail-closed**：查不到就具名失败，⛔ 绝不沿用上次端口 · 只投给 **live 且 `live.sessionId == 目标`** 的会话（T2：不夺会话、不抢 `writer_occupied`） · 超时：读 ≤5 s、写 ≤10 s（T5）；**单次投递、不重试**（T4：绝不反复试凭据、不用重试把自己锁死） · 本程序**不监听任何端口**；⛔ 不改 WorkBuddy 配置 / 插件 / 令牌 / 进程（T3） · 单次短操作 ⇒ 不长期持锁（T6）；无守护、无残留（T7） 用法 ---- python wake-session.py --list # 列出本机所有网关及其 live 会话 python wake-session.py --session current --dry-run # 只做发现 ＋ 前置判定，不投递 python wake-session.py --session current --text "…" # 真投递（默认目标＝当前会话） python wake-session.py --session --text "…" # 指定会话 python wake-session.py --port 59486 --session current --text "…" # 跳过发现（应急） 退出码 ------ 0 成功 ｜ 1 参数错 ｜ 2 前置不满足（目标非 live / 匹配不唯一）｜ 3 发现失败 ｜ 4 投递失败 """ import argparse import json import os import re import subprocess import sys import urllib.error import urllib.request TOKEN_ENV = "CODEBUDDY_GATEWAY_PASSWORD" SESSION_ENV = "CODEBUDDY_SESSION_ID" AUTH_HEADER = "x-access-token" # 唯一实测可用的带法（C1：?password= 在受保护路径上永远失效） MARKERS = ("CodeBuddy Gateway", "CodeBuddy Remote Control") READ_TIMEOUT = 5 # T5：读 ≤5 s WRITE_TIMEOUT = 10 # T5：写 ≤10 s PROBE_TIMEOUT = 1.2 # 具名错误码（⛔ 不把三类失败混成一句 "not found"） E_NO_LISTENER = "no-loopback-listener" E_ENUM_FAILED = "enumeration-failed" E_NOT_GATEWAY = "not-workbuddy-gateway" E_NO_LIVE_MATCH = "no-live-match" E_AMBIGUOUS = "ambiguous-match" E_NOT_LIVE = "target-not-live" def token(): """凭据只从环境读。⛔ 任何日志 / 输出里都不得出现它的值。""" t = os.environ.get(TOKEN_ENV) or "" if not t: die(1, "env-missing", "环境变量 %s 不存在 —— 本程序只接受环境变量取凭据（⛔ 不落盘）" % TOKEN_ENV) return t def die(code, name, msg): print("[FAIL] %s: %s" % (name, msg)) sys.exit(code) # -------------------------------------------------------------------------- # ① 列举本机回环监听口 # -------------------------------------------------------------------------- def listen_ports(): """netstat -ano ⇒ [(port, pid)]，只取 127.0.0.1 上的 LISTENING。 注：Python 的 subprocess 在本沙箱里可以起 netstat（Node 的 spawn 会 EBUSY）。""" try: out = subprocess.run(["netstat", "-ano"], capture_output=True, text=True, timeout=10, encoding="utf-8", errors="replace", creationflags=0x08000000).stdout # CREATE_NO_WINDOW：⛔ 闪窗 except Exception as e: # noqa: BLE001 return None, "%s (%s)" % (E_ENUM_FAILED, e) rows, seen = [], set() for line in out.splitlines(): m = re.match(r"\s*TCP\s+127\.0\.0\.1:(\d+)\s+\S+\s+LISTENING\s+(\d+)", line, re.I) if not m: continue key = (m.group(1), m.group(2)) if key in seen: continue seen.add(key) rows.append((int(m.group(1)), int(m.group(2)))) return rows, None # -------------------------------------------------------------------------- # ② HTTP（纯标准库） # -------------------------------------------------------------------------- def http(port, method, path, body=None, timeout=READ_TIMEOUT, auth=True): """auth=False ⇒ 不带凭据（判指纹时必须不带：带了口令 /api/v1/health 会从 401 变 200）。""" url = "http://127.0.0.1:%d%s" % (port, path) data = None headers = {"Accept": "application/json"} if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" if auth and TIMEOUT_TOKEN[0]: headers[AUTH_HEADER] = TIMEOUT_TOKEN[0] req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "replace") except Exception: # noqa: BLE001 return 0, "" TIMEOUT_TOKEN = [None] # 由 main 填；模块级避免到处传参 def fingerprint(port): """三判据（**全部不带凭据**）：① GET / 含网关标记 ② /api/v1/health = 401 ③ 是 HTTP 服务。 ⚠️ 判指纹必须 **auth=False**：带上 `x-access-token` 时 `/api/v1/health` 会 200。 ⚠️ 这一条也正是「真网关 vs 垫片 20090」的**唯一分水岭** —— 垫片会把根页原样转出去（标记同样命中），但它自己不做鉴权 ⇒ health=200。 """ code, body = http(port, "GET", "/", auth=False) if code != 200 or not any(m in body for m in MARKERS): return False, "root-markers-miss" h, _ = http(port, "GET", "/api/v1/health", auth=False) if h != 401: return False, "health=%s(expect 401)" % h return True, "ok" def live_of(port): """GET /api/v1/sessions/live ⇒ (sessionId, writerOccupied) 或 (None, None)。""" code, body = http(port, "GET", "/api/v1/sessions/live") if code != 200: return None, None, "http=%s" % code try: d = (json.loads(body) or {}).get("data") or {} return d.get("sessionId"), bool(d.get("writerOccupied")), "ok" except Exception as e: # noqa: BLE001 return None, None, "bad-json(%s)" % e # -------------------------------------------------------------------------- # ③ 发现「服务目标会话的那个网关」 # -------------------------------------------------------------------------- def discover(target): ports, err = listen_ports() if ports is None: die(3, err, "无法枚举本机监听口") if not ports: die(3, E_NO_LISTENER, "本机没有任何回环监听口") gateways, matches = [], [] for port, pid in sorted(ports): ok, why = fingerprint(port) if not ok: continue sid, occ, _ = live_of(port) gateways.append({"port": port, "pid": pid, "live": sid, "writerOccupied": occ}) if sid and target and sid == target: matches.append(port) return gateways, matches def cmd_list(): gateways, _ = discover(None) if not gateways: die(3, E_NOT_GATEWAY, "没有端口通过网关指纹（GET / 含 %s ＋ /api/v1/health=401）" % MARKERS[0]) me = os.environ.get(SESSION_ENV) print("本机网关 %d 个（⛔ 端口每次重启都会变，故一律现查）：" % len(gateways)) for g in gateways: flag = "" if me and g["live"] == me: flag = " ← 本会话所在" print(" · 127.0.0.1:%-6d pid=%-7s live=%-38s writerOccupied=%s%s" % (g["port"], g["pid"], g["live"] or "(none)", g["writerOccupied"], flag)) return 0 def resolve_port(args, target): if args.port: ok, why = fingerprint(args.port) if not ok: die(2, E_NOT_GATEWAY, "指定端口 %d 未通过指纹：%s" % (args.port, why)) return args.port gateways, matches = discover(target) if not matches: detail = "；".join(":%d live=%s" % (g["port"], g["live"] or "-") for g in gateways) or "无网关" die(3, E_NO_LIVE_MATCH, "没有网关的 live 会话 == 目标 %s ⇒ 目标当前不是桌面活动会话（%s）" % (target, detail)) if len(matches) > 1: die(3, E_AMBIGUOUS, "多个网关都声称 live==%s：%s" % (target, matches)) return matches[0] # -------------------------------------------------------------------------- # ④ 投递 # -------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(add_help=True) ap.add_argument("--list", action="store_true", help="列出本机网关与各自 live 会话") ap.add_argument("--session", default="current", help="目标会话 id；current = 本会话（默认）") ap.add_argument("--text", default=None, help="投递正文") ap.add_argument("--port", type=int, default=None, help="跳过发现，直连指定网关口（应急）") ap.add_argument("--dry-run", action="store_true", help="只做发现 ＋ 前置判定，不投递") args = ap.parse_args() TIMEOUT_TOKEN[0] = os.environ.get(TOKEN_ENV) or None if args.list: return cmd_list() if not TIMEOUT_TOKEN[0]: die(1, "env-missing", "环境变量 %s 不存在" % TOKEN_ENV) target = os.environ.get(SESSION_ENV, "") if args.session == "current" else args.session if not target: die(1, "no-target", "取不到本会话 id（环境变量 %s）" % SESSION_ENV) port = resolve_port(args, target) sid, occ, why = live_of(port) if sid != target: # T2：只投给「live 且就是目标」的会话 die(2, E_NOT_LIVE, "127.0.0.1:%d 的 live=%s，与目标 %s 不符（%s）" % (port, sid, target, why)) print("[OK] 网关 127.0.0.1:%d ｜ live=%s ｜ writerOccupied=%s" % (port, sid, occ)) if args.dry_run: print("[DRY] 前置全部满足（发现＋live 判定通过），**未投递**。") return 0 if args.text is None: die(1, "no-text", "缺 --text（或加 --dry-run 只做前置判定）") code, body = http(port, "POST", "/api/v1/sessions/%s/reply" % target, body={"text": args.text}, timeout=WRITE_TIMEOUT) print("[POST] /api/v1/sessions/%s/reply -> HTTP %s" % (target, code)) print(" %s" % body[:300].replace("\n", " ")) if code != 200: return 4 print("[DONE] 已投递（⛔ 单次、不重试）。目标会话将在空闲边界收到这条消息。") return 0 if __name__ == "__main__": sys.exit(main())