import ctypes
import os.path
import xml.dom.minidom
from contextlib import contextmanager
import windows
import windows.generated_def as gdef
from windows import winproxy
# Helpers
@contextmanager
def ClosingEvtHandle(handle):
try:
yield handle
finally:
winproxy.EvtClose(handle)
# low-level api helpers
def queryinfo(handle, propertyid):
size = 0x1000
buffer = ctypes.create_string_buffer(size)
evt = ImprovedEVT_VARIANT.from_buffer(buffer)
res = gdef.DWORD()
windows.winproxy.EvtGetLogInfo(handle, propertyid, size, evt, res)
return evt
def arrayproperty(handle, property, index, flags=0):
size = 0x1000
buffer = ctypes.create_string_buffer(size)
evt = ImprovedEVT_VARIANT.from_buffer(buffer)
res = gdef.DWORD()
windows.winproxy.EvtGetObjectArrayProperty(handle, property, index, flags, size, evt, res)
return evt
def generate_query_function(query_function):
def generated_query_function(handle, propertyid, flags=0):
size = 0x10000
buffer = ctypes.create_string_buffer(size)
evt = ImprovedEVT_VARIANT.from_buffer(buffer)
res = gdef.DWORD()
query_function(handle, propertyid, flags, size, evt, res)
return evt
return generated_query_function
chaninfo = generate_query_function(winproxy.EvtGetChannelConfigProperty)
eventinfo = generate_query_function(winproxy.EvtGetEventMetadataProperty)
publishinfo = generate_query_function(winproxy.EvtGetPublisherMetadataProperty)
# Class high-level API
[docs]class EvtQuery(gdef.EVT_HANDLE):
"""Represent an Event-log query"""
TIMEOUT = 0x1000
def __init__(self, handle=0, channel=None):
super(EvtQuery, self).__init__(handle)
self.channel = channel
def __next__(self):
"""Return the next :class:`EvtEvent` matching the query"""
try:
event = EvtEvent(channel=self.channel)
ret = gdef.DWORD()
windows.winproxy.EvtNext(self, 1, event, self.TIMEOUT, 0, ret)
except WindowsError as e:
if e.winerror == gdef.ERROR_NO_MORE_ITEMS:
raise StopIteration
raise
assert ret.value == 1
return event
def __iter__(self):
return self
next = __next__ # Yep.. real name is 'next' in Py2 :D
[docs] def all(self): # SqlAlchemy like :)
"""Return a list with all the query results
:rtype: [:class:`EvtEvent`] -- A list of Event
"""
return list(self)
[docs] def first(self): # SqlAlchemy like :) -> allow testing in interactive console
"""Return the first query result
:rtype: :class:`EvtEvent` -- An Event
"""
return next(iter(self))
[docs]class EvtEvent(gdef.EVT_HANDLE):
"""An Event log"""
def __init__(self, handle=0, channel=None):
super(EvtEvent, self).__init__(handle)
self.channel = channel
def render(self, ctx, rtype):
size = 0x10000
buffer = ctypes.c_buffer(size)
rsize = gdef.DWORD()
elementnb = gdef.DWORD()
try:
windows.winproxy.EvtRender(ctx, self, rtype, size, buffer, rsize, elementnb)
except WindowsError as e:
if e.winerror != gdef.ERROR_INSUFFICIENT_BUFFER:
raise
size = rsize.value
buffer = ctypes.c_buffer(size)
windows.winproxy.EvtRender(ctx, self, rtype, size, buffer, rsize, elementnb)
# Adapting return value type
if rtype != gdef.EvtRenderEventValues:
# import pdb;pdb.set_trace()
# assert elementnb.value == 1
return buffer[:rsize.value]
# print("Got <{0}> elt".format(elementnb.value))
return list((ImprovedEVT_VARIANT * elementnb.value).from_buffer(buffer))
def render_xml(self):
xml = self.render(None, 1).decode("utf-16")
assert xml[-1] == "\x00"
return xml[:-1]
[docs] def value(self, name, **kwargs):
"""Retrieve a value from the event.
``name`` is an XPath expressions that uniquely identify a node or attribute in the event.
(see https://msdn.microsoft.com/en-us/library/windows/desktop/aa385352(v=vs.85).aspx)
"""
values = self.get_values((name,), **kwargs)
assert len(values) == 1
return values[0]
def get_values(self, values, flags=gdef.EvtRenderContextValues):
nbelt = len(values)
pwstr_values = tuple(gdef.LPWSTR(v) for v in values)
pwstr_rarray = (gdef.LPWSTR * nbelt)(*pwstr_values)
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa385352(v=vs.85).aspx
# An array of XPath expressions that uniquely identify a node or attribute in the event that you want to render.
# Each value wil return 1 node :)
ctx = windows.winproxy.EvtCreateRenderContext(nbelt, pwstr_rarray, gdef.EvtRenderContextValues)
result = self.render(ctx, gdef.EvtRenderEventValues)
return [r.value for r in result]
def system_values(self): # POC: use this for all @property based on system data ?
ctx = windows.winproxy.EvtCreateRenderContext(0, None, gdef.EvtRenderContextSystem)
result = self.render(ctx, gdef.EvtRenderEventValues)
return [r.value for r in result]
def event_values(self):
ctx = windows.winproxy.EvtCreateRenderContext(0, None, gdef.EvtRenderContextUser)
result = self.render(ctx, gdef.EvtRenderEventValues)
return [r.value for r in result]
def get_raw_values(self, values, flags=gdef.EvtRenderContextValues):
nbelt = len(values)
pwstr_values = tuple(gdef.LPWSTR(v) for v in values)
pwstr_rarray = (gdef.LPWSTR * nbelt)(*pwstr_values)
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa385352(v=vs.85).aspx
# An array of XPath expressions that uniquely identify a node or attribute in the event that you want to render.
# Each value wil return 1 node :)
ctx = windows.winproxy.EvtCreateRenderContext(nbelt, pwstr_rarray, gdef.EvtRenderContextValues)
result = self.render(ctx, gdef.EvtRenderEventValues)
return list(result)
# Properties arround common Event/System values
@property
def provider(self):
return self.system_values()[gdef.EvtSystemProviderName]
@property
def id(self):
"""The ID of the Event"""
return self.value("Event/System/EventID")
@property
def version(self):
"""The version of the Event"""
return self.value("Event/System/Version")
@property
def level(self):
"""The level of the Event"""
return self.value("Event/System/Level")
@property
def opcode(self):
"""The opcode of the Event"""
return self.value("Event/System/Opcode")
@property
def time_created(self):
"""The creation time of the Event"""
return self.value("Event/System/TimeCreated/@SystemTime")
@property
def pid(self):
"""The process ID of the Event"""
return self.value("Event/System/Execution/@ProcessID")
@property
def tid(self):
"""The process ID of the Event"""
return self.value("Event/System/Execution/@ThreadID")
@property
def metadata(self):
"""The medata for the current Event
:type: :class:`EventMetadata`
"""
try:
return self.channel.get_event_metadata(self.id)
except KeyError as e:
if not self.channel.config.classic:
raise
# id not found: try via the Provider in the event (classic channel)
return self.channel.get_classic_event_metadata(self.id, self.provider)
# Test
@property
def data(self): # user/event specifique data
"""A dict of EventData Name:Value for the current dict.
:type: :class:`dict`
"""
# What about classic channels where there is no event_metadata ?
# Return a dict with [0-1-2-3-4] as key ? raise ?
return {k:v for k,v in zip(self.metadata.event_data, self.event_values())}
def __repr__(self):
creation_time = windows.utils.datetime_from_filetime(self.time_created)
return '<{0} id="{self.id}" time="{creation_time}">'.format(type(self).__name__, self=self, creation_time=creation_time)
class ImprovedEVT_VARIANT(gdef.EVT_VARIANT):
VALUE_MAPPER = {
gdef.EvtVarTypeNull : 'NoneValue',
gdef.EvtVarTypeString : 'StringVal',
gdef.EvtVarTypeAnsiString : 'AnsiStringVal',
gdef.EvtVarTypeSByte : 'SByteVal',
gdef.EvtVarTypeByte : 'ByteVal',
gdef.EvtVarTypeInt16 : 'Int16Val',
gdef.EvtVarTypeUInt16 : 'UInt16Val',
gdef.EvtVarTypeInt32 : 'Int32Val',
gdef.EvtVarTypeUInt32 : 'UInt32Val',
gdef.EvtVarTypeInt64 : 'Int64Val',
gdef.EvtVarTypeUInt64 : 'UInt64Val',
gdef.EvtVarTypeSingle : 'SingleVal',
gdef.EvtVarTypeDouble : 'DoubleVal',
gdef.EvtVarTypeBoolean : 'BooleanVal',
gdef.EvtVarTypeBinary : 'BinaryVal',
gdef.EvtVarTypeGuid : 'GuidVal',
gdef.EvtVarTypeSizeT : 'SizeTVal',
gdef.EvtVarTypeFileTime : 'FileTimeVal',
gdef.EvtVarTypeSysTime : 'SysTimeVal',
gdef.EvtVarTypeSid : 'SidVal',
gdef.EvtVarTypeHexInt32 : 'UInt32Val',
gdef.EvtVarTypeHexInt64 : 'UInt64Val',
gdef.EvtVarTypeEvtHandle : 'EvtHandleVal',
gdef.EvtVarTypeEvtXml : 'XmlVal',
# Array types: TODO: generic stuff
gdef.EvtVarTypeString + gdef.EVT_VARIANT_TYPE_ARRAY : "StringArr",
gdef.EvtVarTypeUInt16 + gdef.EVT_VARIANT_TYPE_ARRAY : "UInt16Arr",
gdef.EvtVarTypeUInt32 + gdef.EVT_VARIANT_TYPE_ARRAY : "UInt32Arr",
gdef.EvtVarTypeUInt64 + gdef.EVT_VARIANT_TYPE_ARRAY : "UInt64Arr",
}
NoneValue = None
@property
def Type(self):
raw_type = super(ImprovedEVT_VARIANT, self).Type
return gdef.EVT_VARIANT_TYPE.mapper.get(raw_type, raw_type)
@property
def value(self): # Prototype !!
attrname = self.VALUE_MAPPER[self.Type]
# print("Resolve type <{0}> -> {1}".format(self.Type, attrname))
v = getattr(self, attrname)
if self.Type == gdef.EvtVarTypeBinary:
v = v[:self.Count] # No need for a raw UBYTE ptr
elif self.Type == gdef.EvtVarTypeGuid:
v = v[0] # Deref LP_GUID
elif self.Type & gdef.EVT_VARIANT_TYPE_ARRAY:
# TODO: handle all array type
v = v[:self.Count]
return v
def __repr__(self):
return "<{0} of type={1}>".format(type(self).__name__, self.Type)
# x = windows.winproxy.EvtQuery(None,
# "Microsoft-Windows-Windows Firewall With Advanced Security/Firewall",
# "Event/System[EventID=2004 or EventID=2006]",
# gdef.EvtQueryChannelPath + gdef.EvtQueryForwardDirection)
# print(x)
# eq = EvtQuery(x)
# event = next(eq)
# xx = gdef.LPWSTR("ModifyingApplication")
# xx = gdef.LPWSTR("Event/System/Channel")
# xx = gdef.LPWSTR('Event/EventData/Data[@Name="ModifyingApplication"]')
# xx = gdef.LPWSTR('Event/System/EventID')
# event = next(eq)
# list(channels())
[docs]class EvtChannel(object):
"""An Event Log channel"""
DEFAULT_QUERY_FLAGS = gdef.EvtQueryChannelPath + gdef.EvtQueryForwardDirection
def __init__(self, name):
self.name = name
self.event_metadata_by_id = {}
self.classic_event_metadata_by_id = {} # For classic only
[docs] def query(self, filter=None, ids=None):
"""Query the event with the ``ids`` or perform a query with the raw query ``filter``
Both parameters are mutually exclusive.
.. note:: Here are some query examples
List all events with a event data attribute named 'RuleName':
``Event/EventData/Data[@Name='RuleName']``
List all events with a event data value of 'C:\\\\WINDOWS\\\\System32\\\\svchost.exe':
``Event/EventData[Data='C:\\WINDOWS\\System32\\svchost.exe']``
List all events with an EventID of 2006:
``Event/System[EventID=2006]``
:rtype: :class:`EvtQuery`
"""
if ids and filter:
raise ValueError("<ids> and <filter> are mutually exclusive")
if ids is not None:
if isinstance(ids, (long, int)):
ids = (ids,)
ids_filter = " or ".join("EventID={0}".format(id) for id in ids)
filter = "Event/System[{0}]".format(ids_filter)
query_handle = winproxy.EvtQuery(None, self.name, filter, self.DEFAULT_QUERY_FLAGS)
return EvtQuery(query_handle, self)
@property
def events(self):
"""The list of all events in the channels, an alias for ``channel.query().all()``
:type: [:class:`EvtEvent`] -- A list of :class:`EvtEvent`
"""
return self.query().all()
@property
def config(self):
"""The configuration of the channel
:type: :class:`ChannelConfig`
"""
return ChannelConfig.from_channel_name(self.name)
def get_classic_event_metadata(self, id, providername):
if providername not in self.classic_event_metadata_by_id:
# print("CALCUL FOR PROVIDER: <{0}> !!!!!!!!!!!!!!".format(providername))
publisher = EvtPublisher(providername)
events_metadata = {x.id: x for x in publisher.metadata.events_metadata}
self.classic_event_metadata_by_id[providername] = events_metadata
return self.classic_event_metadata_by_id[providername][id]
def __repr__(self):
return '<{0} "{1}">'.format(type(self).__name__, self.name)
[docs]class EvtFile(EvtChannel):
"""Represent an Evtx file"""
DEFAULT_QUERY_FLAGS = gdef.EvtQueryFilePath + gdef.EvtQueryForwardDirection
@property
def config(self):
"""Not implemented for EvtFile
:raise: :class:`NotImplementedError`
"""
raise NotImplementedError("Cannot retrieve the configuration of an EvtFile")
[docs]class ChannelConfig(gdef.EVT_HANDLE):
"""The configuration of a event channel"""
def __init__(self, handle, name=None):
super(ChannelConfig, self).__init__(handle)
self.name = name
[docs] @classmethod
def from_channel_name(cls, name):
"""Return the :class:`ChannelConfig` for the channel ``name``"""
return cls(winproxy.EvtOpenChannelConfig(None, name, 0), name)
@property
def publisher(self):
"""The :class:`EvtPublisher` for the channel"""
return EvtPublisher(chaninfo(self, gdef.EvtChannelConfigOwningPublisher).value)
[docs] def publishers(self):
"TEST"
return [EvtPublisher(pub) for pub in chaninfo(self, gdef.EvtChannelPublisherList).value]
@property
def classic(self):
"""``True`` if the channel is a classic event channel (for example the Application or System log)"""
return bool(chaninfo(self, gdef.EvtChannelConfigClassicEventlog).value)
def __repr__(self):
return '<{0} "{1}">'.format(type(self).__name__, self.name)
[docs]class EvtPublisher(object):
"""An Event provider"""
def __init__(self, name):
self.name = name
@property
def metadata(self):
"""Return the metadata for this publisher
:type: :class:`PublisherMetadata`
"""
return PublisherMetadata.from_publisher_name(self.name)
def __repr__(self):
return '<{0} "{1}">'.format(type(self).__name__, self.name)
[docs]class PropertyArray(gdef.EVT_OBJECT_ARRAY_PROPERTY_HANDLE):
"TODO"
@property
def size(self):
array_size = gdef.DWORD()
windows.winproxy.EvtGetObjectArraySize(self, array_size)
return array_size.value
def property(self, type, index):
return arrayproperty(self, type, index).value
[docs]class EvtlogManager(object):
"""The main Evt class to open Evt channel/publisher and evtx file"""
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa385784(v=vs.85).aspx
[docs] def is_implemented(self):
"""Return ``True`` if the new Evt-API is implemented on the current computer
see: https://msdn.microsoft.com/en-us/library/windows/desktop/aa385784(v=vs.85).aspx
"""
return windows.winproxy.is_implemented(windows.winproxy.EvtQuery)
@property
def channels(self):
h = windows.winproxy.EvtOpenChannelEnum(None, 0)
size = 0x1000
buffer = ctypes.create_unicode_buffer(size)
ressize = gdef.DWORD()
with ClosingEvtHandle(h):
while True:
try:
windows.winproxy.EvtNextChannelPath(h, size, buffer, ressize)
except WindowsError as e:
if e.winerror != gdef.ERROR_NO_MORE_ITEMS:
raise
return
assert buffer[ressize.value - 1] == "\x00"
name = buffer[:ressize.value - 1]
chan = EvtChannel(name)
yield chan
@property
def publishers(self):
h = windows.winproxy.EvtOpenPublisherEnum(None, 0)
size = 0x1000
buffer = ctypes.create_unicode_buffer(size)
ressize = gdef.DWORD()
with ClosingEvtHandle(h):
while True:
try:
windows.winproxy.EvtNextPublisherId(h, size, buffer, ressize)
except WindowsError as e:
if e.winerror != gdef.ERROR_NO_MORE_ITEMS:
raise
return
assert buffer[ressize.value - 1] == "\x00"
name = buffer[:ressize.value - 1]
publisher = EvtPublisher(name)
yield publisher
[docs] def open_channel(self, name):
"""Open the Evt channel with ``name``
:rtype: :class:`EvtChannel`
"""
chan = EvtChannel(name)
chan.config # Force to retrieve a handle (check channel exists)
return chan
[docs] def open_evtx_file(self, filename):
"""Open the evtx file with ``filename``
:rtype: :class:`EvtFile`
"""
with windows.utils.DisableWow64FsRedirection():
if not os.path.exists(filename):
raise WindowsError(gdef.ERROR_FILE_NOT_FOUND, "Could not find file <{0}>".format(filename))
file = EvtFile(filename)
return file
[docs] def open_publisher(self, name):
"""Open the Evt publisher with ``name``
:rtype: :class:`EvtPublisher`
"""
publisher = EvtPublisher(name)
publisher.metadata # Force to retrieve a handle (check channel exists)
return publisher
[docs] def __getitem__(self, name):
"""Open the Evt Channel/Publisher or Evtx file with ``name``
:rtype: :class:`EvtChannel` or :class:`EvtPublisher` or :class:`EvtFile`
"""
try:
return self.open_channel(name)
except WindowsError as e:
if e.winerror == gdef.ERROR_ACCESS_DENIED:
raise
try:
return self.open_publisher(name)
except WindowsError as e:
if e.winerror == gdef.ERROR_ACCESS_DENIED:
raise
# Raise FILE_NOT_FOUND if not found (last chance)
return self.open_evtx_file(name)
# CHANNAME = "Microsoft-Windows-Windows Firewall With Advanced Security/Firewall"
# CHANNAME = r"Microsoft-Windows-Windows Defender/Operational"
# print("Working of channel: <{0}>".format(CHANNAME))
# channel = EvtChannel(CHANNAME)
# chanconf = channel.config
# publisher = chanconf.publisher
# open publisher metadata
# print("Provider is <{0}>".format(publisher.name))
# pmd = publisher.metadata
# chansref = pmd.chanrefs
# channame_by_value_id = pmd.channel_name_by_id()
# for event_metadata in pmd.events_metadata:
# id = event_metadata.id
# if id == 2004:
# print("LOL")
# chan = event_metadata.channel_id
# channame = channame_by_value_id[chan]
# print(" * {0}) {1}".format(id, channame))
# if "unexpected" in pmd.message(event_metadata.message_id).lower():
# print("UNEXPECTED in message :D")
# query = channel.query(ids=5008)
# evts = list(query)