711 lines
27 KiB
Python
Vendored
711 lines
27 KiB
Python
Vendored
# -*- coding: utf-8 -*-
|
||
"""
|
||
Ananta 注入器（只挂保龄球瓶 + 气球）—— 共享基础设施
|
||
|
||
★ 名字说明：这一份**不是**「统一加载器 AnantaUnifiedLoader」。
|
||
那份是全通道诊断版（A/B/C/D/E/F 都能开），在
|
||
C:\\Game\\Ananta_mainleakflow_keitaro_gg\\AnantaUnifiedLoader；
|
||
本目录是它的**最小副本**，日常跑服只挂 bowling + balloon 两个模块。
|
||
|
||
★ 由 AnantaTrainSeatLoader v5.0.0.4 的 loader_common.py 改写。保留它踩过的所有坑：
|
||
· 优雅停机三阶段（先让 agent 停手 → 等在途 RPC → 卸脚本）
|
||
· 卸载超时**绝不 detach**（半拆的会话会把游戏拖死）
|
||
· attach 走子线程 + 超时 detach（attach 本身可能永久阻塞）
|
||
· 单次 RPC 失败**不拆会话**，交给主循环打印真实原因 + 退避重连
|
||
|
||
★ 与单脚本版的唯一结构性差别：**一个 frida 会话里开多个 script**。
|
||
为什么必须这样（不是合并成一个大 JS）：
|
||
- 每个 agent.js 顶层都是 `var Il2Cpp2;` + 一堆 IIFE，合并会撞全局命名空间；
|
||
- 更致命的是 world_mod.js / gadget_world.js / train_agent.js 都写
|
||
`rpc.exports = { ... }`（**整体覆盖**），合并后只有最后一个模块的导出会活下来；
|
||
- 分成多个 script：各自独立 realm，导出互不覆盖，且仍是**同一个会话**
|
||
（满足「同一时刻只能有一个 frida 会话」这条硬约束）。
|
||
"""
|
||
import sys, os, time, json, subprocess, atexit, ctypes, threading, re
|
||
|
||
for _s in (sys.stdout, sys.stderr):
|
||
try:
|
||
_s.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
_builtin_print = print
|
||
|
||
|
||
def print(*a, **k):
|
||
k["flush"] = True
|
||
try:
|
||
_builtin_print(*a, **k)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
STOP = {"ask": False, "done": False}
|
||
|
||
RPC_TIMEOUT = 25.0
|
||
ATTACH_TIMEOUT = 60.0
|
||
|
||
# ---- 超时治理（2026-09-29）----
|
||
# ⛑ frida 的同步 RPC 一旦卡住，Python 侧**没有任何办法**让那个调用返回：
|
||
# t.join(timeout) 只是「放弃等待」，线程仍卡在 _frida 的阻塞调用里，
|
||
# daemon 线程要到进程退出才回收 ⇒ 主循环每超时一次就永久多一个线程。
|
||
# 治理靠三层，见 Mod._run_guarded 的注释。
|
||
_STUCK_LOCK = threading.Lock()
|
||
_STUCK_THREADS = [] # 只保留**还活着**的卡住线程（每次统计前先清理）
|
||
# 同一个 script 连续超时这么多次 ⇒ 判定该 script 彻底废了，快速失败不再堆新调用
|
||
STUCK_MAX_CONSECUTIVE = 3
|
||
|
||
|
||
def stuck_thread_count():
|
||
"""当前卡在 frida 同步调用里、join 不回来的线程数（供上层观测/自检）。"""
|
||
with _STUCK_LOCK:
|
||
_STUCK_THREADS[:] = [x for x in _STUCK_THREADS if x.is_alive()]
|
||
return len(_STUCK_THREADS)
|
||
|
||
# 这些是「别的独立加载器 exe」。本统一加载器一开，它们就必须全部关闭，
|
||
# 否则同一个游戏进程上会有两个 frida 会话 —— 实测会把游戏带崩。
|
||
OTHER_EXES = [
|
||
"AnantaGymLoader.exe",
|
||
"AnantaTrainSeatLoader.exe",
|
||
"AnantaBowlingLoader.exe",
|
||
"AnantaBalloonLoader.exe",
|
||
"AnantaGadgetLoader.exe",
|
||
"AnantaWorldMod.exe",
|
||
"AnantaPropScanner.exe",
|
||
]
|
||
|
||
|
||
def _install_ctrl_handler():
|
||
if os.name != "nt":
|
||
return False
|
||
try:
|
||
k32 = ctypes.windll.kernel32
|
||
PROT = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_uint)
|
||
|
||
def handler(ev):
|
||
if ev not in (0, 1, 2, 5, 6):
|
||
return False
|
||
STOP["ask"] = True
|
||
try:
|
||
_atexit_close()
|
||
except Exception:
|
||
pass
|
||
for _ in range(280):
|
||
if STOP["done"]:
|
||
break
|
||
k32.Sleep(50)
|
||
return True
|
||
|
||
cb = PROT(handler)
|
||
handler._cb = cb
|
||
k32.SetConsoleCtrlHandler(cb, True)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
HAS_CTRL = _install_ctrl_handler()
|
||
|
||
|
||
def sleep_slice(sec):
|
||
end = time.time() + sec
|
||
while time.time() < end:
|
||
if STOP["ask"]:
|
||
return
|
||
time.sleep(min(0.2, max(0.0, end - time.time())))
|
||
|
||
|
||
def ts():
|
||
return time.strftime("%H:%M:%S")
|
||
|
||
|
||
def _tasklist_rows():
|
||
"""全量 tasklist 一次，返回 [(进程名小写, pid)]。
|
||
|
||
⛑ 为什么不用 `tasklist /FI "IMAGENAME eq xxx"`：
|
||
实测（2026-09-28）游戏是提权启动的，`/FI` 过滤会直接回
|
||
「没有运行的任务匹配指定标准」，而**不带 filter 的全量列表里明明有**。
|
||
所以一律走全量解析，filter 那条路不可信。
|
||
"""
|
||
rows = []
|
||
try:
|
||
out = subprocess.check_output(
|
||
'tasklist /FO CSV /NH', shell=True,
|
||
text=True, encoding="mbcs", errors="replace")
|
||
except Exception:
|
||
return rows
|
||
for line in out.splitlines():
|
||
parts = [p.strip().strip('"') for p in line.split('","')]
|
||
if len(parts) < 2:
|
||
continue
|
||
try:
|
||
rows.append((parts[0].lower(), int(parts[1].replace(",", ""))))
|
||
except ValueError:
|
||
continue
|
||
return rows
|
||
|
||
|
||
def resolve_pid():
|
||
"""找 Ananta.exe 的 pid：全量 tasklist → frida 枚举（两道保险）。"""
|
||
for name, pid in _tasklist_rows():
|
||
if name in ("ananta.exe", "ananta"):
|
||
return pid
|
||
# 回退：连 tasklist 都看不到时（更严格的权限隔离），用 frida 自己枚举
|
||
try:
|
||
import frida
|
||
dev = frida.get_local_device()
|
||
for p in dev.enumerate_processes():
|
||
if p.name.lower() in ("ananta.exe", "ananta"):
|
||
return p.pid
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def count_exe(name):
|
||
n = name.lower()
|
||
return len([1 for nm, _ in _tasklist_rows() if nm == n])
|
||
|
||
|
||
def _pid_alive_win(pid):
|
||
"""OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION)。
|
||
|
||
⛑ 为什么不用 tasklist 做唯一判据：
|
||
tasklist 在某些环境（受限 shell / 沙箱）会返回**空列表**，
|
||
那时 pid_alive 恒 False ⇒ 互斥锁形同虚设（两份加载器照样起得来）。
|
||
PROCESS_QUERY_LIMITED_INFORMATION(0x1000) 自 Vista 起对**任何**进程都放行
|
||
（含提权进程、含别的会话），只要 pid 存在就拿得到句柄 —— 是最可靠的存在性判据。
|
||
"""
|
||
try:
|
||
k32 = ctypes.windll.kernel32
|
||
k32.OpenProcess.restype = ctypes.c_void_p
|
||
k32.OpenProcess.argtypes = [ctypes.c_uint32, ctypes.c_bool, ctypes.c_uint32]
|
||
h = k32.OpenProcess(0x1000, False, int(pid))
|
||
if not h:
|
||
return False
|
||
try:
|
||
# ⛑ 光拿到句柄**不能**算活着：只要还有人握着该进程的句柄，
|
||
# 进程对象就不会销毁，OpenProcess 照样成功（实测踩到）。
|
||
# 退出码 != STILL_ACTIVE(259) ⇒ 它已经死了。
|
||
code = ctypes.c_uint32()
|
||
ok = k32.GetExitCodeProcess(ctypes.c_void_p(h), ctypes.byref(code))
|
||
return bool(ok) and code.value == 259
|
||
finally:
|
||
k32.CloseHandle(ctypes.c_void_p(h))
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
def pid_alive(pid):
|
||
try:
|
||
pid = int(pid)
|
||
except Exception:
|
||
return False
|
||
if _pid_alive_win(pid):
|
||
return True
|
||
for _, p in _tasklist_rows(): # 兜底：万一 OpenProcess 被拦
|
||
if p == pid:
|
||
return True
|
||
return False
|
||
|
||
|
||
# 常驻加载器的互斥锁。⛑ 为什么必须有（2026-09-29）：
|
||
# 两个 frida 会话同时挂在 Ananta.exe 上 = 游戏必崩（历史踩过 3 次）。
|
||
# 以前加载器是手动开的，人不容易开两份；现在改成「一键启动」随游戏自动起，
|
||
# 重复双击的概率大幅上升，所以必须由程序自己挡住。
|
||
# ⛑ 锁文件里存 pid 而不是单纯「文件存在」—— 崩溃退出会留下僵尸锁，
|
||
# 只判存在会永久卡死启动，必须校验 pid 是否真的活着。
|
||
LOCK_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".loader.lock")
|
||
|
||
|
||
def acquire_instance_lock():
|
||
"""拿常驻锁。返回 None = 拿到；返回占用者 pid = 已经有一份在跑，调用方应立刻退出。"""
|
||
try:
|
||
if os.path.exists(LOCK_PATH):
|
||
try:
|
||
old = int((open(LOCK_PATH, encoding="utf-8").read() or "0").strip() or 0)
|
||
except Exception:
|
||
old = 0
|
||
if old and pid_alive(old) and old != os.getpid():
|
||
return old
|
||
except Exception:
|
||
pass
|
||
try:
|
||
with open(LOCK_PATH, "w", encoding="utf-8") as fh:
|
||
fh.write(str(os.getpid()))
|
||
except Exception:
|
||
pass
|
||
atexit.register(release_instance_lock)
|
||
return None
|
||
|
||
|
||
def release_instance_lock():
|
||
"""只删**自己**写的锁，别把别人的删了。"""
|
||
try:
|
||
if not os.path.exists(LOCK_PATH):
|
||
return
|
||
try:
|
||
cur = int((open(LOCK_PATH, encoding="utf-8").read() or "0").strip() or 0)
|
||
except Exception:
|
||
cur = 0
|
||
if cur == os.getpid():
|
||
os.remove(LOCK_PATH)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def other_loader_running():
|
||
"""有没有别的加载器在跑？返回名字，没有返回 None。必须**每一轮**复查。"""
|
||
for e in OTHER_EXES:
|
||
if count_exe(e) > 0:
|
||
return e
|
||
return None
|
||
|
||
|
||
def apply_placeholders(src, root):
|
||
"""把 agent JS 里「由 Python 端注入」的占位符替换掉。
|
||
|
||
⛑ 2026-09-28 才发现的大坑：`modules/world_mod.js:3376` 是
|
||
var CELLS_RAW = __CELLS_JSON__;
|
||
这是一个**原始占位符**，原版靠 `new-worldmod/inject.py` 在注入前
|
||
`src.replace("__CELLS_JSON__", json.dumps(cells))` 换掉。
|
||
我们这里是直接读文件 ⇒ 未定义的标识符 ⇒ 顶层 ReferenceError
|
||
⇒ **整个脚本加载即死**，连它自己的 `Il2Cpp.perform(...)` 引导都跑不到。
|
||
实测：所有历史会话里 `[worldmod]` 的 console 输出 **0 次** —— 通道 B 一直是死的。
|
||
"""
|
||
if "__CELLS_JSON__" not in src:
|
||
return src
|
||
p = os.path.join(root, "cells.json")
|
||
if not os.path.exists(p):
|
||
raise RuntimeError("JS 需要 __CELLS_JSON__，但找不到 %s" % p)
|
||
data = json.loads(open(p, encoding="utf-8").read())
|
||
rows = data.get("cells", data if isinstance(data, list) else [])
|
||
if not rows:
|
||
raise RuntimeError("%s 里没有 cells" % p)
|
||
payload = json.dumps(rows, ensure_ascii=False, separators=(",", ":"))
|
||
return src.replace("__CELLS_JSON__", payload)
|
||
|
||
|
||
# ★★ 气球 side-channel 路径（2026-09-29 修）
|
||
#
|
||
# 病根：`modules/balloon_fix.js:3417` 的默认值是作者自己开发机上的硬编码
|
||
# var PUSH_PATHS = ['D:\\Ananta\\gym_tools\\_balloon_push.json'];
|
||
# 该文件顶部 R146 注释写得很清楚：**「exe 加载器会把这一行整个改写成运行期算出来的
|
||
# 本机多路径」**。原版 `AnantaBalloonLoader.exe` 确实这么干（实机日志里有
|
||
# `agent PUSH_PATHS -> ["C:\\Users\\…\\AnantaBalloon\\balloon_push.json", …]`），
|
||
# 而我们这里是把 js 原样读进来 ⇒ 没有重写 ⇒ agent 永远去读 D:\Ananta\…
|
||
# ⇒ pushN 恒 0 ⇒ 一颗气球都不会被伪造出来。
|
||
#
|
||
# 服务端（GameRouter.Balloon.cs:1140-1159）按顺序写这三条，这里按**同一顺序**给候选：
|
||
# ① %LOCALAPPDATA%\AnantaBalloon\balloon_push.json ← 与安装位置无关，主用
|
||
# ② <服务端 base 目录>\config\balloon_push.json
|
||
# ③ D:\Ananta\gym_tools\_balloon_push.json ← 作者旧位置，兜底
|
||
_BALLOON_PUSH_RE = re.compile(r"var\s+PUSH_PATHS\s*=\s*\[[^\]]*\]\s*;")
|
||
|
||
|
||
def apply_balloon_paths(src, root):
|
||
"""把 balloon_fix.js 里写死的 PUSH_PATHS 换成本机真实候选列表。"""
|
||
if "PUSH_PATHS" not in src:
|
||
return src
|
||
shared = os.path.join(
|
||
os.environ.get("LOCALAPPDATA") or "", "AnantaBalloon", "balloon_push.json")
|
||
cands = []
|
||
for p in (
|
||
shared,
|
||
os.path.join(root, "config", "balloon_push.json"),
|
||
os.path.join(root, "Drmk.Server", "Drmk.App", "bin", "Debug", "net8.0",
|
||
"config", "balloon_push.json"),
|
||
r"D:\Ananta\gym_tools\_balloon_push.json",
|
||
):
|
||
if not p:
|
||
continue
|
||
# 统一成反斜杠（Windows File API 两种都吃，但 JS 字符串里反斜杠是转义符，
|
||
# 必须成对出现 —— 下面 _js_str 会再翻倍一次）
|
||
p = os.path.normpath(p).replace("/", "\\")
|
||
if p not in cands:
|
||
cands.append(p)
|
||
|
||
def _js_str(s):
|
||
# ⛑ 不能用 json.dumps 直接塞：`"C:\Users\..."` 里的 \U / \b 会被 JS 当转义吃
|
||
# 掉（实测 node 打印成 C:UsersNoiro...）。反斜杠必须翻倍。
|
||
return '"' + s.replace("\\", "\\\\") + '"'
|
||
|
||
arr = "[" + ", ".join(_js_str(p) for p in cands) + "]"
|
||
# ⛑⛑ 替换值**必须用函数**给：re.subn 的字符串 repl 会再做一次反斜杠转义处理，
|
||
# 把 `C:\\Users` 压回 `C:\Users`（实测踩到，JS 侧就又变成转义符了）。
|
||
new, n = _BALLOON_PUSH_RE.subn(lambda _m: "var PUSH_PATHS = %s;" % arr, src, count=1)
|
||
if n == 0:
|
||
return src
|
||
return new
|
||
|
||
|
||
class Mod(object):
|
||
"""一个模块 = 一个 frida script。"""
|
||
|
||
def __init__(self, name, path):
|
||
self.name = name
|
||
self.path = path
|
||
self.script = None
|
||
self.ready = False
|
||
self.note = ""
|
||
self.dead = False # 客户端崩了/重启后，旧 script 会永久失效
|
||
self.timeouts = 0 # 本 script 连续超时次数（load() 成功时清零）
|
||
|
||
def load(self, session, quiet=True):
|
||
try:
|
||
src = open(self.path, encoding="utf-8", errors="replace").read()
|
||
except Exception as e:
|
||
self.note = "读文件失败: %s" % e
|
||
return False
|
||
# ★ 占位符替换（__CELLS_JSON__ 等）—— 见 apply_placeholders 的说明
|
||
try:
|
||
src = apply_placeholders(
|
||
src, os.path.dirname(os.path.dirname(os.path.abspath(self.path))))
|
||
except Exception as e:
|
||
self.note = "占位符替换失败: %s" % str(e)[:160]
|
||
return False
|
||
# ★ 气球 side-channel：js 里的 PUSH_PATHS 是作者机器的硬编码，必须按本机改写
|
||
# root = 服务端根目录（…/DRMK-PS_09_HyNB-v0.3）：
|
||
# <root>/tools/AnantaInjector/modules/balloon_fix.js → 往上 4 层
|
||
try:
|
||
src = apply_balloon_paths(
|
||
src, os.path.dirname(os.path.dirname(os.path.dirname(
|
||
os.path.dirname(os.path.abspath(self.path))))))
|
||
except Exception as e:
|
||
self.note = "气球路径改写失败: %s" % str(e)[:160]
|
||
return False
|
||
try:
|
||
# runtime="v8" 是原加载器的写法（frida 17.18 仍接受该参数）。
|
||
# 万一某个 frida 版本把它拿掉了，退回默认运行时——别因为这一个参数整个模块挂掉。
|
||
try:
|
||
sc = session.create_script(src, runtime="v8")
|
||
except TypeError:
|
||
sc = session.create_script(src)
|
||
except Exception as e:
|
||
self.note = "create_script 失败: %s" % str(e)[:120]
|
||
return False
|
||
|
||
box = {"ready": False}
|
||
|
||
def on_message(msg, data):
|
||
try:
|
||
if msg.get("type") == "send":
|
||
p = msg.get("payload")
|
||
if isinstance(p, str):
|
||
try:
|
||
p = json.loads(p)
|
||
except Exception:
|
||
pass
|
||
if isinstance(p, dict) and (p.get("ready") or p.get("abReady")):
|
||
box["ready"] = True
|
||
elif not quiet:
|
||
print(" [%s] %s" % (self.name, str(p)[:200]))
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if msg.get("type") == "error":
|
||
self.note = "script error: %s" % str(msg.get("description"))[:160]
|
||
if not quiet:
|
||
print(" [!] %s 脚本报错: %s" % (self.name, self.note))
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
sc.on("message", on_message)
|
||
sc.load()
|
||
except Exception as e:
|
||
self.note = "load 失败: %s" % str(e)[:160]
|
||
return False
|
||
|
||
# 有些模块不发 ready，给 3 秒宽限后也放行（它们可能只是没实现握手）
|
||
for _ in range(30):
|
||
if box["ready"]:
|
||
break
|
||
time.sleep(0.1)
|
||
self.script = sc
|
||
self.ready = True
|
||
self.dead = False # 重新加载 = 新 script，旧失效标记要清掉
|
||
self.timeouts = 0 # 同上：连续超时计数也归零
|
||
return True
|
||
|
||
def call3(self, name, arg=None, arg2=None, arg3=None, timeout=RPC_TIMEOUT):
|
||
"""三参 RPC（如 gadget_loader 的 findobj <关键字> <maxNodes> <maxDepth>）。
|
||
|
||
不想把 arg2/arg3 混进 call() 的可选语义里（None 与「不传」会分不清），
|
||
所以单独开一个方法，语义明确。
|
||
"""
|
||
if self.script is None:
|
||
raise RuntimeError("模块 %s 未加载" % self.name)
|
||
E = self.script.exports_sync
|
||
fn = getattr(E, name)
|
||
box = {}
|
||
|
||
def run():
|
||
try:
|
||
box["r"] = fn(arg, arg2, arg3)
|
||
except Exception as e:
|
||
box["e"] = e
|
||
|
||
return self._run_guarded(run, box, name, timeout)
|
||
|
||
def _run_guarded(self, run, box, name, timeout):
|
||
"""起线程跑 RPC + 收尾：超时判定 + 识别「客户端崩了」。
|
||
|
||
⛑ 客户端崩掉/重启后旧 script 会变成 'script has been destroyed'，
|
||
不识别的话上层会一直假死在旧 session 上。
|
||
⛑ 线程在这里起（不在调用方），免得哪天又漏掉 t 的定义。
|
||
|
||
⛑⛑ 2026-09-29 修「超时线程泄漏」（原实现只 join、不回收）：
|
||
Python 杀不掉线程，t.join(timeout) 超时后那个调用**仍然卡在 _frida 里**
|
||
⇒ 主循环每超时一次就永久多一个线程，跑久了线程数单调增长。
|
||
三层治理，从强到弱：
|
||
① **best-effort 真取消**：子线程里挂 frida.Cancellable()，超时后 cancel.cancel()。
|
||
只对 frida 内部会检查 cancellable 的同步调用生效（attach/load 这类稳生效，
|
||
exports_sync 的 RPC 不一定）；**不生效也无害**，纯增强。
|
||
② **超时即判该 script 已废**：置 self.dead ⇒ 上层 `ctx.dead` 为真 ⇒ close()
|
||
→ script.unload() → 卡在等待中的 RPC 抛错 → 线程自然退出。
|
||
**这才是真正的回收路径**，也是原来漏掉的一环（原来超时不置 dead，
|
||
于是下一轮继续往同一个废 script 上发调用，每次再留一个线程）。
|
||
⛑ 本副本还有 armedGuard 兜底（bfStatus 连续 60s 拿不到就主动重挂），
|
||
不置 dead 也不会无限假死 —— 但要白等 60 秒，且这 60 秒里每次探测都再留
|
||
一个线程。置 dead 之后走 `unified_loader.py` 的 `ctx.dead` 分支，**立刻**重挂。
|
||
③ **快速失败**：连续超时达 STUCK_MAX_CONSECUTIVE ⇒ 抛错让上层重建，
|
||
不再往废 script 上堆新调用。
|
||
⛑ 日志按规则 110 —— 不打「只打前 N 条」，每次超时都打（量本来就小）。
|
||
"""
|
||
state = {}
|
||
|
||
def runner():
|
||
# cancellable 是线程局部的，必须在**调用线程**里 push 才有效。
|
||
cancel = None
|
||
try:
|
||
import frida as _frida
|
||
cancel = _frida.Cancellable()
|
||
except Exception:
|
||
cancel = None
|
||
if cancel is None:
|
||
run()
|
||
return
|
||
state["cancel"] = cancel
|
||
try:
|
||
with cancel:
|
||
run()
|
||
except Exception as e:
|
||
if "e" not in box:
|
||
box["e"] = e
|
||
|
||
t = threading.Thread(target=runner)
|
||
t.daemon = True
|
||
with _STUCK_LOCK:
|
||
_STUCK_THREADS.append(t)
|
||
t.start()
|
||
t.join(timeout)
|
||
|
||
if t.is_alive():
|
||
cancel = state.get("cancel")
|
||
if cancel is not None:
|
||
try:
|
||
cancel.cancel()
|
||
except Exception:
|
||
pass
|
||
self.timeouts += 1
|
||
self.dead = True # ② 交给上层 close() 回收
|
||
n = stuck_thread_count()
|
||
print("[%s] [!] rpc %s.%s 超时（>%ss）—— 判定该 script 已废，上层会重建；"
|
||
"卡住线程=%d 连续超时=%d" % (ts(), self.name, name, timeout, n, self.timeouts))
|
||
if self.timeouts >= STUCK_MAX_CONSECUTIVE:
|
||
raise RuntimeError("rpc %s.%s 连续超时 %d 次，放弃该 script，等待重建"
|
||
% (self.name, name, self.timeouts))
|
||
raise RuntimeError("rpc %s.%s 超时（>%ss）" % (self.name, name, timeout))
|
||
|
||
# 正常返回：线程已结束，从账本里摘掉
|
||
self.timeouts = 0
|
||
with _STUCK_LOCK:
|
||
if t in _STUCK_THREADS:
|
||
_STUCK_THREADS.remove(t)
|
||
if "e" in box:
|
||
msg = str(box["e"])
|
||
if ("has been destroyed" in msg or "is detached" in msg
|
||
or "invalid session" in msg or "Session is detached" in msg):
|
||
self.dead = True
|
||
raise RuntimeError("rpc %s.%s 失败: %s" % (self.name, name, msg[:160]))
|
||
return box.get("r")
|
||
|
||
def call(self, name, arg=None, arg2=None, timeout=RPC_TIMEOUT):
|
||
"""arg2 用于双参 RPC（如 addbtn 的 abTick(半径, 批量)）。"""
|
||
if self.script is None:
|
||
raise RuntimeError("模块 %s 未加载" % self.name)
|
||
E = self.script.exports_sync
|
||
fn = getattr(E, name)
|
||
box = {}
|
||
|
||
def run():
|
||
try:
|
||
if arg2 is not None:
|
||
box["r"] = fn(arg, arg2)
|
||
elif arg is not None:
|
||
box["r"] = fn(arg)
|
||
else:
|
||
box["r"] = fn()
|
||
except Exception as e:
|
||
box["e"] = e
|
||
|
||
return self._run_guarded(run, box, name, timeout)
|
||
|
||
|
||
class Ctx(object):
|
||
def __init__(self, modules):
|
||
self.modules = modules # [Mod]
|
||
self.session = None
|
||
self.pid = None
|
||
self.closing = False
|
||
self.fail = 0
|
||
|
||
def attach(self):
|
||
box = {}
|
||
|
||
def run():
|
||
try:
|
||
box["r"] = self._attach_impl(box)
|
||
except Exception as e:
|
||
box["e"] = e
|
||
|
||
t = threading.Thread(target=run)
|
||
t.daemon = True
|
||
t.start()
|
||
t.join(ATTACH_TIMEOUT)
|
||
if t.is_alive():
|
||
sess = box.get("sess")
|
||
if sess is not None:
|
||
def _kill():
|
||
try:
|
||
sess.detach()
|
||
except Exception:
|
||
pass
|
||
k = threading.Thread(target=_kill)
|
||
k.daemon = True
|
||
k.start()
|
||
k.join(6)
|
||
return False
|
||
if "e" in box:
|
||
print("[%s] attach 异常：%s" % (ts(), str(box["e"])[:200]))
|
||
return False
|
||
return box.get("r", False)
|
||
|
||
def _attach_impl(self, box=None):
|
||
import frida
|
||
pid = resolve_pid()
|
||
if pid is None:
|
||
return False
|
||
self.pid = pid
|
||
session = frida.attach(pid)
|
||
if box is not None:
|
||
box["sess"] = session
|
||
ok = 0
|
||
for m in self.modules:
|
||
if m.load(session):
|
||
ok += 1
|
||
print("[%s] ✓ %s" % (ts(), m.name))
|
||
else:
|
||
print("[%s] ✗ %s —— %s" % (ts(), m.name, m.note or "未知原因"))
|
||
if ok == 0:
|
||
try:
|
||
session.detach()
|
||
except Exception:
|
||
pass
|
||
return False
|
||
self.session = session
|
||
self.closing = False
|
||
self.fail = 0
|
||
return True
|
||
|
||
@property
|
||
def dead(self):
|
||
"""客户端崩了/重启 ⇒ 任一模块的 script 失效，或 session 已断。"""
|
||
if self.session is None:
|
||
return True
|
||
for m in self.modules:
|
||
if m.script is not None and m.dead:
|
||
return True
|
||
return False
|
||
|
||
def mod(self, name):
|
||
for m in self.modules:
|
||
if m.name == name and m.script is not None:
|
||
return m
|
||
return None
|
||
|
||
def close(self):
|
||
"""优雅停机：① 请各 agent 停手 ② 卸脚本 ③ detach。卸载超时绝不 detach。"""
|
||
if self.closing or self.session is None:
|
||
STOP["done"] = True
|
||
return
|
||
self.closing = True
|
||
t0 = time.time()
|
||
|
||
for m in self.modules:
|
||
# ⛑ 2026-09-29：dead 的 script 再发 stop 只会再超时一次、再留一个卡住的线程
|
||
# （它的 RPC 通道已经废了，发什么都是白发）⇒ 直接跳过，交给 unload 回收。
|
||
if m.script is None or m.dead:
|
||
continue
|
||
# "stop" 是 AnantaGadgetLoader 的停机 RPC（gadget_world.py 的 finally 里调它）
|
||
for stop_name in ("shutdown", "bowlShutdown", "abShutdown", "stop"):
|
||
try:
|
||
m.call(stop_name, timeout=4)
|
||
break
|
||
except Exception:
|
||
continue
|
||
|
||
time.sleep(0.3)
|
||
sess = self.session
|
||
scrs = [m.script for m in self.modules if m.script is not None]
|
||
self.session = None
|
||
for m in self.modules:
|
||
m.script = None
|
||
|
||
def _unload():
|
||
for sc in scrs:
|
||
try:
|
||
sc.unload()
|
||
except Exception:
|
||
pass
|
||
|
||
t1 = threading.Thread(target=_unload)
|
||
t1.daemon = True
|
||
t1.start()
|
||
t1.join(8.0)
|
||
if t1.is_alive():
|
||
print("[!] 脚本卸载没在 8 秒内完成 —— 已跳过强制断开（避免把游戏拖死）。")
|
||
print(" 各 agent 已停机、不再做任何事；重启游戏即可彻底清掉。")
|
||
STOP["done"] = True
|
||
return
|
||
|
||
def _det():
|
||
try:
|
||
if sess:
|
||
sess.detach()
|
||
except Exception:
|
||
pass
|
||
|
||
t2 = threading.Thread(target=_det)
|
||
t2.daemon = True
|
||
t2.start()
|
||
t2.join(3.0)
|
||
print("[%s] 注入已安全卸载，用时 %.1fs" % (ts(), time.time() - t0))
|
||
STOP["done"] = True
|
||
|
||
|
||
_LIVE = []
|
||
_CLOSE_LOCK = threading.Lock()
|
||
|
||
|
||
def _atexit_close():
|
||
with _CLOSE_LOCK:
|
||
for c in list(_LIVE):
|
||
try:
|
||
c.close()
|
||
except Exception:
|
||
pass
|
||
STOP["done"] = True
|