mirror of
https://github.com/crypt0p3g/dpapi-toolkit
synced 2026-09-24 08:14:49 +00:00
Initial release
This commit is contained in:
ruslan
@@ -0,0 +1,183 @@
|
||||
"""Small, local-only plugin loader for application-specific DPAPI layers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from fnmatch import fnmatch
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
from types import ModuleType
|
||||
|
||||
|
||||
PLUGIN_ROOT = Path(__file__).with_name("plugins")
|
||||
PLUGIN_ID = re.compile(r"^[a-z][a-z0-9_-]{0,31}$")
|
||||
MAX_MANIFEST_BYTES = 64 * 1024
|
||||
WEB_FIELD = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
|
||||
WEB_FIELD_KINDS = frozenset(("text", "password", "file", "checkbox"))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PluginWebField:
|
||||
name: str
|
||||
kind: str
|
||||
label: str
|
||||
placeholder: str
|
||||
help: str
|
||||
optional: bool = False
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PluginManifest:
|
||||
plugin_id: str
|
||||
name: str
|
||||
description: str
|
||||
directory: Path
|
||||
file_patterns: tuple[str, ...]
|
||||
web_help: str
|
||||
web_input_label: str
|
||||
web_workflow_help: str
|
||||
web_fields: tuple[PluginWebField, ...]
|
||||
|
||||
|
||||
def discover_plugins() -> dict[str, PluginManifest]:
|
||||
root = PLUGIN_ROOT.resolve()
|
||||
manifests: dict[str, PluginManifest] = {}
|
||||
if not root.is_dir():
|
||||
return manifests
|
||||
for directory in sorted(path for path in root.iterdir() if path.is_dir()):
|
||||
manifest_path = directory / "plugin.json"
|
||||
module_path = directory / "plugin.py"
|
||||
if not manifest_path.is_file() or not module_path.is_file():
|
||||
continue
|
||||
if manifest_path.stat().st_size > MAX_MANIFEST_BYTES:
|
||||
raise ValueError(f"plugin manifest is too large: {manifest_path}")
|
||||
try:
|
||||
raw = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise ValueError(f"invalid plugin manifest {manifest_path}: {exc}") from None
|
||||
if not isinstance(raw, dict):
|
||||
raise ValueError(f"plugin manifest must be an object: {manifest_path}")
|
||||
plugin_id = raw.get("id")
|
||||
name = raw.get("name")
|
||||
description = raw.get("description", "")
|
||||
patterns = raw.get("file_patterns", [])
|
||||
web = raw.get("web", {})
|
||||
if (
|
||||
not isinstance(plugin_id, str)
|
||||
or not PLUGIN_ID.fullmatch(plugin_id)
|
||||
or plugin_id != directory.name
|
||||
):
|
||||
raise ValueError(f"plugin id must match its safe directory name: {directory}")
|
||||
if not isinstance(name, str) or not name.strip():
|
||||
raise ValueError(f"plugin {plugin_id} has no display name")
|
||||
if not isinstance(description, str):
|
||||
raise ValueError(f"plugin {plugin_id} description must be text")
|
||||
if not isinstance(patterns, list) or not all(
|
||||
isinstance(pattern, str) and pattern and "/" not in pattern and "\\" not in pattern
|
||||
for pattern in patterns
|
||||
):
|
||||
raise ValueError(f"plugin {plugin_id} has invalid file patterns")
|
||||
if not isinstance(web, dict):
|
||||
raise ValueError(f"plugin {plugin_id} web configuration must be an object")
|
||||
web_help = web.get("help", description)
|
||||
web_input_label = web.get("input_label", "plugin artifact")
|
||||
web_workflow_help = web.get(
|
||||
"workflow_help",
|
||||
"Tab 1 is not used for plugins. Add the main input here.",
|
||||
)
|
||||
raw_web_fields = web.get("fields", [])
|
||||
if not isinstance(web_help, str):
|
||||
raise ValueError(f"plugin {plugin_id} web help must be text")
|
||||
if not isinstance(web_input_label, str) or not web_input_label.strip():
|
||||
raise ValueError(f"plugin {plugin_id} web input label must be text")
|
||||
if not isinstance(web_workflow_help, str) or not web_workflow_help.strip():
|
||||
raise ValueError(f"plugin {plugin_id} web workflow help must be text")
|
||||
if not isinstance(raw_web_fields, list) or len(raw_web_fields) > 16:
|
||||
raise ValueError(f"plugin {plugin_id} has invalid web fields")
|
||||
web_fields = []
|
||||
seen_web_fields = set()
|
||||
for field in raw_web_fields:
|
||||
if not isinstance(field, dict):
|
||||
raise ValueError(f"plugin {plugin_id} web field must be an object")
|
||||
field_name = field.get("name")
|
||||
kind = field.get("kind")
|
||||
label = field.get("label")
|
||||
placeholder = field.get("placeholder", "")
|
||||
field_help = field.get("help", "")
|
||||
optional = field.get("optional", False)
|
||||
if (
|
||||
not isinstance(field_name, str)
|
||||
or not WEB_FIELD.fullmatch(field_name)
|
||||
or field_name in seen_web_fields
|
||||
or kind not in WEB_FIELD_KINDS
|
||||
or not isinstance(label, str)
|
||||
or not label.strip()
|
||||
or not isinstance(placeholder, str)
|
||||
or not isinstance(field_help, str)
|
||||
or not isinstance(optional, bool)
|
||||
):
|
||||
raise ValueError(f"plugin {plugin_id} has an invalid web field")
|
||||
seen_web_fields.add(field_name)
|
||||
web_fields.append(PluginWebField(
|
||||
field_name,
|
||||
kind,
|
||||
label.strip(),
|
||||
placeholder,
|
||||
field_help,
|
||||
optional,
|
||||
))
|
||||
if plugin_id in manifests:
|
||||
raise ValueError(f"duplicate plugin id: {plugin_id}")
|
||||
manifests[plugin_id] = PluginManifest(
|
||||
plugin_id,
|
||||
name.strip(),
|
||||
description.strip(),
|
||||
directory.resolve(),
|
||||
tuple(patterns),
|
||||
web_help.strip(),
|
||||
web_input_label.strip(),
|
||||
web_workflow_help.strip(),
|
||||
tuple(web_fields),
|
||||
)
|
||||
return manifests
|
||||
|
||||
|
||||
def detect_plugin(filename: str) -> str | None:
|
||||
matches = [
|
||||
manifest.plugin_id
|
||||
for manifest in discover_plugins().values()
|
||||
if any(fnmatch(filename.casefold(), pattern.casefold()) for pattern in manifest.file_patterns)
|
||||
]
|
||||
return matches[0] if len(matches) == 1 else None
|
||||
|
||||
|
||||
def load_plugin(plugin_id: str) -> tuple[PluginManifest, ModuleType]:
|
||||
manifest = discover_plugins().get(plugin_id)
|
||||
if manifest is None:
|
||||
raise ValueError(f"unknown or unavailable plugin {plugin_id!r}")
|
||||
root = PLUGIN_ROOT.resolve()
|
||||
module_path = (manifest.directory / "plugin.py").resolve()
|
||||
try:
|
||||
module_path.relative_to(root)
|
||||
except ValueError:
|
||||
raise ValueError(f"plugin {plugin_id} escapes the plugin directory") from None
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
f"dpapi_toolkit_plugin_{plugin_id}", module_path
|
||||
)
|
||||
if spec is None or spec.loader is None:
|
||||
raise ValueError(f"cannot load plugin {plugin_id}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
if getattr(module, "PLUGIN_ID", None) != plugin_id or not callable(
|
||||
getattr(module, "run", None)
|
||||
):
|
||||
raise ValueError(f"plugin {plugin_id} does not implement the required contract")
|
||||
return manifest, module
|
||||
|
||||
|
||||
def run_plugin(plugin_id: str, data, source_path, was_hex, args, core, emit):
|
||||
manifest, module = load_plugin(plugin_id)
|
||||
emit(f"[+] plugin: {manifest.name} ({manifest.plugin_id})")
|
||||
return module.run(data, source_path, was_hex, args, core, emit)
|
||||
Reference in New Issue
Block a user