Python Modules

Python is the quickest way to write a Hovel module: there is no build step, and the daemon launches it straight from source through the interpreter. The SDK lives at sdk/python/hovel_sdk; the worked examples are under modules/examples/python/.

The SDK in one import

You subclass HovelModule, set a few class attributes, implement run, and hand an instance to serve.

The generated API reference for the source package is published at Python SDK API.

from hovel_sdk import (
    Artifact, Context, Finding, HovelModule,
    LineShellSession, Requirement, Result, serve,
)

• HovelModule — base class; declares metadata and the run method.
• Requirement — one configuration field.
• Context — inputs, logging, and sessions for one execution.
• Result, Finding, Artifact — what run returns.
• LineShellSession — a base for interactive shell sessions.
• serve — runs the JSON-RPC loop on stdin/stdout.

Treat stdout as reserved for serve. A stray print(), progress bar, subprocess inherited stdout, or banner will corrupt the frame stream and fail the module. Use ctx.log or the standard logging package for operator-visible progress, and capture tool output into artifacts or sessions.

Metadata and schema rules

Class attributes are not just documentation: the daemon calls them during catalog inspection. Keep info, module_schema, and describe_steps deterministic and side-effect free. Put target probes, listener startup, payload generation, random target-side names, and expensive imports inside run, prepare_step, or execute_step, not at import time.

Config values arrive as JSON-compatible Python values. Prefer explicit Requirement entries over ad hoc environment reads so operators, saved chains, and transcripts all agree on what controlled the run.

TLS, workspace PKI, and Mesh bridges

A normal Python module may use ssl or another TLS library inside run without overriding any credential hook. Override credential methods only when the module is a provider that consumes or transforms assignment-bound workspace-PKI material. Base execute does not receive an automatically resolved PKI bundle.

Override only advertised operations; inherited methods fail explicitly. CredentialBytes, CredentialSecretReference, and CredentialProtectedPath redact ordinary repr, but explicit access still reveals the secret to provider code. Keep it scoped to the invocation and out of logs, exceptions, results, and durable configuration. Current external initiation delivers direct runtime certificate DER, public-key SPKI, or private-key PKCS #8 selections before Mesh operations; general non-Mesh delivery and external encoding/stamping initiation are not wired. Continue with TLS and Workspace PKI, Credential Provider Development, and the generated Python SDK API.

Connect to an authenticated Mesh bridge

An existing TCP or UDP client needs no Mesh-provider override. Build an endpoint from the transient OpenMeshBridge response and let the SDK send the bearer capability before application data.

from hovel_sdk import MeshBridgeEndpoint, MeshBridgeNetwork, connect_mesh_bridge


endpoint = MeshBridgeEndpoint.from_rpc(bridge)
with connect_mesh_bridge(endpoint, timeout=5.0) as connection:
    if endpoint.local_network is MeshBridgeNetwork.UDP:
        connection.send(request_bytes)
    else:
        connection.sendall(request_bytes)

MeshBridgeEndpoint.from_rpc() validates and retains localNetwork, preventing a TCP/UDP mismatch and rejecting an unknown local adapter. The provider-defined request.protocol is independent. Never log or persist the capability. Use a provider-native task or session for protocols that cannot preserve TCP stream or UDP datagram semantics. See Build a Module or Provider for the complete decision table.

A survey module

A survey gathers facts and returns them in outputs. Class attributes become the handshake and schema responses; run becomes execute.

from hovel_sdk import Context, HovelModule, Requirement, Result


class MockSurvey(HovelModule):
    name = "mock-survey"
    version = "v0.0.0-example"
    module_type = "survey"
    summary = "Collect example target facts."
    tags = ("example", "survey", "python")
    target_config = (
        Requirement("target.host", "host", description="Target host name or IP address."),
        Requirement("target.port", "port", description="Target TCP port."),
    )

    def run(self, ctx: Context) -> Result:
        host = ctx.input("target.host", ctx.target)
        port = ctx.input("target.port", "unknown")
        ctx.log.info("connecting to target %s:%s", host, port)
        return Result.ok(
            {"facts": {"host": host, "port": port, "reachable": True}},
            summary=f"example survey reached {host}:{port}",
        )

ctx.input(key, default) resolves the value across run inputs, target config, and chain config. ctx.log is a standard Python logger whose records are forwarded to the daemon as module/log notifications.

Findings and artifacts

An exploit reports Findings and Artifacts, and can fail deliberately via an input.

def run(self, ctx: Context) -> Result:
    if ctx.input("failure_mode", "") == "execution":
        return Result.failed("example exploit failed during execution")

    return Result.ok(
        {"target": ctx.target},
        summary="example exploit completed without target interaction",
        findings=[Finding(
            title="example exploit path verified",
            severity="info",
            detail="exercised orchestration without touching a real target",
        )],
        artifacts=[Artifact.inline(
            name="mock-exploit-transcript.txt",
            kind="text/plain",
            data="example module invoked through daemon RPC",
        )],
    )

Artifact also offers text, json, and file constructors for common cases.

Use Result.failed(...) for expected operator-facing failures such as "not vulnerable", "authentication failed", or "confirmation missing". Let exceptions mean bugs or unexpected infrastructure failures; the SDK will report them as JSON-RPC errors and include traceback details in module logs.

Installed payload records

If a Python exploit or provider installs a durable target-side payload, return an explicit installed-payload descriptor. This is the only supported way for a module to ask Hovel to create installed payload inventory.

from hovel_sdk import InstalledPayload, PayloadProviderRecord, Result


payload = InstalledPayload(
    provider="squatter",
    payload_id="squatter/windows/x86/xp-sp3/tcp-bind/pe-exe",
    payload_version="0.1.0",
    target=ctx.target,
    state="installed",
    transport="tcp-bind",
    endpoint="10.0.0.42:9101",
    supports_reconnect=True,
    cleanup=PayloadProviderRecord(
        schema="squatter.cleanup.tcp_bind",
        descriptor={"remotePath": r"C:\Windows\Temp\n4x9q2.exe"},
    ),
)
return Result.ok(
    {"payload": payload.payload_id},
    summary="payload installed and recorded",
).with_installed_payloads(payload)

Do not write workspace.db or invent operator-facing payload handles. Hovel validates descriptors, assigns handles, persists state, and routes future reconnect or cleanup calls back to the provider.

Interactive sessions

An exploit can open a session that outlives run. Subclass LineShellSession and answer commands; ctx.open_session registers it. The daemon keeps the process alive and lets the operator drive the shell.

class MockShell(LineShellSession):
    def __init__(self, target: str) -> None:
        super().__init__(prompt="mock$ ", echo=True)
        self._target = target

    async def handle_command(self, command: str) -> str:
        if command == "whoami":
            return "mock-operator"
        if command == "hostname":
            return self._target.removeprefix("mock://") or "mock-target"
        return f"mock-shell: command not found: {command}"


class MockExploitSession(HovelModule):
    name = "mock-exploit-session"
    module_type = "exploit"
    tags = ("example", "exploit", "python", "session")

    async def run(self, ctx: Context) -> Result:
        session = await ctx.open_session(
            MockShell(ctx.target),
            name=f"mock shell on {ctx.target}",
            capabilities=("read", "write", "exec", "close"),
        )
        return Result.ok(
            {"sessionId": session.id},
            summary="mock exploit opened an interactive shell session",
        )

run may be synchronous or async; the SDK awaits it either way. LineShellSession handles line buffering, echo, the prompt, and the built-in exit/logout commands for you.

Step-provider hooks

Python is suitable for real exploit modules and for lightweight chain-step providers. Override the step methods when the module participates in the capability-native chain planner.

class WindowsCredentialStep(HovelModule):
    name = "windows-credential"
    module_type = "exploit"
    tags = ("dangerous", "credential", "python")

    def describe_steps(self) -> dict[str, object]:
        return {"steps": [{
            "id": "windows.credential.create_local_admin",
            "kind": "credential.create",
            "requires": [{"type": "RemoteExecutionCapability", "states": ["active"]}],
            "produces": [{"type": "CredentialCapability", "states": ["active"]}],
            "prepare": {"materializes": ["username", "password"]},
        }]}

    def prepare_step(self, request: dict[str, object]) -> dict[str, object]:
        return {
            "preparedValues": {
                "username": {"value": "neutral-name", "editable": True},
                "password": {"value": "plain-high-entropy-password", "editable": True},
            },
            "operatorSummary": {
                "warnings": [],
                "targetSideArtifacts": ["temporary local administrator credential"],
            },
        }

    def execute_step(self, request: dict[str, object]) -> dict[str, object]:
        confirmed = dict(request.get("confirmedPreparedValues") or {})
        return {
            "status": "succeeded",
            "capabilities": [{
                "id": "cap_credential_example",
                "type": "CredentialCapability",
                "schemaVersion": "v1",
                "state": "active",
                "attributes": {
                    "protocol": "smb",
                    "username": confirmed["username"],
                    "password": confirmed["password"],
                    "sensitive": True,
                },
            }],
        }

    def cleanup_step(self, request: dict[str, object]) -> dict[str, object]:
        return {"status": "removed"}

The Python SDK leaves step payloads as dictionaries so providers can move fast while the capability schema evolves. The Wire Protocol page is the source of truth for field names.

Entry point

Each module package has a __main__.py so the daemon can run python -m <module>.

from hovel_example_survey.module import MockSurvey
from hovel_sdk import serve


def main() -> None:
    serve(MockSurvey())


if __name__ == "__main__":
    main()

Test without the daemon

Unit-test module logic directly first. Construct a Context, call run, and assert the result shape. Session modules can use SessionManager to exercise ctx.open_session without starting hoveld.

from hovel_example_exploit.module import MockExploit
from hovel_sdk import Context


def test_failure_mode_returns_failed_result() -> None:
    result = MockExploit().run(Context(
        run_id="run-1",
        module_id="mock-exploit",
        target="mock://target",
        inputs={"failure_mode": "execution"},
    ))
    assert result.status == "failed"

Then use hovel_sdk.testing.ModuleRPC for contract tests that exercise the real framed JSON-RPC server. This catches stdout corruption, schema drift, step dispatch mistakes, session RPC errors, and shutdown behavior earlier than a full CLI run.

import base64

from hovel_example_exploit_session.module import MockExploitSession
from hovel_sdk.testing import ModuleRPC


def test_session_over_framed_rpc() -> None:
    with ModuleRPC(MockExploitSession()) as rpc:
        result = rpc.call("execute", {
            "runId": "run-1",
            "moduleId": "mock-exploit-session",
            "target": "mock://target",
        })
        session_id = result["sessions"][0]["id"]

        prompt = rpc.call("session/read", {"sessionId": session_id, "timeoutMs": 100})
        rpc.call("session/write", {
            "sessionId": session_id,
            "data": base64.b64encode(b"whoami\n").decode(),
        })
        output = rpc.call("session/read", {"sessionId": session_id, "timeoutMs": 100})

        assert base64.b64decode(prompt["data"]).endswith(b"$ ")
        assert base64.b64decode(output["data"])

ModuleRPC.notifications records module/log and module/session notifications emitted before each response. drive_module(module, script) is a shorter wrapper for tests that only need the final notification list.

The Python SDK and in-repo Python examples now live outside the core Bazel workspace. Validate them through task sdk:test, task sdk:ci, task modules:examples:test, and task modules:examples:ci; the Task-backed targets remain hermetic.

Move to daemon or CLI tests only after pure parsing, schema, failure paths, artifact construction, and session behavior are covered. Real network-facing exploit tests should keep packet builders and response parsers unit-testable offline, then reserve live target checks for an authorized lab.

Package and run

Package or link the module with a hovel-module.yaml manifest. Python launchers can use Hovel-managed python-build-standalone, an operator-supplied interpreter, or a bundled interpreter.

apiVersion: hovel.dev/v1alpha1
kind: ModulePackage
metadata:
  name: mock-survey
  version: 0.1.0
  moduleType: survey
runtime:
  protocol: jsonrpc-stdio
launch:
  - selector: {}
    python:
      managed:
        versions: ["3.10-3.14"]
      command: ["{python}", "-m", "hovel_example_survey"]

hovel module install --link "$PWD/modules/examples/python/mock_survey"
hovel module check mock-survey
task start
hovel> module installed  # mock-survey, mock-exploit, mock-exploit-session ...

Python modules need no native compile step. See the Go and Rust guides for the same three modules as compiled binaries.