将 Python win32evtlog 对象转换为 xml

Converting Python win32evtlog objects to xml

我有一个使用 win32evtlog 来获取和显示不同事件的应用程序,我想将显示限制为特定级别的事件,但 win32evtlog 不这样做 return。似乎您可以将事件转换为 XML 然后提取此信息,但我无法弄清楚如何将事件从循环中获取到 XML.

我可以达到以下目标并使用它来显示 LogObject 具有的数据,例如 LogObject.TimeGenerated

Log = win32evtlog.OpenEventLog('localhost', 'Application')
while 1:
    LogObjects = winev32tlog.ReadEventLog(Log, win32evtlog.EVENTLOG_BACKWARDS_READ|wine32vtlog.EVENTLOG_SEQUENTIAL_READ, 0)
    if not LogObjects:
        break
    for LogObject in LogObjects:

我尝试使用

进行转换
LogObjectXML = win32evtlog.EvtRender(LogObject, 1)

这很不幸returns

TypeError: The object is not a PyHANDLE object

所以我知道我需要获得某种句柄对象,我可以使用它来将 EvtRender 指向正确的事件,但我不知道该怎么做。

这个问题与 How retrieve from Python win32evtlog rest of info? 非常相似,但是那里的解决方案没有回答我们如何将对象转换为 XML 的关键部分。

--== 编辑了 CristiFati 的 XML ==--

下面是应用程序事件的示例,其中事件消息为:-

Updated Windows Defender status successfully to SECURITY_PRODUCT_STATE_ON.

根据事件查看器的XML如下

- <Event xmlns="http://schemas.microsoft.com/win/2004/08/events/event">
- <System>
  <Provider Name="SecurityCenter" /> 
  <EventID Qualifiers="0">15</EventID> 
  <Level>4</Level> 
  <Task>0</Task> 
  <Keywords>0x80000000000000</Keywords> 
  <TimeCreated SystemTime="2017-05-23T07:36:27.627108000Z" /> 
  <EventRecordID>49419</EventRecordID> 
  <Channel>Application</Channel> 
  <Computer>Name.domain.here</Computer> 
  <Security /> 
  </System>
- <EventData>
  <Data>Windows Defender</Data> 
  <Data>SECURITY_PRODUCT_STATE_ON</Data> 
  </EventData>
  </Event>

ReadEventLog returns PyEventLogRecords(更准确地说是对 [MS.Docs]: _EVENTLOGRECORD structure), while EvtRender expects (you need to work with) PyHANDLEs (PyEVT_HANDLEs (wrapper over EVT_HANDLE ([MS.Docs]: Windows Event Log Data Types 的包装))。 =94=] 因此,要获取 XML 数据,您需要使用适用于此类型的函数系列:例如EvtQuery, EvtNext.

code.py:

#!/usr/bin/env python3

import sys
import pywintypes
import win32evtlog

INFINITE = 0xFFFFFFFF
EVTLOG_READ_BUF_LEN_MAX = 0x7FFFF


def get_record_data(eventlog_record):
    ret = dict()
    for key in dir(eventlog_record):
        if 'A' < key[0] < 'Z':
            ret[key] = getattr(eventlog_record, key)
    return ret


def get_eventlogs(source_name="Application", buf_size=EVTLOG_READ_BUF_LEN_MAX, backwards=True):
    ret = list()
    evt_log = win32evtlog.OpenEventLog(None, source_name)
    read_flags = win32evtlog.EVENTLOG_SEQUENTIAL_READ
    if backwards:
        read_flags |= win32evtlog.EVENTLOG_BACKWARDS_READ
    else:
        read_flags |= win32evtlog.EVENTLOG_FORWARDS_READ
    offset = 0
    eventlog_records = win32evtlog.ReadEventLog(evt_log, read_flags, offset, buf_size)
    while eventlog_records:
        ret.extend(eventlog_records)
        offset += len(eventlog_records)
        eventlog_records = win32evtlog.ReadEventLog(evt_log, read_flags, offset, buf_size)
    win32evtlog.CloseEventLog(evt_log)
    return ret


def get_events_xmls(channel_name="Application", events_batch_num=100, backwards=True):
    ret = list()
    flags = win32evtlog.EvtQueryChannelPath
    if backwards:
        flags |= win32evtlog.EvtQueryReverseDirection
    try:
        query_results = win32evtlog.EvtQuery(channel_name, flags, None, None)
    except pywintypes.error as e:
        print(e)
        return ret
    events = win32evtlog.EvtNext(query_results, events_batch_num, INFINITE, 0)
    while events:
        for event in events:
            ret.append(win32evtlog.EvtRender(event, win32evtlog.EvtRenderEventXml))
        events = win32evtlog.EvtNext(query_results, events_batch_num, INFINITE, 0)
    return ret


def main():
    import sys, os
    from collections import OrderedDict
    standard_log_names = ["Application", "System", "Security"]
    source_channel_dict = OrderedDict()

    for item in standard_log_names:
        source_channel_dict[item] = item

    for item in ["Windows Powershell"]: # !!! This works on my machine (96 events)
        source_channel_dict[item] = item

    for source, channel in source_channel_dict.items():
        print(source, channel)
        logs = get_eventlogs(source_name=source)
        xmls = get_events_xmls(channel_name=channel)
        #print("\n", get_record_data(logs[0]))
        #print(xmls[0])
        #print("\n", get_record_data(logs[-1]))
        #print(xmls[-1])
        print(len(logs))
        print(len(xmls))

if __name__ == "__main__":
    print("Python {:s} on {:s}\n".format(sys.version, sys.platform))
    main()

备注:

  • 2 个列表的长度应该相同。每个函数中的 nth 条目应该引用相同的事件(只要两个函数都以相同的值调用 向后 论点(阅读下文))
  • get_events_xmls:
    • Returns 与事件关联的 XML blob 列表
    • 错误处理不是最好的,您可以将所有 API 调用包装在 try / except 子句中(我没有t 运行 进入错误,所以我不确定在什么情况下会引发异常)
    • 你可以稍微玩一下[MS.Docs]: EvtNext function的参数(TimeoutEventsSize来进行性能微调;对于我,~20k 事件在 <10 秒内处理 - 其中文本打印和转换花费最多)
    • Python3中,XMLbytes ([Python 3.Docs]: Built-in Types - class bytes([source[, encoding[, errors]]])) 而不是普通字符串(我不得不对它们进行编码,因为有些包含一些非 ASCII 字符,并尝试打印它们会引发 UnicodeEncodeError)
    • 可以进行事件过滤,检查 [MS.Docs]: EvtQuery function 的参数(FlagsQuery
    • 请注意 backwards 参数,它允许以相反的(时间顺序)顺序遍历事件(默认设置为 True)。
  • get_record_data:
    • 这只是一个方便的函数,它将一个 PyEventLogRecord 对象转换成一个 Python 字典
    • 转换是基于我们关心的字段以大写字母开头(EventID,ComputerName, TimeGenerated, ...), 这就是为什么它不应在生产中使用
    • 它不转换实际值(TimeGenerated 的值为 pywintypes.datetime(2017, 3, 11, 3, 46, 47)
  • get_eventlogs:
  • 因为我将所有数据存储在 2 个列表中(而不是就地数据处理),所以我选择速度而不是内存消耗。对于 ~20K 事件,这两个列表占用了 ~30MB of RAM(现在我认为这是足够体面)

@EDIT0:我找不到使用 Evt* 获取所有必需信息的方法 函数系列,所以我从两个来源获取它(我增强了我之前发布的脚本):

@EDIT1:根据[MS.Docs]: OpenEventLogW function

If you specify a custom log and it cannot be found, the event logging service opens the Application log; however, there will be no associated message or category string file.

[MS.Docs]: Eventlog Key 列出了 3 个 standard。所以,这就是它打开 Application 日志的原因。我对脚本做了一些小改动来测试源代码。我不知道 mmc 从哪里得到 Setup 事件。