Usage

Own one master

The caller selects an OpenSSH authority and supplies trusted executable paths. An alias delegates host, proxy, host-key, and identity policy to normal OpenSSH configuration:

import asyncio
from pathlib import Path

from ssh_wrapper import ConnectionSpec, OpenSSHMaster, SSHMasterSettings


async def hostname() -> bytes:
    master = OpenSSHMaster(
        SSHMasterSettings(
            ssh_path=Path("/usr/bin/ssh"),
            false_path=Path("/usr/bin/false"),
            connect_timeout=30,
        ),
        ConnectionSpec.from_alias("workstation"),
    )
    await master.start()
    try:
        process = await asyncio.create_subprocess_exec(
            *master.command_argv("hostname"),
            stdout=asyncio.subprocess.PIPE,
        )
        stdout, _stderr = await process.communicate()
        if process.returncode:
            raise RuntimeError(f"remote command exited {process.returncode}")
        return stdout
    finally:
        await master.close()

Direct authority is also available when configuration aliases are not wanted:

connection = ConnectionSpec.from_direct("203.0.113.10", "deploy", 22)

The destination, user, and port remain distinct argv values. A direct host does not opt out of OpenSSH host-key checking.

Use mux-aware consumers

master.mux_transport_argv() returns SSH options without a destination. master.command_argv(remote_program) appends one destination and remote program. master.mux_ssh_command() is a shell-escaped transport string for a mux-aware consumer, and master.create_mux_wrapper() creates a private executable wrapper when a consumer accepts only a path.

These forms contain explicit barriers against new authentication and proxy fallback. If the master disappears, construct a new OpenSSHMaster only after your application deliberately decides to authenticate again. The remote-program string is interpreted by the remote shell. Quote individual application arguments with shlex.join() when constructing such a program; never interpolate untrusted text directly into shell syntax. The embedding application remains responsible for cancelling and reaping ordinary secondary subprocesses, including when its own task is cancelled.

Own a long-lived remote child

Use OwnedRemoteProcess only when the remote target provides python3 and the POSIX process/session operations used by the supervisor, and the child should remain tied to a heartbeat lease (Linux loopback is covered by acceptance):

from ssh_wrapper import OpenSSHMaster, OwnedRemoteProcess


async def supervise(master: OpenSSHMaster) -> None:
    owned = OwnedRemoteProcess(
        master,
        ("worker", "--mode", "safe"),
        heartbeat_interval=5,
        lease_timeout=20,
        grace_timeout=5,
    )
    await owned.start()
    try:
        # Inspect owned.stdout_head, owned.stdout_tail, or owned.stderr_tail.
        await owned.wait()
    finally:
        await owned.close()

Application arguments are encoded as data. Supervision creates and cleans one remote process group; it does not kill unrelated processes or a daemon that intentionally escapes that group.

The default diagnostic tail is 16 KiB. If tail_bytes is customized, pass a positive integer so stdout and stderr retention stays bounded. Select positive finite timeouts and leave a margin between the heartbeat interval and lease. A cancelled or timed-out wait() does not end ownership; the finally block's close() does. Cleanup includes surviving members of the same process group even if its leader exits first.

See the API guide and security guide for the ownership and diagnostic boundaries.