Plugin Class Reference

Inheritance diagram for Plugin:

Collaboration diagram for Plugin:

Public Member Functions
str	desc (self)
GulpRequestStatus	ingest (self, str index, str req_id, int client_id, int operation_id, str context, str|list[dict] source, str ws_id, GulpPluginParams plugin_params=None, GulpIngestionFilter flt=None, **kwargs)
str	name (self)
list[GulpDocument]	record_to_gulp_document (self, int operation_id, int client_id, str context, str source, TmpIngestStats fs, any record, int record_idx, GulpMapping custom_mapping=None, dict index_type_mapping=None, str plugin=None, GulpPluginParams plugin_params=None, **kwargs)
GulpPluginType	type (self)
str	version (self)
Public Member Functions inherited from PluginBase
None	__init__ (self, str path, AsyncEngine collab=None, AsyncElasticsearch elastic=None, **kwargs)
None	cleanup (self)
list[str]	depends_on (self)
str	event_type_field (self)
str	get_unmapped_field_name (self, str field)
tuple[dict, GulpMapping]	ingest_plugin_initialize (self, str index, str|dict source, bool skip_mapping=False, ProcessingPipeline pipeline=None, str mapping_file=None, str mapping_id=None, GulpPluginParams plugin_params=None)
bool	internal (self)
logging.Logger	logger (cls)
list[GulpPluginOption]	options (self)
ProcessingPipeline	pipeline (self, GulpPluginParams plugin_params=None, **kwargs)
dict	run_command (self, GulpPluginParams p)
ProcessingPipeline	sigma_plugin_initialize (self, ProcessingPipeline pipeline=None, str mapping_file=None, str mapping_id=None, str product=None, GulpPluginParams plugin_params=None)
list[str]	tags (self)

Protected Member Functions
dict	_map_evt_code (self, str ev_code)
GulpLogLevel	_normalize_loglevel (self, int|str l)
Protected Member Functions inherited from PluginBase
list[GulpDocument]	_build_gulpdocuments (self, list[FieldMappingEntry] fme, int idx, int operation_id, str context, str plugin, int client_id, str raw_event, str original_id, str src_file, int timestamp=None, int timestamp_nsec=None, str event_code=None, list[str] cat=None, int duration_nsec=0, GulpLogLevel gulp_log_level=None, str original_log_level=None, bool remove_raw_event=False, **kwargs)
list[dict]	_build_ingestion_chunk_for_ws (self, list[dict] docs, GulpIngestionFilter flt=None)
list[GulpDocument]	_call_record_to_gulp_document_funcs (self, int operation_id, int client_id, str context, str source, TmpIngestStats fs, any record, int record_idx, GulpMapping custom_mapping=None, dict index_type_mapping=None, str plugin=None, GulpPluginParams plugin_params=None, Callable record_to_gulp_document_fun=None, **kwargs)
GulpRequestStatus	_finish_ingestion (self, str index, str|dict source, str req_id, int client_id, str ws_id, TmpIngestStats fs, GulpIngestionFilter flt=None)
TmpIngestStats	_flush_buffer (self, str index, TmpIngestStats fs, str ws_id, str req_id, GulpIngestionFilter flt=None, bool wait_for_refresh=False)
TmpIngestStats	_ingest_record (self, str index, GulpDocument|dict doc, TmpIngestStats fs, str ws_id, str req_id, GulpIngestionFilter flt=None, bool flush_enabled=True, **kwargs)
list[FieldMappingEntry]	_map_source_key (self, GulpPluginParams plugin_params, GulpMapping custom_mapping, str source_key, Any v, dict index_type_mapping=None, bool ignore_custom_mapping=False, **kwargs)
TmpIngestStats	_parser_failed (self, TmpIngestStats fs, str|dict source, Exception|str ex)
tuple[GulpMapping, GulpPluginParams]	_process_plugin_params (self, GulpMapping custom_mapping, GulpPluginParams plugin_params=None)
tuple[TmpIngestStats, bool]	_process_record (self, str index, any record, int record_idx, Callable my_record_to_gulp_document_fun, str ws_id, str req_id, int operation_id, int client_id, str context, str source, TmpIngestStats fs, GulpMapping custom_mapping=None, dict index_type_mapping=None, str plugin=None, GulpPluginParams plugin_params=None, GulpIngestionFilter flt=None, **kwargs)
TmpIngestStats	_record_failed (self, TmpIngestStats fs, any entry, str|dict source, Exception|str ex)
any	_type_checks (self, any v, str k, dict index_type_mapping)

Additional Inherited Members
Public Attributes inherited from PluginBase
list	buffer = []
str	client_id = None
	collab = collab
tuple	context = (None,)
	elastic = elastic
str	index = None
str	operation_id = None
	path = path
str	req_id = None
str	ws_id = None

Detailed Description

windows evtx log file processor.

Definition at line 23 of file win_evtx.py.

Member Function Documentation

_map_evt_code()

dict _map_evt_code ( self, str ev_code )

protected

better map an event code to fields

Args:
    ev_code (str): The event code to be converted.
Returns:
    dict: A dictionary with 'event.category' and 'event.type'

Definition at line 64 of file win_evtx.py.

    def _map_evt_code(self, ev_code: str) -> dict:
        """
        better map an event code to fields
 
        Args:
            ev_code (str): The event code to be converted.
        Returns:
            dict: A dictionary with 'event.category' and 'event.type'
        """
        codes = {
            "100": {"event.category": ["package"], "event.type": ["start"]},
            "106": {"event.category": ["package"], "event.type": ["install"]},
            "140": {"event.category": ["package"], "event.type": ["change"]},
            "141": {"event.category": ["package"], "event.type": ["delete"]},
            "1006": {"event.category": ["host"], "event.type": ["change"]},
            "4624": {  # eventid
                "event.category": ["authentication"],
                "event.type": ["start"],
            },
            "4672": {
                "event.category": ["authentication"],
            },
            "4648": {
                "event.category": ["authentication"],
            },
            "4798": {"event.category": ["iam"]},
            "4799": {"event.category": ["iam"]},
            "5379": {"event.category": ["iam"], "event.type": ["access"]},
            "5857": {"event.category": ["process"], "event.type": ["access"]},
            "5858": {"event.category": ["process"], "event.type": ["error"]},
            "5859": {"event.category": ["process"], "event.type": ["change"]},
            "5860": {"event.category": ["process"], "event.type": ["change"]},
            "5861": {"event.category": ["process"], "event.type": ["change"]},
            "7036": {
                "event.category": ["package"],
                "event.type": ["change"],
            },
            "7040": {
                "event.category": ["package"],
                "event.type": ["change"],
            },
            "7045": {
                "event.category": ["package"],
                "event.type": ["install"],
            },
            "13002": {"event.type": ["change"]},
        }
        if ev_code in codes:
            return codes[ev_code]
 
        return None
 

Here is the caller graph for this function:

_normalize_loglevel()

GulpLogLevel _normalize_loglevel ( self, int | str l )

protected

Definition at line 28 of file win_evtx.py.

    def _normalize_loglevel(self, l: int | str) -> GulpLogLevel:
        ll = int(l)
        if ll == 0:
            return GulpLogLevel.ALWAYS
        if ll == 1:
            return GulpLogLevel.CRITICAL
        if ll == 2:
            return GulpLogLevel.ERROR
        if ll == 3:
            return GulpLogLevel.WARNING
        if ll == 4:
            return GulpLogLevel.INFO
        if ll == 5:
            return GulpLogLevel.VERBOSE
        if ll > 5:
            # maps directly to GulpLogLevel.CUSTOM_n
            try:
                return GulpLogLevel(ll)
            except:
                return GulpLogLevel.UNEXPECTED
 
        # wtf ?!
        return GulpLogLevel.VERBOSE
 

Here is the caller graph for this function:

desc()

str desc

(

self

)

Returns a description of the plugin.

Reimplemented from PluginBase.

Definition at line 55 of file win_evtx.py.

    def desc(self) -> str:
        return "Windows EVTX log file processor."
 

Here is the caller graph for this function:

ingest()

GulpRequestStatus ingest	(		self,
		str	index,
		str	req_id,
		int	client_id,
		int	operation_id,
		str	context,
		str | list[dict]	source,
		str	ws_id,
		GulpPluginParams	plugin_params = None,
		GulpIngestionFilter	flt = None,
		**	kwargs )

Ingests a file using the plugin.

NOTE: implementers should call super().ingest() in their implementation.
NOTE: this function *SHOULD NOT* raise exceptions

Args:
    index (str): name of the elasticsearch index to ingest the document into.
    req_id (str): The request ID related to this ingestion (must exist on the collab db).
    client_id (int): The client ID performing the ingestion.
    operation_id (int): The operation ID related to this ingestion.
    context (str): Context related to this ingestion.
    source (str|list[dict]): The path to the file to ingest, or a list of events dicts.
    ws_id (str): The websocket ID
    plugin_params (GulpPluginParams): additional parameters to pass to the ingestion function. Defaults to None.
    flt (GulpIngestionFilter, optional): filter to apply to this ingestion, if any. Defaults to None.
    kwargs: additional arguments if any

Reimplemented from PluginBase.

Definition at line 288 of file win_evtx.py.

    ) -> GulpRequestStatus:
 
        await super().ingest(
            index=index,
            req_id=req_id,
            client_id=client_id,
            operation_id=operation_id,
            context=context,
            source=source,
            ws_id=ws_id,
            plugin_params=plugin_params,
            flt=flt,
            **kwargs,
        )
 
        parser = None
 
        fs = TmpIngestStats(source)
 
        # initialize mapping
        try:
            index_type_mapping, custom_mapping = await self.ingest_plugin_initialize(
                index,
                source=source,
                pipeline=ecs_windows(),
                mapping_file="windows.json",
                plugin_params=plugin_params,
            )
            # Plugin.logger().debug("win_mappings: %s" % (win_mapping))
        except Exception as ex:
            fs=self._parser_failed(fs, source, ex)
            return await self._finish_ingestion(index, source, req_id, client_id, ws_id, fs=fs, flt=flt)
 
        # init parser
        try:
            parser = PyEvtxParser(source)
        except Exception as ex:
            # cannot parse this file at all
            fs=self._parser_failed(fs, source, ex)
            return await self._finish_ingestion(index, source, req_id, client_id, ws_id, fs=fs, flt=flt)
 
        ev_idx = 0
 
        try:
            for rr in parser.records():
                # process (ingest + update stats)
                try:
                    fs, must_break = await self._process_record(
                        index,
                        rr,
                        ev_idx,
                        self.record_to_gulp_document,
                        ws_id,
                        req_id,
                        operation_id,
                        client_id,
                        context,
                        source,
                        fs,
                        custom_mapping=custom_mapping,
                        index_type_mapping=index_type_mapping,
                        plugin_params=plugin_params,
                        flt=flt,
                        **kwargs,
                    )
 
                    ev_idx += 1
                    if must_break:
                        break
 
                except Exception as ex:
                    fs=self._record_failed(fs, rr, source, ex)
 
        except Exception as ex:
            fs=self._parser_failed(fs, source, ex)
 
        # done
        return await self._finish_ingestion(index, source, req_id, client_id, ws_id, fs=fs, flt=flt)

Here is the call graph for this function:

Here is the caller graph for this function:

name()

str name

(

self

)

Returns the name of the plugin.

Reimplemented from PluginBase.

Definition at line 58 of file win_evtx.py.

    def name(self) -> str:
        return "win_evtx"
 

Here is the caller graph for this function:

record_to_gulp_document()

list[GulpDocument] record_to_gulp_document	(		self,
		int	operation_id,
		int	client_id,
		str	context,
		str	source,
		TmpIngestStats	fs,
		any	record,
		int	record_idx,
		GulpMapping	custom_mapping = None,
		dict	index_type_mapping = None,
		str	plugin = None,
		GulpPluginParams	plugin_params = None,
		**	kwargs )

Converts a record to one or more GulpDocument objects based on the provided index mappings.

Args:
    operation_id (int): The operation ID associated with the record.
    client_id (int): The client ID associated with the record.
    context (str): The context associated with the record.
    source (str): The source of the record (source file name or path, usually).
    fs (TmpIngestStats): The temporary ingestion statistics (may be updated on return).
    record (any): record to convert, plugin dependent format: note that here stacked plugins receives a list of GulpDocument objects instead (since the original record may generate one or more documents).
    record_idx (int): The index of the record in source.
    custom_mapping (GulpMapping, optional): The custom mapping to use for the conversion. Defaults to None.
    index_type_mapping (dict, optional): elastic search index type mappings { "ecs_field": "type", ... }. Defaults to None.
    plugin (str, optional): "agent.type" to be set in the GulpDocument. Defaults to None.
    plugin_params (GulpPluginParams, optional): The plugin parameters to use, if any. Defaults to None.
    extra (dict, optional): Additional fields to add to the GulpDocument (after applying mapping). Defaults to None.
    **kwargs: Additional keyword arguments:

Returns:
    list[GulDocument]: The converted GulpDocument objects or None if an exception occurred (fs is updated then).

Raises:
    NotImplementedError: This method is not implemented yet.

Reimplemented from PluginBase.

Definition at line 116 of file win_evtx.py.

    ) -> list[GulpDocument]:
        # process record
        # self.logger().debug(record)
        evt_str: str = record["data"].encode()
 
        data_elem = None
        data_elem = etree.fromstring(evt_str)
 
        # check if we should ignore mapping
        ignore_mapping = False
        if plugin_params is not None and plugin_params.ignore_mapping_ingest:
            ignore_mapping = True
 
        cat_tree = data_elem[0]
 
        # extra mapping from event code
        fme: list[FieldMappingEntry] = []
        entry = FieldMappingEntry(result={})
 
        evt_code = str(muty.xml.child_node_text(cat_tree, "EventID"))
        try:
            evt_channel = str(muty.xml.child_node_text(cat_tree, "Channel"))
        except:
            evt_channel = str(muty.xml.strip_namespace(cat_tree.tag))
 
        entry.result["winlog.channel"] = evt_channel
        try:
            evt_provider = str(muty.xml.child_attrib(cat_tree, "Provider", "Name"))
            entry.result["event.provider"] = evt_provider
        except:
            evt_provider = None
 
        d = self._map_evt_code(evt_code)
        if d is not None:
            entry.result.update(d)
 
        evt_id = muty.xml.child_node_text(cat_tree, "EventRecordID")
        original_log_level: str = muty.xml.child_node_text(cat_tree, "Level")
        gulp_log_level = self._normalize_loglevel(
            muty.xml.child_node_text(cat_tree, "Level")
        )
 
        # evtx timestamp is always UTC string, i.e.: '2021-11-19 16:53:03.836551 UTC'
        time_str = record["timestamp"]
        time_nanosec = muty.time.string_to_epoch_nsec(time_str)
        time_msec = muty.time.nanos_to_millis(time_nanosec)
        # Plugin.logger().debug('%s -  %d' % (time_str, time_msec))
 
        #  raw event as text
        raw_text = str(record["data"])
 
        if ignore_mapping:
            # persist these into the event
            entry.result["EventID"] = evt_code
            entry.result["Level"] = original_log_level
            entry.result["Provider"] = evt_provider
            entry.result["EventRecordID"] = evt_id
            entry.result["Channel"] = evt_channel
 
        # append this entry
        fme.append(entry)
 
        e_tree: etree.ElementTree = etree.ElementTree(data_elem)
 
        # self.logger().debug("e_tree: %s" % (e_tree))
 
        for e in e_tree.iter():
            e.tag = muty.xml.strip_namespace(e.tag)
 
            # these are already processed
            if e.tag in ["EventID", "EventRecordID", "Level", "Provider"]:
                continue
 
            # self.logger().debug(
            #     "found e_tag=%s, value=%s" % (e.tag, e.text)
            # )
 
            # Check XML tag name and set extra mapping
            entries = self._map_source_key(
                plugin_params,
                custom_mapping,
                e.tag,
                e.text,
                index_type_mapping=index_type_mapping,
                **kwargs,
            )
            for entry in entries:
                fme.append(entry)
 
                # add attributes as more extra data
                for attr_k, attr_v in e.attrib.items():
                    kk = "gulp.winevtx.xml.attr.%s" % (attr_k)
                    inner_entries = self._map_source_key(
                        plugin_params,
                        custom_mapping,
                        kk,
                        attr_v,
                        index_type_mapping=index_type_mapping,
                        **kwargs,
                    )
                    for entry in inner_entries:
                        fme.append(entry)
 
            # Check attributes and set extra mapping
            for attr_k, attr_v in e.attrib.items():
                # Plugin.logger().info("name: %s value: %s" % (attr_k, attr_v))
                # self.logger().debug("processing attr_v=%s, value=%s" % (attr_v, e.text))
                entries = self._map_source_key(
                    plugin_params,
                    custom_mapping,
                    attr_v,
                    e.text,
                    index_type_mapping=index_type_mapping,
                    **kwargs,
                )
                for entry in entries:
                    fme.append(entry)
 
                if e.text is None:
                    # no more processing
                    continue
 
                value = e.text.strip()
                if attr_v is not None:
                    # prefer attr_v
                    value = attr_v.strip()
 
                entries = self._map_source_key(
                    plugin_params,
                    custom_mapping,
                    attr_k,
                    value,
                    index_type_mapping=index_type_mapping,
                    **kwargs,
                )
                for entry in entries:
                    fme.append(entry)
 
 
        # finally create documents
        docs = self._build_gulpdocuments(
            fme,
            idx=record_idx,
            operation_id=operation_id,
            context=context,
            plugin=self.name(),
            client_id=client_id,
            raw_event=raw_text,
            original_id=evt_id,
            src_file=os.path.basename(source),
            timestamp=time_msec,
            timestamp_nsec=time_nanosec,
            event_code=evt_code,
            gulp_log_level=gulp_log_level,
            original_log_level=original_log_level
        )
        return docs
 

Here is the call graph for this function:

Here is the caller graph for this function:

type()

GulpPluginType type

(

self

)

Returns the plugin type.

Reimplemented from PluginBase.

Definition at line 52 of file win_evtx.py.

    def type(self) -> GulpPluginType:
        return GulpPluginType.INGESTION
 

version()

str version

(

self

)

Returns plugin version.

Reimplemented from PluginBase.

Definition at line 61 of file win_evtx.py.

    def version(self) -> str:
        return "1.0"
 

---

The documentation for this class was generated from the following file:

• src/gulp/plugins/win_evtx.py