diff --git a/Cargo.lock b/Cargo.lock index 3d970b3..aa39147 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -27,9 +27,9 @@ dependencies = [ [[package]] name = "alkcall" -version = "0.3.1" +version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d4116dba284601e3337e918d4a4256706e43dd8bb6a30095bf96d664ae58d74a" +checksum = "9badefe048a194c93eed09bc5326ebf092fc86817e561d88de2cd12b6302753a" dependencies = [ "async-trait", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 0b46107..b33542f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ default = [] local = ["dep:portable-pty", "dep:tokio-util", "tokio/process", "tokio/rt-multi-thread"] [dependencies] -alkcall = "0.3.1" +alkcall = "0.4.0" # Minimal, wasm-clean tokio features. `local` adds `process` + # `rt-multi-thread` (non-wasm). Do NOT use `features = ["full"]` — it # pulls in `signal`/`fs`/`net` which break `wasm32-unknown-unknown`. diff --git a/docs/reviews/001-code-review.md b/docs/reviews/001-code-review.md index 2f3cbaa..bc3aa38 100644 --- a/docs/reviews/001-code-review.md +++ b/docs/reviews/001-code-review.md @@ -1,6 +1,6 @@ --- -status: partially-resolved (M1, M2, L2, L4, L5, N1, N2, N3, N5) -last_updated: 2026-08-17 +status: partially-resolved (L6, N4, N6) +last_updated: 2026-09-05 reviewed_artifacts: - src/lib.rs - src/wire.rs @@ -532,8 +532,8 @@ Highlights: | M2 | `wait()` swallows `MalformedExitChunk` | propagate the variant | small | low | ✅ resolved | | L2 | consumer stdout/stderr routing untested | add emitting test backend | small | low | ✅ resolved | | M1 | negotiation-rejection frame unhandled | implement disambiguation read | medium | medium (wire-facing) | ✅ resolved | -| L1 | channels `input` ignored | decide drop-vs-pass-through | small | low | open | -| L3 | `open_via_channels` 0% covered | end-to-end channels consumer test | medium | low | open | +| L1 | channels `input` ignored | decide drop-vs-pass-through | small | low | ✅ resolved (2026-09-05) | +| L3 | `open_via_channels` 0% covered | end-to-end channels consumer test | medium | low | ✅ resolved (2026-09-05) | | L6 | pty bridge error paths untested | targeted error-path tests | medium | low | open | | N4 | sleep-based timing | readiness signals | small | low | open | | N6 | MSRV unverified | CI MSRV job or bump | small | none | open | @@ -569,23 +569,74 @@ Post-fix coverage: `session.rs` 79.71% → 87.43% lines, total 90.74% → `cargo test --all-features`, clippy native + wasm, fmt, doc, wasm check). +### Resolution (2026-09-05, L1 + L3 — the channels consumer path) + +**L1 — resolved via the publisher decision: the channels path carries no +second negotiation frame (ADR-009 in `docs/architecture/decisions/`).** + +Two upstream alkcall changes were prerequisites (the review's premise +that the registry "validates `input`" was wrong — alkcall never +enforced `input_schema` on any dispatch path): + +- alkcall 0.4.0 — `OperationSpec.input_schema` is now enforced at call + time by all three registry dispatch entry points (`invoke`, + `invoke_streaming`, `invoke_sink`), compiled once at registration + (fail-closed, same rule as `publish_schema`/CF-003). Violations + return `INVALID_INPUT`. The `channels/tty/sub` spec's input schema + now declares the shared `NegotiateRequest` fields (`carriage`, + `backend`, `cmd` required); `tty`/`cwd`/`env`/backend-params stay + free-form (raw JSON Schema is permissive on unknown keys, so the + opaque ADR-053 params pass through). +- alkcall 0.4.1 — early-arrival chunks for a not-yet-adopted channel + are parked (bounded per-channel buffer) and drained on + `adopt_channel`, instead of dropped. The open-op-response / + producer's-first-write race silently lost the first chunks of any + push-first producer — found by L3's test (the session never resolved + because the stdout sentinel + exit chunk of an + immediately-resolving backend arrived before the adopt). + +Design: the open op's `input` IS the negotiation (design 2 of the +publisher decision). `make_tty_open_handler` parses the +registry-validated `input` into a `NegotiateRequest` and drives +`drive_session_pre_negotiated` (new public API in `adapter.rs`) — the +same three-pump session driver as the direct path, minus the +wire-frame negotiation phase. Validation still runs +(`carriage`/`cmd`/backend lookup + ADR-050 ownership); failures go to +the client as a `0x00`-prefixed negotiation error frame on the channel +stream, so the M1 disambiguation read applies unchanged. The +`tty:open` scope gate is enforced by the registry's `AccessControl` +(not re-checked in the handler). `tty_open_spec()`'s schema is now the +partial `NegotiateRequest` shape. `TtySession::open_via_channels` +parses `params` locally (fail-fast before a channel is allocated), +opens the channel, and starts raw-chunk mode directly +(`from_halves_raw` — no negotiation write, peek retained for the +error-frame path). + +Tests: `open_via_channels_end_to_end_negotiates_and_waits`, +`open_via_channels_surfaces_negotiation_rejected`, +`open_via_channels_fails_fast_on_schema_invalid_params`, +`open_via_channels_fails_fast_on_unparseable_params`, +`open_via_channels_routes_backend_stdout_and_stderr`, +`pre_negotiated_happy_path_over_plain_duplex` (adapter), plus the +shared harness tests in `src/testing.rs` (harness-level handler→client +data flow). The channels harness (`wire_client_and_server`) moved to +`crate::testing` so session tests share it with channels tests. + +**L3 — resolved** by the same work: `open_via_channels` (and +`from_bidi_stream_via`) are covered end-to-end against the real +producer path (`register_openable` + `drive_session_pre_negotiated` +through alkcall's channels stack). + ### Remaining (open) -- **L1** — channels `input` ignored; needs a publisher decision - (drop the parameter vs pass-through to `drive_session`). -- **L3** — `open_via_channels` still 0% covered; needs the channels - harness shared across modules. - **L6** — pty bridge error paths untested. - **N4** — sleep-based timing in signal/cancel tests. - **N6** — MSRV unverified. ### Recommended Order (remaining) -1. **L1 + L3** — the channels consumer path; do together since L3's - test will exercise L1's code. -2. **L6** — pty bridge error paths; medium effort, lower priority than - the consumer-half work. -3. **N4 + N6** — test hardening and MSRV; defer until CI exists. +1. **L6** — pty bridge error paths; medium effort. +2. **N4 + N6** — test hardening and MSRV; defer until CI exists. --- @@ -593,7 +644,8 @@ check). - All line numbers refer to the tree at commit `18c4924` (the last commit on `main` at review time). The resolution section above - reflects the tree at commit `9944153`. + reflects the tree at commit `9944153` and the 2026-09-05 L1+L3 + resolution. - The coverage numbers are from `cargo llvm-cov --all-features` on the same tree. The `--show-missing-lines` output was used to attribute gaps; the full report is at `target/llvm-cov/html`. diff --git a/src/adapter.rs b/src/adapter.rs index 4502125..a893a66 100644 --- a/src/adapter.rs +++ b/src/adapter.rs @@ -192,7 +192,10 @@ async fn send_negotiation_error( /// when the stream is reset (cancel-cleanup path — no exit chunk sent). /// /// This is the per-stream session driver — generalized from the POC's -/// `session::drive_session` to the [`TtyBackend`] trait. +/// `session::drive_session` to the [`TtyBackend`] trait. The negotiation +/// frame is read from the wire (the direct-ALPN path — ADR-052); the +/// channels path (open-op `input` is the negotiation) uses +/// [`drive_session_pre_negotiated`] instead. pub async fn drive_session( client_send: impl AsyncWrite + Send + Unpin + 'static, client_recv: impl AsyncRead + Send + Unpin + 'static, @@ -207,6 +210,131 @@ pub async fn drive_session( } } +/// Drive a `alk/tty` session whose negotiation already happened out of +/// band — the channels path (ADR-009). The open op's registry-validated +/// `input` was parsed into `req` by the channels wrapper; the raw-chunk +/// data plane starts immediately. No negotiation frame is read from the +/// wire, and none is written by the client. +/// +/// Validation still runs here (`carriage`/`cmd`/backend lookup) plus the +/// ADR-050 ownership check — failures go to the client as a negotiation +/// error frame on the channel stream (the same framing the direct path +/// uses, so the consumer's M1 disambiguation read applies unchanged). +/// The `tty:open` scope gate is NOT re-checked: on the channels path the +/// registry's `AccessControl` enforced it at open time (the identity view +/// the registry checked — resolved from the connection — is the +/// authoritative one; the handler-side identity is only the +/// ownership-check subject). +pub async fn drive_session_pre_negotiated( + client_send: impl AsyncWrite + Send + Unpin + 'static, + client_recv: impl AsyncRead + Send + Unpin + 'static, + req: NegotiateRequest, + backends: Arc>>, + ownership: Option>, + identity: Option, +) { + if let Err(e) = drive_session_pre_negotiated_inner( + client_send, + client_recv, + req, + &backends, + &ownership, + &identity, + ) + .await + { + debug!("tty: session ended with error: {e}"); + } +} + +/// Validate a parsed `NegotiateRequest`, select the backend, run the +/// ADR-050 ownership check, and allocate. Shared by the direct +/// (wire-negotiated) and channels (pre-negotiated) paths. Failures are +/// reported to the client as a negotiation error frame on `neg_writer` +/// (which is consumed and shut down); `Err(())` means "session over, the +/// client has been told". On success the unwrapped write half comes back +/// with the allocated handle. +/// +/// `enforce_scope` gates the `tty:open` scope check: `true` on the direct +/// path (the adapter is the only gate), `false` on the channels path (the +/// registry's `AccessControl` enforced the scope at open time — see +/// [`drive_session_pre_negotiated`]). +#[allow(clippy::type_complexity)] +async fn validate_and_allocate( + neg_writer: NegotiationWriter, + req: NegotiateRequest, + backends: &HashMap>, + ownership: &Option>, + identity: &Option, + enforce_scope: bool, +) -> Result<(TtyHandle, W), ()> +where + W: AsyncWrite + Send + Unpin + 'static, +{ + if req.carriage != "raw" { + send_negotiation_error( + neg_writer, + "malformed_negotiation", + &[("message", "carriage must be 'raw'")], + ) + .await; + return Err(()); + } + if req.cmd.is_empty() { + send_negotiation_error( + neg_writer, + "malformed_negotiation", + &[("message", "cmd must be non-empty")], + ) + .await; + return Err(()); + } + + let backend = match backends.get(&req.backend) { + Some(b) => b.clone(), + None => { + send_negotiation_error(neg_writer, "unknown_backend", &[("backend", &req.backend)]) + .await; + return Err(()); + } + }; + + if enforce_scope && !has_scope(identity, TTY_OPEN_SCOPE) { + send_negotiation_error(neg_writer, "forbidden", &[]).await; + return Err(()); + } + + let params = crate::backend::TtyParams::from(req); + + if let Some(provider) = ownership { + if let Some((kind, id)) = backend.resource_id(¶ms) { + let owns = identity + .as_ref() + .map(|id_ref| provider.owns(id_ref, kind, &id, "tty")) + .unwrap_or(false); + if !owns { + send_negotiation_error(neg_writer, "forbidden", &[]).await; + return Err(()); + } + } + } + + let handle = match backend.allocate(¶ms).await { + Ok(h) => h, + Err(e) => { + send_negotiation_error( + neg_writer, + "allocate_failed", + &[("message", &e.to_string())], + ) + .await; + return Err(()); + } + }; + + Ok((handle, neg_writer.into_inner())) +} + async fn drive_session_inner( client_send: W, client_recv: R, @@ -253,72 +381,38 @@ where } }; - if req.carriage != "raw" { - send_negotiation_error( - neg_writer, - "malformed_negotiation", - &[("message", "carriage must be 'raw'")], - ) - .await; - return Ok(()); - } - if req.cmd.is_empty() { - send_negotiation_error( - neg_writer, - "malformed_negotiation", - &[("message", "cmd must be non-empty")], - ) - .await; - return Ok(()); - } - - let backend = match backends.get(&req.backend) { - Some(b) => b.clone(), - None => { - send_negotiation_error(neg_writer, "unknown_backend", &[("backend", &req.backend)]) - .await; - return Ok(()); - } - }; - - if !has_scope(identity, TTY_OPEN_SCOPE) { - send_negotiation_error(neg_writer, "forbidden", &[]).await; - return Ok(()); - } - - let params = crate::backend::TtyParams::from(req); - - if let Some(provider) = ownership { - if let Some((kind, id)) = backend.resource_id(¶ms) { - let owns = identity - .as_ref() - .map(|id_ref| provider.owns(id_ref, kind, &id, "tty")) - .unwrap_or(false); - if !owns { - send_negotiation_error(neg_writer, "forbidden", &[]).await; - return Ok(()); - } - } - } - - let handle = match backend.allocate(¶ms).await { - Ok(h) => h, - Err(e) => { - send_negotiation_error( - neg_writer, - "allocate_failed", - &[("message", &e.to_string())], - ) - .await; - return Ok(()); - } - }; + let (handle, client_write) = + match validate_and_allocate(neg_writer, req, backends, ownership, identity, true).await { + Ok(ok) => ok, + Err(()) => return Ok(()), + }; let client_read = neg_reader.into_inner(); - let client_write = neg_writer.into_inner(); pump_session(client_write, client_read, handle).await } +async fn drive_session_pre_negotiated_inner( + client_send: W, + client_recv: R, + req: NegotiateRequest, + backends: &HashMap>, + ownership: &Option>, + identity: &Option, +) -> Result<(), std::io::Error> +where + W: AsyncWrite + Send + Unpin + 'static, + R: AsyncRead + Send + Unpin + 'static, +{ + let neg_writer = NegotiationWriter::new(client_send); + let (handle, client_write) = + match validate_and_allocate(neg_writer, req, backends, ownership, identity, false).await { + Ok(ok) => ok, + Err(()) => return Ok(()), + }; + + pump_session(client_write, client_recv, handle).await +} + /// Phase 3: the bidirectional pump. Three concurrent tasks plus a drainer. /// /// Enforces the exit-chunk-is-last invariant (ADR-055): the adapter waits for @@ -1519,4 +1613,38 @@ mod tests { exit_tx.send(Ok(0)).unwrap(); let _ = session.await; } + + #[tokio::test] + async fn pre_negotiated_happy_path_over_plain_duplex() { + use crate::backend::MockBackend; + use tokio::io::AsyncReadExt; + + let backend: Arc = Arc::new(MockBackend::with_exit_code(0)); + let mut backends: StdHashMap> = StdHashMap::new(); + backends.insert("mock".to_string(), backend); + let backends = Arc::new(backends); + + let (mut client, server) = duplex(8 * 1024); + let (server_read, server_write) = tokio::io::split(server); + + let identity = identity_with_scope(TTY_OPEN_SCOPE); + let req: NegotiateRequest = + serde_json::from_slice(TEST_NEG.as_bytes()).expect("parse TEST_NEG"); + let session = tokio::spawn(async move { + drive_session_pre_negotiated(server_write, server_read, req, backends, None, identity) + .await; + }); + + let mut buf = [0u8; 5]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + client.read_exact(&mut buf), + ) + .await + .expect("no first chunk from pre-negotiated driver") + .expect("read"); + assert_eq!(buf[0], STREAM_STDOUT, "stdout sentinel comes first"); + + let _ = session.await; + } } diff --git a/src/channels.rs b/src/channels.rs index 0fbadd6..d8bb73e 100644 --- a/src/channels.rs +++ b/src/channels.rs @@ -22,7 +22,7 @@ //! `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 +//! and spawns the session driver. 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. //! @@ -41,8 +41,9 @@ use alkcall::registry::spec::{ use serde_json::{json, Value}; use tracing::debug; -use crate::adapter::{drive_session, TTY_OPEN_SCOPE}; +use crate::adapter::{drive_session_pre_negotiated, TTY_OPEN_SCOPE}; use crate::backend::TtyBackend; +use crate::negotiation::NegotiateRequest; /// The per-ALPN open operation name (`channels//sub` convention, /// ADR-047). TTY is consumer-opens (the client requests a shell), so @@ -78,12 +79,13 @@ pub const TTY_ALPN: &str = "alk/tty"; /// /// `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. +/// is consulted inside `drive_session_pre_negotiated` (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 (here by the registry's `AccessControl`), 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 @@ -117,14 +119,18 @@ pub fn register_openable( /// 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. +/// The input schema is the shared part of the `NegotiateRequest` shape +/// (`carriage`/`backend`/`cmd` required — alkcall 0.4 enforces it at +/// dispatch). The schema is deliberately partial: `tty`, `cwd`, `env`, +/// and backend-specific selector fields are free-form (`additionalProperties` +/// defaults to `true` in raw JSON Schema), because the +/// `NegotiateRequest` wire shape is owned by `negotiation.rs` and the +/// backend params are opaque (ADR-053) — duplicating their full schemas +/// here would create a second definition to drift. The registry check +/// catches structurally-broken opens (missing command, wrong types); +/// the full parse and semantic validation (`carriage == "raw"`, +/// backend lookup) runs in the handler via +/// [`crate::adapter::drive_session_pre_negotiated`]. pub fn tty_open_spec() -> OperationSpec { OperationSpec::new( OP_TTY_OPEN, @@ -162,21 +168,35 @@ pub fn tty_open_spec() -> OperationSpec { /// 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 receives the open op's `input` — the `NegotiateRequest` +/// params, validated against the spec's input schema by the registry +/// (alkcall 0.4: `input_schema` is enforced at dispatch) and passed +/// through as the authoritative negotiation (ADR-009: the channels +/// path does not carry a second negotiation frame on the channel; the +/// open op's `input` *is* the negotiation). The channel's +/// [`Connection`] (data-plane ALPN `alk/tty`) and the peer's +/// [`AuthContext`] come with it. The handler parses `input`, 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 +/// [`crate::adapter::drive_session_pre_negotiated`] on them — the same +/// three-pump session driver as the direct-ALPN path, minus the +/// wire-frame negotiation phase. +/// +/// Parse or validation failures (unknown backend, `allocate_failed`, +/// ADR-050 ownership denial) are reported to the client as a +/// negotiation error frame on the channel stream (the same +/// `0x00`-prefixed framing the direct path uses), so a consumer's +/// negotiation-error disambiguation read applies unchanged. The +/// `tty:open` scope gate is enforced by the registry's `AccessControl` +/// before this handler runs — the handler does not re-check it. /// /// 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). +/// `drive_session_pre_negotiated` 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>>, ownership: Option>, @@ -187,9 +207,15 @@ fn make_tty_open_handler( 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 req: NegotiateRequest = match serde_json::from_value(input) { + Ok(r) => r, + Err(e) => { + debug!("tty: channels open: invalid negotiate params: {e}"); + return; + } + }; let stream = match channel_conn.accept_bi().await { Ok(s) => s, Err(e) => { @@ -198,7 +224,15 @@ fn make_tty_open_handler( } }; let (client_read, client_write) = tokio::io::split(stream); - drive_session(client_write, client_read, backends, ownership, identity).await; + drive_session_pre_negotiated( + client_write, + client_read, + req, + backends, + ownership, + identity, + ) + .await; }) }, ) @@ -208,125 +242,13 @@ fn make_tty_open_handler( mod tests { use super::*; use crate::backend::{MockBackend, TtyError}; - use alkcall::channels::client::ChannelClient; + use crate::testing::wire_client_and_server; use alkcall::channels::operations::ChannelCore; - use alkcall::channels::policy::default_policy; 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>>, - ownership: Option>, - identity: Option, - ) -> 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 { - None - } - fn resolve_from_token( - &self, - _: &alkcall::core::auth::AuthToken, - ) -> Option { - 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 = 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"); @@ -372,6 +294,7 @@ mod tests { /// must be discoverable by name afterwards. #[tokio::test] async fn register_openable_registers_op_on_registry() { + use alkcall::channels::policy::default_policy; let (_client, server) = duplex(1024); let (_reader, writer) = tokio::io::split(server); let (handle, _runner) = alkcall::channels::mux::MuxRunner::new(Box::new(writer)); diff --git a/src/lib.rs b/src/lib.rs index 422a95b..186acb5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,3 +56,6 @@ pub mod wire; #[cfg(feature = "local")] pub mod local; + +#[cfg(test)] +pub(crate) mod testing; diff --git a/src/session.rs b/src/session.rs index dc7f328..9400e0b 100644 --- a/src/session.rs +++ b/src/session.rs @@ -17,8 +17,10 @@ //! `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`. +//! half, and runs the typed-methods flow directly in raw chunk mode +//! (ADR-009: the open op's `params` — validated by the registry's +//! input schema — *are* the negotiation; no second negotiation +//! frame is written on the channel's data stream). //! //! The session handle exposes: //! - [`TtySession::send_stdin`] / [`TtySession::close_stdin`] — write @@ -31,10 +33,11 @@ //! 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. +//! — the caller passes a `NegotiateRequest` (direct path) or a +//! `serde_json::Value` (channels path — the open op's `input` is the +//! negotiation). The direct path writes the frame and switches to +//! raw-chunk mode; the channels path starts in raw-chunk mode. See +//! ADR-052 for the two-carriage model. //! //! # WASM //! @@ -180,22 +183,34 @@ impl TtySession { /// `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`. + /// write half, and runs the typed-methods flow directly in raw + /// chunk mode (ADR-009: the open op's `params` — validated by the + /// registry's input schema — *are* the negotiation; no second + /// negotiation frame is written on the channel's data stream). /// /// `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. + /// the channels open op takes a `Value`, not a typed struct. The + /// session parses it as a `NegotiateRequest` before opening (so a + /// malformed request fails fast, before a channel is allocated) + /// and the producer parses the same value from the open op. + /// + /// Failures before the channel opens (ACL denial, unknown op, + /// channel cap, invalid params) surface as + /// [`TtySessionError::ChannelsOpen`]. Post-open failures (unknown + /// backend, allocate failure, ownership denial) arrive as a + /// negotiation error frame on the channel stream — the session + /// surfaces those as [`TtySessionError::NegotiationRejected`] via + /// the same `0x00` disambiguation read the direct path uses. pub async fn open_via_channels( client: &ChannelClient, params: serde_json::Value, ) -> Result { + let _: NegotiateRequest = serde_json::from_value(params.clone()) + .map_err(TtySessionError::NegotiationSerialize)?; let (channel_id, send, recv) = client .open_channel( crate::channels::OP_TTY_OPEN, - params.clone(), + params, crate::channels::TTY_ALPN, ) .await @@ -207,9 +222,7 @@ impl TtySession { 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 + Self::from_bidi_stream_via(channel_conn).await } /// Shared inner: take a `BiStream`, write the negotiation frame, @@ -225,15 +238,14 @@ impl TtySession { /// 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 { + /// `Connection::from_source`). The negotiation already happened in + /// the open op (ADR-009) — the stream starts in raw-chunk mode. + async fn from_bidi_stream_via(channel_conn: Connection) -> Result { 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 + let (read, write) = tokio::io::split(stream); + Self::from_halves_raw(read, write).await } /// Core inner: take a read half and a write half, write the @@ -274,6 +286,48 @@ impl TtySession { Err(e) => return Err(TtySessionError::Wire(e)), } + Self::start_pump(writer, reader, first_byte_peeked) + } + + /// Core inner for the channels path (ADR-009): the negotiation + /// already happened in the open op — the stream is already in + /// raw-chunk mode. The peek still applies: the producer sends a + /// `0x00`-prefixed error frame on post-open failures (unknown + /// backend, allocate failure, ownership denial), and a raw chunk + /// (`stream_type` in `{1, 2, 4}`) on success. + async fn from_halves_raw(read: R, write: W) -> Result + where + R: AsyncRead + Send + Unpin + 'static, + W: AsyncWrite + Send + Unpin + 'static, + { + let writer = ChunkWriter::new(Box::new(write) as Box); + let mut reader = ChunkReader::new(read); + let mut first_byte_peeked = false; + match reader.peek_stream_type().await { + Ok(0x00) => { + return Err(read_negotiation_error(reader.into_inner()).await); + } + Ok(_) => first_byte_peeked = true, + Err(RawError::ConnectionClosed) => { + // The server closed cleanly without a response. Fall + // through to the read pump, which resolves `NoExitChunk`. + } + Err(e) => return Err(TtySessionError::Wire(e)), + } + + Self::start_pump(writer, reader, first_byte_peeked) + } + + /// Wire up the exit watch + stdout/stderr channels and spawn the + /// read pump. Shared by `from_halves` and `from_halves_raw`. + fn start_pump( + writer: ChunkWriter>, + reader: ChunkReader, + first_byte_peeked: bool, + ) -> Result + where + R: AsyncRead + Send + Unpin + 'static, + { let (stdout_tx, stdout_rx) = mpsc::channel::(64); let (stderr_tx, stderr_rx) = mpsc::channel::(64); let (exit_tx, exit_rx) = tokio::sync::watch::channel::>(None); @@ -934,4 +988,173 @@ mod tests { } let _ = server_handle.await; } + + // --- channels consumer path (L3, over the shared `testing` harness) ---- + + use crate::testing::{tty_identity, wire_client_and_server}; + + /// The negotiate params as the open op's `input` (same JSON shape + /// the direct path serializes from `NegotiateRequest`). + fn test_open_params(backend: &str) -> serde_json::Value { + serde_json::json!({ + "carriage": "raw", + "backend": backend, + "cmd": ["true"], + }) + } + + fn mock_backends(code: i32) -> Arc>> { + let mut backends: HashMap> = HashMap::new(); + backends.insert( + "mock".to_string(), + Arc::new(MockBackend::with_exit_code(code)), + ); + Arc::new(backends) + } + + /// End-to-end (L3): `TtySession::open_via_channels` opens the + /// channel through the real `register_openable` producer path and + /// resolves the session. The negotiation travels in the open op's + /// `input` (ADR-009); the channel stream starts in raw-chunk mode. + #[tokio::test] + async fn open_via_channels_end_to_end_negotiates_and_waits() { + let client = + wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await; + + let session = tokio::time::timeout( + std::time::Duration::from_secs(10), + TtySession::open_via_channels(&client, test_open_params("mock")), + ) + .await + .expect("open_via_channels timed out") + .expect("session opens"); + + let code = tokio::time::timeout(std::time::Duration::from_secs(10), session.wait()) + .await + .expect("wait didn't time out") + .expect("wait returns exit code"); + assert_eq!(code, 0); + } + + /// The same end-to-end path with an unknown backend: the producer + /// sends a `0x00`-prefixed error frame on the channel stream, the + /// consumer's post-open peek disambiguates it, and the session + /// surfaces `NegotiationRejected` (M1 on the channels path). + #[tokio::test] + async fn open_via_channels_surfaces_negotiation_rejected() { + let client = + wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await; + + let result = tokio::time::timeout( + std::time::Duration::from_secs(10), + TtySession::open_via_channels(&client, test_open_params("nope")), + ) + .await + .expect("open_via_channels timed out"); + match result { + Err(TtySessionError::NegotiationRejected { error, fields }) => { + assert_eq!(error, "unknown_backend"); + assert_eq!(fields.get("backend").map(String::as_str), Some("nope")); + } + Ok(_) => panic!("expected NegotiationRejected, got Ok(session)"), + Err(other) => panic!("expected NegotiationRejected, got {other:?}"), + } + } + + /// The open op's `input` is schema-validated by the registry + /// (alkcall 0.4): a params value missing the required `backend` + /// field is rejected before any handler runs, so the failure is a + /// `ChannelsOpen` error (the open op fails), not a session error. + #[tokio::test] + async fn open_via_channels_fails_fast_on_schema_invalid_params() { + let client = + wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await; + + let result = tokio::time::timeout( + std::time::Duration::from_secs(10), + TtySession::open_via_channels( + &client, + serde_json::json!({ "carriage": "raw", "cmd": ["true"] }), + ), + ) + .await + .expect("open_via_channels timed out"); + assert!( + matches!(result, Err(TtySessionError::NegotiationSerialize(_))), + "schema-invalid params fail at the local NegotiateRequest parse (fail-fast, pre-open)" + ); + } + + /// `open_via_channels` with params that fail the local + /// `NegotiateRequest` parse (not just the schema): fails fast, + /// before a channel is allocated. + #[tokio::test] + async fn open_via_channels_fails_fast_on_unparseable_params() { + let client = + wire_client_and_server(mock_backends(0), None, Some(tty_identity("alice"))).await; + + let result = tokio::time::timeout( + std::time::Duration::from_secs(10), + TtySession::open_via_channels( + &client, + serde_json::json!({ "carriage": 42, "backend": "mock", "cmd": ["true"] }), + ), + ) + .await + .expect("open_via_channels timed out"); + assert!( + matches!(result, Err(TtySessionError::NegotiationSerialize(_))), + "unparseable params must fail before the open op" + ); + } + + /// End-to-end with an emitting backend (L2's backend): stdout and + /// stderr route through the channels data plane to the consumer's + /// typed streams — the full producer+consumer channels path with + /// real data. + #[tokio::test] + async fn open_via_channels_routes_backend_stdout_and_stderr() { + let mut backends: HashMap> = HashMap::new(); + backends.insert( + "mock".to_string(), + Arc::new(EmittingBackend { + stdout: vec![Bytes::from_static(b"ch-out")], + stderr: vec![Bytes::from_static(b"ch-err")], + exit_code: 3, + }), + ); + let client = + wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await; + + let session = tokio::time::timeout( + std::time::Duration::from_secs(10), + TtySession::open_via_channels(&client, test_open_params("mock")), + ) + .await + .expect("open_via_channels timed out") + .expect("session opens"); + + let stdout = session.recv_stdout().await; + let collected: Vec = stdout.collect().await; + let data: Vec = collected.into_iter().filter(|b| !b.is_empty()).collect(); + assert_eq!( + data, + vec![Bytes::from_static(b"ch-out")], + "stdout should route through the channels data plane" + ); + + let stderr = session.recv_stderr().await.expect("stderr present"); + let collected: Vec = stderr.collect().await; + assert_eq!( + collected, + vec![Bytes::from_static(b"ch-err")], + "stderr should route through the channels data plane" + ); + + let code = tokio::time::timeout(std::time::Duration::from_secs(10), session.wait()) + .await + .expect("wait didn't time out") + .expect("wait returns exit code"); + assert_eq!(code, 3); + } } diff --git a/src/testing.rs b/src/testing.rs new file mode 100644 index 0000000..18f10a3 --- /dev/null +++ b/src/testing.rs @@ -0,0 +1,209 @@ +//! Shared test wiring for the channels-based tests (the producer +//! harness in `channels.rs::tests` and the consumer end-to-end tests +//! in `session.rs::tests` both need the same `ChannelClient` ↔ +//! server wiring). `#[cfg(test)]`-only — never part of the public +//! API. +use std::collections::HashMap; +use std::sync::Arc; + +use alkcall::channels::client::ChannelClient; +use alkcall::channels::operations::ChannelCore; +use alkcall::channels::policy::default_policy; +use alkcall::core::auth::{AuthContext, Identity}; +use alkcall::core::types::Connection as CoreConnection; +use alkcall::registry::registration::OperationRegistry; + +use crate::backend::TtyBackend; +use crate::channels::register_openable; + +/// 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. 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. +pub(crate) async fn wire_client_and_server( + backends: Arc>>, + ownership: Option>, + identity: Option, +) -> ChannelClient { + use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; + use alkcall::channels::policy::NoCap; + 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 { + None + } + fn resolve_from_token(&self, _: &alkcall::core::auth::AuthToken) -> Option { + 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 = 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) = tokio::io::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") +} + +/// An identity with the `tty:open` scope (the gate +/// `channels/tty/sub` requires). +pub(crate) fn tty_identity(id: &str) -> Identity { + Identity { + id: id.to_string(), + scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()], + resources: HashMap::new(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::backend::MockBackend; + use crate::channels::OP_TTY_OPEN; + + #[test] + fn tty_identity_has_tty_open_scope() { + let id = tty_identity("carol"); + assert_eq!(id.id, "carol"); + assert!(id + .scopes + .iter() + .any(|s| s == crate::adapter::TTY_OPEN_SCOPE)); + } + + #[tokio::test] + async fn harness_open_op_returns_channel_id() { + let mut backends: HashMap> = HashMap::new(); + backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0))); + let client = + wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await; + let response = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.call_open_op( + OP_TTY_OPEN, + serde_json::json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }), + ), + ) + .await + .expect("open op timed out"); + assert!( + response.result.is_ok(), + "open op failed: {:?}", + response.result + ); + } + + #[tokio::test] + async fn harness_handler_to_client_data_flows() { + use tokio::io::AsyncReadExt; + + let mut backends: HashMap> = HashMap::new(); + backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0))); + let client = + wire_client_and_server(Arc::new(backends), None, Some(tty_identity("alice"))).await; + + let (channel_id, send, mut recv) = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.open_channel( + OP_TTY_OPEN, + serde_json::json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }), + crate::channels::TTY_ALPN, + ), + ) + .await + .expect("open_channel timed out") + .expect("open_channel"); + + assert!(channel_id > 0); + + // The producer writes a 5-byte TTY header (stdout sentinel) + // almost immediately after the open (MockBackend resolves + // exit right away) — a push-first producer whose first write + // races the consumer's adopt. Read it: alkcall 0.4.1's + // early-arrival park guarantees the first chunks are not lost. + let mut first = [0u8; 5]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + recv.read_exact(&mut first), + ) + .await + .expect("no first chunk from producer (handler→client direction)") + .expect("read first chunk header"); + assert_eq!(first[0], crate::wire::STREAM_STDOUT); + drop(send); + } +}