Example Plugin

examples/echo_plugin/main.py is a reference implementation that mirrors the Rust test server used in the trustee project's own e2e suite.

What it does

• POST: stores or echoes the request body, depending on PLUGIN_STORE_DATA
• GET: retrieves stored data or echoes the request, depending on PLUGIN_STORE_DATA
• Other methods: returns HTTP 400

Configuration

Configure entirely via environment variables:

Variable	Default	Description
PLUGIN_LISTEN_ADDR	127.0.0.1:50051	gRPC bind address
PLUGIN_TLS_CERT	unset	Path to TLS certificate (enables TLS when both cert and key are set)
PLUGIN_TLS_KEY	unset	Path to TLS private key
PLUGIN_STORE_DATA	false	true → POST stores body, GET retrieves it; otherwise both echo
PLUGIN_ENCRYPT_GET	false	true → GET responses require JWE encryption

POST always requires admin auth; GET never does.

Run with Docker

A Dockerfile.echo-plugin is provided in e2e/:

docker build -f e2e/Dockerfile.echo-plugin -t echo-plugin .
docker run --rm -p 50051:50051 echo-plugin

Implementation walkthrough

The echo plugin implements all three PluginHandler methods.

handle

Routes by method. In store mode, POST writes to an in-memory dict and GET reads from it. Outside store mode, both methods return an echo string with the method, path, query params, and body length.

async def handle(self, request: PluginRequest) -> PluginResponse:
    key = "/".join(request.path)
    if request.method == "POST":
        if self._store_data:
            with self._lock:
                self._store[key] = bytes(request.body)
            return PluginResponse(body=b"", status_code=200, content_type="")
        body = self._echo(request.method, request)
        return PluginResponse(body=body, status_code=200, content_type="text/plain")
    if request.method == "GET":
        with self._lock:
            stored = self._store.get(key)
        if stored is not None:
            return PluginResponse(body=stored, status_code=200,
                                  content_type="application/octet-stream")
        body = self._echo(request.method, request)
        return PluginResponse(body=body, status_code=200, content_type="text/plain")
    return PluginResponse(
        body=f"method not supported: {request.method}".encode(),
        status_code=400, content_type="text/plain",
    )

validate_auth

Returns True only for POST — POST requires admin credentials, GET does not.

async def validate_auth(self, request: ValidateAuthRequest) -> bool:
    return request.method == "POST"

needs_encryption

Returns True for GET when PLUGIN_ENCRYPT_GET=true.

async def needs_encryption(self, request: NeedsEncryptionRequest) -> bool:
    return self._encrypt_get and request.method == "GET"