host.login_proxy¶

A cred entry may declare that its login cannot be reached by direct authentication: to become it, otto authenticates (or starts) as another account (via) and replays the named proxy’s send/expect steps. Proxies are async callables registered by libraries from init modules via register_login_proxy(), mirroring the term/transfer registries. The built-in "su" proxy is the default user-switch mechanism (it replaces the old hardcoded su-switch helper that switch_user/as_user used to call directly).

class otto.host.login_proxy.Cred(login: str, password: str | None = None, proxy: str | None = None, via: str | None = None, params: dict[str, ~typing.Any] = <factory>)¶

Bases: object

One credential entry: a login plus how to become it.

login : str¶: The account name.

password : str | None = None¶: Password, or None (key/agent auth on SSH; empty line on telnet; no password exchange in the built-in su proxy).

proxy : str | None = None¶: Login-proxy registry key; None means directly loginable (switch via the built-in "su").

via : str | None = None¶: Login of the account the proxy starts from. None defaults to the first proxy-less cred entry.

params : dict[str, Any]¶: Free-form data handed to the proxy callable (host-specific knobs).

class otto.host.login_proxy.ProxyIO(*args, **kwargs)¶

Bases: Protocol

Minimal I/O handle a proxy drives.

Satisfied by hosts, HostSession instances, the raw-session adapter used at session establishment, and the interact bridge adapter.

async send(text: str, *, log: LogMode = LogMode.NORMAL) → None¶: Send text to the proxy IO.

async expect(pattern: str | Pattern[str], timeout: float = 10.0) → str¶: Expect a pattern and return the matched output.

class otto.host.login_proxy.ProxyContext(target: Cred, via: Cred, host_id: str)¶

Bases: object

Everything a proxy step may need.

Deliberately NOT the host object — running commands mid-proxy on the session being established deadlocks.

target : Cred¶

via : Cred¶

host_id : str¶

otto.host.login_proxy.LoginProxyFn¶: An async callable that drives one proxy’s steps: async def proxy(io, ctx).

class otto.host.login_proxy.LoginProxy(fn: Callable[[ProxyIO, ProxyContext], Awaitable[None]], undo: Callable[[ProxyIO, ProxyContext], Awaitable[None]] | None = None)¶

Bases: object

A registered proxy: the steps plus an optional reversal.

fn : Callable[[ProxyIO, ProxyContext], Awaitable[None]]¶

undo : Callable[[ProxyIO, ProxyContext], Awaitable[None]] | None = None¶

exception otto.host.login_proxy.LoginProxyError¶

Bases: ConnectionError

A proxy step failed or a chain could not be resolved.

otto.host.login_proxy.register_login_proxy(name: str, fn: Callable[[ProxyIO, ProxyContext], Awaitable[None]], *, undo: Callable[[ProxyIO, ProxyContext], Awaitable[None]] | None = None, overwrite: bool = False) → None¶

undo reverses the steps for as_user restore; None means the default reversal (send exit), correct for any su/sudo-style nested shell.

otto.host.login_proxy.cred_for(creds: list[Cred], login: str) → Cred | None¶: Look up a cred entry by login (None when absent).

otto.host.login_proxy.resolve_chain(creds: list[Cred], target_login: str) → tuple[Cred, list[Cred]]¶

Resolve the direct-auth cred and the hop list for target_login.

Returns (direct, hops) where direct is the cred to authenticate the transport as and hops are the proxied creds to apply afterwards, outermost (first to run) first. Spec validation guarantees termination; the seen set is a runtime backstop against hand-built cred lists.

async otto.host.login_proxy.run_proxy(io: ProxyIO, hop: Cred, via: Cred, host_id: str) → None¶

Run hop’s proxy steps over io, wrapping failures with context.

Ends with a post-transition shell resync (_resync_shell) so the next sentinel-wrapped command otto writes can’t land in the transition’s tty-flush window and be silently discarded (see that function’s docstring). A resync failure surfaces through the same wrapping as any other proxy-step failure below.

async otto.host.login_proxy.run_undo(io: ProxyIO, hop: Cred, via: Cred, host_id: str) → None¶

Reverse hop: the registered undo, or the default exit.

Also ends with a post-transition shell resync, like run_proxy() — the exit back to the prior shell is the same kind of foreground handoff a su/sudo switch is, and races the next command the same way. Failures are wrapped in LoginProxyError with context, like run_proxy().

async otto.host.login_proxy.perform_switch(io: ProxyIO, creds: list[Cred], user: str, password: str | None, current_user: str, host_id: str) → list[Cred]¶

Become user from current_user; return the hops applied, in order.

Semantics preserved from the pre-proxy switch_user: user="" targets root via bare su; an explicit password overrides the creds entry; a user with no creds entry is an ad-hoc su target. A cred whose via differs from current_user first switches to the via account (recursively), so as_user can undo hop-by-hop.