Host

TrueNASClient is a hostctl PosixHost: run, path, spawn, info, close, shell, capabilities, and last_selection are inherited, and it adds the TrueNAS surface (api, conn, connect (the API connection only), login, 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 a target leaves out are defaulted, not probed: a remote target with neither resolves to wss://<host>/api/<version>, and only an explicit ws/http selects plaintext. :attr:~TrueNASConfig.needs_scheme_probe and :attr:~TrueNASConfig.needs_path_probe record that the caller did not say, which is what makes the default visible to a caller that wants to ask the server itself; nothing in this package performs such a request.

AUTO_SCHEME = 'truenas+auto' module-attribute

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

DEFAULT_SOCKET_PATH = DEFAULT_UNIX_SOCKET module-attribute

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

PendingDownload(host, job_id, uri, buffered)

A download the middleware prepared: its job_id, and the bytes on fetch().

TrueNASClient.download(..., wait=False) returns one. The link is single-use and carries a token, so it is not shown in the repr.

Source code in src/pytruenas/host.py

def __init__(self, host: "TrueNASClient", job_id: int, uri: str, buffered: bool):
    self.job_id = job_id
    self._host = host
    self._uri = uri
    self._buffered = buffered

job_id = job_id instance-attribute

__repr__()

Source code in src/pytruenas/host.py

def __repr__(self) -> str:
    return f"PendingDownload(job_id={self.job_id!r})"

fetch(callback=None)

The downloaded bytes, once the job that produced them has finished.

A failed job raises :class:~pytruenas.connection.JobFailed. A buffered download's job is waited for before the fetch; an unbuffered one streams into the response and is checked after it.

Source code in src/pytruenas/host.py

def fetch(
    self, callback: "_ty.Optional[_ty.Callable[..., object]]" = None
) -> bytes:
    """The downloaded bytes, once the job that produced them has finished.

    A failed job raises :class:`~pytruenas.connection.JobFailed`. A
    buffered download's job is waited for before the fetch; an unbuffered
    one streams into the response and is checked after it.
    """
    if self._buffered:
        self._host.wait(self.job_id, callback=callback)
    content = self._host._side_request("GET", self._uri).content
    if not self._buffered:
        self._host.wait(self.job_id, callback=callback)
    return content

TrueNASClient(config=None, **options)

TrueNASClient(
    config: "_ty.Optional[str]" = None,
    *,
    credentials: object = ...,
    username: "_ty.Optional[str]" = ...,
    password: "_ty.Optional[str]" = ...,
    otp: "_ty.Optional[str]" = ...,
    api_key: "_ty.Optional[str]" = ...,
    token: "_ty.Optional[str]" = ...,
    verify: "_ty.Union[bool, str, _os.PathLike[str], _ssl.SSLContext]" = ...,
    ssh: "_ty.Optional[SSHConfig]" = ...,
    version: str = ...,
    shell: "_ty.Optional[str]" = ...,
    known_hosts: object = ...,
    executor: "_ty.Optional[_ty.Union[_ty.Iterable[str], str]]" = ...,
    path: "_ty.Optional[_ty.Union[_ty.Iterable[str], str]]" = ...,
    autologin: bool = ...,
    timeout: "_ty.Union[float, None, Timeout]" = ...,
    executor_providers: "_ty.Optional[_ty.Iterable[ExecutorProvider]]" = ...,
    path_providers: "_ty.Optional[_ty.Iterable[PathProvider]]" = ...,
    info: "_ty.Optional[HostInfo]" = ...,
    initializer: object = ...
)

TrueNASClient(
    config: TrueNASConfig,
    *,
    executor_providers: "_ty.Optional[_ty.Iterable[ExecutorProvider]]" = ...,
    path_providers: "_ty.Optional[_ty.Iterable[PathProvider]]" = ...,
    shell: object = ...,
    info: "_ty.Optional[HostInfo]" = ...,
    initializer: object = ...
)

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)::

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

A string is normalized by :func:_normalize_target, and the keyword options are those of :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 through hostctl's selector, which records a redacted trace of what was tried (host.last_selection).

Source code in src/pytruenas/host.py

def __init__(
    self,
    config: "_ty.Optional[_ty.Union[TrueNASConfig, str]]" = None,
    **options: object,
) -> None:
    # 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: "_ty.Optional[_connection.TrueNASWSConnection]" = None
    # hostctl logs a provider fallback here too, once per skipped provider.
    _PosixHost.logger.fset(self, _resolve_logger(None, config.name))
    #: Guards opening, replacing and dropping `_conn`. Reentrant: logging
    #: in calls `auth.*` through `api`, which reads `conn` again.
    self._conn_lock = _threading.RLock()
    #: How the session last authenticated -- `login()`'s arguments -- so a
    #: reconnect repeats *that* login, not whatever the config would do.
    self._last_login: "_ty.Optional[tuple[object, dict]]" = None
    #: Closes the websocket if this host is collected without close().
    self._conn_finalizer: "_ty.Optional[_weakref.finalize]" = None
    #: One warning per host about credentials over plaintext ws://.
    self._plaintext_warned = False
    #: One warning per host when TLS verification is off.
    self._unverified_warned = False
    #: HTTP(S) session for the side channels; see `_http`.
    self._http_session: "_ty.Optional[_req.Session]" = None
    #: `{service: {method: is a job}}`, read once per service; see `_is_job`.
    self._job_flags: "dict[str, dict[str, bool]]" = {}

    super().__init__(config, **_ty.cast(_ty.Any, options))

JOB_DONE = ('SUCCESS', 'FAILED', 'ABORTED') class-attribute instance-attribute

api cached property

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

Parameterise the client to type it: TrueNASClient[Current]("nas").api completes the generated API stubs.

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; after a successful :meth:login, a reconnect repeats that login -- the same credentials and mechanism -- whatever autologin says.

Thread-safe: concurrent first use opens one connection, not one per thread.

logger property writable

The host's logger: every record it emits starts with [name].

Assign a logger name or a Logger to send the records elsewhere; it is bound to :attr:name the same way. A LoggerAdapter is used as it is. The JSON-RPC connection takes the logger set when it opens.

provider_resolvers = {'ssh': _ssh_pair, 'sftp': _ssh_pair, 'webshell': _webshell, 'tnasws': _tnasws, 'winrm': None} class-attribute instance-attribute

ssh property

The asyncssh connection run() uses (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.

It is :attr:hostctl.transports.ssh.SSHExecutorProvider.connection: the first access connects (raising PermissionError, ConnectionError, TimeoutError or OSError), and the connection is bound to that provider's loop, hostctl's process-wide event-loop thread -- run its coroutines with asyncio.run_coroutine_threadsafe(coro, provider.loop).result(). :meth:close closes it. Raises RuntimeError when this host has no SSH executor.

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.
    """
    finalizer, self._conn_finalizer = self._conn_finalizer, None
    if finalizer is not None:
        # Closing explicitly; nothing left for the collector to do.
        finalizer.detach()
    session, self._http_session = self._http_session, None
    if session is not None:
        session.close()
    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

connect()

Open the JSON-RPC connection, logging in when autologin is set.

This is what with TrueNASClient(...) does on entry. The command and path transports (SSH, the web shell, SFTP) connect when first used, so an API-only script opens no shell session and dials no SSH port. Raises what opening :attr:conn raises.

Source code in src/pytruenas/host.py

def connect(self) -> None:
    """Open the JSON-RPC connection, logging in when ``autologin`` is set.

    This is what ``with TrueNASClient(...)`` does on entry. The command and
    path transports (SSH, the web shell, SFTP) connect when first used, so
    an API-only script opens no shell session and dials no SSH port.
    Raises what opening :attr:`conn` raises.
    """
    self.conn

default_providers()

The provider names for this target when executor=/path= name none.

A local target is served entirely by hostctl's stock local pair: plain subprocess and plain local paths. Reaching this same machine through SSH, a PTY, or the filesystem API would be slower and strictly less capable, so no remote provider is offered.

Remotely, SSH leads on capability (separate streams, real stdin, a rich POSIX path surface) and the websocket legs follow: the web shell as a command channel for a host with no reachable SSH, and tnasws for paths. hostctl drops a default ssh/sftp when its ssh extra is missing; a name the caller gives is kept and fails at use.

Source code in src/pytruenas/host.py

def default_providers(self):
    """The provider names for this target when ``executor=``/``path=`` name none.

    A local target is served entirely by hostctl's stock local pair: plain
    subprocess and plain local paths. Reaching this same machine through
    SSH, a PTY, or the filesystem API would be slower and strictly less
    capable, so no remote provider is offered.

    Remotely, SSH leads on capability (separate streams, real stdin, a rich
    POSIX path surface) and the websocket legs follow: the web shell as a
    command channel for a host with no reachable SSH, and ``tnasws`` for
    paths. hostctl drops a default ``ssh``/``sftp`` when its ``ssh`` extra
    is missing; a name the caller gives is kept and fails at use.
    """
    config = self._config
    if config.is_local:
        return ("local",), ("local",)
    if config.ssh is not None:
        return ("ssh", "webshell"), ("sftp", "tnasws")
    return ("webshell",), ("tnasws",)

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

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

With wait (the default) returns the content, once the job that produced it has finished: a failed job raises :class:~pytruenas.connection.JobFailed rather than returning what was sent before it failed. buffered=True waits before fetching; otherwise the job streams into the response and is checked after it. wait=False returns a :class:PendingDownload instead of the bytes: its job_id, and fetch() when they are wanted.

Source code in src/pytruenas/host.py

def download(
    self,
    method: str,
    *args,
    filename: "_ty.Optional[str]" = None,
    buffered=False,
    wait=True,
):
    """Call ``method`` for a download link and fetch it over HTTP(S).

    With ``wait`` (the default) returns the content, once the job that
    produced it has finished: a failed job raises
    :class:`~pytruenas.connection.JobFailed` rather than returning what
    was sent before it failed. ``buffered=True`` waits before fetching;
    otherwise the job streams into the response and is checked after it.
    ``wait=False`` returns a :class:`PendingDownload` instead of the bytes:
    its ``job_id``, and ``fetch()`` when they are wanted.
    """
    jobid, link = self.api.core.download(
        method, args, filename or "download", buffered
    )
    pending = PendingDownload(self, jobid, self._http_target(link).uri, buffered)
    if wait:
        return pending.fetch(callback=wait if callable(wait) else None)
    return pending

dump_api(*, cache=True, refresh=False, all_versions=False)

Return the middleware API definition (middlewared --dump-api).

The current API version only, pruned on the target before the transfer; all_versions=True keeps every version the appliance serves (current first), which is about five times the bytes on 26.0.

The dump is built and compressed on the target, then fetched over SFTP when this host has an SSH leg and in verified chunks otherwise. Its raw output cannot come back through run(): the dump is 24,103,000 bytes on 26.0, and the web shell's PTY loses the connection on that much output, on the very hosts (no SSH) that need it.

cache=True (the default) reads and writes the per-host, per-version cache under $PYTRUENAS_CACHE; building the dump costs about 75 seconds, which is not a price to pay twice. refresh=True re-fetches and overwrites it. See :mod:pytruenas.apicache.

Source code in src/pytruenas/host.py

def dump_api(
    self, *, cache: bool = True, refresh: bool = False, all_versions: bool = False
):
    """Return the middleware API definition (``middlewared --dump-api``).

    The current API version only, pruned on the target before the
    transfer; ``all_versions=True`` keeps every version the appliance
    serves (current first), which is about five times the bytes on 26.0.

    The dump is built and **compressed on the target**, then fetched over
    SFTP when this host has an SSH leg and in verified chunks otherwise.
    Its raw output cannot come back through ``run()``: the dump is
    24,103,000 bytes on 26.0, and the web shell's PTY loses the connection
    on that much output, on the very hosts (no SSH) that need it.

    ``cache=True`` (the default) reads and writes the per-host, per-version
    cache under ``$PYTRUENAS_CACHE``; building the dump costs about 75
    seconds, which is not a price to pay twice. ``refresh=True`` re-fetches
    and overwrites it. See :mod:`pytruenas.apicache`.
    """
    from . import apicache as _apicache

    if not cache:
        return _apicache.fetch(self, all_versions=all_versions)
    return _apicache.load(self, refresh=refresh, all_versions=all_versions)

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.SSHConfig, as its client_keys; an ssh= the caller passed is copied, not changed.

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: "_ty.Optional[str]" = None, private_key: "_ty.Optional[str]" = 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.SSHConfig`, as its ``client_keys``; an ``ssh=`` the
    caller passed is copied, not changed.

    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: "_ty.Optional[str]" = 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 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()],
            verify=self._config.verify is not False,
            known_hosts=_ty.cast(_ty.Any, self._config.known_hosts),
        )
    elif not existing.password and not existing.client_keys:
        # A copy: the caller's `ssh=` object may serve other hosts too.
        self._config.ssh = _dataclasses.replace(
            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

    # One resolution and one swap: the ssh and sftp legs share a
    # transport, and hostctl closes each replaced one (an open SSH
    # connection among them) once.
    executors, paths = self.resolve_providers()
    self.replace_providers(executor=executors, path=paths)
    return private_key

job_updates(job_id, *, timeout=None, poll=0.25, max_poll=2.0)

Yield job job_id's record each time its state or progress changes.

The record is core.get_jobs's: state (WAITING/RUNNING, then one of :attr:JOB_DONE), progress (percent, description), result, error... The first record is always yielded and the finished one is always the last, so a loop over this can drive a log line or a progress bar and then inspect the outcome::

for job in client.job_updates(job_id):
    print(job["progress"]["percent"], job["progress"]["description"])

Polls with a back-off from poll to max_poll seconds, reset whenever something changes. timeout=None waits indefinitely; otherwise :class:~pytruenas.connection.CallTimeout is raised (the job keeps running on the server). Failure is not raised here -- the final record says so; :meth:wait raises it.

Source code in src/pytruenas/host.py

def job_updates(
    self,
    job_id: int,
    *,
    timeout: "_ty.Optional[float]" = None,
    poll: float = 0.25,
    max_poll: float = 2.0,
) -> "_ty.Iterator[dict]":
    """Yield job ``job_id``'s record each time its state or progress changes.

    The record is ``core.get_jobs``'s: ``state`` (``WAITING``/``RUNNING``,
    then one of :attr:`JOB_DONE`), ``progress`` (``percent``,
    ``description``), ``result``, ``error``... The first record is always
    yielded and the finished one is always the last, so a loop over this
    can drive a log line or a progress bar and then inspect the outcome::

        for job in client.job_updates(job_id):
            print(job["progress"]["percent"], job["progress"]["description"])

    Polls with a back-off from ``poll`` to ``max_poll`` seconds, reset
    whenever something changes. ``timeout=None`` waits indefinitely;
    otherwise :class:`~pytruenas.connection.CallTimeout` is raised (the job
    keeps running on the server). Failure is not raised here -- the final
    record says so; :meth:`wait` raises it.
    """
    import time as _time

    deadline = None if timeout is None else _time.monotonic() + timeout
    backoff = _Backoff(poll, max_delay=max_poll)
    last = None
    while True:
        found = self.api.core.get_jobs([["id", "=", job_id]])
        if not found:
            raise _connection.ClientException(f"no such job: {job_id}")
        job = found[0]
        seen = (job.get("state"), _js.dumps(job.get("progress"), default=str))
        if seen != last:
            last = seen
            backoff.reset()
            yield job
        if job.get("state") in self.JOB_DONE:
            return
        if deadline is not None and _time.monotonic() + backoff.delay > deadline:
            raise _connection.CallTimeout()
        _time.sleep(backoff.delay)
        backoff.advance()

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 if set, else from otp_provider(). A credential with no login_ex form (e.g. local-socket auth) falls back automatically.

Refused credentials raise :class:~pytruenas.auth.AuthenticationError on both paths, and the connection is closed rather than left open and unauthenticated. After a success, a reconnect repeats this login without the credential's one-time code, which is spent: on the login_ex path it asks otp_provider for a new one, and raises AuthenticationError("OTP_REQUIRED") without one.

Source code in src/pytruenas/host.py

def login(
    self,
    creds: "_ty.Optional[_auth.Credentials]" = None,
    *,
    login_ex: bool = False,
    login_options: "_ty.Optional[dict]" = None,
    otp_provider: "_ty.Optional[_ty.Callable[[], str]]" = 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`` if set, else from ``otp_provider()``. A credential with no
    login_ex form (e.g. local-socket auth) falls back automatically.

    Refused credentials raise :class:`~pytruenas.auth.AuthenticationError`
    on both paths, and the connection is closed rather than left open and
    unauthenticated. After a success, a reconnect repeats this login
    without the credential's one-time code, which is spent: on the
    ``login_ex`` path it asks ``otp_provider`` for a new one, and raises
    ``AuthenticationError("OTP_REQUIRED")`` without one.
    """
    options = {
        "login_ex": login_ex,
        "login_options": login_options,
        "otp_provider": otp_provider,
    }
    with self._conn_lock:
        if self._conn and not self._conn._closed.is_set():
            try:
                self._conn.close()
            except Exception:
                pass
        self._track(self._openwss())
        creds = creds or _ty.cast(_auth.Credentials, self._config.credentials)
        self._warn_if_plaintext(creds)
        try:
            if login_ex:
                result = creds.login_ex(
                    _ty.cast(_ty.Any, self),
                    login_options=login_options,
                    otp_provider=otp_provider,
                )
            else:
                result = creds.login(_ty.cast(_ty.Any, self))
        except BaseException:
            conn, self._conn = self._conn, None
            conn.close()
            raise
        self._last_login = (_without_otp(creds), options)
        return result

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())

path(*segments, backend=None)

A path on the target -- :meth:hostctl.host.Host.path, rooted at /.

On a remote target a relative path (and path() with no segments) names the same file on every leg: / joined with it. hostctl's SFTP leg reads a relative path against the SSH login directory, which the websocket leg cannot know, so left relative the same path would name different files depending on which leg serves it. A local target keeps local semantics (the process's working directory).

backend pins the path to that provider for every operation on it.

Source code in src/pytruenas/host.py

def path(self, *segments, backend=None):
    """A path on the target -- :meth:`hostctl.host.Host.path`, rooted at ``/``.

    On a remote target a relative path (and ``path()`` with no segments)
    names the same file on every leg: ``/`` joined with it. hostctl's SFTP
    leg reads a relative path against the SSH login directory, which the
    websocket leg cannot know, so left relative the same path would name
    different files depending on which leg serves it. A local target keeps
    local semantics (the process's working directory).

    ``backend`` pins the path to that provider for every operation on it.
    """
    if not self._config.is_local and (
        not segments or not str(segments[0]).startswith("/")
    ):
        segments = ("/", *segments)
    return super().path(*segments, backend=backend)

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())

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

Upload file via /_upload, then call method(*params) with it.

Source code in src/pytruenas/host.py

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

    if not token:
        # See the web shell: the server defaults match_origin to True and
        # this token is used once, by one HTTP request from this machine.
        token = self.api.auth.generate_token(5, {}, True, True)

    # An upload the middleware refuses answers with an HTTP error and an
    # HTML/JSON body; without the status check that surfaced as a KeyError
    # or a JSON decode error, naming neither the status nor the reason.
    resp = self._side_request(
        "POST",
        target.uri,
        headers={"Authorization": f"Token {token}"},
        files={"data": _js.dumps(data).encode(), "file": file},
    )
    try:
        jobid = resp.json()["job_id"]
    except (ValueError, KeyError, TypeError):
        raise _connection.ClientException(
            f"upload to {target.uri} returned no job id "
            f"(HTTP {resp.status_code}): {resp.text[:200]}"
        ) from None
    if wait:
        self.wait(jobid, callback=wait if callable(wait) else None)
    return jobid

wait(job_id, *, callback=None, timeout=None)

Block until middleware job job_id finishes; return its result.

callback(job) is called with the job record every time its state or progress changes (see :meth:job_updates) -- enough for a log line or a progress bar. SUCCESS returns the job's result; FAILED/ABORTED raise :class:~pytruenas.connection.JobFailed carrying the record.

Not core.job_wait: that method is itself a job, so calling it returns a NEW job id immediately instead of blocking (measured on 26.0.0-BETA.1: 0.4 s) -- every "wait" built on it returned early.

Source code in src/pytruenas/host.py

def wait(
    self,
    job_id: int,
    *,
    callback: "_ty.Optional[_ty.Callable[[dict], object]]" = None,
    timeout: "_ty.Optional[float]" = None,
) -> object:
    """Block until middleware job ``job_id`` finishes; return its result.

    ``callback(job)`` is called with the job record every time its state or
    progress changes (see :meth:`job_updates`) -- enough for a log line or
    a progress bar. ``SUCCESS`` returns the job's ``result``;
    ``FAILED``/``ABORTED`` raise :class:`~pytruenas.connection.JobFailed`
    carrying the record.

    Not ``core.job_wait``: that method is itself a job, so calling it
    returns a NEW job id immediately instead of blocking (measured on
    26.0.0-BETA.1: 0.4 s) -- every "wait" built on it returned early.
    """
    job: dict = {}
    for job in self.job_updates(job_id, timeout=timeout):
        if callback is not None:
            callback(job)
    if job.get("state") != "SUCCESS":
        raise _connection.JobFailed(job)
    return job.get("result")

TrueNASConfig(host='', *, port=0, secure=None, socket_path=None, api_path=None, version='current', verify=True, credentials=None, ssh=None, shell=None, known_hosts=(), executor=None, path=None, autologin=True, timeout=Timeout())

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: "_ty.Optional[bool]" = None,
    socket_path: "_ty.Optional[str]" = None,
    api_path: "_ty.Optional[str]" = None,
    version: str = "current",
    verify: "_ty.Union[bool, str, _os.PathLike[str], _ssl.SSLContext]" = True,
    credentials: object = None,
    ssh: object = None,
    shell: "_ty.Optional[str]" = None,
    known_hosts: object = (),
    executor: "_ty.Optional[_ty.Union[_ty.Iterable[str], str]]" = None,
    path: "_ty.Optional[_ty.Union[_ty.Iterable[str], str]]" = None,
    autologin: bool = True,
    timeout: "_ty.Union[float, None, Timeout]" = Timeout(),
) -> None:
    super().__init__()
    #: A label assigned to `name`, replacing the derived one.
    self._name: "_ty.Optional[str]" = None
    #: Whether the first websocket access logs in automatically.
    self.autologin = autologin
    #: hostctl's `Timeout`: `connect` opens the websocket, `read` bounds one
    #: middleware call (and a side-channel read), `total` is `run()`'s
    #: default. `write` is not supported. See :attr:`timeouts`.
    self.timeout = _timeout(timeout)
    #: 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:`TrueNASClient.default_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)
    if not (host or port or socket_path or api_path) and secure is None:
        # Nothing names a remote host, so this is the local middleware.
        socket_path = DEFAULT_SOCKET_PATH
    self.host = host
    self.port = int(port or 0)
    #: ``True`` -> wss, ``False`` -> ws, ``None`` -> not given, use wss.
    self.secure = secure
    self.socket_path = socket_path
    self.api_path = api_path
    self.version = version
    #: One switch for every check the client makes. ``False`` turns off TLS
    #: certificate verification (the API websocket, the HTTP side channels,
    #: the web shell) and SSH host-key verification (commands and SFTP).
    #: ``True`` trusts the operating system; a CA bundle path (file or hashed
    #: directory) is trusted instead of it; an ``ssl.SSLContext`` is used as
    #: it is (client certificates, pinning). For SSH, a path or a context
    #: means "verify". With ``True``, ``$SSL_CERT_FILE`` and the other
    #: CA-bundle variables are consulted first; certifi is never used.
    self.verify: "_ty.Union[bool, str, _ssl.SSLContext]" = _tls.check(verify)
    #: The SSH leg. Accepts a ready :class:`hostctl.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.
    #:
    #: ``known_hosts`` names the trusted host keys for an SSH leg built
    #: here -- from ``shell=`` or by :meth:`TrueNASClient.install_sshcreds`
    #: -- with hostctl's values: ``()`` is the ssh_config's
    #: ``UserKnownHostsFile``, else ``~/.ssh/known_hosts``; a path or list
    #: names the file(s). ``verify=False`` skips the ``()`` lookup;
    #: named keys are checked either way. An explicit ``ssh=SSHConfig(...)``
    #: keeps its ``known_hosts`` and gets ``verify=False`` on a copy.
    self.known_hosts = known_hosts
    if ssh is not None and verify is False and getattr(ssh, "verify", False):
        ssh = _dataclasses.replace(_ty.cast(_ty.Any, ssh), verify=False)
    self.ssh = (
        ssh
        if ssh is not None
        else _ssh_config_from(shell, known_hosts, verify is not False)
    )
    # 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.

Every option that is not a secret and not at its default is in the query (verify, version, executor, path, autologin, timeout*), so TrueNASConfig.from_target(config.connection_uri) is an equal config. Credentials, the SSH leg and an ssl.SSLContext given as verify are not rendered.

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.

known_hosts = known_hosts instance-attribute

name property writable

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). Assigning it replaces the label.

needs_path_probe property

Whether the target left the API path unsaid (so version is used).

As with :attr:needs_scheme_probe, nothing here queries the server.

needs_scheme_probe property

Whether the target left ws-vs-wss unsaid (so TLS is assumed).

Not a pending action: this package never asks the server. It answers "was this decided by the caller or defaulted", which is what a caller that wants to probe for itself needs to know.

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, known_hosts, verify is not False) instance-attribute

timeout = _timeout(timeout) instance-attribute

timeouts property

:attr:timeout with each DEFAULT replaced by what applies.

connect is :data:~pytruenas.connection.CONNECT_TIMEOUT; read is $PYTRUENAS_CALL_TIMEOUT when set, else :data:~pytruenas.connection.CALL_TIMEOUT (read now, not at import); total stays DEFAULT: the command transport that runs a command applies its own.

uri_credentials = ('username', 'password', 'otp', 'api_key', 'token', 'credentials', 'verify', 'ssh', 'version', 'shell', 'known_hosts', 'executor', 'path', 'autologin', 'timeout') class-attribute instance-attribute

verify = _tls.check(verify) instance-attribute

version = version instance-attribute

__eq__(other)

Equal when :attr:connection_uri is: credentials are not compared.

Source code in src/pytruenas/host.py

def __eq__(self, other: object) -> bool:
    """Equal when :attr:`connection_uri` is: credentials are not compared."""
    if not isinstance(other, TrueNASConfig):
        return NotImplemented
    return self.connection_uri == other.connection_uri

__hash__()

Source code in src/pytruenas/host.py

def __hash__(self) -> int:
    return hash(self.connection_uri)

__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: "_ty.Optional[str]" = 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(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

Providers

pytruenas.providers

hostctl providers for a TrueNAS middleware target.

Only one adapter is TrueNAS-specific: :class:TrueNASWSPathProvider, 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 (:func:hostctl.local_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 -- there is nothing about TrueNAS to add.

On ordering (see TrueNASClient.default_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; hostctl's selector falls back to the websocket when SFTP cannot serve, and 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__ = ['TrueNASWSPathProvider', 'WS_PATH_CAPABILITIES', 'local_providers'] module-attribute

TrueNASWSPathProvider(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. This is :func:hostctl.local_providers unchanged: there is nothing TrueNAS-specific to add, so pytruenas defines no provider class for it. Returns a fresh (executor_provider, path_provider) pair, both named "local".

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. This is :func:`hostctl.local_providers` unchanged: there is nothing
    TrueNAS-specific to add, so pytruenas defines no provider class for it.
    Returns a fresh ``(executor_provider, path_provider)`` pair, both named
    ``"local"``.
    """
    from hostctl import local_providers as _local_providers

    return _local_providers()

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.

Protocol (verified against 26.0.0-BETA.1):

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", ...): a real interactive login shell (zsh for root on TrueNAS), not a request/response API. Input must be sent as BINARY frames -- a text frame reaches os.write(master_fd, ...) as str, the writer thread dies, and the connection resets with no error message.

A command runs through that terminal with :class:hostctl.console.PosixConsoleProfile: the command and any input= are typed base64-encoded inside fixed shell syntax, output is framed by markers the shell prints (never by the echo), stdin is a temporary file or /dev/null, and stderr and the exit status come back in their own frames. :class:WebShellExecutor is hostctl's :class:hostctl.console.ConsoleExecutor over that websocket, which gives it hostctl's subprocess-shaped call surface with stdout delivered as it arrives.

A command that exits the shell (exit 3) ends the session; the next call reconnects. This provider ranks after SSH, whose separate channels need none of the above.

CONNECT_TIMEOUT = 30.0 module-attribute

WEBSHELL_PATH = '/websocket/shell' module-attribute

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

WebShellExecutor(provider)

Bases: ConsoleExecutor

run() over the web shell: :class:hostctl.console.ConsoleExecutor.

Each command takes the provider's session, connecting it first; a connection that fails is a decline, so the host tries its next provider. No args, cwd or env: the terminal takes one line of shell text, which hostctl renders. A timeout or a failure closes the session, whose hang-up ends the command (TimeoutExpired.orphaned is False); the next command reconnects.

Source code in src/pytruenas/webshell.py

def __init__(self, provider: "WebShellExecutorProvider") -> None:
    self._provider = provider
    super().__init__(self._open, profile=_PROFILE)

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: "_ty.Optional[WebShellSession]" = None
    # No `args`: the terminal takes one line of shell text, so hostctl
    # renders the whole invocation to a single string rather than an argv.
    super().__init__("webshell", WebShellExecutor(self), 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()

Open the session; a failure is a decline naming its cause.

Source code in src/pytruenas/webshell.py

def connect(self) -> None:
    """Open the session; a failure is a decline naming its cause."""
    try:
        self.session.connect()
    except _OperationNotStartedError:
        raise
    except Exception as exc:
        raise _OperationNotStartedError(
            f"web shell unavailable: {type(exc).__name__}: "
            f"{_ProviderSelector.redact(exc)}",
            cause=exc,
        ) from exc

probe()

Report availability without dispatching a command.

A local target has the unix socket and does not need this.

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.
    """
    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: "_ty.Optional[dict]" = None
):
    self.client = client
    self.options = options or {}
    self._ws = None
    self._stream: "_ty.Optional[_WebSocketStream]" = None
    #: Guards opening and closing; never held while a command runs, so a
    #: timed-out command's session can be closed from another thread.
    self._lock = _threading.RLock()
    self.shell_id: "_ty.Optional[str]" = None

client = client instance-attribute

options = options or {} instance-attribute

shell_id = None instance-attribute

stream property

The session's terminal as a console stream; connects first.

close()

Source code in src/pytruenas/webshell.py

def close(self) -> None:
    with self._lock:
        ws, self._ws = self._ws, None
        self._stream = 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
        # The same trust as the API websocket -- see pytruenas._tls.
        ws = _websocket.WebSocket(sslopt=_tls.sslopt(self.client.config.verify))
        ws.connect(self._uri(), timeout=CONNECT_TIMEOUT)
        try:
            self._authenticate(ws)
        except BaseException:
            # A failed token, or a refusal of any shape, left the
            # websocket open.
            ws.close()
            raise
        self._ws = ws
        self._stream = _WebSocketStream(ws)
        return ws

clean_output(data)

Render raw PTY bytes as text: escape sequences, prompts and CRs removed.

A display helper for bytes read OUTSIDE a command's frame (a banner, an interactive session). Command output is never passed through it: what :class:WebShellExecutor returns is the program's own bytes.

Source code in src/pytruenas/webshell.py

def clean_output(data: bytes) -> str:
    """Render raw PTY bytes as text: escape sequences, prompts and CRs removed.

    A display helper for bytes read OUTSIDE a command's frame (a banner, an
    interactive session). Command output is never passed through it: what
    :class:`WebShellExecutor` returns is the program's own bytes.
    """
    data = _PROMPT.sub(b"", _ANSI.sub(b"", data)).replace(b"\r", b"")
    return data.decode("utf-8", "replace")