hakril-PythonForWindows/samples/event_log/wevtutil.py

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)