alkdev/alkssh
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(session): land the session mapping — the translation bridge + the exit derivation (w3-tty-session)

Browse Source 
The biggest single-row surface of the spine: pty/shell/pipe-exec serve
through the published alktty seam (drive_session_pre_negotiated) over a
translation bridge, re-landing POC-3 stage D as contract:

- src/tty_map.rs (session feature): the bridge (SSH data → Chunk::stdin
  through a bounded mpsc + framing task; SSH EOF → the zero-length
  stdin sentinel — verified against alktty's pump semantics; stdout →
  data; stderr → extended_data(1); exit chunk → the derivation
  contract), the locally-built negotiation (OQ-08: shell argv verbatim,
  wrapper ++ [cmd], cwd/env session-side, ttymodes → the reserved {}
  modes JSON), and the spawned serving path.
- serve_tty_row: driven chunks→SSH inline so the exit report is the
  path's final act (status before close, ADR-006 decision 4); two
  cancel signals (per-channel watch fired by channel_close, connection
  teardown fired by SessionHandler::Drop) abort the alktty session
  task — the ADR-056 kill-on-drop contract is the cancelled session's
  cleanup.
- Derivation contract rules 1–6 in derive_exit + the crate-local
  signum→name table: −N → exit-signal(name), −1 → exit-status(255),
  core_dumped never, unmapped negatives → 255, the PTY-path rule-5
  fidelity limit documented; wire rows pin the shapes.
- Negotiation-error frames (alktty's 0x00-length-prefixed error
  framing) → negotiation-failed serving decline + exit-status(255) +
  close — the one could-not-start claim this door makes (nothing ran).
- Control rows: window_change/signal → the per-channel BridgeSurface
  (bounded mpsc, try_send, never awaited on the packet loop); signal
  names verbatim (no decline shape — fire-and-forget); pre-serve
  resizes update the stored pty params (final size wins).
- Unset-declines (OQ-08) keep the git row's serving-path shape
  (failure then close, spawned): shell-unconfigured /
  wrapper-unconfigured / seam-unconfigured; pipe-less shell SERVES
  pipe-mode (the recorded vanilla-sshd posture); env_request decline
  unchanged (OQ-09).
- tests: the hand-rolled TtyBackend shim + ownership witness doubles;
  serving rows through the harness, the unset-field decline matrix,
  derivation rows on the wire, negotiation-failure rows, the identity
  ride asserted live at the seam, the kill witness; the real-tool
  smoke — OpenSSH -tt against a real PTY via dev-dep alktty local
  (test graph only), skip-with-reason posture. Wave-2 stub rows
  migrated to the new decline reasons (dispatch_core, observe).

Verification: cargo test (113) / --features session (149) / git (142) /
tunnel (115) / --all-features (161) — all green; cargo clippy
--all-targets -- -D warnings ×5 feature configs; cargo fmt --check;
cargo doc --no-deps ×3 builds clean; taskgraph validate.
This commit is contained in:
glm-5.3-flash committed 2026-10-28 20:29:59 +00:00
1 parent 7aac2d11a2
commit b1e569d1a8
13 files changed
+3018 -155
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/handler.rs
+422 -89
View File
@@ -70,7 +70,7 @@
//! capability leak — the POC-3 stage-A rule); the per-length
//! `max_exec_command_len` cap is enforced before the parse itself.
//!
//! ## The store-and-take channel handoff (`w3-git-mapping`)
//! The store-and-take channel handoff (`w3-git-mapping`)
//!
//! A confirmed `session` open **stores** its [`Channel`] in
//! per-connection handler state ([`Self::take_stored_channel`]'s
@@ -83,9 +83,43 @@
//! mutation happens on the loop's task — single-writer, no lock. A
//! git exec on a channel whose stored handle is already gone
//! (re-served, or never held — unvanilla) declines serving-path.
//! The other session rows currently leave the handle stored until
//! the close releases it (the `w3-tty-session` mapping will take it
//! the same way).
//!
//! ## The session rows' real serving paths (`w3-tty-session`)
//!
//! The pty/shell/pipe rows replaced their wave-3 skeleton stubs with
//! the alktty translation bridge ([`crate::tty_map`]): the pty row
//! stores the negotiated `TerminalParamsWire` ([`Self::ptys`]'s
//! supplier — taken by the later shell/exec request on the same
//! channel, `ssh -t host "cmd"`'s pty-then-exec); the shell and pipe
//! rows take the stored channel the same way the git row does and
//! spawn `serve_tty_row` over the configured seam — locally-built
//! negotiation, unset-fields-decline (OQ-08: `shell_argv` unset →
//! the shell row declines, `exec_wrapper_argv` unset → the pipe row
//! declines, `session_seam` unset → both rows and the pty row
//! decline `seam-unconfigured`; every one of them the serving-path
//! failure+close shape from a spawned task). A routed request
//! answers `channel_success` inline (the request routed; the exit
//! chunk reports the outcome later, status-before-close from the
//! spawned task).
//!
//! ## The control rows' per-channel surface (`w3-tty-session`)
//!
//! `window_change_request` and `signal` arrive on the packet loop
//! **after** the serving task owns the wire — the spawned bridge is
//! reached through [`Self::bridges`]' per-channel surface
//! ([`crate::tty_map::BridgeSurface`]: the bounded control mpsc's
//! `try_send` — never awaited on the packet loop — plus the
//! per-channel cancel). A resize for a channel whose pty is stored
//! but not yet serving updates the stored params (the final size
//! wins); signal names are accepted verbatim (the backend owns the
//! semantics). Unfeatured builds emit the request-level decline
//! event with no reply — the control rows are fire-and-forget, no
//! SSH reply exists to send. [`Self::channel_close`] releases the
//! surface (firing the cancel: the alktty kill-on-drop contract is
//! the cancelled session's cleanup) along with the channel and pty
//! stores. The **connection teardown** signal ([`crate::tty_map::
//! Teardown`]) fires from this handler's `Drop` so a serving task
//! can never outlive the russh session that spawned it.
//!
//! ## The spawn-and-return structural constraint (ADR-006 decision 4)
//!
@@ -94,15 +128,17 @@
//! routing and deadlocks against the peer's follow-up bytes (POC-3
//! proved by construction with receive-pack). Every serving path is
//! therefore spawned through [`crate::serve::spawn_serving_path`] and
//! the handler returns fast: the session-row seam stubs clone
//! `session.handle()`, spawn their decline through the clone
//! ([`crate::serve`]), and are gone from the hook's stack — no hook
//! the handler returns fast: the serving rows (the git row's alkgit
//! dispatch, the session rows' config-driven declines and
//! [`crate::tty_map`]'s bridge) all clone `session.handle()`, spawn
//! through the clone, and are gone from the hook's stack — no hook
//! awaits a serving path inline. The decline keeps its wire shape
//! (failure then close), riding the spawn. Exit/status reporting from
//! spawned tasks rides the same cloned handle
//! ([`crate::serve::report_exit_status_then_close`] — the mechanics;
//! the derivation rules are wave-3 `mapping.md` material), pinned
//! end-to-end by the mechanics rows in `tests/spawn_return.rs`.
//! the tty rows' derivation rules live in [`crate::tty_map`]), pinned
//! end-to-end by the mechanics rows in `tests/spawn_return.rs` and the
//! mapping rows in `tests/git_mapping.rs` / `tests/tty_mapping.rs`.
//!
//! The open-level declines (unfeatured/never-feature rows, the tunnel
//! seam stub) stay inline by class: they answer the open *before* any
@@ -141,16 +177,21 @@
//! session-survives-declines posture) — including the git grammar's
//! declinables (`unknown-service` / `service-only` /
//! `invalid-repo-id`, the parsed reply shapes, `w3-git-mapping`);
//! a *serving-path* decline (a routable row that owns the channel:
//! the seam-unconfigured git row and the unvanilla
//! channel-not-held / identity-less guards) answers failure and
//! closes the channel — the vanilla-clean shape for a row that owns
//! the channel. The serving-path decline rides the spawned serving
//! task (ADR-006 decision 4, §The spawn-and-return structural
//! constraint above); the wire shape is unchanged. The real
//! serve's own outcomes are not declines: the substrate maps them
//! (denial band before any protocol byte; exit-status then close —
//! the `exit_code` mapping decision recorded in [`crate::git_map`]).
//! a *serving-path* decline (a routable row that owns the request's
//! whole outcome: the seam-unconfigured git row, the session rows'
//! unset-field and unset-seam declines — `w3-tty-session`'s OQ-08
//! shapes — and the identity-less guard) answers failure and closes
//! the channel — the vanilla-clean shape for a row whose serving
//! path cannot exist. The serving-path decline rides the spawned
//! serving task (ADR-006 decision 4, §The spawn-and-return
//! structural constraint above); the wire shape is unchanged. The
//! `channel-not-held` guard is the inline request-level shape (the
//! channel that could not be held stays usable). The real serve's
//! own outcomes are not declines: the substrate maps them (denial
//! band before any protocol byte; exit-status then close — the
//! `exit_code` mapping decision recorded in [`crate::git_map`];
//! the tty bridge's negotiation-error and exit-chunk decisions in
//! [`crate::tty_map`]).
//!
//! ## The event vocabulary on every decline (`w3-decline-tracing`)
//!
@@ -199,20 +240,27 @@ use std::sync::Arc;
use alkcall::core::auth::Identity;
use russh::keys::Algorithm;
use russh::server::{
Auth, ChannelOpenHandle, Handle, Msg, PublicKeyOffer, Session, VerifiedPublicKey,
};
use russh::{Channel, ChannelId, Pty};
#[cfg(any(feature = "git", feature = "session"))]
use russh::server::Handle;
use russh::server::{Auth, ChannelOpenHandle, Msg, PublicKeyOffer, Session, VerifiedPublicKey};
use russh::{Channel, ChannelId, Pty, Sig};
use crate::channel_cap::ChannelCap;
use crate::config::ServeConfig;
#[cfg(feature = "session")]
use crate::config::SessionSeam;
use crate::dispatch::{self, ExecRoute};
use crate::error::AdapterError;
use crate::fingerprint::fingerprint;
#[cfg(feature = "git")]
use crate::git_map;
use crate::observe;
#[cfg(any(feature = "git", feature = "session"))]
use crate::serve;
#[cfg(feature = "session")]
use crate::tty_map;
#[cfg(feature = "session")]
use alktty::negotiation::NegotiateRequest;
/// The per-connection handler (one per mounted stream): russh constructs
/// nothing here — [`crate::SshAdapter::handle`] builds one per accepted
@@ -231,10 +279,39 @@ pub struct SessionHandler {
/// The store-and-take handoff (`w3-git-mapping`): confirmed
/// session channels held from the open until their row takes or
/// the close releases them. Written only by the packet loop's
/// hooks (`channel_open_session` inserts; `exec_request` removes
/// for the git row; `channel_close` removes) — single-writer, no
/// lock.
/// hooks (`channel_open_session` inserts; the git row's and the
/// session rows' serving dispatches remove for their taken
/// channel; `channel_close` removes) — single-writer, no lock.
channels: HashMap<ChannelId, Channel<Msg>>,
/// The session rows' negotiated pty params (`w3-tty-session`):
/// `pty_request` stores, the later `shell_request`/`exec_request`
/// on the same channel consumes (`ssh -t host "cmd"`'s
/// pty-then-exec), the close releases. Packet-loop only.
#[cfg(feature = "session")]
ptys: HashMap<ChannelId, alktty::negotiation::TerminalParamsWire>,
/// The per-channel control surfaces (`w3-tty-session`): installed
/// when a session row's serving task spawns, forwarded to by
/// `window_change_request`/`signal`, fired-and-removed by
/// `channel_close`. Packet-loop only.
#[cfg(feature = "session")]
bridges: HashMap<ChannelId, tty_map::BridgeSurface>,
/// The connection-teardown signal (`w3-tty-session`): fired by
/// this handler's [`Drop`](Self::drop) so a serving task never
/// outlives the russh session that spawned it (the alktty
/// kill-on-drop contract is the cancelled session's cleanup).
#[cfg(feature = "session")]
teardown: tty_map::Teardown,
}
#[cfg(feature = "session")]
impl Drop for SessionHandler {
/// Fires the session rows' teardown signal (`w3-tty-session`):
/// every spawned serving task of this connection watches it and
/// cancels its alktty session (the kill-on-drop cleanup) instead
/// of outliving the connection.
fn drop(&mut self) {
self.teardown.fire();
}
}
impl SessionHandler {
@@ -248,15 +325,22 @@ impl SessionHandler {
channel_cap: ChannelCap::default(),
resolved_identity: None,
channels: HashMap::new(),
#[cfg(feature = "session")]
ptys: HashMap::new(),
#[cfg(feature = "session")]
bridges: HashMap::new(),
#[cfg(feature = "session")]
teardown: tty_map::Teardown::new(),
}
}
/// Takes a stored channel handle (the store half of the
/// store-and-take handoff): the git row's dispatch removes the
/// handle the open stored, handing the stream into the spawned
/// serving task. `None` when nothing is held — a re-served or
/// never-held channel (unvanilla; declined serving-path).
#[cfg(feature = "git")]
/// store-and-take handoff): the git row's and the session rows'
/// dispatches remove the handle the open stored, handing the
/// stream into the spawned serving task. `None` when nothing is
/// held — a re-served or never-held channel (unvanilla; declined
/// serving-path).
#[cfg(any(feature = "git", feature = "session"))]
fn take_stored_channel(&mut self, channel: ChannelId) -> Option<Channel<Msg>> {
self.channels.remove(&channel)
}
@@ -343,6 +427,7 @@ impl SessionHandler {
/// The session-row gate (`pty`/`shell`/pipe-exec): only the
/// `session` feature serves these rows.
#[cfg(feature = "session")]
fn serves_session_rows(&self) -> bool {
cfg!(feature = "session")
}
@@ -452,6 +537,13 @@ impl russh::server::Handler for SessionHandler {
_session: &mut Session,
) -> Result<(), Self::Error> {
self.channels.remove(&channel);
#[cfg(feature = "session")]
{
self.ptys.remove(&channel);
if let Some(surface) = self.bridges.remove(&channel) {
tty_map::fire_cancel(&surface.cancel);
}
}
self.channel_cap.release();
Ok(())
}
@@ -644,62 +736,220 @@ impl russh::server::Handler for SessionHandler {
decline_unfeatured_request(channel, session).await;
}
ExecRoute::Pipe => {
#[cfg(feature = "session")]
if self.serves_session_rows() {
pipe_row_stub(channel, session);
} else {
observe::decline_request(
"exec_request",
"unfeatured",
channel.number(),
Some("pipe"),
);
decline_unfeatured_request(channel, session).await;
pipe_row_dispatch(self, channel, session, command.to_owned());
return Ok(());
}
#[cfg(not(feature = "session"))]
let _ = &command;
observe::decline_request(
"exec_request",
"unfeatured",
channel.number(),
Some("pipe"),
);
decline_unfeatured_request(channel, session).await;
}
}
Ok(())
}
/// The `pty_request` row route (`session`): featured routes to
/// the session-row seam stub; unfeatured declines the capability.
/// The `pty_request` row (`session`): the args →
/// `TerminalParamsWire` (ttymodes → the reserved `{}` JSON,
/// OQ-44), stored per channel for the later shell/exec request on
/// the same channel — after the seam check (a pty whose session
/// can never serve is a declined allocation,
/// `seam-unconfigured`). Unfeatured builds decline the
/// capability. The successful row answers `channel_success`
/// (POC-3 §D.1's shape — serving starts at the shell/exec
/// request).
async fn pty_request(
&mut self,
channel: ChannelId,
_term: &str,
_col_width: u32,
_row_height: u32,
_pix_width: u32,
_pix_height: u32,
term: &str,
col_width: u32,
row_height: u32,
pix_width: u32,
pix_height: u32,
_modes: &[(Pty, u32)],
session: &mut Session,
) -> Result<(), Self::Error> {
if self.serves_session_rows() {
pty_row_stub(channel, session);
} else {
#[cfg(feature = "session")]
match self.serve.session_seam.clone() {
Some(_) => {
self.ptys.insert(
channel,
tty_map::terminal_params(term, col_width, row_height, pix_width, pix_height),
);
let _ = session.channel_success(channel);
}
None => {
spawn_seam_stub(channel, session, "pty_request", None, "seam-unconfigured");
}
}
#[cfg(not(feature = "session"))]
{
let _ = (term, col_width, row_height, pix_width, pix_height);
observe::decline_request("pty_request", "unfeatured", channel.number(), None);
decline_unfeatured_request(channel, session).await;
}
Ok(())
}
/// The `shell_request` row route (`session`): the same shape as
/// `pty_request`. The OQ-08 unset-shell check is the serving
/// path's consumption (wave 3): this wave the stub declines the
/// row regardless, so the unset-declines posture already holds.
/// The `shell_request` row (`session`): the serve-config shell
/// argv, verbatim, as `NegotiateRequest.cmd`; the stored pty
/// params consumed (`None` → pipe-mode shell — decided: a pty-less
/// shell request is served, `ssh -T host` is legitimate vanilla);
/// unset-declines first (OQ-08: `shell_argv` unset →
/// `shell-unconfigured`; the seam unset → `seam-unconfigured`
/// — both the spawned serving-path shape). A routed request
/// answers `channel_success` inline; the exit chunk reports the
/// outcome, status-before-close from the spawned task.
async fn shell_request(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), Self::Error> {
if self.serves_session_rows() {
shell_row_stub(channel, session);
} else {
#[cfg(feature = "session")]
match (
self.serve.shell_argv.clone(),
self.serve.session_seam.clone(),
) {
(None, _) => {
spawn_seam_stub(
channel,
session,
"shell_request",
None,
"shell-unconfigured",
);
}
(_, None) => {
spawn_seam_stub(channel, session, "shell_request", None, "seam-unconfigured");
}
(Some(shell_argv), Some(seam)) => {
let tty = self.ptys.remove(&channel);
let Some(taken) = self.take_stored_channel(channel) else {
observe::decline_request(
"shell_request",
"channel-not-held",
channel.number(),
None,
);
let _ = session.channel_failure(channel);
return Ok(());
};
let request = tty_map::negotiate_request(
seam.backend.clone(),
tty,
shell_argv,
self.serve.session_cwd.clone(),
self.serve.session_env.clone(),
);
spawn_tty_row(
self,
channel,
session,
"shell_request",
None,
taken,
request,
seam,
);
}
}
#[cfg(not(feature = "session"))]
{
observe::decline_request("shell_request", "unfeatured", channel.number(), None);
decline_unfeatured_request(channel, session).await;
}
Ok(())
}
/// `window_change_request` (`session`): the resize control row —
/// forwarded into the serving bridge's bounded control surface
/// (`try_send`, never awaited on the packet loop); for a channel
/// whose pty is stored but not yet serving, the stored params
/// resize instead (the final size wins — vanilla sshd's own
/// ordering); with neither, the event is dropped. Fire-and-forget
/// on the wire — no SSH reply exists, so a decline sends none
/// (the event is the record). Signal names: see [`Self::signal`].
async fn window_change_request(
&mut self,
channel: ChannelId,
col_width: u32,
row_height: u32,
pix_width: u32,
pix_height: u32,
_session: &mut Session,
) -> Result<(), Self::Error> {
#[cfg(feature = "session")]
{
let cols = tty_map::clamp_u16(col_width);
let rows = tty_map::clamp_u16(row_height);
let pix_w = tty_map::clamp_u16(pix_width);
let pix_h = tty_map::clamp_u16(pix_height);
if let Some(surface) = self.bridges.get(&channel) {
let message = alktty::ControlMessage::Resize {
cols,
rows,
pixel_width: pix_w,
pixel_height: pix_h,
};
let _ = surface.forward_control(&message);
} else if let Some(params) = self.ptys.get_mut(&channel) {
params.cols = cols;
params.rows = rows;
params.pixel_width = pix_w;
params.pixel_height = pix_h;
params.modes = serde_json::json!({});
}
}
#[cfg(not(feature = "session"))]
{
let _ = (col_width, row_height, pix_width, pix_height);
observe::decline_request(
"window_change_request",
"unfeatured",
channel.number(),
None,
);
}
Ok(())
}
/// The `signal` row (`session`): accepted verbatim — no
/// alkssh-side name filter (the backend owns the semantics;
/// alktty's control handle applies its own default kill for
/// names it rejects, and the pipe-mode pid-only vs PTY
/// process-group scope is backend semantics) — so there is no
/// decline shape: the request is fire-and-forget on the SSH
/// wire, and the bridge's bounded surface is the only path. An
/// unfeatured build emits the decline event with no reply.
async fn signal(
&mut self,
channel: ChannelId,
signal: Sig,
_session: &mut Session,
) -> Result<(), Self::Error> {
#[cfg(feature = "session")]
{
if let Some(surface) = self.bridges.get(&channel) {
let message = alktty::ControlMessage::Signal {
name: tty_map::sig_control_name(&signal),
};
let _ = surface.forward_control(&message);
}
}
#[cfg(not(feature = "session"))]
{
let _ = signal;
observe::decline_request("signal", "unfeatured", channel.number(), None);
}
Ok(())
}
/// Every subsystem name declines (unimplemented on this door):
/// `subsystem("sftp")` is the OQ-13 placeholder — the feature
/// exists, the route does not until alkfs lands; git service
@@ -823,21 +1073,22 @@ async fn decline_unfeatured_request(channel: ChannelId, session: &mut Session) {
let _ = session.channel_failure(channel);
}
/// The family-seam decline shape for a **routed-but-not-yet-served**
/// row — now the spawned serving task's body: fail the request and
/// close the channel, through the cloned handle the task was handed.
/// The vanilla-clean shape for a serving path that owns the channel
/// (wave 3 replaces these stubs with the real serving paths; a
/// declined seam never leaks a half-served channel, since the decline
/// IS the row this wave). Decline-first-alloc-never holds structurally:
/// nothing is allocated before the failure lands — the spawn's handle
/// clone is the only resource ever taken.
/// The spawned serving-path decline body (`w3-git-mapping`'s shape,
/// reused by `w3-tty-session`'s config-driven declines): fail the
/// request and close the channel, through the cloned handle the task
/// was handed — the vanilla-clean shape for a routed row that owns
/// the request's whole outcome. The helper name keeps the wave-3
/// history (the decline IS the path it spawns). Decline-first-alloc-
/// never holds structurally: nothing is allocated before the failure
/// lands — the spawn's handle clone is the only resource ever taken.
///
/// The serving-path lifecycle events ride this body
/// ([`crate::observe`]): started (the task's first action), the
/// serving-path decline (class `serving`, reason `seam-stub`), and —
/// after the wire reporting has completed — ended, or failed when a
/// report send could not complete (the one warn in the vocabulary).
/// serving-path decline (class `serving`, the caller's vocabulary
/// reason), and — after the wire reporting has completed — ended, or
/// failed when a report send could not complete (the one warn in the
/// vocabulary).
#[cfg(any(feature = "git", feature = "session"))]
async fn seam_stub_decline(
handle: Handle,
channel: ChannelId,
@@ -862,6 +1113,7 @@ async fn seam_stub_decline(
/// task, and the hook returns — no serving path awaits inline. The
/// row never serves, so the decline *is* the path: failure + close,
/// event-shaped ([`crate::observe`]) with the caller's reason.
#[cfg(any(feature = "git", feature = "session"))]
fn spawn_seam_stub(
channel: ChannelId,
session: &mut Session,
@@ -929,28 +1181,109 @@ fn git_row_dispatch(
let _ = session.channel_success(channel);
}
/// The session pipe row's seam stub — wave 3 lands the alktty pipe
/// seam here (`wrapper ++ [cmd]`, the OQ-08 argv append), spawned per
/// ADR-006 decision 4; the skeleton declines at the seam, on its own
/// task.
fn pipe_row_stub(channel: ChannelId, session: &mut Session) {
spawn_seam_stub(channel, session, "exec_request", Some("pipe"), "seam-stub");
/// The session pipe row's dispatch (`w3-tty-session`): unset-declines
/// first (OQ-08 — `exec_wrapper_argv` unset → `wrapper-unconfigured`;
/// the seam unset → `seam-unconfigured`; both the spawned serving-path
/// shape, never a fallback), then the store-and-take handoff
/// (`channel-not-held` → the inline unvanilla decline, the git row's
/// shape), the wire command appended to the wrapper argv (`wrapper ++
/// [cmd]` — never naively split), the stored pty params consumed if a
/// pty preceded, and the spawned bridge. A routed exec answers
/// `channel_success` inline (the request routed; the exit chunk
/// reports the serving outcome later).
#[cfg(feature = "session")]
fn pipe_row_dispatch(
handler: &mut SessionHandler,
channel: ChannelId,
session: &mut Session,
command: String,
) {
let Some(mut cmd) = handler.serve.exec_wrapper_argv.clone() else {
spawn_seam_stub(
channel,
session,
"exec_request",
Some("pipe"),
"wrapper-unconfigured",
);
return;
};
let Some(seam) = handler.serve.session_seam.clone() else {
spawn_seam_stub(
channel,
session,
"exec_request",
Some("pipe"),
"seam-unconfigured",
);
return;
};
let tty = handler.ptys.remove(&channel);
let Some(taken) = handler.take_stored_channel(channel) else {
observe::decline_request(
"exec_request",
"channel-not-held",
channel.number(),
Some("pipe"),
);
let _ = session.channel_failure(channel);
return;
};
cmd.push(command);
let request = tty_map::negotiate_request(
seam.backend.clone(),
tty,
cmd,
handler.serve.session_cwd.clone(),
handler.serve.session_env.clone(),
);
spawn_tty_row(
handler,
channel,
session,
"exec_request",
Some("pipe"),
taken,
request,
seam,
);
}
/// The session pty row's seam stub — wave 3 lands the alktty
/// `NegotiateRequest.tty` construction here
/// (`drive_session_pre_negotiated`), spawned per ADR-006 decision 4;
/// the skeleton declines at the seam, on its own task.
fn pty_row_stub(channel: ChannelId, session: &mut Session) {
spawn_seam_stub(channel, session, "pty_request", None, "seam-stub");
}
/// The session shell row's seam stub — wave 3 lands the alktty
/// `NegotiateRequest.cmd` = serve-config shell argv here (OQ-08, no
/// crate default), spawned per ADR-006 decision 4; the skeleton
/// declines at the seam, on its own task.
fn shell_row_stub(channel: ChannelId, session: &mut Session) {
spawn_seam_stub(channel, session, "shell_request", None, "seam-stub");
/// Installs the per-channel bridge surface, clones the session
/// handle, and spawns the spawned serving path — the session rows'
/// spawn step (ADR-006 decision 4: the handler returns fast; the
/// routed request answers `channel_success` inline).
#[cfg(feature = "session")]
#[allow(clippy::too_many_arguments)]
fn spawn_tty_row(
handler: &mut SessionHandler,
channel: ChannelId,
session: &mut Session,
row: &'static str,
exec_route: Option<&'static str>,
taken: Channel<Msg>,
request: NegotiateRequest,
seam: SessionSeam,
) {
let (surface, chunk_rx, cancel_rx) = tty_map::build_surface();
let stdin_tx = surface.sender.clone();
handler.bridges.insert(channel, surface);
let identity = handler.resolved_identity.clone();
let handle = session.handle();
serve::spawn_serving_path(tty_map::serve_tty_row(
row,
exec_route,
handle,
taken,
request,
identity,
seam,
stdin_tx,
chunk_rx,
cancel_rx,
handler.teardown.subscribe(),
));
let _ = session.channel_success(channel);
}
/// The tunnel row's seam stub: the explicit `ConnectFailed` reject
src/lib.rs
+2
View File
@@ -37,6 +37,8 @@ mod host_key;
mod key_source;
mod observe;
mod serve;
#[cfg(feature = "session")]
mod tty_map;
pub use adapter::SshAdapter;
#[cfg(feature = "git")]
src/observe.rs
+17 -4
View File
@@ -65,9 +65,15 @@
//! classifications: `"unfeatured"` (feature-gate absence),
//! `"never-feature"` (ADR-003 §5 rows), `"cap-exceeded"`,
//! `"connect-failed"` (the tunnel seam stub's `ConnectFailed`),
//! `"seam-stub"` (the session-row seam stubs),
//! `"seam-unconfigured"` (the config-driven git-row decline —
//! feature on, [`crate::GitSeam`] unset; `w3-git-mapping`),
//! `"seam-unconfigured"` (the config-driven row decline — the git
//! row and, now, the session rows: feature on, seam unset;
//! `w3-git-mapping` / `w3-tty-session`),
//! `"shell-unconfigured"` / `"wrapper-unconfigured"` (`w3-tty-
//! session`: the OQ-08 unset-field rows — feature on, the
//! serve-config field unset, never a fallback),
//! `"negotiation-failed"` (`w3-tty-session`: the alktty
//! negotiation error frame the bridge received — the session never
//! started; the frame's JSON detail stays alktty-side),
//! `"unknown-service"` / `"service-only"` / `"invalid-repo-id"`
//! (the exec grammar's git declinables — the parse core's two and
//! the git row's validation gate; no service token or repo string
@@ -210,8 +216,9 @@ pub(crate) fn decline_global_request(row: &'static str, reason: &'static str) {
/// A serving-path decline: a routed row whose serving task answers
/// failure-then-close rather than serving. This wave that is the seam
/// stub (the wire shape pinned in `tests/spawn_return.rs`); the wave-3
/// mappings emit it from their real serving tasks wherever serving
/// mappings emit it from their real spawned paths wherever serving
/// itself cannot proceed — the class the spawned paths inherit.
#[cfg(any(feature = "git", feature = "session"))]
pub(crate) fn decline_serving(
row: &'static str,
reason: &'static str,
@@ -232,6 +239,7 @@ pub(crate) fn decline_serving(
/// Spawned-path started — emitted as the spawned task's first action,
/// so "started" is the task's own honest start, not the hook's spawn
/// call. The mapping tasks' serving paths inherit this shape.
#[cfg(any(feature = "git", feature = "session"))]
pub(crate) fn serving_path_started(
row: &'static str,
channel_id: u32,
@@ -249,6 +257,7 @@ pub(crate) fn serving_path_started(
/// Spawned-path ended — the serving future finished (cleanly, or with
/// the decline it was wired to answer); emitted *after* the path's
/// wire reporting has completed.
#[cfg(any(feature = "git", feature = "session"))]
pub(crate) fn serving_path_ended(
row: &'static str,
channel_id: u32,
@@ -269,6 +278,7 @@ pub(crate) fn serving_path_ended(
/// the family's recoverable-failure level. Only russh protocol errors
/// ride the `error` field (§Safety posture); the mapping tasks emit
/// this from their real paths when serving fails before reporting.
#[cfg(any(feature = "git", feature = "session"))]
pub(crate) fn serving_path_failed<E: std::fmt::Debug>(
row: &'static str,
channel_id: u32,
@@ -292,7 +302,9 @@ mod tests {
/// A fake error shape — the failure row only pins that a Debug
/// error value rides the `error` field.
#[derive(Debug)]
#[cfg(any(feature = "git", feature = "session"))]
struct Probe;
#[cfg(any(feature = "git", feature = "session"))]
static PROBE: Probe = Probe;
use super::*;
@@ -420,6 +432,7 @@ mod tests {
/// - *Contract stamp:* the `exec_route` field — present on the
/// exec row, absent elsewhere (a `None` field is not recorded
/// by `tracing`). The pty/shell lifecycle rows carry no route.
#[cfg(any(feature = "git", feature = "session"))]
#[test]
fn serving_path_lifecycle_carries_started_ended_and_the_warn_failure() {
let fields = capture(|| serving_path_started("pty_request", 1, None));
src/tty_map.rs
+895
View File
@@ -0,0 +1,895 @@
//! The session mapping (`w3-tty-session`) — the biggest single-row
//! surface of the spine ([`crate::dispatch::ExecRoute::Pipe`] + the
//! `pty_request`/`shell_request` rows): pty/shell/pipe-exec serve
//! through the published alktty seam
//! (`drive_session_pre_negotiated(writer, reader, req, backends,
//! ownership, identity)`) over a **translation bridge, not a byte
//! pump** (POC-3 §D.1): alktty's data plane is chunk-framed both
//! directions (`alktty::wire` — 5-byte header + payload), the SSH
//! channel is a plain byte plane, so the bridge wraps the channel
//! with small adapters around a pair of in-memory duplexes.
//!
//! ## The negotiation is built locally, never wire-derived (OQ-08)
//!
//! The serve-config is the whole truth:
//!
//! - `pty_request` args → `alktty::negotiation::TerminalParamsWire`
//! (`term`/`cols`/`rows`/pixel dims; SSH ttymodes → the **reserved
//! modes JSON `{}`** — alktty OQ-44's reservation: backends must
//! ignore modes in v1, so the door writes `{}` and defers
//! faithfully). Stored per channel (the POC-3 accepted-but-held
//! shape, [`crate::handler`]'s store) and consumed by the later
//! `shell_request`/`exec_request` on the same channel
//! (`ssh -t host "cmd"` sends pty-then-exec).
//! - `shell_request` → `NegotiateRequest{carriage: "raw", backend:
//! the seam selector, tty: the consumed pty params, cmd: the
//! serve-config shell argv **verbatim**, cwd: session_cwd, env:
//! session_env}` — the wire never carries command, args, cwd or
//! env (OQ-08/OQ-09; the `env_request` row stays declined
//! unconditionally). **A `shell_request` without a preceding pty
//! serves pipe-mode (`tty = None`)** — decided: vanilla sshd serves
//! a shell without a terminal too (`ssh -T host` is legitimate
//! vanilla), so a pty-less shell request is served, never declined.
//! - `exec_request` (pipe row) → `cmd = exec_wrapper_argv ++ [cmd]`
//! (the OQ-08 argv append — vanilla quoting semantics; the wire
//! command string is **never naively split**), `tty =
//! Some(consumed pty)` when a pty preceded, else `None` (pipe
//! mode).
//! - **Unset-declines** (OQ-08, never a fallback): `shell_argv`
//! unset → `shell_request` declines; `exec_wrapper_argv` unset →
//! the pipe row declines. Wire shape (decided + recorded): the
//! **serving-path shape — `channel_failure` then channel close,
//! from a spawned task** ([`crate::serve`]'s mechanics; the same
//! spawned-stub body the git row uses) — the class of a routable
//! row that owns the request's whole outcome; the session channel
//! closes the way it would when the request itself cannot exist.
//! `session_seam` unset (feature on, seam absent) → the same
//! shape, reason `seam-unconfigured` (the decline matrix's
//! config-driven note); `pty_request` checks the seam too — a pty
//! whose session can never serve is a doomed allocation. Field
//! checks precede the seam check (the row's own unset field is the
//! first decline); the decline never takes the channel and never
//! consumes stored pty state.
//! - The unvanilla guards keep the git row's shapes: `channel-not-
//! held` (a routed row whose channel handle is already gone)
//! answers request-level failure inline; `no-identity` (post-auth
//! unreachable, fail-closed) is the spawned serving-path decline.
//!
//! ## The translation bridge
//!
//! Two small adapters around a duplex pair the serving task drives
//! (the POC-3 §D.1 shape, re-landed as contract):
//!
//! - **SSH → chunks** (the `client_recv` side of alktty's session):
//! a bounded [`tokio::sync::mpsc`] (the
//! [`CONTROL_SURFACE_BOUND`] bound) feeds a framing task that
//! writes `alktty::wire::Chunk`s into the duplex — the same mpsc
//! the control rows ride ([`BridgeSurface`], §below). SSH `data` →
//! `Chunk::stdin`; SSH channel EOF → **the zero-length
//! stdin-chunk sentinel** — decided (verified against alktty 0.8.1's
//! adapter pump, `adapter.rs::pump_client_to_backend`): a
//! zero-length stdin chunk shuts the backend's stdin, *identical
//! in effect* to `ControlMessage::Eof` on `STREAM_CTRL_IN`; the
//! sentinel rides the data path the SSH EOF already chose, so no
//! extra control frame is sent. Bounded end-to-end: the mpsc is
//! bounded, the duplexes bounded, and russh's per-channel window
//! bounds the SSH side.
//! - **Chunks → SSH**: the `client_send` side of alktty's session is
//! read through a **4-byte first-window disambiguation** (alktty's
//! own framing-disambiguation invariant, ADR-052 §5: a negotiation
//! *error frame* is length-prefixed JSON whose first length byte
//! is `0x00`; a raw chunk's first byte is a `stream_type` ≤ 4).
//! stdout chunks → SSH `data`; stderr chunks → SSH
//! `extended_data(1)` (PTY mode merges upstream — the backend's
//! stderr stream is `None` there, so the translation is universal:
//! no mode-switch exists); the exit chunk (`STREAM_CTRL_OUT`) →
//! the derivation contract (§below). No unbounded buffering in
//! either adapter (AGENTS.md convention 9).
//! - **The negotiation-error frame is the bridge's serving-path
//! decline class**: alktty re-runs validation + allocation inline
//! even on the pre-negotiated path (carriage/cmd/backend lookup/
//! ownership), and a failure answers a length-prefixed error
//! frame instead of chunks. Decided + recorded: the bridge maps
//! that frame to a serving-path decline event
//! (`negotiation-failed`) and an **`exit-status(255)` then close**
//! — the one place this door claims 255: the session **never
//! started** (no exit chunk exists, nothing ran, so 255 is the
//! honest could-not-start semantic — contrast the git row, whose
//! service *runs* its substrate and honestly reports `1`). The
//! error frame's JSON detail stays alktty-side; only the
//! vocabulary value rides the event.
//! - A bridge read that ends **without** an exit chunk reports
//! nothing (the chunk is the sole fidelity input; its absence is
//! neither 0 nor 255 — no fabricated report): plain close, and
//! the lifecycle's ended event.
//!
//! ## The control rows
//!
//! The requests arrive on the packet loop after the serving task
//! owns the wire, and reach the spawned bridge through the
//! **per-channel control surface** ([`BridgeSurface`] — handler
//! state while the row serves): a bounded mpsc whose send half
//! forward_control `try_send`s the control chunk (never awaited on
//! the packet loop). `window_change_request` →
//! `ControlMessage::Resize`; `signal` → `ControlMessage::Signal`
//! (alktty's uppercase-name vocabulary). Decisions, recorded:
//! the SSH control rows are fire-and-forget (no reply exists on
//! the wire), so **a full or closed surface drops the control
//! event** — the bound keeps a stalled bridge from stalling the
//! packet loop (bounded resources, convention 9); **signal names
//! are accepted verbatim** — no alkssh-side name filter: the
//! backend owns the rejection semantics (alktty passes the name to
//! the backend's control handle, whose semantics decide), and the
//! pipe-mode pid-only vs PTY process-group *scope* is backend
//! semantics (the mapping row note), so there is **no decline
//! shape** for a name — nothing on the wire can answer. A resize
//! arriving for a channel whose pty is stored but not yet serving
//! updates the stored params instead (the final size wins —
//! vanilla sshd's own ordering); a resize with neither a serving
//! bridge nor a stored pty is dropped.
//!
//! ## The exit derivation contract (mapping.md §Exit chunk, rules
//! 1–6 — normative)
//!
//! The alktty exit chunk (`{"type":"exit","code":i32}`,
//! exit-chunk-is-last per alktty's ADR-005 lineage) is the sole
//! fidelity input, **never widened**: [`derive_exit`] maps it to
//! exactly one SSH exit report shape:
//!
//! 1. `code ≥ 0` → `exit-status(N)` (rule 1).
//! 2. negative `−N`, N ≥ 2 → `exit-signal(name)` via the
//! crate-local **Unix signum→name table** ([`SIGNAL_NAMES`]).
//! Server-side work — the signal semantics belong to the serving
//! platform where the backend reaps. Deliberately *not* derived
//! from alktty's `signal_from_name` inverse, which is a forward
//! name→number map for the control row, not a derivation table.
//! `core_dumped` is **never asserted** — alktty discards the
//! core-dump bit entirely, so nothing better is knowable from the
//! chunk, and SSH's `exit-signal` bool stays false on every row.
//! 3. `−1` → `exit-status(255)`, never a signal claim (rule 3 — the
//! could-not-determine sentinel collides with the SIGHUP
//! encoding on the pipe path, accepted symmetrically per the
//! mapping: neither a fabricated signal for a wait failure nor a
//! real one).
//! 4. Non-Unix adds no other negatives (rule 4): the table above is
//! the only negative recognizer.
//! 5. **The PTY-path fidelity limit (rule 5, upstream)**:
//! portable-pty 0.9 reaps a signal-terminated Unix child as a
//! plain code (the signal number is unrecoverable through its
//! trait surface), so signal-terminated PTY sessions arrive as
//! `exit-status(1)` — a documented alktty/portable-pty ask; this
//! module's table in practice sees the *pipe* path's negative
//! encoding only. Faithful `exit-signal` there needs the alktty
//! backend to conform to its own ADR's negative-code convention;
//! alkssh's contract adapts automatically.
//! 6. Unmapped negatives → rule 3 (`exit-status(255)`) — the
//! mapping never invents a name (defensive case: the
//! convention's producers are `−1` and `−signum` only).
//!
//! The report is sent **before** close, from the spawned serving
//! task (ADR-006 decision 4, via the cloned `Session::handle()`).
//!
//! ## Cancellation (bounded resources + alktty's kill-on-drop)
//!
//! The serving task watches two signals: the **per-channel cancel**
//! (fired by `channel_close`, or surfaced by the surface's drop) and
//! the **connection teardown** ([`Teardown`] — fired by the
//! handler's `Drop`; the russh session ended, however it ended). On
//! either, the task aborts the alktty session task — dropping its
//! not-yet-resolved `exit_code` future, which is alktty's
//! kill-on-drop contract firing (its ADR-056) — then closes the SSH
//! channel and ends. Nothing is fabricated on a cancel. The
//! detached stdin/framing tasks are bounded by their duplex/mpsc/
//! russh-window bounds and end on the same signals' effects.
use alkcall::core::auth::Identity;
use alktty::ControlMessage;
use alktty::negotiation::{NegotiateRequest, TerminalParamsWire};
use alktty::wire::{
Chunk, ChunkWriter, MAX_CHUNK_LEN, STREAM_CTRL_IN, STREAM_CTRL_OUT, STREAM_STDERR,
STREAM_STDIN, STREAM_STDOUT,
};
use bytes::Bytes;
use russh::ChannelId;
use russh::server::Handle;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::{mpsc, watch};
use crate::config::SessionSeam;
use crate::observe;
/// The negotiation frame the bridge speaks locally — every locally
/// constructed request reads `"raw"` (alktty's v1 carriage).
pub(crate) const CARRIAGE: &str = "raw";
/// The per-channel control surface's bound: control rows are rare
/// (resize on window drag, signal), so the bound never bites vanilla
/// traffic while keeping the forwarding surface bounded
/// (§the control rows). The shim's scripted rows exercise a handful.
pub(crate) const CONTROL_SURFACE_BOUND: usize = 64;
/// The translation bridge duplex capacity — the wave-2 loopback
/// harness's per-direction convention, kept for the bridge pairs.
const BRIDGE_DUPLEX_CAPACITY: usize = 65_536;
/// The stdin pump's read buffer — the SSH-side chunk granularity
/// (per-chunk payloads are bounded in practice by russh's
/// `maximum_packet_size`, here by [`MAX_CHUNK_LEN`] on the frame).
const STDIN_PUMP_BUF: usize = 16 * 1024;
/// One SSH exit report shape derived from an exit chunk —
/// [`derive_exit`]'s outcome.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ExitReport {
/// `exit-status(N)` — codes ≥ 0, the `−1` sentinel, and every
/// unmapped negative (rules 1, 3, 4, 6).
Status(u32),
/// `exit-signal(name)` — a mapped negative `−N`, N ≥ 2 (rule 2).
/// The name is the table's value, never an invention.
Signal(&'static str),
}
/// The crate-local Unix signum→name table (rule 2): the 1–31
/// conventional Unix signals, as the exit chunk's producers encode
/// them (`−signum`). Server-side work — the serving platform owns
/// the signal semantics where the backend reaps. Deliberately **not**
/// derived from alktty's `signal_from_name` (a forward name→number
/// map for the control row, not a derivation table; deliberately
/// narrower too — it exists for what a front-end sends, not for what
/// a child can die by).
pub(crate) const SIGNAL_NAMES: &[(u32, &str)] = &[
(1, "HUP"),
(2, "INT"),
(3, "QUIT"),
(4, "ILL"),
(5, "TRAP"),
(6, "ABRT"),
(7, "BUS"),
(8, "FPE"),
(9, "KILL"),
(10, "USR1"),
(11, "SEGV"),
(12, "USR2"),
(13, "PIPE"),
(14, "ALRM"),
(15, "TERM"),
(16, "STKFLT"),
(17, "CHLD"),
(18, "CONT"),
(19, "STOP"),
(20, "TSTP"),
(21, "TTIN"),
(22, "TTOU"),
(23, "URG"),
(24, "XCPU"),
(25, "XFSZ"),
(26, "VTALRM"),
(27, "PROF"),
(28, "WINCH"),
(29, "IO"),
(30, "PWR"),
(31, "SYS"),
];
/// The derivation contract's whole decision procedure (§module docs,
/// mapping.md rules 1–6 — normative).
pub(crate) fn derive_exit(code: i32) -> ExitReport {
match code {
code if code >= 0 => ExitReport::Status(code as u32),
-1 => ExitReport::Status(255),
other => match other.checked_abs() {
Some(signum) => match SIGNAL_NAMES.iter().find(|(n, _)| *n == signum as u32) {
Some((_, name)) => ExitReport::Signal(name),
None => ExitReport::Status(255),
},
None => ExitReport::Status(255),
},
}
}
/// The table's name → russh's `Sig` vocabulary (the
/// `exit_signal_request` parameter). russh's known set rides its
/// variant names; anything else rides `Sig::Custom` (RFC 4254 §6.10's
/// signal-name encoding — vanilla clients parse arbitrary names).
pub(crate) fn sig_from_name(name: &str) -> russh::Sig {
match name {
"ABRT" => russh::Sig::ABRT,
"ALRM" => russh::Sig::ALRM,
"FPE" => russh::Sig::FPE,
"HUP" => russh::Sig::HUP,
"ILL" => russh::Sig::ILL,
"INT" => russh::Sig::INT,
"KILL" => russh::Sig::KILL,
"PIPE" => russh::Sig::PIPE,
"QUIT" => russh::Sig::QUIT,
"SEGV" => russh::Sig::SEGV,
"TERM" => russh::Sig::TERM,
"USR1" => russh::Sig::USR1,
other => russh::Sig::Custom(other.to_owned()),
}
}
/// The SSH signal row's name translation: russh's `Sig` → the name
/// string alktty's control vocabulary carries (the uppercase names;
/// `Custom` carrying its own). Names flow to the backend verbatim —
/// no alkssh-side filter (§the control rows).
pub(crate) fn sig_control_name(sig: &russh::Sig) -> String {
match sig {
russh::Sig::ABRT => "ABRT".to_owned(),
russh::Sig::ALRM => "ALRM".to_owned(),
russh::Sig::FPE => "FPE".to_owned(),
russh::Sig::HUP => "HUP".to_owned(),
russh::Sig::ILL => "ILL".to_owned(),
russh::Sig::INT => "INT".to_owned(),
russh::Sig::KILL => "KILL".to_owned(),
russh::Sig::PIPE => "PIPE".to_owned(),
russh::Sig::QUIT => "QUIT".to_owned(),
russh::Sig::SEGV => "SEGV".to_owned(),
russh::Sig::TERM => "TERM".to_owned(),
russh::Sig::USR1 => "USR1".to_owned(),
russh::Sig::Custom(name) => name.clone(),
}
}
/// The session dimension clamp: u32 wire dims → the wire's u16 (a
/// dimension beyond 65535 is nonsense on both papers; the cap, not
/// a truncation of useful data).
pub(crate) fn clamp_u16(value: u32) -> u16 {
value.min(u16::MAX as u32) as u16
}
/// The `pty_request` args → `TerminalParamsWire` conversion (§the
/// negotiation section): SSH ttymodes → the **reserved modes JSON
/// `{}`** (alktty OQ-44's reservation — the map is honest: nothing
/// is carried, exactly what the reservation asks).
pub(crate) fn terminal_params(
term: &str,
cols: u32,
rows: u32,
pix_width: u32,
pix_height: u32,
) -> TerminalParamsWire {
TerminalParamsWire {
term: Some(term.to_owned()),
cols: clamp_u16(cols),
rows: clamp_u16(rows),
pixel_width: clamp_u16(pix_width),
pixel_height: clamp_u16(pix_height),
modes: serde_json::json!({}),
}
}
/// Builds the locally-constructed `NegotiateRequest` (§the
/// negotiation section): the serve-config's values verbatim; the
/// reserved backend_params empty (the seam selector is the `backend`
/// string; no backend-specific selector fields exist on this door).
pub(crate) fn negotiate_request(
seam_backend: String,
tty: Option<TerminalParamsWire>,
cmd: Vec<String>,
cwd: Option<std::path::PathBuf>,
env: std::collections::BTreeMap<String, String>,
) -> NegotiateRequest {
NegotiateRequest {
carriage: CARRIAGE.to_owned(),
backend: seam_backend,
tty,
cmd,
cwd,
env: env.into_iter().collect(),
backend_params: Default::default(),
}
}
/// One parsed bridge-side frame from alktty's `client_send` half —
/// the 4-byte first-window disambiguation's outcome (§the
/// translation bridge).
enum BridgeFrame {
/// A raw alktty chunk (data or server→client control).
Chunk(Chunk),
/// The length-prefixed negotiation error frame — the session
/// failed validation/allocation/ownership at the seam. The bytes
/// are alktty-side detail (never read here — only the
/// vocabulary value rides the decline event).
NegotiationError,
/// The half ended (alktty's task completed or was cancelled)
/// with no exit chunk having been observed.
Closed,
/// The half broke (a non-EOF I/O error, or a rogue frame).
Break,
}
/// Reads one bridge frame — the disambiguation invariant: the first
/// byte `0x00` is a negotiation error frame's length-prefix high
/// byte; anything ≤ 4 is a chunk `stream_type`. Bounds-checked
/// against [`MAX_CHUNK_LEN`] both directions — a rogue length never
/// triggers an oversized read.
async fn read_bridge_frame(reader: &mut (impl tokio::io::AsyncRead + Unpin)) -> BridgeFrame {
let mut head = [0u8; 4];
match reader.read_exact(&mut head).await {
Ok(_) => {}
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return BridgeFrame::Closed,
Err(_) => return BridgeFrame::Break,
}
if head[0] == 0x00 {
let length = u32::from_be_bytes([0, head[1], head[2], head[3]]);
if length as usize > MAX_CHUNK_LEN as usize {
return BridgeFrame::Break;
}
let mut body = vec![0u8; length as usize];
return match reader.read_exact(&mut body).await {
Ok(_) => BridgeFrame::NegotiationError,
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => BridgeFrame::Closed,
Err(_) => BridgeFrame::Break,
};
}
let stream_type = head[0];
let mut last = [0u8; 1];
match reader.read_exact(&mut last).await {
Ok(_) => {}
Err(_) => return BridgeFrame::Break,
}
let length = u32::from_be_bytes([head[1], head[2], head[3], last[0]]);
if length as usize > MAX_CHUNK_LEN as usize {
return BridgeFrame::Break;
}
let mut body = vec![0u8; length as usize];
match reader.read_exact(&mut body).await {
Ok(_) => BridgeFrame::Chunk(Chunk {
stream_type,
bytes: Bytes::from(body),
}),
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => BridgeFrame::Closed,
Err(_) => BridgeFrame::Break,
}
}
/// The exit chunk's JSON payload → the derived code
/// (`{"type":"exit","code":N}`). Anything else on `STREAM_CTRL_OUT`
/// has no contract meaning — and an exit chunk that cannot parse has
/// no fidelity input — both map to `None` (no report; the plain
/// close posture).
fn exit_chunk_code(payload: &[u8]) -> Option<i32> {
ControlMessage::from_slice(payload)
.ok()
.and_then(|msg| match msg {
ControlMessage::Exit { code } => Some(code),
_ => None,
})
}
/// The per-channel bridge surface — handler state while a session row
/// serves (§the control rows). The handler's `window_change_request`
/// / `signal` hooks reach the spawned bridge **through this struct**
/// (the requests arrive on the packet loop after the serving task
/// owns the wire): the bounded control-surface send half, plus the
/// per-channel cancel signal `channel_close` fires. Packet-loop only
/// (single-writer; no lock).
pub(crate) struct BridgeSurface {
/// The control-surface send half — the same bounded mpsc the
/// serving task's stdin pump feeds.
pub(crate) sender: mpsc::Sender<Chunk>,
/// The per-channel cancel signal — `channel_close` fires it. The
/// serving task interprets the fire *or* the half's drop as the
/// cancel.
pub(crate) cancel: watch::Sender<bool>,
}
impl BridgeSurface {
/// Forwards one control row into the bridge — `try_send`, never
/// awaited on the packet loop (§the control rows); a full or
/// closed surface drops the event.
pub(crate) fn forward_control(&self, message: &ControlMessage) -> bool {
let Ok(payload) = message.to_json() else {
return false;
};
self.sender
.try_send(Chunk {
stream_type: STREAM_CTRL_IN,
bytes: payload,
})
.is_ok()
}
}
/// Builds the pair the dispatch hands into the serving task: the
/// surface the handler stores and its consumed receiver counterpart
/// (the bridge's framed-input mpsc) plus the cancel watch halves.
/// The surface's `sender` clone is the stdin pump's producer inside
/// the serving path.
pub(crate) fn build_surface() -> (BridgeSurface, mpsc::Receiver<Chunk>, watch::Receiver<bool>) {
let (sender, receiver) = mpsc::channel(CONTROL_SURFACE_BOUND);
let (cancel, cancel_rx) = watch::channel(false);
(BridgeSurface { sender, cancel }, receiver, cancel_rx)
}
/// Fires the cancel watch (the `channel_close` hook's action).
pub(crate) fn fire_cancel(cancel: &watch::Sender<bool>) {
let _ = cancel.send(true);
}
/// The connection-teardown signal: fired by the handler's `Drop`
/// (the russh session ended, however it ended). Subscribe before the
/// serving task spawns; the receiver also reads the sender's drop as
/// the cancel.
pub(crate) struct Teardown {
sender: watch::Sender<bool>,
}
impl Teardown {
pub(crate) fn new() -> Self {
Self {
sender: watch::Sender::new(false),
}
}
pub(crate) fn subscribe(&self) -> watch::Receiver<bool> {
self.sender.subscribe()
}
pub(crate) fn fire(&mut self) {
let _ = self.sender.send(true);
}
}
/// Reads the SSH half of the channel and feeds stdin chunks — the
/// SSH `data` → `Chunk::stdin` adapter (§the translation bridge).
/// On EOF or a read error it sends **the zero-length stdin-chunk
/// sentinel** (the EOF translation decision: identical in effect to
/// alktty's `ControlMessage::Eof` — the backend's stdin shuts while
/// the session keeps pumping) and ends.
async fn stdin_pump(
ssh_read: impl tokio::io::AsyncRead + Unpin + 'static,
chunks: mpsc::Sender<Chunk>,
) {
let mut read_half = ssh_read;
let mut buf = vec![0u8; STDIN_PUMP_BUF];
loop {
match read_half.read(&mut buf).await {
Ok(0) => {
let _ = chunks
.send(Chunk {
stream_type: STREAM_STDIN,
bytes: Bytes::new(),
})
.await;
return;
}
Ok(n) => {
let chunk = Chunk {
stream_type: STREAM_STDIN,
bytes: Bytes::copy_from_slice(&buf[..n]),
};
if chunks.send(chunk).await.is_err() {
return;
}
}
Err(_) => {
let _ = chunks
.send(Chunk {
stream_type: STREAM_STDIN,
bytes: Bytes::new(),
})
.await;
return;
}
}
}
}
/// Consumes the framed-input mpsc and writes alktty chunks into the
/// session's `client_recv` duplex half — the framing task. Ends after
/// the EOF sentinel is written (nothing rides the data path past
/// SSH EOF), on a mpsc close (every sender gone — the cancelled or
/// completed session), or on a framing failure.
async fn framing_task(mut chunks: mpsc::Receiver<Chunk>, writer: tokio::io::DuplexStream) {
let mut writer = ChunkWriter::new(writer);
while let Some(chunk) = chunks.recv().await {
if writer.write_chunk(&chunk).await.is_err() {
return;
}
if chunk.stream_type == STREAM_STDIN && chunk.bytes.is_empty() {
return;
}
}
}
/// The exit report + close + lifecycle event, in the ADR-006
/// decision-4 order: **status before close, from the spawned task**.
/// A failed send means the session is gone — the lifecycle's one
/// `warn` records it (the report that could not land).
async fn report_exit(
handle: &Handle,
row: &'static str,
exec_route: Option<&'static str>,
channel_id: ChannelId,
channel_number: u32,
report: ExitReport,
) {
let error = match report {
ExitReport::Status(code) => handle.exit_status_request(channel_id, code).await.err(),
ExitReport::Signal(name) => handle
.exit_signal_request(
channel_id,
sig_from_name(name),
false,
String::new(),
"en".to_owned(),
)
.await
.err(),
};
let error = match error {
Some(error) => Some(error),
None => handle.close(channel_id).await.err(),
};
match error {
Some(error) => observe::serving_path_failed(row, channel_number, &error, exec_route),
None => observe::serving_path_ended(row, channel_number, exec_route),
}
}
/// Fails the request and closes the channel from the serving path —
/// the serving-path decline shape ([`crate::serve`]'s mechanics) with
/// the lifecycle events and the given reason. Used for `no-identity`.
async fn fail_and_close(
handle: &Handle,
row: &'static str,
exec_route: Option<&'static str>,
channel_id: ChannelId,
channel_number: u32,
reason: &'static str,
) {
observe::decline_serving(row, reason, channel_number, exec_route);
let error = match handle.channel_failure(channel_id).await {
Ok(()) => handle.close(channel_id).await.err(),
Err(error) => Some(error),
};
match error {
Some(error) => observe::serving_path_failed(row, channel_number, &error, exec_route),
None => observe::serving_path_ended(row, channel_number, exec_route),
}
}
/// The spawned session-row serving path (§module docs): the store's
/// taken channel rides the translation bridge into
/// [`alktty::drive_session_pre_negotiated`]; the alktty session task
/// and the SSH→chunks pump are spawned here, and *this* future drives
/// the chunks→SSH direction inline so the exit report is the path's
/// own final act (status before close — ADR-006 decision 4). The
/// channel was taken by the dispatch; the spawned task owns the
/// stream and its report surface.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn serve_tty_row(
row: &'static str,
exec_route: Option<&'static str>,
handle: Handle,
channel: russh::Channel<russh::server::Msg>,
request: NegotiateRequest,
identity: Option<Identity>,
seam: SessionSeam,
stdin_tx: mpsc::Sender<Chunk>,
chunk_rx: mpsc::Receiver<Chunk>,
cancel_rx: watch::Receiver<bool>,
teardown_rx: watch::Receiver<bool>,
) {
let channel_id = channel.id();
let channel_number = channel_id.number();
observe::serving_path_started(row, channel_number, exec_route);
let Some(identity) = identity else {
fail_and_close(
&handle,
row,
exec_route,
channel_id,
channel_number,
"no-identity",
)
.await;
return;
};
let (ssh_read, mut ssh_write) = tokio::io::split(channel.into_stream());
let (chunk_in, chunk_in_b) = tokio::io::duplex(BRIDGE_DUPLEX_CAPACITY);
let (chunk_out, chunk_out_b) = tokio::io::duplex(BRIDGE_DUPLEX_CAPACITY);
let mut chunk_out = chunk_out;
tokio::spawn(framing_task(chunk_rx, chunk_in));
tokio::spawn(stdin_pump(ssh_read, stdin_tx));
let alktty_session = tokio::spawn(alktty::drive_session_pre_negotiated(
chunk_out_b,
chunk_in_b,
request,
seam.backends,
seam.ownership,
Some(identity),
));
drive_chunks_to_ssh(
&handle,
row,
exec_route,
channel_id,
channel_number,
&mut ssh_write,
&mut chunk_out,
cancel_rx,
teardown_rx,
alktty_session,
)
.await;
}
/// The chunks→SSH direction, driven inline by the serving path
/// (§the translation bridge): data/`extended_data(1)`/exit-chunk
/// translation, the negotiation-error-frame decline class, and the
/// two cancel signals. Each terminal path ends the lifecycle exactly
/// once.
#[allow(clippy::too_many_arguments)]
async fn drive_chunks_to_ssh(
handle: &Handle,
row: &'static str,
exec_route: Option<&'static str>,
channel_id: ChannelId,
channel_number: u32,
ssh_write: &mut (impl tokio::io::AsyncWrite + Unpin),
chunk_out: &mut tokio::io::DuplexStream,
cancel_rx: watch::Receiver<bool>,
teardown_rx: watch::Receiver<bool>,
alktty_session: tokio::task::JoinHandle<()>,
) {
let mut cancel_rx = cancel_rx;
let mut teardown_rx = teardown_rx;
loop {
tokio::select! {
frame = read_bridge_frame(chunk_out) => {
match frame {
BridgeFrame::Chunk(chunk) => {
match chunk.stream_type {
STREAM_STDOUT => {
if ssh_write.write_all(&chunk.bytes).await.is_err() {
break;
}
}
STREAM_STDERR => {
let _ = handle
.extended_data(channel_id, 1, chunk.bytes.to_vec())
.await;
}
STREAM_CTRL_OUT => {
match exit_chunk_code(&chunk.bytes).map(derive_exit) {
Some(report) => {
report_exit(
handle,
row,
exec_route,
channel_id,
channel_number,
report,
)
.await;
}
None => {
let _ = handle.close(channel_id).await;
observe::serving_path_ended(
row,
channel_number,
exec_route,
);
}
}
return;
}
_ => {}
}
}
BridgeFrame::NegotiationError => {
observe::decline_serving(
row,
"negotiation-failed",
channel_number,
exec_route,
);
let _ = handle.exit_status_request(channel_id, 255).await;
let _ = handle.close(channel_id).await;
observe::serving_path_ended(row, channel_number, exec_route);
return;
}
BridgeFrame::Closed | BridgeFrame::Break => {
let _ = handle.close(channel_id).await;
observe::serving_path_ended(row, channel_number, exec_route);
return;
}
}
}
_ = cancel_rx.changed() => break,
_ = teardown_rx.changed() => break,
}
}
alktty_session.abort();
let _ = handle.close(channel_id).await;
observe::serving_path_ended(row, channel_number, exec_route);
}
#[cfg(test)]
mod tests {
use super::{ExitReport, SIGNAL_NAMES, derive_exit, sig_control_name, sig_from_name};
use russh::Sig;
/// - *Contract stamp:* mapping.md §Exit chunk rule 1 (`code ≥ 0`
/// → `exit-status(N)` — the chunk carries the whole truth for
/// clean exits).
#[test]
fn non_negative_codes_map_to_exit_status() {
assert_eq!(derive_exit(0), ExitReport::Status(0));
assert_eq!(derive_exit(1), ExitReport::Status(1));
assert_eq!(derive_exit(42), ExitReport::Status(42));
assert_eq!(derive_exit(255), ExitReport::Status(255));
}
/// - *Contract stamp:* mapping.md §Exit chunk rule 2 (negative
/// `−N`, N ≥ 2 → `exit-signal(name)` via the table — the POC-3
/// §D.2 live evidence: `−15` → `exit-signal("TERM")`).
#[test]
fn negative_signals_map_through_the_table() {
for (signum, name) in SIGNAL_NAMES.iter().skip(1) {
assert_eq!(derive_exit(-(*signum as i32)), ExitReport::Signal(name));
}
assert_eq!(derive_exit(-15), ExitReport::Signal("TERM"));
assert_eq!(derive_exit(-2), ExitReport::Signal("INT"));
}
/// - *Contract stamp:* mapping.md §Exit chunk rule 3 (`−1` →
/// `exit-status(255)`, never a signal claim — the sentinel/
/// SIGHUP collision accepted symmetrically).
#[test]
fn the_sentinel_is_not_a_signal() {
assert_eq!(derive_exit(-1), ExitReport::Status(255));
}
/// - *Contract stamp:* mapping.md §Exit chunk rule 6 (unmapped
/// negatives → rule 3 — the mapping never invents a name).
#[test]
fn unmapped_negatives_report_255() {
for code in [i32::MIN, -33, -32, -100, -1_000_000] {
assert_eq!(derive_exit(code), ExitReport::Status(255));
}
}
/// - *Contract stamp:* the table bound — every entry is the
/// conventional Unix signal set 1–31, sorted, names unique and
/// uppercase (the vocabulary the derivation table promises).
#[test]
fn the_table_is_the_conventional_unix_set() {
for (n, name) in SIGNAL_NAMES.iter() {
assert!((1..=31).contains(n), "signum {n} outside 1..=31");
assert!(
name.chars()
.all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()),
"signal names are uppercase-and-digit vocabulary: {name}"
);
}
let mut nums: Vec<u32> = SIGNAL_NAMES.iter().map(|(n, _)| *n).collect();
nums.sort_unstable();
nums.dedup();
assert_eq!(nums.len(), SIGNAL_NAMES.len(), "no duplicate signa");
assert_eq!(nums.first(), Some(&1));
assert_eq!(nums.last(), Some(&31));
}
/// The name→`Sig` translation: russh's known set rides variant
/// names; everything else rides `Sig::Custom` (the RFC's
/// signal-name encoding).
#[test]
fn sig_from_name_covers_known_and_custom() {
assert!(matches!(sig_from_name("TERM"), Sig::TERM));
assert!(matches!(sig_from_name("USR1"), Sig::USR1));
assert!(matches!(sig_from_name("USR2"), Sig::Custom(_)));
assert!(matches!(sig_from_name("TSTP"), Sig::Custom(_)));
assert_eq!(sig_control_name(&Sig::TERM), "TERM");
assert_eq!(sig_control_name(&Sig::USR1), "USR1");
assert_eq!(sig_control_name(&Sig::Custom("ABRT".to_owned())), "ABRT");
}
}