Outpost Plugin System

The Arbitex Hybrid Outpost includes an extensible plugin architecture that allows you to attach custom processing hooks at key lifecycle points in the request pipeline. Plugins run in-process alongside the outpost, have direct access to DLP scan results, and can extend functionality — logging, alerting, annotation, or side-effect triggering — without requiring modifications to core outpost code.

This guide covers the plugin interface protocol, how to write and load custom plugins, the built-in webhook emitter plugin, and custom entity redaction patterns for organization-specific DLP detection.

---

Plugin System Overview

The plugin system is disabled by default. Enable it and point the outpost at a plugin directory using the following environment variables.

Variable	Default	Type	Description
PLUGINS_ENABLED	false	true/false	Enable the plugin loading subsystem
PLUGIN_DIR	/etc/arbitex-outpost/plugins	Path	Directory scanned for plugin .py files at startup

When PLUGINS_ENABLED=true, the outpost scans PLUGIN_DIR at startup, imports each .py file it finds, and calls register_plugin() to register the plugin instance with the plugin manager. Registered plugins receive lifecycle hook calls as the outpost processes requests.

Plugins are isolated by responsibility, not by process. They share the outpost’s runtime and have access to the same scan result structures. This enables low-latency integrations but means a poorly written plugin can affect outpost stability. Review the cautions in the Loading Plugins section before deploying custom plugins to production.

---

Plugin Interface Protocol

Every plugin must implement the following interface. The outpost plugin manager validates the presence of all required properties and hooks at load time and will refuse to register plugins that do not conform.

from typing import Protocol, runtime_checkable

@runtime_checkable
class OutpostPlugin(Protocol):
    """
    Interface that every Arbitex Outpost plugin must implement.
    All hooks are called synchronously in the outpost request thread.
    Hooks must not block for extended periods.
    """

    name: str
    """Unique string identifier for this plugin. Must be unique across all loaded plugins."""

    version: str
    """Plugin version string in semver format (e.g. '1.0.0')."""

    def on_startup(self) -> None:
        """
        Called once when the plugin is loaded during outpost startup.
        Use this hook to initialize connections, load configuration,
        or validate required environment variables. Raising an exception
        here will prevent the plugin from being registered.
        """
        ...

    def on_shutdown(self) -> None:
        """
        Called during graceful outpost shutdown.
        Use this hook to flush buffers, close connections, or persist state.
        The outpost will wait up to 5 seconds for this hook to complete.
        """
        ...

    def on_scan_complete(self, scan_result: dict) -> None:
        """
        Called after the DLP scan pipeline finishes for a single request.
        Receives the full scan result object as a dictionary.

        The scan_result dict includes:
          scan_id        str   — unique identifier for this scan
          severity       str   — 'none', 'low', 'medium', 'high', 'critical'
          detections     list  — list of detection objects (entity, offset, score)
          policy_action  str   — 'allow', 'block', 'redact', 'warn'
          metadata       dict  — source_ip, user_id, model, request_id, timestamp
          content_hash   str   — SHA-256 of the original content (pre-redaction)

        Mutations to scan_result are permitted and will be visible to
        subsequently registered plugins. Core outpost behavior is not
        affected by mutations after this hook fires.
        """
        ...

Caution

All plugin hooks are called synchronously in the outpost request processing thread. Hooks that perform blocking I/O (network calls, file writes, database queries) will add latency to every proxied request. Use async queues or background threads for any operation that may take more than a few milliseconds.

---

Writing a Custom Plugin

The following is a complete, production-ready custom plugin that logs high-severity scan results to an external HTTP endpoint. It demonstrates proper protocol implementation, environment variable configuration, error handling, and the required register_plugin() call.

"""
high_severity_logger.py — Arbitex Outpost Plugin

Logs high and critical severity scan results to an external audit log
endpoint via HTTP POST. Runs I/O in a background thread to avoid
blocking the request pipeline.
"""

import json
import logging
import os
import queue
import threading
import urllib.request
import urllib.error
from typing import Optional

logger = logging.getLogger("arbitex.plugin.high_severity_logger")


class HighSeverityLoggerPlugin:
    """
    Plugin that forwards high-severity and critical DLP scan results
    to a configurable external audit log endpoint.
    """

    name: str = "high_severity_logger"
    version: str = "1.2.0"

    def __init__(self) -> None:
        self._endpoint: Optional[str] = None
        self._api_key: Optional[str] = None
        self._queue: queue.Queue = queue.Queue(maxsize=1000)
        self._worker: Optional[threading.Thread] = None
        self._running: bool = False

    def on_startup(self) -> None:
        self._endpoint = os.environ.get("AUDIT_LOG_ENDPOINT")
        self._api_key = os.environ.get("AUDIT_LOG_API_KEY")

        if not self._endpoint:
            raise RuntimeError(
                "HighSeverityLoggerPlugin requires AUDIT_LOG_ENDPOINT to be set"
            )

        self._running = True
        self._worker = threading.Thread(
            target=self._dispatch_loop,
            name="high-severity-logger-worker",
            daemon=True,
        )
        self._worker.start()
        logger.info(
            "HighSeverityLoggerPlugin started, endpoint=%s", self._endpoint
        )

    def on_shutdown(self) -> None:
        self._running = False
        # Sentinel to unblock the worker queue
        self._queue.put(None)
        if self._worker:
            self._worker.join(timeout=5.0)
        logger.info("HighSeverityLoggerPlugin shut down cleanly")

    def on_scan_complete(self, scan_result: dict) -> None:
        severity = scan_result.get("severity", "none")
        if severity not in ("high", "critical"):
            return

        payload = {
            "scan_id": scan_result.get("scan_id"),
            "severity": severity,
            "policy_action": scan_result.get("policy_action"),
            "detections_count": len(scan_result.get("detections", [])),
            "metadata": scan_result.get("metadata", {}),
        }

        try:
            self._queue.put_nowait(payload)
        except queue.Full:
            logger.warning(
                "HighSeverityLoggerPlugin: dispatch queue full, dropping event "
                "scan_id=%s", payload.get("scan_id")
            )

    def _dispatch_loop(self) -> None:
        while self._running:
            try:
                payload = self._queue.get(timeout=1.0)
            except queue.Empty:
                continue

            if payload is None:
                break

            self._send(payload)

    def _send(self, payload: dict) -> None:
        body = json.dumps(payload).encode("utf-8")
        headers = {"Content-Type": "application/json"}
        if self._api_key:
            headers["Authorization"] = f"Bearer {self._api_key}"

        req = urllib.request.Request(
            self._endpoint, data=body, headers=headers, method="POST"
        )
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                logger.debug(
                    "Audit log event sent scan_id=%s status=%d",
                    payload.get("scan_id"), resp.status
                )
        except urllib.error.HTTPError as exc:
            logger.error(
                "Audit log HTTP error scan_id=%s status=%d",
                payload.get("scan_id"), exc.code
            )
        except Exception as exc:  # noqa: BLE001
            logger.error(
                "Audit log send failed scan_id=%s error=%s",
                payload.get("scan_id"), exc
            )


def register_plugin() -> HighSeverityLoggerPlugin:
    """
    Required entry point. Called by the outpost plugin manager at startup.
    Must return an instance of a class that implements OutpostPlugin.
    """
    return HighSeverityLoggerPlugin()

Save this file to your PLUGIN_DIR (e.g. /etc/arbitex-outpost/plugins/high_severity_logger.py). Set the required environment variables and restart the outpost to activate the plugin.

---

Loading Plugins

Auto-Discovery

When the outpost starts with PLUGINS_ENABLED=true, it performs the following steps:

1. Scans PLUGIN_DIR for files matching *.py.
2. Imports each file as a module.
3. Calls register_plugin() from the module’s top-level namespace.
4. Validates that the returned object implements OutpostPlugin (checks for name, version, on_startup, on_scan_complete, on_shutdown).
5. Calls on_startup() on the plugin instance.
6. Registers the plugin in the plugin manager’s internal registry.

Load order is alphabetical by filename. If load order matters for your plugins (e.g. one plugin depends on state set by another), name your files accordingly — 00_base_plugin.py, 01_dependent_plugin.py, etc.

Caution

Plugin load failures are isolated. If a plugin’s register_plugin() raises an exception, or if on_startup() raises an exception, that plugin is skipped and an error is logged. Other plugins continue loading normally. Always check outpost startup logs to confirm all expected plugins loaded successfully.

Constraints

• Plugins are loaded once at startup. There is no hot-reload mechanism. To add, remove, or update a plugin, update the files in PLUGIN_DIR and restart the outpost.
• Plugin filenames must not start with _ (underscore-prefixed files are ignored by the auto-discovery scanner).
• Two plugins with the same name property cannot both be registered. The second plugin encountered during alphabetical scan will be rejected with a logged error.

---

Plugin Management API

The outpost admin API provides endpoints to inspect and toggle plugins at runtime without a restart. The admin API requires a valid Authorization: Bearer <ADMIN_API_TOKEN> header on all requests.

List Registered Plugins

GET /admin/api/plugins

Returns the list of all plugins currently registered with the plugin manager, including their enabled/disabled state.

Response fields:

Field	Type	Description
name	string	Unique plugin identifier
version	string	Plugin version (semver)
enabled	boolean	Whether the plugin is currently active and receiving hook calls
hooks_registered	array of string	List of hook names this plugin implements

Example response:

[
  {
    "name": "high_severity_logger",
    "version": "1.2.0",
    "enabled": true,
    "hooks_registered": ["on_startup", "on_shutdown", "on_scan_complete"]
  },
  {
    "name": "internal_audit_annotator",
    "version": "0.9.1",
    "enabled": false,
    "hooks_registered": ["on_startup", "on_scan_complete"]
  }
]

---

Enable a Plugin

POST /admin/api/plugins/{name}/enable

Enables a previously disabled plugin. The plugin will begin receiving hook calls immediately after this request completes.

• Returns 200 with the updated PluginInfo object on success.
• Returns 404 if no plugin with the given name is registered.

Example response (200):

{
  "name": "internal_audit_annotator",
  "version": "0.9.1",
  "enabled": true,
  "hooks_registered": ["on_startup", "on_scan_complete"]
}

---

Disable a Plugin

POST /admin/api/plugins/{name}/disable

Disables an active plugin. The plugin instance remains registered and in memory but will not receive further hook calls until re-enabled.

• Returns 200 with the updated PluginInfo object on success.
• Returns 404 if no plugin with the given name is registered.

Example response (200):

{
  "name": "high_severity_logger",
  "version": "1.2.0",
  "enabled": false,
  "hooks_registered": ["on_startup", "on_shutdown", "on_scan_complete"]
}

Note

Disabling a plugin via the API does not call on_shutdown(). The on_shutdown() hook is only called during outpost graceful shutdown. Use the enable/disable API to temporarily suppress a plugin’s hook calls without losing its in-memory state.

---

Built-in: Webhook Emitter Plugin

The outpost ships with a built-in webhook emitter plugin. When enabled, it posts a structured JSON payload to a configurable URL after every DLP scan completes. This is the recommended integration point for SIEM systems, alerting pipelines, and custom dashboards that need real-time scan event data.

Configuration

Variable	Default	Type	Description
WEBHOOK_EMIT_ENABLED	false	true/false	Enable the webhook emitter plugin
WEBHOOK_EMIT_URL	(none)	URL	Destination URL for webhook POST requests
WEBHOOK_EMIT_SECRET	(none)	string	Shared secret used to compute the HMAC-SHA256 signature on each payload

Danger

WEBHOOK_EMIT_URL must be set when WEBHOOK_EMIT_ENABLED=true. The outpost will fail to start if this variable is missing.

Payload Schema

The webhook emitter sends an HTTP POST to WEBHOOK_EMIT_URL with Content-Type: application/json. The body is a JSON object with the following structure:

{
  "event": "scan_complete",
  "timestamp": "2026-03-15T14:30:00Z",
  "scan_id": "a3f2c1d4-e5b6-7890-abcd-ef1234567890",
  "severity": "high",
  "detections_count": 3,
  "policy_action": "block",
  "metadata": {
    "source_ip": "10.0.1.50",
    "user_id": "user-42",
    "model": "gpt-4"
  }
}

Field	Type	Description
event	string	Always "scan_complete" in this version
timestamp	string	ISO 8601 UTC timestamp of the scan completion
scan_id	string	UUID uniquely identifying this scan event
severity	string	Highest severity among all detections: none, low, medium, high, critical
detections_count	integer	Total number of entity detections in this scan
policy_action	string	Action taken: allow, block, redact, or warn
metadata.source_ip	string	IP address of the originating client
metadata.user_id	string	Authenticated user identifier, if available
metadata.model	string	LLM model name targeted by this request

HMAC Signing

When WEBHOOK_EMIT_SECRET is configured, each webhook POST includes an X-Webhook-Signature header containing the HMAC-SHA256 of the raw JSON request body, encoded as a lowercase hexadecimal string.

Receiver verification example (Python):

import hashlib
import hmac

def verify_webhook(raw_body: bytes, signature_header: str, secret: str) -> bool:
    """
    Returns True if the webhook payload signature is valid.

    raw_body         — the raw bytes of the request body (do not parse JSON first)
    signature_header — value of the X-Webhook-Signature header
    secret           — the shared secret configured in WEBHOOK_EMIT_SECRET
    """
    expected = hmac.new(
        key=secret.encode("utf-8"),
        msg=raw_body,
        digestmod=hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(expected, signature_header)

Always use hmac.compare_digest for signature comparison to prevent timing-based side-channel attacks.

Caution

If WEBHOOK_EMIT_SECRET is not set, payloads are sent without a signature and the X-Webhook-Signature header is omitted. For production deployments, always configure a secret and verify signatures on your receiver to prevent spoofed webhook deliveries.

Retry Behavior

The webhook emitter retries failed deliveries up to 3 times with exponential backoff:

Attempt	Delay before retry
1 (initial)	—
2	1 second
3	2 seconds
4 (final)	4 seconds

A delivery is considered failed if the HTTP response status is not in the 2xx range, if the connection times out (5-second timeout per attempt), or if a network error occurs.

Caution

Webhook delivery is best-effort. If all retry attempts fail, the event is dropped and the failure is recorded in the emitter stats. The outpost does not queue events to disk. If your receiver is unavailable for an extended period, events will be lost. Monitor the events_failed counter in the stats endpoint and configure alerting on sustained failure rates.

Webhook Emitter Stats

GET /admin/api/webhook-emitter/stats

Returns delivery statistics for the webhook emitter since the last outpost startup.

Response fields:

Field	Type	Description
enabled	boolean	Whether the webhook emitter plugin is currently active
events_sent	integer	Total events successfully delivered (HTTP 2xx received)
events_failed	integer	Total events dropped after exhausting all retry attempts
last_event_at	string or null	ISO 8601 timestamp of the most recently delivered event, or null if no events have been sent

Example response:

{
  "enabled": true,
  "events_sent": 14382,
  "events_failed": 7,
  "last_event_at": "2026-03-15T14:29:58Z"
}

---

Custom Entity Redaction Patterns

Custom redaction patterns extend the outpost’s DLP engine with organization-specific entity detection rules. Each pattern is a named regular expression that, when matched in a scanned payload, is replaced with a configurable redaction string before the content is forwarded downstream.

Custom patterns are evaluated as a post-processing step after the core DLP engine completes its built-in entity detection pass. This means they operate on content that may have already been partially redacted by built-in rules.

Configuration

Variable	Default	Type	Description
CUSTOM_REDACTION_PATTERNS	(none)	JSON string	Initial set of custom patterns loaded at startup before the admin API is available

The value must be a JSON array of pattern objects. This is most useful for patterns that must be available immediately at startup, before any admin API calls can be made.

Example CUSTOM_REDACTION_PATTERNS value:

[
  {
    "name": "internal_employee_id",
    "pattern": "EMP-[0-9]{6}",
    "replacement": "[REDACTED_EMPLOYEE_ID]",
    "enabled": true
  },
  {
    "name": "project_codename",
    "pattern": "\\b(PROJECT_HELIOS|PROJECT_TITAN|PROJECT_NOVA)\\b",
    "replacement": "[REDACTED_PROJECT_NAME]",
    "enabled": true
  }
]

Note

When embedding a JSON string in a shell environment variable, remember to escape backslashes. A regex \b must be written as \\b in the JSON string value.

List Custom Patterns

GET /admin/api/custom-patterns

Returns all custom redaction patterns currently registered, including both patterns loaded from CUSTOM_REDACTION_PATTERNS at startup and patterns added via the API.

Response fields:

Field	Type	Description
name	string	Unique pattern name
pattern	string	Regular expression string
replacement	string	Text substituted for each match (e.g. "[REDACTED_INTERNAL_ID]")
enabled	boolean	Whether this pattern is currently applied during scans

Example response:

[
  {
    "name": "internal_employee_id",
    "pattern": "EMP-[0-9]{6}",
    "replacement": "[REDACTED_EMPLOYEE_ID]",
    "enabled": true
  },
  {
    "name": "project_codename",
    "pattern": "\\b(PROJECT_HELIOS|PROJECT_TITAN|PROJECT_NOVA)\\b",
    "replacement": "[REDACTED_PROJECT_NAME]",
    "enabled": false
  }
]

---

Create a Custom Pattern

POST /admin/api/custom-patterns

Creates a new custom redaction pattern. The pattern is validated as a regular expression before being stored. If the pattern is syntactically invalid, the request is rejected with a 422 response.

Request body:

{
  "name": "vendor_contract_id",
  "pattern": "VCI-[A-Z]{3}-[0-9]{8}",
  "replacement": "[REDACTED_CONTRACT_ID]",
  "enabled": true
}

Success response (201):

{
  "name": "vendor_contract_id",
  "pattern": "VCI-[A-Z]{3}-[0-9]{8}",
  "replacement": "[REDACTED_CONTRACT_ID]",
  "enabled": true
}

Error response — invalid regex (422):

{
  "error": "invalid_pattern",
  "message": "The provided pattern is not a valid regular expression.",
  "detail": "unterminated character class at position 12: '[A-Z{3}'"
}

Error response — name conflict (409):

{
  "error": "pattern_exists",
  "message": "A custom pattern with name 'vendor_contract_id' already exists.",
  "detail": "Use DELETE /admin/api/custom-patterns/vendor_contract_id to remove the existing pattern before re-creating it."
}

---

Delete a Custom Pattern

DELETE /admin/api/custom-patterns/{name}

Permanently removes a custom redaction pattern. The pattern stops being applied to new scans immediately.

• Returns 204 No Content on success.
• Returns 404 if no pattern with the given name exists.

Caution

Deleting a custom pattern is immediate and irreversible via the API. Patterns loaded from the CUSTOM_REDACTION_PATTERNS environment variable are re-loaded on the next outpost restart, so deleting an env-originated pattern via the API will result in it reappearing after restart. Remove the entry from the environment variable as well if you want the deletion to be permanent.

Tip

Custom patterns are applied in the order they appear in the pattern list, which reflects creation order. The first pattern that matches a given substring wins — subsequent patterns are still evaluated against the remainder of the content. If you need a specific evaluation order, create patterns in the desired sequence. There is currently no API to reorder patterns; delete and recreate them in the desired order if reordering is needed.

---

Summary and Next Steps

The Arbitex Outpost plugin system provides a structured, low-friction path to extending DLP behavior without forking or patching the outpost binary. The three-hook protocol (on_startup, on_shutdown, on_scan_complete) covers the majority of integration use cases, and the built-in webhook emitter handles the most common production requirement — delivering scan events to external systems — out of the box.

Custom entity redaction patterns complement the plugin system by allowing fine-grained, regex-based additions to the DLP detection surface without requiring any code.

Related guides:

• Outpost Configuration Reference — full environment variable reference for the outpost
• DLP Policy Management — built-in entity types, policy actions, and severity thresholds
• DLP Accuracy and Tuning — improving detection quality, reducing false positives, and understanding the scoring model
• DLP Detection Hierarchy — how built-in rules, custom patterns, and policy overrides interact
• Outpost Security Hardening — mTLS, admin API access controls, and network isolation recommendations
• Outpost Administration — budget enforcement, CredInt, health monitoring, JWT validation, PVC recovery, security hardening, SIEM direct integration