Tunnel

The otto.tunnel package builds and tears down host-resident bidirectional tunnels — one end-to-end forwarding path per add_tunnel call, realized as mirrored chains of tagged processes riding the topology edges modelled by otto.link. What each process actually executes is a pluggable carrier (otto.tunnel.carrier’s TunnelCarrier contract + CARRIERS registry); socat remains the first-party default. otto.tunnel.manage and otto.tunnel.discovery are the callable library API behind otto tunnel add / list / remove; otto.tunnel.socat is the socat carrier plus the pure command-builder layer it wraps, and otto.tunnel.sentinel is the argv-tag codec that makes every running process self-describing. For CLI usage, multi-hop chains, docker endpoints, and host requirements, see the user guide.

otto.tunnel — host-resident bidirectional tunnels (#2b spec).

Library-first: the CLI is a thin consumer of this package’s callable API. Grown task-by-task; final re-export surface lands with the manage layer.

Runtime Tunnel model — one end-to-end forwarding path (#2b spec §4).

A tunnel is what one otto tunnel add builds: service port + protocol + ordered host path (+ optional far-end delivery override). Its per-hop segments conceptually ride links (otto.link edges), but the packages are fully decoupled: NEITHER imports the other; shared daemon machinery lives in otto.host.daemon.

class otto.tunnel.model.Direction(value)

Bases: Enum

Which mirrored chain a process belongs to (spec §6.1).

FWD = 'fwd'

First-listed host toward last-listed host.

REV = 'rev'

Last-listed host toward first-listed host.

class otto.tunnel.model.Role(value)

Bases: Enum

What a tagged process does within its direction’s chain.

INGRESS = 'ingress'

Binds the service port on its endpoint’s data-plane ip.

RELAY = 'relay'

TCP carrier pass-through on an intermediate hop.

EGRESS = 'egress'

Delivers carrier traffic to the local service (or --dest).

otto.tunnel.model.ProcKey

One expected/observed tunnel process: (host_id, direction, role).

class otto.tunnel.model.TunnelHop(host: str, interface: str | None = None)

Bases: object

One position in a tunnel’s ordered path.

host : str

Host id (containers use their dotted parent.project.service id).

interface : str | None

Netdev key on the host; None = single/assumed interface (containers: always None — they have no modeled interfaces).

otto.tunnel.model.make_tunnel_id(path: tuple[TunnelHop, ...], protocol: str, service_port: int) → str

Deterministic id: tun-<12 hex>-<service_port> (spec §4).

The hash covers the ordered chain + protocol; the service port is a readable suffix, not hashed (same scheme as the retired lnk-<hex>-<port>). a,c,b != b,c,a — a reversed duplicate is rejected by the endpoint-bind conflict rule instead (spec §7).

class otto.tunnel.model.Tunnel(protocol: str, service_port: int, path: tuple[TunnelHop, ...], dest: str | None = None, id: str = '')

Bases: object

One end-to-end forwarding path (one add = one tunnel, spec D4).

protocol : str

"udp" or "tcp" (validated at the manage layer).

service_port : int

The port that binds on both endpoint hosts.

path : tuple[TunnelHop, ...]

Ordered chain, first = one endpoint, last = the other; len >= 2.

dest : str | None

Far-end delivery override host id (--dest), or None for loopback delivery on the last-listed host (spec §6.3).

id : str

Auto-computed via make_tunnel_id() when empty. A sentinel-parsed tunnel passes its wire id through verbatim (never recomputed).

expected_processes() → set[tuple[str, Direction, Role]]

Return the 2 * len(path) processes a healthy tunnel runs (spec §6.1).

Every host carries exactly two: its FWD-chain process and its REV-chain process, with the role determined by position (ingress at the direction’s origin, egress at its far end, relay between).

Sentinel v2 wire format: the argv marker every tunnel process carries (spec §5).

The running processes on the hosts ARE the tunnel record — zero persisted state. Each process is launched with argv[0]:

otto-tunnel:v1:<id>:<proto>:<svc-port>:<carrier-port>:<direction>:<role>:<hop-index>:<dest>:<path>

Prefix + version + 9 payload segments (framing: otto.host.daemon), each percent-encoded; empty segment = None. The path segment carries the full fwd-ordered chain (entries host@iface or bare host, each percent-encoded, joined with ,, the joined string percent-encoded once more), so any single surviving process reconstructs the entire intended tunnel — the record survives every other host being down.

Owner-agnostic (no username). Unknown versions/prefixes parse to None, never an error. STABILITY CONTRACT restarts at otto-tunnel:v1: from the first release with users, evolve only by adding versions and keeping old ones parseable. (The otto-link:v1 era predates users and is deleted.)

class otto.tunnel.sentinel.ParsedSentinel(tunnel: Tunnel, direction: Direction, role: Role, hop_index: int, carrier_port: int)

Bases: object

One decoded tunnel process: the whole tunnel + this process’s slot.

tunnel : Tunnel
direction : Direction
role : Role
hop_index : int
carrier_port : int
otto.tunnel.sentinel.encode_sentinel(tunnel: Tunnel, *, direction: Direction, role: Role, hop_index: int, carrier_port: int) → str

Return the wire token for one process of tunnel.

otto.tunnel.sentinel.parse_sentinel(token: str) → ParsedSentinel | None

Parse one wire token; None for non-otto / other-version / malformed.

Pluggable tunnel carriers: the TunnelCarrier contract + CARRIERS registry.

Mirrors the impairer registry (otto.link.impairer): custom carriers register from init modules under a name; otto tunnel add --carrier selects one per tunnel (chain-wide). Socat is the only first-party registrant (otto.tunnel.socat).

A carrier decides what each tagged process EXECUTES — nothing more. The 2n process topology (ingress/relay/egress x fwd/rev), the sentinel v1 wire format, free-port allocation, discovery, verify, and remove are all carrier-agnostic (spec 2026-07-11). The carrier name is deliberately not on the wire: a tunnel’s identity is its path+protocol+port, and remove reaps by pid, so it tears down any carrier’s processes.

otto.tunnel.carrier.DEFAULT_CARRIER = 'socat'

Name of the first-party carrier — the --carrier default at the CLI and the add_tunnel default in the library; one source of truth for both.

class otto.tunnel.carrier.TunnelCarrier

Bases: object

Builds the argv each tunnel process executes.

Stateless: implementations build argv lists; the orchestration layer (otto.tunnel.manage) launches them on hosts via otto.host.daemon.launch_command.

supported_protocols : ClassVar[frozenset[str]] = frozenset({})

Service protocols this carrier can forward (e.g. frozenset({"tcp"})).

requirements_command : ClassVar[str] = ''

Complete shell probe run on every chain host; prints the bare word ok on a line of its own iff satisfied (the check is a whole-line match, so failure text may safely contain the substring ok).

tools_description : ClassVar[str] = ''

Human summary of the required tools, for the missing-tools error.

ingress_args(protocol: str, service_port: int, bind_ip: str, next_ip: str, carrier_port: int) → list[str]

Argv accepting client traffic on the service port, shipping to the carrier.

relay_args(carrier_port: int, next_ip: str) → list[str]

Argv for an intermediate-hop pass-through (same carrier port both sides).

egress_args(protocol: str, service_port: int, deliver_ip: str, carrier_port: int) → list[str]

Argv accepting the carrier and delivering to the local service.

otto.tunnel.carrier.register_carrier(name: str, cls: type[TunnelCarrier], *, overwrite: bool = False) → None

Make a custom carrier selectable via --carrier <name>.

Call from an init module listed in .otto/settings.toml. The carrier must declare a non-empty TunnelCarrier.supported_protocols; otherwise it could never validate any tunnel and is rejected here.

otto.tunnel.carrier.build_carrier(name: str) → type[TunnelCarrier]

Return the carrier class registered under name (rich unknown-name error).

The socat carrier: pure command/argv builders + the SocatCarrier registrant — no I/O.

Bidirectional ingress/relay/egress builders (#2b); every value is a string or list of strings destined for host.exec; running nothing keeps the whole module unit-testable (assert exact argv).

otto.tunnel.socat.FREE_PORT_PROBE_COMMAND : str = 'ss -Htln 2>/dev/null || netstat -tln 2>/dev/null || true'

Free-port probe run on each chain host — ss preferred, netstat fallback (both exist on CentOS 6). Parsed by parse_listening_ports().

otto.tunnel.socat.ingress_socat_args(protocol: str, service_port: int, bind_ip: str, next_ip: str, carrier_port: int) → list[str]

Accept client traffic on the service port, ship over the TCP carrier.

Binds the endpoint’s data-plane ip specifically (never wildcard) so the reverse chain’s loopback delivery on this same host cannot U-turn into this listener (spec §6.3 loop hazard).

otto.tunnel.socat.relay_socat_args(carrier_port: int, next_ip: str) → list[str]

Intermediate-hop pass-through: same carrier port on both sides (§6.2).

otto.tunnel.socat.egress_socat_args(protocol: str, service_port: int, deliver_ip: str, carrier_port: int) → list[str]

Accept the TCP carrier, deliver to the service (loopback or --dest).

otto.tunnel.socat.parse_listening_ports(output: str) → set[int]

Extract every port appearing as :<port> in ss/netstat output.

Safe superset of used ports — we only need to avoid them.

otto.tunnel.socat.pick_free_port(used: set[int], lo: int = 49152, hi: int = 65535) → int

First port in [lo, hi] not in used. Raises when exhausted.

class otto.tunnel.socat.SocatCarrier

Bases: TunnelCarrier

socat over a TCP4 carrier — the first-party tunnel transport (#2b).

supported_protocols : ClassVar[frozenset[str]] = frozenset({'tcp', 'udp'})

Service protocols this carrier can forward (e.g. frozenset({"tcp"})).

requirements_command : ClassVar[str] = 'command -v socat >/dev/null 2>&1 && command -v bash >/dev/null 2>&1 && echo ok || echo no'

Complete shell probe run on every chain host; prints the bare word ok on a line of its own iff satisfied (the check is a whole-line match, so failure text may safely contain the substring ok).

tools_description : ClassVar[str] = 'socat and/or bash'

Human summary of the required tools, for the missing-tools error.

ingress_args(protocol: str, service_port: int, bind_ip: str, next_ip: str, carrier_port: int) → list[str]

Delegate to ingress_socat_args() (the proven builder).

relay_args(carrier_port: int, next_ip: str) → list[str]

Delegate to relay_socat_args().

egress_args(protocol: str, service_port: int, deliver_ip: str, carrier_port: int) → list[str]

Delegate to egress_socat_args().

Live tunnel discovery — the processes on the hosts ARE the record (spec §9).

_scan_hosts gathers DISCOVERY_PS_COMMAND across hosts (best-effort, bounded, transparent about unreachables); parse_process_discovery decodes each tagged process; and discover_tunnels groups observations by tunnel id, comparing what was observed against what the sentinel-encoded path says must exist.

otto.tunnel.discovery.DISCOVERY_PS_COMMAND : str = "ps -eo pid= -eo etime= -eo args= 2>/dev/null | \\grep -a ' otto-tunnel:' || true"

The lab-wide daemon scan for tunnel processes. Built by otto.host.daemon.ps_scan_command() — see it for the procps portability story. The bytes are pinned by TestPsScanCommand in tests/unit/host/test_daemon.py (STABILITY CONTRACT).

class otto.tunnel.discovery.Observation(pid: int, age_seconds: int, parsed: ParsedSentinel)

Bases: object

One tagged tunnel process seen on one host.

pid : int
age_seconds : int
parsed : ParsedSentinel
otto.tunnel.discovery.parse_process_discovery(ps_output: str) → list[Observation]

Reconstruct observations from DISCOVERY_PS_COMMAND output.

async otto.tunnel.discovery.discover_observations(lab: Lab) → tuple[list[tuple[str, Observation]], list[str]]

Every tagged tunnel process across the lab’s has_bash hosts.

class otto.tunnel.discovery.DiscoveredTunnel(tunnel: Tunnel, present: set[tuple[str, Direction, Role]], missing: set[tuple[str, Direction, Role]], age_seconds: int, uncertain: bool)

Bases: object

One live tunnel: intended shape + what was actually observed.

tunnel : Tunnel
present : set[tuple[str, Direction, Role]]
missing : set[tuple[str, Direction, Role]]

Expected-but-absent processes on hosts that WERE scanned. Absence on an unreachable host is unknown, not missing (spec §9).

age_seconds : int

Oldest observed process age (max etime) — the tunnel’s creation age.

uncertain : bool

True when >=1 chain host was unreachable during the scan.

property status : str

ok / degraded (<present>/<expected>), ?-suffixed if uncertain.

class otto.tunnel.discovery.TunnelDiscovery(tunnels: list[DiscoveredTunnel], unreachable: list[str])

Bases: object

A full scan: the tunnels seen plus the hosts that couldn’t be scanned.

tunnels : list[DiscoveredTunnel]
unreachable : list[str]
otto.tunnel.discovery.group_observations(observations: list[tuple[str, Observation]], unreachable: list[str]) → list[DiscoveredTunnel]

Group per-host observations by tunnel id and compute per-tunnel status.

async otto.tunnel.discovery.discover_tunnels(lab: Lab) → TunnelDiscovery

Discover live otto tunnels across the lab (the monitor-facing surface).

Async orchestration for tunnels — the callable library API (spec §6-§8, §10-§12).

The CLI is a thin consumer of add_tunnel / remove_tunnel / remove_all_tunnels; each is usable standalone from any Python code.

class otto.tunnel.manage.ResolvedHop(hop: TunnelHop, ip: str, host: Any)

Bases: object

One chain position resolved against the live lab.

hop : TunnelHop
ip : str
host : Any
class otto.tunnel.manage.AddedTunnel(tunnel: Tunnel, carrier_fwd: int, carrier_rev: int)

Bases: object

A successfully added + verified tunnel.

tunnel : Tunnel
carrier_fwd : int
carrier_rev : int
async otto.tunnel.manage.add_tunnel(lab: Lab, hosts: list[tuple[str, str | None]], *, port: int, protocol: str = 'tcp', dest: tuple[str, str | None] | None = None, carrier: str = 'socat') → AddedTunnel

Build a bidirectional host-resident tunnel and verify it came up (spec §6).

Launch order is downstream-first per direction; any launch failure or a failed post-add verify reaps everything already started — no half-tunnels survive a failed add. “Started” is tracked from the moment a launch is attempted, not from a confirmed ack: a launch exec that times out only bounds how long we waited for the reply, not whether the command reached the host, so even a first-launch timeout triggers rollback. The carrier names a registered TunnelCarrier (chain-wide; default "socat").

class otto.tunnel.manage.RemovedReport(removed_ids: list[str], killed: dict[str, list[int]], unreachable: list[str], survivors: list[tuple[str, int]])

Bases: object

What a reap pass tore down — and what refused to die (spec §10).

removed_ids : list[str]
killed : dict[str, list[int]]
unreachable : list[str]
survivors : list[tuple[str, int]]

(host_id, pid) processes still present in the post-kill verify scan.

async otto.tunnel.manage.remove_tunnel(lab: Lab, tunnel_id: str) → RemovedReport

Reap one tunnel by id, then verify its processes are actually gone.

async otto.tunnel.manage.remove_all_tunnels(lab: Lab) → RemovedReport

Reap every otto tunnel (owner-agnostic), with the same verify pass.