455 lines
22 KiB
Python
455 lines
22 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,
|
|
"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)
|
|
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} 系列，但当前没有该补丁的精确接口布局。"
|
|
"已阻止协议写入；请导出此报告，在该版本完成一次接口与服务器回执验证后再启用。"
|
|
f"当前已登记适配：{result['supportedClient']}。"
|
|
),
|
|
)
|
|
else:
|
|
result.update(status="unsupported_client_version", label="企业微信版本尚未适配",
|
|
message=f"本机检测到 {result['clientVersion'] or '未知版本'}；当前协议适配范围为 {result['supportedClient']}。安装盘符不影响识别；其他版本需要单独验证发送接口与回执。")
|
|
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)
|
|
return result
|
|
profile = build_for_version(live["clientVersion"])
|
|
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="需要首次发送验证", 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())
|