Writing a plugin

Deployment and alerting are plugin seams. Adding a plugin requires no registration code and no import changes: the loader discovers the directory on startup, and the UI renders a config form from the manifest.

Plugin layout

A plugin is a directory under plugins/{deploy,alert}/<name>/ with two files:

plugins/alert/slack/
  manifest.yaml   # metadata + config_schema (the UI renders a form from this)
  plugin.py       # a class named `Plugin`

manifest.yaml

Metadata plus a config_schema. Each field becomes a form input in the UI:

name: slack                 # unique id (must match the directory)
kind: alert                 # deploy | alert
display_name: "Slack"
version: "0.1.0"
author: your-handle
description: "Posts a message to a Slack channel when a tripwire fires."
config_schema:              # each field becomes a form input in the UI
  - key: webhook_url
    label: "Incoming webhook URL"
    type: secret            # string | secret | boolean
    required: true
    placeholder: "https://hooks.slack.com/services/…"
    help: "Optional helper text shown under the field."

secret fields are stored as given but masked (••••••••) when read back.

plugin.py

The class must be named Plugin and is constructed with the saved config dict (available as self.config). Subclass AlertPlugin or DeployPlugin from thumper.plugins.base.

Alert plugin
Deploy plugin

Implement alert(event) - deliver the event, raise on failure. The router logs a failure and keeps going, so one bad plugin never drops the alert.

import httpx
from thumper.plugins.base import AlertPlugin, PluginError

class Plugin(AlertPlugin):
    def alert(self, event: dict) -> None:
        url = self.config.get("webhook_url")
        if not url:
            raise PluginError("slack: webhook_url is required")
        text = (f":rotating_light: Honeytoken read on *{event['endpoint_hostname']}* "
                f"by `{event.get('process')}` (user {event.get('os_user')}) "
                f"- {event['tripwire_name']} at {event.get('accessed_path')}")
        httpx.post(url, json={"text": text}, timeout=10).raise_for_status()

See Alert integrations for the full event dict. The base class also provides a test() that sends a clearly-labeled synthetic event through your real send path.

Implement deploy(install, targets) - run install.command on the machines you target. The plugin never handles token content or secrets; running the command makes each machine self-enroll and pull its own unique instance.

from thumper.plugins.base import AgentInstall, DeployPlugin, DeployResult, PluginError

class Plugin(DeployPlugin):
    def deploy(self, install: AgentInstall, targets: list[str]) -> DeployResult:
        # Run install.command on machines from self.config or targets.
        ...
        return DeployResult(state="deployed", deployed_count=N, message="…")

AgentInstall carries tripwire_id, server_url, enroll_token, and the ready-to-run command.

Raise PluginError for expected failures (missing config, unreachable target); the API surfaces the message.

Test it

Drop the directory in, restart the server, and the plugin appears under Integrations with a generated config form. Verify it loads:

python -c "from thumper.plugins.registry import load_plugin; load_plugin('slack', {})"

Get started

Concepts

Integrations

Writing a plugin

Plugin layout

manifest.yaml

plugin.py

Test it

​Plugin layout

​manifest.yaml

​plugin.py

​Test it

Plugin layout

manifest.yaml

plugin.py

Test it