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.

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.

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.
    """
    name = name or "pytruenas"
    keypair = self.api.keychaincredential._get(type="SSH_KEY_PAIR", name=name)
    if not keypair and not private_key:
        private_key = self.api.keychaincredential.generate_ssh_key_pair()[
            "private_key"
        ]
    elif not private_key:
        private_key = keypair["attributes"]["private_key"]

    pubkey = (
        _asyncssh()
        .import_private_key(private_key)
        .export_public_key()
        .decode()
        .strip()
    )
    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)
        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 are one stream -- a PTY has no separate error channel, so capture_output="stderr" cannot be honoured.
• 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.
• 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

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

run_script(script, *, timeout=None)

Run one shell script; return its cleaned output and exit status.

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

Source code in src/pytruenas/webshell.py

def run_script(
    self, script: str, *, timeout: "float | None" = None
) -> "tuple[str, int]":
    """Run one shell script; return its cleaned output and exit status.

    ``script`` must be a single line -- an embedded newline submits a
    partial line to the PTY and desynchronises every later read.
    """
    if "\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.
        ws.send_binary(f'{script}; printf "{end}%s\\n" "$?"\n'.encode())

        ws.settimeout(timeout)
        deadline = _time.monotonic() + timeout
        buffer = b""
        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 found:
                return self._strip(text, script, end), int(found.group(1))

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

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)