290 lines
16 KiB
Python
290 lines
16 KiB
Python
"""Passive Windows process/network metadata monitor for WeCom.
|
|
No process memory reads, injection, packet capture, credentials, or message sends.
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import ctypes as c
|
|
from ctypes import wintypes as w
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import struct
|
|
import threading
|
|
import time
|
|
import uuid
|
|
|
|
TARGET_NAMES = frozenset({'wxwork.exe', 'wecom-agent.exe', 'wironboxd32.exe', 'wironboxd64.exe'})
|
|
MODULE_MARKERS = ('wecom', 'weclaw', 'ironbox', 'wxworkcommand', 'ilink', 'native.dll')
|
|
TCP_STATES = {1:'CLOSED',2:'LISTEN',3:'SYN_SENT',4:'SYN_RECEIVED',5:'ESTABLISHED',6:'FIN_WAIT_1',7:'FIN_WAIT_2',8:'CLOSE_WAIT',9:'CLOSING',10:'LAST_ACK',11:'TIME_WAIT',12:'DELETE_TCB'}
|
|
DEFAULT_OUTPUT = Path(__file__).resolve().parent / 'logs' / 'wecom_process_monitor'
|
|
|
|
def utc_now():
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
def process_identity(item):
|
|
return (item['pid'], item.get('createdAt'))
|
|
|
|
def changes(previous, current):
|
|
"""Do not infer exits from a failed snapshot; callers retain the last good one."""
|
|
result = []
|
|
for field, started, stopped, key in (
|
|
('processes','process_started','process_stopped',process_identity),
|
|
('connections','connection_observed','connection_gone',lambda x:json.dumps(x,sort_keys=True)),
|
|
('modules','module_observed','module_gone',lambda x:(x['pid'],x['path'].lower())),
|
|
):
|
|
before = {key(x):x for x in previous.get(field, [])}
|
|
after = {key(x):x for x in current.get(field, [])}
|
|
result.extend({'type':stopped,'item':v} for k,v in before.items() if k not in after)
|
|
result.extend({'type':started,'item':v} for k,v in after.items() if k not in before)
|
|
return result
|
|
|
|
def atomic_json(path, value):
|
|
# Windows readers/antivirus can temporarily deny replacement of an open file.
|
|
temporary = path.with_name(path.name + '.tmp-' + uuid.uuid4().hex)
|
|
try:
|
|
temporary.write_text(json.dumps(value,ensure_ascii=False,indent=2),encoding='utf-8')
|
|
for attempt in range(21):
|
|
try:
|
|
os.replace(temporary,path)
|
|
return
|
|
except PermissionError:
|
|
if attempt == 20:
|
|
raise
|
|
time.sleep(0.05)
|
|
finally:
|
|
if temporary.exists():
|
|
temporary.unlink()
|
|
|
|
|
|
def append_events(path, events):
|
|
if not events:
|
|
return
|
|
if path.exists() and path.stat().st_size > 4 * 1024 * 1024:
|
|
os.replace(path,path.with_suffix('.previous.jsonl'))
|
|
with path.open('a',encoding='utf-8') as f:
|
|
for event in events:
|
|
f.write(json.dumps(event,ensure_ascii=False,separators=(',',':'))+'\n')
|
|
|
|
class ProcessEntry(c.Structure):
|
|
_fields_=[('dwSize',w.DWORD),('cntUsage',w.DWORD),('th32ProcessID',w.DWORD),('th32DefaultHeapID',c.c_size_t),('th32ModuleID',w.DWORD),('cntThreads',w.DWORD),('th32ParentProcessID',w.DWORD),('pcPriClassBase',w.LONG),('dwFlags',w.DWORD),('szExeFile',w.WCHAR*260)]
|
|
class ModuleEntry(c.Structure):
|
|
_fields_=[('dwSize',w.DWORD),('th32ModuleID',w.DWORD),('th32ProcessID',w.DWORD),('GlblcntUsage',w.DWORD),('ProccntUsage',w.DWORD),('modBaseAddr',c.c_void_p),('modBaseSize',w.DWORD),('hModule',w.HMODULE),('szModule',w.WCHAR*256),('szExePath',w.WCHAR*260)]
|
|
|
|
class WindowsCollector:
|
|
def __init__(self):
|
|
if os.name != 'nt':
|
|
raise RuntimeError('This monitor requires Windows')
|
|
self.kernel=c.WinDLL('kernel32',use_last_error=True)
|
|
definitions=[
|
|
('CreateToolhelp32Snapshot',w.HANDLE,[w.DWORD,w.DWORD]),
|
|
('Process32FirstW',w.BOOL,[w.HANDLE,c.POINTER(ProcessEntry)]),
|
|
('Process32NextW',w.BOOL,[w.HANDLE,c.POINTER(ProcessEntry)]),
|
|
('Module32FirstW',w.BOOL,[w.HANDLE,c.POINTER(ModuleEntry)]),
|
|
('Module32NextW',w.BOOL,[w.HANDLE,c.POINTER(ModuleEntry)]),
|
|
('OpenProcess',w.HANDLE,[w.DWORD,w.BOOL,w.DWORD]),
|
|
('QueryFullProcessImageNameW',w.BOOL,[w.HANDLE,w.DWORD,w.LPWSTR,c.POINTER(w.DWORD)]),
|
|
('GetProcessTimes',w.BOOL,[w.HANDLE,c.POINTER(w.FILETIME),c.POINTER(w.FILETIME),c.POINTER(w.FILETIME),c.POINTER(w.FILETIME)]),
|
|
('CloseHandle',w.BOOL,[w.HANDLE]),
|
|
('CreateMutexW',w.HANDLE,[c.c_void_p,w.BOOL,w.LPCWSTR]),
|
|
('ReleaseMutex',w.BOOL,[w.HANDLE]),
|
|
]
|
|
for name,restype,args in definitions:
|
|
f=getattr(self.kernel,name);f.restype=restype;f.argtypes=args
|
|
self.ip=c.WinDLL('iphlpapi',use_last_error=True)
|
|
for name in ['GetExtendedTcpTable','GetExtendedUdpTable']:
|
|
f=getattr(self.ip,name);f.restype=w.DWORD;f.argtypes=[c.c_void_p,c.POINTER(w.DWORD),w.BOOL,w.ULONG,c.c_int,w.ULONG]
|
|
self.metadata={}
|
|
def snapshot_handle(self,flags,pid):
|
|
h=self.kernel.CreateToolhelp32Snapshot(flags,pid)
|
|
if h==c.c_void_p(-1).value:
|
|
raise OSError(c.get_last_error(),'process/module snapshot failed')
|
|
return h
|
|
def processes(self):
|
|
h=self.snapshot_handle(2,0);entry=ProcessEntry();entry.dwSize=c.sizeof(entry);all_items={}
|
|
try:
|
|
ok=self.kernel.Process32FirstW(h,c.byref(entry))
|
|
if not ok:
|
|
raise OSError(c.get_last_error(),'process enumeration failed')
|
|
while ok:
|
|
all_items[entry.th32ProcessID]={'pid':entry.th32ProcessID,'parentPid':entry.th32ParentProcessID,'name':entry.szExeFile}
|
|
ok=self.kernel.Process32NextW(h,c.byref(entry))
|
|
finally:
|
|
self.kernel.CloseHandle(h)
|
|
selected={pid for pid,p in all_items.items() if p['name'].lower() in TARGET_NAMES}
|
|
while True:
|
|
more={pid for pid,p in all_items.items() if p['parentPid'] in selected and pid not in selected}
|
|
if not more:
|
|
break
|
|
selected.update(more)
|
|
result=[]
|
|
for pid in sorted(selected):
|
|
item=all_items[pid];item['path']='';item['createdAt']=None
|
|
# Only query metadata. No PROCESS_VM_READ / WRITE / CREATE_THREAD.
|
|
process=self.kernel.OpenProcess(0x1000,False,pid)
|
|
if not process:
|
|
item['metadataError']=c.get_last_error()
|
|
else:
|
|
try:
|
|
name=c.create_unicode_buffer(32768);length=w.DWORD(len(name))
|
|
if self.kernel.QueryFullProcessImageNameW(process,0,name,c.byref(length)):
|
|
item['path']=name.value
|
|
ts=[w.FILETIME() for _ in range(4)]
|
|
if self.kernel.GetProcessTimes(process,*(c.byref(t) for t in ts)):
|
|
ticks=(ts[0].dwHighDateTime<<32)|ts[0].dwLowDateTime
|
|
item['createdAt']=datetime.fromtimestamp(ticks/10000000-11644473600,timezone.utc).isoformat()
|
|
finally:
|
|
self.kernel.CloseHandle(process)
|
|
result.append(item)
|
|
return result
|
|
def modules(self,processes):
|
|
result=[];errors=[]
|
|
for process in processes:
|
|
pid=process['pid']
|
|
try:
|
|
h=self.snapshot_handle(0x18,pid)
|
|
try:
|
|
m=ModuleEntry();m.dwSize=c.sizeof(m);ok=self.kernel.Module32FirstW(h,c.byref(m))
|
|
if not ok:
|
|
raise OSError(c.get_last_error(),'module enumeration failed')
|
|
while ok:
|
|
if any(marker in m.szModule.lower() for marker in MODULE_MARKERS):
|
|
result.append({'pid':pid,'name':m.szModule,'path':m.szExePath})
|
|
ok=self.kernel.Module32NextW(h,c.byref(m))
|
|
finally:
|
|
self.kernel.CloseHandle(h)
|
|
except OSError as exc:
|
|
errors.append({'kind':'module_metadata','pid':pid,'error':exc.errno})
|
|
return result,errors
|
|
def connections(self,pids):
|
|
results=[];errors=[]
|
|
for family,protocol in [(2,'tcp'),(23,'tcp'),(2,'udp'),(23,'udp')]:
|
|
fn=self.ip.GetExtendedTcpTable if protocol=='tcp' else self.ip.GetExtendedUdpTable
|
|
klass=5 if protocol=='tcp' else 1
|
|
size=w.DWORD();code=fn(None,c.byref(size),False,family,klass,0)
|
|
if code not in (0,122):
|
|
errors.append({'kind':'network_table','family':family,'protocol':protocol,'error':code});continue
|
|
for attempt in range(3):
|
|
buf=c.create_string_buffer(max(size.value,4));code=fn(buf,c.byref(size),False,family,klass,0)
|
|
if code!=122:
|
|
break
|
|
if code:
|
|
errors.append({'kind':'network_table','family':family,'protocol':protocol,'error':code});continue
|
|
raw=buf.raw;count=struct.unpack_from('<I',raw)[0]
|
|
stride=24 if (family,protocol)==(2,'tcp') else 56 if protocol=='tcp' else 12 if family==2 else 28
|
|
if 4+count*stride>len(raw):
|
|
errors.append({'kind':'network_table','error':'truncated'});continue
|
|
for offset in range(4,4+count*stride,stride):
|
|
row=raw[offset:offset+stride];pid=struct.unpack_from('<I',row,stride-4)[0]
|
|
if pid not in pids:
|
|
continue
|
|
if family==2:
|
|
start=4 if protocol=='tcp' else 0
|
|
local=socket.inet_ntop(socket.AF_INET,row[start:start+4]);port=struct.unpack_from('>H',row,start+4)[0]
|
|
remote=socket.inet_ntop(socket.AF_INET,row[12:16]) if protocol=='tcp' else ''
|
|
remote_port=struct.unpack_from('>H',row,16)[0] if protocol=='tcp' else 0
|
|
state=struct.unpack_from('<I',row)[0] if protocol=='tcp' else 0
|
|
else:
|
|
local=socket.inet_ntop(socket.AF_INET6,row[:16]);port=struct.unpack_from('>H',row,20)[0]
|
|
remote=socket.inet_ntop(socket.AF_INET6,row[24:40]) if protocol=='tcp' else ''
|
|
remote_port=struct.unpack_from('>H',row,44)[0] if protocol=='tcp' else 0
|
|
state=struct.unpack_from('<I',row,48)[0] if protocol=='tcp' else 0
|
|
results.append({'pid':pid,'protocol':protocol,'family':'ipv4' if family==2 else 'ipv6','localAddress':local,'localPort':port,'remoteAddress':remote,'remotePort':remote_port,'state':TCP_STATES.get(state,str(state)) if protocol=='tcp' else 'BOUND'})
|
|
return sorted(results,key=lambda x:json.dumps(x,sort_keys=True)),errors
|
|
def snapshot(self):
|
|
processes=self.processes();pids={p['pid'] for p in processes}
|
|
connections,errors=self.connections(pids);modules,module_errors=self.modules(processes)
|
|
return {'checkedAt':utc_now(),'processes':processes,'connections':connections,'modules':modules,'errors':errors+module_errors}
|
|
|
|
def summarize(snapshot):
|
|
processes=snapshot['processes']
|
|
return {'processCount':len(processes),'agents':[p['pid'] for p in processes if p['name'].lower()=='wecom-agent.exe'],
|
|
'ironbox':[p['pid'] for p in processes if p['name'].lower().startswith('wironboxd')],
|
|
'tcpListeners':[{'pid':x['pid'],'address':x['localAddress'],'port':x['localPort']} for x in snapshot['connections'] if x['protocol']=='tcp' and x['state']=='LISTEN'],
|
|
'connectionCount':len(snapshot['connections']),'moduleCount':len(snapshot['modules']),'collectionErrors':snapshot['errors']}
|
|
|
|
def monitor(output,interval,duration):
|
|
output.mkdir(parents=True,exist_ok=True);collector=WindowsCollector();kernel=collector.kernel
|
|
mutex_name='Local\\WeChatRpaWecomProcessMonitor_'+hashlib.sha256(str(output.resolve()).lower().encode()).hexdigest()[:24]
|
|
mutex=kernel.CreateMutexW(None,True,mutex_name)
|
|
if not mutex:
|
|
raise OSError(c.get_last_error(),'monitor lock failed')
|
|
if c.get_last_error()==183:
|
|
kernel.CloseHandle(mutex);raise RuntimeError('A monitor already owns this output directory')
|
|
run_id=uuid.uuid4().hex;stop=threading.Event();status_path=output/'status.json';events_path=output/'events.jsonl'
|
|
started=time.monotonic();previous={};count=0;reason='duration_elapsed'
|
|
status={'schemaVersion':1,'runId':run_id,'monitorPid':os.getpid(),'startedAt':utc_now(),'state':'starting','intervalSeconds':interval,'durationSeconds':duration,'readOnly':True,'capturesMessageContent':False,'protocolSendVerified':False,'processObservation':'snapshot_polling','shortLivedProcessesMayBeMissed':True}
|
|
atomic_json(status_path,status)
|
|
append_events(events_path,[{'type':'monitor_started','runId':run_id,'at':utc_now()}])
|
|
try:
|
|
while time.monotonic()-started<duration:
|
|
tick=time.monotonic();control=output/'stop.json'
|
|
if control.exists():
|
|
try:
|
|
if json.loads(control.read_text(encoding='utf-8')).get('runId')==run_id:
|
|
reason='stop_requested';break
|
|
except (OSError,ValueError):
|
|
pass
|
|
observed=[]
|
|
try:
|
|
current=collector.snapshot()
|
|
# Retain rows for metadata tables that failed, so failures are not mistaken for exits.
|
|
failed_modules={e['pid'] for e in current['errors'] if e['kind']=='module_metadata'}
|
|
current['modules'].extend(x for x in previous.get('modules',[]) if x['pid'] in failed_modules)
|
|
if any(e['kind']=='network_table' for e in current['errors']):
|
|
current['connections']=previous.get('connections',current['connections'])
|
|
observed=changes(previous,current)
|
|
if count==0:
|
|
for event in observed:
|
|
event['baseline']=True
|
|
if event['type']=='process_started':
|
|
event['type']='process_observed'
|
|
previous=current;count+=1
|
|
atomic_json(output/'snapshot.json',current)
|
|
status.update(state='running',checkedAt=current['checkedAt'],snapshotCount=count,summary=summarize(current),lastError=None)
|
|
except Exception as exc:
|
|
status.update(state='degraded',checkedAt=utc_now(),lastError=type(exc).__name__+': '+str(exc))
|
|
for event in observed:
|
|
event.update(runId=run_id,at=utc_now())
|
|
append_events(events_path,observed)
|
|
status['lastChangeAt']=utc_now() if observed else status.get('lastChangeAt')
|
|
atomic_json(status_path,status)
|
|
stop.wait(max(0,interval-(time.monotonic()-tick)))
|
|
except KeyboardInterrupt:
|
|
reason='interrupted'
|
|
finally:
|
|
stop.set()
|
|
status.update(state='stopped',stoppedAt=utc_now(),stopReason=reason,checkedAt=utc_now())
|
|
try:
|
|
atomic_json(status_path,status);append_events(events_path,[{'type':'monitor_stopped','runId':run_id,'at':utc_now(),'reason':reason}])
|
|
finally:
|
|
kernel.ReleaseMutex(mutex);kernel.CloseHandle(mutex)
|
|
return status
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__)
|
|
mode=p.add_mutually_exclusive_group();mode.add_argument('--watch',action='store_true');mode.add_argument('--stop',action='store_true');mode.add_argument('--status',action='store_true')
|
|
p.add_argument('--output',type=Path,default=DEFAULT_OUTPUT);p.add_argument('--interval',type=float,default=0.5);p.add_argument('--duration',type=float,default=7200)
|
|
args=p.parse_args()
|
|
if not 0.2<=args.interval<=60 or not 1<=args.duration<=86400:
|
|
p.error('interval must be 0.2..60 seconds; duration must be 1..86400 seconds')
|
|
if args.status or args.stop:
|
|
result=json.loads((args.output/'status.json').read_text(encoding='utf-8'))
|
|
if args.status and result.get('state') in {'running','degraded','starting'}:
|
|
age=(datetime.now(timezone.utc)-datetime.fromisoformat(result.get('checkedAt',result['startedAt']))).total_seconds()
|
|
if age>max(10,result.get('intervalSeconds',1)*4):
|
|
result['state']='stale'
|
|
result['statusAgeSeconds']=round(age,1)
|
|
if args.stop:
|
|
atomic_json(args.output/'stop.json',{'runId':result['runId'],'requestedAt':utc_now()})
|
|
result={'stopRequested':True,'monitorPid':result['monitorPid'],'runId':result['runId']}
|
|
elif args.watch:
|
|
result=monitor(args.output,args.interval,args.duration)
|
|
else:
|
|
result=WindowsCollector().snapshot()
|
|
print(json.dumps(result,ensure_ascii=False,indent=2))
|
|
|
|
if __name__=='__main__':
|
|
main() |