mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
[event log] Add example of what can be done with events
This commit is contained in:
1orenz0
@@ -0,0 +1,220 @@
|
||||
import os
|
||||
import sys
|
||||
import ctypes
|
||||
|
||||
import windows
|
||||
import windows.generated_def as gdef
|
||||
from windows.winobject import event_log
|
||||
|
||||
|
||||
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(
|
||||
level=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):
|
||||
|
||||
manager = event_log.EvtlogManager()
|
||||
for publisher in sorted(list(manager.publishers), key=lambda pub:pub.name.lower()):
|
||||
print(publisher.name)
|
||||
|
||||
def get_publisher(args):
|
||||
|
||||
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
|
||||
|
||||
print(publisher_infos)
|
||||
|
||||
|
||||
|
||||
def main(args):
|
||||
|
||||
if args.action == "enum-publishers":
|
||||
enum_publishers(args)
|
||||
|
||||
elif args.action == "get-publisher":
|
||||
get_publisher(args)
|
||||
else:
|
||||
raise NotImplementedError("Unknown action : %s" % args.action)
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser("wevtutil script reimplementation using PythonForWindows")
|
||||
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
|
||||
|
||||
|
||||
args = parser.parse_args()
|
||||
main(args)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user