Host

TrueNASHost is a hostctl PosixHost: run, path, spawn, info, connect, close, shell, capabilities, and last_selection are inherited, and it adds the TrueNAS surface (api, websocket, login, subscribe, upload, download, dump_api, install_sshcreds).

TrueNASConfig is the matching HostConfig. It accepts every connection string TrueNASClient does and normalizes to a truenas+* scheme, so HostConfig("truenas+wss://nas") resolves through hostctl's own registry.

pytruenas.host

hostctl integration: the TrueNAS host configuration.

:class:TrueNASConfig is the hostctl :class:~hostctl.host.HostConfig for a TrueNAS middleware target. It carries parsed connection settings and nothing more -- constructing one performs no network I/O, which is what lets a config be built offline, logged, and round-tripped through its canonical URI.

Scheme handling has two deliberately separate layers:

• The hostctl registry sees only truenas/truenas+auto/truenas+ws/ truenas+wss/truenas+unix. pytruenas does not claim bare wss:// or https:// globally -- hostctl is protocol-agnostic, and hijacking a generic scheme would risk an "ambiguous host URI matched" collision with any other configuration that legitimately wants it.
• pytruenas' own entry point additionally understands every connection string :class:~pytruenas.TrueNASClient has always accepted -- a bare host, host:port, ws/wss, http/https, a unix socket path, or None -- by rewriting it to a truenas+* URI first. That rewrite is :func:_normalize_target: one pure string function, no I/O.

The scheme and API-path probes that :class:TrueNASClient historically ran inside __init__ are recorded here (:attr:~TrueNASConfig.needs_scheme_probe, :attr:~TrueNASConfig.needs_path_probe) and performed later, on connect.

AUTO_SCHEME = 'truenas+auto' module-attribute

ApiVersion = _ty.TypeVar('ApiVersion', bound=_Namespace, default=Current) module-attribute

DEFAULT_SOCKET_PATH = DEFAULT_UNIX_SOCKET module-attribute

EXECUTOR_NAMES = ('local', 'ssh', 'webshell') module-attribute

PATH_NAMES = ('local', 'sftp', 'tnasws') module-attribute

__all__ = ['AUTO_SCHEME', 'DEFAULT_SOCKET_PATH', 'TrueNASConfig', 'TrueNASHost'] module-attribute

TrueNASConfig(host='', *, port=0, secure=None, socket_path=None, api_path=None, version='current', sslverify=True, credentials=None, ssh=None, shell=None, executor=None, path=None, autologin=True, logger=None)

Bases: HostConfig

Parsed, credential-safe connection settings for a TrueNAS middleware host.

Source code in src/pytruenas/host.py

def __init__(
    self,
    host: str = "",
    *,
    port: int = 0,
    secure: "bool | None" = None,
    socket_path: "str | None" = None,
    api_path: "str | None" = None,
    version: str = "current",
    sslverify: bool = True,
    credentials: object = None,
    ssh: object = None,
    shell: "str | None" = None,
    executor: "_ty.Iterable[str] | str | None" = None,
    path: "_ty.Iterable[str] | str | None" = None,
    autologin: bool = True,
    logger: object = None,
) -> None:
    super().__init__()
    #: Whether the first websocket access logs in automatically.
    self.autologin = autologin
    #: Logger for the client built from this config; a name or a Logger.
    self.logger = logger
    #: Explicit provider selection, overriding the defaults. Each is a name
    #: or a sequence of names, in preference order; ``None`` means "decide
    #: from the target". See :meth:`TrueNASHost._build_providers`.
    #:
    #: This replaced a ``webshell: bool`` flag, which was only ever the one
    #: hardcoded case of it (``executor=["ssh"]`` says the same thing, and
    #: says it in a form that also covers "SSH only", "force the web
    #: shell", or any other combination).
    self.executors = _as_names(executor)
    self.paths = _as_names(path)
    self.host = host
    self.port = int(port or 0)
    #: ``True`` -> wss, ``False`` -> ws, ``None`` -> probe on connect.
    self.secure = secure
    self.socket_path = socket_path
    self.api_path = api_path
    self.version = version
    self.sslverify = sslverify
    #: The SSH leg. Accepts a ready :class:`hostctl.host.SshConfig` or, via
    #: ``shell=``, the connection string ``TrueNASClient`` has always taken
    #: (``"ssh://root@nas"``, ``"root@nas:22"``) -- so a caller does not
    #: have to import and assemble an SshConfig for the common case.
    self.ssh = ssh if ssh is not None else _ssh_config_from(shell)
    # Credentials are normalized once, here, so every downstream consumer
    # sees a Credentials instance rather than "maybe a string, maybe a
    # tuple, maybe None".
    self.credentials = (
        credentials
        if isinstance(credentials, _auth.Credentials)
        else _auth.Credentials(credentials)
    )

api_path = api_path instance-attribute

autologin = autologin instance-attribute

connection_uri property

The canonical, credential-free URI for this configuration.

credentials = credentials if isinstance(credentials, _auth.Credentials) else _auth.Credentials(credentials) instance-attribute

executors = _as_names(executor) instance-attribute

host = host instance-attribute

is_local property

Whether this target is the local middleware unix socket.

logger = logger instance-attribute

name property

A short label for this target: the hostname, not a whole URI.

This is what belongs in a log prefix or a progress line. The scheme, port, API path and userinfo that :attr:connection_uri carries are noise once every record on the line repeats them -- and on a fan-out across ten hosts, the one thing a reader needs is which machine.

localhost for the local middleware socket; the bare hostname otherwise, with the port appended only when it is non-default (a :8443 is a real distinguisher between two entries for one host, where :443 just repeats the scheme).

needs_path_probe property

Whether the API path still has to be resolved against the server.

needs_scheme_probe property

Whether ws-vs-wss still has to be resolved against the server.

paths = _as_names(path) instance-attribute

port = int(port or 0) instance-attribute

secure = secure instance-attribute

socket_path = socket_path instance-attribute

ssh = ssh if ssh is not None else _ssh_config_from(shell) instance-attribute

sslverify = sslverify instance-attribute

uri_credentials = ('password', 'otp', 'api_key', 'token', 'credentials', 'sslverify', 'ssh', 'version', 'shell', 'executor', 'path', 'autologin', 'logger') class-attribute instance-attribute

version = version instance-attribute

__repr__()

Source code in src/pytruenas/host.py

def __repr__(self) -> str:
    # Never render credentials: a config is logged, and connection_uri is
    # already the credential-free canonical form.
    return f"{type(self).__name__}({self.connection_uri!r})"

from_target(target=None, **options) classmethod

Build a config from any connection string TrueNASClient accepts.

This is pytruenas' entry point: it normalizes the string first (so bare wss:// and friends work) and then hands it to hostctl's registry, which strips and parses any credentials in the userinfo.

Source code in src/pytruenas/host.py

@classmethod
def from_target(
    cls, target: "str | None" = None, **options: object
) -> "TrueNASConfig":
    """Build a config from any connection string ``TrueNASClient`` accepts.

    This is pytruenas' entry point: it normalizes the string first (so bare
    ``wss://`` and friends work) and then hands it to hostctl's registry,
    which strips and parses any credentials in the userinfo.
    """
    uri = _normalize_target(target)
    config = _HostConfig._from_uri(uri, **options)
    if not isinstance(config, cls):  # pragma: no cover - registry misconfig
        raise TypeError(f"{uri!r} did not resolve to a {cls.__name__}")
    return config

TrueNASHost(config=None, credentials=None, *, client=None, **options)

Bases: PosixHost, Generic[ApiVersion]

A TrueNAS middleware host: POSIX semantics over composed transports.

Everything generic -- run, path, spawn, info, connect, close, shell, capabilities, last_selection -- is inherited from :class:hostctl.host.PosixHost, which selects between the providers assembled below. What this class adds is only the part no other host has: the middleware JSON-RPC websocket and the API surface built on it.

Construct it from a connection string, a :class:TrueNASConfig, or nothing at all (the local middleware socket)::

TrueNASHost("wss://nas")
TrueNASHost("nas", credentials="1-...", executor=["ssh"])
TrueNASHost(TrueNASConfig.from_target("wss://nas"))
TrueNASHost()

A string accepts every form :class:~pytruenas.TrueNASClient does and takes the same keyword options as :meth:TrueNASConfig.from_target.

Provider order depends on the target, and is overridable per selector with executor=/path=:

• local -- hostctl's stock local pair, and nothing else. Reaching this same machine over SSH, a PTY, or the filesystem API would be slower and strictly less capable.
• remote -- ssh then webshell for commands, sftp then tnasws for paths. SSH leads on capability; the websocket legs follow. This reproduces :class:~pytruenas.fs.truenas.TruenasPath's hand-rolled fallback through hostctl's selector, which additionally records a redacted trace of what was tried (host.last_selection).

Source code in src/pytruenas/host.py

def __init__(
    self,
    config: "TrueNASConfig | str | None" = None,
    credentials: object = None,
    *,
    client: object = None,
    **options: object,
) -> None:
    # `TrueNASClient(target, creds)` passed credentials positionally, and
    # that is the single most common call in the wild -- accept it.
    if credentials is not None:
        if "credentials" in options:
            raise TypeError("credentials given both positionally and by keyword")
        options["credentials"] = credentials
    # A connection string is the common case, so accept it directly rather
    # than making every caller reach for TrueNASConfig.from_target first.
    # hostctl's own `Host("uri")` shortcut cannot help here: its metaclass
    # only intercepts when `cls is Host`, so a subclass falls straight
    # through to normal construction.
    #
    # Keywords are split by destination: everything TrueNASConfig accepts
    # builds the config, and the rest (`info=`, `initializer=`, ...) goes
    # on to SystemHost. Splitting rather than guessing keeps a typo an
    # error from whichever layer owns the name.
    if config is None or isinstance(config, str):
        config_options = {
            key: options.pop(key) for key in list(options) if key in _CONFIG_OPTIONS
        }
        # Anything left that SystemHost does not take is a typo. Catch it
        # here rather than letting it reach SystemHost, which would raise a
        # TypeError naming an internal class -- unhelpful for a caller who
        # wrote `passwrd=` and needs to be told *that*.
        unknown = sorted(set(options) - _HOST_OPTIONS)
        if unknown:
            raise ValueError(
                f"unknown credential argument: {unknown[0]!r} "
                f"(configuration options: {', '.join(sorted(_CONFIG_OPTIONS))})"
            )
        config = TrueNASConfig.from_target(
            config, **_ty.cast(_ty.Any, config_options)
        )
    elif any(key in _CONFIG_OPTIONS for key in options):
        unexpected = ", ".join(
            sorted(key for key in options if key in _CONFIG_OPTIONS)
        )
        raise TypeError(
            "configuration options may not be combined with an existing "
            f"TrueNASConfig; pass them to from_target instead: {unexpected}"
        )
    self._config = config
    #: The live JSON-RPC connection, opened on first `.conn` access.
    self._conn: "_connection.TrueNASWSConnection | None" = None
    self.logger = _resolve_logger(config.logger, config.name)
    # `client=` is accepted and ignored: the host *is* the client now.
    # Kept so existing callers (and tests that injected a stand-in) do not
    # break on an unexpected keyword.
    del client

    executors, paths = self._build_providers(config)
    super().__init__(
        config,
        executor_providers=executors,
        path_providers=paths,
        **_ty.cast(_ty.Any, options),
    )

api cached property

The root API namespace (host.api.<namespace>.<method>(...)).

Parameterise the host to type it: TrueNASHost[Current]("nas").api completes exactly as the old TrueNASClient[Current] did.

client property

Deprecated alias for self.

The host is the client -- they were two objects forwarding halves of their surface to each other, which is now one class. Kept so host.client.api and similar keep working.

config_type = TrueNASConfig class-attribute instance-attribute

conn property

The live JSON-RPC connection; opens on first access.

Logs in first when autologin is set (the default) and there is no live connection. Reconnects if the previous one closed.

logger = _resolve_logger(config.logger, config.name) instance-attribute

name property

This host's short label -- see :attr:TrueNASConfig.name.

ssh property

The underlying asyncssh connection (requires the ssh extra).

For the rare caller that needs the raw connection -- port forwarding, an SFTP client of its own. Ordinary command and path work should go through :meth:run and :meth:path, which pick a transport rather than assuming this one exists.

sslverify property

TLS verification for this host -- see :attr:TrueNASConfig.sslverify.

Read from the config rather than stored, so every transport -- the JSON-RPC connection, the REST calls, and the web shell -- answers from the one value the caller set, with no second copy to drift.

websocket property

Former name of :attr:conn, kept because it is public API.

Defined as a property that reads self.conn rather than as websocket = conn: the latter makes two independent class attributes, so patching or overriding one would silently leave the other pointing at the original.

close()

Close the transports, then the websocket.

Order matters: the tnasws path provider talks over the websocket, so it must be torn down before the connection it depends on.

Source code in src/pytruenas/host.py

def close(self) -> None:
    """Close the transports, then the websocket.

    Order matters: the ``tnasws`` path provider talks over the websocket,
    so it must be torn down before the connection it depends on.
    """
    try:
        super().close()
    finally:
        conn, self._conn = self._conn, None
        if conn is not None:
            try:
                conn.close()
            except Exception:
                # close() must be safe to call repeatedly and must not mask
                # an error raised by the provider teardown above.
                pass

download(method, *args, filename=None, buffered=False, wait=True, **kwargs)

Call method for a download link and fetch it over HTTP(S).

Source code in src/pytruenas/host.py

def download(
    self,
    method: str,
    *args,
    filename: "str | None" = None,
    buffered=False,
    wait=True,
    **kwargs,
):
    """Call ``method`` for a download link and fetch it over HTTP(S)."""
    jobid, link = self.api.core.download(
        method, args, filename or "download", buffered, **kwargs
    )
    target = self._http_target(link)

    if wait:
        if buffered:
            self.api.core.job_wait(jobid, job=True, _timeout=None)
        resp = _req.get(target.uri, verify=self._config.sslverify)
        resp.raise_for_status()
        return resp.content
    return jobid

dump_api()

Run middlewared --dump-api on the target and parse the JSON.

Source code in src/pytruenas/host.py

def dump_api(self):
    """Run ``middlewared --dump-api`` on the target and parse the JSON."""
    import json

    from .models.apidump import Api

    api: Api = json.loads(
        self.run("middlewared --dump-api", capture_output=True).stdout
    )
    return api

install_sshcreds(name=None, private_key=None)

Provision an SSH keypair and wire it into this host's SSH transport.

Generates (or reuses) a SSH_KEY_PAIR keychain credential, installs the public half on root's authorized_keys, and stores the private half on :attr:TrueNASConfig.ssh as a real :class:hostctl.host.SshConfig -- with client_keys as an actual field rather than the "client_keys|root" string that the pre-hostctl client packed into a username.

Adding an SSH transport changes what this host can do, so the providers are rebuilt: a host that had no executor at all (remote, no SSH) gains one, and paths gain the richer SFTP leg.

Does not require the optional ssh extra: provisioning runs entirely over the middleware API and opens no SSH connection. The one exception is passing private_key= for a key the host does not already know, where the public half has to be derived locally (see :func:_public_key). Using the SSH transport this configures still needs the extra, as it always did.

Source code in src/pytruenas/host.py

def install_sshcreds(
    self, name: "str | None" = None, private_key: "str | None" = None
):
    """Provision an SSH keypair and wire it into this host's SSH transport.

    Generates (or reuses) a ``SSH_KEY_PAIR`` keychain credential, installs
    the public half on root's ``authorized_keys``, and stores the private
    half on :attr:`TrueNASConfig.ssh` as a real
    :class:`hostctl.host.SshConfig` -- with ``client_keys`` as an actual
    field rather than the ``"client_keys|root"`` string that the pre-hostctl
    client packed into a username.

    Adding an SSH transport changes what this host can do, so the providers
    are rebuilt: a host that had no executor at all (remote, no SSH) gains
    one, and paths gain the richer SFTP leg.

    Does **not** require the optional ``ssh`` extra: provisioning runs
    entirely over the middleware API and opens no SSH connection. The one
    exception is passing ``private_key=`` for a key the host does not
    already know, where the public half has to be derived locally (see
    :func:`_public_key`). *Using* the SSH transport this configures still
    needs the extra, as it always did.
    """
    name = name or "pytruenas"
    keypair = self.api.keychaincredential._get(type="SSH_KEY_PAIR", name=name)
    # The middleware hands back BOTH halves on the paths it generates or
    # stores, so the public key is usually already known -- see `_public_key`
    # for why that matters.
    pubkey: "str | None" = None
    if not keypair and not private_key:
        generated = self.api.keychaincredential.generate_ssh_key_pair()
        private_key = generated["private_key"]
        pubkey = generated.get("public_key")
    elif not private_key:
        attributes = keypair["attributes"]
        private_key = attributes["private_key"]
        pubkey = attributes.get("public_key")

    pubkey = (pubkey or "").strip() or _public_key(_ty.cast(str, private_key))
    keypair = self.api.keychaincredential._upsert(
        ("name", "type"),
        type="SSH_KEY_PAIR",
        name=name,
        attributes={"private_key": private_key, "public_key": pubkey},
    )
    root = self.api.user._get(username="root")
    authorized = (root.get("sshpubkey") or "").splitlines()
    if pubkey not in authorized:
        authorized.append(pubkey)
        # `("username",)` -- a one-item SEQUENCE, not the bare string. A
        # bare `str` selector is read as a record *id* (see
        # `DbAction.execute`); the tuple says "match on this field name",
        # which is what selecting root by username means.
        self.api.user._upsert(
            ("username",), username="root", sshpubkey="\n".join(authorized)
        )

    private_key = _ty.cast(str, keypair["attributes"]["private_key"])

    from hostctl.host import SshConfig

    # A local target has no host to SSH *to*, and needs none -- commands
    # already run here. The keypair is still provisioned (it is installed
    # on root's authorized_keys, so other machines can use it), but there
    # is no leg to wire it into.
    if self._config.is_local:
        return private_key

    existing = self._config.ssh
    if existing is None:
        self._config.ssh = SshConfig(
            host=self._config.host,
            username="root",
            client_keys=[private_key.encode()],
        )
    elif not existing.password and not existing.client_keys:
        existing.client_keys = [private_key.encode()]
    else:
        # Explicit credentials win: a caller who configured their own auth
        # is not silently overridden by a provisioning call.
        return private_key

    executors, paths = self._build_providers(self._config)
    from hostctl.provider import ProviderSelector

    self._executor_selector = ProviderSelector(executors)
    self._path_selector = ProviderSelector(paths)
    return private_key

login(creds=None, *, login_ex=False, login_options=None, otp_provider=None)

Open a fresh connection and authenticate.

By default uses the legacy auth.login/login_with_* path. Pass login_ex=True for the modern mechanism, which supports 2FA via an OTP_REQUIRED continuation: the OTP comes from the credential's own otp_token if set, else from otp_provider(). A credential with no login_ex form (e.g. local-socket auth) falls back automatically.

Source code in src/pytruenas/host.py

def login(
    self,
    creds: "_auth.Credentials | None" = None,
    *,
    login_ex: bool = False,
    login_options: "dict | None" = None,
    otp_provider: "_ty.Callable[[], str] | None" = None,
):
    """Open a fresh connection and authenticate.

    By default uses the legacy ``auth.login``/``login_with_*`` path. Pass
    ``login_ex=True`` for the modern mechanism, which supports 2FA via an
    ``OTP_REQUIRED`` continuation: the OTP comes from the credential's own
    ``otp_token`` if set, else from ``otp_provider()``. A credential with no
    login_ex form (e.g. local-socket auth) falls back automatically.
    """
    if self._conn and not self._conn._closed.is_set():
        try:
            self._conn.close()
        except Exception:
            pass
    self._conn = self._openwss()
    creds = creds or _ty.cast(_auth.Credentials, self._config.credentials)
    if login_ex:
        return creds.login_ex(
            _ty.cast(_ty.Any, self),
            login_options=login_options,
            otp_provider=otp_provider,
        )
    creds.login(_ty.cast(_ty.Any, self))

logout()

End the current session (auth.logout).

Source code in src/pytruenas/host.py

def logout(self) -> None:
    """End the current session (``auth.logout``)."""
    self.api.auth.logout()

me()

The current session's authenticated user (auth.me).

Source code in src/pytruenas/host.py

def me(self) -> dict:
    """The current session's authenticated user (``auth.me``)."""
    return _ty.cast(dict, self.api.auth.me())

ping()

Round-trip the middleware (core.ping -> "pong").

Source code in src/pytruenas/host.py

def ping(self) -> str:
    """Round-trip the middleware (``core.ping`` -> ``"pong"``)."""
    return _ty.cast(str, self.api.core.ping())

subscribe(event, callback=None, *, maxsize=_connection.DEFAULT_EVENT_QUEUE_SIZE)

Subscribe to a middleware event; return a Subscription.

host.subscribe("alert.list") is shorthand for host.api.alert.list.subscribe(). A subscription is bound to the current websocket and does not survive a reconnect -- the events() iterator ending is that signal.

Source code in src/pytruenas/host.py

def subscribe(
    self,
    event: str,
    callback: "_ty.Callable[..., object] | None" = None,
    *,
    maxsize: int = _connection.DEFAULT_EVENT_QUEUE_SIZE,
):
    """Subscribe to a middleware event; return a ``Subscription``.

    ``host.subscribe("alert.list")`` is shorthand for
    ``host.api.alert.list.subscribe()``. A subscription is bound to the
    current websocket and does **not** survive a reconnect -- the
    ``events()`` iterator ending is that signal.
    """
    return self.conn.subscribe(event, callback, maxsize=maxsize)

upload(file, method, *params, token=None, wait=True, **kwargs)

Upload file via /_upload, then call method with it.

Source code in src/pytruenas/host.py

def upload(
    self, file: "str | bytes", method: str, *params, token=None, wait=True, **kwargs
):
    """Upload ``file`` via ``/_upload``, then call ``method`` with it."""
    target = self._http_target("/_upload")
    data = {"method": method, "params": params}
    if isinstance(file, str):
        file = file.encode()

    if not token:
        token = self.api.auth.generate_token(5, {}, False, **kwargs)

    resp = _req.post(
        target.uri,
        headers={"Authorization": f"Token {token}"},
        verify=self._config.sslverify,
        files={"data": _js.dumps(data).encode(), "file": file},
    )
    jobid = resp.json()["job_id"]
    if wait:
        self.api.core.job_wait(jobid, job=True, _timeout=None)
    return jobid

Providers

pytruenas.providers

hostctl providers for a TrueNAS middleware target.

Only one adapter is TrueNAS-specific: :class:TnasWsPathProvider, paths served by the middleware filesystem.* API. It is deliberately thin -- it owns connection behaviour only, and leaves operating-system semantics to :class:hostctl.host.PosixHost.

Local execution needs no adapter at all: :func:local_providers returns hostctl's stock pair. A target reached over the middleware unix socket is this machine, so a command there is a plain subprocess call and a path is a plain local path -- there is nothing about TrueNAS to add.

On ordering (see TrueNASHost._build_providers):

local -> ssh -> webshell, and local -> sftp -> tnasws.

Local comes first because reaching this machine through SSH, a PTY, or the filesystem API would be slower and strictly less capable; it is only built when the target is local, and everything after it is a way of reaching a machine somewhere else. Among the remote options SSH wins on capability, which also reproduces :class:~pytruenas.fs.truenas.TruenasPath's hand-rolled "try SFTP, fall back to the websocket" behaviour -- now through hostctl's selector, which additionally records a redacted trace of what was tried and why (host.last_selection).

On honesty. probe() reports what a transport can actually do rather than what would be convenient. The JSON-RPC endpoint is not a general command channel: it exposes filesystem.* and friends, not arbitrary exec (verified on 26.0.0-BETA.1 -- of 781 methods only core.resize_shell and user.shell_choices are shell-adjacent, and the former merely resizes an already-open session). That limitation belongs to the JSON-RPC endpoint, not to middlewared as a whole: it also serves the PTY behind /websocket/shell, which :mod:pytruenas.webshell drives to give a remote host without SSH a real command channel.

WS_PATH_CAPABILITIES = frozenset(('stat', 'scandir', 'open', 'open_read', 'open_write', 'read', 'write', 'exists', 'is_file', 'is_dir', 'mkdir', 'chmod', 'unlink', 'rmdir')) module-attribute

__all__ = ['TnasWsPathProvider', 'WS_PATH_CAPABILITIES', 'local_providers'] module-attribute

TnasWsPathProvider(client)

Bases: PathProvider

Paths served by the middleware filesystem.* websocket API.

Always available for a connected client -- the middleware socket is the client's own connection, so there is no separate transport to fail. It is ordered after SFTP because its operation surface is narrower, not because it is less reliable.

Source code in src/pytruenas/providers.py

def __init__(self, client: "TrueNASClient") -> None:
    self.client = client
    super().__init__(
        "tnasws",
        self._make_path,
        capabilities=WS_PATH_CAPABILITIES,
    )

client = client instance-attribute

probe()

Source code in src/pytruenas/providers.py

def probe(self) -> _ProviderProbe:
    return _ProviderProbe("available", capabilities=self.capabilities)

local_providers()

hostctl's stock local executor and path providers.

A target reached over the middleware unix socket is this machine, so a command there is a plain subprocess call and a path is a plain local path. hostctl already provides both, and this is the same one-liner its own system.py:_local_provider uses -- there is nothing TrueNAS-specific to add, so pytruenas does not define a provider class for it.

(There was one. It wrapped LocalExecutor behind an is_local guard and called itself MiddlewareExecutorProvider, which was doubly misleading: nothing about the dispatch involved middlewared, and the guard only duplicated the decision the caller had already made by choosing to build it.)

Source code in src/pytruenas/providers.py

def local_providers():
    """hostctl's stock local executor and path providers.

    A target reached over the middleware unix socket *is* this machine, so a
    command there is a plain ``subprocess`` call and a path is a plain local
    path. hostctl already provides both, and this is the same one-liner its own
    ``system.py:_local_provider`` uses -- there is nothing TrueNAS-specific to
    add, so pytruenas does not define a provider class for it.

    (There was one. It wrapped ``LocalExecutor`` behind an ``is_local`` guard
    and called itself ``MiddlewareExecutorProvider``, which was doubly
    misleading: nothing about the dispatch involved ``middlewared``, and the
    guard only duplicated the decision the caller had already made by choosing
    to build it.)
    """
    from hostctl.executor import LocalExecutor
    from pathlib_next import Path as _LocalPath

    return (
        _ExecutorProvider("local", LocalExecutor()),
        _PathProvider("local", lambda *parts: _LocalPath(*parts)),
    )

Web shell

pytruenas.webshell

Command execution over the TrueNAS web-shell endpoint.

Reached at /websocket/shell (nginx), which proxies to /_shell on the middleware's own port -- see :data:WEBSHELL_PATH for why that distinction costs an afternoon if you get it wrong.

This is the executor for a host reachable on the API port but not on 22 -- NAT without a forwarded SSH port, a firewall allowing only 443, an appliance behind a reverse proxy. Such a host otherwise has no run() at all: the JSON-RPC API exposes no remote command execution (of 781 methods on 26.0.0-BETA.1 only core.resize_shell and user.shell_choices are shell-adjacent, and the former only resizes an already-open session).

middlewared serves /_shell as a separate websocket app beside the RPC socket -- a sibling of /_upload and /_download, which pytruenas already uses. It is what the web UI's Shell page drives, so a command channel demonstrably exists wherever the API does. nginx proxies it on 443 with a 7-day timeout, so it reaches exactly the hosts SSH cannot.

Protocol (verified against 26.0.0-BETA.1, 18/18 live):

1. Open a websocket to /websocket/shell.
2. Receive {"msg": "connected", "id": "<uuid>"}.
3. Send one JSON frame {"token": ..., "options": {}}. The token comes from auth.generate_token; the server validates it via auth.get_token_for_shell_application, which requires a token with no attributes and a user holding the web_shell privilege.
4. Every frame after that is raw PTY bytes, both directions.

Server-side it is os.forkpty() + os.execve("/usr/bin/login", ...), so this is a real login shell -- not a request/response API.

Input must be sent as BINARY frames. The handler queues msg.data verbatim and the writer thread calls os.write(master_fd, ...), which requires bytes. A text frame delivers a str, os.write raises, the worker thread dies without closing the pty, and the connection resets with no error message. This is the single least obvious thing about the endpoint.

Known limits, all deliberate and declared rather than papered over:

• stdout and stderr share one fd, because a PTY has no second channel. They are separated in band when the caller asks for it: the command is wrapped so its stderr passes through a process substitution that brackets it with OSC 1337 markers, and the reader routes the marked regions to stderr. The bracketing is per-line (see :meth:WebShellSession.wrap_stderr -- an earlier version bracketed the whole process substitution's lifetime instead, via a bare cat, which let interleaved stdout writes land inside an open bracket and get misattributed as stderr).

This needs 2> >(...), which only bash/zsh/ksh parse -- under sh it is a syntax error, the command never runs, and the call would TIME OUT rather than fail usefully. So it is gated on the login shell reported by auth.me() (see :meth:WebShellSession.supports_stderr_split) and simply stays merged when that is not positive. Merged output is always correct, just less informative.

A caller who asks for no separation (no stderr=, no capture_output="stderr") pays nothing: the wrapper is only applied when the difference is observable.

Output that is not captured is written through to stdout (defaulting to sys.stdout.buffer) incrementally, as frames arrive, rather than buffered until the command finishes. Those writes carry the raw PTY bytes, so colour and other escape sequences survive; the captured CompletedProcess.stdout value stays cleaned, since a caller parsing it wants the text rather than the terminal's rendering of it. capture_output/stdout resolution goes through hostctl.executor.capture_streams, the same helper the SSH executor uses, so the option surface matches whichever provider a host selects. * No exit-status channel -- the return code is recovered by appending a sentinel (printf "__END__%s\n" "$?") and reading until it appears. * No raw multi-line input -- an embedded newline submits a partial line to the PTY and desynchronises every later read. Pipes and here-strings work because they are ordinary single-line shell syntax, and a here-DOCUMENT is the one legal multi-line form (its newlines are the document's own, and the shell reads to the delimiter). input= uses exactly that. * Input works, in two shapes. input= is a value known in full up front and rides along as a here-document -- no timing, no second channel, and the preferred form. stdin= takes a readable object and PUMPS it on a background thread, for a stream the caller is still producing; it races the pty's echo of the command line and is mitigated, not cured, by a short delay (see :meth:WebShellSession._pump_stdin). A file DESCRIPTOR is rejected -- there is no pty fd to attach one to. * A command that exits the shell (exit 3) ends the session; the next call reconnects.

Because of those, this provider is ordered after SSH. It is a real executor for ordinary commands, not a degraded fallback -- but SSH's clean separate channels are better when available.

DEFAULT_TIMEOUT = 120.0 module-attribute

WEBSHELL_PATH = '/websocket/shell' module-attribute

__all__ = ['WebShellExecutorProvider', 'WebShellSession', 'clean_output'] module-attribute

WebShellExecutorProvider(client)

Bases: ExecutorProvider

Executor for hosts reachable on the API port but not over SSH.

Source code in src/pytruenas/webshell.py

def __init__(self, client: "TrueNASClient"):
    self.client = client
    self._session: "WebShellSession | None" = None
    # No `args`: a PTY takes one line of shell text, so hostctl should
    # render the whole invocation to a single string rather than an argv.
    super().__init__("webshell", self._execute, capabilities=())

client = client instance-attribute

session property

close()

Source code in src/pytruenas/webshell.py

def close(self) -> None:
    if self._session is not None:
        self._session.close()
        self._session = None

connect()

Source code in src/pytruenas/webshell.py

def connect(self) -> None:
    try:
        self.session.connect()
    except _OperationNotStarted:
        raise
    except Exception as exc:
        raise _OperationNotStarted(
            f"web shell unavailable: {type(exc).__name__}", cause=exc
        ) from exc

probe()

Report availability without dispatching a command.

A local target has the unix socket and does not need this. A user without the web_shell privilege would be rejected at the handshake, so decline up front rather than fail mid-command.

Source code in src/pytruenas/webshell.py

def probe(self) -> _ProviderProbe:
    """Report availability without dispatching a command.

    A local target has the unix socket and does not need this. A user
    without the ``web_shell`` privilege would be rejected at the handshake,
    so decline up front rather than fail mid-command.
    """
    config = getattr(self.client, "config", None)
    if config is not None and getattr(config, "is_local", False):
        return _ProviderProbe(
            "unavailable", reason="local target uses the middleware socket"
        )
    return _ProviderProbe("available", capabilities=self.capabilities)

WebShellSession(client, *, options=None)

One authenticated /_shell websocket running a login shell.

Source code in src/pytruenas/webshell.py

def __init__(self, client: "TrueNASClient", *, options: "dict | None" = None):
    self.client = client
    self.options = options or {}
    self._ws = None
    self._lock = _threading.RLock()
    self.shell_id: "str | None" = None
    #: Tri-state: None = not yet probed, then the cached answer. See
    #: `supports_stderr_split`.
    self._procsub: "bool | None" = None
    #: Guards websocket SENDS only. Deliberately not `_lock`: the stdin
    #: pump runs while `run_script` holds that one and blocks in recv, so
    #: sharing it would deadlock. See `_pump_stdin`.
    self._send_lock = _threading.Lock()

client = client instance-attribute

options = options or {} instance-attribute

shell_id = None instance-attribute

close()

Source code in src/pytruenas/webshell.py

def close(self) -> None:
    with self._lock:
        ws, self._ws = self._ws, None
        self.shell_id = None
        if ws is not None:
            try:
                ws.close()
            except Exception:
                pass

connect()

Open and authenticate the session; idempotent.

Source code in src/pytruenas/webshell.py

def connect(self):
    """Open and authenticate the session; idempotent."""
    with self._lock:
        if self._ws is not None:
            return self._ws
        try:
            import websocket as _websocket
        except ImportError as exc:  # pragma: no cover - dependency present
            raise ImportError("the web shell requires websocket-client") from exc

        import ssl

        sslopt = {} if self.client.sslverify else {"cert_reqs": ssl.CERT_NONE}
        ws = _websocket.WebSocket(sslopt=sslopt)
        ws.connect(self._uri())

        token = self.client.api.auth.generate_token(60, {}, False)
        ws.send(_json.dumps({"token": token, "options": self.options}))

        # Drain the connect frame and the login banner. A short read
        # timeout is the only way to know the prompt has settled -- the
        # server sends no "ready" marker.
        ws.settimeout(_SETTLE)
        while True:
            try:
                frame = ws.recv()
            except Exception:
                break
            if isinstance(frame, str):
                try:
                    message = _json.loads(frame)
                except ValueError:
                    continue
                if message.get("msg") == "connected":
                    self.shell_id = message.get("id")
                elif message.get("msg") == "failed":
                    ws.close()
                    raise _OperationNotStarted(
                        "web shell rejected the token: "
                        f"{message.get('error', {}).get('reason', 'unknown')}"
                    )
        self._ws = ws
        return ws

login_shell()

The login shell of the account this session authenticates as.

auth.me() reports the authenticated account's passwd entry, so pw_shell is the shell /usr/bin/login will exec -- exactly what the PTY ends up running. None when the API cannot answer.

Source code in src/pytruenas/webshell.py

def login_shell(self) -> "str | None":
    """The login shell of the account this session authenticates as.

    ``auth.me()`` reports the authenticated account's passwd entry, so
    ``pw_shell`` is the shell ``/usr/bin/login`` will exec -- exactly what
    the PTY ends up running. ``None`` when the API cannot answer.
    """
    try:
        return _ty.cast(
            "str | None", self.client.api.auth.me().get("pw_shell")
        )
    except Exception:
        return None

run_script(script, *, timeout=None, sink=None, errsink=None, heredoc=False, stdin=None)

Run one shell script; return cleaned output, raw bytes, and status.

script must be a single line -- an embedded newline submits a partial line to the PTY and desynchronises every later read.

sink, when given, receives the raw PTY bytes AS THEY ARRIVE rather than only at completion, so a long-running command reports progress instead of going silent until its sentinel appears. Raw is deliberate: the sink is what a caller sees on their terminal, and stripping the escape sequences there would discard exactly the colour they are watching for. The cleaned text is still what the return value carries.

Streaming has to hold back a suffix. The completion sentinel arrives split across frames like any other output, so the last len(end) bytes are never emitted until more arrive behind them -- otherwise half a sentinel reaches the terminal and the caller sees the marker this module exists to hide. The withheld tail is flushed on completion, minus the sentinel line itself.

Source code in src/pytruenas/webshell.py

def run_script(
    self,
    script: str,
    *,
    timeout: "float | None" = None,
    sink: "_ty.Callable[[bytes], None] | None" = None,
    errsink: "_ty.Callable[[bytes], None] | None" = None,
    heredoc: bool = False,
    stdin: object = None,
) -> "tuple[str, bytes, int]":
    """Run one shell script; return cleaned output, raw bytes, and status.

    ``script`` must be a single line -- an embedded newline submits a
    partial line to the PTY and desynchronises every later read.

    ``sink``, when given, receives the raw PTY bytes AS THEY ARRIVE rather
    than only at completion, so a long-running command reports progress
    instead of going silent until its sentinel appears. Raw is deliberate:
    the sink is what a caller sees on their terminal, and stripping the
    escape sequences there would discard exactly the colour they are
    watching for. The cleaned text is still what the return value carries.

    Streaming has to hold back a suffix. The completion sentinel arrives
    split across frames like any other output, so the last ``len(end)``
    bytes are never emitted until more arrive behind them -- otherwise half
    a sentinel reaches the terminal and the caller sees the marker this
    module exists to hide. The withheld tail is flushed on completion,
    minus the sentinel line itself.
    """
    # A here-document is the one legal multi-line form: its newlines are
    # the document's own separators and the shell keeps reading until the
    # delimiter, so the pty stays in sync. Anything else with an embedded
    # newline submits a partial line and desynchronises every later read.
    if not heredoc and ("\n" in script or "\r" in script):
        raise ValueError(
            "the web shell takes single-line commands; encode a multi-line "
            "payload as a here-string or a pipe"
        )
    timeout = DEFAULT_TIMEOUT if timeout is None else timeout
    marker = _uuid.uuid4().hex[:12]
    end = f"__EN{marker}__"
    pattern = _re.compile(_re.escape(end) + r"(\d+)")

    with self._lock:
        ws = self.connect()
        # BINARY: the server writes msg.data straight to the pty fd, which
        # rejects str and kills its writer thread. See the module docstring.
        #
        # The sentinel's separator depends on the shape of the command. A
        # here-document ends at a line containing ONLY its delimiter, so
        # appending `; printf ...` to that last line stops it terminating:
        # the shell keeps reading input forever and the command never runs.
        # A NEWLINE puts the sentinel on its own line, after the document,
        # where it is an ordinary next command. Everything else keeps `;`,
        # which is what makes the status apply to the command just run.
        separator = "\n" if heredoc else "; "
        ws.send_binary(
            f'{script}{separator}printf "{end}%s\\n" "$?"\n'.encode()
        )

        if stdin is not None:
            self._pump_stdin(ws, stdin)

        ws.settimeout(timeout)
        deadline = _time.monotonic() + timeout
        buffer = b""
        # How much of `buffer` the sink has already seen. Bytes are emitted
        # once and never re-sent, so this only ever moves forward.
        emitted = 0
        # Which stream the NEXT byte belongs to. This has to persist across
        # batches: a stderr burst can span two emits, and splitting each
        # chunk independently would put its tail back on stdout the moment
        # a batch boundary fell inside the region.
        in_err = [False]
        # Never emit a tail that could still be a partial marker -- of
        # EITHER kind. An OSC boundary split across two frames would
        # otherwise reach the terminal as escape-sequence garbage AND put
        # the following bytes on the wrong stream, so the longest marker
        # bounds how much is withheld, not just the sentinel.
        hold = (
            max(
                len(end.encode()) + 12,  # + room for the status digits
                len(_ERR_START),
                len(_ERR_END),
            )
        )
        flush_at = _time.monotonic() + _FLUSH_INTERVAL
        while _time.monotonic() < deadline:
            try:
                frame = ws.recv()
            except Exception as exc:
                self.close()
                raise _subprocess.SubprocessError(
                    f"web shell connection lost: {type(exc).__name__}"
                ) from exc
            buffer += frame if isinstance(frame, bytes) else frame.encode()
            text = clean_output(buffer)
            found = pattern.search(text)
            if sink is not None and not found:
                safe = len(buffer) - hold
                # Batch: a PTY delivers many tiny frames (often a frame per
                # keystroke echo), and one write+flush each turns a noisy
                # command into thousands of syscalls. Accumulate to
                # _BATCH before emitting; the deadline below bounds how
                # long a quiet partial batch can sit unflushed, so this
                # trades syscalls for latency without ever withholding
                # output indefinitely.
                if safe - emitted >= _BATCH or (
                    safe > emitted and _time.monotonic() >= flush_at
                ):
                    self._emit(buffer[emitted:safe], sink, errsink, in_err)
                    emitted = safe
                    flush_at = _time.monotonic() + _FLUSH_INTERVAL
            if found:
                if sink is not None:
                    self._emit(
                        self._tail(buffer[emitted:], end),
                        sink,
                        errsink,
                        in_err,
                    )
                return (
                    self._strip(text, script, end),
                    buffer,
                    int(found.group(1)),
                )

    self.close()
    raise _subprocess.TimeoutExpired(script, timeout)

split_streams(data) staticmethod

Split raw PTY bytes into (stdout, stderr) on the OSC markers.

Everything between :data:_ERR_START and :data:_ERR_END came from the command's stderr; everything else is stdout. The markers themselves are dropped from both -- they are this module's framing, not output.

Runs against the RAW stream. clean_output strips OSC sequences, so splitting the cleaned text would find no markers at all.

An unterminated final region (the command died mid-write, or the buffer was cut at a batch boundary) counts as stderr through the end: the marker said the stream switched, and nothing said it switched back.

Source code in src/pytruenas/webshell.py

@staticmethod
def split_streams(data: bytes) -> "tuple[bytes, bytes]":
    """Split raw PTY bytes into ``(stdout, stderr)`` on the OSC markers.

    Everything between :data:`_ERR_START` and :data:`_ERR_END` came from
    the command's stderr; everything else is stdout. The markers themselves
    are dropped from both -- they are this module's framing, not output.

    Runs against the RAW stream. `clean_output` strips OSC sequences, so
    splitting the cleaned text would find no markers at all.

    An unterminated final region (the command died mid-write, or the
    buffer was cut at a batch boundary) counts as stderr through the end:
    the marker said the stream switched, and nothing said it switched back.
    """
    out = bytearray()
    err = bytearray()
    rest = data
    while True:
        start = rest.find(_ERR_START)
        if start < 0:
            out += rest
            break
        out += rest[:start]
        rest = rest[start + len(_ERR_START) :]
        stop = rest.find(_ERR_END)
        if stop < 0:
            err += rest
            break
        err += rest[:stop]
        rest = rest[stop + len(_ERR_END) :]
    return bytes(out), bytes(err)

supports_stderr_split()

Whether the login shell can split stderr; resolved once, then cached.

The split wraps the command in 2> >(...) process substitution, which only bash/zsh/ksh parse. Under sh/dash it is a syntax error and the command never runs -- so the completion sentinel never arrives and the call TIMES OUT instead of failing usefully. Hence a positive answer is required before wrapping anything.

The answer comes from the API (auth.me()'s pw_shell), not from running a command. Asking the shell to identify itself would mean driving the very PTY this is meant to make safe: if the terminal is wedged, so is the probe, and the timeout it exists to prevent happens during the check. The API path also costs no PTY round-trip at all.

Falls back to False when the shell is unknown -- merged output is always correct, just less informative, so an unknown shell degrades rather than risking the syntax error.

Source code in src/pytruenas/webshell.py

def supports_stderr_split(self) -> bool:
    """Whether the login shell can split stderr; resolved once, then cached.

    The split wraps the command in ``2> >(...)`` process substitution,
    which only bash/zsh/ksh parse. Under ``sh``/``dash`` it is a syntax
    error and the command never runs -- so the completion sentinel never
    arrives and the call TIMES OUT instead of failing usefully. Hence a
    positive answer is required before wrapping anything.

    The answer comes from the **API** (``auth.me()``'s ``pw_shell``), not
    from running a command. Asking the shell to identify itself would mean
    driving the very PTY this is meant to make safe: if the terminal is
    wedged, so is the probe, and the timeout it exists to prevent happens
    during the check. The API path also costs no PTY round-trip at all.

    Falls back to ``False`` when the shell is unknown -- merged output is
    always correct, just less informative, so an unknown shell degrades
    rather than risking the syntax error.
    """
    if self._procsub is None:
        shell = self.login_shell()
        self._procsub = bool(shell) and any(
            # Match the basename: `/usr/bin/zsh` -> `zsh`. A substring test
            # over the whole path would let `/usr/bin/nozsh` through, and
            # would miss nothing a basename match catches.
            shell.rsplit("/", 1)[-1] == name  # type: ignore[union-attr]
            for name in _PROCSUB_SHELLS
        )
    return self._procsub

wrap_stderr(script) staticmethod

Wrap script so its stderr arrives fenced in OSC markers.

stderr is redirected into a process substitution that brackets it with :data:_ERR_START/:data:_ERR_END and forwards it to the shared PTY. Both streams stay on one fd -- a PTY has no second channel -- but the markers say which bytes were which, so the reader can separate them.

A per-LINE read loop, not a single cat. cat looks like the obviously-correct choice -- it forwards bytes unchanged -- but ``2>

(...)gives the substitution its own subshell with its own scheduling:printf start; cat; printf endemitsstartonce, when the subshell is FIRST scheduled (which can be before the wrapped command has written anything), andendonce, only when its input pipe hits EOF (which is when the WHOLE wrapped command finishes, not after any one stderr write). Every stdout byte written in between -- byscriptitself, on the pty's other fd -- lands inside that one open bracket and reads back as stderr. Verified against real bash:{ echo out1; echo err1 >&2; echo out2; echo err2 >&2; }wrapped with thecatform puts BOTHout1andout2inside the bracket alongside the twoerr`` lines, nondeterministically depending on subshell scheduling.

Bracketing per read (here, per line -- the granularity read offers) closes each region before the next stdout write can land inside it. The trade made for this: a stderr write with no trailing newline still needs a line boundary to be forwarded at all, hence the || [ -n "$line" ] to flush the final partial line, and every forwarded line gains a synthetic trailing \n (harmless -- clean_output already collapses blank-line runs). read -r and printf %s (not %b) keep the payload itself byte-for-byte: backslashes and % in the command's actual stderr text are not reinterpreted, only the marker halves of the format string are.

A byte-exact (NUL-safe) alternative was tried and rejected: bracket each raw OS-level read via dd bs=N count=1 into a scratch file (mktemp), instead of each shell line. It IS byte-exact where this line loop is not -- but every burst costs 4 forked processes (mktemp, dd, wc, cat), and that startup cost loses the race against a short-lived command: 2> >(...) does not wait for the substitution's subshell to actually be scheduled before the wrapped command can run and exit, so a fast printf ... >&2 can close its write end before dd's first read ever happens. Verified against real bash: a trivial printf "abc\n" >&2 lost the stderr content outright in 2 of 6 trials with the dd version, and 0 of 6 with this line loop. A shell builtin with no per-burst fork beats a byte-exact external pipeline that cannot reliably start in time.

Source code in src/pytruenas/webshell.py

@staticmethod
def wrap_stderr(script: str) -> str:
    """Wrap ``script`` so its stderr arrives fenced in OSC markers.

    stderr is redirected into a process substitution that brackets it with
    :data:`_ERR_START`/:data:`_ERR_END` and forwards it to the shared PTY.
    Both streams stay on one fd -- a PTY has no second channel -- but the
    markers say which bytes were which, so the reader can separate them.

    A per-LINE ``read`` loop, not a single ``cat``. ``cat`` looks like the
    obviously-correct choice -- it forwards bytes unchanged -- but ``2>
    >(...)`` gives the substitution its own subshell with its own
    scheduling: ``printf start; cat; printf end`` emits ``start`` once,
    when the subshell is FIRST scheduled (which can be before the wrapped
    command has written anything), and ``end`` once, only when its input
    pipe hits EOF (which is when the WHOLE wrapped command finishes, not
    after any one stderr write). Every stdout byte written in between --
    by ``script`` itself, on the pty's other fd -- lands inside that one
    open bracket and reads back as stderr. Verified against real bash:
    ``{ echo out1; echo err1 >&2; echo out2; echo err2 >&2; }`` wrapped
    with the ``cat`` form puts BOTH ``out1`` and ``out2`` inside the
    bracket alongside the two ``err`` lines, nondeterministically
    depending on subshell scheduling.

    Bracketing per read (here, per line -- the granularity ``read``
    offers) closes each region before the next stdout write can land
    inside it. The trade made for this: a stderr write with no trailing
    newline still needs a line boundary to be forwarded at all, hence the
    ``|| [ -n "$line" ]`` to flush the final partial line, and every
    forwarded line gains a synthetic trailing ``\\n`` (harmless --
    `clean_output` already collapses blank-line runs). ``read -r`` and
    ``printf %s`` (not ``%b``) keep the payload itself byte-for-byte:
    backslashes and ``%`` in the command's actual stderr text are not
    reinterpreted, only the marker halves of the format string are.

    A byte-exact (NUL-safe) alternative was tried and rejected: bracket
    each raw OS-level read via ``dd bs=N count=1`` into a scratch file
    (``mktemp``), instead of each shell line. It IS byte-exact where this
    line loop is not -- but every burst costs 4 forked processes
    (``mktemp``, ``dd``, ``wc``, ``cat``), and that startup cost loses the
    race against a short-lived command: ``2> >(...)`` does not wait for
    the substitution's subshell to actually be scheduled before the
    wrapped command can run and exit, so a fast `printf ... >&2` can close
    its write end before ``dd``'s first read ever happens. Verified
    against real bash: a trivial ``printf "abc\\n" >&2`` lost the stderr
    content outright in 2 of 6 trials with the ``dd`` version, and 0 of 6
    with this line loop. A shell builtin with no per-burst fork beats a
    byte-exact external pipeline that cannot reliably start in time.
    """
    # The markers hold ESC and BEL, which cannot appear literally in a
    # single-line command, so they are written as printf %b escapes.
    start = "\\033]%d;PytruenasStderr\\007" % _OSC
    end = "\\033]%d;PytruenasStdout\\007" % _OSC
    # One marker pair per LINE read from stderr, so a bracket never spans
    # past the stderr write it belongs to and cannot swallow a stdout
    # write that happens to land in between.
    forward = (
        'while IFS= read -r line || [ -n "$line" ]; do '
        f'printf "%b%s\\n%b" "{start}" "$line" "{end}"; '
        "done"
    )
    return f"{{ {script} ; }} 2> >({forward})"

clean_output(data)

Strip escape sequences, prompts, and CRs from raw PTY bytes.

What comes off a PTY is a terminal rendering, not a clean stream: cursor moves, line redraws, and the shell's own prompt are interleaved with the command's output. This removes the rendering so a caller sees what the command actually printed.

Source code in src/pytruenas/webshell.py

def clean_output(data: bytes) -> str:
    """Strip escape sequences, prompts, and CRs from raw PTY bytes.

    What comes off a PTY is a *terminal rendering*, not a clean stream: cursor
    moves, line redraws, and the shell's own prompt are interleaved with the
    command's output. This removes the rendering so a caller sees what the
    command actually printed.
    """
    text = _ANSI.sub(b"", data).replace(b"\r", b"").decode("utf-8", "replace")
    text = _PROMPT.sub("", text)
    # Collapse the blank runs the prompt removal leaves behind.
    return _re.sub(r"\n{2,}", "\n", text)