From 4324a96034d7bbdf83d843173e54c341387ed3ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Moreau?= Date: Mon, 13 Jul 2026 09:16:15 -0400 Subject: [PATCH 1/4] feat(protocols): add NOW execution client Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Cargo.lock | 10 + Cargo.toml | 2 +- protocols/rust/now-client/Cargo.toml | 21 + protocols/rust/now-client/README.md | 17 + protocols/rust/now-client/src/capabilities.rs | 94 ++ protocols/rust/now-client/src/client.rs | 1092 +++++++++++++++++ protocols/rust/now-client/src/config.rs | 88 ++ protocols/rust/now-client/src/error.rs | 83 ++ protocols/rust/now-client/src/exec.rs | 494 ++++++++ protocols/rust/now-client/src/frame.rs | 141 +++ protocols/rust/now-client/src/lib.rs | 18 + 11 files changed, 2059 insertions(+), 1 deletion(-) create mode 100644 protocols/rust/now-client/Cargo.toml create mode 100644 protocols/rust/now-client/README.md create mode 100644 protocols/rust/now-client/src/capabilities.rs create mode 100644 protocols/rust/now-client/src/client.rs create mode 100644 protocols/rust/now-client/src/config.rs create mode 100644 protocols/rust/now-client/src/error.rs create mode 100644 protocols/rust/now-client/src/exec.rs create mode 100644 protocols/rust/now-client/src/frame.rs create mode 100644 protocols/rust/now-client/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 41d05be..30285e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -607,6 +607,15 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "now-client" +version = "0.1.0" +dependencies = [ + "now-proto-pdu", + "thiserror", + "tokio", +] + [[package]] name = "now-policy" version = "0.1.0" @@ -1063,6 +1072,7 @@ version = "1.52.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" dependencies = [ + "bytes", "pin-project-lite", "tokio-macros", ] diff --git a/Cargo.toml b/Cargo.toml index aa3ba62..8aa6e1c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ keywords = ["rdp", "remote-desktop", "network", "client", "protocol"] categories = ["network-programming"] [workspace.dependencies] -now-proto-pdu = { version = "0.1", path = "protocols/rust/now-proto-pdu" } +now-proto-pdu = { version = "=0.4.3", path = "protocols/rust/now-proto-pdu" } now-proto-fuzzing = { version = "0.1", path = "protocols/rust/now-proto-fuzzing" } [profile.test.package.proptest] diff --git a/protocols/rust/now-client/Cargo.toml b/protocols/rust/now-client/Cargo.toml new file mode 100644 index 0000000..6bf0a75 --- /dev/null +++ b/protocols/rust/now-client/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "now-client" +version = "0.1.0" +readme = "README.md" +description = "High-level Tokio client for NOW execution channels" +edition.workspace = true +license.workspace = true +homepage.workspace = true +repository.workspace = true +authors.workspace = true +keywords.workspace = true +categories.workspace = true +publish = true + +[lints] +workspace = true + +[dependencies] +now-proto-pdu = { workspace = true, features = ["std"] } +thiserror = "2" +tokio = { version = "1.52", features = ["io-util", "macros", "rt", "sync", "time"] } diff --git a/protocols/rust/now-client/README.md b/protocols/rust/now-client/README.md new file mode 100644 index 0000000..e5ceba0 --- /dev/null +++ b/protocols/rust/now-client/README.md @@ -0,0 +1,17 @@ +# NOW client + +`now-client` is a high-level, transport-agnostic Rust client for the NOW execution +channel. It accepts a caller-provided Tokio `AsyncRead + AsyncWrite` byte stream; +consumers create DVCs, pipes, sockets, and replacement clients after reconnects. + +Connect with `NowClient::connect`, then use the returned handle for Run, Process, +Batch, Windows PowerShell, or PowerShell 7. The client defensively negotiates +capabilities, applies bounded frame/command/event queues, and permits **one tracked +execution at a time per stream**. Tracked operations expose raw `Vec` stdout/stderr +chunks, stdin forwarding, normal cancellation, and terminal status. + +Run and detached requests are submission-only. Gateway may emit an immediate +Started/Data/Result sequence for Run; the client records and discards those matching +frames so they cannot affect the next tracked operation. This crate deliberately does +not provide Abort, Shell submission, DVC/pipe setup, reconnect policy, or retained +operation output. diff --git a/protocols/rust/now-client/src/capabilities.rs b/protocols/rust/now-client/src/capabilities.rs new file mode 100644 index 0000000..3b25ade --- /dev/null +++ b/protocols/rust/now-client/src/capabilities.rs @@ -0,0 +1,94 @@ +use core::time::Duration; + +use now_proto_pdu::{NowChannelCapsetMsg, NowExecCapsetFlags, NowProtoVersion}; + +use crate::NowClientError; + +/// Capabilities mutually supported by the client and its connected peer. +/// +/// The value is always calculated as an intersection with the local advertised capset, +/// even if the peer echoes unsupported flags. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NegotiatedCapabilities { + capset: NowChannelCapsetMsg, +} + +impl NegotiatedCapabilities { + pub(crate) fn negotiate( + requested: &NowChannelCapsetMsg, + peer: &NowChannelCapsetMsg, + ) -> Result { + if requested.version().major != peer.version().major { + return Err(NowClientError::IncompatibleVersion { + client: requested.version(), + peer: peer.version(), + }); + } + + Ok(Self { + capset: requested.downgrade(peer), + }) + } + + /// Returns the negotiated capability-set PDU. + pub fn capset(&self) -> &NowChannelCapsetMsg { + &self.capset + } + + /// Returns the negotiated NOW protocol version. + pub fn version(&self) -> NowProtoVersion { + self.capset.version() + } + + /// Returns the negotiated heartbeat interval, if both peers requested one. + pub fn heartbeat_interval(&self) -> Option { + self.capset.heartbeat_interval() + } + + /// Returns whether the generic Run style is available. + pub fn supports_run(&self) -> bool { + self.has(NowExecCapsetFlags::STYLE_RUN) + } + + /// Returns whether CreateProcess execution is available. + pub fn supports_process(&self) -> bool { + self.has(NowExecCapsetFlags::STYLE_PROCESS) + } + + /// Returns whether Batch execution is available. + pub fn supports_batch(&self) -> bool { + self.has(NowExecCapsetFlags::STYLE_BATCH) + } + + /// Returns whether Windows PowerShell execution is available. + pub fn supports_win_ps(&self) -> bool { + self.has(NowExecCapsetFlags::STYLE_WINPS) + } + + /// Returns whether PowerShell 7 execution is available. + pub fn supports_pwsh(&self) -> bool { + self.has(NowExecCapsetFlags::STYLE_PWSH) + } + + /// Returns whether tracked I/O redirection is available. + pub fn supports_io_redirection(&self) -> bool { + self.has(NowExecCapsetFlags::IO_REDIRECTION) && self.at_least(1, 3) + } + + /// Returns whether UTF-8 and Unicode-console encoding controls are available. + pub fn supports_unicode_console(&self) -> bool { + self.has(NowExecCapsetFlags::UNICODE_CONSOLE) && self.version().supports_exec_unicode_console() + } + + pub(crate) fn supports_detached(&self) -> bool { + self.at_least(1, 4) + } + + fn has(&self, capability: NowExecCapsetFlags) -> bool { + self.capset.exec_capset().contains(capability) + } + + fn at_least(&self, major: u16, minor: u16) -> bool { + self.version() >= NowProtoVersion { major, minor } + } +} diff --git a/protocols/rust/now-client/src/client.rs b/protocols/rust/now-client/src/client.rs new file mode 100644 index 0000000..d0b8b02 --- /dev/null +++ b/protocols/rust/now-client/src/client.rs @@ -0,0 +1,1092 @@ +use now_proto_pdu::ironrdp_core::encode_vec; +use now_proto_pdu::{ + NowChannelCapsetMsg, NowChannelMessage, NowExecDataStreamKind, NowExecMessage, NowMessage, OwnedNowMessage, +}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, RwLock}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::sync::{mpsc, oneshot}; +use tokio::time::{self, Instant}; + +use crate::exec::{stdin_message, EncodedRequest, RequestSpec}; +use crate::{NegotiatedCapabilities, NowClientConfig, NowClientError}; + +/// Entry point for connecting a Tokio byte stream to the NOW execution protocol. +pub struct NowClient; + +impl NowClient { + /// Negotiates NOW capabilities over `stream` and starts its single-owner protocol worker. + pub async fn connect( + stream: S, + config: NowClientConfig, + ) -> Result<(NowClientHandle, NegotiatedCapabilities), NowClientError> + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + config.validate()?; + let (stream, messages, peer_capset) = handshake(stream, &config).await?; + let capabilities = NegotiatedCapabilities::negotiate(&config.client_capset, &peer_capset)?; + let shared_capabilities = Arc::new(RwLock::new(capabilities.clone())); + let (command_sender, command_receiver) = mpsc::channel(config.command_queue_capacity); + let worker_sender = command_sender.downgrade(); + let worker = OwnedWorker { + stream, + messages, + requested_capset: config.client_capset.clone(), + capabilities: Arc::clone(&shared_capabilities), + command_receiver, + command_sender: worker_sender, + operations: HashMap::new(), + discarded_run_sessions: HashSet::new(), + run_discard_capacity: config.run_discard_capacity, + next_session_id: 1, + read_buffer: vec![0; config.read_buffer_size], + }; + + tokio::spawn(async move { + worker.run().await; + }); + + Ok(( + NowClientHandle { + command_sender, + capabilities: shared_capabilities, + event_queue_capacity: config.event_queue_capacity, + }, + capabilities, + )) + } +} + +/// Cloneable command handle for one connected NOW channel. +#[derive(Clone)] +pub struct NowClientHandle { + command_sender: mpsc::Sender, + capabilities: Arc>, + event_queue_capacity: usize, +} + +impl NowClientHandle { + /// Returns the latest negotiated capabilities, including capset refreshes from the peer. + pub fn capabilities(&self) -> NegotiatedCapabilities { + self.capabilities + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + /// Submits an untracked Run request. + pub async fn run(&self, request: crate::RunRequest) -> Result { + match self.submit(RequestSpec::Run(request)).await? { + ExecutionSubmission::Detached(execution) => Ok(execution), + ExecutionSubmission::Tracked(_) => Err(NowClientError::Protocol( + "Run request unexpectedly started a tracked operation".to_owned(), + )), + } + } + + /// Submits a Process request. + pub async fn process(&self, request: crate::ProcessRequest) -> Result { + self.submit(RequestSpec::Process(request)).await + } + + /// Submits a Batch request. + pub async fn batch(&self, request: crate::BatchRequest) -> Result { + self.submit(RequestSpec::Batch(request)).await + } + + /// Submits a Windows PowerShell request. + pub async fn win_ps(&self, request: crate::WinPsRequest) -> Result { + self.submit(RequestSpec::WinPs(request)).await + } + + /// Submits a PowerShell 7 request. + pub async fn pwsh(&self, request: crate::PwshRequest) -> Result { + self.submit(RequestSpec::Pwsh(request)).await + } + + async fn submit(&self, spec: RequestSpec) -> Result { + let tracked = spec.is_tracked(); + let (start_sender, start_receiver) = oneshot::channel(); + let (registration, events, terminal) = if tracked { + let (event_sender, event_receiver) = mpsc::channel(self.event_queue_capacity); + let (terminal_sender, terminal_receiver) = oneshot::channel(); + ( + Some(OperationRegistration { + event_sender, + terminal_sender, + }), + Some(event_receiver), + Some(terminal_receiver), + ) + } else { + (None, None, None) + }; + + self.command_sender + .send(WorkerCommand::Start { + spec, + registration, + response: start_sender, + }) + .await + .map_err(|_| NowClientError::WorkerClosed("command queue is closed".to_owned()))?; + let session_id = start_receiver + .await + .map_err(|_| NowClientError::WorkerClosed("worker stopped while starting operation".to_owned()))??; + + if tracked { + let events = events.ok_or_else(|| { + NowClientError::Protocol("tracked operation did not receive an event receiver".to_owned()) + })?; + let terminal = terminal.ok_or_else(|| { + NowClientError::Protocol("tracked operation did not receive a terminal receiver".to_owned()) + })?; + Ok(ExecutionSubmission::Tracked(Execution { + session_id, + events, + terminal, + command_sender: self.command_sender.clone(), + })) + } else { + Ok(ExecutionSubmission::Detached(DetachedExecution { session_id })) + } + } +} + +/// Either a tracked execution handle or a detached submission identity. +pub enum ExecutionSubmission { + /// A tracked operation that emits output and has a terminal result. + Tracked(Execution), + /// A detached operation that only reports local submission. + Detached(DetachedExecution), +} + +impl ExecutionSubmission { + /// Returns the NOW session identity allocated for the submission. + pub fn id(&self) -> u32 { + match self { + Self::Tracked(execution) => execution.id(), + Self::Detached(execution) => execution.id(), + } + } +} + +/// Fire-and-forget NOW execution submission. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DetachedExecution { + session_id: u32, +} + +impl DetachedExecution { + /// Returns the NOW session identity allocated for this submission. + pub fn id(&self) -> u32 { + self.session_id + } +} + +/// A tracked NOW execution. +pub struct Execution { + session_id: u32, + events: mpsc::Receiver, + terminal: oneshot::Receiver>, + command_sender: mpsc::Sender, +} + +impl Execution { + /// Returns the NOW session identity allocated for this execution. + pub fn id(&self) -> u32 { + self.session_id + } + + /// Receives the next execution event, or `None` when the event stream is closed. + pub async fn next_event(&mut self) -> Option { + self.events.recv().await + } + + /// Forwards raw bytes to the remote standard input stream. + pub async fn send_stdin(&self, data: Vec, last: bool) -> Result<(), NowClientError> { + let (response_sender, response_receiver) = oneshot::channel(); + self.command_sender + .send(WorkerCommand::SendStdin { + session_id: self.session_id, + data, + last, + response: response_sender, + }) + .await + .map_err(|_| NowClientError::WorkerClosed("command queue is closed".to_owned()))?; + response_receiver + .await + .map_err(|_| NowClientError::WorkerClosed("worker stopped while writing stdin".to_owned()))? + } + + /// Sends a cancel request and waits until the peer responds to it. + /// + /// Call [`Self::wait`] afterwards to observe the matching terminal result. + pub async fn cancel(&self) -> Result<(), NowClientError> { + let (response_sender, response_receiver) = oneshot::channel(); + self.command_sender + .send(WorkerCommand::Cancel { + session_id: self.session_id, + response: Some(response_sender), + }) + .await + .map_err(|_| NowClientError::WorkerClosed("command queue is closed".to_owned()))?; + response_receiver + .await + .map_err(|_| NowClientError::WorkerClosed("worker stopped while waiting for cancel response".to_owned()))? + } + + /// Waits for the remote terminal result. + /// + /// A nonzero remote exit code is returned as [`ExecutionStatus::Completed`], not an error. + pub async fn wait(self) -> Result { + self.terminal + .await + .map_err(|_| NowClientError::WorkerClosed("worker stopped before terminal result".to_owned()))? + } +} + +/// Events emitted by a tracked execution. +#[derive(Debug, PartialEq, Eq)] +pub enum ExecutionEvent { + /// The peer accepted and started the execution. + Started, + /// Raw stdout bytes and their final-stream marker. + Stdout { + /// Bytes exactly as received from NOW. + data: Vec, + /// Whether this is the final stdout chunk. + last: bool, + }, + /// Raw stderr bytes and their final-stream marker. + Stderr { + /// Bytes exactly as received from NOW. + data: Vec, + /// Whether this is the final stderr chunk. + last: bool, + }, + /// The peer accepted a previously requested cancellation. + CancelAccepted, +} + +/// Terminal state of a tracked execution. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ExecutionStatus { + /// The peer completed the command with this exit code. + Completed { + /// Process exit code, including nonzero values. + exit_code: u32, + }, + /// The peer accepted cancellation and later returned a terminal result. + Cancelled, +} + +struct OperationRegistration { + event_sender: mpsc::Sender, + terminal_sender: oneshot::Sender>, +} + +struct Operation { + event_sender: mpsc::Sender, + terminal_sender: oneshot::Sender>, + cancel_response: Option>>, + cancel_pending: bool, + cancel_accepted: bool, + stdin_closed: bool, +} + +#[derive(Default)] +struct BufferUpdate { + capset_updated: bool, + heartbeat_received: bool, +} + +impl BufferUpdate { + fn merge(&mut self, other: Self) { + self.capset_updated |= other.capset_updated; + self.heartbeat_received |= other.heartbeat_received; + } +} + +enum WorkerCommand { + Start { + spec: RequestSpec, + registration: Option, + response: oneshot::Sender>, + }, + SendStdin { + session_id: u32, + data: Vec, + last: bool, + response: oneshot::Sender>, + }, + Cancel { + session_id: u32, + response: Option>>, + }, +} + +struct OwnedWorker { + stream: S, + messages: crate::frame::MessageBuffer, + requested_capset: NowChannelCapsetMsg, + capabilities: Arc>, + command_receiver: mpsc::Receiver, + command_sender: mpsc::WeakSender, + operations: HashMap, + discarded_run_sessions: HashSet, + run_discard_capacity: usize, + next_session_id: u32, + read_buffer: Vec, +} + +impl OwnedWorker +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + async fn run(mut self) { + let mut heartbeat_deadline = self.heartbeat_deadline(); + loop { + match self.process_buffer() { + Ok(update) => { + if update.capset_updated || update.heartbeat_received { + heartbeat_deadline = self.heartbeat_deadline(); + } + } + Err(error) => { + self.fail_all(&error); + return; + } + } + + enum Event { + Command(Option), + Read(std::io::Result), + HeartbeatTimeout, + } + + let event = { + let heartbeat = async { + if let Some(deadline) = heartbeat_deadline { + time::sleep_until(deadline).await; + } else { + core::future::pending::<()>().await; + } + }; + tokio::pin!(heartbeat); + tokio::select! { + command = self.command_receiver.recv() => Event::Command(command), + read = self.stream.read(&mut self.read_buffer) => Event::Read(read), + _ = &mut heartbeat => Event::HeartbeatTimeout, + } + }; + + let result = match event { + Event::Command(Some(command)) => self.handle_command(command).await, + Event::Command(None) => { + let error = NowClientError::WorkerClosed("all client handles were dropped".to_owned()); + self.fail_all(&error); + return; + } + Event::Read(Ok(0)) => Err(NowClientError::WorkerClosed("transport reached EOF".to_owned())), + Event::Read(Ok(read)) => self.messages.push(&self.read_buffer[..read]), + Event::Read(Err(error)) => Err(error.into()), + Event::HeartbeatTimeout => Err(NowClientError::WorkerClosed("peer heartbeat timed out".to_owned())), + }; + if let Err(error) = result { + self.fail_all(&error); + return; + } + } + } + + fn heartbeat_deadline(&self) -> Option { + self.capabilities() + .heartbeat_interval() + .filter(|interval| !interval.is_zero()) + .map(|interval| Instant::now() + interval.saturating_mul(2)) + } + + fn capabilities(&self) -> NegotiatedCapabilities { + self.capabilities + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + fn process_buffer(&mut self) -> Result { + let messages = self.messages.take_ready(); + let mut update = BufferUpdate::default(); + for message in messages { + update.merge(self.dispatch_message(message)?); + } + Ok(update) + } + + fn dispatch_message(&mut self, message: OwnedNowMessage) -> Result { + match message { + NowMessage::Channel(NowChannelMessage::Capset(capset)) => { + let capabilities = NegotiatedCapabilities::negotiate(&self.requested_capset, &capset)?; + *self + .capabilities + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = capabilities; + Ok(BufferUpdate { + capset_updated: true, + heartbeat_received: false, + }) + } + NowMessage::Channel(NowChannelMessage::Close(_)) => { + Err(NowClientError::WorkerClosed("peer closed the NOW channel".to_owned())) + } + NowMessage::Exec(NowExecMessage::Started(message)) => { + let session_id = message.session_id(); + if !self.discarded_run_sessions.contains(&session_id) { + self.emit_event(session_id, ExecutionEvent::Started); + } + Ok(BufferUpdate::default()) + } + NowMessage::Exec(NowExecMessage::Data(message)) => { + if self.discarded_run_sessions.contains(&message.session_id()) { + return Ok(BufferUpdate::default()); + } + let event = match message + .stream_kind() + .map_err(|error| NowClientError::PduDecode(error.to_string()))? + { + NowExecDataStreamKind::Stdout => Some(ExecutionEvent::Stdout { + data: message.data().to_vec(), + last: message.is_last(), + }), + NowExecDataStreamKind::Stderr => Some(ExecutionEvent::Stderr { + data: message.data().to_vec(), + last: message.is_last(), + }), + NowExecDataStreamKind::Stdin => None, + }; + if let Some(event) = event { + self.emit_event(message.session_id(), event); + } + Ok(BufferUpdate::default()) + } + NowMessage::Exec(NowExecMessage::CancelRsp(message)) => { + let session_id = message.session_id(); + match message.to_result() { + Ok(()) => { + self.emit_event(session_id, ExecutionEvent::CancelAccepted); + if let Some(operation) = self.operations.get_mut(&session_id) { + operation.cancel_accepted = true; + if let Some(response) = operation.cancel_response.take() { + let _ = response.send(Ok(())); + } + } + } + Err(error) => { + if let Some(operation) = self.operations.get_mut(&session_id) { + operation.cancel_pending = false; + if let Some(response) = operation.cancel_response.take() { + let _ = response.send(Err(NowClientError::RemoteStatus { session_id, error })); + } + } + } + } + Ok(BufferUpdate::default()) + } + NowMessage::Exec(NowExecMessage::Result(message)) => { + let session_id = message.session_id(); + if self.discarded_run_sessions.remove(&session_id) { + return Ok(BufferUpdate::default()); + } + match message.to_result() { + Ok(exit_code) => { + let status = if self + .operations + .get(&session_id) + .is_some_and(|operation| operation.cancel_accepted) + { + ExecutionStatus::Cancelled + } else { + ExecutionStatus::Completed { exit_code } + }; + self.finish(session_id, Ok(status)); + } + Err(error) => self.finish(session_id, Err(NowClientError::RemoteStatus { session_id, error })), + } + Ok(BufferUpdate::default()) + } + NowMessage::Channel(NowChannelMessage::Heartbeat(_)) => Ok(BufferUpdate { + capset_updated: false, + heartbeat_received: true, + }), + _ => Ok(BufferUpdate::default()), + } + } + + async fn handle_command(&mut self, command: WorkerCommand) -> Result<(), NowClientError> { + match command { + WorkerCommand::Start { + spec, + registration, + response, + } => { + let tracked = spec.is_tracked(); + let is_run = spec.is_run(); + if tracked != registration.is_some() { + let _ = response.send(Err(NowClientError::Protocol( + "worker received inconsistent operation registration".to_owned(), + ))); + return Ok(()); + } + if tracked && !self.operations.is_empty() { + let _ = response.send(Err(NowClientError::OperationInProgress)); + return Ok(()); + } + if is_run && self.discarded_run_sessions.len() == self.run_discard_capacity { + let _ = response.send(Err(NowClientError::RunDiscardQueueFull)); + return Ok(()); + } + let session_id = self.allocate_session_id(); + let request = match spec.build(session_id, &self.capabilities()) { + Ok(request) => request, + Err(error) => { + let _ = response.send(Err(error)); + return Ok(()); + } + }; + let initial_stdin = match spec + .initial_stdin() + .map(|data| stdin_message(session_id, data.to_vec(), true)) + .transpose() + { + Ok(stdin) => stdin, + Err(error) => { + let _ = response.send(Err(error)); + return Ok(()); + } + }; + let stdin_closed = initial_stdin.is_some(); + + if let Err(error) = self.write_message(request).await { + let response_error = NowClientError::WorkerClosed(error.to_string()); + let _ = response.send(Err(response_error)); + return Err(error); + } + if let Some(stdin) = initial_stdin { + if let Err(error) = self + .write_message(EncodedRequest { + message: stdin, + non_interactive: false, + }) + .await + { + let response_error = NowClientError::WorkerClosed(error.to_string()); + let _ = response.send(Err(response_error)); + return Err(error); + } + } + + if let Some(registration) = registration { + self.operations.insert( + session_id, + Operation { + event_sender: registration.event_sender, + terminal_sender: registration.terminal_sender, + cancel_response: None, + cancel_pending: false, + cancel_accepted: false, + stdin_closed, + }, + ); + if let Some(timeout) = spec.timeout() { + if let Some(command_sender) = self.command_sender.upgrade() { + tokio::spawn(async move { + time::sleep(timeout).await; + let _ = command_sender + .send(WorkerCommand::Cancel { + session_id, + response: None, + }) + .await; + }); + } + } + } + if is_run { + self.discarded_run_sessions.insert(session_id); + } + let _ = response.send(Ok(session_id)); + Ok(()) + } + WorkerCommand::SendStdin { + session_id, + data, + last, + response, + } => { + let stdin_closed = match self.operations.get(&session_id) { + Some(operation) => operation.stdin_closed, + None => { + let _ = response.send(Err(NowClientError::OperationFinished { session_id })); + return Ok(()); + } + }; + if stdin_closed { + let _ = response.send(Err(NowClientError::InvalidRequest( + "standard input is already closed".to_owned(), + ))); + return Ok(()); + } + match stdin_message(session_id, data, last) { + Ok(message) => match self + .write_message(EncodedRequest { + message, + non_interactive: false, + }) + .await + { + Ok(()) => { + if last { + if let Some(operation) = self.operations.get_mut(&session_id) { + operation.stdin_closed = true; + } + } + let _ = response.send(Ok(())); + Ok(()) + } + Err(error) => { + let response_error = NowClientError::WorkerClosed(error.to_string()); + let _ = response.send(Err(response_error)); + Err(error) + } + }, + Err(error) => { + let _ = response.send(Err(error)); + Ok(()) + } + } + } + WorkerCommand::Cancel { session_id, response } => { + let already_pending = match self.operations.get(&session_id) { + Some(operation) => operation.cancel_pending, + None => { + if let Some(response) = response { + let _ = response.send(Err(NowClientError::OperationFinished { session_id })); + } + return Ok(()); + } + }; + if already_pending { + if let Some(response) = response { + let _ = response.send(Err(NowClientError::InvalidRequest( + "cancellation is already pending".to_owned(), + ))); + } + return Ok(()); + } + + let message = now_proto_pdu::NowExecCancelReqMsg::new(session_id).into(); + match self + .write_message(EncodedRequest { + message, + non_interactive: false, + }) + .await + { + Ok(()) => { + if let Some(operation) = self.operations.get_mut(&session_id) { + operation.cancel_pending = true; + operation.cancel_response = response; + } + Ok(()) + } + Err(error) => { + if let Some(response) = response { + let _ = response.send(Err(NowClientError::WorkerClosed(error.to_string()))); + } + Err(error) + } + } + } + } + } + + async fn write_message(&mut self, request: EncodedRequest) -> Result<(), NowClientError> { + let mut bytes = encode_vec(&request.message).map_err(|error| NowClientError::PduEncode(error.to_string()))?; + if request.non_interactive { + // now-proto-pdu 0.4.3 exposes the PowerShell non-interactive flag for decoding but + // intentionally has no setter. The common NOW header stores flags at bytes 6..8. + let flags = u16::from_le_bytes([bytes[6], bytes[7]]) | 0x0020; + bytes[6..8].copy_from_slice(&flags.to_le_bytes()); + } + self.stream.write_all(&bytes).await?; + self.stream.flush().await?; + Ok(()) + } + + fn allocate_session_id(&mut self) -> u32 { + loop { + let session_id = self.next_session_id; + self.next_session_id = self.next_session_id.wrapping_add(1); + if self.next_session_id == 0 { + self.next_session_id = 1; + } + if session_id != 0 + && !self.operations.contains_key(&session_id) + && !self.discarded_run_sessions.contains(&session_id) + { + return session_id; + } + } + } + + fn emit_event(&mut self, session_id: u32, event: ExecutionEvent) { + let overflowed = match self.operations.get_mut(&session_id) { + Some(operation) => match operation.event_sender.try_send(event) { + Ok(()) => false, + Err(mpsc::error::TrySendError::Full(_)) => true, + Err(mpsc::error::TrySendError::Closed(_)) => false, + }, + None => false, + }; + if overflowed { + self.finish(session_id, Err(NowClientError::EventQueueFull { session_id })); + } + } + + fn finish(&mut self, session_id: u32, result: Result) { + if let Some(mut operation) = self.operations.remove(&session_id) { + if let Some(cancel_response) = operation.cancel_response.take() { + let _ = cancel_response.send(Err(NowClientError::OperationFinished { session_id })); + } + let _ = operation.terminal_sender.send(result); + } + } + + fn fail_all(&mut self, error: &NowClientError) { + let reason = error.to_string(); + for (session_id, mut operation) in self.operations.drain() { + if let Some(cancel_response) = operation.cancel_response.take() { + let _ = cancel_response.send(Err(NowClientError::WorkerClosed(reason.clone()))); + } + let _ = operation.terminal_sender.send(Err(NowClientError::WorkerClosed(format!( + "{reason} (session {session_id})" + )))); + } + } +} + +async fn handshake( + mut stream: S, + config: &NowClientConfig, +) -> Result<(S, crate::frame::MessageBuffer, NowChannelCapsetMsg), NowClientError> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let capset: NowMessage<'static> = config.client_capset.clone().into(); + let bytes = encode_vec(&capset).map_err(|error| NowClientError::PduEncode(error.to_string()))?; + stream.write_all(&bytes).await?; + stream.flush().await?; + + let receive_capset = async { + let mut messages = crate::frame::MessageBuffer::new(config.max_frame_body_size); + let mut read_buffer = vec![0; config.read_buffer_size]; + loop { + let read = stream.read(&mut read_buffer).await?; + if read == 0 { + return Err(NowClientError::WorkerClosed( + "transport reached EOF during capability handshake".to_owned(), + )); + } + messages.push(&read_buffer[..read])?; + let decoded = messages.take_ready(); + let mut decoded = decoded.into_iter(); + while let Some(message) = decoded.next() { + match message { + NowMessage::Channel(NowChannelMessage::Heartbeat(_)) => {} + NowMessage::Channel(NowChannelMessage::Capset(capset)) => { + messages.restore_ready(decoded); + return Ok((messages, capset)); + } + _ => { + return Err(NowClientError::Protocol( + "expected peer capability set during handshake".to_owned(), + )) + } + } + } + } + }; + + match time::timeout(config.connect_timeout, receive_capset).await { + Ok(result) => { + let (messages, capset) = result?; + Ok((stream, messages, capset)) + } + Err(_) => Err(NowClientError::HandshakeTimeout), + } +} + +#[cfg(test)] +mod tests { + use now_proto_pdu::ironrdp_core::{encode_vec, Decode, IntoOwned, ReadCursor}; + use now_proto_pdu::{ + NowChannelCapsetMsg, NowChannelHeartbeatMsg, NowExecCancelRspMsg, NowExecCapsetFlags, NowExecDataMsg, + NowExecDataStreamKind, NowExecMessage, NowExecResultMsg, NowExecStartedMsg, NowMessage, + }; + use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + + use super::{ExecutionEvent, ExecutionStatus, ExecutionSubmission, NowClient}; + use crate::{NowClientConfig, NowClientError, ProcessRequest, RunRequest}; + + fn encode(message: impl Into>) -> Vec { + match encode_vec(&message.into()) { + Ok(bytes) => bytes, + Err(error) => panic!("test PDU must encode: {error}"), + } + } + + async fn read_message(stream: &mut S) -> NowMessage<'static> + where + S: AsyncRead + Unpin, + { + let mut header = [0; 8]; + if let Err(error) = stream.read_exact(&mut header).await { + panic!("test peer must receive a header: {error}"); + } + let body_size = u32::from_le_bytes([header[0], header[1], header[2], header[3]]) as usize; + let mut bytes = header.to_vec(); + bytes.resize(8 + body_size, 0); + if let Err(error) = stream.read_exact(&mut bytes[8..]).await { + panic!("test peer must receive a body: {error}"); + } + let mut cursor = ReadCursor::new(&bytes); + match NowMessage::decode(&mut cursor) { + Ok(message) => message.into_owned(), + Err(error) => panic!("test peer must decode a PDU: {error}"), + } + } + + async fn write_message(stream: &mut S, message: impl Into>) + where + S: AsyncWrite + Unpin, + { + let bytes = encode(message); + if let Err(error) = stream.write_all(&bytes).await { + panic!("test peer must write a PDU: {error}"); + } + } + + fn capset(flags: NowExecCapsetFlags) -> NowChannelCapsetMsg { + NowChannelCapsetMsg::default().with_exec_capset(flags) + } + + #[tokio::test] + async fn handshake_rejects_an_incompatible_major_version() { + let (client_stream, mut peer_stream) = tokio::io::duplex(1024); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + let mut bytes = encode(capset(NowExecCapsetFlags::STYLE_RUN)); + bytes[8..10].copy_from_slice(&2u16.to_le_bytes()); + if let Err(error) = peer_stream.write_all(&bytes).await { + panic!("test peer must write an incompatible capset: {error}"); + } + }); + + let result = NowClient::connect(client_stream, NowClientConfig::default()).await; + assert!(matches!(result, Err(NowClientError::IncompatibleVersion { .. }))); + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + + #[tokio::test] + async fn handshake_intersects_peer_capabilities_defensively() { + let (client_stream, mut peer_stream) = tokio::io::duplex(1024); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message(&mut peer_stream, NowChannelHeartbeatMsg::default()).await; + write_message( + &mut peer_stream, + capset( + NowExecCapsetFlags::STYLE_RUN + | NowExecCapsetFlags::STYLE_PROCESS + | NowExecCapsetFlags::IO_REDIRECTION, + ), + ) + .await; + }); + + let config = NowClientConfig { + client_capset: capset(NowExecCapsetFlags::STYLE_RUN), + ..NowClientConfig::default() + }; + let (handle, capabilities) = match NowClient::connect(client_stream, config).await { + Ok(connection) => connection, + Err(error) => panic!("handshake must succeed: {error}"), + }; + assert!(capabilities.supports_run()); + assert!(!capabilities.supports_process()); + assert!(!handle.capabilities().supports_process()); + drop(handle); + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + + #[tokio::test] + async fn run_frames_are_discarded_before_the_following_process() { + let (client_stream, mut peer_stream) = tokio::io::duplex(4096); + let (release_sender, release_receiver) = tokio::sync::oneshot::channel(); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message( + &mut peer_stream, + capset( + NowExecCapsetFlags::STYLE_RUN + | NowExecCapsetFlags::STYLE_PROCESS + | NowExecCapsetFlags::IO_REDIRECTION, + ), + ) + .await; + + let run_session = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Run(message)) => message.session_id(), + message => panic!("expected Run request, got {message:?}"), + }; + write_message(&mut peer_stream, NowExecStartedMsg::new(run_session)).await; + let run_data = match NowExecDataMsg::new(run_session, NowExecDataStreamKind::Stdout, true, vec![0xaa, 0xbb]) + { + Ok(message) => message, + Err(error) => panic!("test Run data PDU must encode: {error}"), + }; + write_message(&mut peer_stream, run_data).await; + write_message(&mut peer_stream, NowExecResultMsg::new_success(run_session, 0)).await; + + let process_session = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Process(message)) => message.session_id(), + message => panic!("expected Process request, got {message:?}"), + }; + match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Data(message)) => { + assert_eq!(message.session_id(), process_session); + assert_eq!(message.data(), [0x01, 0xff]); + assert!(message.is_last()); + } + message => panic!("expected Process stdin, got {message:?}"), + } + if release_receiver.await.is_err() { + panic!("test must release the Process response"); + } + write_message(&mut peer_stream, NowExecStartedMsg::new(process_session)).await; + let stdout = match NowExecDataMsg::new( + process_session, + NowExecDataStreamKind::Stdout, + true, + vec![0xff, 0x00, 0x80], + ) { + Ok(message) => message, + Err(error) => panic!("test Process data PDU must encode: {error}"), + }; + write_message(&mut peer_stream, stdout).await; + write_message(&mut peer_stream, NowExecResultMsg::new_success(process_session, 17)).await; + }); + + let (handle, _) = match NowClient::connect(client_stream, NowClientConfig::default()).await { + Ok(connection) => connection, + Err(error) => panic!("handshake must succeed: {error}"), + }; + let run = match handle.run(RunRequest::new("run.exe")).await { + Ok(submission) => submission, + Err(error) => panic!("Run request must be accepted: {error}"), + }; + assert_eq!(run.id(), 1); + let mut process = match handle + .process(ProcessRequest::new("process.exe").with_stdin(vec![0x01, 0xff])) + .await + { + Ok(ExecutionSubmission::Tracked(execution)) => execution, + Ok(ExecutionSubmission::Detached(_)) => panic!("default process request must be tracked"), + Err(error) => panic!("Process request must start: {error}"), + }; + match handle.process(ProcessRequest::new("second.exe")).await { + Err(NowClientError::OperationInProgress) => {} + Ok(_) => panic!("second tracked execution must be rejected"), + Err(error) => panic!("unexpected second execution error: {error}"), + } + if release_sender.send(()).is_err() { + panic!("test peer stopped before Process response"); + } + + assert_eq!(process.next_event().await, Some(ExecutionEvent::Started)); + assert_eq!( + process.next_event().await, + Some(ExecutionEvent::Stdout { + data: vec![0xff, 0x00, 0x80], + last: true, + }) + ); + match process.wait().await { + Ok(ExecutionStatus::Completed { exit_code: 17 }) => {} + Ok(status) => panic!("Process returned unexpected status: {status:?}"), + Err(error) => panic!("Process must complete: {error}"), + } + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + + #[tokio::test] + async fn cancellation_waits_for_the_matching_response_and_result() { + let (client_stream, mut peer_stream) = tokio::io::duplex(4096); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message( + &mut peer_stream, + capset(NowExecCapsetFlags::STYLE_PROCESS | NowExecCapsetFlags::IO_REDIRECTION), + ) + .await; + + let session_id = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Process(message)) => message.session_id(), + message => panic!("expected Process request, got {message:?}"), + }; + write_message(&mut peer_stream, NowExecStartedMsg::new(session_id)).await; + match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::CancelReq(message)) => assert_eq!(message.session_id(), session_id), + message => panic!("expected cancel request, got {message:?}"), + } + write_message(&mut peer_stream, NowExecCancelRspMsg::new_success(session_id)).await; + write_message(&mut peer_stream, NowExecResultMsg::new_success(session_id, 0)).await; + }); + + let (handle, _) = match NowClient::connect(client_stream, NowClientConfig::default()).await { + Ok(connection) => connection, + Err(error) => panic!("handshake must succeed: {error}"), + }; + let mut execution = match handle.process(ProcessRequest::new("process.exe")).await { + Ok(ExecutionSubmission::Tracked(execution)) => execution, + Ok(ExecutionSubmission::Detached(_)) => panic!("default process request must be tracked"), + Err(error) => panic!("Process request must start: {error}"), + }; + assert_eq!(execution.next_event().await, Some(ExecutionEvent::Started)); + match execution.cancel().await { + Ok(()) => {} + Err(error) => panic!("cancellation must be accepted: {error}"), + } + assert_eq!(execution.next_event().await, Some(ExecutionEvent::CancelAccepted)); + match execution.wait().await { + Ok(ExecutionStatus::Cancelled) => {} + Ok(status) => panic!("unexpected terminal status: {status:?}"), + Err(error) => panic!("cancellation must complete: {error}"), + } + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } +} diff --git a/protocols/rust/now-client/src/config.rs b/protocols/rust/now-client/src/config.rs new file mode 100644 index 0000000..d579b68 --- /dev/null +++ b/protocols/rust/now-client/src/config.rs @@ -0,0 +1,88 @@ +use core::time::Duration; + +use now_proto_pdu::{NowChannelCapsetMsg, NowExecCapsetFlags}; + +use crate::NowClientError; + +/// Limits and advertised capabilities used to establish a NOW client connection. +#[derive(Clone, Debug)] +pub struct NowClientConfig { + /// Capabilities sent to the peer during the initial handshake. + pub client_capset: NowChannelCapsetMsg, + /// Largest accepted PDU body, excluding the eight-byte NOW header. + pub max_frame_body_size: usize, + /// Maximum number of bytes requested from the transport in one read. + pub read_buffer_size: usize, + /// Bound for requests waiting to be written by the worker. + pub command_queue_capacity: usize, + /// Bound for unread events per tracked execution. + pub event_queue_capacity: usize, + /// Maximum number of Run submissions awaiting their terminal frame for disposal. + pub run_discard_capacity: usize, + /// Deadline for receiving the peer capability set. + pub connect_timeout: Duration, +} + +impl Default for NowClientConfig { + fn default() -> Self { + let exec_capset = NowExecCapsetFlags::STYLE_RUN + | NowExecCapsetFlags::STYLE_PROCESS + | NowExecCapsetFlags::STYLE_BATCH + | NowExecCapsetFlags::STYLE_WINPS + | NowExecCapsetFlags::STYLE_PWSH + | NowExecCapsetFlags::IO_REDIRECTION + | NowExecCapsetFlags::UNICODE_CONSOLE; + let capset = NowChannelCapsetMsg::default().with_exec_capset(exec_capset); + let client_capset = match capset.with_heartbeat_interval(Duration::from_secs(60)) { + Ok(capset) => capset, + Err(_) => unreachable!("the fixed default heartbeat interval is valid"), + }; + + Self { + client_capset, + max_frame_body_size: 16 * 1024 * 1024, + read_buffer_size: 16 * 1024, + command_queue_capacity: 64, + event_queue_capacity: 64, + run_discard_capacity: 64, + connect_timeout: Duration::from_secs(15), + } + } +} + +impl NowClientConfig { + pub(crate) fn validate(&self) -> Result<(), NowClientError> { + if self.max_frame_body_size == 0 { + return Err(NowClientError::InvalidConfiguration( + "max_frame_body_size must be greater than zero".to_owned(), + )); + } + if self.read_buffer_size == 0 { + return Err(NowClientError::InvalidConfiguration( + "read_buffer_size must be greater than zero".to_owned(), + )); + } + if self.command_queue_capacity == 0 { + return Err(NowClientError::InvalidConfiguration( + "command_queue_capacity must be greater than zero".to_owned(), + )); + } + if self.event_queue_capacity == 0 { + return Err(NowClientError::InvalidConfiguration( + "event_queue_capacity must be greater than zero".to_owned(), + )); + } + if self.run_discard_capacity == 0 { + return Err(NowClientError::InvalidConfiguration( + "run_discard_capacity must be greater than zero".to_owned(), + )); + } + if self.connect_timeout.is_zero() { + return Err(NowClientError::InvalidConfiguration( + "connect_timeout must be greater than zero".to_owned(), + )); + } + + Ok(()) + } +} diff --git a/protocols/rust/now-client/src/error.rs b/protocols/rust/now-client/src/error.rs new file mode 100644 index 0000000..838c31e --- /dev/null +++ b/protocols/rust/now-client/src/error.rs @@ -0,0 +1,83 @@ +use now_proto_pdu::{NowProtoVersion, NowStatusError}; +use thiserror::Error; + +/// Errors produced while connecting to or using a NOW execution channel. +#[derive(Debug, Error)] +pub enum NowClientError { + /// The underlying transport failed. + #[error("NOW transport I/O failed: {0}")] + Io(#[from] std::io::Error), + /// A NOW PDU could not be encoded. + #[error("failed to encode NOW PDU: {0}")] + PduEncode(String), + /// A NOW PDU could not be decoded. + #[error("failed to decode NOW PDU: {0}")] + PduDecode(String), + /// The peer uses a different NOW protocol major version. + #[error("incompatible NOW protocol major versions: client {client:?}, peer {peer:?}")] + IncompatibleVersion { + /// The version advertised by this client. + client: NowProtoVersion, + /// The version advertised by the peer. + peer: NowProtoVersion, + }, + /// The handshake did not complete before its configured deadline. + #[error("NOW capability handshake timed out")] + HandshakeTimeout, + /// A message arrived in an invalid protocol state. + #[error("NOW protocol violation: {0}")] + Protocol(String), + /// A PDU body exceeded the configured frame limit. + #[error("NOW frame body size {declared} exceeds configured maximum {maximum}")] + FrameTooLarge { + /// Body size declared by the PDU header. + declared: usize, + /// Configured maximum body size. + maximum: usize, + }, + /// An incomplete frame exceeded the bounded deframer storage. + #[error("NOW frame buffer exceeds configured maximum {maximum}")] + FrameBufferTooLarge { + /// Maximum retained frame size. + maximum: usize, + }, + /// The supplied client configuration is invalid. + #[error("invalid NOW client configuration: {0}")] + InvalidConfiguration(String), + /// The execution request is invalid before it was sent. + #[error("invalid NOW execution request: {0}")] + InvalidRequest(String), + /// The peer did not negotiate a capability required by the request. + #[error("peer does not support required NOW capability: {0}")] + UnsupportedCapability(&'static str), + /// A tracked execution is already active on this client stream. + #[error("NOW client already has an active tracked execution")] + OperationInProgress, + /// Too many Run submissions are awaiting their terminal frame for disposal. + #[error("NOW Run discard queue reached its configured capacity")] + RunDiscardQueueFull, + /// A remote NOW result or cancellation response reported an error status. + #[error("remote NOW operation {session_id} failed: {error}")] + RemoteStatus { + /// Session that reported the error. + session_id: u32, + /// Error status returned by the peer. + #[source] + error: NowStatusError, + }, + /// The worker stopped before the requested operation completed. + #[error("NOW worker is closed: {0}")] + WorkerClosed(String), + /// A caller did not drain an operation's bounded event queue. + #[error("NOW operation {session_id} event queue is full")] + EventQueueFull { + /// Session whose event queue overflowed. + session_id: u32, + }, + /// An operation is already terminal or no longer tracked. + #[error("NOW operation {session_id} is no longer active")] + OperationFinished { + /// Session that is no longer active. + session_id: u32, + }, +} diff --git a/protocols/rust/now-client/src/exec.rs b/protocols/rust/now-client/src/exec.rs new file mode 100644 index 0000000..87134b1 --- /dev/null +++ b/protocols/rust/now-client/src/exec.rs @@ -0,0 +1,494 @@ +use core::fmt::Display; +use core::time::Duration; + +use now_proto_pdu::{ + NowExecBatchMsg, NowExecDataMsg, NowExecDataStreamKind, NowExecProcessMsg, NowExecPwshMsg, NowExecRunMsg, + NowExecWinPsMsg, NowMessage, +}; + +use crate::{NegotiatedCapabilities, NowClientError}; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct CommonRequest { + command: String, + directory: Option, + detached: bool, + stdin: Option>, + timeout: Option, +} + +impl CommonRequest { + fn new(command: impl Into) -> Self { + Self { + command: command.into(), + directory: None, + detached: false, + stdin: None, + timeout: None, + } + } +} + +/// Request for the generic fire-and-forget Run style. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RunRequest { + command: String, + directory: Option, +} + +impl RunRequest { + /// Creates a Run request. + pub fn new(command: impl Into) -> Self { + Self { + command: command.into(), + directory: None, + } + } + + /// Sets the working directory. + #[must_use] + pub fn with_directory(mut self, directory: impl Into) -> Self { + self.directory = Some(directory.into()); + self + } +} + +/// Request for Windows CreateProcess execution. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProcessRequest { + filename: String, + parameters: Option, + directory: Option, + detached: bool, + stdin: Option>, + timeout: Option, +} + +impl ProcessRequest { + /// Creates a CreateProcess request. + pub fn new(filename: impl Into) -> Self { + Self { + filename: filename.into(), + parameters: None, + directory: None, + detached: false, + stdin: None, + timeout: None, + } + } + + /// Sets command-line parameters. + #[must_use] + pub fn with_parameters(mut self, parameters: impl Into) -> Self { + self.parameters = Some(parameters.into()); + self + } + + /// Sets the working directory. + #[must_use] + pub fn with_directory(mut self, directory: impl Into) -> Self { + self.directory = Some(directory.into()); + self + } + + /// Requests detached execution. + #[must_use] + pub fn detached(mut self) -> Self { + self.detached = true; + self + } + + /// Supplies the first and final stdin chunk. + #[must_use] + pub fn with_stdin(mut self, data: impl Into>) -> Self { + self.stdin = Some(data.into()); + self + } + + /// Sets the tracked execution timeout. + #[must_use] + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = Some(timeout); + self + } +} + +/// Request for a Windows Batch command. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BatchRequest { + common: CommonRequest, +} + +impl BatchRequest { + /// Creates a Batch request. + pub fn new(command: impl Into) -> Self { + Self { + common: CommonRequest::new(command), + } + } + + /// Sets the working directory. + #[must_use] + pub fn with_directory(mut self, directory: impl Into) -> Self { + self.common.directory = Some(directory.into()); + self + } + + /// Requests detached execution. + #[must_use] + pub fn detached(mut self) -> Self { + self.common.detached = true; + self + } + + /// Supplies the first and final stdin chunk. + #[must_use] + pub fn with_stdin(mut self, data: impl Into>) -> Self { + self.common.stdin = Some(data.into()); + self + } + + /// Sets the tracked execution timeout. + #[must_use] + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.common.timeout = Some(timeout); + self + } +} + +/// Request shared by Windows PowerShell and PowerShell 7. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PowerShellRequest { + common: CommonRequest, + no_profile: bool, + non_interactive: bool, +} + +impl PowerShellRequest { + /// Creates a PowerShell request. + pub fn new(command: impl Into) -> Self { + Self { + common: CommonRequest::new(command), + no_profile: false, + non_interactive: false, + } + } + + /// Sets the working directory. + #[must_use] + pub fn with_directory(mut self, directory: impl Into) -> Self { + self.common.directory = Some(directory.into()); + self + } + + /// Requests detached execution. + #[must_use] + pub fn detached(mut self) -> Self { + self.common.detached = true; + self + } + + /// Supplies the first and final stdin chunk. + #[must_use] + pub fn with_stdin(mut self, data: impl Into>) -> Self { + self.common.stdin = Some(data.into()); + self + } + + /// Sets the tracked execution timeout. + #[must_use] + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.common.timeout = Some(timeout); + self + } + + /// Requests PowerShell's `-NoProfile` option. + #[must_use] + pub fn with_no_profile(mut self) -> Self { + self.no_profile = true; + self + } + + /// Requests PowerShell's `-NonInteractive` option. + #[must_use] + pub fn with_non_interactive(mut self) -> Self { + self.non_interactive = true; + self + } +} + +/// Request for Windows PowerShell (`powershell.exe`). +pub type WinPsRequest = PowerShellRequest; +/// Request for PowerShell 7 (`pwsh`). +pub type PwshRequest = PowerShellRequest; + +pub(crate) enum RequestSpec { + Run(RunRequest), + Process(ProcessRequest), + Batch(BatchRequest), + WinPs(WinPsRequest), + Pwsh(PwshRequest), +} + +pub(crate) struct EncodedRequest { + pub(crate) message: NowMessage<'static>, + pub(crate) non_interactive: bool, +} + +impl RequestSpec { + pub(crate) fn is_run(&self) -> bool { + matches!(self, Self::Run(_)) + } + + pub(crate) fn is_tracked(&self) -> bool { + match self { + Self::Run(_) => false, + Self::Process(request) => !request.detached, + Self::Batch(request) => !request.common.detached, + Self::WinPs(request) | Self::Pwsh(request) => !request.common.detached, + } + } + + pub(crate) fn initial_stdin(&self) -> Option<&[u8]> { + match self { + Self::Run(_) => None, + Self::Process(request) => request.stdin.as_deref(), + Self::Batch(request) => request.common.stdin.as_deref(), + Self::WinPs(request) | Self::Pwsh(request) => request.common.stdin.as_deref(), + } + } + + pub(crate) fn timeout(&self) -> Option { + match self { + Self::Run(_) => None, + Self::Process(request) => request.timeout, + Self::Batch(request) => request.common.timeout, + Self::WinPs(request) | Self::Pwsh(request) => request.common.timeout, + } + } + + pub(crate) fn build( + &self, + session_id: u32, + capabilities: &NegotiatedCapabilities, + ) -> Result { + self.validate(capabilities)?; + + match self { + Self::Run(request) => { + let mut message = pdu(NowExecRunMsg::new(session_id, request.command.clone()))?; + if let Some(directory) = &request.directory { + message = pdu(message.with_directory(directory.clone()))?; + } + Ok(EncodedRequest { + message: message.into(), + non_interactive: false, + }) + } + Self::Process(request) => { + let mut message = pdu(NowExecProcessMsg::new(session_id, request.filename.clone()))?; + if let Some(parameters) = &request.parameters { + message = pdu(message.with_parameters(parameters.clone()))?; + } + if let Some(directory) = &request.directory { + message = pdu(message.with_directory(directory.clone()))?; + } + if capabilities.supports_unicode_console() { + message = message.with_encoding_utf8(); + } + if request.detached { + message = message.with_detached(); + } else { + message = message.with_io_redirection(); + } + Ok(EncodedRequest { + message: message.into(), + non_interactive: false, + }) + } + Self::Batch(request) => { + let common = &request.common; + let mut message = pdu(NowExecBatchMsg::new(session_id, common.command.clone()))?; + if let Some(directory) = &common.directory { + message = pdu(message.with_directory(directory.clone()))?; + } + if capabilities.supports_unicode_console() { + message = message.with_raw_encoding().with_unicode_console(); + } + if common.detached { + message = message.with_detached(); + } else { + message = message.with_io_redirection(); + } + Ok(EncodedRequest { + message: message.into(), + non_interactive: false, + }) + } + Self::WinPs(request) => build_power_shell(session_id, request, false, capabilities), + Self::Pwsh(request) => build_power_shell(session_id, request, true, capabilities), + } + } + + fn validate(&self, capabilities: &NegotiatedCapabilities) -> Result<(), NowClientError> { + match self { + Self::Run(request) => { + require(capabilities.supports_run(), "Run")?; + validate_command(&request.command)?; + validate_directory(request.directory.as_deref())?; + } + Self::Process(request) => { + require(capabilities.supports_process(), "Process")?; + validate_filename(&request.filename)?; + validate_optional(&request.parameters, "parameters")?; + validate_common(request.detached, request.stdin.as_deref(), request.timeout)?; + validate_tracking(capabilities, request.detached)?; + } + Self::Batch(request) => { + require(capabilities.supports_batch(), "Batch")?; + validate_common_request(&request.common, capabilities)?; + } + Self::WinPs(request) => { + require(capabilities.supports_win_ps(), "WinPs")?; + validate_common_request(&request.common, capabilities)?; + } + Self::Pwsh(request) => { + require(capabilities.supports_pwsh(), "Pwsh")?; + validate_common_request(&request.common, capabilities)?; + } + } + Ok(()) + } +} + +macro_rules! apply_power_shell_options { + ($message:ident, $common:ident, $request:ident, $capabilities:ident) => { + if let Some(directory) = &$common.directory { + $message = pdu($message.with_directory(directory.clone()))?; + } + if $request.no_profile { + $message = $message.set_no_profile(); + } + if $capabilities.supports_unicode_console() { + $message = $message.with_raw_encoding().with_unicode_console(); + } + if $common.detached { + $message = $message.with_detached(); + } else { + $message = $message.with_io_redirection(); + } + }; +} + +fn build_power_shell( + session_id: u32, + request: &PowerShellRequest, + pwsh: bool, + capabilities: &NegotiatedCapabilities, +) -> Result { + let common = &request.common; + if pwsh { + let mut message = pdu(NowExecPwshMsg::new(session_id, common.command.clone()))?; + apply_power_shell_options!(message, common, request, capabilities); + Ok(EncodedRequest { + message: message.into(), + non_interactive: request.non_interactive, + }) + } else { + let mut message = pdu(NowExecWinPsMsg::new(session_id, common.command.clone()))?; + apply_power_shell_options!(message, common, request, capabilities); + Ok(EncodedRequest { + message: message.into(), + non_interactive: request.non_interactive, + }) + } +} + +fn validate_common_request( + request: &CommonRequest, + capabilities: &NegotiatedCapabilities, +) -> Result<(), NowClientError> { + validate_command(&request.command)?; + validate_directory(request.directory.as_deref())?; + validate_common(request.detached, request.stdin.as_deref(), request.timeout)?; + validate_tracking(capabilities, request.detached) +} + +fn validate_common(detached: bool, stdin: Option<&[u8]>, timeout: Option) -> Result<(), NowClientError> { + if detached && stdin.is_some() { + return Err(NowClientError::InvalidRequest( + "detached execution cannot include inline stdin".to_owned(), + )); + } + if detached && timeout.is_some() { + return Err(NowClientError::InvalidRequest( + "detached execution cannot include a timeout".to_owned(), + )); + } + if timeout.is_some_and(|timeout| timeout.is_zero()) { + return Err(NowClientError::InvalidRequest( + "execution timeout must not be zero".to_owned(), + )); + } + Ok(()) +} + +fn validate_tracking(capabilities: &NegotiatedCapabilities, detached: bool) -> Result<(), NowClientError> { + if detached { + require(capabilities.supports_detached(), "detached execution") + } else { + require(capabilities.supports_io_redirection(), "I/O redirection") + } +} + +fn validate_command(command: &str) -> Result<(), NowClientError> { + if command.trim().is_empty() { + return Err(NowClientError::InvalidRequest("command must not be empty".to_owned())); + } + Ok(()) +} + +fn validate_filename(filename: &str) -> Result<(), NowClientError> { + if filename.trim().is_empty() { + return Err(NowClientError::InvalidRequest("filename must not be empty".to_owned())); + } + Ok(()) +} + +fn validate_directory(directory: Option<&str>) -> Result<(), NowClientError> { + if directory.is_some_and(str::is_empty) { + return Err(NowClientError::InvalidRequest( + "directory must not be empty when specified".to_owned(), + )); + } + Ok(()) +} + +fn validate_optional(value: &Option, name: &str) -> Result<(), NowClientError> { + if value.as_deref().is_some_and(str::is_empty) { + return Err(NowClientError::InvalidRequest(format!( + "{name} must not be empty when specified" + ))); + } + Ok(()) +} + +fn require(supported: bool, capability: &'static str) -> Result<(), NowClientError> { + supported + .then_some(()) + .ok_or(NowClientError::UnsupportedCapability(capability)) +} + +fn pdu(result: Result) -> Result { + result.map_err(|error| NowClientError::PduEncode(error.to_string())) +} + +pub(crate) fn stdin_message(session_id: u32, data: Vec, last: bool) -> Result, NowClientError> { + pdu(NowExecDataMsg::new( + session_id, + NowExecDataStreamKind::Stdin, + last, + data, + )) + .map(Into::into) +} diff --git a/protocols/rust/now-client/src/frame.rs b/protocols/rust/now-client/src/frame.rs new file mode 100644 index 0000000..65124ef --- /dev/null +++ b/protocols/rust/now-client/src/frame.rs @@ -0,0 +1,141 @@ +use now_proto_pdu::ironrdp_core::{Decode, IntoOwned, ReadCursor}; +use now_proto_pdu::{NowMessage, OwnedNowMessage}; + +use crate::NowClientError; + +/// Incrementally decodes bounded NOW PDU frames from a byte stream. +pub(crate) struct MessageBuffer { + bytes: Vec, + ready: Vec, + max_body_size: usize, +} + +impl MessageBuffer { + const HEADER_SIZE: usize = 8; + + pub(crate) fn new(max_body_size: usize) -> Self { + Self { + bytes: Vec::new(), + ready: Vec::new(), + max_body_size, + } + } + + pub(crate) fn push(&mut self, input: &[u8]) -> Result<(), NowClientError> { + let maximum = self + .max_body_size + .checked_add(Self::HEADER_SIZE) + .ok_or(NowClientError::FrameBufferTooLarge { + maximum: self.max_body_size, + })?; + let mut input = input; + loop { + if self.bytes.len() < Self::HEADER_SIZE { + if input.is_empty() { + break; + } + let bytes_to_copy = input.len().min(Self::HEADER_SIZE - self.bytes.len()); + self.bytes.extend_from_slice(&input[..bytes_to_copy]); + input = &input[bytes_to_copy..]; + continue; + } + + let body_size = u32::from_le_bytes([self.bytes[0], self.bytes[1], self.bytes[2], self.bytes[3]]) as usize; + if body_size > self.max_body_size { + return Err(NowClientError::FrameTooLarge { + declared: body_size, + maximum: self.max_body_size, + }); + } + + let frame_size = Self::HEADER_SIZE + .checked_add(body_size) + .ok_or(NowClientError::FrameTooLarge { + declared: body_size, + maximum: self.max_body_size, + })?; + if frame_size > maximum { + return Err(NowClientError::FrameBufferTooLarge { maximum }); + } + if self.bytes.len() < frame_size { + if input.is_empty() { + break; + } + let bytes_to_copy = input.len().min(frame_size - self.bytes.len()); + self.bytes.extend_from_slice(&input[..bytes_to_copy]); + input = &input[bytes_to_copy..]; + continue; + } + + let mut cursor = ReadCursor::new(&self.bytes[..frame_size]); + let message = + NowMessage::decode(&mut cursor).map_err(|error| NowClientError::PduDecode(error.to_string()))?; + self.ready.push(message.into_owned()); + self.bytes.drain(..frame_size); + } + + Ok(()) + } + + pub(crate) fn take_ready(&mut self) -> Vec { + core::mem::take(&mut self.ready) + } + + pub(crate) fn restore_ready(&mut self, messages: impl IntoIterator) { + self.ready.extend(messages); + } +} + +#[cfg(test)] +mod tests { + use now_proto_pdu::ironrdp_core::encode_vec; + use now_proto_pdu::{NowChannelCapsetMsg, NowChannelHeartbeatMsg, NowMessage}; + + use super::MessageBuffer; + use crate::NowClientError; + + fn encode(message: impl Into>) -> Vec { + match encode_vec(&message.into()) { + Ok(bytes) => bytes, + Err(error) => panic!("test PDU must encode: {error}"), + } + } + + #[test] + fn fragmented_and_coalesced_frames_are_retained_and_decoded() { + let frame = encode(NowChannelCapsetMsg::default()); + let heartbeat = encode(NowChannelHeartbeatMsg::default()); + let mut buffer = MessageBuffer::new(64); + + let first = buffer.push(&frame[..10]); + assert!(first.is_ok()); + assert!(buffer.take_ready().is_empty()); + + let mut coalesced = frame[10..].to_vec(); + coalesced.extend_from_slice(&heartbeat); + match buffer.push(&coalesced) { + Ok(()) => {} + Err(error) => panic!("frames must decode: {error}"), + }; + let messages = buffer.take_ready(); + + assert_eq!(messages.len(), 2); + } + + #[test] + fn oversized_body_is_rejected_before_buffering_it() { + let mut buffer = MessageBuffer::new(16); + let error = match buffer.push(&[17, 0, 0, 0, 0, 0, 0, 0]) { + Ok(()) => panic!("oversized frame must be rejected"), + Err(error) => error, + }; + + assert!(matches!( + error, + NowClientError::FrameTooLarge { + declared: 17, + maximum: 16 + } + )); + } +} diff --git a/protocols/rust/now-client/src/lib.rs b/protocols/rust/now-client/src/lib.rs new file mode 100644 index 0000000..432402f --- /dev/null +++ b/protocols/rust/now-client/src/lib.rs @@ -0,0 +1,18 @@ +#![doc = include_str!("../README.md")] + +//! Tokio-based high-level client support for the NOW execution protocol. + +mod capabilities; +mod client; +mod config; +mod error; +mod exec; +mod frame; + +pub use capabilities::NegotiatedCapabilities; +pub use client::{ + DetachedExecution, Execution, ExecutionEvent, ExecutionStatus, ExecutionSubmission, NowClient, NowClientHandle, +}; +pub use config::NowClientConfig; +pub use error::NowClientError; +pub use exec::{BatchRequest, PowerShellRequest, ProcessRequest, PwshRequest, RunRequest, WinPsRequest}; From 5a54ff1f36e40e2b7d058bee775a3eb9c5a269df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Moreau?= Date: Mon, 13 Jul 2026 10:20:47 -0400 Subject: [PATCH 2/4] refactor(protocols): refine NOW client API Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- protocols/rust/now-client/README.md | 12 +- protocols/rust/now-client/src/capabilities.rs | 4 +- protocols/rust/now-client/src/client.rs | 254 ++++++++++-------- protocols/rust/now-client/src/error.rs | 3 + protocols/rust/now-client/src/exec.rs | 81 ++---- protocols/rust/now-client/src/lib.rs | 6 +- 6 files changed, 177 insertions(+), 183 deletions(-) diff --git a/protocols/rust/now-client/README.md b/protocols/rust/now-client/README.md index e5ceba0..3daf7c2 100644 --- a/protocols/rust/now-client/README.md +++ b/protocols/rust/now-client/README.md @@ -4,11 +4,13 @@ channel. It accepts a caller-provided Tokio `AsyncRead + AsyncWrite` byte stream; consumers create DVCs, pipes, sockets, and replacement clients after reconnects. -Connect with `NowClient::connect`, then use the returned handle for Run, Process, -Batch, Windows PowerShell, or PowerShell 7. The client defensively negotiates -capabilities, applies bounded frame/command/event queues, and permits **one tracked -execution at a time per stream**. Tracked operations expose raw `Vec` stdout/stderr -chunks, stdin forwarding, normal cancellation, and terminal status. +Connect with `NowClient::connect`, then query negotiated `NowCapabilities` from the +returned handle. `process`, `batch`, `win_ps`, and `pwsh` submit tracked executions; +their `_detached` counterparts submit detached executions. `run` is submission-only. +The client defensively negotiates capabilities, applies bounded frame/command/event +queues, and permits **one tracked execution at a time per stream**. Tracked operations +expose raw `Vec` stdout/stderr chunks, stdin forwarding, normal cancellation, and +terminal status. Run and detached requests are submission-only. Gateway may emit an immediate Started/Data/Result sequence for Run; the client records and discards those matching diff --git a/protocols/rust/now-client/src/capabilities.rs b/protocols/rust/now-client/src/capabilities.rs index 3b25ade..b31f3f2 100644 --- a/protocols/rust/now-client/src/capabilities.rs +++ b/protocols/rust/now-client/src/capabilities.rs @@ -9,11 +9,11 @@ use crate::NowClientError; /// The value is always calculated as an intersection with the local advertised capset, /// even if the peer echoes unsupported flags. #[derive(Clone, Debug, PartialEq, Eq)] -pub struct NegotiatedCapabilities { +pub struct NowCapabilities { capset: NowChannelCapsetMsg, } -impl NegotiatedCapabilities { +impl NowCapabilities { pub(crate) fn negotiate( requested: &NowChannelCapsetMsg, peer: &NowChannelCapsetMsg, diff --git a/protocols/rust/now-client/src/client.rs b/protocols/rust/now-client/src/client.rs index d0b8b02..09b1380 100644 --- a/protocols/rust/now-client/src/client.rs +++ b/protocols/rust/now-client/src/client.rs @@ -9,24 +9,21 @@ use tokio::sync::{mpsc, oneshot}; use tokio::time::{self, Instant}; use crate::exec::{stdin_message, EncodedRequest, RequestSpec}; -use crate::{NegotiatedCapabilities, NowClientConfig, NowClientError}; +use crate::{NowCapabilities, NowClientConfig, NowClientError}; /// Entry point for connecting a Tokio byte stream to the NOW execution protocol. pub struct NowClient; impl NowClient { /// Negotiates NOW capabilities over `stream` and starts its single-owner protocol worker. - pub async fn connect( - stream: S, - config: NowClientConfig, - ) -> Result<(NowClientHandle, NegotiatedCapabilities), NowClientError> + pub async fn connect(stream: S, config: NowClientConfig) -> Result where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { config.validate()?; let (stream, messages, peer_capset) = handshake(stream, &config).await?; - let capabilities = NegotiatedCapabilities::negotiate(&config.client_capset, &peer_capset)?; - let shared_capabilities = Arc::new(RwLock::new(capabilities.clone())); + let capabilities = NowCapabilities::negotiate(&config.client_capset, &peer_capset)?; + let shared_capabilities = Arc::new(RwLock::new(capabilities)); let (command_sender, command_receiver) = mpsc::channel(config.command_queue_capacity); let worker_sender = command_sender.downgrade(); let worker = OwnedWorker { @@ -47,14 +44,11 @@ impl NowClient { worker.run().await; }); - Ok(( - NowClientHandle { - command_sender, - capabilities: shared_capabilities, - event_queue_capacity: config.event_queue_capacity, - }, - capabilities, - )) + Ok(NowClientHandle { + command_sender, + capabilities: shared_capabilities, + event_queue_capacity: config.event_queue_capacity, + }) } } @@ -62,13 +56,13 @@ impl NowClient { #[derive(Clone)] pub struct NowClientHandle { command_sender: mpsc::Sender, - capabilities: Arc>, + capabilities: Arc>, event_queue_capacity: usize, } impl NowClientHandle { /// Returns the latest negotiated capabilities, including capset refreshes from the peer. - pub fn capabilities(&self) -> NegotiatedCapabilities { + pub fn capabilities(&self) -> NowCapabilities { self.capabilities .read() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -77,98 +71,95 @@ impl NowClientHandle { /// Submits an untracked Run request. pub async fn run(&self, request: crate::RunRequest) -> Result { - match self.submit(RequestSpec::Run(request)).await? { - ExecutionSubmission::Detached(execution) => Ok(execution), - ExecutionSubmission::Tracked(_) => Err(NowClientError::Protocol( - "Run request unexpectedly started a tracked operation".to_owned(), - )), - } + let session_id = self.start(RequestSpec::Run(request), None, false).await?; + Ok(DetachedExecution { session_id }) } - /// Submits a Process request. - pub async fn process(&self, request: crate::ProcessRequest) -> Result { + /// Submits a tracked Process request. + pub async fn process(&self, request: crate::ProcessRequest) -> Result { self.submit(RequestSpec::Process(request)).await } - /// Submits a Batch request. - pub async fn batch(&self, request: crate::BatchRequest) -> Result { + /// Submits a detached Process request. + pub async fn process_detached(&self, request: crate::ProcessRequest) -> Result { + self.submit_detached(RequestSpec::Process(request)).await + } + + /// Submits a tracked Batch request. + pub async fn batch(&self, request: crate::BatchRequest) -> Result { self.submit(RequestSpec::Batch(request)).await } - /// Submits a Windows PowerShell request. - pub async fn win_ps(&self, request: crate::WinPsRequest) -> Result { + /// Submits a detached Batch request. + pub async fn batch_detached(&self, request: crate::BatchRequest) -> Result { + self.submit_detached(RequestSpec::Batch(request)).await + } + + /// Submits a tracked Windows PowerShell request. + pub async fn win_ps(&self, request: crate::WinPsRequest) -> Result { self.submit(RequestSpec::WinPs(request)).await } - /// Submits a PowerShell 7 request. - pub async fn pwsh(&self, request: crate::PwshRequest) -> Result { + /// Submits a detached Windows PowerShell request. + pub async fn win_ps_detached(&self, request: crate::WinPsRequest) -> Result { + self.submit_detached(RequestSpec::WinPs(request)).await + } + + /// Submits a tracked PowerShell 7 request. + pub async fn pwsh(&self, request: crate::PwshRequest) -> Result { self.submit(RequestSpec::Pwsh(request)).await } - async fn submit(&self, spec: RequestSpec) -> Result { - let tracked = spec.is_tracked(); - let (start_sender, start_receiver) = oneshot::channel(); - let (registration, events, terminal) = if tracked { - let (event_sender, event_receiver) = mpsc::channel(self.event_queue_capacity); - let (terminal_sender, terminal_receiver) = oneshot::channel(); - ( + /// Submits a detached PowerShell 7 request. + pub async fn pwsh_detached(&self, request: crate::PwshRequest) -> Result { + self.submit_detached(RequestSpec::Pwsh(request)).await + } + + async fn submit(&self, spec: RequestSpec) -> Result { + let (event_sender, events) = mpsc::channel(self.event_queue_capacity); + let (terminal_sender, terminal) = oneshot::channel(); + let session_id = self + .start( + spec, Some(OperationRegistration { event_sender, terminal_sender, }), - Some(event_receiver), - Some(terminal_receiver), + false, ) - } else { - (None, None, None) - }; + .await?; + Ok(Execution { + session_id, + events, + terminal, + command_sender: self.command_sender.clone(), + }) + } + + async fn submit_detached(&self, spec: RequestSpec) -> Result { + let session_id = self.start(spec, None, true).await?; + Ok(DetachedExecution { session_id }) + } + async fn start( + &self, + spec: RequestSpec, + registration: Option, + detached: bool, + ) -> Result { + let (start_sender, start_receiver) = oneshot::channel(); self.command_sender .send(WorkerCommand::Start { spec, registration, + detached, response: start_sender, }) .await .map_err(|_| NowClientError::WorkerClosed("command queue is closed".to_owned()))?; - let session_id = start_receiver + start_receiver .await - .map_err(|_| NowClientError::WorkerClosed("worker stopped while starting operation".to_owned()))??; - - if tracked { - let events = events.ok_or_else(|| { - NowClientError::Protocol("tracked operation did not receive an event receiver".to_owned()) - })?; - let terminal = terminal.ok_or_else(|| { - NowClientError::Protocol("tracked operation did not receive a terminal receiver".to_owned()) - })?; - Ok(ExecutionSubmission::Tracked(Execution { - session_id, - events, - terminal, - command_sender: self.command_sender.clone(), - })) - } else { - Ok(ExecutionSubmission::Detached(DetachedExecution { session_id })) - } - } -} - -/// Either a tracked execution handle or a detached submission identity. -pub enum ExecutionSubmission { - /// A tracked operation that emits output and has a terminal result. - Tracked(Execution), - /// A detached operation that only reports local submission. - Detached(DetachedExecution), -} - -impl ExecutionSubmission { - /// Returns the NOW session identity allocated for the submission. - pub fn id(&self) -> u32 { - match self { - Self::Tracked(execution) => execution.id(), - Self::Detached(execution) => execution.id(), - } + .map_err(|_| NowClientError::WorkerClosed("worker stopped while starting operation".to_owned()))? } } @@ -314,6 +305,7 @@ enum WorkerCommand { Start { spec: RequestSpec, registration: Option, + detached: bool, response: oneshot::Sender>, }, SendStdin { @@ -332,7 +324,7 @@ struct OwnedWorker { stream: S, messages: crate::frame::MessageBuffer, requested_capset: NowChannelCapsetMsg, - capabilities: Arc>, + capabilities: Arc>, command_receiver: mpsc::Receiver, command_sender: mpsc::WeakSender, operations: HashMap, @@ -409,7 +401,7 @@ where .map(|interval| Instant::now() + interval.saturating_mul(2)) } - fn capabilities(&self) -> NegotiatedCapabilities { + fn capabilities(&self) -> NowCapabilities { self.capabilities .read() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -428,7 +420,7 @@ where fn dispatch_message(&mut self, message: OwnedNowMessage) -> Result { match message { NowMessage::Channel(NowChannelMessage::Capset(capset)) => { - let capabilities = NegotiatedCapabilities::negotiate(&self.requested_capset, &capset)?; + let capabilities = NowCapabilities::negotiate(&self.requested_capset, &capset)?; *self .capabilities .write() @@ -529,11 +521,12 @@ where WorkerCommand::Start { spec, registration, + detached, response, } => { - let tracked = spec.is_tracked(); + let tracked = registration.is_some(); let is_run = spec.is_run(); - if tracked != registration.is_some() { + if (is_run && (tracked || detached)) || (!is_run && tracked == detached) { let _ = response.send(Err(NowClientError::Protocol( "worker received inconsistent operation registration".to_owned(), ))); @@ -547,16 +540,23 @@ where let _ = response.send(Err(NowClientError::RunDiscardQueueFull)); return Ok(()); } - let session_id = self.allocate_session_id(); - let request = match spec.build(session_id, &self.capabilities()) { + let session_id = match self.allocate_session_id() { + Ok(session_id) => session_id, + Err(error) => { + let _ = response.send(Err(error)); + return Ok(()); + } + }; + let request = match spec.build(session_id, &self.capabilities(), detached) { Ok(request) => request, Err(error) => { let _ = response.send(Err(error)); return Ok(()); } }; - let initial_stdin = match spec - .initial_stdin() + let initial_stdin = match tracked + .then(|| spec.initial_stdin()) + .flatten() .map(|data| stdin_message(session_id, data.to_vec(), true)) .transpose() { @@ -725,20 +725,12 @@ where Ok(()) } - fn allocate_session_id(&mut self) -> u32 { - loop { - let session_id = self.next_session_id; - self.next_session_id = self.next_session_id.wrapping_add(1); - if self.next_session_id == 0 { - self.next_session_id = 1; - } - if session_id != 0 - && !self.operations.contains_key(&session_id) - && !self.discarded_run_sessions.contains(&session_id) - { - return session_id; - } - } + fn allocate_session_id(&mut self) -> Result { + let session_id = self.next_session_id; + self.next_session_id = self.next_session_id.checked_add(1).unwrap_or_default(); + (session_id != 0) + .then_some(session_id) + .ok_or(NowClientError::SessionIdExhausted) } fn emit_event(&mut self, session_id: u32, event: ExecutionEvent) { @@ -837,7 +829,7 @@ mod tests { }; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; - use super::{ExecutionEvent, ExecutionStatus, ExecutionSubmission, NowClient}; + use super::{ExecutionEvent, ExecutionStatus, NowClient}; use crate::{NowClientConfig, NowClientError, ProcessRequest, RunRequest}; fn encode(message: impl Into>) -> Vec { @@ -923,12 +915,11 @@ mod tests { client_capset: capset(NowExecCapsetFlags::STYLE_RUN), ..NowClientConfig::default() }; - let (handle, capabilities) = match NowClient::connect(client_stream, config).await { - Ok(connection) => connection, + let handle = match NowClient::connect(client_stream, config).await { + Ok(handle) => handle, Err(error) => panic!("handshake must succeed: {error}"), }; - assert!(capabilities.supports_run()); - assert!(!capabilities.supports_process()); + assert!(handle.capabilities().supports_run()); assert!(!handle.capabilities().supports_process()); drop(handle); match peer.await { @@ -995,8 +986,8 @@ mod tests { write_message(&mut peer_stream, NowExecResultMsg::new_success(process_session, 17)).await; }); - let (handle, _) = match NowClient::connect(client_stream, NowClientConfig::default()).await { - Ok(connection) => connection, + let handle = match NowClient::connect(client_stream, NowClientConfig::default()).await { + Ok(handle) => handle, Err(error) => panic!("handshake must succeed: {error}"), }; let run = match handle.run(RunRequest::new("run.exe")).await { @@ -1008,10 +999,10 @@ mod tests { .process(ProcessRequest::new("process.exe").with_stdin(vec![0x01, 0xff])) .await { - Ok(ExecutionSubmission::Tracked(execution)) => execution, - Ok(ExecutionSubmission::Detached(_)) => panic!("default process request must be tracked"), + Ok(execution) => execution, Err(error) => panic!("Process request must start: {error}"), }; + assert_eq!(process.id(), 2); match handle.process(ProcessRequest::new("second.exe")).await { Err(NowClientError::OperationInProgress) => {} Ok(_) => panic!("second tracked execution must be rejected"), @@ -1040,6 +1031,38 @@ mod tests { } } + #[tokio::test] + async fn detached_submission_is_explicit_and_untracked() { + let (client_stream, mut peer_stream) = tokio::io::duplex(1024); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message(&mut peer_stream, capset(NowExecCapsetFlags::STYLE_PROCESS)).await; + + match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Process(message)) => { + assert_eq!(message.session_id(), 1); + assert!(message.is_detached()); + } + message => panic!("expected detached Process request, got {message:?}"), + } + }); + + let handle = match NowClient::connect(client_stream, NowClientConfig::default()).await { + Ok(handle) => handle, + Err(error) => panic!("handshake must succeed: {error}"), + }; + let submission = match handle.process_detached(ProcessRequest::new("process.exe")).await { + Ok(submission) => submission, + Err(error) => panic!("detached Process request must start: {error}"), + }; + assert_eq!(submission.id(), 1); + drop(handle); + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + #[tokio::test] async fn cancellation_waits_for_the_matching_response_and_result() { let (client_stream, mut peer_stream) = tokio::io::duplex(4096); @@ -1064,13 +1087,12 @@ mod tests { write_message(&mut peer_stream, NowExecResultMsg::new_success(session_id, 0)).await; }); - let (handle, _) = match NowClient::connect(client_stream, NowClientConfig::default()).await { - Ok(connection) => connection, + let handle = match NowClient::connect(client_stream, NowClientConfig::default()).await { + Ok(handle) => handle, Err(error) => panic!("handshake must succeed: {error}"), }; let mut execution = match handle.process(ProcessRequest::new("process.exe")).await { - Ok(ExecutionSubmission::Tracked(execution)) => execution, - Ok(ExecutionSubmission::Detached(_)) => panic!("default process request must be tracked"), + Ok(execution) => execution, Err(error) => panic!("Process request must start: {error}"), }; assert_eq!(execution.next_event().await, Some(ExecutionEvent::Started)); diff --git a/protocols/rust/now-client/src/error.rs b/protocols/rust/now-client/src/error.rs index 838c31e..77c8552 100644 --- a/protocols/rust/now-client/src/error.rs +++ b/protocols/rust/now-client/src/error.rs @@ -56,6 +56,9 @@ pub enum NowClientError { /// Too many Run submissions are awaiting their terminal frame for disposal. #[error("NOW Run discard queue reached its configured capacity")] RunDiscardQueueFull, + /// All nonzero NOW session IDs have been allocated by this client. + #[error("NOW session ID space is exhausted")] + SessionIdExhausted, /// A remote NOW result or cancellation response reported an error status. #[error("remote NOW operation {session_id} failed: {error}")] RemoteStatus { diff --git a/protocols/rust/now-client/src/exec.rs b/protocols/rust/now-client/src/exec.rs index 87134b1..fba3989 100644 --- a/protocols/rust/now-client/src/exec.rs +++ b/protocols/rust/now-client/src/exec.rs @@ -6,13 +6,12 @@ use now_proto_pdu::{ NowExecWinPsMsg, NowMessage, }; -use crate::{NegotiatedCapabilities, NowClientError}; +use crate::{NowCapabilities, NowClientError}; #[derive(Clone, Debug, PartialEq, Eq)] struct CommonRequest { command: String, directory: Option, - detached: bool, stdin: Option>, timeout: Option, } @@ -22,7 +21,6 @@ impl CommonRequest { Self { command: command.into(), directory: None, - detached: false, stdin: None, timeout: None, } @@ -59,7 +57,6 @@ pub struct ProcessRequest { filename: String, parameters: Option, directory: Option, - detached: bool, stdin: Option>, timeout: Option, } @@ -71,7 +68,6 @@ impl ProcessRequest { filename: filename.into(), parameters: None, directory: None, - detached: false, stdin: None, timeout: None, } @@ -91,13 +87,6 @@ impl ProcessRequest { self } - /// Requests detached execution. - #[must_use] - pub fn detached(mut self) -> Self { - self.detached = true; - self - } - /// Supplies the first and final stdin chunk. #[must_use] pub fn with_stdin(mut self, data: impl Into>) -> Self { @@ -134,13 +123,6 @@ impl BatchRequest { self } - /// Requests detached execution. - #[must_use] - pub fn detached(mut self) -> Self { - self.common.detached = true; - self - } - /// Supplies the first and final stdin chunk. #[must_use] pub fn with_stdin(mut self, data: impl Into>) -> Self { @@ -181,13 +163,6 @@ impl PowerShellRequest { self } - /// Requests detached execution. - #[must_use] - pub fn detached(mut self) -> Self { - self.common.detached = true; - self - } - /// Supplies the first and final stdin chunk. #[must_use] pub fn with_stdin(mut self, data: impl Into>) -> Self { @@ -240,15 +215,6 @@ impl RequestSpec { matches!(self, Self::Run(_)) } - pub(crate) fn is_tracked(&self) -> bool { - match self { - Self::Run(_) => false, - Self::Process(request) => !request.detached, - Self::Batch(request) => !request.common.detached, - Self::WinPs(request) | Self::Pwsh(request) => !request.common.detached, - } - } - pub(crate) fn initial_stdin(&self) -> Option<&[u8]> { match self { Self::Run(_) => None, @@ -270,9 +236,10 @@ impl RequestSpec { pub(crate) fn build( &self, session_id: u32, - capabilities: &NegotiatedCapabilities, + capabilities: &NowCapabilities, + detached: bool, ) -> Result { - self.validate(capabilities)?; + self.validate(capabilities, detached)?; match self { Self::Run(request) => { @@ -296,7 +263,7 @@ impl RequestSpec { if capabilities.supports_unicode_console() { message = message.with_encoding_utf8(); } - if request.detached { + if detached { message = message.with_detached(); } else { message = message.with_io_redirection(); @@ -315,7 +282,7 @@ impl RequestSpec { if capabilities.supports_unicode_console() { message = message.with_raw_encoding().with_unicode_console(); } - if common.detached { + if detached { message = message.with_detached(); } else { message = message.with_io_redirection(); @@ -325,12 +292,12 @@ impl RequestSpec { non_interactive: false, }) } - Self::WinPs(request) => build_power_shell(session_id, request, false, capabilities), - Self::Pwsh(request) => build_power_shell(session_id, request, true, capabilities), + Self::WinPs(request) => build_power_shell(session_id, request, false, capabilities, detached), + Self::Pwsh(request) => build_power_shell(session_id, request, true, capabilities, detached), } } - fn validate(&self, capabilities: &NegotiatedCapabilities) -> Result<(), NowClientError> { + fn validate(&self, capabilities: &NowCapabilities, detached: bool) -> Result<(), NowClientError> { match self { Self::Run(request) => { require(capabilities.supports_run(), "Run")?; @@ -341,20 +308,20 @@ impl RequestSpec { require(capabilities.supports_process(), "Process")?; validate_filename(&request.filename)?; validate_optional(&request.parameters, "parameters")?; - validate_common(request.detached, request.stdin.as_deref(), request.timeout)?; - validate_tracking(capabilities, request.detached)?; + validate_common(detached, request.stdin.as_deref(), request.timeout)?; + validate_tracking(capabilities, detached)?; } Self::Batch(request) => { require(capabilities.supports_batch(), "Batch")?; - validate_common_request(&request.common, capabilities)?; + validate_common_request(&request.common, capabilities, detached)?; } Self::WinPs(request) => { require(capabilities.supports_win_ps(), "WinPs")?; - validate_common_request(&request.common, capabilities)?; + validate_common_request(&request.common, capabilities, detached)?; } Self::Pwsh(request) => { require(capabilities.supports_pwsh(), "Pwsh")?; - validate_common_request(&request.common, capabilities)?; + validate_common_request(&request.common, capabilities, detached)?; } } Ok(()) @@ -362,7 +329,7 @@ impl RequestSpec { } macro_rules! apply_power_shell_options { - ($message:ident, $common:ident, $request:ident, $capabilities:ident) => { + ($message:ident, $common:ident, $request:ident, $capabilities:ident, $detached:ident) => { if let Some(directory) = &$common.directory { $message = pdu($message.with_directory(directory.clone()))?; } @@ -372,7 +339,7 @@ macro_rules! apply_power_shell_options { if $capabilities.supports_unicode_console() { $message = $message.with_raw_encoding().with_unicode_console(); } - if $common.detached { + if $detached { $message = $message.with_detached(); } else { $message = $message.with_io_redirection(); @@ -384,19 +351,20 @@ fn build_power_shell( session_id: u32, request: &PowerShellRequest, pwsh: bool, - capabilities: &NegotiatedCapabilities, + capabilities: &NowCapabilities, + detached: bool, ) -> Result { let common = &request.common; if pwsh { let mut message = pdu(NowExecPwshMsg::new(session_id, common.command.clone()))?; - apply_power_shell_options!(message, common, request, capabilities); + apply_power_shell_options!(message, common, request, capabilities, detached); Ok(EncodedRequest { message: message.into(), non_interactive: request.non_interactive, }) } else { let mut message = pdu(NowExecWinPsMsg::new(session_id, common.command.clone()))?; - apply_power_shell_options!(message, common, request, capabilities); + apply_power_shell_options!(message, common, request, capabilities, detached); Ok(EncodedRequest { message: message.into(), non_interactive: request.non_interactive, @@ -406,12 +374,13 @@ fn build_power_shell( fn validate_common_request( request: &CommonRequest, - capabilities: &NegotiatedCapabilities, + capabilities: &NowCapabilities, + detached: bool, ) -> Result<(), NowClientError> { validate_command(&request.command)?; validate_directory(request.directory.as_deref())?; - validate_common(request.detached, request.stdin.as_deref(), request.timeout)?; - validate_tracking(capabilities, request.detached) + validate_common(detached, request.stdin.as_deref(), request.timeout)?; + validate_tracking(capabilities, detached) } fn validate_common(detached: bool, stdin: Option<&[u8]>, timeout: Option) -> Result<(), NowClientError> { @@ -433,7 +402,7 @@ fn validate_common(detached: bool, stdin: Option<&[u8]>, timeout: Option Result<(), NowClientError> { +fn validate_tracking(capabilities: &NowCapabilities, detached: bool) -> Result<(), NowClientError> { if detached { require(capabilities.supports_detached(), "detached execution") } else { diff --git a/protocols/rust/now-client/src/lib.rs b/protocols/rust/now-client/src/lib.rs index 432402f..131db21 100644 --- a/protocols/rust/now-client/src/lib.rs +++ b/protocols/rust/now-client/src/lib.rs @@ -9,10 +9,8 @@ mod error; mod exec; mod frame; -pub use capabilities::NegotiatedCapabilities; -pub use client::{ - DetachedExecution, Execution, ExecutionEvent, ExecutionStatus, ExecutionSubmission, NowClient, NowClientHandle, -}; +pub use capabilities::NowCapabilities; +pub use client::{DetachedExecution, Execution, ExecutionEvent, ExecutionStatus, NowClient, NowClientHandle}; pub use config::NowClientConfig; pub use error::NowClientError; pub use exec::{BatchRequest, PowerShellRequest, ProcessRequest, PwshRequest, RunRequest, WinPsRequest}; From de81fe682d6f5dd968372313cf801768895b280b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Moreau?= Date: Mon, 13 Jul 2026 10:48:30 -0400 Subject: [PATCH 3/4] fix(protocols): harden NOW client worker Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- protocols/rust/now-client/README.md | 7 +- protocols/rust/now-client/src/capabilities.rs | 2 +- protocols/rust/now-client/src/client.rs | 542 ++++++++++++++---- protocols/rust/now-client/src/config.rs | 2 +- protocols/rust/now-client/src/error.rs | 3 - protocols/rust/now-client/src/exec.rs | 1 + 6 files changed, 450 insertions(+), 107 deletions(-) diff --git a/protocols/rust/now-client/README.md b/protocols/rust/now-client/README.md index 3daf7c2..9b53f81 100644 --- a/protocols/rust/now-client/README.md +++ b/protocols/rust/now-client/README.md @@ -14,6 +14,7 @@ terminal status. Run and detached requests are submission-only. Gateway may emit an immediate Started/Data/Result sequence for Run; the client records and discards those matching -frames so they cannot affect the next tracked operation. This crate deliberately does -not provide Abort, Shell submission, DVC/pipe setup, reconnect policy, or retained -operation output. +frames so they cannot affect the next tracked operation. This recent-session quarantine +is bounded and evicts its oldest entry; evicted Run traffic remains harmless because +session IDs are never reused. This crate deliberately does not provide Abort, Shell +submission, DVC/pipe setup, reconnect policy, or retained operation output. diff --git a/protocols/rust/now-client/src/capabilities.rs b/protocols/rust/now-client/src/capabilities.rs index b31f3f2..e92bb50 100644 --- a/protocols/rust/now-client/src/capabilities.rs +++ b/protocols/rust/now-client/src/capabilities.rs @@ -40,7 +40,7 @@ impl NowCapabilities { self.capset.version() } - /// Returns the negotiated heartbeat interval, if both peers requested one. + /// Returns the negotiated heartbeat interval, if either peer requested one. pub fn heartbeat_interval(&self) -> Option { self.capset.heartbeat_interval() } diff --git a/protocols/rust/now-client/src/client.rs b/protocols/rust/now-client/src/client.rs index 09b1380..56a8366 100644 --- a/protocols/rust/now-client/src/client.rs +++ b/protocols/rust/now-client/src/client.rs @@ -1,10 +1,12 @@ +use core::time::Duration; + use now_proto_pdu::ironrdp_core::encode_vec; use now_proto_pdu::{ NowChannelCapsetMsg, NowChannelMessage, NowExecDataStreamKind, NowExecMessage, NowMessage, OwnedNowMessage, }; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, RwLock}; -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf}; use tokio::sync::{mpsc, oneshot}; use tokio::time::{self, Instant}; @@ -26,17 +28,21 @@ impl NowClient { let shared_capabilities = Arc::new(RwLock::new(capabilities)); let (command_sender, command_receiver) = mpsc::channel(config.command_queue_capacity); let worker_sender = command_sender.downgrade(); + let (reader, writer) = tokio::io::split(stream); let worker = OwnedWorker { - stream, + reader, + writer, messages, requested_capset: config.client_capset.clone(), capabilities: Arc::clone(&shared_capabilities), command_receiver, command_sender: worker_sender, operations: HashMap::new(), - discarded_run_sessions: HashSet::new(), + discarded_run_sessions: VecDeque::new(), run_discard_capacity: config.run_discard_capacity, - next_session_id: 1, + next_session_id: Some(1), + pending_writes: VecDeque::new(), + write_queue_capacity: config.command_queue_capacity, read_buffer: vec![0; config.read_buffer_size], }; @@ -288,6 +294,31 @@ struct Operation { stdin_closed: bool, } +enum PendingWriteCompletion { + Start { + session_id: u32, + response: oneshot::Sender>, + timeout: Option, + }, + Stdin { + session_id: u32, + last: bool, + response: oneshot::Sender>, + }, + Cancel, +} + +struct PendingWrite { + bytes: Vec, + written: usize, + completion: PendingWriteCompletion, +} + +enum WriteProgress { + Written(usize), + Flushed, +} + #[derive(Default)] struct BufferUpdate { capset_updated: bool, @@ -321,16 +352,19 @@ enum WorkerCommand { } struct OwnedWorker { - stream: S, + reader: ReadHalf, + writer: WriteHalf, messages: crate::frame::MessageBuffer, requested_capset: NowChannelCapsetMsg, capabilities: Arc>, command_receiver: mpsc::Receiver, command_sender: mpsc::WeakSender, operations: HashMap, - discarded_run_sessions: HashSet, + discarded_run_sessions: VecDeque, run_discard_capacity: usize, - next_session_id: u32, + next_session_id: Option, + pending_writes: VecDeque, + write_queue_capacity: usize, read_buffer: Vec, } @@ -348,6 +382,7 @@ where } } Err(error) => { + self.fail_pending_writes(&error); self.fail_all(&error); return; } @@ -356,9 +391,12 @@ where enum Event { Command(Option), Read(std::io::Result), + Write(std::io::Result), HeartbeatTimeout, } + let has_pending_write = !self.pending_writes.is_empty(); + let can_receive_command = self.pending_writes.len() < self.write_queue_capacity; let event = { let heartbeat = async { if let Some(deadline) = heartbeat_deadline { @@ -368,26 +406,52 @@ where } }; tokio::pin!(heartbeat); + let reader = &mut self.reader; + let writer = &mut self.writer; + let read_buffer = &mut self.read_buffer; + let pending_writes = &mut self.pending_writes; tokio::select! { - command = self.command_receiver.recv() => Event::Command(command), - read = self.stream.read(&mut self.read_buffer) => Event::Read(read), + command = self.command_receiver.recv(), if can_receive_command => Event::Command(command), + read = reader.read(read_buffer) => Event::Read(read), + write = async { + let pending = pending_writes + .front() + .expect("a pending write must be available when the write branch is enabled"); + if pending.written == pending.bytes.len() { + writer.flush().await.map(|()| WriteProgress::Flushed) + } else { + writer + .write(&pending.bytes[pending.written..]) + .await + .map(WriteProgress::Written) + } + }, if has_pending_write => Event::Write(write), _ = &mut heartbeat => Event::HeartbeatTimeout, } }; let result = match event { - Event::Command(Some(command)) => self.handle_command(command).await, + Event::Command(Some(command)) => self.handle_command(command), Event::Command(None) => { let error = NowClientError::WorkerClosed("all client handles were dropped".to_owned()); + self.fail_pending_writes(&error); self.fail_all(&error); return; } Event::Read(Ok(0)) => Err(NowClientError::WorkerClosed("transport reached EOF".to_owned())), Event::Read(Ok(read)) => self.messages.push(&self.read_buffer[..read]), Event::Read(Err(error)) => Err(error.into()), + Event::Write(Ok(WriteProgress::Written(0))) => Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "transport accepted no bytes for a pending NOW PDU", + ) + .into()), + Event::Write(Ok(progress)) => self.complete_pending_write(progress), + Event::Write(Err(error)) => Err(error.into()), Event::HeartbeatTimeout => Err(NowClientError::WorkerClosed("peer heartbeat timed out".to_owned())), }; if let Err(error) = result { + self.fail_pending_writes(&error); self.fail_all(&error); return; } @@ -436,7 +500,7 @@ where NowMessage::Exec(NowExecMessage::Started(message)) => { let session_id = message.session_id(); if !self.discarded_run_sessions.contains(&session_id) { - self.emit_event(session_id, ExecutionEvent::Started); + self.emit_event(session_id, ExecutionEvent::Started)?; } Ok(BufferUpdate::default()) } @@ -459,7 +523,7 @@ where NowExecDataStreamKind::Stdin => None, }; if let Some(event) = event { - self.emit_event(message.session_id(), event); + self.emit_event(message.session_id(), event)?; } Ok(BufferUpdate::default()) } @@ -467,7 +531,7 @@ where let session_id = message.session_id(); match message.to_result() { Ok(()) => { - self.emit_event(session_id, ExecutionEvent::CancelAccepted); + self.emit_event(session_id, ExecutionEvent::CancelAccepted)?; if let Some(operation) = self.operations.get_mut(&session_id) { operation.cancel_accepted = true; if let Some(response) = operation.cancel_response.take() { @@ -488,7 +552,7 @@ where } NowMessage::Exec(NowExecMessage::Result(message)) => { let session_id = message.session_id(); - if self.discarded_run_sessions.remove(&session_id) { + if self.remove_discarded_run_session(session_id) { return Ok(BufferUpdate::default()); } match message.to_result() { @@ -516,7 +580,7 @@ where } } - async fn handle_command(&mut self, command: WorkerCommand) -> Result<(), NowClientError> { + fn handle_command(&mut self, command: WorkerCommand) -> Result<(), NowClientError> { match command { WorkerCommand::Start { spec, @@ -536,11 +600,7 @@ where let _ = response.send(Err(NowClientError::OperationInProgress)); return Ok(()); } - if is_run && self.discarded_run_sessions.len() == self.run_discard_capacity { - let _ = response.send(Err(NowClientError::RunDiscardQueueFull)); - return Ok(()); - } - let session_id = match self.allocate_session_id() { + let session_id = match allocate_session_id(&mut self.next_session_id) { Ok(session_id) => session_id, Err(error) => { let _ = response.send(Err(error)); @@ -566,25 +626,13 @@ where return Ok(()); } }; + let mut bytes = Self::encode_request(request)?; let stdin_closed = initial_stdin.is_some(); - - if let Err(error) = self.write_message(request).await { - let response_error = NowClientError::WorkerClosed(error.to_string()); - let _ = response.send(Err(response_error)); - return Err(error); - } if let Some(stdin) = initial_stdin { - if let Err(error) = self - .write_message(EncodedRequest { - message: stdin, - non_interactive: false, - }) - .await - { - let response_error = NowClientError::WorkerClosed(error.to_string()); - let _ = response.send(Err(response_error)); - return Err(error); - } + bytes.extend(Self::encode_request(EncodedRequest { + message: stdin, + non_interactive: false, + })?); } if let Some(registration) = registration { @@ -599,24 +647,19 @@ where stdin_closed, }, ); - if let Some(timeout) = spec.timeout() { - if let Some(command_sender) = self.command_sender.upgrade() { - tokio::spawn(async move { - time::sleep(timeout).await; - let _ = command_sender - .send(WorkerCommand::Cancel { - session_id, - response: None, - }) - .await; - }); - } - } } if is_run { - self.discarded_run_sessions.insert(session_id); + self.discard_run_session(session_id); } - let _ = response.send(Ok(session_id)); + self.pending_writes.push_back(PendingWrite { + bytes, + written: 0, + completion: PendingWriteCompletion::Start { + session_id, + response, + timeout: tracked.then(|| spec.timeout()).flatten(), + }, + }); Ok(()) } WorkerCommand::SendStdin { @@ -638,29 +681,24 @@ where ))); return Ok(()); } - match stdin_message(session_id, data, last) { - Ok(message) => match self - .write_message(EncodedRequest { - message, - non_interactive: false, - }) - .await - { - Ok(()) => { - if last { - if let Some(operation) = self.operations.get_mut(&session_id) { - operation.stdin_closed = true; - } - } - let _ = response.send(Ok(())); - Ok(()) - } - Err(error) => { - let response_error = NowClientError::WorkerClosed(error.to_string()); - let _ = response.send(Err(response_error)); - Err(error) - } - }, + match stdin_message(session_id, data, last).and_then(|message| { + Self::encode_request(EncodedRequest { + message, + non_interactive: false, + }) + }) { + Ok(bytes) => { + self.pending_writes.push_back(PendingWrite { + bytes, + written: 0, + completion: PendingWriteCompletion::Stdin { + session_id, + last, + response, + }, + }); + Ok(()) + } Err(error) => { let _ = response.send(Err(error)); Ok(()) @@ -687,32 +725,33 @@ where } let message = now_proto_pdu::NowExecCancelReqMsg::new(session_id).into(); - match self - .write_message(EncodedRequest { - message, - non_interactive: false, - }) - .await - { - Ok(()) => { + match Self::encode_request(EncodedRequest { + message, + non_interactive: false, + }) { + Ok(bytes) => { if let Some(operation) = self.operations.get_mut(&session_id) { operation.cancel_pending = true; operation.cancel_response = response; } - Ok(()) + self.pending_writes.push_back(PendingWrite { + bytes, + written: 0, + completion: PendingWriteCompletion::Cancel, + }); } Err(error) => { if let Some(response) = response { - let _ = response.send(Err(NowClientError::WorkerClosed(error.to_string()))); + let _ = response.send(Err(error)); } - Err(error) } } + Ok(()) } } } - async fn write_message(&mut self, request: EncodedRequest) -> Result<(), NowClientError> { + fn encode_request(request: EncodedRequest) -> Result, NowClientError> { let mut bytes = encode_vec(&request.message).map_err(|error| NowClientError::PduEncode(error.to_string()))?; if request.non_interactive { // now-proto-pdu 0.4.3 exposes the PowerShell non-interactive flag for decoding but @@ -720,20 +759,102 @@ where let flags = u16::from_le_bytes([bytes[6], bytes[7]]) | 0x0020; bytes[6..8].copy_from_slice(&flags.to_le_bytes()); } - self.stream.write_all(&bytes).await?; - self.stream.flush().await?; + Ok(bytes) + } + + fn complete_pending_write(&mut self, progress: WriteProgress) -> Result<(), NowClientError> { + match progress { + WriteProgress::Written(written) => { + let pending = self + .pending_writes + .front_mut() + .expect("a write completion must have a pending write"); + pending.written += written; + } + WriteProgress::Flushed => { + let pending = self + .pending_writes + .pop_front() + .expect("a flush completion must have a pending write"); + self.finish_pending_write(pending.completion); + } + } Ok(()) } - fn allocate_session_id(&mut self) -> Result { - let session_id = self.next_session_id; - self.next_session_id = self.next_session_id.checked_add(1).unwrap_or_default(); - (session_id != 0) - .then_some(session_id) - .ok_or(NowClientError::SessionIdExhausted) + fn finish_pending_write(&mut self, completion: PendingWriteCompletion) { + match completion { + PendingWriteCompletion::Start { + session_id, + response, + timeout, + } => { + if let Some(timeout) = timeout { + if let Some(command_sender) = self.command_sender.upgrade() { + tokio::spawn(async move { + time::sleep(timeout).await; + let _ = command_sender + .send(WorkerCommand::Cancel { + session_id, + response: None, + }) + .await; + }); + } + } + let _ = response.send(Ok(session_id)); + } + PendingWriteCompletion::Stdin { + session_id, + last, + response, + } => { + if last { + if let Some(operation) = self.operations.get_mut(&session_id) { + operation.stdin_closed = true; + } + } + let _ = response.send(Ok(())); + } + PendingWriteCompletion::Cancel => {} + } + } + + fn fail_pending_writes(&mut self, error: &NowClientError) { + let reason = error.to_string(); + for pending in self.pending_writes.drain(..) { + match pending.completion { + PendingWriteCompletion::Start { response, .. } => { + let _ = response.send(Err(NowClientError::WorkerClosed(reason.clone()))); + } + PendingWriteCompletion::Stdin { response, .. } => { + let _ = response.send(Err(NowClientError::WorkerClosed(reason.clone()))); + } + PendingWriteCompletion::Cancel => {} + } + } } - fn emit_event(&mut self, session_id: u32, event: ExecutionEvent) { + fn discard_run_session(&mut self, session_id: u32) { + if self.discarded_run_sessions.len() == self.run_discard_capacity { + self.discarded_run_sessions.pop_front(); + } + self.discarded_run_sessions.push_back(session_id); + } + + fn remove_discarded_run_session(&mut self, session_id: u32) -> bool { + let Some(position) = self + .discarded_run_sessions + .iter() + .position(|discarded_session_id| *discarded_session_id == session_id) + else { + return false; + }; + self.discarded_run_sessions.remove(position); + true + } + + fn emit_event(&mut self, session_id: u32, event: ExecutionEvent) -> Result<(), NowClientError> { let overflowed = match self.operations.get_mut(&session_id) { Some(operation) => match operation.event_sender.try_send(event) { Ok(()) => false, @@ -743,8 +864,9 @@ where None => false, }; if overflowed { - self.finish(session_id, Err(NowClientError::EventQueueFull { session_id })); + return Err(NowClientError::EventQueueFull { session_id }); } + Ok(()) } fn finish(&mut self, session_id: u32, result: Result) { @@ -769,6 +891,12 @@ where } } +fn allocate_session_id(next_session_id: &mut Option) -> Result { + let session_id = (*next_session_id).ok_or(NowClientError::SessionIdExhausted)?; + *next_session_id = session_id.checked_add(1); + Ok(session_id) +} + async fn handshake( mut stream: S, config: &NowClientConfig, @@ -822,6 +950,8 @@ where #[cfg(test)] mod tests { + use core::time::Duration; + use now_proto_pdu::ironrdp_core::{encode_vec, Decode, IntoOwned, ReadCursor}; use now_proto_pdu::{ NowChannelCapsetMsg, NowChannelHeartbeatMsg, NowExecCancelRspMsg, NowExecCapsetFlags, NowExecDataMsg, @@ -829,7 +959,7 @@ mod tests { }; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; - use super::{ExecutionEvent, ExecutionStatus, NowClient}; + use super::{allocate_session_id, ExecutionEvent, ExecutionStatus, NowClient}; use crate::{NowClientConfig, NowClientError, ProcessRequest, RunRequest}; fn encode(message: impl Into>) -> Vec { @@ -874,6 +1004,20 @@ mod tests { NowChannelCapsetMsg::default().with_exec_capset(flags) } + #[test] + fn session_id_exhaustion_is_sticky() { + let mut next_session_id = Some(u32::MAX); + assert!(matches!(allocate_session_id(&mut next_session_id), Ok(u32::MAX))); + assert!(matches!( + allocate_session_id(&mut next_session_id), + Err(NowClientError::SessionIdExhausted) + )); + assert!(matches!( + allocate_session_id(&mut next_session_id), + Err(NowClientError::SessionIdExhausted) + )); + } + #[tokio::test] async fn handshake_rejects_an_incompatible_major_version() { let (client_stream, mut peer_stream) = tokio::io::duplex(1024); @@ -1031,6 +1175,206 @@ mod tests { } } + #[tokio::test] + async fn run_quarantine_evicts_old_sessions_without_blocking_submission() { + let (client_stream, mut peer_stream) = tokio::io::duplex(1024); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message(&mut peer_stream, capset(NowExecCapsetFlags::STYLE_RUN)).await; + + let first_session = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Run(message)) => message.session_id(), + message => panic!("expected first Run request, got {message:?}"), + }; + let second_session = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Run(message)) => message.session_id(), + message => panic!("expected second Run request, got {message:?}"), + }; + assert_eq!((first_session, second_session), (1, 2)); + write_message(&mut peer_stream, NowExecResultMsg::new_success(first_session, 0)).await; + }); + + let config = NowClientConfig { + run_discard_capacity: 1, + ..NowClientConfig::default() + }; + let handle = match NowClient::connect(client_stream, config).await { + Ok(handle) => handle, + Err(error) => panic!("handshake must succeed: {error}"), + }; + assert_eq!( + handle + .run(RunRequest::new("first.exe")) + .await + .expect("first Run request must start") + .id(), + 1 + ); + assert_eq!( + handle + .run(RunRequest::new("second.exe")) + .await + .expect("second Run request must start after eviction") + .id(), + 2 + ); + drop(handle); + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + + #[tokio::test] + async fn process_rejects_an_empty_directory_before_writing() { + let (client_stream, mut peer_stream) = tokio::io::duplex(1024); + let (release_sender, release_receiver) = tokio::sync::oneshot::channel(); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message( + &mut peer_stream, + capset(NowExecCapsetFlags::STYLE_PROCESS | NowExecCapsetFlags::IO_REDIRECTION), + ) + .await; + let _ = release_receiver.await; + }); + + let handle = match NowClient::connect(client_stream, NowClientConfig::default()).await { + Ok(handle) => handle, + Err(error) => panic!("handshake must succeed: {error}"), + }; + assert!(matches!( + handle + .process(ProcessRequest::new("process.exe").with_directory("")) + .await, + Err(NowClientError::InvalidRequest(_)) + )); + let _ = release_sender.send(()); + drop(handle); + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + + #[tokio::test] + async fn stdin_write_does_not_block_reading_peer_output() { + let (client_stream, mut peer_stream) = tokio::io::duplex(64); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message( + &mut peer_stream, + capset(NowExecCapsetFlags::STYLE_PROCESS | NowExecCapsetFlags::IO_REDIRECTION), + ) + .await; + + let session_id = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Process(message)) => message.session_id(), + message => panic!("expected Process request, got {message:?}"), + }; + write_message(&mut peer_stream, NowExecStartedMsg::new(session_id)).await; + let stdout = match NowExecDataMsg::new(session_id, NowExecDataStreamKind::Stdout, true, vec![0xa5; 512]) { + Ok(message) => message, + Err(error) => panic!("test stdout PDU must encode: {error}"), + }; + write_message(&mut peer_stream, stdout).await; + match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Data(message)) => { + assert_eq!(message.session_id(), session_id); + assert_eq!(message.data(), vec![0x5a; 512]); + assert!(message.is_last()); + } + message => panic!("expected stdin data, got {message:?}"), + } + write_message(&mut peer_stream, NowExecResultMsg::new_success(session_id, 0)).await; + }); + + let config = NowClientConfig { + read_buffer_size: 32, + ..NowClientConfig::default() + }; + let handle = match NowClient::connect(client_stream, config).await { + Ok(handle) => handle, + Err(error) => panic!("handshake must succeed: {error}"), + }; + let mut execution = match handle.process(ProcessRequest::new("process.exe")).await { + Ok(execution) => execution, + Err(error) => panic!("Process request must start: {error}"), + }; + assert_eq!(execution.next_event().await, Some(ExecutionEvent::Started)); + match tokio::time::timeout(Duration::from_secs(2), execution.send_stdin(vec![0x5a; 512], true)).await { + Ok(Ok(())) => {} + Ok(Err(error)) => panic!("stdin must be written: {error}"), + Err(_) => panic!("stdin write timed out while peer output was backpressured"), + } + assert_eq!( + execution.next_event().await, + Some(ExecutionEvent::Stdout { + data: vec![0xa5; 512], + last: true, + }) + ); + match execution.wait().await { + Ok(ExecutionStatus::Completed { exit_code: 0 }) => {} + Ok(status) => panic!("unexpected terminal status: {status:?}"), + Err(error) => panic!("Process must complete: {error}"), + } + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + + #[tokio::test] + async fn event_queue_overflow_closes_the_worker_before_another_execution_starts() { + let (client_stream, mut peer_stream) = tokio::io::duplex(1024); + let (release_sender, release_receiver) = tokio::sync::oneshot::channel(); + let peer = tokio::spawn(async move { + let _ = read_message(&mut peer_stream).await; + write_message( + &mut peer_stream, + capset(NowExecCapsetFlags::STYLE_PROCESS | NowExecCapsetFlags::IO_REDIRECTION), + ) + .await; + + let session_id = match read_message(&mut peer_stream).await { + NowMessage::Exec(NowExecMessage::Process(message)) => message.session_id(), + message => panic!("expected Process request, got {message:?}"), + }; + write_message(&mut peer_stream, NowExecStartedMsg::new(session_id)).await; + let stdout = match NowExecDataMsg::new(session_id, NowExecDataStreamKind::Stdout, true, vec![0x01]) { + Ok(message) => message, + Err(error) => panic!("test stdout PDU must encode: {error}"), + }; + write_message(&mut peer_stream, stdout).await; + let _ = release_receiver.await; + }); + + let config = NowClientConfig { + event_queue_capacity: 1, + ..NowClientConfig::default() + }; + let handle = match NowClient::connect(client_stream, config).await { + Ok(handle) => handle, + Err(error) => panic!("handshake must succeed: {error}"), + }; + let execution = match handle.process(ProcessRequest::new("process.exe")).await { + Ok(execution) => execution, + Err(error) => panic!("Process request must start: {error}"), + }; + assert!(matches!(execution.wait().await, Err(NowClientError::WorkerClosed(_)))); + assert!(matches!( + handle.process(ProcessRequest::new("second.exe")).await, + Err(NowClientError::WorkerClosed(_)) + )); + let _ = release_sender.send(()); + drop(handle); + match peer.await { + Ok(()) => {} + Err(error) => panic!("test peer task failed: {error}"), + } + } + #[tokio::test] async fn detached_submission_is_explicit_and_untracked() { let (client_stream, mut peer_stream) = tokio::io::duplex(1024); diff --git a/protocols/rust/now-client/src/config.rs b/protocols/rust/now-client/src/config.rs index d579b68..53d07f0 100644 --- a/protocols/rust/now-client/src/config.rs +++ b/protocols/rust/now-client/src/config.rs @@ -17,7 +17,7 @@ pub struct NowClientConfig { pub command_queue_capacity: usize, /// Bound for unread events per tracked execution. pub event_queue_capacity: usize, - /// Maximum number of Run submissions awaiting their terminal frame for disposal. + /// Maximum number of recent Run submissions whose post-submission frames are discarded. pub run_discard_capacity: usize, /// Deadline for receiving the peer capability set. pub connect_timeout: Duration, diff --git a/protocols/rust/now-client/src/error.rs b/protocols/rust/now-client/src/error.rs index 77c8552..ab046af 100644 --- a/protocols/rust/now-client/src/error.rs +++ b/protocols/rust/now-client/src/error.rs @@ -53,9 +53,6 @@ pub enum NowClientError { /// A tracked execution is already active on this client stream. #[error("NOW client already has an active tracked execution")] OperationInProgress, - /// Too many Run submissions are awaiting their terminal frame for disposal. - #[error("NOW Run discard queue reached its configured capacity")] - RunDiscardQueueFull, /// All nonzero NOW session IDs have been allocated by this client. #[error("NOW session ID space is exhausted")] SessionIdExhausted, diff --git a/protocols/rust/now-client/src/exec.rs b/protocols/rust/now-client/src/exec.rs index fba3989..ac97b40 100644 --- a/protocols/rust/now-client/src/exec.rs +++ b/protocols/rust/now-client/src/exec.rs @@ -308,6 +308,7 @@ impl RequestSpec { require(capabilities.supports_process(), "Process")?; validate_filename(&request.filename)?; validate_optional(&request.parameters, "parameters")?; + validate_directory(request.directory.as_deref())?; validate_common(detached, request.stdin.as_deref(), request.timeout)?; validate_tracking(capabilities, detached)?; } From d33270243c3578262df8a82c31d7bc8ccc7c7867 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Moreau?= Date: Mon, 13 Jul 2026 11:15:35 -0400 Subject: [PATCH 4/4] chore(protocols): relax NOW PDU dependency Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Cargo.toml | 2 +- protocols/rust/now-client/Cargo.toml | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 8aa6e1c..4078ba9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ keywords = ["rdp", "remote-desktop", "network", "client", "protocol"] categories = ["network-programming"] [workspace.dependencies] -now-proto-pdu = { version = "=0.4.3", path = "protocols/rust/now-proto-pdu" } +now-proto-pdu = { version = "0.4", path = "protocols/rust/now-proto-pdu" } now-proto-fuzzing = { version = "0.1", path = "protocols/rust/now-proto-fuzzing" } [profile.test.package.proptest] diff --git a/protocols/rust/now-client/Cargo.toml b/protocols/rust/now-client/Cargo.toml index 6bf0a75..193112f 100644 --- a/protocols/rust/now-client/Cargo.toml +++ b/protocols/rust/now-client/Cargo.toml @@ -10,7 +10,6 @@ repository.workspace = true authors.workspace = true keywords.workspace = true categories.workspace = true -publish = true [lints] workspace = true