NIKKE 启动器 CDN 投毒：一次只改四个 JS 文件就落地的远控链

// verify-asset.mjs —— 校验 CDN 上的可执行资产是否与构建清单一致
// 用法：node verify-asset.mjs manifest.json
// 清单：[{ "url": "...", "sha256": "<64 位小写十六进制>" }]
import { createHash } from 'node:crypto'
import { readFile } from 'node:fs/promises'

const TIMEOUT_MS = 15_000

async function fetchBytes(url) {
  const ctrl = new AbortController()
  const timer = setTimeout(() => ctrl.abort(), TIMEOUT_MS)
  try {
    // no-cache 让边缘回源，避免拿到还没清的旧副本
    const res = await fetch(url, {
      signal: ctrl.signal,
      headers: { 'user-agent': 'asset-audit/1.0', 'cache-control': 'no-cache' },
    })
    if (!res.ok) throw new Error(`HTTP ${res.status} ${res.statusText}`)
    return Buffer.from(await res.arrayBuffer())
  } finally {
    clearTimeout(timer)
  }
}

const sha256 = (buf) => createHash('sha256').update(buf).digest('hex')

const manifest = JSON.parse(await readFile(process.argv[2], 'utf8'))
let bad = 0
for (const { url, sha256: expected } of manifest) {
  try {
    const buf = await fetchBytes(url)
    const actual = sha256(buf)
    const ok = actual === expected.toLowerCase()
    console.log(`${ok ? 'OK  ' : 'FAIL'} ${buf.length}B ${actual} ${url}`)
    if (!ok) { bad++; console.error(`     期望 ${expected}\n     实际 ${actual}`) }
  } catch (err) {
    bad++
    console.error(`FAIL 抓取失败 ${url}: ${err.message}`)
  }
}
// 非零退出码可直接挂到 CI 或定时任务，失败即阻断发布或告警
if (bad > 0) { console.error(`\n${bad} 个资产不一致`); process.exit(1) }
console.log('\n全部一致')

#!/usr/bin/env python3
"""离线排查启动器是否被植入 loader / 远控代理（只读，不发网络请求）"""
import argparse, gzip, os, re, sys

# 回退域名 insightjs.org 由 String.fromCharCode 拼出，明文域名搜不到
SIG = b"String.fromCharCode(111,114,103)"
LOGS = (b"k2f_chan registered", b"lua downloaded to log\\tinyget")
STATE = b"k2f.agent.state"
DIRS = ("nikke_launcher", "nikke_launcher_hmt", "nikke_launcher_google")

def read_tail(path, limit=200 * 1024):
    with open(path, "rb") as f:              # 只读，别先删缓存再查
        f.seek(max(0, os.path.getsize(path) - limit))
        return f.read()

def scan_log(path):
    buf, hits = read_tail(path), []
    for hint in LOGS:
        if hint in buf:
            for line in buf.splitlines():
                if hint in line and line not in hits:
                    hits.append(line)
    # 日志里可能夹 token / cookie，对外发之前必须打码
    return [(path, re.sub(rb"[A-Za-z0-9_.-]{24,}", b"<redacted>", h)[:160]) for h in hits]

def scan_gzip(paths):
    """缓存按 gzip 原样落盘，明文 grep 必漏，必须先解压"""
    found = []
    for path in paths:
        try:
            if os.path.getsize(path) > 8 * 1024 * 1024:
                continue
            buf = read_tail(path, 8 * 1024 * 1024)
            start = buf.find(b"\x1f\x8b\x08")     # gzip 魔数，前面可能是 HTTP 头
            if start < 0:
                continue
            if SIG in gzip.decompress(buf[start:]):
                found.append(path)
        except (OSError, EOFError):
            continue                              # 尾部块头残留解压失败就跳过
    return found

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--prefix", help="Wine/Proton 前缀")
    ap.add_argument("--appdata", help="启动器数据目录")
    args = ap.parse_args()

    roots = [args.appdata] if args.appdata else []
    if args.prefix:
        base = os.path.join(args.prefix, "drive_c", "users")
        if os.path.isdir(base):
            roots += [os.path.join(base, u, "AppData", "Roaming") for u in os.listdir(base)]
    if not roots:
        ap.error("至少给一个 --prefix 或 --appdata")

    log, state, cache = [], [], []
    for root in roots:
        for d in DIRS:
            base = os.path.join(root, d)
            if not os.path.isdir(base):
                continue
            for rel in ("Launcher/intl_service/debug.log", "intl_service/debug.log"):
                if os.path.isfile(os.path.join(base, rel)):
                    log += scan_log(os.path.join(base, rel))
            ls = os.path.join(base, "tbs_cache", "Local Storage", "leveldb")
            if os.path.isdir(ls):
                state += [os.path.join(ls, n) for n in os.listdir(ls)
                          if os.path.isfile(os.path.join(ls, n))
                          and STATE in read_tail(os.path.join(ls, n))]
            cdir = os.path.join(base, "tbs_cache", "Cache")
            if os.path.isdir(cdir):
                cache += scan_gzip([os.path.join(cdir, n) for n in os.listdir(cdir)])

    for name, items in (("log", log), ("state", state), ("cache", cache)):
        print(f"{name:6s} {len(items)} 条")
        for item in items[:5]:
            print("   ", item)
    print("\n结论：" + ("命中感染痕迹" if (log or state or cache) else "未发现痕迹（缓存已删则无法回溯）"))
    return 1 if (log or state or cache) else 0

if __name__ == "__main__":
    sys.exit(main())