From d756cb1027aa1f31c599afcac3e8d8e457a9ed87 Mon Sep 17 00:00:00 2001 From: hakril Date: Wed, 22 May 2024 17:52:23 +0200 Subject: [PATCH] windows.system.etw support unicode + tests --- tests/test_event_trace.py | 38 +++++++++++++++++++++++++++++++- windows/winobject/event_trace.py | 24 ++++++++++---------- 2 files changed, 49 insertions(+), 13 deletions(-) diff --git a/tests/test_event_trace.py b/tests/test_event_trace.py index 3004c67..9a8abb4 100644 --- a/tests/test_event_trace.py +++ b/tests/test_event_trace.py @@ -1,9 +1,13 @@ +# -*- coding: utf-8 -*- + import collections import os import windows import windows.generated_def as gdef +from windows.winobject.event_trace import EventTraceProperties + TASK_SCHEDULER_PROVIDER = "047311A9-FA52-4A68-A1E4-4E289FBB8D17" @@ -19,16 +23,48 @@ def test_etw_trace_open_with_guid(): def test_etw_trace_registration_and_processing(): EVENT_ID_COUNT.clear() # RealTime Test + trace = windows.system.etw.open_trace("PFW_test_etw_2", logfile="pfw_test_trace.etl") + assert not trace.exists() trace.start() + assert trace.exists() trace.enable(TASK_SCHEDULER_PROVIDER, 0xff, 0xff) # Scheduler code that generate event windows.system.task_scheduler(r"\Microsoft\Windows\Chkdsk")["SyspartRepair"] # End of scheduler code trace.stop() + assert not trace.exists() trace.process(show) # Task scheduler generate event id 10,11,12 assert EVENT_ID_COUNT[10] >= 1 assert EVENT_ID_COUNT[11] >= 1 assert EVENT_ID_COUNT[12] >= 1 - os.unlink("pfw_test_trace.etl") \ No newline at end of file + os.unlink("pfw_test_trace.etl") + +def test_etw_EventTraceProperties_struct(): + traceprop = EventTraceProperties.create() + traceprop.set_logger_name(u"中国银行网银助手") + assert traceprop.get_logger_name() == u"中国银行网银助手" + + traceprop.set_logfilename(u"中国银行网银助手_中国银行网银助手") + assert traceprop.get_logfilename() == u"中国银行网银助手_中国银行网银助手" + +def test_etw_trace_unicode(): + trace = windows.system.etw.open_trace(u"中国银行网银助手", logfile="pfw_test_trace_unicode.etl") + try: + assert not trace.exists() + trace.start() + assert trace.exists() + trace.flush() + # Test session listing for unicode string as well + assert u"中国银行网银助手" in [session.name for session in windows.system.etw.sessions] + trace.stop() + assert not trace.exists() + finally: + if trace.exists(): + trace.stop() + if os.path.exists("pfw_test_trace_unicode.etl"): + os.unlink("pfw_test_trace_unicode.etl") + +def test_etw_trace_sessions(): + assert any(u"Kernel" in session.name for session in windows.system.etw.sessions) \ No newline at end of file diff --git a/windows/winobject/event_trace.py b/windows/winobject/event_trace.py index 65e94ad..baddceb 100644 --- a/windows/winobject/event_trace.py +++ b/windows/winobject/event_trace.py @@ -97,28 +97,28 @@ class EventTraceProperties(gdef.EVENT_TRACE_PROPERTIES): def get_logfilename(self): assert self.LogFileNameOffset - return windows.current_process.read_string(ctypes.addressof(self) + self.LogFileNameOffset) + return windows.current_process.read_wstring(ctypes.addressof(self) + self.LogFileNameOffset) def set_logfilename(self, filename): assert self.LogFileNameOffset if not filename.endswith("\x00"): filename += "\x00" - return windows.current_process.write_memory(ctypes.addressof(self) + self.LogFileNameOffset, filename) + return windows.current_process.write_memory(ctypes.addressof(self) + self.LogFileNameOffset, filename.encode("utf-16-le")) logfile = property(get_logfilename, set_logfilename) #: The logfile associated with the session def get_logger_name(self): assert self.LoggerNameOffset - return windows.current_process.read_string(ctypes.addressof(self) + self.LoggerNameOffset) + return windows.current_process.read_wstring(ctypes.addressof(self) + self.LoggerNameOffset) - def set_logfilename(self, filename): + def set_logger_name(self, filename): assert self.LoggerNameOffset if not filename.endswith("\x00"): filename += "\x00" - return windows.current_process.write_memory(ctypes.addressof(self) + self.LoggerNameOffset, filename) + return windows.current_process.write_memory(ctypes.addressof(self) + self.LoggerNameOffset, filename.encode("utf-16-le")) - name = property(get_logger_name, set_logfilename) #: The name of the session + name = property(get_logger_name, set_logger_name) #: The name of the session @property def guid(self): @@ -168,7 +168,7 @@ class CtxProcess(object): class EtwTrace(object): """Represent an ETW Trace for tracing/processing events""" def __init__(self, name, logfile=None, guid=None): - self.name = windows.pycompat.raw_encode(name) #: The name of the trace + self.name = name #: The name of the trace self.logfile = logfile #: The logging file of the trace (``None`` means real time trace) if guid and isinstance(guid, basestring): guid = gdef.GUID.from_string(guid) @@ -179,7 +179,7 @@ class EtwTrace(object): """Return ``True`` if the trace already exist (based on its name)""" prop = EventTraceProperties.create() try: - windows.winproxy.ControlTraceA(self.handle, self.name, prop, gdef.EVENT_TRACE_CONTROL_QUERY) + windows.winproxy.ControlTraceW(self.handle, self.name, prop, gdef.EVENT_TRACE_CONTROL_QUERY) except WindowsError as e: if e.winerror == gdef.ERROR_WMI_INSTANCE_NOT_FOUND: return False # Not found -> does not exists @@ -199,7 +199,7 @@ class EtwTrace(object): if self.name: # Base REAL_TIME on option ? name presence ? logfile presence ? prop.LogFileMode |= gdef.EVENT_TRACE_REAL_TIME_MODE handle = gdef.TRACEHANDLE() - windows.winproxy.StartTraceA(handle, self.name, prop) + windows.winproxy.StartTraceW(handle, self.name, prop) if not self.guid: self.guid = prop.Wnode.Guid self.handle = handle @@ -212,7 +212,7 @@ class EtwTrace(object): """ prop = EventTraceProperties.create() try: - windows.winproxy.ControlTraceA(0, self.name, prop, gdef.EVENT_TRACE_CONTROL_STOP) + windows.winproxy.ControlTraceW(0, self.name, prop, gdef.EVENT_TRACE_CONTROL_STOP) except WindowsError as e: if soft and e.winerror == gdef.ERROR_WMI_INSTANCE_NOT_FOUND: return False @@ -222,7 +222,7 @@ class EtwTrace(object): def flush(self): """Flush the trace""" prop = EventTraceProperties.create() - windows.winproxy.ControlTraceA(0, self.name, prop, gdef.EVENT_TRACE_CONTROL_FLUSH) + windows.winproxy.ControlTraceW(0, self.name, prop, gdef.EVENT_TRACE_CONTROL_FLUSH) def enable(self, guid, flags=0xff, level=0xff): @@ -424,7 +424,7 @@ class EtwManager(object): # Cast as array/ptr does not handle subtypes very-well tarray = ctypes.cast(array, ctypes.POINTER(ctypes.POINTER(gdef.EVENT_TRACE_PROPERTIES))) count = gdef.DWORD() - windows.winproxy.QueryAllTracesA(tarray, MAX_ETW_SESSIONS, count) + windows.winproxy.QueryAllTracesW(tarray, MAX_ETW_SESSIONS, count) return t[:count.value]