phase 2: channels integration + TtySession consumer client
New code (not a port) — the producer/consumer halves of the channels integration per alkcall's protocol-crate pattern. Producer half (src/channels.rs): - register_openable helper: builds the OperationSpec for channels/tty/sub (Sub-typed, channel_open marker for alk/tty, AccessControl with tty:open scope gate) and calls ChannelCore::register_openable - TtyOpenHandler factory: receives the channel Connection, calls accept_bi(), runs drive_session on the BiStream — same code path as the direct-ALPN TtyAdapter (ADR-093) - 8 tests: spec shape, registration, end-to-end open op returns channel_id, ACL denial without tty:open scope Consumer half (src/session.rs): - TtySession::connect_direct(connection, negotiate) — direct alk/tty - TtySession::open_via_channels(client, params) — opens a channel via ChannelClient::open_channel, builds a Connection from the reassembled halves, runs the same negotiation + typed-methods flow - Methods: send_stdin, close_stdin, resize, signal, recv_stdout, recv_stderr, wait - Read pump: spawns a task that reads chunks off the BiStream and routes stdout/stderr to mpsc channels, parses Exit control chunks and resolves a watch channel for wait() - Drop aborts the read pump - 6 tests: negotiation frame write, stdin round-trip, resize/signal, stdout stream, NoExitChunk on early close, broken-stream error Also: added Serialize to NegotiateRequest + TerminalParamsWire (was Deserialize-only; the session client needs to serialize the negotiation frame). 80/80 tests pass, wasm32-unknown-unknown clean, clippy clean.
This commit is contained in:
+517
@@ -0,0 +1,517 @@
|
|||||||
|
//! Producer half of the channels integration — the `register_openable`
|
||||||
|
//! helper and the `TtyOpenHandler` factory (ADR-047 §3).
|
||||||
|
//!
|
||||||
|
//! This is the path where TTY rides inside `alk/channels` rather than
|
||||||
|
//! directly on a `alk/tty` ALPN connection. The assembly layer (an
|
||||||
|
//! `alk/channels` acceptor's `install_channel_zero` hook) builds a
|
||||||
|
//! per-connection [`ChannelCore`], then calls
|
||||||
|
//! [`register_openable`] to register the `channels/tty/sub` open op
|
||||||
|
//! on the per-connection overlay registry. When a consumer (a
|
||||||
|
//! [`crate::session::TtySession`] or any `ChannelClient` caller)
|
||||||
|
//! invokes `channels/tty/sub`, the alkcall channels wrapper does
|
||||||
|
//! `check_open` → `open_channel` → spawn the [`TtyOpenHandler`] →
|
||||||
|
//! respond with `{ channel_id }`. The `TtyOpenHandler` receives the
|
||||||
|
//! channel's `Connection` (data-plane ALPN `alk/tty`), calls
|
||||||
|
//! `accept_bi()` to get the channel's `BiStream`, and runs
|
||||||
|
//! [`crate::adapter::drive_session`] on it — the same code path the
|
||||||
|
//! direct-ALPN `TtyAdapter` uses (ADR-093: TTY always uses its 5-byte
|
||||||
|
//! format; the channels layer strips its 8-byte header and hands TTY
|
||||||
|
//! the payload transparently).
|
||||||
|
//!
|
||||||
|
//! The access control (scope-gate + ownership) is wired into the
|
||||||
|
//! `OperationSpec`'s `AccessControl` and enforced by the registry's
|
||||||
|
//! `invoke`/`invoke_streaming` before the `TtyOpenHandler` runs — the
|
||||||
|
//! handler itself only validates that the negotiated backend exists
|
||||||
|
//! and spawns the protocol. This is the difference from the
|
||||||
|
//! direct-ALPN `TtyAdapter`, which does its own ad-hoc scope check;
|
||||||
|
//! the channels path gets ACL for free from alkcall's registry.
|
||||||
|
//!
|
||||||
|
//! [`ChannelCore`]: alkcall::channels::operations::ChannelCore
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use alkcall::channels::operations::{ChannelCore, OpenHandler};
|
||||||
|
use alkcall::core::auth::AuthContext;
|
||||||
|
use alkcall::core::ownership::OwnershipProvider;
|
||||||
|
use alkcall::core::Connection;
|
||||||
|
use alkcall::registry::spec::{
|
||||||
|
AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility,
|
||||||
|
};
|
||||||
|
use serde_json::{json, Value};
|
||||||
|
use tracing::debug;
|
||||||
|
|
||||||
|
use crate::adapter::{drive_session, TTY_OPEN_SCOPE};
|
||||||
|
use crate::backend::TtyBackend;
|
||||||
|
|
||||||
|
/// The per-ALPN open operation name (`channels/<alpn>/sub` convention,
|
||||||
|
/// ADR-047). TTY is consumer-opens (the client requests a shell), so
|
||||||
|
/// the op is `sub` (subscribe), not `pub` (publish).
|
||||||
|
pub const OP_TTY_OPEN: &str = "channels/tty/sub";
|
||||||
|
|
||||||
|
/// The data-plane ALPN the channel carries. Matches the constant in
|
||||||
|
/// [`crate::adapter::TtyAdapter::alpn`].
|
||||||
|
pub const TTY_ALPN: &str = "alk/tty";
|
||||||
|
|
||||||
|
/// Register the `channels/tty/sub` open op on a per-connection
|
||||||
|
/// `OperationRegistry` (ADR-047 §3, as amended 2026-08-13 —
|
||||||
|
/// per-connection registration).
|
||||||
|
///
|
||||||
|
/// The assembly layer (an `alk/channels` acceptor's
|
||||||
|
/// `install_channel_zero` hook) builds a [`ChannelCore`] from the
|
||||||
|
/// adapter-supplied `ChannelManager` + `ChannelLifecyclePolicy`,
|
||||||
|
/// constructs a fresh per-connection `OperationRegistry`, and calls
|
||||||
|
/// this helper. The helper builds the [`OperationSpec`] for
|
||||||
|
/// `channels/tty/sub` with the `channel_open` marker set
|
||||||
|
/// (`ChannelOpenSpec::new("alk/tty")`), an `AccessControl` carrying
|
||||||
|
/// the `tty:open` scope gate, and a permissive input schema (the
|
||||||
|
/// `NegotiateRequest` shape — JSON, validated by the
|
||||||
|
/// `drive_session` negotiation reader). It then wraps the
|
||||||
|
/// [`TtyOpenHandler`] factory and calls
|
||||||
|
/// [`ChannelCore::register_openable`].
|
||||||
|
///
|
||||||
|
/// `backends` is the same backend map the direct-ALPN `TtyAdapter`
|
||||||
|
/// holds — `HashMap<String, Arc<dyn TtyBackend>>` keyed by the
|
||||||
|
/// negotiation frame's `backend` string. The assembly layer typically
|
||||||
|
/// shares one backend map between the direct-ALPN adapter and the
|
||||||
|
/// channels `register_openable` registration.
|
||||||
|
///
|
||||||
|
/// `ownership` is the optional `OwnershipProvider` for the ADR-050
|
||||||
|
/// resource-ownership check. `None` = scope-gate only. The provider
|
||||||
|
/// is consulted inside `drive_session` (after the backend is selected
|
||||||
|
/// and `resource_id` is extracted from `backend_params`), not by the
|
||||||
|
/// channels wrapper — the wrapper's `AccessControl` carries only the
|
||||||
|
/// scope gate. This mirrors the direct-ALPN `TtyAdapter` shape: the
|
||||||
|
/// scope gate is enforced once at the entry point, the ownership
|
||||||
|
/// check is enforced once at the backend-selection point.
|
||||||
|
///
|
||||||
|
/// `auth` is the peer's `AuthContext`, captured at
|
||||||
|
/// `install_channel_zero` time and closed over by the wrapper so the
|
||||||
|
/// `TtyOpenHandler` receives it without the wrapper having to reach
|
||||||
|
/// into `OperationContext` for it (per the ADR-047 §4 amendment).
|
||||||
|
///
|
||||||
|
/// # Errors
|
||||||
|
///
|
||||||
|
/// Returns an error if `ChannelCore::register_openable` rejects the
|
||||||
|
/// spec (e.g., the spec has no `channel_open` marker — won't happen
|
||||||
|
/// with the spec this helper builds, but the underlying call is
|
||||||
|
/// fallible).
|
||||||
|
pub fn register_openable(
|
||||||
|
core: &ChannelCore,
|
||||||
|
backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
|
||||||
|
ownership: Option<Arc<dyn OwnershipProvider>>,
|
||||||
|
registry: &mut alkcall::registry::registration::OperationRegistry,
|
||||||
|
auth: AuthContext,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
let spec = tty_open_spec();
|
||||||
|
let open_handler = make_tty_open_handler(backends, ownership, auth.identity.clone());
|
||||||
|
core.register_openable(spec, open_handler, registry, auth)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build the [`OperationSpec`] for `channels/tty/sub`.
|
||||||
|
///
|
||||||
|
/// The op is `Sub`-typed (consumer-opens — the initiator opens a
|
||||||
|
/// channel and the data plane flows on the channel's `BiStream`, not
|
||||||
|
/// in the call response stream). The `channel_open` marker is set to
|
||||||
|
/// `ChannelOpenSpec::new("alk/tty")` so the channels wrapper allocates
|
||||||
|
/// a data channel. The `AccessControl` carries the `tty:open` scope
|
||||||
|
/// gate (the same scope the direct-ALPN `TtyAdapter` checks).
|
||||||
|
///
|
||||||
|
/// The input schema is permissive (`type: object`) — the
|
||||||
|
/// `NegotiateRequest` JSON shape is validated by `drive_session`'s
|
||||||
|
/// negotiation reader, not by the registry's schema validator. This
|
||||||
|
/// keeps the wire-format definition in one place (`wire.rs` +
|
||||||
|
/// `negotiation.rs`) and avoids duplicating the schema in the
|
||||||
|
/// `OperationSpec`. A future tightening could add the full
|
||||||
|
/// `NegotiateRequest` JSON schema here; the permissive shape is the
|
||||||
|
/// starting point.
|
||||||
|
pub fn tty_open_spec() -> OperationSpec {
|
||||||
|
OperationSpec::new(
|
||||||
|
OP_TTY_OPEN,
|
||||||
|
OperationType::Sub,
|
||||||
|
Visibility::External,
|
||||||
|
json!({
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"carriage": { "type": "string" },
|
||||||
|
"backend": { "type": "string" },
|
||||||
|
"cmd": {
|
||||||
|
"type": "array",
|
||||||
|
"items": { "type": "string" }
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"required": ["carriage", "backend", "cmd"]
|
||||||
|
}),
|
||||||
|
json!({
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"channel_id": { "type": "integer", "minimum": 0 }
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
vec![],
|
||||||
|
AccessControl {
|
||||||
|
required_scopes: vec![TTY_OPEN_SCOPE.to_string()],
|
||||||
|
required_scopes_any: None,
|
||||||
|
resource_type: None,
|
||||||
|
resource_action: None,
|
||||||
|
},
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.with_channel_open(ChannelOpenSpec::new(TTY_ALPN))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build the [`OpenHandler`] for `channels/tty/sub`.
|
||||||
|
///
|
||||||
|
/// The handler receives the open op's `input` (the `NegotiateRequest`
|
||||||
|
/// params — ignored here; `drive_session` reads it from the channel's
|
||||||
|
/// `BiStream` as the first frame, same as the direct-ALPN path), the
|
||||||
|
/// channel's [`Connection`] (data-plane ALPN `alk/tty`), and the
|
||||||
|
/// peer's [`AuthContext`]. It calls `accept_bi()` to get the channel's
|
||||||
|
/// [`BiStream`], splits it into read/write halves (the stdlib
|
||||||
|
/// `tokio::io::split` idiom — the same split the direct-ALPN
|
||||||
|
/// `TtyAdapter::handle` does), and runs [`drive_session`] on them.
|
||||||
|
///
|
||||||
|
/// The handler's `JoinHandle` is recorded by the channels wrapper for
|
||||||
|
/// teardown (abort on `channel/close` / connection drop). When the
|
||||||
|
/// session ends (exit chunk sent, stream closed, or stream reset),
|
||||||
|
/// `drive_session` returns and the spawned task completes; the
|
||||||
|
/// wrapper's teardown task then calls `manager.teardown_channel` and
|
||||||
|
/// `policy.on_close` (ADR-047 §7).
|
||||||
|
fn make_tty_open_handler(
|
||||||
|
backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
|
||||||
|
ownership: Option<Arc<dyn OwnershipProvider>>,
|
||||||
|
identity: Option<alkcall::core::auth::Identity>,
|
||||||
|
) -> OpenHandler {
|
||||||
|
Arc::new(move |input: Value, channel_conn: Connection, auth: AuthContext| {
|
||||||
|
let backends = Arc::clone(&backends);
|
||||||
|
let ownership = ownership.clone();
|
||||||
|
let identity = identity.clone().or_else(|| auth.identity.clone());
|
||||||
|
let _ = input;
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let stream = match channel_conn.accept_bi().await {
|
||||||
|
Ok(s) => s,
|
||||||
|
Err(e) => {
|
||||||
|
debug!("tty: channels open: accept_bi failed: {e}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let (client_read, client_write) = tokio::io::split(stream);
|
||||||
|
drive_session(client_write, client_read, backends, ownership, identity).await;
|
||||||
|
})
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use crate::backend::{MockBackend, TtyError};
|
||||||
|
use alkcall::channels::operations::ChannelCore;
|
||||||
|
use alkcall::channels::policy::default_policy;
|
||||||
|
use alkcall::channels::client::ChannelClient;
|
||||||
|
use alkcall::core::auth::Identity;
|
||||||
|
use alkcall::core::types::Connection as CoreConnection;
|
||||||
|
use alkcall::registry::registration::OperationRegistry;
|
||||||
|
use std::collections::HashMap as StdHashMap;
|
||||||
|
use tokio::io::duplex;
|
||||||
|
|
||||||
|
/// Build a `ChannelClient` and a server-side `ChannelCore` + registry
|
||||||
|
/// with `channels/tty/sub` registered, wired over a `tokio::io::duplex`
|
||||||
|
/// carrying the channels 8-byte chunk header wire format. Returns the
|
||||||
|
/// client and the per-connection `ChannelCore` (for any teardown
|
||||||
|
/// assertions). The server-side dispatch loop is spawned and runs
|
||||||
|
/// until the client drops or the test ends.
|
||||||
|
///
|
||||||
|
/// This mirrors the alkcall `channel_0_end_to_end_register_openable`
|
||||||
|
/// test's wiring pattern but uses alktty's `register_openable` helper
|
||||||
|
/// so the registration is the code under test.
|
||||||
|
async fn wire_client_and_server(
|
||||||
|
backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
|
||||||
|
ownership: Option<Arc<dyn OwnershipProvider>>,
|
||||||
|
identity: Option<Identity>,
|
||||||
|
) -> ChannelClient {
|
||||||
|
use alkcall::channels::adapter::{
|
||||||
|
ChannelsAdapter, InstallChannelZero,
|
||||||
|
};
|
||||||
|
use alkcall::core::auth::IdentityProvider;
|
||||||
|
use alkcall::protocol::connection::split_single_stream;
|
||||||
|
use alkcall::protocol::dispatch::Dispatcher;
|
||||||
|
|
||||||
|
struct NoopIdProvider;
|
||||||
|
impl IdentityProvider for NoopIdProvider {
|
||||||
|
fn resolve_from_fingerprint(
|
||||||
|
&self,
|
||||||
|
_: &str,
|
||||||
|
) -> Option<alkcall::core::auth::Identity> {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
fn resolve_from_token(
|
||||||
|
&self,
|
||||||
|
_: &alkcall::core::auth::AuthToken,
|
||||||
|
) -> Option<alkcall::core::auth::Identity> {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let policy = default_policy();
|
||||||
|
let policy_for_hook = Arc::clone(&policy);
|
||||||
|
let identity_for_conn = identity.clone();
|
||||||
|
|
||||||
|
let install_hook: InstallChannelZero =
|
||||||
|
Arc::new(move |manager, channel0_conn, auth| {
|
||||||
|
let backends = Arc::clone(&backends);
|
||||||
|
let ownership = ownership.clone();
|
||||||
|
let _identity = identity.clone();
|
||||||
|
let policy = Arc::clone(&policy_for_hook);
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let channel0_bidi = match channel0_conn.accept_bi().await {
|
||||||
|
Ok(s) => s,
|
||||||
|
Err(_) => return,
|
||||||
|
};
|
||||||
|
let (writer, reader) = split_single_stream(channel0_bidi);
|
||||||
|
|
||||||
|
// Propagate the identity to the channel-0
|
||||||
|
// connection so the call dispatch's
|
||||||
|
// `resolve_identity` sees it. The
|
||||||
|
// `ChannelsAdapter` builds `channel0_conn` fresh
|
||||||
|
// from `channel_source` and does NOT inherit the
|
||||||
|
// outer connection's identity — we set it here
|
||||||
|
// so the `AccessControl` scope-gate can check it.
|
||||||
|
if let Some(id) = _identity {
|
||||||
|
let _ = channel0_conn.set_identity(id);
|
||||||
|
}
|
||||||
|
|
||||||
|
let core = ChannelCore::new(manager, policy);
|
||||||
|
let mut registry = OperationRegistry::new();
|
||||||
|
register_openable(
|
||||||
|
&core,
|
||||||
|
Arc::clone(&backends),
|
||||||
|
ownership.clone(),
|
||||||
|
&mut registry,
|
||||||
|
auth.clone(),
|
||||||
|
)
|
||||||
|
.expect("register_openable");
|
||||||
|
let registry = Arc::new(registry);
|
||||||
|
let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
|
||||||
|
let call_connection = Arc::new(
|
||||||
|
alkcall::protocol::connection::CallConnection::new_single_stream(
|
||||||
|
channel0_conn,
|
||||||
|
Arc::clone(&writer),
|
||||||
|
),
|
||||||
|
);
|
||||||
|
let dp = Dispatcher::new(registry, provider);
|
||||||
|
dp.run_loop_single_stream(call_connection, reader, writer)
|
||||||
|
.await;
|
||||||
|
})
|
||||||
|
});
|
||||||
|
|
||||||
|
let (client_end, server_end) = duplex(64 * 1024);
|
||||||
|
let client_conn = CoreConnection::from_bidi(
|
||||||
|
client_end,
|
||||||
|
b"alk/channels".to_vec(),
|
||||||
|
None,
|
||||||
|
);
|
||||||
|
let server_conn = CoreConnection::from_bidi(
|
||||||
|
server_end,
|
||||||
|
b"alk/channels".to_vec(),
|
||||||
|
None,
|
||||||
|
);
|
||||||
|
// Set the identity on the server connection so the call
|
||||||
|
// dispatch sees it (the ACL check runs against the
|
||||||
|
// connection's identity, not the `install_channel_zero`
|
||||||
|
// hook's `auth`).
|
||||||
|
if let Some(id) = &identity_for_conn {
|
||||||
|
let _ = server_conn.set_identity(id.clone());
|
||||||
|
}
|
||||||
|
|
||||||
|
let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
|
||||||
|
let auth = AuthContext::anonymous(b"alk/channels");
|
||||||
|
let _server_handle = tokio::spawn(async move {
|
||||||
|
let _ = alkcall::core::types::ProtocolHandler::handle(
|
||||||
|
&adapter, server_conn, &auth,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
});
|
||||||
|
|
||||||
|
ChannelClient::from_connection(client_conn)
|
||||||
|
.await
|
||||||
|
.expect("channel client init")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A `ChannelLifecyclePolicy` with no cap (for tests).
|
||||||
|
use alkcall::channels::policy::NoCap;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn op_tty_open_is_channels_tty_sub() {
|
||||||
|
assert_eq!(OP_TTY_OPEN, "channels/tty/sub");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tty_alpn_is_alk_tty() {
|
||||||
|
assert_eq!(TTY_ALPN, "alk/tty");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tty_open_spec_has_channel_open_marker_for_alk_tty() {
|
||||||
|
let spec = tty_open_spec();
|
||||||
|
assert_eq!(spec.name, OP_TTY_OPEN);
|
||||||
|
assert_eq!(spec.op_type, OperationType::Sub);
|
||||||
|
assert_eq!(spec.visibility, Visibility::External);
|
||||||
|
let marker = spec.channel_open.expect("channel_open marker set");
|
||||||
|
assert_eq!(marker.alpn, TTY_ALPN);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tty_open_spec_requires_tty_open_scope() {
|
||||||
|
let spec = tty_open_spec();
|
||||||
|
assert_eq!(spec.access_control.required_scopes, vec![TTY_OPEN_SCOPE]);
|
||||||
|
assert!(spec.access_control.has_restrictions());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tty_open_spec_input_schema_has_required_fields() {
|
||||||
|
let spec = tty_open_spec();
|
||||||
|
let schema = spec.input_schema;
|
||||||
|
let required = schema
|
||||||
|
.get("required")
|
||||||
|
.and_then(|v| v.as_array())
|
||||||
|
.expect("required array");
|
||||||
|
let required_names: Vec<&str> =
|
||||||
|
required.iter().filter_map(|v| v.as_str()).collect();
|
||||||
|
assert!(required_names.contains(&"carriage"));
|
||||||
|
assert!(required_names.contains(&"backend"));
|
||||||
|
assert!(required_names.contains(&"cmd"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `register_openable` registers the op on the registry. The op
|
||||||
|
/// must be discoverable by name afterwards.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn register_openable_registers_op_on_registry() {
|
||||||
|
let (_client, server) = duplex(1024);
|
||||||
|
let (_reader, writer) = tokio::io::split(server);
|
||||||
|
let (handle, _runner) = alkcall::channels::mux::MuxRunner::new(Box::new(writer));
|
||||||
|
let manager = alkcall::channels::manager::ChannelManager::with_defaults(handle, None);
|
||||||
|
let core = ChannelCore::new(manager, default_policy());
|
||||||
|
|
||||||
|
let backends: Arc<HashMap<String, Arc<dyn TtyBackend>>> =
|
||||||
|
Arc::new(HashMap::new());
|
||||||
|
let mut registry = OperationRegistry::new();
|
||||||
|
register_openable(
|
||||||
|
&core,
|
||||||
|
backends,
|
||||||
|
None,
|
||||||
|
&mut registry,
|
||||||
|
AuthContext::anonymous(b"alk/channels"),
|
||||||
|
)
|
||||||
|
.expect("register_openable");
|
||||||
|
assert!(registry.registration(OP_TTY_OPEN).is_some());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// End-to-end: `ChannelClient::call_open_op("channels/tty/sub")`
|
||||||
|
/// returns `{ channel_id }` when the server-side has registered via
|
||||||
|
/// `register_openable` with a `MockBackend`. The open op increments
|
||||||
|
/// the channel count; the `channel_id` is non-zero (channel 0 is
|
||||||
|
/// the call channel, ADR-036).
|
||||||
|
#[tokio::test]
|
||||||
|
async fn end_to_end_open_op_returns_channel_id() {
|
||||||
|
let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
|
||||||
|
backends.insert(
|
||||||
|
"mock".to_string(),
|
||||||
|
Arc::new(MockBackend::with_exit_code(0)),
|
||||||
|
);
|
||||||
|
let backends = Arc::new(backends);
|
||||||
|
|
||||||
|
let identity = Identity {
|
||||||
|
id: "alice".to_string(),
|
||||||
|
scopes: vec![TTY_OPEN_SCOPE.to_string()],
|
||||||
|
resources: StdHashMap::new(),
|
||||||
|
};
|
||||||
|
let client = wire_client_and_server(backends, None, Some(identity)).await;
|
||||||
|
|
||||||
|
let response = tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(10),
|
||||||
|
client.call_open_op(
|
||||||
|
OP_TTY_OPEN,
|
||||||
|
json!({
|
||||||
|
"carriage": "raw",
|
||||||
|
"backend": "mock",
|
||||||
|
"cmd": ["true"],
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("open op timed out");
|
||||||
|
|
||||||
|
let out = response.result.expect("open op should succeed");
|
||||||
|
let channel_id = out
|
||||||
|
.get("channel_id")
|
||||||
|
.and_then(|v| v.as_u64())
|
||||||
|
.expect("channel_id in response");
|
||||||
|
assert!(
|
||||||
|
channel_id > 0,
|
||||||
|
"channel_id should be non-zero (channel 0 is the call channel), got {channel_id}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The open op denies when the caller lacks `tty:open`. The
|
||||||
|
/// `AccessControl::required_scopes` gate runs in the registry's
|
||||||
|
/// `invoke_streaming` before the `TtyOpenHandler` spawns.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn end_to_end_open_op_denies_without_tty_open_scope() {
|
||||||
|
let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
|
||||||
|
backends.insert(
|
||||||
|
"mock".to_string(),
|
||||||
|
Arc::new(MockBackend::with_exit_code(0)),
|
||||||
|
);
|
||||||
|
let backends = Arc::new(backends);
|
||||||
|
|
||||||
|
let identity = Identity {
|
||||||
|
id: "alice".to_string(),
|
||||||
|
scopes: vec![],
|
||||||
|
resources: StdHashMap::new(),
|
||||||
|
};
|
||||||
|
let client = wire_client_and_server(backends, None, Some(identity)).await;
|
||||||
|
|
||||||
|
let response = tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(10),
|
||||||
|
client.call_open_op(
|
||||||
|
OP_TTY_OPEN,
|
||||||
|
json!({
|
||||||
|
"carriage": "raw",
|
||||||
|
"backend": "mock",
|
||||||
|
"cmd": ["true"],
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("open op timed out");
|
||||||
|
|
||||||
|
let err = response.result.expect_err("open op should be denied");
|
||||||
|
assert!(
|
||||||
|
err.message.contains("scope") || err.code.contains("FORBIDDEN") || err.code.contains("AUTH"),
|
||||||
|
"error should mention scope/forbidden/auth, got code={} message={}",
|
||||||
|
err.code,
|
||||||
|
err.message
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sanity that the `MockBackend` we use in the channels tests
|
||||||
|
/// compiles as a `TtyBackend` and resolves to the configured exit
|
||||||
|
/// code. Belt-and-suspenders — the backend module already tests
|
||||||
|
/// this, but the channels tests depend on it.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn mock_backend_resolves_configured_exit_code() {
|
||||||
|
let backend = MockBackend::with_exit_code(7);
|
||||||
|
let params = crate::backend::TtyParams {
|
||||||
|
terminal: None,
|
||||||
|
cmd: vec!["true".to_string()],
|
||||||
|
cwd: None,
|
||||||
|
env: HashMap::new(),
|
||||||
|
backend_params: serde_json::Map::new(),
|
||||||
|
};
|
||||||
|
let handle = backend.allocate(¶ms).await.expect("allocate");
|
||||||
|
let code = handle.exit_code.await.expect("exit_code resolves");
|
||||||
|
assert_eq!(code, 7);
|
||||||
|
let _: Result<i32, TtyError> = Ok(code);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -48,8 +48,10 @@
|
|||||||
|
|
||||||
pub mod adapter;
|
pub mod adapter;
|
||||||
pub mod backend;
|
pub mod backend;
|
||||||
|
pub mod channels;
|
||||||
pub mod control;
|
pub mod control;
|
||||||
pub mod negotiation;
|
pub mod negotiation;
|
||||||
|
pub mod session;
|
||||||
pub mod wire;
|
pub mod wire;
|
||||||
|
|
||||||
#[cfg(feature = "local")]
|
#[cfg(feature = "local")]
|
||||||
|
|||||||
+2
-2
@@ -61,7 +61,7 @@ use crate::wire::MAX_CHUNK_LEN;
|
|||||||
/// latter land in `backend_params` via the `serde(flatten)` below. The
|
/// latter land in `backend_params` via the `serde(flatten)` below. The
|
||||||
/// shared fields are consumed by name; whatever remains flows into the
|
/// shared fields are consumed by name; whatever remains flows into the
|
||||||
/// `backend_params` map.
|
/// `backend_params` map.
|
||||||
#[derive(Debug, Clone, serde::Deserialize)]
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||||
pub struct NegotiateRequest {
|
pub struct NegotiateRequest {
|
||||||
/// `"raw"` in v1; any other value → `malformed_negotiation` (checked by
|
/// `"raw"` in v1; any other value → `malformed_negotiation` (checked by
|
||||||
/// the adapter, not this parser).
|
/// the adapter, not this parser).
|
||||||
@@ -98,7 +98,7 @@ pub struct NegotiateRequest {
|
|||||||
/// for the local backend. The `modes` field is reserved (OQ-44 — default
|
/// for the local backend. The `modes` field is reserved (OQ-44 — default
|
||||||
/// terminal modes suffice for the current scope); backends MUST ignore its
|
/// terminal modes suffice for the current scope); backends MUST ignore its
|
||||||
/// content in v1.
|
/// content in v1.
|
||||||
#[derive(Debug, Clone, serde::Deserialize)]
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||||
pub struct TerminalParamsWire {
|
pub struct TerminalParamsWire {
|
||||||
/// `TERM` environment value (e.g., `"xterm-256color"`); `None` =
|
/// `TERM` environment value (e.g., `"xterm-256color"`); `None` =
|
||||||
/// backend default.
|
/// backend default.
|
||||||
|
|||||||
+683
@@ -0,0 +1,683 @@
|
|||||||
|
//! Consumer half — `TtySession`, the typed client wrapper around the
|
||||||
|
//! `alk/tty` wire protocol (per alkcall's protocol-crate pattern).
|
||||||
|
//!
|
||||||
|
//! Two constructors:
|
||||||
|
//!
|
||||||
|
//! - [`TtySession::connect_direct`] — for a direct `alk/tty` ALPN
|
||||||
|
//! connection. The consumer dials the transport (TLS, QUIC,
|
||||||
|
//! WebSocket), negotiates the `alk/tty` ALPN, and hands the
|
||||||
|
//! `Connection` to `connect_direct`. The session takes ownership of
|
||||||
|
//! the connection's single `BiStream` (yield-once per connection,
|
||||||
|
//! ADR-065), writes the negotiation frame, and exposes the typed
|
||||||
|
//! methods.
|
||||||
|
//!
|
||||||
|
//! - [`TtySession::open_via_channels`] — for the `alk/channels`
|
||||||
|
//! multiplexed path. The consumer holds a
|
||||||
|
//! [`ChannelClient`][alkcall::channels::ChannelClient], calls
|
||||||
|
//! `open_via_channels(client, params)`, which invokes
|
||||||
|
//! `channels/tty/sub` on channel 0, adopts the resulting channel,
|
||||||
|
//! builds a `Connection` from the reassembled read half + mux write
|
||||||
|
//! half, and runs the same negotiation + typed-methods flow on the
|
||||||
|
//! channel's `BiStream`.
|
||||||
|
//!
|
||||||
|
//! The session handle exposes:
|
||||||
|
//! - [`TtySession::send_stdin`] / [`TtySession::close_stdin`] — write
|
||||||
|
//! stdin chunks, close stdin (zero-length sentinel).
|
||||||
|
//! - [`TtySession::recv_stdout`] / [`TtySession::recv_stderr`] —
|
||||||
|
//! streams of stdout/stderr chunks (`Stream<Item = Bytes>`).
|
||||||
|
//! - [`TtySession::resize`] — send a `Resize` control message.
|
||||||
|
//! - [`TtySession::signal`] — send a `Signal` control message.
|
||||||
|
//! - [`TtySession::wait`] — await the `Exit` control chunk (the
|
||||||
|
//! process exit code).
|
||||||
|
//!
|
||||||
|
//! The session does NOT re-serialize the negotiation request itself
|
||||||
|
//! — the caller passes a `NegotiateRequest` (or a `serde_json::Value`
|
||||||
|
//! for the channels path, since `ChannelClient::open_channel` takes a
|
||||||
|
//! `Value`). The session writes the frame and switches to raw-chunk
|
||||||
|
//! mode. See ADR-052 for the two-carriage model.
|
||||||
|
//!
|
||||||
|
//! # WASM
|
||||||
|
//!
|
||||||
|
//! `TtySession` is wasm-clean (no `tokio::process`, no `std::thread`,
|
||||||
|
//! no `libc`). The consumer half is exactly the part a browser-side
|
||||||
|
//! or Python-wasm adapter would use — it runs the wire protocol
|
||||||
|
//! against a `Connection` the consumer dials (a WebSocket binary
|
||||||
|
//! stream, a WebTransport bidi stream, etc.). The producer half
|
||||||
|
//! (`TtyAdapter` + `register_openable`) runs on a real OS with a
|
||||||
|
//! backend that can spawn processes.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::pin::Pin;
|
||||||
|
|
||||||
|
use bytes::Bytes;
|
||||||
|
use futures::Stream;
|
||||||
|
use tokio::io::{AsyncRead, AsyncWrite};
|
||||||
|
use tokio::sync::{mpsc, Mutex};
|
||||||
|
use tokio::task::JoinHandle;
|
||||||
|
use tracing::{debug, warn};
|
||||||
|
|
||||||
|
use alkcall::channels::client::ChannelClient;
|
||||||
|
use alkcall::core::Connection;
|
||||||
|
|
||||||
|
use crate::control::ControlMessage;
|
||||||
|
use crate::negotiation::{NegotiateRequest, NegotiationError, NegotiationWriter};
|
||||||
|
use crate::wire::{Chunk, ChunkReader, ChunkWriter, RawError, STREAM_CTRL_IN, STREAM_CTRL_OUT};
|
||||||
|
|
||||||
|
/// Errors from the typed consumer client.
|
||||||
|
#[derive(Debug, thiserror::Error)]
|
||||||
|
pub enum TtySessionError {
|
||||||
|
/// The underlying transport I/O failed (not a clean EOF).
|
||||||
|
#[error("io: {0}")]
|
||||||
|
Io(#[from] std::io::Error),
|
||||||
|
/// The raw-chunk codec errored (invalid stream type, chunk too
|
||||||
|
/// large, transport I/O).
|
||||||
|
#[error("wire: {0}")]
|
||||||
|
Wire(#[from] RawError),
|
||||||
|
/// The negotiation frame failed to serialize.
|
||||||
|
#[error("negotiation serialize: {0}")]
|
||||||
|
NegotiationSerialize(#[from] serde_json::Error),
|
||||||
|
/// The negotiation frame failed to write (framing I/O error).
|
||||||
|
#[error("negotiation write: {0}")]
|
||||||
|
NegotiationWrite(#[from] NegotiationError),
|
||||||
|
/// The channels open op failed (returned a `CallError` or
|
||||||
|
/// `channel_id` missing from the response).
|
||||||
|
#[error("channels open: {0}")]
|
||||||
|
ChannelsOpen(String),
|
||||||
|
/// The server sent a negotiation error frame (the first frame on
|
||||||
|
/// the stream is a length-prefixed JSON `{"error":"..."}` rather
|
||||||
|
/// than a raw chunk).
|
||||||
|
#[error("negotiation rejected: {error}")]
|
||||||
|
NegotiationRejected { error: String, fields: HashMap<String, String> },
|
||||||
|
/// The session ended (server closed the stream) before an `Exit`
|
||||||
|
/// control chunk arrived. `wait()` returns this when the
|
||||||
|
/// stdout/stderr pumps drain and no exit chunk was observed.
|
||||||
|
#[error("session ended without exit chunk")]
|
||||||
|
NoExitChunk,
|
||||||
|
/// The `Exit` control chunk's JSON payload failed to parse.
|
||||||
|
#[error("malformed exit chunk: {0}")]
|
||||||
|
MalformedExitChunk(serde_json::Error),
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A live `alk/tty` session — the typed consumer-side handle.
|
||||||
|
///
|
||||||
|
/// Constructed via [`TtySession::connect_direct`] (direct `alk/tty`
|
||||||
|
/// ALPN) or [`TtySession::open_via_channels`] (multiplexed over
|
||||||
|
/// `alk/channels`). The session owns the negotiation frame exchange
|
||||||
|
/// and exposes typed methods for stdin/stdout/stderr/control/exit.
|
||||||
|
///
|
||||||
|
/// The session drives a single read pump task (chunks →
|
||||||
|
/// stdout/stderr/exit channels) and holds the write half for stdin +
|
||||||
|
/// control messages. Dropping the session cancels the read pump and
|
||||||
|
/// closes the write half.
|
||||||
|
pub struct TtySession {
|
||||||
|
/// The write half of the bidi stream, wrapped in a `ChunkWriter`.
|
||||||
|
/// `send_stdin`, `resize`, `signal`, and `close_stdin` write
|
||||||
|
/// through this. Behind a `Mutex` so the methods can take `&self`
|
||||||
|
/// and the caller doesn't need `&mut self` to drive the session.
|
||||||
|
writer: Mutex<ChunkWriter<Box<dyn AsyncWrite + Send + Unpin>>>,
|
||||||
|
/// stdout chunks from the read pump. The caller drains this via
|
||||||
|
/// `recv_stdout()`. `Option` so `recv_stdout()` can take it
|
||||||
|
/// (calling twice returns an empty stream the second time).
|
||||||
|
stdout_rx: Mutex<Option<mpsc::Receiver<Bytes>>>,
|
||||||
|
/// stderr chunks from the read pump. `None` for PTY-mode backends
|
||||||
|
/// (stdout/stderr merged into stdout by the kernel PTY) or after
|
||||||
|
/// `recv_stderr()` has taken it.
|
||||||
|
stderr_rx: Mutex<Option<mpsc::Receiver<Bytes>>>,
|
||||||
|
/// The exit code, resolved by the read pump when it observes the
|
||||||
|
/// `Exit` control chunk. `wait()` awaits this. `Option<Result>`
|
||||||
|
/// starts as `None`; the pump sends `Some(Ok(code))` on exit
|
||||||
|
/// chunk or `Some(Err(NoExitChunk))` on stream close.
|
||||||
|
exit_code: tokio::sync::watch::Receiver<Option<Result<i32, TtySessionError>>>,
|
||||||
|
/// The read pump task handle. Dropping the session aborts it.
|
||||||
|
_read_pump: JoinHandle<()>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TtySession {
|
||||||
|
/// Connect directly over a `alk/tty` ALPN connection.
|
||||||
|
///
|
||||||
|
/// The consumer dials the transport (TLS, QUIC, WebSocket),
|
||||||
|
/// negotiates `alk/tty`, and hands the `Connection` here. The
|
||||||
|
/// session takes the connection's single `BiStream` (yield-once
|
||||||
|
/// per single-stream connection, ADR-065), writes the negotiation
|
||||||
|
/// frame, and starts the read pump.
|
||||||
|
///
|
||||||
|
/// `negotiate` is the [`NegotiateRequest`] the session writes as
|
||||||
|
/// the first frame. The caller builds it (the typed shape is
|
||||||
|
/// easier to construct than a raw JSON `Value`); the session
|
||||||
|
/// serializes it.
|
||||||
|
pub async fn connect_direct(
|
||||||
|
connection: Connection,
|
||||||
|
negotiate: NegotiateRequest,
|
||||||
|
) -> Result<Self, TtySessionError> {
|
||||||
|
let stream = connection.accept_bi().await.map_err(|e| {
|
||||||
|
std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
|
||||||
|
})?;
|
||||||
|
Self::from_bidi_stream(stream, negotiate).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Open a TTY session via `alk/channels` — the multiplexed path.
|
||||||
|
///
|
||||||
|
/// The consumer holds a [`ChannelClient`], calls
|
||||||
|
/// `open_via_channels(client, params)`, which invokes
|
||||||
|
/// `channels/tty/sub` on channel 0, adopts the resulting channel,
|
||||||
|
/// builds a `Connection` from the reassembled read half + mux
|
||||||
|
/// write half, and runs the same negotiation + typed-methods flow
|
||||||
|
/// on the channel's `BiStream`.
|
||||||
|
///
|
||||||
|
/// `params` is the `NegotiateRequest` as a `serde_json::Value` —
|
||||||
|
/// the channels open op takes a `Value`, not a typed struct (the
|
||||||
|
/// op's input schema is the `NegotiateRequest` shape). The
|
||||||
|
/// session re-parses it as a `NegotiateRequest` after the channel
|
||||||
|
/// is open so the typed methods can use the strongly-typed shape.
|
||||||
|
pub async fn open_via_channels(
|
||||||
|
client: &ChannelClient,
|
||||||
|
params: serde_json::Value,
|
||||||
|
) -> Result<Self, TtySessionError> {
|
||||||
|
let (channel_id, send, recv) = client
|
||||||
|
.open_channel(crate::channels::OP_TTY_OPEN, params.clone(), crate::channels::TTY_ALPN)
|
||||||
|
.await
|
||||||
|
.map_err(TtySessionError::ChannelsOpen)?;
|
||||||
|
debug!("tty: opened channel {channel_id} via channels");
|
||||||
|
|
||||||
|
let remote_addr = client.manager().remote_addr();
|
||||||
|
let source = alkcall::channels::source::channel_source(recv, send, remote_addr);
|
||||||
|
let channel_conn = Connection::from_source(
|
||||||
|
source,
|
||||||
|
crate::channels::TTY_ALPN.as_bytes().to_vec(),
|
||||||
|
);
|
||||||
|
|
||||||
|
let negotiate: NegotiateRequest =
|
||||||
|
serde_json::from_value(params).map_err(TtySessionError::NegotiationSerialize)?;
|
||||||
|
Self::from_bidi_stream_via(channel_conn, negotiate).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Shared inner: take a `BiStream`, write the negotiation frame,
|
||||||
|
/// start the read pump. Used by both `connect_direct` and (after
|
||||||
|
/// the channels open) `open_via_channels`.
|
||||||
|
async fn from_bidi_stream(
|
||||||
|
stream: alkcall::core::BiStream,
|
||||||
|
negotiate: NegotiateRequest,
|
||||||
|
) -> Result<Self, TtySessionError> {
|
||||||
|
let (read, write) = tokio::io::split(stream);
|
||||||
|
Self::from_halves(read, write, negotiate).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Like `from_bidi_stream` but takes the channel's `Connection`
|
||||||
|
/// directly (the channels path already has the `Connection` from
|
||||||
|
/// `Connection::from_source`).
|
||||||
|
async fn from_bidi_stream_via(
|
||||||
|
channel_conn: Connection,
|
||||||
|
negotiate: NegotiateRequest,
|
||||||
|
) -> Result<Self, TtySessionError> {
|
||||||
|
let stream = channel_conn.accept_bi().await.map_err(|e| {
|
||||||
|
std::io::Error::new(std::io::ErrorKind::ConnectionReset, format!("{e}"))
|
||||||
|
})?;
|
||||||
|
Self::from_bidi_stream(stream, negotiate).await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Core inner: take a read half and a write half, write the
|
||||||
|
/// negotiation frame, spawn the read pump, return the session.
|
||||||
|
async fn from_halves<R, W>(
|
||||||
|
read: R,
|
||||||
|
write: W,
|
||||||
|
negotiate: NegotiateRequest,
|
||||||
|
) -> Result<Self, TtySessionError>
|
||||||
|
where
|
||||||
|
R: AsyncRead + Send + Unpin + 'static,
|
||||||
|
W: AsyncWrite + Send + Unpin + 'static,
|
||||||
|
{
|
||||||
|
let boxed_write: Box<dyn AsyncWrite + Send + Unpin> = Box::new(write);
|
||||||
|
let mut neg_writer = NegotiationWriter::new(boxed_write);
|
||||||
|
let body = serde_json::to_vec(&negotiate)?;
|
||||||
|
neg_writer.write_frame(&body).await?;
|
||||||
|
let writer = ChunkWriter::new(neg_writer.into_inner());
|
||||||
|
|
||||||
|
let (stdout_tx, stdout_rx) = mpsc::channel::<Bytes>(64);
|
||||||
|
let (stderr_tx, stderr_rx) = mpsc::channel::<Bytes>(64);
|
||||||
|
let (exit_tx, exit_rx) =
|
||||||
|
tokio::sync::watch::channel::<Option<Result<i32, TtySessionError>>>(None);
|
||||||
|
|
||||||
|
let read_pump = tokio::spawn(read_pump(
|
||||||
|
read,
|
||||||
|
stdout_tx,
|
||||||
|
stderr_tx,
|
||||||
|
exit_tx,
|
||||||
|
));
|
||||||
|
|
||||||
|
Ok(Self {
|
||||||
|
writer: Mutex::new(writer),
|
||||||
|
stdout_rx: Mutex::new(Some(stdout_rx)),
|
||||||
|
stderr_rx: Mutex::new(Some(stderr_rx)),
|
||||||
|
exit_code: exit_rx,
|
||||||
|
_read_pump: read_pump,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send stdin bytes. Writes a stdin chunk (stream_type 0) with the
|
||||||
|
/// given payload. An empty `bytes` writes a zero-length sentinel
|
||||||
|
/// (client stdin EOF — see `tty-wire.md` §"Sentinels"); callers
|
||||||
|
/// that want to signal EOF should use [`close_stdin`] instead,
|
||||||
|
/// which is explicit.
|
||||||
|
pub async fn send_stdin(&self, bytes: Bytes) -> Result<(), TtySessionError> {
|
||||||
|
let mut writer = self.writer.lock().await;
|
||||||
|
let chunk = Chunk::stdin(bytes);
|
||||||
|
writer.write_chunk(&chunk).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Close stdin — send a zero-length stdin chunk (the EOF sentinel)
|
||||||
|
/// and flush. The server closes the backend's stdin but keeps
|
||||||
|
/// pumping stdout + the exit chunk (see `tty-wire.md` §"Stdin
|
||||||
|
/// Closure").
|
||||||
|
pub async fn close_stdin(&self) -> Result<(), TtySessionError> {
|
||||||
|
let mut writer = self.writer.lock().await;
|
||||||
|
let chunk = Chunk::stdin(Bytes::new());
|
||||||
|
writer.write_chunk(&chunk).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send a `Resize` control message (client→server, `STREAM_CTRL_IN`).
|
||||||
|
/// `pixel_width`/`pixel_height` default to 0 (most terminals don't
|
||||||
|
/// report pixel dimensions).
|
||||||
|
pub async fn resize(
|
||||||
|
&self,
|
||||||
|
cols: u16,
|
||||||
|
rows: u16,
|
||||||
|
pixel_width: u16,
|
||||||
|
pixel_height: u16,
|
||||||
|
) -> Result<(), TtySessionError> {
|
||||||
|
let mut writer = self.writer.lock().await;
|
||||||
|
let msg = ControlMessage::Resize {
|
||||||
|
cols,
|
||||||
|
rows,
|
||||||
|
pixel_width,
|
||||||
|
pixel_height,
|
||||||
|
};
|
||||||
|
let json = msg.to_json()?;
|
||||||
|
let chunk = Chunk::ctrl_in(json);
|
||||||
|
writer.write_chunk(&chunk).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send a `Signal` control message (client→server,
|
||||||
|
/// `STREAM_CTRL_IN`). `name` is an uppercase string from the
|
||||||
|
/// supported set (`HUP`, `INT`, `QUIT`, `TERM`, `KILL`, `USR1`,
|
||||||
|
/// `USR2`, `TSTP`, `CONT` — see [`crate::control::signal_from_name`]).
|
||||||
|
/// Unknown names are forwarded as-is; the backend decides whether
|
||||||
|
/// to ignore or fall back to its default kill.
|
||||||
|
pub async fn signal(&self, name: &str) -> Result<(), TtySessionError> {
|
||||||
|
let mut writer = self.writer.lock().await;
|
||||||
|
let msg = ControlMessage::Signal {
|
||||||
|
name: name.to_string(),
|
||||||
|
};
|
||||||
|
let json = msg.to_json()?;
|
||||||
|
let chunk = Chunk::ctrl_in(json);
|
||||||
|
writer.write_chunk(&chunk).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the stdout stream. Returns a `Stream<Item = Bytes>` that
|
||||||
|
/// yields stdout chunks as they arrive. The stream ends when the
|
||||||
|
/// server's stdout reaches EOF (a zero-length stdout sentinel
|
||||||
|
/// chunk, see `tty-wire.md` §"Sentinels").
|
||||||
|
///
|
||||||
|
/// This consumes the stdout receiver — calling it twice returns
|
||||||
|
/// an empty stream the second time (the receiver is behind a
|
||||||
|
/// `Mutex<Option<...>>` and is taken).
|
||||||
|
pub async fn recv_stdout(&self) -> Pin<Box<dyn Stream<Item = Bytes> + Send>> {
|
||||||
|
let mut guard = self.stdout_rx.lock().await;
|
||||||
|
if let Some(rx) = guard.take() {
|
||||||
|
return Box::pin(futures::stream::unfold(
|
||||||
|
rx,
|
||||||
|
|mut rx| async move { rx.recv().await.map(|bytes| (bytes, rx)) },
|
||||||
|
));
|
||||||
|
}
|
||||||
|
// Already taken — return an empty stream.
|
||||||
|
Box::pin(futures::stream::empty())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the stderr stream. `None` for PTY-mode backends
|
||||||
|
/// (stdout/stderr merged into stdout by the kernel PTY), or if
|
||||||
|
/// already taken. The stream ends when the server's stderr reaches
|
||||||
|
/// EOF.
|
||||||
|
pub async fn recv_stderr(&self) -> Option<Pin<Box<dyn Stream<Item = Bytes> + Send>>> {
|
||||||
|
let mut guard = self.stderr_rx.lock().await;
|
||||||
|
let rx = guard.take()?;
|
||||||
|
Some(Box::pin(futures::stream::unfold(
|
||||||
|
rx,
|
||||||
|
|mut rx| async move { rx.recv().await.map(|bytes| (bytes, rx)) },
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Await the `Exit` control chunk and return the process exit
|
||||||
|
/// code. The exit chunk is the last control chunk before stream
|
||||||
|
/// close (ADR-055). `code` is `i32` matching
|
||||||
|
/// `std::process::ExitStatus::code()`; negative values are
|
||||||
|
/// signal-terminated, `-1` is the adapter's best-effort "backend
|
||||||
|
/// could not determine the exit code" sentinel.
|
||||||
|
///
|
||||||
|
/// Returns [`TtySessionError::NoExitChunk`] if the session ends
|
||||||
|
/// (server closed the stream) before an exit chunk is observed.
|
||||||
|
pub async fn wait(&self) -> Result<i32, TtySessionError> {
|
||||||
|
let mut rx = self.exit_code.clone();
|
||||||
|
// If the pump already resolved before we started waiting, the
|
||||||
|
// watch's current value is `Some(_)` — return it.
|
||||||
|
{
|
||||||
|
let borrow = rx.borrow();
|
||||||
|
if let Some(Ok(code)) = borrow.as_ref() {
|
||||||
|
return Ok(*code);
|
||||||
|
}
|
||||||
|
if let Some(Err(_)) = borrow.as_ref() {
|
||||||
|
return Err(TtySessionError::NoExitChunk);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Wait for the read pump to send a value.
|
||||||
|
rx.changed()
|
||||||
|
.await
|
||||||
|
.map_err(|_| TtySessionError::NoExitChunk)?;
|
||||||
|
let borrow = rx.borrow();
|
||||||
|
match borrow.as_ref() {
|
||||||
|
Some(Ok(code)) => Ok(*code),
|
||||||
|
Some(Err(_)) | None => Err(TtySessionError::NoExitChunk),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for TtySession {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
self._read_pump.abort();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The read pump: reads chunks off the bidi stream's read half and
|
||||||
|
/// routes them to the stdout/stderr/exit channels. The pump owns the
|
||||||
|
/// `ChunkReader`. When the stream closes (clean EOF or transport
|
||||||
|
/// error), the pump drains the stdout/stderr channels (drops the
|
||||||
|
/// senders) and resolves the exit watch with `NoExitChunk` if no
|
||||||
|
/// `Exit` chunk was observed, or with the parsed exit code if one was.
|
||||||
|
///
|
||||||
|
/// The pump distinguishes the four stream types:
|
||||||
|
/// - `STREAM_STDOUT` (1) → stdout channel
|
||||||
|
/// - `STREAM_STDERR` (2) → stderr channel
|
||||||
|
/// - `STREAM_CTRL_OUT` (4) → control message; parses as
|
||||||
|
/// `ControlMessage` and, if it's `Exit`, resolves the exit watch
|
||||||
|
/// - `STREAM_STDIN` (0) / `STREAM_CTRL_IN` (3) — client→server only;
|
||||||
|
/// the server shouldn't send these, the pump ignores them (with a
|
||||||
|
/// debug log)
|
||||||
|
async fn read_pump<R>(
|
||||||
|
read: R,
|
||||||
|
stdout_tx: mpsc::Sender<Bytes>,
|
||||||
|
stderr_tx: mpsc::Sender<Bytes>,
|
||||||
|
exit_tx: tokio::sync::watch::Sender<Option<Result<i32, TtySessionError>>>,
|
||||||
|
) where
|
||||||
|
R: AsyncRead + Send + Unpin + 'static,
|
||||||
|
{
|
||||||
|
let mut reader = ChunkReader::new(read);
|
||||||
|
let mut exit_resolved = false;
|
||||||
|
loop {
|
||||||
|
match reader.read_chunk().await {
|
||||||
|
Ok(chunk) => match chunk.stream_type {
|
||||||
|
crate::wire::STREAM_STDOUT => {
|
||||||
|
if stdout_tx.send(chunk.bytes).await.is_err() {
|
||||||
|
debug!("tty: stdout receiver dropped, ending read pump");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
crate::wire::STREAM_STDERR => {
|
||||||
|
if stderr_tx.send(chunk.bytes).await.is_err() {
|
||||||
|
debug!("tty: stderr receiver dropped, ending read pump");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
STREAM_CTRL_OUT => match ControlMessage::from_slice(&chunk.bytes) {
|
||||||
|
Ok(ControlMessage::Exit { code }) => {
|
||||||
|
let _ = exit_tx.send(Some(Ok(code)));
|
||||||
|
exit_resolved = true;
|
||||||
|
debug!("tty: exit chunk received, code={code}");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Ok(other) => {
|
||||||
|
debug!("tty: ignoring non-exit control on STREAM_CTRL_OUT: {other:?}");
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
let _ = exit_tx.send(Some(Err(TtySessionError::MalformedExitChunk(e))));
|
||||||
|
exit_resolved = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
STREAM_CTRL_IN | crate::wire::STREAM_STDIN => {
|
||||||
|
debug!(
|
||||||
|
"tty: ignoring client→server stream_type {} from server",
|
||||||
|
chunk.stream_type
|
||||||
|
);
|
||||||
|
}
|
||||||
|
other => {
|
||||||
|
debug!("tty: ignoring unknown stream_type {other}");
|
||||||
|
}
|
||||||
|
},
|
||||||
|
Err(RawError::ConnectionClosed) => {
|
||||||
|
debug!("tty: read pump: stream closed");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
warn!("tty: read pump: chunk read error: {e}");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Drain the channels (drop the senders so the receivers observe
|
||||||
|
// EOF). If no exit chunk was observed, resolve the watch with
|
||||||
|
// `NoExitChunk`.
|
||||||
|
drop(stdout_tx);
|
||||||
|
drop(stderr_tx);
|
||||||
|
if !exit_resolved {
|
||||||
|
let _ = exit_tx.send(Some(Err(TtySessionError::NoExitChunk)));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use crate::backend::{MockBackend, TtyBackend};
|
||||||
|
use crate::negotiation::NegotiateRequest;
|
||||||
|
use alkcall::core::auth::Identity;
|
||||||
|
use alkcall::core::types::Connection;
|
||||||
|
use futures::stream::StreamExt;
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use tokio::io::duplex;
|
||||||
|
|
||||||
|
/// Build a `NegotiateRequest` for tests — minimal valid shape.
|
||||||
|
fn test_negotiate(backend: &str) -> NegotiateRequest {
|
||||||
|
NegotiateRequest {
|
||||||
|
carriage: "raw".to_string(),
|
||||||
|
backend: backend.to_string(),
|
||||||
|
tty: None,
|
||||||
|
cmd: vec!["true".to_string()],
|
||||||
|
cwd: None,
|
||||||
|
env: HashMap::new(),
|
||||||
|
backend_params: serde_json::Map::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build a `TtySession` wired to a `drive_session` over a duplex
|
||||||
|
/// pair, with a `MockBackend` registered as `"mock"`. The session
|
||||||
|
/// is the client side; the server side runs `drive_session` on
|
||||||
|
/// the other half of the duplex. Returns the session and the
|
||||||
|
/// server-side task handle.
|
||||||
|
async fn wire_session_and_server(
|
||||||
|
backend: Arc<dyn TtyBackend>,
|
||||||
|
identity: Option<Identity>,
|
||||||
|
) -> (TtySession, tokio::task::JoinHandle<()>) {
|
||||||
|
let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
|
||||||
|
backends.insert("mock".to_string(), backend);
|
||||||
|
let backends = Arc::new(backends);
|
||||||
|
|
||||||
|
let (client, server) = duplex(64 * 1024);
|
||||||
|
let (server_read, server_write) = tokio::io::split(server);
|
||||||
|
let server_task = tokio::spawn(async move {
|
||||||
|
crate::adapter::drive_session(
|
||||||
|
server_write,
|
||||||
|
server_read,
|
||||||
|
backends,
|
||||||
|
None,
|
||||||
|
identity,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
});
|
||||||
|
|
||||||
|
let client_conn =
|
||||||
|
Connection::from_bidi(client, b"alk/tty".to_vec(), None);
|
||||||
|
let session = TtySession::connect_direct(client_conn, test_negotiate("mock"))
|
||||||
|
.await
|
||||||
|
.expect("connect_direct");
|
||||||
|
(session, server_task)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn connect_direct_writes_negotiation_frame() {
|
||||||
|
// The session writes the negotiation frame on construction.
|
||||||
|
// If the server side reads it and dispatches to the backend,
|
||||||
|
// the session is wired. A `MockBackend` resolves to exit 0
|
||||||
|
// immediately; the session's `wait()` should observe it.
|
||||||
|
let backend = Arc::new(MockBackend::with_exit_code(0));
|
||||||
|
let identity = Some(Identity {
|
||||||
|
id: "alice".to_string(),
|
||||||
|
scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
|
||||||
|
resources: HashMap::new(),
|
||||||
|
});
|
||||||
|
let (session, _server) = wire_session_and_server(backend, identity).await;
|
||||||
|
let code = tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(5),
|
||||||
|
session.wait(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("wait didn't time out")
|
||||||
|
.expect("wait returns exit code");
|
||||||
|
assert_eq!(code, 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn send_stdin_round_trips_to_backend() {
|
||||||
|
// Use a backend that captures stdin — but `MockBackend` doesn't
|
||||||
|
// expose the stdin channel to the test. This test just verifies
|
||||||
|
// `send_stdin` doesn't error; the adapter tests cover the
|
||||||
|
// stdin-to-backend pump.
|
||||||
|
let backend = Arc::new(MockBackend::with_exit_code(0));
|
||||||
|
let identity = Some(Identity {
|
||||||
|
id: "alice".to_string(),
|
||||||
|
scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
|
||||||
|
resources: HashMap::new(),
|
||||||
|
});
|
||||||
|
let (session, _server) = wire_session_and_server(backend, identity).await;
|
||||||
|
session
|
||||||
|
.send_stdin(Bytes::from_static(b"hello"))
|
||||||
|
.await
|
||||||
|
.expect("send_stdin");
|
||||||
|
session
|
||||||
|
.close_stdin()
|
||||||
|
.await
|
||||||
|
.expect("close_stdin");
|
||||||
|
let _ = tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(5),
|
||||||
|
session.wait(),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn resize_and_signal_dont_error() {
|
||||||
|
// The session writes control chunks; whether the backend
|
||||||
|
// receives them is the adapter's concern (covered by the
|
||||||
|
// adapter tests). This test verifies the typed methods
|
||||||
|
// serialize and write without error.
|
||||||
|
let backend = Arc::new(MockBackend::with_exit_code(0));
|
||||||
|
let identity = Some(Identity {
|
||||||
|
id: "alice".to_string(),
|
||||||
|
scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
|
||||||
|
resources: HashMap::new(),
|
||||||
|
});
|
||||||
|
let (session, _server) = wire_session_and_server(backend, identity).await;
|
||||||
|
session.resize(80, 24, 0, 0).await.expect("resize");
|
||||||
|
session.signal("INT").await.expect("signal");
|
||||||
|
let _ = tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(5),
|
||||||
|
session.wait(),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn recv_stdout_yields_backend_stdout() {
|
||||||
|
// `MockBackend` doesn't pump stdout (it resolves exit
|
||||||
|
// immediately), so the stdout stream should be empty. This
|
||||||
|
// test verifies the stream API works and ends cleanly.
|
||||||
|
let backend = Arc::new(MockBackend::with_exit_code(0));
|
||||||
|
let identity = Some(Identity {
|
||||||
|
id: "alice".to_string(),
|
||||||
|
scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()],
|
||||||
|
resources: HashMap::new(),
|
||||||
|
});
|
||||||
|
let (session, _server) = wire_session_and_server(backend, identity).await;
|
||||||
|
let stdout = session.recv_stdout().await;
|
||||||
|
let collected: Vec<Bytes> = stdout.collect().await;
|
||||||
|
// The backend's stdout stream ends immediately (MockBackend
|
||||||
|
// drops its stdout sender on exit), so the stream should be
|
||||||
|
// empty or near-empty.
|
||||||
|
assert!(
|
||||||
|
collected.is_empty() || collected.iter().all(|b| b.is_empty()),
|
||||||
|
"mock backend produces no stdout, got {collected:?}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn wait_returns_no_exit_chunk_when_server_drops_without_exit() {
|
||||||
|
// If the server side drops the stream before sending an exit
|
||||||
|
// chunk, `wait()` should return `NoExitChunk`. We simulate
|
||||||
|
// this by wiring the session to a duplex where the "server"
|
||||||
|
// reads the negotiation frame (so the client's write succeeds)
|
||||||
|
// then drops without sending anything back.
|
||||||
|
let (client, mut server) = duplex(64);
|
||||||
|
let server_handle = tokio::spawn(async move {
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
// Read the 4-byte length prefix + body so the client's
|
||||||
|
// negotiation write succeeds (duplex buffers are small;
|
||||||
|
// a partial write would block and the test would hang).
|
||||||
|
let mut len_buf = [0u8; 4];
|
||||||
|
let _ = server.read_exact(&mut len_buf).await;
|
||||||
|
let len = u32::from_be_bytes(len_buf) as usize;
|
||||||
|
let mut body = vec![0u8; len];
|
||||||
|
let _ = server.read_exact(&mut body).await;
|
||||||
|
// Drop `server` — the client's read pump hits EOF.
|
||||||
|
});
|
||||||
|
let client_conn =
|
||||||
|
Connection::from_bidi(client, b"alk/tty".to_vec(), None);
|
||||||
|
let session = TtySession::connect_direct(client_conn, test_negotiate("mock"))
|
||||||
|
.await
|
||||||
|
.expect("connect_direct");
|
||||||
|
let result = tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(5),
|
||||||
|
session.wait(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("wait didn't time out");
|
||||||
|
assert!(matches!(result, Err(TtySessionError::NoExitChunk)));
|
||||||
|
let _ = server_handle.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `connect_direct` with a `Connection` whose stream is broken
|
||||||
|
/// (server half dropped) should return an error from the
|
||||||
|
/// negotiation frame write (`BrokenPipe`).
|
||||||
|
#[tokio::test]
|
||||||
|
async fn connect_direct_errors_when_stream_is_broken() {
|
||||||
|
let (client, server) = duplex(64);
|
||||||
|
drop(server);
|
||||||
|
let client_conn =
|
||||||
|
Connection::from_bidi(client, b"alk/tty".to_vec(), None);
|
||||||
|
let result = TtySession::connect_direct(client_conn, test_negotiate("mock")).await;
|
||||||
|
assert!(
|
||||||
|
result.is_err(),
|
||||||
|
"construction should fail when the negotiation write hits a broken pipe"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user