472 lines
24 KiB
Python
472 lines
24 KiB
Python
"""Read-only inspection of the installed WeCom client; never loads client DLLs."""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from contextlib import contextmanager
|
||
import hashlib
|
||
import json
|
||
import mmap
|
||
import os
|
||
import platform
|
||
from pathlib import Path
|
||
import re
|
||
import struct
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
|
||
MARKERS = (
|
||
b"SendTextMessage", b"sendTextMessage", b"SendText", b"sendText",
|
||
b"SendMessage", b"sendMessage", b"SendMsg", b"sendMsg",
|
||
b"SendMsgReq", b"SendMsgRsp", b"SendMsgResp",
|
||
b"MessageService", b"MsgService", b"SendMessageService",
|
||
b"WM_COPYDATA", b"CreateNamedPipe", b"send_chat_message",
|
||
b"WeClawMcpToolManager", b"SendTextToClawConversation",
|
||
)
|
||
TARGET_MODULES = ("WXWorkCommand.exe", "ilink2.dll", "native.dll",
|
||
"tencent.mm.pbc.dll", "ilink_live_lite_wrapper.dll")
|
||
MAX_EXPORTS = 20000
|
||
|
||
|
||
def pe_header(path):
|
||
with open(path, "rb") as handle:
|
||
if handle.read(2) != b"MZ":
|
||
raise ValueError("不是 Windows PE 文件")
|
||
handle.seek(0x3c)
|
||
raw = handle.read(4)
|
||
if len(raw) != 4:
|
||
raise ValueError("PE 头不完整")
|
||
offset = struct.unpack("<I", raw)[0]
|
||
if not 64 <= offset <= 16 * 1024 * 1024:
|
||
raise ValueError("PE 头偏移无效")
|
||
handle.seek(offset)
|
||
raw = handle.read(24)
|
||
if len(raw) != 24 or raw[:4] != b"PE\x00\x00":
|
||
raise ValueError("PE 签名无效")
|
||
machine, sections, stamp, _, _, _, characteristics = struct.unpack("<HHIIIHH", raw[4:])
|
||
return {
|
||
"architecture": {0x14c: "x86", 0x8664: "x64", 0xaa64: "arm64"}.get(machine, "unknown"),
|
||
"machine": hex(machine), "sections": sections, "buildTimestamp": stamp,
|
||
"isDll": bool(characteristics & 0x2000),
|
||
}
|
||
|
||
|
||
def file_version(path):
|
||
if os.name != "nt":
|
||
return ""
|
||
import win32api
|
||
info = win32api.GetFileVersionInfo(str(path), "\\")
|
||
ms, ls = info["FileVersionMS"], info["FileVersionLS"]
|
||
return ".".join(str(value) for value in (ms >> 16, ms & 65535, ls >> 16, ls & 65535))
|
||
|
||
|
||
def running_clients():
|
||
"""Read process image paths with PROCESS_QUERY_LIMITED_INFORMATION only."""
|
||
if os.name != "nt":
|
||
return []
|
||
import ctypes
|
||
from ctypes import wintypes
|
||
from wxwork_key import find_wxwork_pids
|
||
|
||
kernel = ctypes.WinDLL("kernel32", use_last_error=True)
|
||
kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
|
||
kernel.OpenProcess.restype = wintypes.HANDLE
|
||
kernel.QueryFullProcessImageNameW.argtypes = [
|
||
wintypes.HANDLE, wintypes.DWORD, wintypes.LPWSTR, ctypes.POINTER(wintypes.DWORD)]
|
||
kernel.QueryFullProcessImageNameW.restype = wintypes.BOOL
|
||
kernel.CloseHandle.argtypes = [wintypes.HANDLE]
|
||
kernel.CloseHandle.restype = wintypes.BOOL
|
||
result = []
|
||
binary_metadata = {}
|
||
for pid in find_wxwork_pids(include_web=False, include_extra=False):
|
||
process = kernel.OpenProcess(0x1000, False, pid)
|
||
if not process:
|
||
result.append({"pid": pid, "path": "", "error": "进程信息不可读取"})
|
||
continue
|
||
try:
|
||
buffer = ctypes.create_unicode_buffer(32768)
|
||
length = wintypes.DWORD(len(buffer))
|
||
if not kernel.QueryFullProcessImageNameW(process, 0, buffer, ctypes.byref(length)):
|
||
result.append({"pid": pid, "path": "", "error": "进程路径不可读取"})
|
||
continue
|
||
path = buffer.value
|
||
entry = {"pid": pid, "path": path}
|
||
try:
|
||
identity = os.path.normcase(os.path.abspath(path))
|
||
metadata = binary_metadata.get(identity)
|
||
if metadata is None:
|
||
metadata = {**pe_header(path), "version": file_version(path), "size": Path(path).stat().st_size}
|
||
binary_metadata[identity] = metadata
|
||
entry.update(metadata)
|
||
except Exception as exc:
|
||
entry["error"] = type(exc).__name__
|
||
result.append(entry)
|
||
finally:
|
||
kernel.CloseHandle(process)
|
||
return result
|
||
|
||
|
||
def inspect_binary(path, *, deep=False):
|
||
path = Path(path)
|
||
info = {"name": path.name, "path": str(path), "size": path.stat().st_size,
|
||
**pe_header(path)}
|
||
try:
|
||
info["version"] = file_version(path)
|
||
except Exception:
|
||
info["version"] = ""
|
||
if not deep:
|
||
return info
|
||
with path.open("rb") as handle:
|
||
digest = hashlib.sha256()
|
||
while chunk := handle.read(1024 * 1024):
|
||
digest.update(chunk)
|
||
info["sha256"] = digest.hexdigest()
|
||
handle.seek(0)
|
||
with mmap.mmap(handle.fileno(), 0, access=mmap.ACCESS_READ) as data:
|
||
found = []
|
||
for marker in MARKERS:
|
||
for encoding, needle in (("ascii", marker), ("utf16le", marker.decode("ascii").encode("utf-16le"))):
|
||
position, offsets = 0, []
|
||
while len(offsets) < 12:
|
||
position = data.find(needle, position)
|
||
if position < 0:
|
||
break
|
||
offsets.append(position)
|
||
position += len(needle)
|
||
if offsets:
|
||
found.append({"symbol": marker.decode("ascii"), "encoding": encoding, "fileOffsets": offsets})
|
||
info["markers"] = found
|
||
try:
|
||
import pefile
|
||
with pefile.PE(str(path), fast_load=True) as pe:
|
||
pe.parse_data_directories(directories=[
|
||
pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"],
|
||
pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_IMPORT"]])
|
||
exports = list(getattr(getattr(pe, "DIRECTORY_ENTRY_EXPORT", None), "symbols", []))
|
||
info["exportCount"] = len(exports)
|
||
relevant = re.compile(r"send|recv|receive|message|task|init|connect|create|ipc|command", re.I)
|
||
info["relevantExports"] = [
|
||
{"name": item.name.decode("ascii", "replace")[:240], "rva": item.address}
|
||
for item in exports[:MAX_EXPORTS]
|
||
if item.name and relevant.search(item.name.decode("ascii", "replace"))
|
||
][:80]
|
||
info["imports"] = [item.dll.decode("ascii", "replace")
|
||
for item in getattr(pe, "DIRECTORY_ENTRY_IMPORT", [])]
|
||
except ImportError:
|
||
info["exportInspection"] = "未安装 pefile，仅完成 PE 头与标记检查"
|
||
except Exception as exc:
|
||
info["exportInspection"] = "导出表分析失败：" + type(exc).__name__
|
||
return info
|
||
|
||
|
||
def bundled_agent_info(executable, version):
|
||
"""Inspect packaged SDK files, without executing the agent or importing its code."""
|
||
root = Path(executable).parent / version / "WeComAgent"
|
||
sdk = root / "python" / "3.10" / "site-packages" / "wecomagent"
|
||
info = {
|
||
"installed": (root / "wecom-agent.exe").is_file(),
|
||
"pythonSdkInstalled": (sdk / "_rpc.py").is_file(),
|
||
"rpcMethod": "", "requiresRuntimeAuthorization": False,
|
||
"externalChatSendVerified": False,
|
||
}
|
||
if info["pythonSdkInstalled"]:
|
||
try:
|
||
# Bound reads; no credential/configuration files are accessed.
|
||
with (sdk / "_rpc.py").open(encoding="utf-8") as handle:
|
||
source = handle.read(128 * 1024)
|
||
info["requiresRuntimeAuthorization"] = all(
|
||
marker in source for marker in
|
||
("WECOMAGENT_RPC_HOST", "WECOMAGENT_RPC_PORT", "WECOMAGENT_RPC_TOKEN"))
|
||
if 'wecom.exec' in source:
|
||
info["rpcMethod"] = "wecom.exec"
|
||
except (OSError, UnicodeError):
|
||
info["inspectionError"] = "内置 SDK 文件不可读"
|
||
return info
|
||
|
||
|
||
@contextmanager
|
||
def _probe_phase(timings, name):
|
||
started=time.perf_counter()
|
||
try:yield
|
||
finally:timings[name]=round((time.perf_counter()-started)*1000,3)
|
||
|
||
|
||
def probe_installed_client(*, deep=False, initialize_database=False):
|
||
"""Run fresh identity checks; static build hashes live for this call only."""
|
||
from wecom_native_sender import readonly_build_validation_scope
|
||
started=time.perf_counter()
|
||
timings={}
|
||
with readonly_build_validation_scope() as builds:
|
||
result=_probe_installed_client(deep=deep,initialize_database=initialize_database,timings=timings)
|
||
timings['total']=round((time.perf_counter()-started)*1000,3)
|
||
result['timingsMs']=timings
|
||
result['buildValidation']={key:builds[key] for key in ('hashedFiles','reusedFiles','bytesHashed')}
|
||
return result
|
||
|
||
|
||
def _probe_installed_client(*, deep=False, initialize_database=False, timings):
|
||
"""File/process presence is deliberately not treated as a login or send proof."""
|
||
from wecom_native_builds import (
|
||
KNOWN_VERSION_FAMILIES, build_for_hash, build_for_version,
|
||
classify_version, client_support_text, sha256_file,
|
||
)
|
||
result = {
|
||
"schemaVersion": 3, "checkedAt": time.time(), "readOnly": not initialize_database,
|
||
"status": "not_checked", "label": "待检测", "message": "",
|
||
"canSend": False, "loginVerified": False, "adapterInstalled": False,
|
||
"sendVerified": False, "receiptVerified": False,
|
||
"clients": [], "binaries": [], "bundledAgents": [], "errors": [],
|
||
"versionDetection": [], "clientVersionFamily": "",
|
||
"buildFingerprints": [], "buildMatchMode": "",
|
||
"needsAdapterUpdate": False, "autoAdapted": False, "adaptationErrors": [],
|
||
"supportedClient": client_support_text(),
|
||
"host": {"os": platform.system(), "release": platform.release(),
|
||
"machine": platform.machine(), "processBits": struct.calcsize("P") * 8},
|
||
"databaseReady": False, "verificationScope": "adapter_version",
|
||
"targetSendVerified": False, "localSendVerified": False, "needsSendValidation": False,
|
||
}
|
||
if os.name != "nt":
|
||
result.update(status="unsupported_platform", label="不支持当前平台",
|
||
message="原账号客户端协议模式需要 Windows 企业微信。")
|
||
return result
|
||
try:
|
||
with _probe_phase(timings,"processes"):
|
||
result["clients"] = running_clients()
|
||
except Exception as exc:
|
||
result.update(status="probe_failed", label="检测失败", message="无法读取企业微信进程信息。")
|
||
result["errors"].append(type(exc).__name__)
|
||
return result
|
||
if not result["clients"]:
|
||
from wecom_environment import installed_executables
|
||
for path in installed_executables():
|
||
try:
|
||
result["binaries"].append(inspect_binary(path))
|
||
except (OSError, ValueError) as exc:
|
||
result["errors"].append(type(exc).__name__)
|
||
result.update(status="client_not_running", label="客户端未运行",
|
||
clientInstallPath=" / ".join(item["path"] for item in result["binaries"]),
|
||
message="请先启动并登录本机企业微信，再检测。安装路径会自动识别，无需安装在 C 盘。")
|
||
return result
|
||
paths = sorted({item.get("path") for item in result["clients"] if item.get("path")})
|
||
result["clientInstallPath"] = " / ".join(paths)
|
||
seen = set()
|
||
for path in paths:
|
||
version = next((item.get("version") for item in result["clients"] if item.get("path") == path), "")
|
||
if version:
|
||
result["bundledAgents"].append(bundled_agent_info(path, version))
|
||
targets = [Path(path)]
|
||
if deep and version:
|
||
folder = Path(path).parent / version
|
||
targets.extend(folder / name for name in TARGET_MODULES if (folder / name).is_file())
|
||
for target in targets:
|
||
if str(target) in seen:
|
||
continue
|
||
seen.add(str(target))
|
||
try:
|
||
metadata=next((item for item in result['clients'] if item.get('path')==str(target)),{})
|
||
if not deep and all(key in metadata for key in ('size','architecture','machine','sections','buildTimestamp','isDll','version')):
|
||
result['binaries'].append({'name':target.name,'path':str(target),**{
|
||
key:metadata[key] for key in ('size','architecture','machine','sections','buildTimestamp','isDll','version')}})
|
||
else:
|
||
result["binaries"].append(inspect_binary(target, deep=deep))
|
||
except Exception as exc:
|
||
result["errors"].append(f"{target.name}: {type(exc).__name__}")
|
||
versions = sorted({item.get("version") for item in result["clients"] if item.get("version")})
|
||
detections = [classify_version(value) for value in versions]
|
||
families = sorted({item["family"] for item in detections if item.get("family")})
|
||
result["versionDetection"] = detections
|
||
result["clientVersionFamily"] = " / ".join(families)
|
||
result["clientVersion"] = " / ".join(versions)
|
||
result["architecture"] = " / ".join(sorted({
|
||
item.get("architecture") for item in result["clients"] if item.get("architecture")}))
|
||
# SDK metadata is optional. This native adapter does not depend on a WeClaw runtime.
|
||
if deep:
|
||
from wecom_sdk_client import probe_sdk_runtime
|
||
try:
|
||
result["sdkRuntime"] = probe_sdk_runtime()
|
||
except Exception as exc:
|
||
result["sdkRuntime"] = {"status": "probe_failed", "rpcVerified": False, "error": type(exc).__name__}
|
||
from wecom_native_protocol import load_native_trial
|
||
trial = load_native_trial(client_versions=versions)
|
||
if trial:
|
||
result["nativeTrial"] = trial
|
||
if not paths:
|
||
result.update(status="probe_failed", label="进程信息不可读",
|
||
message="检测到企业微信进程，但无法读取程序路径与版本。请确认本程序与企业微信在同一 Windows 用户会话内、以相同权限运行。")
|
||
return result
|
||
from wecom_native_sender import discover, NativeUnavailable
|
||
version_matches = [item for item in result["clients"] if build_for_version(item.get("version")) is not None]
|
||
# The version resource is only a hint. Hash an unregistered image once so
|
||
# a localized/malformed version string cannot prevent an exact known build
|
||
# from being selected. A matching hash is still checked again by
|
||
# NativeClient against the live process before any write is possible.
|
||
hash_matches = []
|
||
fingerprint_paths = set()
|
||
if not version_matches:
|
||
for item in result["clients"]:
|
||
path = item.get("path")
|
||
if not path:
|
||
continue
|
||
identity = os.path.normcase(os.path.abspath(path))
|
||
if identity in fingerprint_paths:
|
||
continue
|
||
fingerprint_paths.add(identity)
|
||
try:
|
||
digest = sha256_file(path)
|
||
except (OSError, ValueError, TypeError):
|
||
continue
|
||
match = build_for_hash(digest)
|
||
if match is None and item.get("architecture") == "x86":
|
||
from wecom_native_adaptation import resolve_build, AdaptationUnavailable
|
||
try:
|
||
match = resolve_build(path, digest)
|
||
except (AdaptationUnavailable, OSError, ValueError) as exc:
|
||
result["adaptationErrors"].append(str(exc))
|
||
result["buildFingerprints"].append({
|
||
"path": str(path), "sha256": digest,
|
||
"adapterId": match.adapter_id if match is not None else "",
|
||
})
|
||
if match is not None:
|
||
hash_matches.append(item)
|
||
if hash_matches:
|
||
version_matches = hash_matches
|
||
result["buildMatchMode"] = "sha256"
|
||
if not version_matches:
|
||
known_families = sorted({
|
||
detection.get("family") for detection in detections
|
||
if detection.get("state") == "known_family_new_patch"
|
||
and detection.get("family") in KNOWN_VERSION_FAMILIES
|
||
})
|
||
if known_families:
|
||
family_text = "、".join(known_families)
|
||
result.update(
|
||
status="known_client_family_requires_adapter",
|
||
label="新版接口自动适配未通过",
|
||
needsAdapterUpdate=True,
|
||
message=(
|
||
f"已自动识别本机企业微信 {result['clientVersion'] or '未知版本'} "
|
||
f"属于 {family_text} 系列，已尝试自动定位接口，但接口结构未通过完整校验。"
|
||
"请导出此报告以更新适配规则。"
|
||
+ ("检测详情：" + "；".join(result["adaptationErrors"]) + "。" if result["adaptationErrors"] else "")
|
||
),
|
||
)
|
||
else:
|
||
result.update(status="unsupported_client_version", label="企业微信版本尚未适配",
|
||
needsAdapterUpdate=True,
|
||
message=f"本机检测到 {result['clientVersion'] or '未知版本'}；自动接口适配未通过。请导出报告更新适配规则。" + "；".join(result["adaptationErrors"]))
|
||
return result
|
||
if not result.get("buildMatchMode"):
|
||
result["buildMatchMode"] = "version"
|
||
if not any(item.get("architecture") == "x86" for item in version_matches):
|
||
result.update(status="unsupported_client_architecture", label="企业微信架构尚未适配",
|
||
message=f"本机企业微信为 {result['architecture'] or '未知架构'}；协议适配器目前要求 x86 企业微信（可运行在 64 位 Windows 上）。")
|
||
return result
|
||
try:
|
||
with _probe_phase(timings,"clientValidation"):
|
||
live = discover(result["clients"])
|
||
except (NativeUnavailable, OSError, ValueError) as exc:
|
||
error = str(exc)
|
||
status, label = "native_unavailable", "原账号协议未就绪"
|
||
if any(term in error for term in ("权限", "不可读取", "无法访问")):
|
||
status, label = "client_access_denied", "企业微信进程无法读取"
|
||
error += "。请确认两者在同一 Windows 用户会话内、以相同权限运行。"
|
||
elif any(term in error for term in ("文件未适配", "入口校验", "接口不匹配", "自动适配未通过", "接口与文件不一致")):
|
||
status, label = "client_build_mismatch", "企业微信构建尚未适配"
|
||
elif "多个企业微信" in error:
|
||
status, label = "multiple_clients", "存在多个已登录主客户端"
|
||
elif any(term in error for term in ("未登录", "尚未就绪", "状态已变化")):
|
||
status, label = "client_not_ready", "等待企业微信登录完成"
|
||
result.update(status=status, label=label, message=error, nativeAdapterError=error,
|
||
needsAdapterUpdate=status == "client_build_mismatch")
|
||
return result
|
||
profile = build_for_hash(live.get("clientSha256"))
|
||
if profile is None:
|
||
result.update(status="client_build_mismatch", label="客户端构建未通过核对",
|
||
message="无法核对检测到的客户端构建，请重新检测。", needsAdapterUpdate=True)
|
||
return result
|
||
result.update(clientVersion=profile.version, autoAdapted=profile.automatic,
|
||
buildMatchMode="structural" if profile.automatic else result["buildMatchMode"])
|
||
if profile.automatic:
|
||
result["buildFingerprints"] = [{"sha256": profile.sha256, "adapterId": profile.adapter_id}]
|
||
|
||
from wecom_native_evidence import local_send_receipt
|
||
local_receipt = local_send_receipt(profile, live["accountId"]) if not profile.verified else None
|
||
verified = bool(profile.verified or local_receipt)
|
||
result.update(nativeAdapter=profile.adapter_id, nativeAccount=live["accountId"], clientSha256=profile.sha256,
|
||
localSendVerified=bool(local_receipt), verificationScope="current_account" if local_receipt else "adapter_version",
|
||
loginVerified=True, adapterInstalled=True, sendVerified=verified, receiptVerified=verified)
|
||
from wecom_environment import protocol_database_status
|
||
with _probe_phase(timings,"databaseValidation"):
|
||
database = protocol_database_status(live["accountId"], initialize=initialize_database)
|
||
result.update(database=database, databaseReady=database["ready"])
|
||
if not database["ready"]:
|
||
result.update(status=database["status"], label=database["label"], message=database["message"])
|
||
return result
|
||
# Recheck identity after a possibly slow first-install initialization.
|
||
try:
|
||
with _probe_phase(timings,"identityRecheck"):
|
||
current = discover()
|
||
except (NativeUnavailable, OSError, ValueError) as exc:
|
||
result.update(status="client_not_ready", label="企业微信状态已变化", loginVerified=False,
|
||
message="初始化后无法核对当前登录账号：" + str(exc))
|
||
return result
|
||
if any(current.get(key) != live.get(key) for key in ("pid", "accountId", "clientVersion", "clientSha256")):
|
||
result.update(status="client_not_ready", label="登录账号已变化", loginVerified=False,
|
||
message="初始化期间企业微信进程或登录账号发生变化，请重新检测。")
|
||
return result
|
||
if not verified:
|
||
result.update(status="native_validation_required", label="自动适配完成，待首次发送验证" if profile.automatic else "需要首次发送验证", needsSendValidation=True,
|
||
message=f"已自动匹配 {profile.version} 的接口并核对当前账号和消息库。点击本页首次发送验证，取得真实服务器回执后自动启用此账号的协议回复。")
|
||
return result
|
||
result.update(status="native_ready", label="原账号协议环境已就绪", canSend=True, checkedAt=time.time(),
|
||
message=f"已自动选择 {profile.version} 对应接口，当前账号、客户端构建和消息库均已核对，可启动协议自动回复。" + ("本机当前账号已有首次发送及回执实测记录。" if local_receipt else "适配版本已有发送及回执实测记录。") + "本次环境检测未发送消息。")
|
||
return result
|
||
|
||
|
||
def export_probe_report(report, path):
|
||
"""Export only diagnostics; no account directory, chat text, or login keys."""
|
||
target = Path(path)
|
||
content = json.dumps(report, ensure_ascii=False, indent=2, allow_nan=False)
|
||
# Atomic export leaves an existing report intact if writing fails.
|
||
temporary = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile(
|
||
"w", encoding="utf-8", dir=target.resolve().parent,
|
||
prefix=".protocol-report-", suffix=".tmp", delete=False,
|
||
) as handle:
|
||
temporary = Path(handle.name)
|
||
handle.write(content + "\n")
|
||
os.replace(temporary, target)
|
||
finally:
|
||
if temporary is not None:
|
||
temporary.unlink(missing_ok=True)
|
||
return target
|
||
|
||
|
||
def run_packaged_diagnostics(argv):
|
||
"""Run before GUI/login initialization; metadata only, never reads chat content."""
|
||
parser = argparse.ArgumentParser(description="导出企业微信协议诊断（不启动监听）")
|
||
parser.add_argument("--protocol-diagnostics", type=Path, required=True, metavar="REPORT.json")
|
||
args = parser.parse_args(argv)
|
||
report = probe_installed_client(deep=True)
|
||
export_probe_report(report, args.protocol_diagnostics)
|
||
return 1 if report["status"] == "probe_failed" else 0
|
||
|
||
|
||
def main():
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--deep", action="store_true", help="静态分析已安装程序的导出表和符号标记")
|
||
parser.add_argument("--output", type=Path)
|
||
args = parser.parse_args()
|
||
report = probe_installed_client(deep=args.deep)
|
||
if args.output:
|
||
export_probe_report(report, args.output)
|
||
print(json.dumps(report, ensure_ascii=False, indent=2))
|
||
return 1 if report["status"] == "probe_failed" else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|