mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
hakril
550 lines
20 KiB
Python
550 lines
20 KiB
Python
import os
|
|
import re
|
|
import sys
|
|
import xml
|
|
import ctypes
|
|
import struct
|
|
import pickle
|
|
import logging
|
|
import binascii
|
|
import traceback
|
|
|
|
from io import BytesIO
|
|
from collections import namedtuple
|
|
|
|
# PythonForWindows
|
|
import windows
|
|
import windows.generated_def as gdef
|
|
from windows.winobject import event_log
|
|
|
|
""" ParsedElement : This represent a deserialized element for the EventRecord user data, using the xml template associated with the event. """
|
|
ParsedElement = namedtuple('ParsedElement', 'index, name, value, format')
|
|
|
|
class RealtimeEventLoggerBase(object):
|
|
""" Virtual class, used mainly to inject the instance in the EventRecord's context """
|
|
|
|
def __init__(self):
|
|
self.etw_trace = None
|
|
|
|
def start_process_trace(self):
|
|
""" """
|
|
|
|
# Blocking call here, you can't do anything anymore except killing the trace
|
|
self.etw_trace.process(RealtimeEventLogger.callback, context = self)
|
|
|
|
@staticmethod
|
|
def callback(event):
|
|
""" Custom callback for listening to ETW events and process them. """
|
|
self = event.context
|
|
self.process_event(event)
|
|
|
|
|
|
def process_event(self, event):
|
|
raise NotImplementedError("You should implement this method in a subclass !")
|
|
|
|
|
|
|
|
class RealtimeEventLogger(RealtimeEventLoggerBase):
|
|
""" Custom reimplementation of a event logger. """
|
|
|
|
def __init__(self, trace_name, publisher = None):
|
|
super(RealtimeEventLogger, self).__init__()
|
|
|
|
# Open ETW publisher
|
|
self.publisher = None
|
|
self.publisher_name = publisher
|
|
self.publisher_guid = None
|
|
|
|
if self.publisher_name:
|
|
manager = event_log.EvtlogManager()
|
|
self.publisher = manager.open_publisher(self.publisher_name)
|
|
self.publisher_guid = self.publisher.metadata.guid
|
|
logging.debug("publisher guid : %s" % self.publisher_guid.to_string())
|
|
|
|
# ETW trace
|
|
self.any_keywords = 0
|
|
self.etw_trace_name = trace_name
|
|
|
|
def setup_trace(self):
|
|
|
|
# Collecting all keywords for listening to events to
|
|
event_keywords = set(map(lambda event_metadata: event_metadata.keyword, self.publisher.metadata.events_metadata))
|
|
self.any_keywords = 0x00
|
|
for keyword in list(event_keywords):
|
|
self.any_keywords |= keyword
|
|
|
|
logging.debug("events KeywordsAny : 0x%x" % self.any_keywords)
|
|
|
|
|
|
# Create a custom realtime ETW trace for printing out events
|
|
self.etw_trace = windows.system.etw.open_trace(self.etw_trace_name)
|
|
|
|
|
|
def start_trace(self):
|
|
self.etw_trace.stop(soft=True)
|
|
self.etw_trace.start()
|
|
|
|
# We can't configure etw trace if it's not started previously
|
|
self.etw_trace.enable_ex(
|
|
self.publisher_guid.to_string(),
|
|
flags=0,
|
|
level=0xff,
|
|
any_keyword=self.any_keywords
|
|
)
|
|
|
|
self.start_process_trace()
|
|
|
|
def stop_trace(self):
|
|
# No way to stop it from python :(
|
|
os.system("logman -ets stop %s" % self.etw_trace_name)
|
|
|
|
def process_event(self, event):
|
|
""" Custom callback for listening to ETW events and parse them correctly. """
|
|
|
|
try:
|
|
if event.guid == self.publisher.metadata.guid:
|
|
|
|
# Check this is an event we can parse
|
|
event_metadata = self.lookup_event_metadata(self.publisher, event.id)
|
|
if not event_metadata:
|
|
return
|
|
|
|
logging.debug("recv event user data: %s" % event.user_data)
|
|
logging.debug("recv xml template : %s" % event_metadata.template)
|
|
|
|
# Deserialize event.user_data based on the event_metadata xml template
|
|
message_data = event.user_data
|
|
message_params = self.parse_user_data(event_metadata, message_data)
|
|
logging.debug("message params : %s" % message_params)
|
|
|
|
|
|
try:
|
|
template_message = self.publisher.metadata.message(event_metadata.message_id)
|
|
logging.debug("template_message : %s" % message_params)
|
|
except WindowsError as e:
|
|
if e.winerror != gdef.ERROR_INVALID_PARAMETER:
|
|
raise
|
|
return
|
|
|
|
# import pdb;pdb.set_trace()
|
|
xx = tuple(event_log.ImprovedEVT_VARIANT.from_value(x.value) for x in message_params)
|
|
# xx = tuple(event_log.ImprovedEVT_VARIANT.from_value(x.value) for x in message_params)
|
|
res = ctypes.c_buffer(0x1000)
|
|
res_size = gdef.DWORD()
|
|
yolo_ptr = (gdef.EVT_VARIANT * len(xx))(*xx)
|
|
# import pdb;pdb.set_trace()
|
|
windows.winproxy.EvtFormatMessage(self.publisher.metadata, None, event_metadata.message_id, len(message_params), yolo_ptr, gdef.EvtFormatMessageId, 0x1000, ctypes.cast(res, gdef.LPCWSTR), res_size)
|
|
str = res[:res_size.value * 2].decode("utf-16-le")
|
|
print("=== MINE ===")
|
|
print(str)
|
|
|
|
print("=== REAL ===")
|
|
# "sprintf" the message using the event format message as well as the deserialized elements
|
|
event_message = self.format_event_log_message(template_message, message_params)
|
|
print(event_message)
|
|
import pdb;pdb.set_trace()
|
|
|
|
except Exception as unke:
|
|
print("Unhandled exception in Evtlogger.process_event : %s" % unke)
|
|
traceback.print_tb(sys.exc_info()[2])
|
|
sys.exit(0) # Exiting on unknown error, since this is the only way to have some control
|
|
finally:
|
|
pass
|
|
|
|
def parse_unicode_string(self, stream):
|
|
""" Deserialize a wide string """
|
|
|
|
uni_string = b""
|
|
while True:
|
|
unicode_byte = stream.read(2)
|
|
|
|
if unicode_byte == b"\x00\x00":
|
|
break
|
|
|
|
uni_string += unicode_byte
|
|
|
|
return uni_string.decode("utf-16")
|
|
|
|
def parse_element(self, in_type, stream, length = None):
|
|
|
|
# Types definis dans les templates
|
|
# https://docs.microsoft.com/en-us/windows/win32/wes/eventmanifestschema-inputtype-complextype
|
|
if in_type == "win:UnicodeString":
|
|
value = self.parse_unicode_string(stream)
|
|
elif in_type == "win:UInt8":
|
|
value = struct.unpack("B", stream.read(1))[0]
|
|
|
|
elif in_type == "win:Int32":
|
|
value = struct.unpack("I", stream.read(4))[0]
|
|
elif in_type == "win:UInt32":
|
|
value = struct.unpack("I", stream.read(4))[0]
|
|
elif in_type == "win:HexInt32":
|
|
value = struct.unpack("I", stream.read(4))[0]
|
|
elif in_type == "xs:unsignedLong":
|
|
value = struct.unpack("I", stream.read(4))[0]
|
|
|
|
elif in_type == "win:UInt64":
|
|
value = struct.unpack("Q", stream.read(8))[0]
|
|
elif in_type == "win:Pointer":
|
|
value = struct.unpack("Q", stream.read(8))[0]
|
|
elif in_type == "win:Double":
|
|
value = struct.unpack("Q", stream.read(8))[0]
|
|
|
|
elif in_type == "win:Boolean":
|
|
value = struct.unpack("I", stream.read(4))[0] == 1
|
|
|
|
elif in_type == "win:Binary":
|
|
if not length:
|
|
raise ValueError(" param_in_type (%s) cannot be used with a null length value" % in_type)
|
|
|
|
# TODO : we should return the raw bytes buffer, since get_param_str_format is too crude
|
|
# to properly display win:SocketAddress parameters
|
|
value = binascii.hexlify(stream.read(length))
|
|
|
|
elif in_type == "win:GUID":
|
|
guid_data = struct.unpack("IHHBBBBBBBB", stream.read(16))
|
|
value = gdef.GUID.from_raw(*guid_data).to_string()
|
|
else:
|
|
raise ValueError("unrecognized param_in_type : %s" % in_type)
|
|
|
|
return value
|
|
|
|
def get_param_str_format(self, param_out_type):
|
|
PYTHON_FORMAT_DICT = {
|
|
"xs:boolean" : "s", # "True" or "False"
|
|
"xs:unsignedByte" : "02x",
|
|
"xs:int" : "d",
|
|
"xs:double" : "f",
|
|
"xs:unsignedInt" : "d",
|
|
"xs:unsignedLong" : "d",
|
|
"win:ErrorCode" : "x",
|
|
"win:HexInt64" : "0x08x",
|
|
"win:HexInt32" : "0x04x",
|
|
"xs:string" : "s",
|
|
"win:SocketAddress" : "s" # TODO
|
|
}
|
|
|
|
if param_out_type not in PYTHON_FORMAT_DICT:
|
|
raise ValueError("unrecognized out param : %s" % param_out_type)
|
|
|
|
return PYTHON_FORMAT_DICT[param_out_type]
|
|
|
|
def parse_user_data(self, event_metadata, data):
|
|
"""
|
|
Deserialize event.user_data based on the associated publisher's template.
|
|
Return a list of ParsedElement(Name:string, Value:py_object, Type:py_type).
|
|
"""
|
|
stream = BytesIO(data)
|
|
event_items = event_metadata.event_data
|
|
params = []
|
|
if not event_items:
|
|
return []
|
|
|
|
context = {} # saving parsed items for "count" elements
|
|
for (i, param_data) in enumerate(event_items):
|
|
# Some param are repeating, and "count" refers to the variable holding the number of repetitions
|
|
param_in_type = param_data["inType"]
|
|
param_out_type = param_data["outType"]
|
|
param_name = param_data["name"]
|
|
param_count = param_data.get("count", None)
|
|
if param_count:
|
|
param_count = context[param_count] # must be already set
|
|
if param_count == 0:
|
|
continue
|
|
|
|
# Some param (win:Binary) have a length attribute
|
|
param_length = param_data.get("length", None)
|
|
if param_length:
|
|
param_length = context[param_length] # must be already set
|
|
|
|
# Parse element
|
|
if param_count:
|
|
value = [self.parse_element(param_in_type, stream, param_length) for c in range(param_count)]
|
|
else:
|
|
value = self.parse_element(param_in_type, stream, param_length)
|
|
|
|
context[param_name] = value
|
|
format_type = self.get_param_str_format(param_out_type)
|
|
|
|
logging.debug(ParsedElement(i, param_name, value, format_type))
|
|
params.append(ParsedElement(i, param_name, value, format_type)) # Yield ?
|
|
|
|
return params
|
|
|
|
def lookup_event_metadata(self, publisher, event_id):
|
|
matching_events_metadata = list(filter(lambda event_meta: event_meta.id == event_id, publisher.metadata.events_metadata))
|
|
|
|
if not len(matching_events_metadata):
|
|
return None
|
|
|
|
if len(matching_events_metadata) > 1:
|
|
return None
|
|
|
|
return matching_events_metadata[0]
|
|
|
|
def format_event_log_message(self, template, event_args):
|
|
|
|
py_template = ""
|
|
last_span = (0,0)
|
|
|
|
# Convert message template to python string formating
|
|
# e.g. : "ParseError: HResult: %1, Error: %2." into "ParseError: HResult: {arg0:x}, Error: {arg1:d}."
|
|
pattern = re.compile(r"%(\d)")
|
|
for match in re.finditer(pattern, template):
|
|
|
|
arg_id = int(match.groups()[0]) - 1 # event's template message index params from 1 to N, wtf
|
|
span = match.span()
|
|
str_format = ""
|
|
|
|
str_format = "{a%d:%s}" % (arg_id, event_args[arg_id].format)
|
|
|
|
py_template += template[last_span[1]:span[0]] + str_format
|
|
last_span = span
|
|
|
|
py_template += template[last_span[1]:]
|
|
|
|
logging.debug(py_template)
|
|
|
|
# string formating using Python .format()
|
|
events_kwargs = {"a%d" % (x.index) : x.value for x in event_args}
|
|
logging.debug(events_kwargs)
|
|
message = py_template.format(**events_kwargs)
|
|
logging.debug(message)
|
|
return message
|
|
|
|
|
|
|
|
def get_message(publisher_metadata, message_id, get_str_message):
|
|
""" if --gm is set, try to return the str message associated. If not, return the raw value id """
|
|
if not get_str_message:
|
|
return "%d" % message_id
|
|
|
|
try:
|
|
return publisher_metadata.message(message_id)
|
|
except WindowsError as e:
|
|
if e.winerror != gdef.ERROR_INVALID_PARAMETER:
|
|
raise
|
|
return ""
|
|
|
|
def format_channel_metadata(publisher_metadata, channel_metadata, args):
|
|
""" Str formating channel metadata """
|
|
return "\n".join([
|
|
" channel:",
|
|
" name: {channel.name:s}",
|
|
" id: {channel.id:d}",
|
|
" flags: {channel.flags:d}",
|
|
" message: {channel_message:s}",
|
|
]).format(
|
|
channel=channel_metadata,
|
|
channel_message=get_message(publisher_metadata, channel_metadata.message_id, args.gm)
|
|
)
|
|
|
|
def format_level_metadata(publisher_metadata, level_metadata, args):
|
|
""" Str formating level metadata """
|
|
return "\n".join([
|
|
" level:",
|
|
" name: {level.name:s}",
|
|
" value: {level.value:d}",
|
|
" message: {level_message:s}",
|
|
]).format(
|
|
level=level_metadata,
|
|
level_message=get_message(publisher_metadata, level_metadata.message_id, args.gm)
|
|
)
|
|
|
|
def format_opcode_metadata(publisher_metadata, opcode_metadata, args):
|
|
""" Str formating opcode metadata """
|
|
return "\n".join([
|
|
" opcode:",
|
|
" name: {opcode.name:s}",
|
|
" value: {opcode.value:d}",
|
|
#" task: {opcode.task:d}", # TODO
|
|
#" opcode: {opcode.task_value:d}", # TODO
|
|
" message: {opcode_message:s}",
|
|
]).format(
|
|
opcode=opcode_metadata,
|
|
opcode_message=get_message(publisher_metadata, opcode_metadata.message_id, args.gm)
|
|
)
|
|
|
|
def format_task_metadata(publisher_metadata, task_metadata, args):
|
|
""" Str formating task metadata """
|
|
return "\n".join([
|
|
" task:",
|
|
" name: {task.name:s}",
|
|
" value: {task.value:d}",
|
|
" eventGUID: {task.event_guid:s}",
|
|
" message: {task_message:s}",
|
|
]).format(
|
|
task=task_metadata,
|
|
task_message=get_message(publisher_metadata, task_metadata.message_id, args.gm)
|
|
)
|
|
|
|
def format_keyword_metadata(publisher_metadata, keyword_metadata, args):
|
|
""" Str formating keyword metadata """
|
|
return "\n".join([
|
|
" keyword:",
|
|
" name: {keyword.name:s}",
|
|
" mask: {keyword.value:x}",
|
|
" message: {keyword_message:s}",
|
|
]).format(
|
|
keyword=keyword_metadata,
|
|
keyword_message=get_message(publisher_metadata, keyword_metadata.message_id, args.gm)
|
|
)
|
|
|
|
def format_event_metadata(publisher_metadata, event_metadata, args):
|
|
""" Str formating keyword metadata """
|
|
return "\n".join([
|
|
" event:",
|
|
" value: {event.id:d}",
|
|
" version: {event.version:d}",
|
|
" opcode: {event.opcode:d}",
|
|
" channel: {event.channel_id:d}",
|
|
" level: {event.level:d}",
|
|
" task: {event.task:d}",
|
|
" keywords: 0x{event.keyword:016x}",
|
|
" message: {event_message:s}"
|
|
]).format(
|
|
event=event_metadata,
|
|
event_message=get_message(publisher_metadata, event_metadata.message_id, args.gm)
|
|
)
|
|
|
|
def enum_publishers(args):
|
|
""" enum-publishers verb implementation """
|
|
manager = event_log.EvtlogManager()
|
|
for publisher in sorted(list(manager.publishers), key=lambda pub:pub.name.lower()):
|
|
print(publisher.name)
|
|
|
|
def get_publisher(args):
|
|
""" get-publisher verb implementation """
|
|
|
|
manager = event_log.EvtlogManager()
|
|
publisher = manager.open_publisher(args.publisher_name)
|
|
|
|
channels_info = "\n".join(map(lambda c: format_channel_metadata(publisher.metadata, c, args), publisher.metadata.channels_metadata))
|
|
levels_info = "\n".join(map(lambda l: format_level_metadata(publisher.metadata, l, args), publisher.metadata.levels_metadata))
|
|
opcodes_info = "\n".join(map(lambda o: format_opcode_metadata(publisher.metadata, o, args), publisher.metadata.opcodes_metadata))
|
|
tasks_info = "\n".join(map(lambda t: format_task_metadata(publisher.metadata, t, args), publisher.metadata.tasks_metadata))
|
|
keywords_info = "\n".join(map(lambda k: format_keyword_metadata(publisher.metadata, k, args), publisher.metadata.keywords_metadata))
|
|
events_info = "\n".join(map(lambda e: format_event_metadata(publisher.metadata, e, args), publisher.metadata.events_metadata))
|
|
|
|
publisher_infos = "\n".join([
|
|
"name: {pub_name:s}",
|
|
"guid: {pub_guid:s}",
|
|
]).format(
|
|
pub_name=publisher.name,
|
|
pub_guid=publisher.metadata.guid.to_string(),
|
|
)
|
|
|
|
if publisher.metadata.message_resource_filepath != None:
|
|
publisher_infos += "\n"
|
|
publisher_infos += "resourceFileName: {:s}".format(publisher.metadata.message_resource_filepath)
|
|
if publisher.metadata.message_parameter_filepath != None:
|
|
publisher_infos += "\n"
|
|
publisher_infos += "parameterFileName: {:s}".format(publisher.metadata.message_parameter_filepath)
|
|
if publisher.metadata.message_filepath != None:
|
|
publisher_infos += "\n"
|
|
publisher_infos += "messageFileName: {:s}".format(publisher.metadata.message_filepath)
|
|
|
|
publisher_infos += "\n"
|
|
publisher_infos += "message: {:s}".format(get_message(publisher.metadata, publisher.metadata.message_id, args.gm))
|
|
|
|
# Channels
|
|
publisher_infos += "\n"
|
|
publisher_infos += "channels:"
|
|
if channels_info != "":
|
|
publisher_infos += "\n"
|
|
publisher_infos += channels_info
|
|
|
|
# Levels
|
|
publisher_infos += "\n"
|
|
publisher_infos += "levels:"
|
|
if levels_info != "":
|
|
publisher_infos += "\n"
|
|
publisher_infos += levels_info
|
|
|
|
# Opcodes
|
|
publisher_infos += "\n"
|
|
publisher_infos += "opcodes:"
|
|
if opcodes_info != "":
|
|
publisher_infos += "\n"
|
|
publisher_infos += opcodes_info
|
|
|
|
# Tasks
|
|
publisher_infos += "\n"
|
|
publisher_infos += "tasks:"
|
|
if tasks_info != "":
|
|
publisher_infos += "\n"
|
|
publisher_infos += tasks_info
|
|
|
|
# Keywords
|
|
publisher_infos += "\n"
|
|
publisher_infos += "keywords:"
|
|
if keywords_info != "":
|
|
publisher_infos += "\n"
|
|
publisher_infos += keywords_info
|
|
|
|
# Events
|
|
if args.ge:
|
|
publisher_infos += "\n"
|
|
publisher_infos += "events:"
|
|
if events_info != "":
|
|
publisher_infos += "\n"
|
|
publisher_infos += events_info
|
|
|
|
publisher_infos += "\n"
|
|
print(publisher_infos)
|
|
|
|
|
|
def main(args):
|
|
|
|
if args.action == "enum-publishers":
|
|
enum_publishers(args)
|
|
elif args.action == "get-publisher":
|
|
get_publisher(args)
|
|
elif args.action == "start-trace":
|
|
evl = RealtimeEventLogger(args.etw_name, publisher = args.publisher_name)
|
|
evl.setup_trace()
|
|
evl.start_trace()
|
|
elif args.action == "stop-trace":
|
|
evl = RealtimeEventLogger(args.etw_name)
|
|
evl.stop_trace()
|
|
else:
|
|
raise NotImplementedError("Unknown action : %s" % args.action)
|
|
|
|
if __name__ == '__main__':
|
|
|
|
import argparse
|
|
parser = argparse.ArgumentParser("wevtutil script reimplementation using PythonForWindows")
|
|
parser.add_argument("-v", "--verbose", action="store_true", help="active verbose logging")
|
|
action_parsers = parser.add_subparsers(dest="action", help="subparsers for action specific arguments")
|
|
|
|
# we can't express shorthands easily like ep for enum-publishers since only Python3's argparse surpport parser "aliases"
|
|
enum_publishers_parser = action_parsers.add_parser("enum-publishers", help="enum-publishers verb")
|
|
|
|
get_publisher_parser = action_parsers.add_parser("get-publisher", help="get-publisher verb")
|
|
get_publisher_parser.add_argument("publisher_name", type=str, help="registered publisher name")
|
|
get_publisher_parser.add_argument("--ge", action="store_true", help="get event metadata")
|
|
get_publisher_parser.add_argument("--gm", action="store_true", help="get message name instead of raw id")
|
|
# get_publisher_parser.add_argument("--f", type=str, help="format") # TODO
|
|
# get_publisher_parser.add_argument("--im", "--install-manifest", type=str, help="install manifest ???") # TODO
|
|
|
|
|
|
# Not a wevutil command, but something nice to have ;p
|
|
start_trace_parser = action_parsers.add_parser("start-trace", help="start a realtime ETW trace")
|
|
start_trace_parser.add_argument("etw_name", type=str, help="ETW trace name")
|
|
start_trace_parser.add_argument("publisher_name", type=str, help="registered publisher name")
|
|
|
|
stop_trace_parser = action_parsers.add_parser("stop-trace", help="start a realtime ETW trace")
|
|
stop_trace_parser.add_argument("etw_name", type=str, help="ETW trace name")
|
|
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.verbose:
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
else:
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
main(args)
|
|
|
|
|