""" A module for working with the Windows Event log system. .. versionadded:: 3006.0 """ # https://docs.microsoft.com/en-us/windows/win32/eventlog/event-logging import collections import logging import salt.utils.platform import salt.utils.stringutils from salt.exceptions import CommandExecutionError try: import pywintypes import win32evtlog import win32evtlogutil import winerror # Only windows needs this dependency at runtime import xmltodict IMPORT_STATUS = True except ImportError: IMPORT_STATUS = False log = logging.getLogger(__name__) __virtualname__ = "win_event" def __virtual__(): """ Load only on minions running on Windows. """ if not salt.utils.platform.is_windows(): return False, "win_event: Must be on Windows" if not IMPORT_STATUS: return False, "win_event: Missing PyWin32" return __virtualname__ def _to_bytes(data, encoding="utf-8", encode_keys=False): """ Convert string objects to byte objects. .. warning:: This function will destroy the data object and objects that data links to. Args: data (object): The string object to encode encoding (:obj:`str`, optional): The encoding type. Default is "utf-8". encode_keys (:obj:`bool`, optional): If ``False``, key strings will not be encoded. Defaults is ``False``. Returns: (object): An object with the new encoding """ if isinstance(data, dict): new_dict = {} # recursively check every item in the dict for key in data: item = _to_bytes(data[key], encoding) if encode_keys: # keys that are strings most be made into bytes key = _to_bytes(key, encoding) new_dict[key] = item data = new_dict elif isinstance(data, list): new_list = [] # recursively check every item in the list for item in data: new_list.append(_to_bytes(item, encoding)) data = new_list elif isinstance(data, tuple): new_list = [] # recursively check every item in the tuple for item in data: new_list.append(_to_bytes(item, encoding)) data = tuple(new_list) elif isinstance(data, str): # encode string data to bytes data = data.encode(encoding) return data def _raw_time(time): """ Will make a pywintypes.datetime into a TimeTuple. Args: time (obj): A datetime object Returns: TimeTuple: A TimeTuple """ TimeTuple = collections.namedtuple( "TimeTuple", "year, month, day, hour, minute, second" ) return TimeTuple( time.year, time.month, time.day, time.hour, time.minute, time.second ) def _make_event_dict(event): """ Will make a PyEventLogRecord into a dictionary Args: event (PyEventLogRecord): An event to convert to a dictionary Returns: dict: A dictionary containing the event information """ # keys of all the parts of a Event supported by the API event_parts = ( "closingRecordNumber", "computerName", "data", "eventCategory", "eventID", "eventType", "recordNumber", "reserved", "reservedFlags", "sid", "sourceName", "stringInserts", "timeGenerated", "timeWritten", ) event_dict = {} for event_part in event_parts: # get object value and add it to the event dict event_dict[event_part] = getattr( event, event_part[0].upper() + event_part[1:], None ) # format items event_dict["eventID"] = winerror.HRESULT_CODE(event_dict["eventID"]) if event_dict["sid"] is not None: event_dict["sid"] = event_dict["sid"].GetSidIdentifierAuthority() event_dict["timeGenerated"] = _raw_time(event_dict["timeGenerated"]) event_dict["timeWritten"] = _raw_time(event_dict["timeWritten"]) return _to_bytes(event_dict) def _get_handle(log_name): """ Will try to open a PyHANDLE to the Event System Args: log_name (str): The name of the log to open Returns: PyHANDLE: A handle to the event log """ # TODO: upgrade windows token # "log close" can fail if this is not done try: return win32evtlog.OpenEventLog(None, log_name) except pywintypes.error as exc: raise FileNotFoundError( f"Failed to open log: {log_name}\nError: {exc.strerror}" ) def _close_handle(handle): """ Will close the handle to the event log Args: handle (PyHANDLE): The handle to the event log to close """ # TODO: downgrade windows token win32evtlog.CloseEventLog(handle) def _event_generator(log_name): """ Get all log events one by one. Events are not ordered Args: log_name(str): The name of the log to retrieve Yields: dict: A dictionary object for each event """ # Get events from the local machine (None) handle = _get_handle(log_name) flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ while True: # get list of some of the events events = win32evtlog.ReadEventLog(handle, flags, 0) if not events: # event log was updated and events are not ready to be given yet # rather than wait just return break for event in events: yield _make_event_dict(event) _close_handle(handle) def _event_generator_with_time(log_name): """ Sorts the results of the event generator Args: log_name (str): The name of the log to retrieve Yields: dict: A dictionary object for each event """ # keys time time_parts = ( "year", "month", "day", "hour", "minute", "second", ) for event in _event_generator(log_name): event_info = {} for part in event: event_info[part] = event[part] for spot, key in enumerate(time_parts): event_info[key] = event["timeGenerated"][spot] yield event, event_info def _event_generator_filter(log_name, all_requirements=True, **kwargs): """ Will find events that meet the requirements in the filter. Can be any item in the return for the event. Args: log_name (str): The name of the log to retrieve all_requirements (bool): Should the results match all requirements. ``True`` matches all requirements. ``False`` matches any requirement. Kwargs: eventID (int): The event ID number eventType (int): The event type number. Valid options and their corresponding meaning are: - 0 : Success - 1 : Error - 2 : Warning - 4 : Information - 8 : Audit Success - 10 : Audit Failure year (int): The year month (int): The month day (int): The day of the month hour (int): The hour minute (int): The minute second (int): The second eventCategory (int): The event category number sid (sid): The SID of the user that created the event sourceName (str): The name of the event source Yields: dict: A dictionary object for each event CLI Example: .. code-block:: python # Return all events from the Security log with an ID of 1100 _event_generator_filter("Security", eventID=1100) # Return all events from the System log with an Error (1) event type _event_generator_filter("System", eventType=1) # Return all events from System log with an Error (1) type, source is Service Control Manager, and data is netprofm _event_generator_filter("System", eventType=1, sourceName="Service Control Manager", data="netprofm") """ for event, info in _event_generator_with_time(log_name): if all_requirements: # all keys need to match each other for key in kwargs: # ignore kwargs built-ins if key.startswith("__"): continue # ignore function parameters if key in ["log_name", "all_arguments"]: continue # Try to handle bytestrings if isinstance(info[key], bytes): # try utf-8 first try: log.trace( "utf-8: Does %s == %s", repr(kwargs[key]), repr(info[key].decode("utf-8")), ) if kwargs[key] != info[key].decode("utf-8"): # try utf-16 and strip null bytes try: log.trace( "utf-16: Does %s == %s", repr(kwargs[key]), repr(info[key].decode("utf-16").strip("\x00")), ) if kwargs[key] != info[key].decode("utf-16").strip( "\x00" ): break except UnicodeDecodeError: log.trace("Failed to decode (utf-16): %s", info[key]) break except UnicodeDecodeError: log.trace("Failed to decode (utf-8): %s", info[key]) break elif kwargs[key] != info[key]: break else: yield info else: # just a single key pair needs to match for key in kwargs: # ignore kwargs built-ins if key.startswith("__"): continue # ignore function parameters if key in ["log_name", "all_arguments"]: continue # Try to handle bytestrings if isinstance(info[key], bytes): # try utf-8 first try: log.trace( "utf-8: Does %s == %s", repr(kwargs[key]), repr(info[key].decode("utf-8")), ) if kwargs[key] == info[key].decode("utf-8"): yield info except UnicodeDecodeError: log.trace("Failed to decode (utf-8): %s", info[key]) # try utf-16 and strip null bytes try: log.trace( "utf-16: Does %s == %s", repr(kwargs[key]), repr(info[key].decode("utf-16").strip("\x00")), ) if kwargs[key] == info[key].decode("utf-16").strip("\x00"): yield info except UnicodeDecodeError: log.trace("Failed to decode (utf-16): %s", info[key]) break elif kwargs[key] == info[key]: yield info def get(log_name): """ Get events from the specified log. Get a list of available logs using the :py:func:`win_event.get_log_names ` function. .. warning:: Running this command on a log with thousands of events, such as the ``Applications`` log, can take a long time. Args: log_name (str): The name of the log to retrieve. Returns tuple: A tuple of events as dictionaries CLI Example: .. code-block:: bash salt '*' win_event.get Application """ return tuple(_event_generator(log_name)) def query(log_name, query_text=None, records=20, latest=True, raw=False): """ Query a log for a specific event_id. Return the top number of records specified. Use the :py:func:`win_event.get_log_names ` to see a list of available logs on the system. .. Note:: You can use the Windows Event Viewer to create the XPath query for the ``query_text`` parameter. Click on ``Filter Current Log``, configure the filter, then click on the XML tab. Copy the text between the two ``