Skip to content

loader

File-based plugin loading with security checks.

This module handles loading plugins from Python files in user directories. It includes security checks for symlinks and world-writable directories.

Classes

Functions

clear_plugin_cache 

clear_plugin_cache() -> None

Clear the cached user plugins, forcing a reload on next access.

Source code in snakesee/plugins/loader.py 
244
245
246
247
def clear_plugin_cache() -> None:
    """Clear the cached user plugins, forcing a reload on next access."""
    global _user_plugins
    _user_plugins = None

load_user_plugins 

load_user_plugins(plugin_dirs: list[Path] | None = None, force_reload: bool = False) -> list[ToolProgressPlugin]

Load custom user plugins from plugin directories.

User plugins are Python files in ~/.snakesee/plugins/ or ~/.config/snakesee/plugins/ that define classes inheriting from ToolProgressPlugin.

Parameters:

Name Type Description Default
plugin_dirs list[Path] | None

List of directories to search. Defaults to USER_PLUGIN_DIRS.

 None
force_reload bool

If True, reload plugins even if already cached.

 False

Returns:

Type Description
list[ToolProgressPlugin]

List of loaded user plugin instances.

Example plugin file (~/.snakesee/plugins/my_tool.py)::

from snakesee.plugins.base import ToolProgress, ToolProgressPlugin
import re

class MyToolPlugin(ToolProgressPlugin):
    @property
    def tool_name(self) -> str:
        return "mytool"

    def can_parse(self, rule_name: str, log_content: str) -> bool:
        return "mytool" in rule_name.lower()

    def parse_progress(self, log_content: str) -> ToolProgress | None:
        match = re.search(r"Processed (\d+) items", log_content)
        if match:
            return ToolProgress(items_processed=int(match.group(1)), unit="items")
        return None
Source code in snakesee/plugins/loader.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def load_user_plugins(
    plugin_dirs: list[Path] | None = None,
    force_reload: bool = False,
) -> list[ToolProgressPlugin]:
    """
    Load custom user plugins from plugin directories.

    User plugins are Python files in ~/.snakesee/plugins/ or ~/.config/snakesee/plugins/
    that define classes inheriting from ToolProgressPlugin.

    Args:
        plugin_dirs: List of directories to search. Defaults to USER_PLUGIN_DIRS.
        force_reload: If True, reload plugins even if already cached.

    Returns:
        List of loaded user plugin instances.

    Example plugin file (~/.snakesee/plugins/my_tool.py)::

        from snakesee.plugins.base import ToolProgress, ToolProgressPlugin
        import re

        class MyToolPlugin(ToolProgressPlugin):
            @property
            def tool_name(self) -> str:
                return "mytool"

            def can_parse(self, rule_name: str, log_content: str) -> bool:
                return "mytool" in rule_name.lower()

            def parse_progress(self, log_content: str) -> ToolProgress | None:
                match = re.search(r"Processed (\\d+) items", log_content)
                if match:
                    return ToolProgress(items_processed=int(match.group(1)), unit="items")
                return None
    """
    global _user_plugins

    if _user_plugins is not None and not force_reload:
        return _user_plugins

    if plugin_dirs is None:
        plugin_dirs = USER_PLUGIN_DIRS

    loaded_plugins: list[ToolProgressPlugin] = []

    for plugin_dir in plugin_dirs:
        if not plugin_dir.exists() or not plugin_dir.is_dir():
            continue

        # Security checks
        _check_plugin_dir_security(plugin_dir)

        # Find all Python files in the plugin directory
        for plugin_file in plugin_dir.glob("*.py"):
            if plugin_file.name.startswith("_"):
                continue  # Skip private modules

            try:
                plugins = _load_plugins_from_file(plugin_file)
                loaded_plugins.extend(plugins)
            except (ImportError, SyntaxError, OSError) as e:
                logger.debug("Failed to load plugin from %s: %s", plugin_file, e)
                continue

    _user_plugins = loaded_plugins
    return loaded_plugins

validate_plugin 

validate_plugin(plugin: ToolProgressPlugin, source: str = 'unknown') -> PluginMetadata | None

Validate that a plugin instance is compatible and properly implemented.

Uses PluginMetadata for structured validation of plugin attributes.

Parameters:

Name Type Description Default
plugin ToolProgressPlugin

The plugin instance to validate.

 required
source str

Description of where the plugin came from (for logging).

 'unknown'

Returns:

Type Description
PluginMetadata | None

PluginMetadata if the plugin is valid and compatible, None otherwise.
Source code in snakesee/plugins/loader.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def validate_plugin(plugin: ToolProgressPlugin, source: str = "unknown") -> PluginMetadata | None:
    """
    Validate that a plugin instance is compatible and properly implemented.

    Uses PluginMetadata for structured validation of plugin attributes.

    Args:
        plugin: The plugin instance to validate.
        source: Description of where the plugin came from (for logging).

    Returns:
        PluginMetadata if the plugin is valid and compatible, None otherwise.
    """
    # Validate required interface methods exist and are callable
    required_methods = ["can_parse", "parse_progress"]

    for method_name in required_methods:
        method = getattr(plugin, method_name, None)
        if method is None or not callable(method):
            logger.warning(
                "Plugin from %s is missing required method '%s'. Skipping.",
                source,
                method_name,
            )
            return None

    # Use PluginMetadata for structured validation
    try:
        metadata = PluginMetadata.from_plugin(plugin)
    except (ValueError, AttributeError) as e:
        logger.warning(
            "Plugin from %s failed metadata validation: %s. Skipping.",
            source,
            e,
        )
        return None

    # Check API version compatibility
    if not metadata.is_compatible(PLUGIN_API_VERSION):
        logger.warning(
            "Plugin %s from %s requires API version %d, but current version is %d. Skipping.",
            metadata.name,
            source,
            metadata.api_version,
            PLUGIN_API_VERSION,
        )
        return None

    return metadata