Classes
class	PluginBase

Functions
str	_get_plugin_path (str plugin, **kwargs)

list[str]	get_plugin_tags (str plugin)

	init (logging.Logger logger)

list[dict]	list_plugins ()

PluginBase	load_plugin (str plugin, AsyncEngine collab=None, AsyncElasticsearch elastic=None, **kwargs)

None	plugin_cache_add (ModuleType m, str name)

None	plugin_cache_clear ()

ModuleType	plugin_cache_get (str plugin)

None	plugin_cache_remove (str plugin)

None	unload_plugin (PluginBase mod)

Variables
dict	_cache = {}

logging	_logger = None

Detailed Description

Gulp plugin base class and plugin utilities.

Function Documentation

◆ _get_plugin_path()

str _get_plugin_path	(	str	plugin,
		**	kwargs )

protected

Definition at line 1010 of file plugin.py.

def _get_plugin_path(plugin: str, **kwargs) -> str:
 
    # try plain .py first
    # TODO: on license manager, disable plain .py load (only encrypted pyc)
    # get path according to plugin type
    plugin_type = kwargs.get("plugin_type", GulpPluginType.INGESTION)
    path_plugins = config.path_plugins(plugin_type)
 
    ppy = muty.file.safe_path_join(path_plugins, plugin + ".py")
    if not muty.file.exists(ppy):
        # try pyc
        ppyc = muty.file.safe_path_join(path_plugins, plugin + ".pyc")
        if not muty.file.exists(ppyc):
            raise ObjectNotFound(
                "plugin %s not found (tried %s, %s)" % (plugin, ppy, ppyc)
            )
        return ppyc
    return ppy
 
 

Here is the caller graph for this function:

◆ get_plugin_tags()

list[str] get_plugin_tags ( str plugin )

Get the tags for a given (ingestion) plugin.

Args:
    plugin (str): The name of the plugin to get the tags for.
Returns:
    list[str]: The tags for the given plugin.

Definition at line 1118 of file plugin.py.

async def get_plugin_tags(plugin: str) -> list[str]:
    """
    Get the tags for a given (ingestion) plugin.
 
    Args:
        plugin (str): The name of the plugin to get the tags for.
    Returns:
        list[str]: The tags for the given plugin.
    """
    p = load_plugin(plugin)
    tags = p.tags()
    return tags
 
 

Here is the call graph for this function:

◆ init()

init ( logging.Logger logger )

Initializes the plugin base module.

Definition at line 47 of file plugin.py.

def init(logger: logging.Logger):
    """
    Initializes the plugin base module.
    """
    global _logger
    _logger = logger
 
 

◆ list_plugins()

list[dict] list_plugins ( )

List all available plugins.

Returns:
    list[dict]: The list of available plugins.

Definition at line 1082 of file plugin.py.

async def list_plugins() -> list[dict]:
    """
    List all available plugins.
 
    Returns:
        list[dict]: The list of available plugins.
    """
    path_plugins = config.path_plugins()
    files = await muty.file.list_directory_async(path_plugins, "*.py*", recursive=True)
    l = []
    for f in files:
        if "__init__" not in f and "__pycache__" not in f:
            try:
                p = load_plugin(f)
                n = {
                    "name": p.name(),
                    "type": str(p.type()),
                    "desc": p.desc(),
                    "filename": os.path.basename(p.path),
                    "internal": p.internal(),
                    "options": [o.to_dict() for o in p.options()],
                    "depends_on": p.depends_on(),
                    "tags": p.tags(),
                    "event_type_field": p.event_type_field(),
                    "version": p.version(),
                }
                l.append(n)
                unload_plugin(p)
            except Exception as ex:
                _logger.exception(ex)
                _logger.error("could not load plugin %s" % (f))
                continue
 
    return l
 
 

Here is the call graph for this function:

◆ load_plugin()

PluginBase load_plugin	(	str	plugin,
		AsyncEngine	collab = None,
		AsyncElasticsearch	elastic = None,
		**	kwargs )

Load a plugin from a given path or from the default plugin path.

Args:
    plugin (str): The name or path of the plugin to load.
    req_id (str, optional): The request ID to associate with the plugin loading. Defaults to None.
    collab (AsyncEngine, optional): The SQLAlchemy engine instance. Defaults to None (not needed for pysigma plugins).
    elastic (AsyncElasticsearch, optional): The Elasticsearch client instance. Defaults to None (not needed for pysigma plugins).
    kwargs: additional arguments:
        "plugin_type" (GulpPluginType, default=GulpPluginType.INGESTION)
Returns:
    PluginBase: The loaded plugin.

Raises:
    Exception: If the plugin could not be loaded.

Definition at line 1030 of file plugin.py.

) -> PluginBase:
    """
    Load a plugin from a given path or from the default plugin path.
 
    Args:
        plugin (str): The name or path of the plugin to load.
        req_id (str, optional): The request ID to associate with the plugin loading. Defaults to None.
        collab (AsyncEngine, optional): The SQLAlchemy engine instance. Defaults to None (not needed for pysigma plugins).
        elastic (AsyncElasticsearch, optional): The Elasticsearch client instance. Defaults to None (not needed for pysigma plugins).
        kwargs: additional arguments:
            "plugin_type" (GulpPluginType, default=GulpPluginType.INGESTION)
    Returns:
        PluginBase: The loaded plugin.
 
    Raises:
        Exception: If the plugin could not be loaded.
    """
    _logger.debug(
        "load_plugin %s, collab=%s, elastic=%s ..." % (plugin, collab, elastic)
    )
 
    m = plugin_cache_get(plugin)
    if m is not None:
        return m.Plugin(plugin, collab, elastic, **kwargs)
 
    if "/" in plugin:
        # plugins is an absolute path
        path = muty.file.abspath(plugin)
    else:
        # uses "plugin_type" to load from the correct subfolder
        path = _get_plugin_path(plugin, **kwargs)
 
    try:
        m: PluginBase = muty.dynload.load_dynamic_module_from_file(path)
    except Exception as ex:
        raise Exception(
            "load_dynamic_module_from_file() failed, could not load plugin %s !"
            % (path)
        ) from ex
 
    mod: PluginBase = m.Plugin(path, collab, elastic, **kwargs)
    _logger.debug("loaded plugin: %s" % (mod.name()))
    plugin_cache_add(m, plugin)
 
    return mod
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ plugin_cache_add()

None plugin_cache_add	(	ModuleType	m,
		str	name )

Add a plugin to the process's own plugin cache.

Args:
    m (ModuleType): The plugin module to add to the cache.
    name (str): The name/path of the plugin.

Returns:
    None

Definition at line 1193 of file plugin.py.

def plugin_cache_add(m: ModuleType, name: str) -> None:
    """
    Add a plugin to the process's own plugin cache.
 
    Args:
        m (ModuleType): The plugin module to add to the cache.
        name (str): The name/path of the plugin.
 
    Returns:
        None
    """
    global _cache
    if not config.plugin_cache_enabled():
        return
 
    mm = _cache.get(name, None)
    if mm is None:
        _logger.debug("adding plugin %s (%s) to cache" % (name, m))
        _cache[name] = m
 
 

Here is the caller graph for this function:

◆ plugin_cache_clear()

None plugin_cache_clear ( )

Clear the process's own plugin cache.

Returns:
    None

Definition at line 1157 of file plugin.py.

def plugin_cache_clear() -> None:
    """
    Clear the process's own plugin cache.
 
    Returns:
        None
    """
    global _cache
    if not config.plugin_cache_enabled():
        return
 
    _cache = {}
 
 

◆ plugin_cache_get()

ModuleType plugin_cache_get ( str plugin )

Retrieve a plugin from the process's own plugin cache.

Args:
    plugin (str): The name/path of the plugin to retrieve.

Returns:
    ModuleType: The plugin module if found in the cache, otherwise None.

Definition at line 1214 of file plugin.py.

def plugin_cache_get(plugin: str) -> ModuleType:
    """
    Retrieve a plugin from the process's own plugin cache.
 
    Args:
        plugin (str): The name/path of the plugin to retrieve.
 
    Returns:
        ModuleType: The plugin module if found in the cache, otherwise None.
    """
    global _cache
    if not config.plugin_cache_enabled():
        return None
 
    p = _cache.get(plugin, None)
    if p is not None:
        _logger.debug("found plugin %s in cache" % (plugin))
    return p

Here is the caller graph for this function:

◆ plugin_cache_remove()

None plugin_cache_remove ( str plugin )

Remove a plugin from the process's own plugin cache.

Args:
    plugin (str): The name/path of the plugin to remove from the cache.

Returns:
    None

Definition at line 1171 of file plugin.py.

def plugin_cache_remove(plugin: str) -> None:
    """
    Remove a plugin from the process's own plugin cache.
 
    Args:
        plugin (str): The name/path of the plugin to remove from the cache.
 
    Returns:
        None
    """
    global _cache
    if not config.plugin_cache_enabled():
        return
 
    if plugin in _cache:
        _logger.debug("removing plugin %s from cache" % (plugin))
 
        # cleanup module and delete
        m = _cache[plugin]
        del _cache[plugin]
 
 

◆ unload_plugin()

None unload_plugin ( PluginBase mod )

Unloads a plugin module by calling its `unload` method and deletes the module object

NOTE: mod is **no more valid** after this function returns.

Args:
    mod (PluginBase): The plugin module to unload.
    run_gc (bool): if set, garbage collector is called after unloading the module. Defaults to True.

Returns:
    None

Definition at line 1132 of file plugin.py.

def unload_plugin(mod: PluginBase) -> None:
    """
    Unloads a plugin module by calling its `unload` method and deletes the module object
 
    NOTE: mod is **no more valid** after this function returns.
 
    Args:
        mod (PluginBase): The plugin module to unload.
        run_gc (bool): if set, garbage collector is called after unloading the module. Defaults to True.
 
    Returns:
        None
    """
    if config.plugin_cache_enabled():
        return
 
    if mod is not None:
        # delete from cache if any
        # plugin_cache_delete(mod)
 
        _logger.debug("unloading plugin: %s" % (mod.name()))
        mod.cleanup()
        del mod
 
 

Here is the caller graph for this function:

Variable Documentation

◆ _cache

dict _cache = {}

protected

Definition at line 44 of file plugin.py.

◆ _logger

logging _logger = None

protected

Definition at line 41 of file plugin.py.

Classes

Functions

Variables

Detailed Description

Function Documentation

◆ _get_plugin_path()

◆ get_plugin_tags()

◆ init()

◆ list_plugins()

◆ load_plugin()

◆ plugin_cache_add()

◆ plugin_cache_clear()

◆ plugin_cache_get()

◆ plugin_cache_remove()

◆ unload_plugin()

Variable Documentation

◆ _cache

◆ _logger