admin
19101acd65
- 按用户指示清空原有 25 技能内容，只提交 session-mechanism（57 文件） - 附 .gitignore（产物 + 本机凭据） - 令牌明文已脱敏（历史 .neodata_token 与 pitfalls 引用均不入库） - 本提交为孤儿提交（父提交为空），历史自此重新开始
257 lines
11 KiB
Python
257 lines
11 KiB
Python
# -*- 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 <uuid> --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())
|