From c816627ca887cf71fbb0ed3a917071dda6ef552a Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 7 Sep 2026 16:06:28 -0400 Subject: [PATCH 01/17] feat(dgw): route JMUX channels through agents Select an advertised agent for each JMUX channel by destination. Fall back to direct TCP only when no agent route matches. Keep channel failures isolated with bounded client errors and emit ConnectFailure audit events when the target IP is known. --- crates/jmux-proxy/src/lib.rs | 121 +++++++++++++++++++++++++--- devolutions-gateway/src/api/jmux.rs | 16 +++- devolutions-gateway/src/jmux.rs | 44 +++++++++- 3 files changed, 164 insertions(+), 17 deletions(-) diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index 7373fce85..2f90df599 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -12,7 +12,10 @@ mod id_allocator; use std::collections::{HashMap, HashSet}; use std::convert::TryFrom; +use std::future::Future; use std::io; +use std::net::IpAddr; +use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::time::SystemTime; @@ -22,7 +25,6 @@ use bytes::Bytes; use jmux_proto::{ChannelData, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::net::TcpStream; -use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::sync::{Notify, mpsc, oneshot}; use tokio::task::JoinHandle; use tokio_util::codec::FramedRead; @@ -53,6 +55,28 @@ pub type ApiResponseReceiver = oneshot::Receiver; pub type ApiRequestSender = mpsc::Sender; pub type ApiRequestReceiver = mpsc::Receiver; +trait TargetStream: AsyncRead + AsyncWrite + Unpin + Send {} + +impl TargetStream for T where T: AsyncRead + AsyncWrite + Unpin + Send {} + +type ErasedTargetStream = Box; +type TargetConnectorFuture = Pin>> + Send>>; +type TargetConnector = Arc TargetConnectorFuture + Send + Sync>; + +pub struct ConnectedTarget { + stream: ErasedTargetStream, + target_ip: Option, +} + +impl ConnectedTarget { + pub fn new(stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static, target_ip: Option) -> Self { + Self { + stream: Box::new(stream), + target_ip, + } + } +} + #[derive(Debug)] pub enum JmuxApiRequest { OpenChannel { @@ -84,6 +108,7 @@ pub struct JmuxProxy { jmux_reader: Box, jmux_writer: Box, traffic_callback: Option, + target_connector: Option, } impl JmuxProxy { @@ -98,6 +123,7 @@ impl JmuxProxy { jmux_reader, jmux_writer, traffic_callback: None, + target_connector: None, } } @@ -113,6 +139,19 @@ impl JmuxProxy { self } + /// Tries a custom target connection before falling back to direct TCP. + /// + /// Return `Ok(None)` when the target should use the default direct connection. + #[must_use] + pub fn with_target_connector(mut self, connector: C) -> Self + where + C: Fn(DestinationUrl) -> F + Send + Sync + 'static, + F: Future>> + Send + 'static, + { + self.target_connector = Some(Arc::new(move |destination_url| Box::pin(connector(destination_url)))); + self + } + /// Configures an outgoing-traffic callback for lifecycle event monitoring. /// /// The provided callback will be invoked exactly once per outgoing stream at the end of its @@ -186,6 +225,7 @@ async fn run_proxy_impl(proxy: JmuxProxy, span: Span) -> anyhow::Result<()> { jmux_reader, jmux_writer, traffic_callback, + target_connector, } = proxy; let (msg_to_send_tx, msg_to_send_rx) = mpsc::channel::(JMUX_MESSAGE_MPSC_CHANNEL_SIZE); @@ -206,6 +246,7 @@ async fn run_proxy_impl(proxy: JmuxProxy, span: Span) -> anyhow::Result<()> { msg_to_send_tx, api_request_rx, traffic_callback, + target_connector, parent_span: span, } .spawn(); @@ -255,7 +296,7 @@ struct JmuxChannelCtx { // Traffic audit metadata target_host: String, /// Target server resolved address IP - target_ip: Option, + target_ip: Option, /// Target server port target_port: u16, /// Time the connection with target peer was established at @@ -344,7 +385,6 @@ type DataReceiver = mpsc::Receiver; type DataSender = mpsc::Sender; type InternalMessageSender = mpsc::Sender; -#[derive(Debug)] enum InternalMessage { Eof { id: LocalChannelId, @@ -353,7 +393,7 @@ enum InternalMessage { // Boxing reduces enum size from 224 bytes to ~16 bytes // (clippy::large_enum_variant) channel: Box, - stream: TcpStream, + stream: ErasedTargetStream, }, AbnormalTermination { id: LocalChannelId, @@ -427,6 +467,7 @@ struct JmuxSchedulerTask { msg_to_send_tx: MessageSender, api_request_rx: ApiRequestReceiver, traffic_callback: Option, + target_connector: Option, parent_span: Span, } @@ -448,6 +489,7 @@ async fn scheduler_task_impl(task: JmuxSc msg_to_send_tx, mut api_request_rx, traffic_callback, + target_connector, parent_span, } = task; @@ -501,7 +543,8 @@ async fn scheduler_task_impl(task: JmuxSc error!(%error, "Couldn't send leftover bytes"); } - let (reader, writer) = stream.into_split(); + let stream = Box::new(stream) as ErasedTargetStream; + let (reader, writer) = tokio::io::split(stream); DataWriterTask { writer, @@ -626,7 +669,7 @@ async fn scheduler_task_impl(task: JmuxSc debug!("Channel accepted"); }); - let (reader, writer) = stream.into_split(); + let (reader, writer) = tokio::io::split(stream); DataWriterTask { writer, @@ -760,6 +803,7 @@ async fn scheduler_task_impl(task: JmuxSc internal_msg_tx: internal_msg_tx.clone(), msg_to_send_tx: msg_to_send_tx.clone(), traffic_callback: traffic_callback.clone(), + target_connector: target_connector.clone(), } .spawn() .detach(); @@ -958,7 +1002,7 @@ async fn scheduler_task_impl(task: JmuxSc // ---------------------- // struct DataReaderTask { - reader: OwnedReadHalf, + reader: tokio::io::ReadHalf, local_id: LocalChannelId, distant_id: DistantChannelId, window_size_updated: Arc, @@ -1087,7 +1131,7 @@ impl DataReaderTask { // ---------------------- // struct DataWriterTask { - writer: OwnedWriteHalf, + writer: tokio::io::WriteHalf, data_rx: DataReceiver, /// Tracks bytes written into the stream. bytes_tx: Arc, @@ -1139,6 +1183,7 @@ struct StreamResolverTask { internal_msg_tx: InternalMessageSender, msg_to_send_tx: MessageSender, traffic_callback: Option, + target_connector: Option, } impl StreamResolverTask { @@ -1164,6 +1209,7 @@ impl StreamResolverTask { internal_msg_tx, msg_to_send_tx, traffic_callback, + target_connector, } = self; let scheme = destination_url.scheme(); @@ -1172,6 +1218,59 @@ impl StreamResolverTask { match scheme { "tcp" => { + if let Some(connector) = target_connector { + match connector(destination_url.clone()).await { + Ok(Some(ConnectedTarget { stream, target_ip })) => { + channel.target_ip = target_ip; + channel.connect_at = SystemTime::now(); + + internal_msg_tx + .send(InternalMessage::StreamResolved { + channel: Box::new(channel), + stream, + }) + .await + .map_err(|_| { + anyhow::anyhow!("couldn't send back resolved stream through internal mpsc channel") + })?; + + return Ok(()); + } + Ok(None) => {} + Err(error) => { + if let Some(callback) = &traffic_callback + && let Ok(target_ip) = host.parse::() + { + let connect_and_disconnect_time = SystemTime::now(); + + callback(TrafficEvent { + outcome: EventOutcome::ConnectFailure, + protocol: TransportProtocol::Tcp, + target_host: channel.target_host.clone(), + target_ip, + target_port: channel.target_port, + connect_at: connect_and_disconnect_time, + disconnect_at: connect_and_disconnect_time, + active_duration: std::time::Duration::ZERO, + bytes_tx: 0, + bytes_rx: 0, + }); + } + + msg_to_send_tx + .send(Message::open_failure( + channel.distant_id, + ReasonCode::GENERAL_FAILURE, + "target connection failed", + )) + .await + .context("couldn't send OPEN FAILURE message through mpsc channel")?; + + return Err(error.context(format!("couldn't connect to {host}:{port}"))); + } + } + } + // Perform DNS resolution first to get concrete IP addresses. let socket_addrs = match tokio::net::lookup_host((host, port)).await { Ok(addrs) => addrs, @@ -1203,10 +1302,12 @@ impl StreamResolverTask { internal_msg_tx .send(InternalMessage::StreamResolved { channel: Box::new(channel), - stream, + stream: Box::new(stream), }) .await - .context("couldn't send back resolved stream through internal mpsc channel")?; + .map_err(|_| { + anyhow::anyhow!("couldn't send back resolved stream through internal mpsc channel") + })?; return Ok(()); } diff --git a/devolutions-gateway/src/api/jmux.rs b/devolutions-gateway/src/api/jmux.rs index b9199c01f..e40bcf3e6 100644 --- a/devolutions-gateway/src/api/jmux.rs +++ b/devolutions-gateway/src/api/jmux.rs @@ -22,6 +22,7 @@ pub async fn handler( shutdown_signal, conf_handle, traffic_audit_handle, + agent_tunnel_handle, .. }): State, JmuxToken(claims): JmuxToken, @@ -35,6 +36,7 @@ pub async fn handler( sessions, subscriber_tx, traffic_audit_handle, + agent_tunnel_handle, claims, source_addr, Duration::from_secs(conf_handle.get_conf().debug.ws_keep_alive_interval), @@ -54,6 +56,7 @@ async fn handle_socket( sessions: SessionMessageSender, subscriber_tx: SubscriberSender, traffic_audit_handle: TrafficAuditHandle, + agent_tunnel_handle: Option>, claims: JmuxTokenClaims, source_addr: SocketAddr, keep_alive_interval: Duration, @@ -65,9 +68,16 @@ async fn handle_socket( ); let session_id = claims.jet_aid; - let result = crate::jmux::handle(stream, claims, sessions, subscriber_tx, traffic_audit_handle) - .instrument(info_span!("jmux", client = %source_addr, %session_id)) - .await; + let result = crate::jmux::handle( + stream, + claims, + sessions, + subscriber_tx, + traffic_audit_handle, + agent_tunnel_handle, + ) + .instrument(info_span!("jmux", client = %source_addr, %session_id)) + .await; if let Err(error) = result { close_handle.server_error("JMUX failure".to_owned()).await; diff --git a/devolutions-gateway/src/jmux.rs b/devolutions-gateway/src/jmux.rs index 500ec0963..f50dd46e7 100644 --- a/devolutions-gateway/src/jmux.rs +++ b/devolutions-gateway/src/jmux.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use anyhow::Context as _; use devolutions_gateway_task::ChildTask; -use jmux_proxy::{FilteringRule, JmuxConfig, JmuxProxy}; +use jmux_proxy::{ConnectedTarget, DestinationUrl, FilteringRule, JmuxConfig, JmuxProxy}; use tap::prelude::*; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::sync::Notify; @@ -10,8 +10,10 @@ use transport::{ErasedRead, ErasedWrite}; use crate::session::{ConnectionModeDetails, SessionInfo, SessionMessageSender}; use crate::subscriber::SubscriberSender; +use crate::target_addr::TargetAddr; use crate::token::{JmuxTokenClaims, RecordingPolicy}; use crate::traffic_audit::TrafficAuditHandle; +use crate::upstream::route_target_from_target_addr; pub async fn handle( stream: impl AsyncRead + AsyncWrite + Send + 'static, @@ -19,6 +21,7 @@ pub async fn handle( sessions: SessionMessageSender, subscriber_tx: SubscriberSender, traffic_audit_handle: TrafficAuditHandle, + agent_tunnel_handle: Option>, ) -> anyhow::Result<()> { match claims.jet_rec { RecordingPolicy::None | RecordingPolicy::Stream => (), @@ -105,10 +108,43 @@ pub async fn handle( }); }; - let proxy_fut = JmuxProxy::new(reader, writer) + let mut proxy = JmuxProxy::new(reader, writer) .with_config(config) - .with_outgoing_traffic_event_callback(traffic_event_callback) - .run(); + .with_outgoing_traffic_event_callback(traffic_event_callback); + + if let Some(agent_tunnel_handle) = agent_tunnel_handle { + proxy = proxy.with_target_connector(move |destination_url: DestinationUrl| { + let agent_tunnel_handle = Arc::clone(&agent_tunnel_handle); + + async move { + let target = TargetAddr::from_components( + destination_url.scheme(), + destination_url.host(), + destination_url.port(), + ) + .context("invalid JMUX target")?; + let route_target = route_target_from_target_addr(&target); + + let routed = agent_tunnel::routing::try_route( + Some(agent_tunnel_handle.as_ref()), + // TODO: Pass `jet_agent_id` after JMUX consumers start issuing it. + None, + &route_target, + session_id, + target.as_addr(), + ) + .await?; + + let Some((stream, _agent)) = routed else { + return Ok(None); + }; + + Ok(Some(ConnectedTarget::new(stream, target.host_ip()))) + } + }); + } + + let proxy_fut = proxy.run(); let proxy_handle = ChildTask::spawn(proxy_fut); let join_fut = proxy_handle.join(); tokio::pin!(join_fut); From 8995646df5c39b8dfe669730a8c109e16025d814 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 7 Sep 2026 16:48:22 -0400 Subject: [PATCH 02/17] fix(dgw,agent): preserve routed target addresses Return the Agent-selected socket address so JMUX can audit hostname targets accurately. Keep opaque connector failures out of target-failure audit events and cover connector success, fallback, failure, and channel isolation. --- crates/agent-tunnel-proto/src/session.rs | 24 ++- .../agent-tunnel-proto/src/session_codec.rs | 25 ++- crates/agent-tunnel-proto/tests/session.rs | 29 ++++ crates/agent-tunnel/src/listener.rs | 11 +- crates/agent-tunnel/src/routing.rs | 10 +- crates/jmux-proxy/src/lib.rs | 24 +-- crates/jmux-proxy/tests/target_connector.rs | 155 ++++++++++++++++++ devolutions-agent/src/tunnel.rs | 2 +- devolutions-gateway/src/jmux.rs | 6 +- devolutions-gateway/src/kdc_connector.rs | 2 +- devolutions-gateway/src/upstream.rs | 10 +- 11 files changed, 252 insertions(+), 46 deletions(-) create mode 100644 crates/jmux-proxy/tests/target_connector.rs diff --git a/crates/agent-tunnel-proto/src/session.rs b/crates/agent-tunnel-proto/src/session.rs index 4a3f153e5..8e2e563f1 100644 --- a/crates/agent-tunnel-proto/src/session.rs +++ b/crates/agent-tunnel-proto/src/session.rs @@ -2,6 +2,8 @@ //! //! Encoding and decoding live in [`crate::session_codec`]. +use std::net::SocketAddr; + use uuid::Uuid; use crate::version::CURRENT_PROTOCOL_VERSION; @@ -33,12 +35,18 @@ pub enum ConnectRequest { /// Agent's response to a [`ConnectRequest`]. /// /// Wire layout: -/// - Success: `[1B tag=0x00][2B version]` +/// - Success: `[1B tag=0x00][2B version][optional 4B address length + address bytes]` /// - Error: `[1B tag=0x01][2B version][4B reason_len][reason bytes]` #[derive(Debug, Clone, PartialEq, Eq)] pub enum ConnectResponse { - Success { protocol_version: u16 }, - Error { protocol_version: u16, reason: String }, + Success { + protocol_version: u16, + target_addr: Option, + }, + Error { + protocol_version: u16, + reason: String, + }, } impl ConnectRequest { @@ -77,6 +85,14 @@ impl ConnectResponse { pub fn success() -> Self { Self::Success { protocol_version: CURRENT_PROTOCOL_VERSION, + target_addr: None, + } + } + + pub fn success_with_target(target_addr: SocketAddr) -> Self { + Self::Success { + protocol_version: CURRENT_PROTOCOL_VERSION, + target_addr: Some(target_addr), } } @@ -94,7 +110,7 @@ impl ConnectResponse { /// Extract the protocol version from any variant. pub fn protocol_version(&self) -> u16 { match self { - Self::Success { protocol_version } | Self::Error { protocol_version, .. } => *protocol_version, + Self::Success { protocol_version, .. } | Self::Error { protocol_version, .. } => *protocol_version, } } } diff --git a/crates/agent-tunnel-proto/src/session_codec.rs b/crates/agent-tunnel-proto/src/session_codec.rs index 725d8b1ca..3f729cdea 100644 --- a/crates/agent-tunnel-proto/src/session_codec.rs +++ b/crates/agent-tunnel-proto/src/session_codec.rs @@ -57,9 +57,15 @@ impl Decode for ConnectRequest { impl Encode for ConnectResponse { fn encode(&self, buf: &mut BytesMut) { match self { - Self::Success { protocol_version } => { + Self::Success { + protocol_version, + target_addr, + } => { buf.put_u8(TAG_RESPONSE_SUCCESS); buf.put_u16(*protocol_version); + if let Some(target_addr) = target_addr { + codec::put_string(buf, &target_addr.to_string()); + } } Self::Error { protocol_version, @@ -80,7 +86,22 @@ impl Decode for ConnectResponse { let protocol_version = buf.get_u16(); match tag { - TAG_RESPONSE_SUCCESS => Ok(Self::Success { protocol_version }), + TAG_RESPONSE_SUCCESS => { + let target_addr = if buf.has_remaining() { + let value = codec::get_string(&mut buf)?; + Some(value.parse().map_err(|_| ProtoError::InvalidField { + field: "target_addr", + reason: "not a socket address", + })?) + } else { + None + }; + + Ok(Self::Success { + protocol_version, + target_addr, + }) + } TAG_RESPONSE_ERROR => { let reason = codec::get_string(&mut buf)?; Ok(Self::Error { diff --git a/crates/agent-tunnel-proto/tests/session.rs b/crates/agent-tunnel-proto/tests/session.rs index c8c4168e6..43816a2c9 100644 --- a/crates/agent-tunnel-proto/tests/session.rs +++ b/crates/agent-tunnel-proto/tests/session.rs @@ -1,3 +1,5 @@ +use std::net::{Ipv4Addr, SocketAddr}; + use agent_tunnel_proto::{ConnectRequest, ConnectResponse, MAX_SESSION_MESSAGE_SIZE, ProtoError, SessionStream}; use uuid::Uuid; @@ -27,6 +29,20 @@ async fn roundtrip_connect_response_success() { assert_eq!(msg, decoded); } +#[tokio::test] +async fn roundtrip_connect_response_success_with_target() { + let target_addr = SocketAddr::from((Ipv4Addr::new(192, 0, 2, 10), 3389)); + let msg = ConnectResponse::success_with_target(target_addr); + + let mut buf = Vec::new(); + let mut stream = SessionStream::new(&mut buf, &[][..]); + stream.send_response(&msg).await.expect("send should succeed"); + + let mut stream = SessionStream::new(tokio::io::sink(), buf.as_slice()); + let decoded = stream.recv_response().await.expect("recv should succeed"); + assert_eq!(msg, decoded); +} + #[tokio::test] async fn roundtrip_connect_response_error() { let msg = ConnectResponse::error("connection refused"); @@ -109,6 +125,19 @@ async fn decode_rejects_unknown_connect_response_tag() { assert!(matches!(err, ProtoError::UnknownTag { tag: 0xFF }), "got {err:?}"); } +#[tokio::test] +async fn decode_rejects_invalid_success_target_address() { + let payload = &[0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x03, b'b', b'a', b'd']; + let err = recv_response_payload(payload).await; + assert!(matches!( + err, + ProtoError::InvalidField { + field: "target_addr", + .. + } + )); +} + #[tokio::test] async fn decode_rejects_unknown_connect_request_tag() { // Minimum bytes for header-length check (1 tag + 2 version + 16 uuid = 19), diff --git a/crates/agent-tunnel/src/listener.rs b/crates/agent-tunnel/src/listener.rs index 1e548c05c..a0708c824 100644 --- a/crates/agent-tunnel/src/listener.rs +++ b/crates/agent-tunnel/src/listener.rs @@ -91,7 +91,7 @@ impl AgentTunnelHandle { agent_id: Uuid, session_id: Uuid, target: &str, - ) -> anyhow::Result { + ) -> anyhow::Result<(TunnelStream, Option)> { let conn = self .agent_connections .read() @@ -122,9 +122,10 @@ impl AgentTunnelHandle { agent_tunnel_proto::validate_protocol_version(response.protocol_version()) .map_err(|e| anyhow::anyhow!("ConnectResponse: {e}"))?; - if let ConnectResponse::Error { reason, .. } = &response { - anyhow::bail!("agent refused connection: {reason}"); - } + let target_addr = match response { + ConnectResponse::Success { target_addr, .. } => target_addr, + ConnectResponse::Error { reason, .. } => anyhow::bail!("agent refused connection: {reason}"), + }; info!( %agent_id, @@ -134,7 +135,7 @@ impl AgentTunnelHandle { ); let (send, recv) = session.into_inner(); - Ok(TunnelStream { send, recv }) + Ok((TunnelStream { send, recv }, target_addr)) } } diff --git a/crates/agent-tunnel/src/routing.rs b/crates/agent-tunnel/src/routing.rs index b6d9a967a..629910598 100644 --- a/crates/agent-tunnel/src/routing.rs +++ b/crates/agent-tunnel/src/routing.rs @@ -3,7 +3,7 @@ //! Consumed by the upstream connection paths (forwarding, RDP clean path, //! generic client) to ensure consistent routing behavior and error messages. -use std::net::IpAddr; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use agent_tunnel_proto::DomainName; @@ -95,7 +95,7 @@ pub async fn try_route( target: &RouteTarget, session_id: Uuid, target_addr: &str, -) -> Result)>> { +) -> Result, Option)>> { let Some(handle) = handle else { // An explicit `jet_agent_id` claim means the token requires routing via that // specific agent; silently falling back to a direct connect would bypass the @@ -131,7 +131,7 @@ pub async fn route_and_connect( candidates: &[Arc], session_id: Uuid, target: &str, -) -> Result<(TunnelStream, Arc)> { +) -> Result<(TunnelStream, Arc, Option)> { if candidates.is_empty() { return Err(anyhow!("route_and_connect called with empty candidates")); } @@ -147,14 +147,14 @@ pub async fn route_and_connect( ); match handle.connect_via_agent(agent.agent_id, session_id, target).await { - Ok(stream) => { + Ok((stream, target_addr)) => { info!( agent_id = %agent.agent_id, agent_name = %agent.name, %target, "Agent tunnel connection established" ); - return Ok((stream, Arc::clone(agent))); + return Ok((stream, Arc::clone(agent), target_addr)); } Err(error) => { warn!( diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index 2f90df599..eb4bc7a4a 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -69,6 +69,9 @@ pub struct ConnectedTarget { } impl ConnectedTarget { + /// Wraps a connected stream and the concrete peer IP used for traffic auditing. + /// + /// When `target_ip` is `None`, the channel works normally but emits no traffic event. pub fn new(stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static, target_ip: Option) -> Self { Self { stream: Box::new(stream), @@ -142,6 +145,8 @@ impl JmuxProxy { /// Tries a custom target connection before falling back to direct TCP. /// /// Return `Ok(None)` when the target should use the default direct connection. + /// Return the concrete peer IP in `ConnectedTarget` to enable traffic auditing. + /// Connector errors do not emit `ConnectFailure` because they do not prove that a concrete address was attempted. #[must_use] pub fn with_target_connector(mut self, connector: C) -> Self where @@ -1238,25 +1243,6 @@ impl StreamResolverTask { } Ok(None) => {} Err(error) => { - if let Some(callback) = &traffic_callback - && let Ok(target_ip) = host.parse::() - { - let connect_and_disconnect_time = SystemTime::now(); - - callback(TrafficEvent { - outcome: EventOutcome::ConnectFailure, - protocol: TransportProtocol::Tcp, - target_host: channel.target_host.clone(), - target_ip, - target_port: channel.target_port, - connect_at: connect_and_disconnect_time, - disconnect_at: connect_and_disconnect_time, - active_duration: std::time::Duration::ZERO, - bytes_tx: 0, - bytes_rx: 0, - }); - } - msg_to_send_tx .send(Message::open_failure( channel.distant_id, diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs new file mode 100644 index 000000000..4549a01a2 --- /dev/null +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -0,0 +1,155 @@ +#![allow(unused_crate_dependencies)] +#![allow(clippy::unwrap_used)] + +use std::net::{IpAddr, Ipv4Addr}; +use std::time::Duration; + +use jmux_proto::{BytesMut, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; +use jmux_proxy::{ConnectedTarget, DestinationUrl, EventOutcome, JmuxConfig, JmuxProxy, TrafficEvent}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::sync::mpsc; +use tokio::time::timeout; + +const TEST_TIMEOUT: Duration = Duration::from_secs(5); + +async fn send_message(writer: &mut (impl AsyncWrite + Unpin), message: Message) { + let mut bytes = BytesMut::new(); + message.encode(&mut bytes).expect("encode JMUX message"); + writer.write_all(&bytes).await.expect("send JMUX message"); +} + +async fn receive_message(reader: &mut (impl AsyncRead + Unpin)) -> Message { + timeout(TEST_TIMEOUT, async { + let mut header = [0; Header::SIZE]; + reader.read_exact(&mut header).await.expect("read JMUX header"); + let message_size = usize::from(u16::from_be_bytes([header[1], header[2]])); + let mut body = vec![0; message_size - Header::SIZE]; + reader.read_exact(&mut body).await.expect("read JMUX body"); + + let mut bytes = BytesMut::with_capacity(message_size); + bytes.extend_from_slice(&header); + bytes.extend_from_slice(&body); + Message::decode(bytes.freeze()).expect("decode JMUX message") + }) + .await + .expect("JMUX response timed out") +} + +#[tokio::test] +async fn connected_target_ip_is_used_for_audit() { + let (proxy_stream, peer_stream) = tokio::io::duplex(8192); + let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); + let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); + let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); + let target_ip = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 10)); + + let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) + .with_config(JmuxConfig::permissive()) + .with_target_connector(move |destination| async move { + assert_eq!(destination.host(), "agent.example"); + let (target_stream, mut target_peer) = tokio::io::duplex(64); + tokio::spawn(async move { + target_peer.shutdown().await.expect("close target stream"); + }); + Ok(Some(ConnectedTarget::new(target_stream, Some(target_ip)))) + }) + .with_outgoing_traffic_event_callback(move |event| { + event_tx.send(event).expect("capture traffic event"); + }); + let proxy_task = tokio::spawn(proxy.run()); + + send_message( + &mut peer_writer, + Message::open( + LocalChannelId::from(7), + 4096, + DestinationUrl::new("tcp", "agent.example", 443), + ), + ) + .await; + + let Message::OpenSuccess(open_success) = receive_message(&mut peer_reader).await else { + panic!("expected OPEN SUCCESS"); + }; + let local_id = DistantChannelId::from(open_success.sender_channel_id); + + assert!(matches!(receive_message(&mut peer_reader).await, Message::Eof(_))); + send_message(&mut peer_writer, Message::eof(local_id)).await; + assert!(matches!(receive_message(&mut peer_reader).await, Message::Close(_))); + send_message(&mut peer_writer, Message::close(local_id)).await; + + let event = timeout(TEST_TIMEOUT, event_rx.recv()) + .await + .expect("traffic event timed out") + .expect("traffic event channel closed"); + assert_eq!(event.outcome, EventOutcome::NormalTermination); + assert_eq!(event.target_host, "agent.example"); + assert_eq!(event.target_ip, target_ip); + assert_eq!(event.target_port, 443); + + proxy_task.abort(); +} + +#[tokio::test] +async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { + let (proxy_stream, peer_stream) = tokio::io::duplex(8192); + let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); + let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); + let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); + + let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) + .with_config(JmuxConfig::permissive()) + .with_target_connector(|destination| async move { + if destination.host() == "fail.example" { + anyhow::bail!("{}", "agent error ".repeat(8192)); + } + Ok(None) + }) + .with_outgoing_traffic_event_callback(move |event| { + event_tx.send(event).expect("capture traffic event"); + }); + let proxy_task = tokio::spawn(proxy.run()); + + send_message( + &mut peer_writer, + Message::open( + LocalChannelId::from(11), + 4096, + DestinationUrl::new("tcp", "fail.example", 443), + ), + ) + .await; + + let Message::OpenFailure(open_failure) = receive_message(&mut peer_reader).await else { + panic!("expected OPEN FAILURE"); + }; + assert_eq!(open_failure.reason_code, ReasonCode::GENERAL_FAILURE); + assert_eq!(open_failure.description, "target connection failed"); + assert!(timeout(Duration::from_millis(100), event_rx.recv()).await.is_err()); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind direct target"); + let target_port = listener.local_addr().expect("read direct target address").port(); + let server_task = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.expect("accept direct connection"); + std::future::pending::<()>().await; + }); + + send_message( + &mut peer_writer, + Message::open( + LocalChannelId::from(12), + 4096, + DestinationUrl::new("tcp", "127.0.0.1", target_port), + ), + ) + .await; + assert!(matches!( + receive_message(&mut peer_reader).await, + Message::OpenSuccess(_) + )); + + server_task.abort(); + proxy_task.abort(); +} diff --git a/devolutions-agent/src/tunnel.rs b/devolutions-agent/src/tunnel.rs index e19c69d65..f261adfa7 100644 --- a/devolutions-agent/src/tunnel.rs +++ b/devolutions-agent/src/tunnel.rs @@ -778,7 +778,7 @@ async fn run_session_proxy( info!(target = %selected_target, "TCP connection established"); session - .send_response(&ConnectResponse::success()) + .send_response(&ConnectResponse::success_with_target(selected_target)) .await .context("send ConnectResponse")?; info!("Sent ConnectResponse::Success"); diff --git a/devolutions-gateway/src/jmux.rs b/devolutions-gateway/src/jmux.rs index f50dd46e7..174094ee3 100644 --- a/devolutions-gateway/src/jmux.rs +++ b/devolutions-gateway/src/jmux.rs @@ -135,11 +135,13 @@ pub async fn handle( ) .await?; - let Some((stream, _agent)) = routed else { + let Some((stream, _agent, target_addr)) = routed else { return Ok(None); }; - Ok(Some(ConnectedTarget::new(stream, target.host_ip()))) + let target_ip = target_addr.map(|addr| addr.ip()).or_else(|| target.host_ip()); + + Ok(Some(ConnectedTarget::new(stream, target_ip))) } }); } diff --git a/devolutions-gateway/src/kdc_connector.rs b/devolutions-gateway/src/kdc_connector.rs index f1840b643..bf14aa078 100644 --- a/devolutions-gateway/src/kdc_connector.rs +++ b/devolutions-gateway/src/kdc_connector.rs @@ -81,7 +81,7 @@ impl KdcConnector { .await .map_err(|e| HttpError::bad_gateway().build(format!("KDC routing through agent tunnel failed: {e:#}")))?; - if let Some((mut stream, _)) = route_result { + if let Some((mut stream, _, _)) = route_result { // The agent tunnel currently carries only TCP (`ConnectRequest::tcp`). If the // routing pipeline picked an agent for a udp:// KDC target — either by subnet // match or by explicit pin — we must reject explicitly. Silently falling diff --git a/devolutions-gateway/src/upstream.rs b/devolutions-gateway/src/upstream.rs index c6f5d9ad0..3b38069da 100644 --- a/devolutions-gateway/src/upstream.rs +++ b/devolutions-gateway/src/upstream.rs @@ -264,15 +264,11 @@ impl<'a> RoutePlan<'a> { .connect_via_agent(agent.agent_id, session_id, target.as_addr()) .await { - Ok(stream) => { - // The TCP peer lives on the agent side; surface the target - // IP:port for logs/PCAP when the target is a literal IP, or - // 0.0.0.0: when it's a hostname the gateway never - // resolved itself. Either is more useful than 0.0.0.0:0. - let server_addr = match target.host_ip() { + Ok((stream, target_addr)) => { + let server_addr = target_addr.unwrap_or_else(|| match target.host_ip() { Some(ip) => SocketAddr::new(ip, target.port()), None => SocketAddr::from((std::net::Ipv4Addr::UNSPECIFIED, target.port())), - }; + }); return Ok(ConnectedUpstream { leg: UpstreamLeg::Tunnel(stream), From 54c106ca757186709c6b77d5659c9d22edad2fc9 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 7 Sep 2026 19:41:11 -0400 Subject: [PATCH 03/17] fix(dgw,jetsocat): release failed JMUX channel IDs --- crates/jmux-proxy/src/lib.rs | 101 ++++++++++++++------ crates/jmux-proxy/tests/target_connector.rs | 8 +- 2 files changed, 77 insertions(+), 32 deletions(-) diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index eb4bc7a4a..9fdbdcccd 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -400,6 +400,12 @@ enum InternalMessage { channel: Box, stream: ErasedTargetStream, }, + StreamResolutionFailed { + id: LocalChannelId, + distant_id: DistantChannelId, + reason_code: ReasonCode, + description: String, + }, AbnormalTermination { id: LocalChannelId, }, @@ -700,6 +706,18 @@ async fn scheduler_task_impl(task: JmuxSc .spawn(channel_span) .detach(); } + InternalMessage::StreamResolutionFailed { + id, + distant_id, + reason_code, + description, + } => { + jmux_ctx.id_allocator.free(id); + msg_to_send_tx + .send(Message::open_failure(distant_id, reason_code, description)) + .await + .context("couldn't send OPEN FAILURE message through mpsc channel")?; + } } } msg = jmux_stream.next() => { @@ -806,7 +824,6 @@ async fn scheduler_task_impl(task: JmuxSc channel, destination_url: msg.destination_url, internal_msg_tx: internal_msg_tx.clone(), - msg_to_send_tx: msg_to_send_tx.clone(), traffic_callback: traffic_callback.clone(), target_connector: target_connector.clone(), } @@ -1186,7 +1203,6 @@ struct StreamResolverTask { channel: JmuxChannelCtx, destination_url: DestinationUrl, internal_msg_tx: InternalMessageSender, - msg_to_send_tx: MessageSender, traffic_callback: Option, target_connector: Option, } @@ -1207,12 +1223,28 @@ impl StreamResolverTask { ChildTask(handle) } + async fn report_failure( + internal_msg_tx: &InternalMessageSender, + channel: &JmuxChannelCtx, + reason_code: ReasonCode, + description: String, + ) -> anyhow::Result<()> { + internal_msg_tx + .send(InternalMessage::StreamResolutionFailed { + id: channel.local_id, + distant_id: channel.distant_id, + reason_code, + description, + }) + .await + .map_err(|_| anyhow::anyhow!("couldn't report stream resolution failure through internal mpsc channel")) + } + async fn run(self) -> anyhow::Result<()> { let Self { mut channel, destination_url, internal_msg_tx, - msg_to_send_tx, traffic_callback, target_connector, } = self; @@ -1243,14 +1275,13 @@ impl StreamResolverTask { } Ok(None) => {} Err(error) => { - msg_to_send_tx - .send(Message::open_failure( - channel.distant_id, - ReasonCode::GENERAL_FAILURE, - "target connection failed", - )) - .await - .context("couldn't send OPEN FAILURE message through mpsc channel")?; + Self::report_failure( + &internal_msg_tx, + &channel, + ReasonCode::GENERAL_FAILURE, + "target connection failed".to_owned(), + ) + .await?; return Err(error.context(format!("couldn't connect to {host}:{port}"))); } @@ -1263,14 +1294,13 @@ impl StreamResolverTask { Err(error) => { debug!(?error, "DNS resolution failed"); // No event emission for DNS failures - cannot determine target IP. - msg_to_send_tx - .send(Message::open_failure( - channel.distant_id, - ReasonCode::from(error.kind()), - error.to_string(), - )) - .await - .context("couldn't send OPEN FAILURE message through mpsc channel")?; + Self::report_failure( + &internal_msg_tx, + &channel, + ReasonCode::from(error.kind()), + error.to_string(), + ) + .await?; anyhow::bail!("couldn't resolve {host}:{port}: {error}"); } }; @@ -1324,21 +1354,36 @@ impl StreamResolverTask { }); } - msg_to_send_tx - .send(Message::open_failure( - channel.distant_id, - ReasonCode::from(error.kind()), - error.to_string(), - )) - .await - .context("couldn't send OPEN FAILURE message through mpsc channel")?; + Self::report_failure( + &internal_msg_tx, + &channel, + ReasonCode::from(error.kind()), + error.to_string(), + ) + .await?; anyhow::bail!("couldn't open TCP stream to {host}:{port}: {error}"); } else { + Self::report_failure( + &internal_msg_tx, + &channel, + ReasonCode::GENERAL_FAILURE, + "no addresses resolved".to_owned(), + ) + .await?; anyhow::bail!("no addresses resolved for {host}:{port}"); } } - _ => anyhow::bail!("unsupported scheme: {scheme}"), + _ => { + Self::report_failure( + &internal_msg_tx, + &channel, + ReasonCode::GENERAL_FAILURE, + format!("unsupported scheme: {scheme}"), + ) + .await?; + anyhow::bail!("unsupported scheme: {scheme}") + } } } } diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs index 4549a01a2..1741fe8a7 100644 --- a/crates/jmux-proxy/tests/target_connector.rs +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -145,10 +145,10 @@ async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { ), ) .await; - assert!(matches!( - receive_message(&mut peer_reader).await, - Message::OpenSuccess(_) - )); + let Message::OpenSuccess(open_success) = receive_message(&mut peer_reader).await else { + panic!("expected OPEN SUCCESS"); + }; + assert_eq!(open_success.sender_channel_id, 0); server_task.abort(); proxy_task.abort(); From d954814f19bd8d30b1c6b8c17dee2a2104da43e8 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 7 Sep 2026 19:45:57 -0400 Subject: [PATCH 04/17] docs(dgw): document routed target address --- crates/agent-tunnel/src/routing.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/agent-tunnel/src/routing.rs b/crates/agent-tunnel/src/routing.rs index 629910598..80752ef16 100644 --- a/crates/agent-tunnel/src/routing.rs +++ b/crates/agent-tunnel/src/routing.rs @@ -122,7 +122,8 @@ pub async fn try_route( /// Try connecting to target through agent candidates (try-fail-retry). /// -/// Returns the connected `TunnelStream` and the agent that succeeded. +/// Returns the connected stream, the agent that succeeded, and its reported target address. +/// The address is `None` when the agent sends the legacy success response. /// /// Callers must handle `RoutingDecision::ExplicitAgentNotFound` and /// `RoutingDecision::Direct` before calling this function. From c10d66a75a7e6ac9ac84d70f614dfa18b2483d81 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 7 Sep 2026 19:55:56 -0400 Subject: [PATCH 05/17] docs(dgw): clarify agent routing result --- crates/agent-tunnel/src/routing.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/agent-tunnel/src/routing.rs b/crates/agent-tunnel/src/routing.rs index 80752ef16..fdad13ab3 100644 --- a/crates/agent-tunnel/src/routing.rs +++ b/crates/agent-tunnel/src/routing.rs @@ -86,9 +86,10 @@ pub async fn resolve_route( /// Attempt to route a connection via the agent tunnel. /// -/// Returns `Ok(Some(stream))` if routed through an agent, `Ok(None)` if the caller -/// should fall through to direct connect, or `Err` if an explicit agent was specified -/// but not found (or all candidates failed). +/// Returns `Ok(Some((stream, agent, target_addr)))` when routed through an agent. +/// `target_addr` is `None` when the agent sends the legacy success response. +/// Returns `Ok(None)` when the caller should use a direct connection. +/// Returns `Err` when the requested agent is unavailable or all matching agents fail. pub async fn try_route( handle: Option<&AgentTunnelHandle>, explicit_agent_id: Option, From 65722608ff805f647ce0ad69deca6106c105f466 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Tue, 8 Sep 2026 10:41:32 -0400 Subject: [PATCH 06/17] fix(dgw,agent,jetsocat): simplify JMUX routing Keep Agent Tunnel ConnectResponse unchanged and pass only the routed stream into JMUX. Keep connector errors channel-local, reclaim failed IDs, and half-close type-erased streams when EOF arrives. --- crates/agent-tunnel-proto/src/session.rs | 24 +--- .../agent-tunnel-proto/src/session_codec.rs | 25 +--- crates/agent-tunnel-proto/tests/session.rs | 29 ----- crates/agent-tunnel/src/listener.rs | 11 +- crates/agent-tunnel/src/routing.rs | 20 ++- crates/jmux-proxy/src/lib.rs | 117 +++++++----------- crates/jmux-proxy/tests/target_connector.rs | 30 +---- devolutions-agent/src/tunnel.rs | 2 +- devolutions-gateway/src/jmux.rs | 6 +- devolutions-gateway/src/kdc_connector.rs | 2 +- devolutions-gateway/src/upstream.rs | 10 +- 11 files changed, 76 insertions(+), 200 deletions(-) diff --git a/crates/agent-tunnel-proto/src/session.rs b/crates/agent-tunnel-proto/src/session.rs index 8e2e563f1..4a3f153e5 100644 --- a/crates/agent-tunnel-proto/src/session.rs +++ b/crates/agent-tunnel-proto/src/session.rs @@ -2,8 +2,6 @@ //! //! Encoding and decoding live in [`crate::session_codec`]. -use std::net::SocketAddr; - use uuid::Uuid; use crate::version::CURRENT_PROTOCOL_VERSION; @@ -35,18 +33,12 @@ pub enum ConnectRequest { /// Agent's response to a [`ConnectRequest`]. /// /// Wire layout: -/// - Success: `[1B tag=0x00][2B version][optional 4B address length + address bytes]` +/// - Success: `[1B tag=0x00][2B version]` /// - Error: `[1B tag=0x01][2B version][4B reason_len][reason bytes]` #[derive(Debug, Clone, PartialEq, Eq)] pub enum ConnectResponse { - Success { - protocol_version: u16, - target_addr: Option, - }, - Error { - protocol_version: u16, - reason: String, - }, + Success { protocol_version: u16 }, + Error { protocol_version: u16, reason: String }, } impl ConnectRequest { @@ -85,14 +77,6 @@ impl ConnectResponse { pub fn success() -> Self { Self::Success { protocol_version: CURRENT_PROTOCOL_VERSION, - target_addr: None, - } - } - - pub fn success_with_target(target_addr: SocketAddr) -> Self { - Self::Success { - protocol_version: CURRENT_PROTOCOL_VERSION, - target_addr: Some(target_addr), } } @@ -110,7 +94,7 @@ impl ConnectResponse { /// Extract the protocol version from any variant. pub fn protocol_version(&self) -> u16 { match self { - Self::Success { protocol_version, .. } | Self::Error { protocol_version, .. } => *protocol_version, + Self::Success { protocol_version } | Self::Error { protocol_version, .. } => *protocol_version, } } } diff --git a/crates/agent-tunnel-proto/src/session_codec.rs b/crates/agent-tunnel-proto/src/session_codec.rs index 3f729cdea..725d8b1ca 100644 --- a/crates/agent-tunnel-proto/src/session_codec.rs +++ b/crates/agent-tunnel-proto/src/session_codec.rs @@ -57,15 +57,9 @@ impl Decode for ConnectRequest { impl Encode for ConnectResponse { fn encode(&self, buf: &mut BytesMut) { match self { - Self::Success { - protocol_version, - target_addr, - } => { + Self::Success { protocol_version } => { buf.put_u8(TAG_RESPONSE_SUCCESS); buf.put_u16(*protocol_version); - if let Some(target_addr) = target_addr { - codec::put_string(buf, &target_addr.to_string()); - } } Self::Error { protocol_version, @@ -86,22 +80,7 @@ impl Decode for ConnectResponse { let protocol_version = buf.get_u16(); match tag { - TAG_RESPONSE_SUCCESS => { - let target_addr = if buf.has_remaining() { - let value = codec::get_string(&mut buf)?; - Some(value.parse().map_err(|_| ProtoError::InvalidField { - field: "target_addr", - reason: "not a socket address", - })?) - } else { - None - }; - - Ok(Self::Success { - protocol_version, - target_addr, - }) - } + TAG_RESPONSE_SUCCESS => Ok(Self::Success { protocol_version }), TAG_RESPONSE_ERROR => { let reason = codec::get_string(&mut buf)?; Ok(Self::Error { diff --git a/crates/agent-tunnel-proto/tests/session.rs b/crates/agent-tunnel-proto/tests/session.rs index 43816a2c9..c8c4168e6 100644 --- a/crates/agent-tunnel-proto/tests/session.rs +++ b/crates/agent-tunnel-proto/tests/session.rs @@ -1,5 +1,3 @@ -use std::net::{Ipv4Addr, SocketAddr}; - use agent_tunnel_proto::{ConnectRequest, ConnectResponse, MAX_SESSION_MESSAGE_SIZE, ProtoError, SessionStream}; use uuid::Uuid; @@ -29,20 +27,6 @@ async fn roundtrip_connect_response_success() { assert_eq!(msg, decoded); } -#[tokio::test] -async fn roundtrip_connect_response_success_with_target() { - let target_addr = SocketAddr::from((Ipv4Addr::new(192, 0, 2, 10), 3389)); - let msg = ConnectResponse::success_with_target(target_addr); - - let mut buf = Vec::new(); - let mut stream = SessionStream::new(&mut buf, &[][..]); - stream.send_response(&msg).await.expect("send should succeed"); - - let mut stream = SessionStream::new(tokio::io::sink(), buf.as_slice()); - let decoded = stream.recv_response().await.expect("recv should succeed"); - assert_eq!(msg, decoded); -} - #[tokio::test] async fn roundtrip_connect_response_error() { let msg = ConnectResponse::error("connection refused"); @@ -125,19 +109,6 @@ async fn decode_rejects_unknown_connect_response_tag() { assert!(matches!(err, ProtoError::UnknownTag { tag: 0xFF }), "got {err:?}"); } -#[tokio::test] -async fn decode_rejects_invalid_success_target_address() { - let payload = &[0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x03, b'b', b'a', b'd']; - let err = recv_response_payload(payload).await; - assert!(matches!( - err, - ProtoError::InvalidField { - field: "target_addr", - .. - } - )); -} - #[tokio::test] async fn decode_rejects_unknown_connect_request_tag() { // Minimum bytes for header-length check (1 tag + 2 version + 16 uuid = 19), diff --git a/crates/agent-tunnel/src/listener.rs b/crates/agent-tunnel/src/listener.rs index a0708c824..1e548c05c 100644 --- a/crates/agent-tunnel/src/listener.rs +++ b/crates/agent-tunnel/src/listener.rs @@ -91,7 +91,7 @@ impl AgentTunnelHandle { agent_id: Uuid, session_id: Uuid, target: &str, - ) -> anyhow::Result<(TunnelStream, Option)> { + ) -> anyhow::Result { let conn = self .agent_connections .read() @@ -122,10 +122,9 @@ impl AgentTunnelHandle { agent_tunnel_proto::validate_protocol_version(response.protocol_version()) .map_err(|e| anyhow::anyhow!("ConnectResponse: {e}"))?; - let target_addr = match response { - ConnectResponse::Success { target_addr, .. } => target_addr, - ConnectResponse::Error { reason, .. } => anyhow::bail!("agent refused connection: {reason}"), - }; + if let ConnectResponse::Error { reason, .. } = &response { + anyhow::bail!("agent refused connection: {reason}"); + } info!( %agent_id, @@ -135,7 +134,7 @@ impl AgentTunnelHandle { ); let (send, recv) = session.into_inner(); - Ok((TunnelStream { send, recv }, target_addr)) + Ok(TunnelStream { send, recv }) } } diff --git a/crates/agent-tunnel/src/routing.rs b/crates/agent-tunnel/src/routing.rs index fdad13ab3..b6d9a967a 100644 --- a/crates/agent-tunnel/src/routing.rs +++ b/crates/agent-tunnel/src/routing.rs @@ -3,7 +3,7 @@ //! Consumed by the upstream connection paths (forwarding, RDP clean path, //! generic client) to ensure consistent routing behavior and error messages. -use std::net::{IpAddr, SocketAddr}; +use std::net::IpAddr; use std::sync::Arc; use agent_tunnel_proto::DomainName; @@ -86,17 +86,16 @@ pub async fn resolve_route( /// Attempt to route a connection via the agent tunnel. /// -/// Returns `Ok(Some((stream, agent, target_addr)))` when routed through an agent. -/// `target_addr` is `None` when the agent sends the legacy success response. -/// Returns `Ok(None)` when the caller should use a direct connection. -/// Returns `Err` when the requested agent is unavailable or all matching agents fail. +/// Returns `Ok(Some(stream))` if routed through an agent, `Ok(None)` if the caller +/// should fall through to direct connect, or `Err` if an explicit agent was specified +/// but not found (or all candidates failed). pub async fn try_route( handle: Option<&AgentTunnelHandle>, explicit_agent_id: Option, target: &RouteTarget, session_id: Uuid, target_addr: &str, -) -> Result, Option)>> { +) -> Result)>> { let Some(handle) = handle else { // An explicit `jet_agent_id` claim means the token requires routing via that // specific agent; silently falling back to a direct connect would bypass the @@ -123,8 +122,7 @@ pub async fn try_route( /// Try connecting to target through agent candidates (try-fail-retry). /// -/// Returns the connected stream, the agent that succeeded, and its reported target address. -/// The address is `None` when the agent sends the legacy success response. +/// Returns the connected `TunnelStream` and the agent that succeeded. /// /// Callers must handle `RoutingDecision::ExplicitAgentNotFound` and /// `RoutingDecision::Direct` before calling this function. @@ -133,7 +131,7 @@ pub async fn route_and_connect( candidates: &[Arc], session_id: Uuid, target: &str, -) -> Result<(TunnelStream, Arc, Option)> { +) -> Result<(TunnelStream, Arc)> { if candidates.is_empty() { return Err(anyhow!("route_and_connect called with empty candidates")); } @@ -149,14 +147,14 @@ pub async fn route_and_connect( ); match handle.connect_via_agent(agent.agent_id, session_id, target).await { - Ok((stream, target_addr)) => { + Ok(stream) => { info!( agent_id = %agent.agent_id, agent_name = %agent.name, %target, "Agent tunnel connection established" ); - return Ok((stream, Arc::clone(agent), target_addr)); + return Ok((stream, Arc::clone(agent))); } Err(error) => { warn!( diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index 9fdbdcccd..acf3dbf8f 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -65,17 +65,12 @@ type TargetConnector = Arc TargetConnectorFuture + Sen pub struct ConnectedTarget { stream: ErasedTargetStream, - target_ip: Option, } impl ConnectedTarget { - /// Wraps a connected stream and the concrete peer IP used for traffic auditing. - /// - /// When `target_ip` is `None`, the channel works normally but emits no traffic event. - pub fn new(stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static, target_ip: Option) -> Self { + pub fn new(stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static) -> Self { Self { stream: Box::new(stream), - target_ip, } } } @@ -145,8 +140,6 @@ impl JmuxProxy { /// Tries a custom target connection before falling back to direct TCP. /// /// Return `Ok(None)` when the target should use the default direct connection. - /// Return the concrete peer IP in `ConnectedTarget` to enable traffic auditing. - /// Connector errors do not emit `ConnectFailure` because they do not prove that a concrete address was attempted. #[must_use] pub fn with_target_connector(mut self, connector: C) -> Self where @@ -400,11 +393,9 @@ enum InternalMessage { channel: Box, stream: ErasedTargetStream, }, - StreamResolutionFailed { + TargetConnectionFailed { id: LocalChannelId, distant_id: DistantChannelId, - reason_code: ReasonCode, - description: String, }, AbnormalTermination { id: LocalChannelId, @@ -706,15 +697,14 @@ async fn scheduler_task_impl(task: JmuxSc .spawn(channel_span) .detach(); } - InternalMessage::StreamResolutionFailed { - id, - distant_id, - reason_code, - description, - } => { + InternalMessage::TargetConnectionFailed { id, distant_id } => { jmux_ctx.id_allocator.free(id); msg_to_send_tx - .send(Message::open_failure(distant_id, reason_code, description)) + .send(Message::open_failure( + distant_id, + ReasonCode::GENERAL_FAILURE, + "target connection failed", + )) .await .context("couldn't send OPEN FAILURE message through mpsc channel")?; } @@ -824,6 +814,7 @@ async fn scheduler_task_impl(task: JmuxSc channel, destination_url: msg.destination_url, internal_msg_tx: internal_msg_tx.clone(), + msg_to_send_tx: msg_to_send_tx.clone(), traffic_callback: traffic_callback.clone(), target_connector: target_connector.clone(), } @@ -1189,6 +1180,8 @@ impl DataWriterTask { bytes_tx.fetch_add(data.len() as u64, Ordering::SeqCst); } + + let _ = writer.shutdown().await; } .instrument(span), ); @@ -1203,6 +1196,7 @@ struct StreamResolverTask { channel: JmuxChannelCtx, destination_url: DestinationUrl, internal_msg_tx: InternalMessageSender, + msg_to_send_tx: MessageSender, traffic_callback: Option, target_connector: Option, } @@ -1223,28 +1217,12 @@ impl StreamResolverTask { ChildTask(handle) } - async fn report_failure( - internal_msg_tx: &InternalMessageSender, - channel: &JmuxChannelCtx, - reason_code: ReasonCode, - description: String, - ) -> anyhow::Result<()> { - internal_msg_tx - .send(InternalMessage::StreamResolutionFailed { - id: channel.local_id, - distant_id: channel.distant_id, - reason_code, - description, - }) - .await - .map_err(|_| anyhow::anyhow!("couldn't report stream resolution failure through internal mpsc channel")) - } - async fn run(self) -> anyhow::Result<()> { let Self { mut channel, destination_url, internal_msg_tx, + msg_to_send_tx, traffic_callback, target_connector, } = self; @@ -1257,8 +1235,7 @@ impl StreamResolverTask { "tcp" => { if let Some(connector) = target_connector { match connector(destination_url.clone()).await { - Ok(Some(ConnectedTarget { stream, target_ip })) => { - channel.target_ip = target_ip; + Ok(Some(ConnectedTarget { stream })) => { channel.connect_at = SystemTime::now(); internal_msg_tx @@ -1275,13 +1252,17 @@ impl StreamResolverTask { } Ok(None) => {} Err(error) => { - Self::report_failure( - &internal_msg_tx, - &channel, - ReasonCode::GENERAL_FAILURE, - "target connection failed".to_owned(), - ) - .await?; + internal_msg_tx + .send(InternalMessage::TargetConnectionFailed { + id: channel.local_id, + distant_id: channel.distant_id, + }) + .await + .map_err(|_| { + anyhow::anyhow!( + "couldn't report target connection failure through internal mpsc channel" + ) + })?; return Err(error.context(format!("couldn't connect to {host}:{port}"))); } @@ -1294,13 +1275,14 @@ impl StreamResolverTask { Err(error) => { debug!(?error, "DNS resolution failed"); // No event emission for DNS failures - cannot determine target IP. - Self::report_failure( - &internal_msg_tx, - &channel, - ReasonCode::from(error.kind()), - error.to_string(), - ) - .await?; + msg_to_send_tx + .send(Message::open_failure( + channel.distant_id, + ReasonCode::from(error.kind()), + error.to_string(), + )) + .await + .context("couldn't send OPEN FAILURE message through mpsc channel")?; anyhow::bail!("couldn't resolve {host}:{port}: {error}"); } }; @@ -1354,36 +1336,21 @@ impl StreamResolverTask { }); } - Self::report_failure( - &internal_msg_tx, - &channel, - ReasonCode::from(error.kind()), - error.to_string(), - ) - .await?; + msg_to_send_tx + .send(Message::open_failure( + channel.distant_id, + ReasonCode::from(error.kind()), + error.to_string(), + )) + .await + .context("couldn't send OPEN FAILURE message through mpsc channel")?; anyhow::bail!("couldn't open TCP stream to {host}:{port}: {error}"); } else { - Self::report_failure( - &internal_msg_tx, - &channel, - ReasonCode::GENERAL_FAILURE, - "no addresses resolved".to_owned(), - ) - .await?; anyhow::bail!("no addresses resolved for {host}:{port}"); } } - _ => { - Self::report_failure( - &internal_msg_tx, - &channel, - ReasonCode::GENERAL_FAILURE, - format!("unsupported scheme: {scheme}"), - ) - .await?; - anyhow::bail!("unsupported scheme: {scheme}") - } + _ => anyhow::bail!("unsupported scheme: {scheme}"), } } } diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs index 1741fe8a7..885aebd68 100644 --- a/crates/jmux-proxy/tests/target_connector.rs +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -1,13 +1,11 @@ #![allow(unused_crate_dependencies)] #![allow(clippy::unwrap_used)] -use std::net::{IpAddr, Ipv4Addr}; use std::time::Duration; use jmux_proto::{BytesMut, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; -use jmux_proxy::{ConnectedTarget, DestinationUrl, EventOutcome, JmuxConfig, JmuxProxy, TrafficEvent}; +use jmux_proxy::{ConnectedTarget, DestinationUrl, JmuxConfig, JmuxProxy}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio::sync::mpsc; use tokio::time::timeout; const TEST_TIMEOUT: Duration = Duration::from_secs(5); @@ -36,13 +34,10 @@ async fn receive_message(reader: &mut (impl AsyncRead + Unpin)) -> Message { } #[tokio::test] -async fn connected_target_ip_is_used_for_audit() { +async fn connector_success_opens_channel() { let (proxy_stream, peer_stream) = tokio::io::duplex(8192); let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); - let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); - let target_ip = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 10)); - let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) .with_config(JmuxConfig::permissive()) .with_target_connector(move |destination| async move { @@ -51,10 +46,7 @@ async fn connected_target_ip_is_used_for_audit() { tokio::spawn(async move { target_peer.shutdown().await.expect("close target stream"); }); - Ok(Some(ConnectedTarget::new(target_stream, Some(target_ip)))) - }) - .with_outgoing_traffic_event_callback(move |event| { - event_tx.send(event).expect("capture traffic event"); + Ok(Some(ConnectedTarget::new(target_stream))) }); let proxy_task = tokio::spawn(proxy.run()); @@ -78,15 +70,6 @@ async fn connected_target_ip_is_used_for_audit() { assert!(matches!(receive_message(&mut peer_reader).await, Message::Close(_))); send_message(&mut peer_writer, Message::close(local_id)).await; - let event = timeout(TEST_TIMEOUT, event_rx.recv()) - .await - .expect("traffic event timed out") - .expect("traffic event channel closed"); - assert_eq!(event.outcome, EventOutcome::NormalTermination); - assert_eq!(event.target_host, "agent.example"); - assert_eq!(event.target_ip, target_ip); - assert_eq!(event.target_port, 443); - proxy_task.abort(); } @@ -95,8 +78,6 @@ async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { let (proxy_stream, peer_stream) = tokio::io::duplex(8192); let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); - let (event_tx, mut event_rx) = mpsc::unbounded_channel::(); - let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) .with_config(JmuxConfig::permissive()) .with_target_connector(|destination| async move { @@ -104,9 +85,6 @@ async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { anyhow::bail!("{}", "agent error ".repeat(8192)); } Ok(None) - }) - .with_outgoing_traffic_event_callback(move |event| { - event_tx.send(event).expect("capture traffic event"); }); let proxy_task = tokio::spawn(proxy.run()); @@ -125,8 +103,6 @@ async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { }; assert_eq!(open_failure.reason_code, ReasonCode::GENERAL_FAILURE); assert_eq!(open_failure.description, "target connection failed"); - assert!(timeout(Duration::from_millis(100), event_rx.recv()).await.is_err()); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind direct target"); diff --git a/devolutions-agent/src/tunnel.rs b/devolutions-agent/src/tunnel.rs index f261adfa7..e19c69d65 100644 --- a/devolutions-agent/src/tunnel.rs +++ b/devolutions-agent/src/tunnel.rs @@ -778,7 +778,7 @@ async fn run_session_proxy( info!(target = %selected_target, "TCP connection established"); session - .send_response(&ConnectResponse::success_with_target(selected_target)) + .send_response(&ConnectResponse::success()) .await .context("send ConnectResponse")?; info!("Sent ConnectResponse::Success"); diff --git a/devolutions-gateway/src/jmux.rs b/devolutions-gateway/src/jmux.rs index 174094ee3..cbf6f3dec 100644 --- a/devolutions-gateway/src/jmux.rs +++ b/devolutions-gateway/src/jmux.rs @@ -135,13 +135,11 @@ pub async fn handle( ) .await?; - let Some((stream, _agent, target_addr)) = routed else { + let Some((stream, _agent)) = routed else { return Ok(None); }; - let target_ip = target_addr.map(|addr| addr.ip()).or_else(|| target.host_ip()); - - Ok(Some(ConnectedTarget::new(stream, target_ip))) + Ok(Some(ConnectedTarget::new(stream))) } }); } diff --git a/devolutions-gateway/src/kdc_connector.rs b/devolutions-gateway/src/kdc_connector.rs index bf14aa078..f1840b643 100644 --- a/devolutions-gateway/src/kdc_connector.rs +++ b/devolutions-gateway/src/kdc_connector.rs @@ -81,7 +81,7 @@ impl KdcConnector { .await .map_err(|e| HttpError::bad_gateway().build(format!("KDC routing through agent tunnel failed: {e:#}")))?; - if let Some((mut stream, _, _)) = route_result { + if let Some((mut stream, _)) = route_result { // The agent tunnel currently carries only TCP (`ConnectRequest::tcp`). If the // routing pipeline picked an agent for a udp:// KDC target — either by subnet // match or by explicit pin — we must reject explicitly. Silently falling diff --git a/devolutions-gateway/src/upstream.rs b/devolutions-gateway/src/upstream.rs index 3b38069da..c6f5d9ad0 100644 --- a/devolutions-gateway/src/upstream.rs +++ b/devolutions-gateway/src/upstream.rs @@ -264,11 +264,15 @@ impl<'a> RoutePlan<'a> { .connect_via_agent(agent.agent_id, session_id, target.as_addr()) .await { - Ok((stream, target_addr)) => { - let server_addr = target_addr.unwrap_or_else(|| match target.host_ip() { + Ok(stream) => { + // The TCP peer lives on the agent side; surface the target + // IP:port for logs/PCAP when the target is a literal IP, or + // 0.0.0.0: when it's a hostname the gateway never + // resolved itself. Either is more useful than 0.0.0.0:0. + let server_addr = match target.host_ip() { Some(ip) => SocketAddr::new(ip, target.port()), None => SocketAddr::from((std::net::Ipv4Addr::UNSPECIFIED, target.port())), - }); + }; return Ok(ConnectedUpstream { leg: UpstreamLeg::Tunnel(stream), From 201f03c9d6018c4b7ea52b0f87b08249c52cbed5 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Tue, 8 Sep 2026 13:30:20 -0400 Subject: [PATCH 07/17] fix(agent): preserve tunneled TCP half-close Preserve half-close semantics when JMUX traffic crosses an Agent tunnel. Tokio's bidirectional copy keeps the other direction open after EOF. --- devolutions-agent/src/tunnel.rs | 20 +++++--------------- 1 file changed, 5 insertions(+), 15 deletions(-) diff --git a/devolutions-agent/src/tunnel.rs b/devolutions-agent/src/tunnel.rs index e19c69d65..8ca7e7980 100644 --- a/devolutions-agent/src/tunnel.rs +++ b/devolutions-agent/src/tunnel.rs @@ -756,7 +756,7 @@ async fn run_session_proxy( // Whatever went wrong has to travel back as a ConnectResponse::Error — returning // early instead drops the stream and the Gateway just sees an unexplained EOF. - let (tcp_stream, selected_target) = match connect_result { + let (mut tcp_stream, selected_target) = match connect_result { Ok(connected) => connected, Err(error) => { let reason = format!("{error:#}"); @@ -783,20 +783,10 @@ async fn run_session_proxy( .context("send ConnectResponse")?; info!("Sent ConnectResponse::Success"); - let (mut send, mut recv) = session.into_inner(); - let (mut tcp_read, mut tcp_write) = tcp_stream.into_split(); - - // Use join! (not select!) to wait for BOTH directions to finish. - // select! would cancel in-flight data when one direction closes first. - let (r1, r2) = tokio::join!( - tokio::io::copy(&mut recv, &mut tcp_write), - tokio::io::copy(&mut tcp_read, &mut send), - ); - r1.inspect_err(|e| debug!(%e, "QUIC->TCP copy ended"))?; - r2.inspect_err(|e| debug!(%e, "TCP->QUIC copy ended"))?; - - // Gracefully finish the QUIC send stream (signals EOF to peer). - let _ = send.finish(); + let (send, recv) = session.into_inner(); + tokio::io::copy_bidirectional(&mut tokio::io::join(recv, send), &mut tcp_stream) + .await + .context("proxy session traffic")?; Ok(()) } From 9bfef41fd766b7c745bb7d7014c97a1dc6e81bd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 12:18:45 -0400 Subject: [PATCH 08/17] fix(dgw,agent,jetsocat): harden JMUX routing Centralize stream resolution outcomes so every failed path releases its reserved channel ID without bypassing a selected Agent route. Validate Agent Tunnel configuration before startup and preserve independent draining and half-close propagation in both relay directions. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/jmux-proxy/src/lib.rs | 304 ++++++++++---------- crates/jmux-proxy/tests/target_connector.rs | 67 +++-- devolutions-agent/src/config.rs | 249 ++++++++++++++-- devolutions-agent/src/tunnel.rs | 86 +++--- devolutions-gateway/src/jmux.rs | 6 +- 5 files changed, 456 insertions(+), 256 deletions(-) diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index acf3dbf8f..54f6594b8 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -55,25 +55,14 @@ pub type ApiResponseReceiver = oneshot::Receiver; pub type ApiRequestSender = mpsc::Sender; pub type ApiRequestReceiver = mpsc::Receiver; +// A supertrait is required because trait objects may include only one non-auto trait. trait TargetStream: AsyncRead + AsyncWrite + Unpin + Send {} impl TargetStream for T where T: AsyncRead + AsyncWrite + Unpin + Send {} type ErasedTargetStream = Box; -type TargetConnectorFuture = Pin>> + Send>>; -type TargetConnector = Arc TargetConnectorFuture + Send + Sync>; - -pub struct ConnectedTarget { - stream: ErasedTargetStream, -} - -impl ConnectedTarget { - pub fn new(stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static) -> Self { - Self { - stream: Box::new(stream), - } - } -} +type TargetConnectorOverrideFuture = Pin>> + Send>>; +type TargetConnectorOverride = Arc TargetConnectorOverrideFuture + Send + Sync>; #[derive(Debug)] pub enum JmuxApiRequest { @@ -106,7 +95,7 @@ pub struct JmuxProxy { jmux_reader: Box, jmux_writer: Box, traffic_callback: Option, - target_connector: Option, + target_connector_override: Option, } impl JmuxProxy { @@ -121,7 +110,7 @@ impl JmuxProxy { jmux_reader, jmux_writer, traffic_callback: None, - target_connector: None, + target_connector_override: None, } } @@ -137,16 +126,27 @@ impl JmuxProxy { self } - /// Tries a custom target connection before falling back to direct TCP. + /// Overrides the default direct TCP connector when applicable. /// - /// Return `Ok(None)` when the target should use the default direct connection. + /// Return `Ok(Some(stream))` to use the override, `Ok(None)` to delegate to the default connector, + /// or an error to reject the connection without falling back. + /// Overridden streams do not emit outgoing traffic events because their resolved target IP is unknown. #[must_use] - pub fn with_target_connector(mut self, connector: C) -> Self + pub fn with_target_connector_override(mut self, connector: C) -> Self where C: Fn(DestinationUrl) -> F + Send + Sync + 'static, - F: Future>> + Send + 'static, + F: Future>> + Send + 'static, + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { - self.target_connector = Some(Arc::new(move |destination_url| Box::pin(connector(destination_url)))); + self.target_connector_override = Some(Arc::new(move |destination_url| { + let connect = connector(destination_url); + + Box::pin(async move { + connect + .await + .map(|stream| stream.map(|stream| Box::new(stream) as ErasedTargetStream)) + }) + })); self } @@ -223,7 +223,7 @@ async fn run_proxy_impl(proxy: JmuxProxy, span: Span) -> anyhow::Result<()> { jmux_reader, jmux_writer, traffic_callback, - target_connector, + target_connector_override, } = proxy; let (msg_to_send_tx, msg_to_send_rx) = mpsc::channel::(JMUX_MESSAGE_MPSC_CHANNEL_SIZE); @@ -244,7 +244,7 @@ async fn run_proxy_impl(proxy: JmuxProxy, span: Span) -> anyhow::Result<()> { msg_to_send_tx, api_request_rx, traffic_callback, - target_connector, + target_connector_override, parent_span: span, } .spawn(); @@ -342,7 +342,7 @@ impl JmuxCtx { fn unregister(&mut self, id: LocalChannelId, traffic_callback: &Option, is_abnormal_error: bool) { if let Some(channel) = self.channels.remove(&id) { // Emit audit event if we have a callback and haven't already emitted. - // For now, we only emit an event when the IP address is known = on the "server side". + // Streams without a known target IP do not emit an event. if let Some(callback) = traffic_callback && let Some(target_ip) = channel.target_ip && !channel.audit_emitted.swap(true, Ordering::SeqCst) @@ -393,9 +393,11 @@ enum InternalMessage { channel: Box, stream: ErasedTargetStream, }, - TargetConnectionFailed { + StreamResolutionFailed { id: LocalChannelId, distant_id: DistantChannelId, + reason_code: ReasonCode, + description: String, }, AbnormalTermination { id: LocalChannelId, @@ -469,7 +471,7 @@ struct JmuxSchedulerTask { msg_to_send_tx: MessageSender, api_request_rx: ApiRequestReceiver, traffic_callback: Option, - target_connector: Option, + target_connector_override: Option, parent_span: Span, } @@ -491,7 +493,7 @@ async fn scheduler_task_impl(task: JmuxSc msg_to_send_tx, mut api_request_rx, traffic_callback, - target_connector, + target_connector_override, parent_span, } = task; @@ -697,14 +699,15 @@ async fn scheduler_task_impl(task: JmuxSc .spawn(channel_span) .detach(); } - InternalMessage::TargetConnectionFailed { id, distant_id } => { + InternalMessage::StreamResolutionFailed { + id, + distant_id, + reason_code, + description, + } => { jmux_ctx.id_allocator.free(id); msg_to_send_tx - .send(Message::open_failure( - distant_id, - ReasonCode::GENERAL_FAILURE, - "target connection failed", - )) + .send(Message::open_failure(distant_id, reason_code, description)) .await .context("couldn't send OPEN FAILURE message through mpsc channel")?; } @@ -814,9 +817,8 @@ async fn scheduler_task_impl(task: JmuxSc channel, destination_url: msg.destination_url, internal_msg_tx: internal_msg_tx.clone(), - msg_to_send_tx: msg_to_send_tx.clone(), traffic_callback: traffic_callback.clone(), - target_connector: target_connector.clone(), + target_connector_override: target_connector_override.clone(), } .spawn() .detach(); @@ -1196,9 +1198,14 @@ struct StreamResolverTask { channel: JmuxChannelCtx, destination_url: DestinationUrl, internal_msg_tx: InternalMessageSender, - msg_to_send_tx: MessageSender, traffic_callback: Option, - target_connector: Option, + target_connector_override: Option, +} + +struct StreamResolutionFailure { + error: anyhow::Error, + reason_code: ReasonCode, + description: String, } impl StreamResolverTask { @@ -1222,135 +1229,130 @@ impl StreamResolverTask { mut channel, destination_url, internal_msg_tx, - msg_to_send_tx, traffic_callback, - target_connector, + target_connector_override, } = self; let scheme = destination_url.scheme(); let host = destination_url.host(); let port = destination_url.port(); - match scheme { - "tcp" => { - if let Some(connector) = target_connector { - match connector(destination_url.clone()).await { - Ok(Some(ConnectedTarget { stream })) => { - channel.connect_at = SystemTime::now(); - - internal_msg_tx - .send(InternalMessage::StreamResolved { - channel: Box::new(channel), - stream, - }) - .await - .map_err(|_| { - anyhow::anyhow!("couldn't send back resolved stream through internal mpsc channel") - })?; - - return Ok(()); - } - Ok(None) => {} - Err(error) => { - internal_msg_tx - .send(InternalMessage::TargetConnectionFailed { - id: channel.local_id, - distant_id: channel.distant_id, - }) - .await - .map_err(|_| { - anyhow::anyhow!( - "couldn't report target connection failure through internal mpsc channel" - ) - })?; + let resolution = if scheme != "tcp" { + let description = format!("unsupported scheme: {scheme}"); + Err(StreamResolutionFailure { + error: anyhow::anyhow!(description.clone()), + reason_code: ReasonCode::GENERAL_FAILURE, + description, + }) + } else if let Some(connector) = target_connector_override { + match connector(destination_url.clone()).await { + Ok(Some(stream)) => Ok(stream), + Ok(None) => Self::connect_direct(host, port, &mut channel, traffic_callback.as_ref()).await, + Err(error) => Err(StreamResolutionFailure { + error: error.context(format!("couldn't connect to {host}:{port}")), + reason_code: ReasonCode::GENERAL_FAILURE, + description: "target connection failed".to_owned(), + }), + } + } else { + Self::connect_direct(host, port, &mut channel, traffic_callback.as_ref()).await + }; - return Err(error.context(format!("couldn't connect to {host}:{port}"))); - } - } - } + match resolution { + Ok(stream) => { + channel.connect_at = SystemTime::now(); + internal_msg_tx + .send(InternalMessage::StreamResolved { + channel: Box::new(channel), + stream, + }) + .await + .ok() + .context("couldn't send back resolved stream through internal mpsc channel") + } + Err(StreamResolutionFailure { + error, + reason_code, + description, + }) => { + internal_msg_tx + .send(InternalMessage::StreamResolutionFailed { + id: channel.local_id, + distant_id: channel.distant_id, + reason_code, + description, + }) + .await + .ok() + .context("couldn't report stream resolution failure through internal mpsc channel")?; + Err(error) + } + } + } - // Perform DNS resolution first to get concrete IP addresses. - let socket_addrs = match tokio::net::lookup_host((host, port)).await { - Ok(addrs) => addrs, - Err(error) => { - debug!(?error, "DNS resolution failed"); - // No event emission for DNS failures - cannot determine target IP. - msg_to_send_tx - .send(Message::open_failure( - channel.distant_id, - ReasonCode::from(error.kind()), - error.to_string(), - )) - .await - .context("couldn't send OPEN FAILURE message through mpsc channel")?; - anyhow::bail!("couldn't resolve {host}:{port}: {error}"); - } - }; + async fn connect_direct( + host: &str, + port: u16, + channel: &mut JmuxChannelCtx, + traffic_callback: Option<&TrafficCallback>, + ) -> Result { + let socket_addrs = match tokio::net::lookup_host((host, port)).await { + Ok(addrs) => addrs, + Err(error) => { + debug!(?error, "DNS resolution failed"); + // No event emission for DNS failures - cannot determine target IP. + return Err(StreamResolutionFailure { + reason_code: ReasonCode::from(error.kind()), + description: error.to_string(), + error: anyhow::Error::new(error).context(format!("couldn't resolve {host}:{port}")), + }); + } + }; - // Try connecting to each resolved address (Happy Eyeballs style). - let mut last_error = None; - - for socket_addr in socket_addrs { - match TcpStream::connect(socket_addr).await { - Ok(stream) => { - // Update channel with resolved target IP and connect time. - channel.target_ip = Some(socket_addr.ip()); - channel.connect_at = SystemTime::now(); - - internal_msg_tx - .send(InternalMessage::StreamResolved { - channel: Box::new(channel), - stream: Box::new(stream), - }) - .await - .map_err(|_| { - anyhow::anyhow!("couldn't send back resolved stream through internal mpsc channel") - })?; + let mut last_error = None; - return Ok(()); - } - Err(error) => { - debug!(?error, ?socket_addr, "TcpStream::connect failed"); - last_error = Some((socket_addr, error)); - } - } + for socket_addr in socket_addrs { + match TcpStream::connect(socket_addr).await { + Ok(stream) => { + channel.target_ip = Some(socket_addr.ip()); + return Ok(Box::new(stream)); } + Err(error) => { + debug!(?error, ?socket_addr, "TcpStream::connect failed"); + last_error = Some((socket_addr, error)); + } + } + } - // All connection attempts failed - emit ConnectFailure for the last attempted address. - if let Some((failed_addr, error)) = last_error { - // Emit ConnectFailure event - we always have a concrete IP at this point. - if let Some(callback) = &traffic_callback { - let connect_and_disconnect_time = SystemTime::now(); - - callback(TrafficEvent { - outcome: EventOutcome::ConnectFailure, - protocol: TransportProtocol::Tcp, - target_host: channel.target_host.clone(), - target_ip: failed_addr.ip(), - target_port: failed_addr.port(), - connect_at: connect_and_disconnect_time, - disconnect_at: connect_and_disconnect_time, - active_duration: std::time::Duration::ZERO, - bytes_tx: 0, - bytes_rx: 0, - }); - } - - msg_to_send_tx - .send(Message::open_failure( - channel.distant_id, - ReasonCode::from(error.kind()), - error.to_string(), - )) - .await - .context("couldn't send OPEN FAILURE message through mpsc channel")?; + if let Some((failed_addr, error)) = last_error { + if let Some(callback) = traffic_callback { + let connect_and_disconnect_time = SystemTime::now(); - anyhow::bail!("couldn't open TCP stream to {host}:{port}: {error}"); - } else { - anyhow::bail!("no addresses resolved for {host}:{port}"); - } + callback(TrafficEvent { + outcome: EventOutcome::ConnectFailure, + protocol: TransportProtocol::Tcp, + target_host: channel.target_host.clone(), + target_ip: failed_addr.ip(), + target_port: failed_addr.port(), + connect_at: connect_and_disconnect_time, + disconnect_at: connect_and_disconnect_time, + active_duration: std::time::Duration::ZERO, + bytes_tx: 0, + bytes_rx: 0, + }); } - _ => anyhow::bail!("unsupported scheme: {scheme}"), + + Err(StreamResolutionFailure { + reason_code: ReasonCode::from(error.kind()), + description: error.to_string(), + error: anyhow::Error::new(error).context(format!("couldn't open TCP stream to {host}:{port}")), + }) + } else { + Err(StreamResolutionFailure { + error: anyhow::anyhow!("no addresses resolved for {host}:{port}"), + reason_code: ReasonCode::GENERAL_FAILURE, + description: "no addresses resolved".to_owned(), + }) } } } diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs index 885aebd68..20c05dc7d 100644 --- a/crates/jmux-proxy/tests/target_connector.rs +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -1,10 +1,7 @@ -#![allow(unused_crate_dependencies)] -#![allow(clippy::unwrap_used)] - use std::time::Duration; -use jmux_proto::{BytesMut, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; -use jmux_proxy::{ConnectedTarget, DestinationUrl, JmuxConfig, JmuxProxy}; +use jmux_proto::{Bytes, BytesMut, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; +use jmux_proxy::{DestinationUrl, JmuxConfig, JmuxProxy}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::time::timeout; @@ -34,21 +31,22 @@ async fn receive_message(reader: &mut (impl AsyncRead + Unpin)) -> Message { } #[tokio::test] -async fn connector_success_opens_channel() { +async fn override_stream_carries_channel_data() { let (proxy_stream, peer_stream) = tokio::io::duplex(8192); let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) .with_config(JmuxConfig::permissive()) - .with_target_connector(move |destination| async move { - assert_eq!(destination.host(), "agent.example"); + .with_target_connector_override(|_| async move { let (target_stream, mut target_peer) = tokio::io::duplex(64); tokio::spawn(async move { - target_peer.shutdown().await.expect("close target stream"); + let mut payload = [0; 4]; + target_peer.read_exact(&mut payload).await.expect("read target data"); + target_peer.write_all(&payload).await.expect("echo target data"); }); - Ok(Some(ConnectedTarget::new(target_stream))) + Ok(Some(target_stream)) }); - let proxy_task = tokio::spawn(proxy.run()); + let _proxy_task = tokio::spawn(proxy.run()); send_message( &mut peer_writer, @@ -65,28 +63,28 @@ async fn connector_success_opens_channel() { }; let local_id = DistantChannelId::from(open_success.sender_channel_id); - assert!(matches!(receive_message(&mut peer_reader).await, Message::Eof(_))); - send_message(&mut peer_writer, Message::eof(local_id)).await; - assert!(matches!(receive_message(&mut peer_reader).await, Message::Close(_))); - send_message(&mut peer_writer, Message::close(local_id)).await; - - proxy_task.abort(); + send_message(&mut peer_writer, Message::data(local_id, Bytes::from_static(b"ping"))).await; + let Message::Data(data) = receive_message(&mut peer_reader).await else { + panic!("expected CHANNEL DATA"); + }; + assert_eq!(data.recipient_channel_id, 7); + assert_eq!(data.transfer_data, b"ping"[..]); } #[tokio::test] -async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { +async fn resolution_failures_free_id_and_keep_direct_fallback() { let (proxy_stream, peer_stream) = tokio::io::duplex(8192); let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) .with_config(JmuxConfig::permissive()) - .with_target_connector(|destination| async move { + .with_target_connector_override(|destination| async move { if destination.host() == "fail.example" { - anyhow::bail!("{}", "agent error ".repeat(8192)); + anyhow::bail!("agent error"); } - Ok(None) + Ok(None::) }); - let proxy_task = tokio::spawn(proxy.run()); + let _proxy_task = tokio::spawn(proxy.run()); send_message( &mut peer_writer, @@ -103,19 +101,30 @@ async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { }; assert_eq!(open_failure.reason_code, ReasonCode::GENERAL_FAILURE); assert_eq!(open_failure.description, "target connection failed"); + + send_message( + &mut peer_writer, + Message::open( + LocalChannelId::from(12), + 4096, + DestinationUrl::new("tcp", "127.0.0.1", 0), + ), + ) + .await; + assert!(matches!( + receive_message(&mut peer_reader).await, + Message::OpenFailure(_) + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind direct target"); let target_port = listener.local_addr().expect("read direct target address").port(); - let server_task = tokio::spawn(async move { - let (_stream, _) = listener.accept().await.expect("accept direct connection"); - std::future::pending::<()>().await; - }); send_message( &mut peer_writer, Message::open( - LocalChannelId::from(12), + LocalChannelId::from(13), 4096, DestinationUrl::new("tcp", "127.0.0.1", target_port), ), @@ -124,8 +133,6 @@ async fn connector_failure_is_bounded_and_does_not_stop_direct_fallback() { let Message::OpenSuccess(open_success) = receive_message(&mut peer_reader).await else { panic!("expected OPEN SUCCESS"); }; + // The failed channel released ID 0 for reuse. assert_eq!(open_success.sender_channel_id, 0); - - server_task.abort(); - proxy_task.abort(); } diff --git a/devolutions-agent/src/config.rs b/devolutions-agent/src/config.rs index 81755248a..3c9d186bb 100644 --- a/devolutions-agent/src/config.rs +++ b/devolutions-agent/src/config.rs @@ -1,11 +1,14 @@ use std::fs::File; use std::io::BufReader; -use std::net::SocketAddr; +use std::net::{Ipv6Addr, SocketAddr}; +use std::num::NonZeroU16; use std::sync::Arc; +use agent_tunnel_proto::DomainName; use anyhow::{Context as _, bail}; use camino::{Utf8Path, Utf8PathBuf}; use devolutions_agent_shared::{default_schedule_window_start, get_data_dir}; +use ipnetwork::Ipv4Network; use serde::{Deserialize, Serialize}; use tap::prelude::*; use url::Url; @@ -27,19 +30,18 @@ pub struct Conf { pub debug: dto::DebugConf, } -/// Validated tunnel configuration — required fields are guaranteed present. +/// Validated tunnel configuration. /// -/// Constructed from `dto::TunnelConf` via `TryFrom`. If the tunnel is disabled -/// or not yet enrolled, the `enabled` field is `false` and path fields are empty -/// (but the struct is always constructible). +/// Required fields and field values are validated when this is constructed from [`dto::TunnelConf`]. +/// A disabled or unenrolled tunnel is represented by `enabled` being `false`. #[derive(Debug, Clone)] pub struct TunnelConf { pub enabled: bool, - pub gateway_endpoint: String, + gateway_endpoint: Option, pub client_cert_path: Utf8PathBuf, pub client_key_path: Utf8PathBuf, pub gateway_ca_cert_path: Utf8PathBuf, - pub advertise_subnets: Vec, + pub advertise_subnets: Vec, pub advertise_domains: Vec, pub auto_detect_domain: bool, pub heartbeat_interval_secs: u64, @@ -47,15 +49,73 @@ pub struct TunnelConf { pub server_spki_sha256: Option, } -impl TryFrom for TunnelConf { - type Error = anyhow::Error; +#[derive(Debug, Clone)] +struct GatewayEndpoint { + host: String, + port: NonZeroU16, +} + +impl std::str::FromStr for GatewayEndpoint { + type Err = anyhow::Error; + + fn from_str(endpoint: &str) -> anyhow::Result { + anyhow::ensure!(!endpoint.is_empty(), "value is required when Tunnel.Enabled is true"); + anyhow::ensure!( + endpoint.trim() == endpoint, + "value must not contain leading or trailing whitespace" + ); + + if let Ok(socket_addr) = endpoint.parse::() { + let port = NonZeroU16::new(socket_addr.port()).context("port must be greater than zero")?; + return Ok(Self { + host: socket_addr.ip().to_string(), + port, + }); + } + + let (hostname, port) = endpoint + .rsplit_once(':') + .context("expected an endpoint in host:port format")?; + anyhow::ensure!(!hostname.is_empty(), "hostname must not be empty"); + anyhow::ensure!( + hostname.parse::().is_ok() + || !hostname.chars().any(|character| matches!(character, ':' | '[' | ']')), + "IPv6 addresses must use bracketed host:port notation" + ); + + rustls_pki_types::ServerName::try_from(hostname.to_owned()) + .map_err(|_| anyhow::anyhow!("invalid hostname `{hostname}`"))?; - fn try_from(conf: dto::TunnelConf) -> anyhow::Result { + let port = port + .parse::() + .with_context(|| format!("invalid port `{port}`"))?; + + Ok(Self { + host: hostname.to_owned(), + port, + }) + } +} + +impl TunnelConf { + pub(crate) fn gateway_hostname(&self) -> &str { + self.gateway_endpoint().0 + } + + pub(crate) fn gateway_endpoint(&self) -> (&str, u16) { + let endpoint = self + .gateway_endpoint + .as_ref() + .expect("enabled tunnel has a gateway endpoint"); + (&endpoint.host, endpoint.port.get()) + } + + pub(crate) fn from_dto(conf: dto::TunnelConf) -> anyhow::Result { if !conf.enabled { // Disabled tunnel — return a placeholder with defaults. return Ok(Self { enabled: false, - gateway_endpoint: String::new(), + gateway_endpoint: None, client_cert_path: Utf8PathBuf::new(), client_key_path: Utf8PathBuf::new(), gateway_ca_cert_path: Utf8PathBuf::new(), @@ -68,38 +128,79 @@ impl TryFrom for TunnelConf { }); } - // Enabled tunnel — all required fields must be present. - let client_cert_path = conf - .client_cert_path - .context("tunnel enabled but client_cert_path not configured")?; - let client_key_path = conf - .client_key_path - .context("tunnel enabled but client_key_path not configured")?; - let gateway_ca_cert_path = conf - .gateway_ca_cert_path - .context("tunnel enabled but gateway_ca_cert_path not configured")?; + let gateway_endpoint = conf + .gateway_endpoint + .parse() + .context("invalid Tunnel.GatewayEndpoint")?; + + let client_cert_path = required_tunnel_path(conf.client_cert_path).context("invalid Tunnel.ClientCertPath")?; + let client_key_path = required_tunnel_path(conf.client_key_path).context("invalid Tunnel.ClientKeyPath")?; + let gateway_ca_cert_path = + required_tunnel_path(conf.gateway_ca_cert_path).context("invalid Tunnel.GatewayCaCertPath")?; + + let advertise_subnets = conf + .advertise_subnets + .into_iter() + .enumerate() + .map(|(index, subnet)| { + subnet + .parse::() + .with_context(|| format!("invalid Tunnel.AdvertiseSubnets[{index}] value `{subnet}`")) + }) + .collect::>>()?; + + for (index, route) in conf.advertise_domains.iter().enumerate() { + anyhow::ensure!( + DomainName::is_valid_route(route), + "invalid Tunnel.AdvertiseDomains[{index}] value `{route}`" + ); + } + let heartbeat_interval_secs = conf.heartbeat_interval_secs.unwrap_or(60); anyhow::ensure!( - !conf.gateway_endpoint.is_empty(), - "tunnel enabled but gateway_endpoint is empty" + heartbeat_interval_secs > 0, + "invalid Tunnel.HeartbeatIntervalSecs: value must be greater than zero" ); + let route_advertise_interval_secs = conf.route_advertise_interval_secs.unwrap_or(30); + anyhow::ensure!( + route_advertise_interval_secs > 0, + "invalid Tunnel.RouteAdvertiseIntervalSecs: value must be greater than zero" + ); + + let server_spki_sha256 = conf + .server_spki_sha256 + .map(|hash| { + anyhow::ensure!( + hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit()), + "invalid Tunnel.ServerSpkiSha256: expected 64 hexadecimal characters" + ); + Ok(hash.to_ascii_lowercase()) + }) + .transpose()?; + Ok(Self { enabled: true, - gateway_endpoint: conf.gateway_endpoint, + gateway_endpoint: Some(gateway_endpoint), client_cert_path, client_key_path, gateway_ca_cert_path, - advertise_subnets: conf.advertise_subnets, + advertise_subnets, advertise_domains: conf.advertise_domains, auto_detect_domain: conf.auto_detect_domain, - heartbeat_interval_secs: conf.heartbeat_interval_secs.unwrap_or(60), - route_advertise_interval_secs: conf.route_advertise_interval_secs.unwrap_or(30), - server_spki_sha256: conf.server_spki_sha256, + heartbeat_interval_secs, + route_advertise_interval_secs, + server_spki_sha256, }) } } +fn required_tunnel_path(path: Option) -> anyhow::Result { + let path = path.context("value is required when Tunnel.Enabled is true")?; + anyhow::ensure!(!path.as_str().trim().is_empty(), "path must not be empty"); + Ok(path) +} + /// Validated PSU agent configuration. /// /// Constructed from `dto::PsuConf` via `TryFrom for Option`. @@ -178,7 +279,7 @@ impl Conf { .tunnel .clone() .unwrap_or_default() - .pipe(TunnelConf::try_from) + .pipe(TunnelConf::from_dto) .context("invalid tunnel config")?, proxy: conf_file.proxy.clone().unwrap_or_default(), debug: conf_file.debug.clone().unwrap_or_default(), @@ -954,6 +1055,96 @@ pub fn handle_cli(command: &str) -> Result<(), anyhow::Error> { mod tests { use super::*; + fn valid_tunnel_json() -> serde_json::Value { + serde_json::json!({ + "Enabled": true, + "GatewayEndpoint": "[::1]:4433", + "ClientCertPath": "client.crt", + "ClientKeyPath": "client.key", + "GatewayCaCertPath": "gateway-ca.crt", + "HeartbeatIntervalSecs": 60, + "ServerSpkiSha256": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + }) + } + + fn load_tunnel_json(tunnel: serde_json::Value) -> anyhow::Result { + let conf_file = serde_json::from_value(serde_json::json!({ "Tunnel": tunnel })) + .context("deserialize test configuration")?; + Conf::from_conf_file(&conf_file) + } + + #[test] + fn tunnel_config_normalizes_spki_and_hostname() { + let conf = load_tunnel_json(valid_tunnel_json()).expect("load valid tunnel configuration"); + + assert_eq!(conf.tunnel.gateway_hostname(), "::1"); + assert_eq!( + conf.tunnel.server_spki_sha256.as_deref(), + Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + ); + } + + #[test] + fn tunnel_config_accepts_legacy_unbracketed_ipv6_endpoint() { + let mut tunnel = valid_tunnel_json(); + tunnel["GatewayEndpoint"] = serde_json::json!("::1:4433"); + + let conf = load_tunnel_json(tunnel).expect("load legacy tunnel endpoint"); + + assert_eq!(conf.tunnel.gateway_hostname(), "::1"); + assert_eq!(conf.tunnel.gateway_endpoint().1, 4433); + } + + #[test] + fn tunnel_config_errors_identify_invalid_fields() { + let cases = [ + ( + "GatewayEndpoint", + serde_json::json!("gateway.example.com"), + "invalid Tunnel.GatewayEndpoint", + ), + ("ClientKeyPath", serde_json::json!(""), "invalid Tunnel.ClientKeyPath"), + ( + "AdvertiseDomains", + serde_json::json!(["invalid.*.example.com"]), + "invalid Tunnel.AdvertiseDomains[0]", + ), + ( + "HeartbeatIntervalSecs", + serde_json::json!(0), + "invalid Tunnel.HeartbeatIntervalSecs", + ), + ( + "ServerSpkiSha256", + serde_json::json!("not-a-sha256-hash"), + "invalid Tunnel.ServerSpkiSha256", + ), + ]; + + for (field, value, expected) in cases { + let mut tunnel = valid_tunnel_json(); + tunnel[field] = value; + + let error = match load_tunnel_json(tunnel) { + Ok(_) => panic!("invalid {field} should fail loading"), + Err(error) => error, + }; + let error = format!("{error:#}"); + assert!(error.contains(expected), "expected `{expected}` in `{error}`"); + } + } + + #[test] + fn disabled_tunnel_skips_validation() { + let conf = load_tunnel_json(serde_json::json!({ + "Enabled": false, + "GatewayEndpoint": "invalid" + })) + .expect("load disabled tunnel configuration"); + + assert!(!conf.tunnel.enabled); + } + #[test] fn psu_config_deserializes() { let conf_file: dto::ConfFile = serde_json::from_value(serde_json::json!({ diff --git a/devolutions-agent/src/tunnel.rs b/devolutions-agent/src/tunnel.rs index 8ca7e7980..4e925c9f7 100644 --- a/devolutions-agent/src/tunnel.rs +++ b/devolutions-agent/src/tunnel.rs @@ -16,6 +16,7 @@ use async_trait::async_trait; use devolutions_gateway_task::{ShutdownSignal, Task}; use ipnetwork::Ipv4Network; use sha2::Digest as _; +use tokio::io::AsyncWriteExt as _; use crate::config::ConfHandle; use crate::tunnel_helpers::{Target, connect_to_target, resolve_target}; @@ -257,25 +258,12 @@ async fn run_single_connection( let key_path = &tunnel_conf.client_key_path; let ca_path = &tunnel_conf.gateway_ca_cert_path; - let advertise_subnets: Vec = tunnel_conf - .advertise_subnets - .iter() - .map(|subnet| subnet.parse()) - .collect::, _>>() - .context("failed to parse advertise_subnets")?; + let advertise_subnets = tunnel_conf.advertise_subnets.clone(); if advertise_subnets.is_empty() { warn!("No subnets configured to advertise"); } - if let Some(route) = tunnel_conf - .advertise_domains - .iter() - .find(|route| !DomainName::is_valid_route(route)) - { - bail!("invalid advertise domain route: {route}"); - } - let detected_domain = if tunnel_conf.auto_detect_domain && tunnel_conf.advertise_domains.is_empty() { crate::domain_detect::detect_domain() } else { @@ -454,12 +442,9 @@ async fn connect_to_gateway( // -- DNS resolve -- // Extract hostname for TLS server name validation. - let (gateway_hostname, _) = tunnel_conf - .gateway_endpoint - .rsplit_once(':') - .context("gateway_endpoint missing port separator")?; + let gateway_hostname = tunnel_conf.gateway_hostname(); - let gateway_addr = tokio::net::lookup_host(&tunnel_conf.gateway_endpoint) + let gateway_addr = tokio::net::lookup_host(tunnel_conf.gateway_endpoint()) .await .context("failed to resolve gateway endpoint")? .next() @@ -517,7 +502,7 @@ pub async fn probe_connectivity(tunnel_conf: &crate::config::TunnelConf, timeout } async fn reach_gateway(tunnel_conf: &crate::config::TunnelConf) -> anyhow::Result<()> { - let gateway_addr = tokio::net::lookup_host(&tunnel_conf.gateway_endpoint) + let gateway_addr = tokio::net::lookup_host(tunnel_conf.gateway_endpoint()) .await .context("failed to resolve gateway endpoint")? .next() @@ -756,7 +741,7 @@ async fn run_session_proxy( // Whatever went wrong has to travel back as a ConnectResponse::Error — returning // early instead drops the stream and the Gateway just sees an unexplained EOF. - let (mut tcp_stream, selected_target) = match connect_result { + let (tcp_stream, selected_target) = match connect_result { Ok(connected) => connected, Err(error) => { let reason = format!("{error:#}"); @@ -783,10 +768,31 @@ async fn run_session_proxy( .context("send ConnectResponse")?; info!("Sent ConnectResponse::Success"); - let (send, recv) = session.into_inner(); - tokio::io::copy_bidirectional(&mut tokio::io::join(recv, send), &mut tcp_stream) - .await - .context("proxy session traffic")?; + let (mut send, mut recv) = session.into_inner(); + let (mut tcp_read, mut tcp_write) = tcp_stream.into_split(); + + let quic_to_tcp = async { + tokio::io::copy(&mut recv, &mut tcp_write) + .await + .context("proxy QUIC to TCP")?; + if let Err(error) = tcp_write.shutdown().await { + debug!(%error, "TCP write shutdown failed"); + } + Ok::<_, anyhow::Error>(()) + }; + let tcp_to_quic = async { + tokio::io::copy(&mut tcp_read, &mut send) + .await + .context("proxy TCP to QUIC")?; + if let Err(error) = send.shutdown().await { + debug!(%error, "QUIC send shutdown failed"); + } + Ok::<_, anyhow::Error>(()) + }; + + let (quic_to_tcp, tcp_to_quic) = tokio::join!(quic_to_tcp, tcp_to_quic); + quic_to_tcp?; + tcp_to_quic?; Ok(()) } @@ -799,28 +805,23 @@ mod tests { use camino::Utf8PathBuf; use super::*; - use crate::config::TunnelConf; + use crate::config::{TunnelConf, dto}; - fn tunnel_conf_template() -> TunnelConf { - TunnelConf { + fn tunnel_conf(endpoint: impl Into) -> TunnelConf { + let conf = dto::TunnelConf { enabled: true, - gateway_endpoint: String::new(), - client_cert_path: Utf8PathBuf::new(), - client_key_path: Utf8PathBuf::new(), - gateway_ca_cert_path: Utf8PathBuf::new(), - advertise_subnets: Vec::new(), - advertise_domains: Vec::new(), - auto_detect_domain: false, - heartbeat_interval_secs: 15, - route_advertise_interval_secs: 60, - server_spki_sha256: None, - } + gateway_endpoint: endpoint.into(), + client_cert_path: Some(Utf8PathBuf::from("client.crt")), + client_key_path: Some(Utf8PathBuf::from("client.key")), + gateway_ca_cert_path: Some(Utf8PathBuf::from("gateway-ca.crt")), + ..dto::TunnelConf::default() + }; + TunnelConf::from_dto(conf).expect("validate tunnel configuration") } #[tokio::test] async fn probe_fails_fast_when_tunnel_disabled() { - let mut conf = tunnel_conf_template(); - conf.enabled = false; + let conf = TunnelConf::from_dto(dto::TunnelConf::default()).expect("validate disabled tunnel configuration"); let error = probe_connectivity(&conf, Duration::from_millis(200)) .await @@ -842,8 +843,7 @@ mod tests { .expect("bind blackhole socket"); let blackhole_addr = blackhole.local_addr().expect("blackhole addr"); - let mut conf = tunnel_conf_template(); - conf.gateway_endpoint = blackhole_addr.to_string(); + let conf = tunnel_conf(blackhole_addr.to_string()); let started = std::time::Instant::now(); let result = probe_connectivity(&conf, Duration::from_millis(300)).await; diff --git a/devolutions-gateway/src/jmux.rs b/devolutions-gateway/src/jmux.rs index cbf6f3dec..a2c8ea2fd 100644 --- a/devolutions-gateway/src/jmux.rs +++ b/devolutions-gateway/src/jmux.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use anyhow::Context as _; use devolutions_gateway_task::ChildTask; -use jmux_proxy::{ConnectedTarget, DestinationUrl, FilteringRule, JmuxConfig, JmuxProxy}; +use jmux_proxy::{DestinationUrl, FilteringRule, JmuxConfig, JmuxProxy}; use tap::prelude::*; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::sync::Notify; @@ -113,7 +113,7 @@ pub async fn handle( .with_outgoing_traffic_event_callback(traffic_event_callback); if let Some(agent_tunnel_handle) = agent_tunnel_handle { - proxy = proxy.with_target_connector(move |destination_url: DestinationUrl| { + proxy = proxy.with_target_connector_override(move |destination_url: DestinationUrl| { let agent_tunnel_handle = Arc::clone(&agent_tunnel_handle); async move { @@ -139,7 +139,7 @@ pub async fn handle( return Ok(None); }; - Ok(Some(ConnectedTarget::new(stream))) + Ok(Some(stream)) } }); } From 69f4f7c0c47af0fc8b7a35340a728ea729d93699 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 12:57:32 -0400 Subject: [PATCH 09/17] test(dgw,agent): cover routed JMUX boundaries Pin half-close propagation in both relay directions and exercise the Gateway JMUX binding against matched Agent routes. Verify successful routed data flow and reject Agent failures without attempting a direct TCP fallback. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Cargo.lock | 1 + devolutions-agent/src/tunnel.rs | 138 ++++++++++++---- testsuite/Cargo.toml | 1 + testsuite/tests/agent_tunnel/integration.rs | 164 +++++++++++++++++++- 4 files changed, 276 insertions(+), 28 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7e40a46d0..312f081db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7539,6 +7539,7 @@ dependencies = [ "fastrand", "futures-util", "ipnetwork", + "jmux-proto", "libsql", "mcp-proxy", "network-scanner", diff --git a/devolutions-agent/src/tunnel.rs b/devolutions-agent/src/tunnel.rs index 4e925c9f7..dfc174747 100644 --- a/devolutions-agent/src/tunnel.rs +++ b/devolutions-agent/src/tunnel.rs @@ -16,7 +16,7 @@ use async_trait::async_trait; use devolutions_gateway_task::{ShutdownSignal, Task}; use ipnetwork::Ipv4Network; use sha2::Digest as _; -use tokio::io::AsyncWriteExt as _; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _}; use crate::config::ConfHandle; use crate::tunnel_helpers::{Target, connect_to_target, resolve_target}; @@ -583,8 +583,8 @@ async fn try_renew_certificate( ca_path: &camino::Utf8Path, ) -> anyhow::Result> where - S: tokio::io::AsyncWrite + Unpin, - R: tokio::io::AsyncRead + Unpin, + S: AsyncWrite + Unpin, + R: AsyncRead + Unpin, { const RENEWAL_THRESHOLD_DAYS: u32 = 15; const RENEWAL_TIMEOUT: Duration = Duration::from_secs(30); @@ -660,7 +660,7 @@ where // Control stream reader // --------------------------------------------------------------------------- -async fn run_control_reader(mut ctrl: FramedRecv) { +async fn run_control_reader(mut ctrl: FramedRecv) { let _ = async move { loop { let message: ControlMessage = ctrl.recv().await.context("recv control message")?; @@ -702,6 +702,34 @@ async fn run_control_reader(mut ctrl: FramedRec /// black-holed target outlives its deadline and it never hears why we failed. const CONNECT_DEADLINE: Duration = Duration::from_secs(20); +async fn proxy_session_traffic( + (mut tunnel_send, mut tunnel_recv): (impl AsyncWrite + Unpin, impl AsyncRead + Unpin), + (mut target_read, mut target_write): (impl AsyncRead + Unpin, impl AsyncWrite + Unpin), +) -> anyhow::Result<()> { + let tunnel_to_target = async { + tokio::io::copy(&mut tunnel_recv, &mut target_write) + .await + .context("proxy QUIC to TCP")?; + if let Err(error) = target_write.shutdown().await { + debug!(%error, "TCP write shutdown failed"); + } + Ok::<_, anyhow::Error>(()) + }; + let target_to_tunnel = async { + tokio::io::copy(&mut target_read, &mut tunnel_send) + .await + .context("proxy TCP to QUIC")?; + if let Err(error) = tunnel_send.shutdown().await { + debug!(%error, "QUIC send shutdown failed"); + } + Ok::<_, anyhow::Error>(()) + }; + + let (tunnel_to_target, target_to_tunnel) = tokio::join!(tunnel_to_target, target_to_tunnel); + tunnel_to_target?; + target_to_tunnel +} + async fn run_session_proxy( advertise_subnets: Vec, advertise_domains: Vec, @@ -770,29 +798,7 @@ async fn run_session_proxy( let (mut send, mut recv) = session.into_inner(); let (mut tcp_read, mut tcp_write) = tcp_stream.into_split(); - - let quic_to_tcp = async { - tokio::io::copy(&mut recv, &mut tcp_write) - .await - .context("proxy QUIC to TCP")?; - if let Err(error) = tcp_write.shutdown().await { - debug!(%error, "TCP write shutdown failed"); - } - Ok::<_, anyhow::Error>(()) - }; - let tcp_to_quic = async { - tokio::io::copy(&mut tcp_read, &mut send) - .await - .context("proxy TCP to QUIC")?; - if let Err(error) = send.shutdown().await { - debug!(%error, "QUIC send shutdown failed"); - } - Ok::<_, anyhow::Error>(()) - }; - - let (quic_to_tcp, tcp_to_quic) = tokio::join!(quic_to_tcp, tcp_to_quic); - quic_to_tcp?; - tcp_to_quic?; + proxy_session_traffic((&mut send, &mut recv), (&mut tcp_read, &mut tcp_write)).await?; Ok(()) } @@ -803,10 +809,88 @@ async fn run_session_proxy( #[cfg(test)] mod tests { use camino::Utf8PathBuf; + use tokio::io::AsyncReadExt as _; use super::*; use crate::config::{TunnelConf, dto}; + async fn tcp_pair() -> (tokio::net::TcpStream, tokio::net::TcpStream) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind TCP listener"); + let client = tokio::net::TcpStream::connect(listener.local_addr().expect("read listener address")) + .await + .expect("connect TCP client"); + let (server, _) = listener.accept().await.expect("accept TCP client"); + (client, server) + } + + async fn spawn_relay() -> ( + tokio::io::DuplexStream, + tokio::net::TcpStream, + tokio::task::JoinHandle>, + ) { + let (tunnel, gateway) = tokio::io::duplex(64); + let (tunnel_recv, tunnel_send) = tokio::io::split(tunnel); + let (target, peer) = tcp_pair().await; + let (target_read, target_write) = target.into_split(); + let relay = tokio::spawn(proxy_session_traffic( + (tunnel_send, tunnel_recv), + (target_read, target_write), + )); + (gateway, peer, relay) + } + + #[tokio::test] + async fn tunnel_eof_half_closes_target_and_preserves_response() { + tokio::time::timeout(Duration::from_secs(5), async { + let (mut gateway, mut peer, relay) = spawn_relay().await; + + gateway.write_all(b"request").await.expect("write tunnel request"); + gateway.shutdown().await.expect("finish tunnel request"); + + let mut request = [0; 7]; + peer.read_exact(&mut request).await.expect("read target request"); + assert_eq!(&request, b"request"); + assert_eq!(peer.read(&mut [0]).await.expect("read target EOF"), 0); + + peer.write_all(b"response").await.expect("write target response"); + peer.shutdown().await.expect("finish target response"); + + let mut response = [0; 8]; + gateway.read_exact(&mut response).await.expect("read tunnel response"); + assert_eq!(&response, b"response"); + relay.await.expect("relay task panicked").expect("relay traffic"); + }) + .await + .expect("tunnel EOF test timed out"); + } + + #[tokio::test] + async fn target_eof_finishes_tunnel_send_and_preserves_request() { + tokio::time::timeout(Duration::from_secs(5), async { + let (mut gateway, mut peer, relay) = spawn_relay().await; + + peer.write_all(b"response").await.expect("write target response"); + peer.shutdown().await.expect("finish target response"); + + let mut response = [0; 8]; + gateway.read_exact(&mut response).await.expect("read tunnel response"); + assert_eq!(&response, b"response"); + assert_eq!(gateway.read(&mut [0]).await.expect("read tunnel EOF"), 0); + + gateway.write_all(b"request").await.expect("write tunnel request"); + gateway.shutdown().await.expect("finish tunnel request"); + + let mut request = [0; 7]; + peer.read_exact(&mut request).await.expect("read target request"); + assert_eq!(&request, b"request"); + relay.await.expect("relay task panicked").expect("relay traffic"); + }) + .await + .expect("target EOF test timed out"); + } + fn tunnel_conf(endpoint: impl Into) -> TunnelConf { let conf = dto::TunnelConf { enabled: true, diff --git a/testsuite/Cargo.toml b/testsuite/Cargo.toml index 1cc45390a..88e163f60 100644 --- a/testsuite/Cargo.toml +++ b/testsuite/Cargo.toml @@ -40,6 +40,7 @@ devolutions-gateway-task = { path = "../crates/devolutions-gateway-task" } devolutions-gateway = { path = "../devolutions-gateway" } futures-util = "0.3" ipnetwork = "0.20" +jmux-proto = { path = "../crates/jmux-proto" } libsql = { version = "0.9", default-features = false, features = ["core"] } mcp-proxy.path = "../crates/mcp-proxy" network-scanner = { path = "../crates/network-scanner", features = ["test-utils"] } diff --git a/testsuite/tests/agent_tunnel/integration.rs b/testsuite/tests/agent_tunnel/integration.rs index b36433689..1389324e0 100644 --- a/testsuite/tests/agent_tunnel/integration.rs +++ b/testsuite/tests/agent_tunnel/integration.rs @@ -1,3 +1,4 @@ +use std::sync::Arc; use std::time::Duration; use agent_tunnel::AgentTunnelHandle; @@ -5,10 +6,17 @@ use agent_tunnel::registry::AgentRegistry; use agent_tunnel_proto::{ CertRenewalResult, ConnectResponse, ControlMessage, ControlStream, DomainAdvertisement, DomainName, }; +use devolutions_gateway::recording::recording_message_channel; +use devolutions_gateway::session::SessionManagerTask; +use devolutions_gateway::subscriber::subscriber_channel; use devolutions_gateway::target_addr::TargetAddr; +use devolutions_gateway::token::{ApplicationProtocol, JmuxTokenClaims, RecordingPolicy, SessionTtl}; +use devolutions_gateway::traffic_audit::TrafficAuditHandle; use devolutions_gateway::upstream::{ConnectedUpstream, UpstreamLeg, connect_upstream}; +use devolutions_gateway_task::{ShutdownHandle, Task}; +use jmux_proto::{Bytes, BytesMut, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; use nonempty::NonEmpty; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; use uuid::Uuid; @@ -21,6 +29,32 @@ fn target(host: &str, port: u16) -> TargetAddr { TargetAddr::from_components("tcp", host, port).expect("build target address") } +async fn send_jmux_message(writer: &mut (impl AsyncWrite + Unpin), message: Message) { + let mut bytes = BytesMut::new(); + message.encode(&mut bytes).expect("encode JMUX message"); + writer.write_all(&bytes).await.expect("send JMUX message"); +} + +async fn receive_jmux_message(reader: &mut (impl AsyncRead + Unpin)) -> Message { + tokio::time::timeout(Duration::from_secs(5), async { + let mut header = [0; Header::SIZE]; + reader.read_exact(&mut header).await.expect("read JMUX header"); + let header = Header::decode(Bytes::copy_from_slice(&header)).expect("decode JMUX header"); + let body_size = usize::from(header.size) + .checked_sub(Header::SIZE) + .expect("JMUX message size smaller than header"); + let mut body = vec![0; body_size]; + reader.read_exact(&mut body).await.expect("read JMUX body"); + + let mut bytes = BytesMut::with_capacity(usize::from(header.size)); + header.encode(&mut bytes); + bytes.extend_from_slice(&body); + Message::decode(bytes.freeze()).expect("decode JMUX message") + }) + .await + .expect("JMUX response timed out") +} + async fn advertise_domain( connection: &quinn::Connection, registry: &AgentRegistry, @@ -311,6 +345,134 @@ async fn gateway_connect_upstream_does_not_bypass_failed_agent_routes() { listener.shutdown().await; } +#[tokio::test] +async fn gateway_jmux_uses_agent_route_without_direct_fallback() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("jmux-agent").await; + let _ctrl = advertise_routes( + &connection, + listener.handle.registry(), + agent_id, + 1, + vec!["127.0.0.0/8".parse().expect("parse test subnet")], + vec![], + ) + .await; + let direct_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind direct target"); + let target_port = direct_listener.local_addr().expect("read direct target address").port(); + let target = target("127.0.0.1", target_port); + let session_id = Uuid::new_v4(); + + let (recordings, _recording_rx) = recording_message_channel(); + let session_manager = SessionManagerTask::init(recordings); + let sessions = session_manager.handle(); + let (session_shutdown, session_shutdown_signal) = ShutdownHandle::new(); + let session_task = tokio::spawn(session_manager.run(session_shutdown_signal)); + let (subscriber_tx, _subscriber_rx) = subscriber_channel(); + let (traffic_audit_handle, _traffic_audit_rx) = TrafficAuditHandle::new(); + let claims = JmuxTokenClaims { + jet_aid: session_id, + hosts: NonEmpty::new(target.clone()), + jet_ap: ApplicationProtocol::unknown(), + jet_rec: RecordingPolicy::None, + jet_ttl: SessionTtl::Unlimited, + exp: i64::MAX, + jti: Uuid::new_v4(), + }; + let (proxy_stream, mut peer_stream) = tokio::io::duplex(8192); + let proxy_task = tokio::spawn(devolutions_gateway::jmux::handle( + proxy_stream, + claims, + sessions, + subscriber_tx, + traffic_audit_handle, + Some(Arc::new(listener.handle.clone())), + )); + + send_jmux_message( + &mut peer_stream, + Message::open( + LocalChannelId::from(20), + 4096, + jmux_proto::DestinationUrl::new("tcp", "127.0.0.1", target_port), + ), + ) + .await; + let mut routed_session = tokio::time::timeout( + Duration::from_secs(5), + accept_session_request(&connection, session_id, target.as_addr()), + ) + .await + .expect("routed JMUX request timed out"); + routed_session + .send_response(&ConnectResponse::success()) + .await + .expect("accept routed JMUX request"); + let Message::OpenSuccess(success) = receive_jmux_message(&mut peer_stream).await else { + panic!("expected OPEN SUCCESS"); + }; + assert_eq!(success.recipient_channel_id, 20); + + let local_id = DistantChannelId::from(success.sender_channel_id); + send_jmux_message(&mut peer_stream, Message::data(local_id, Bytes::from_static(b"ping"))).await; + let (mut routed_send, mut routed_recv) = routed_session.into_inner(); + let mut request = [0; 4]; + routed_recv + .read_exact(&mut request) + .await + .expect("read routed JMUX payload"); + assert_eq!(&request, b"ping"); + routed_send + .write_all(b"pong") + .await + .expect("write routed JMUX response"); + let Message::Data(response) = receive_jmux_message(&mut peer_stream).await else { + panic!("expected CHANNEL DATA"); + }; + assert_eq!(response.recipient_channel_id, 20); + assert_eq!(response.transfer_data, b"pong"[..]); + + send_jmux_message( + &mut peer_stream, + Message::open( + LocalChannelId::from(21), + 4096, + jmux_proto::DestinationUrl::new("tcp", "127.0.0.1", target_port), + ), + ) + .await; + let mut failed_session = tokio::time::timeout( + Duration::from_secs(5), + accept_session_request(&connection, session_id, target.as_addr()), + ) + .await + .expect("failed routed JMUX request timed out"); + failed_session + .send_response(&ConnectResponse::error("connection refused")) + .await + .expect("reject routed JMUX request"); + let Message::OpenFailure(failure) = receive_jmux_message(&mut peer_stream).await else { + panic!("expected OPEN FAILURE"); + }; + assert_eq!(failure.recipient_channel_id, 21); + assert_eq!(failure.reason_code, ReasonCode::GENERAL_FAILURE); + assert!( + tokio::time::timeout(Duration::from_millis(100), direct_listener.accept()) + .await + .is_err(), + "matched Agent route must not fall back to direct TCP" + ); + + proxy_task.abort(); + session_shutdown.signal(); + session_task + .await + .expect("session manager task panicked") + .expect("session manager shutdown"); + connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} + #[tokio::test] async fn gateway_listener_rejects_certificate_renewal_key_rotation() { let listener = bind_test_listener().await; From a2a58b47e4b397ef7578e18fedb4487f7413029b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:03:13 -0400 Subject: [PATCH 10/17] test(jmux): cover connector half-close Verify JMUX EOF reaches an overridden target stream without cancelling traffic in the reverse direction. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/jmux-proxy/tests/target_connector.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs index 20c05dc7d..31c363acb 100644 --- a/crates/jmux-proxy/tests/target_connector.rs +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -42,6 +42,7 @@ async fn override_stream_carries_channel_data() { tokio::spawn(async move { let mut payload = [0; 4]; target_peer.read_exact(&mut payload).await.expect("read target data"); + assert_eq!(target_peer.read(&mut [0]).await.expect("read target EOF"), 0); target_peer.write_all(&payload).await.expect("echo target data"); }); Ok(Some(target_stream)) @@ -64,6 +65,7 @@ async fn override_stream_carries_channel_data() { let local_id = DistantChannelId::from(open_success.sender_channel_id); send_message(&mut peer_writer, Message::data(local_id, Bytes::from_static(b"ping"))).await; + send_message(&mut peer_writer, Message::eof(local_id)).await; let Message::Data(data) = receive_message(&mut peer_reader).await else { panic!("expected CHANNEL DATA"); }; From 8a30605d2f776de67fc1b19d1229f2bf2d23ee16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:11:12 -0400 Subject: [PATCH 11/17] test(jmux): make EOF probes explicit Use named one-byte buffers so half-close assertions cannot be mistaken for unconditional zero-length reads. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/jmux-proxy/tests/target_connector.rs | 3 ++- devolutions-agent/src/tunnel.rs | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs index 31c363acb..ea2d5c6c4 100644 --- a/crates/jmux-proxy/tests/target_connector.rs +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -42,7 +42,8 @@ async fn override_stream_carries_channel_data() { tokio::spawn(async move { let mut payload = [0; 4]; target_peer.read_exact(&mut payload).await.expect("read target data"); - assert_eq!(target_peer.read(&mut [0]).await.expect("read target EOF"), 0); + let mut eof_probe = [0u8; 1]; + assert_eq!(target_peer.read(&mut eof_probe).await.expect("read target EOF"), 0); target_peer.write_all(&payload).await.expect("echo target data"); }); Ok(Some(target_stream)) diff --git a/devolutions-agent/src/tunnel.rs b/devolutions-agent/src/tunnel.rs index dfc174747..279727fe8 100644 --- a/devolutions-agent/src/tunnel.rs +++ b/devolutions-agent/src/tunnel.rs @@ -852,7 +852,8 @@ mod tests { let mut request = [0; 7]; peer.read_exact(&mut request).await.expect("read target request"); assert_eq!(&request, b"request"); - assert_eq!(peer.read(&mut [0]).await.expect("read target EOF"), 0); + let mut eof_probe = [0u8; 1]; + assert_eq!(peer.read(&mut eof_probe).await.expect("read target EOF"), 0); peer.write_all(b"response").await.expect("write target response"); peer.shutdown().await.expect("finish target response"); @@ -877,7 +878,8 @@ mod tests { let mut response = [0; 8]; gateway.read_exact(&mut response).await.expect("read tunnel response"); assert_eq!(&response, b"response"); - assert_eq!(gateway.read(&mut [0]).await.expect("read tunnel EOF"), 0); + let mut eof_probe = [0u8; 1]; + assert_eq!(gateway.read(&mut eof_probe).await.expect("read tunnel EOF"), 0); gateway.write_all(b"request").await.expect("write tunnel request"); gateway.shutdown().await.expect("finish tunnel request"); From 70bbb44c0c609b244ecc426a75b024a58670b925 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:28:12 -0400 Subject: [PATCH 12/17] fix(dgw,agent): preserve Agent route liveness Share the Gateway offline timeout and require at least one Agent liveness interval to refresh within one third of that window. Scope JMUX traffic-event guarantees to direct connections whose resolved target IP is available. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/agent-tunnel-proto/src/lib.rs | 3 +++ crates/agent-tunnel/src/registry.rs | 4 ++-- crates/jmux-proxy/src/lib.rs | 12 +++++++----- devolutions-agent/src/config.rs | 21 +++++++++++++++++++++ 4 files changed, 33 insertions(+), 7 deletions(-) diff --git a/crates/agent-tunnel-proto/src/lib.rs b/crates/agent-tunnel-proto/src/lib.rs index 17af4919e..32bc5ce14 100644 --- a/crates/agent-tunnel-proto/src/lib.rs +++ b/crates/agent-tunnel-proto/src/lib.rs @@ -29,6 +29,9 @@ pub use session::{ConnectRequest, ConnectResponse, MAX_SESSION_MESSAGE_SIZE}; pub use stream::{ControlStream, FramedRecv, FramedSend, SessionStream}; pub use version::{ALPN_PROTOCOL, CURRENT_PROTOCOL_VERSION, MIN_SUPPORTED_VERSION, validate_protocol_version}; +/// Maximum time the Gateway keeps an Agent route online without a liveness message. +pub const AGENT_OFFLINE_TIMEOUT_SECS: u64 = 90; + /// Current wall-clock time in milliseconds since UNIX epoch. pub fn current_time_millis() -> u64 { u64::try_from( diff --git a/crates/agent-tunnel/src/registry.rs b/crates/agent-tunnel/src/registry.rs index eaa92df38..5587804f9 100644 --- a/crates/agent-tunnel/src/registry.rs +++ b/crates/agent-tunnel/src/registry.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; -use agent_tunnel_proto::{DomainAdvertisement, current_time_millis}; +use agent_tunnel_proto::{AGENT_OFFLINE_TIMEOUT_SECS, DomainAdvertisement, current_time_millis}; use ipnetwork::Ipv4Network; use parking_lot::RwLock; use serde::Serialize; @@ -13,7 +13,7 @@ use uuid::Uuid; use crate::routing::RouteTarget; /// Duration after which an agent is considered offline if no heartbeat has been received. -pub const AGENT_OFFLINE_TIMEOUT: Duration = Duration::from_secs(90); +pub const AGENT_OFFLINE_TIMEOUT: Duration = Duration::from_secs(AGENT_OFFLINE_TIMEOUT_SECS); /// Tracks route advertisements received from an agent. /// diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index 54f6594b8..ec55187a0 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -130,7 +130,8 @@ impl JmuxProxy { /// /// Return `Ok(Some(stream))` to use the override, `Ok(None)` to delegate to the default connector, /// or an error to reject the connection without falling back. - /// Overridden streams do not emit outgoing traffic events because their resolved target IP is unknown. + /// Connection attempts handled by the override do not emit outgoing traffic events because + /// their resolved target IP is unknown. #[must_use] pub fn with_target_connector_override(mut self, connector: C) -> Self where @@ -152,9 +153,9 @@ impl JmuxProxy { /// Configures an outgoing-traffic callback for lifecycle event monitoring. /// - /// The provided callback will be invoked exactly once per outgoing stream at the end of its - /// lifecycle, providing comprehensive audit information including connection metadata, - /// byte counts, timing, and termination classification. + /// The provided callback is invoked exactly once at the end of each outgoing stream whose + /// resolved target IP is known. + /// It provides connection metadata, byte counts, timing, and termination classification. /// /// # Event Emission /// @@ -165,6 +166,7 @@ impl JmuxProxy { /// /// Events are **NOT** emitted for: /// - DNS resolution failures (no concrete IP address available) + /// - Connection attempts handled by a target connector override (no concrete IP address available) /// - Internal JMUX protocol errors before stream establishment /// /// For hostnames with multiple IP addresses, connection attempts follow a Happy Eyeballs @@ -172,7 +174,7 @@ impl JmuxProxy { /// /// # Callback Contract /// - /// - **Exactly once**: Each traffic item generates precisely one event, protected by atomic guards + /// - **Exactly once**: Each eligible traffic item generates one event, protected by atomic guards /// - **At stream end**: Events are emitted during cleanup, not during operation /// - **Synchronous**: The callback is called synchronously from JMUX task contexts /// - **Thread safe**: Must be `Send + Sync + 'static` for multi-threaded access diff --git a/devolutions-agent/src/config.rs b/devolutions-agent/src/config.rs index 3c9d186bb..4e8defe63 100644 --- a/devolutions-agent/src/config.rs +++ b/devolutions-agent/src/config.rs @@ -167,6 +167,13 @@ impl TunnelConf { route_advertise_interval_secs > 0, "invalid Tunnel.RouteAdvertiseIntervalSecs: value must be greater than zero" ); + anyhow::ensure!( + heartbeat_interval_secs.min(route_advertise_interval_secs) + <= agent_tunnel_proto::AGENT_OFFLINE_TIMEOUT_SECS / 3, + "invalid Tunnel.HeartbeatIntervalSecs and Tunnel.RouteAdvertiseIntervalSecs: \ + at least one value must be at most {} seconds", + agent_tunnel_proto::AGENT_OFFLINE_TIMEOUT_SECS / 3 + ); let server_spki_sha256 = conf .server_spki_sha256 @@ -1145,6 +1152,20 @@ mod tests { assert!(!conf.tunnel.enabled); } + #[test] + fn tunnel_config_requires_liveness_margin() { + let mut tunnel = valid_tunnel_json(); + let invalid_interval = agent_tunnel_proto::AGENT_OFFLINE_TIMEOUT_SECS / 3 + 1; + tunnel["HeartbeatIntervalSecs"] = serde_json::json!(invalid_interval); + tunnel["RouteAdvertiseIntervalSecs"] = serde_json::json!(invalid_interval); + + let error = load_tunnel_json(tunnel).expect_err("stale liveness intervals should fail loading"); + + assert!( + format!("{error:#}").contains("invalid Tunnel.HeartbeatIntervalSecs and Tunnel.RouteAdvertiseIntervalSecs") + ); + } + #[test] fn psu_config_deserializes() { let conf_file: dto::ConfFile = serde_json::from_value(serde_json::json!({ From 83d6fb77b1b752feb882fd88b8abef311849d479 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:35:23 -0400 Subject: [PATCH 13/17] fix(dgw,agent): enforce route refresh margin Require a heartbeat or route advertisement every 30 seconds so transient scheduling delays cannot age a connected Agent out of the Gateway registry. Document that connector-managed JMUX attempts cannot emit traffic events until the Gateway receives their resolved target IP. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/jmux-proxy/src/event.rs | 16 ++++++++-------- crates/jmux-proxy/src/lib.rs | 9 +++------ 2 files changed, 11 insertions(+), 14 deletions(-) diff --git a/crates/jmux-proxy/src/event.rs b/crates/jmux-proxy/src/event.rs index 5ecfe1845..212405848 100644 --- a/crates/jmux-proxy/src/event.rs +++ b/crates/jmux-proxy/src/event.rs @@ -66,8 +66,8 @@ pub enum EventOutcome { /// Complete audit information for one traffic item's lifecycle. /// -/// A single `TrafficEvent` is emitted exactly once per JMUX traffic item when -/// it ends (successfully or with error). +/// A single `TrafficEvent` is emitted exactly once when an eligible JMUX traffic item ends. +/// An item is eligible only when direct target resolution provides a concrete IP address. /// /// # Timestamp semantics /// @@ -94,7 +94,7 @@ pub enum EventOutcome { /// - connect failure: the last IP that was attempted /// - `target_port`: the destination port. /// -/// DNS failures do **not** produce an event because `target_ip` is unknown. +/// DNS failures and connection attempts handled by connector overrides do **not** produce an event because `target_ip` is unknown. #[derive(Clone, Debug)] pub struct TrafficEvent { /// How the traffic item's lifecycle ended. @@ -142,13 +142,13 @@ pub struct TrafficEvent { /// Type-erased traffic audit callback. /// -/// Invoked exactly once per JMUX traffic item at end-of-lifecycle. The callback -/// itself is **synchronous**; perform any asynchronous work by spawning within -/// the callback (e.g., `tokio::spawn`) or by sending to an internal channel. +/// Invoked exactly once at the end of each eligible JMUX traffic item. +/// An item is eligible only when direct target resolution provides a concrete IP address. +/// The callback itself is **synchronous**; perform asynchronous work by spawning within the callback (e.g., `tokio::spawn`) or by sending to an internal channel. /// /// # Exactly-once /// -/// - Each traffic item yields exactly one event. +/// - Each eligible traffic item yields exactly one event. /// - Emitted at cleanup time, not during operation. /// - Guarded to prevent duplicate emission. /// - No aggregation—each event stands alone. @@ -162,7 +162,7 @@ pub struct TrafficEvent { /// /// ```rust,ignore /// let proxy = JmuxProxy::new(reader, writer) -/// .with_traffic_event_callback(|event| { +/// .with_outgoing_traffic_event_callback(|event| { /// // Log quickly... /// tracing::info!( /// outcome = ?event.outcome, diff --git a/crates/jmux-proxy/src/lib.rs b/crates/jmux-proxy/src/lib.rs index ec55187a0..4c5c7d00f 100644 --- a/crates/jmux-proxy/src/lib.rs +++ b/crates/jmux-proxy/src/lib.rs @@ -128,10 +128,8 @@ impl JmuxProxy { /// Overrides the default direct TCP connector when applicable. /// - /// Return `Ok(Some(stream))` to use the override, `Ok(None)` to delegate to the default connector, - /// or an error to reject the connection without falling back. - /// Connection attempts handled by the override do not emit outgoing traffic events because - /// their resolved target IP is unknown. + /// Return `Ok(Some(stream))` to use the override, `Ok(None)` to delegate to the default connector, or an error to reject the connection without falling back. + /// Connection attempts handled by the override do not emit outgoing traffic events because their resolved target IP is unknown. #[must_use] pub fn with_target_connector_override(mut self, connector: C) -> Self where @@ -153,8 +151,7 @@ impl JmuxProxy { /// Configures an outgoing-traffic callback for lifecycle event monitoring. /// - /// The provided callback is invoked exactly once at the end of each outgoing stream whose - /// resolved target IP is known. + /// The provided callback is invoked exactly once at the end of each outgoing stream whose resolved target IP is known. /// It provides connection metadata, byte counts, timing, and termination classification. /// /// # Event Emission From c28ac7aed25da1936438e8a2f6fb4bad009147e5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:37:39 -0400 Subject: [PATCH 14/17] ci: update PSTools checksum Microsoft republished the signed PSTools archive. Pin the verified replacement hash so Windows policy and PEDM jobs can install PsExec again. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 928eb54be..b8ce52c19 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1236,7 +1236,7 @@ jobs: - name: Install PsExec shell: pwsh run: | - $expectedHash = '4F49964CC9CBAC2B5D87BDC8F9526012E9C4B243D8B7D0C0BB51F254A721CA2E' + $expectedHash = '2B10B3D9DAE0403B06D90B13BFB53E723A8B14A788F78CDFD43A445D8991415E' $zipPath = Join-Path $env:RUNNER_TEMP 'PSTools.zip' $toolsDir = Join-Path $env:RUNNER_TEMP 'PSTools' Invoke-WebRequest -Uri 'https://download.sysinternals.com/files/PSTools.zip' -OutFile $zipPath From 64c9f680c6f77f6cb303a8234adaf58fce12752b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:41:43 -0400 Subject: [PATCH 15/17] ci: align Agent policy PSTools checksum The latest master adds a second PsExec installation for the Agent policy end-to-end job. Keep it pinned to the same verified Microsoft archive as the PEDM simulator job. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9b88e92ea..4d2952c0e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1290,7 +1290,7 @@ jobs: - name: Install PsExec shell: pwsh run: | - $expectedHash = '4F49964CC9CBAC2B5D87BDC8F9526012E9C4B243D8B7D0C0BB51F254A721CA2E' + $expectedHash = '2B10B3D9DAE0403B06D90B13BFB53E723A8B14A788F78CDFD43A445D8991415E' $zipPath = Join-Path $env:RUNNER_TEMP 'PSTools.zip' $toolsDir = Join-Path $env:RUNNER_TEMP 'PSTools' Invoke-WebRequest -Uri 'https://download.sysinternals.com/files/PSTools.zip' -OutFile $zipPath From 1df771d76572fd2d523d8f5d6846d21ca5ee1db9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:49:10 -0400 Subject: [PATCH 16/17] test(jmux): cover connector audit exclusion Exercise complete connector-handled channel lifecycles with an installed traffic callback and assert that neither successful override streams nor connector failures emit events without a resolved target IP. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- crates/jmux-proxy/tests/target_connector.rs | 28 ++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/crates/jmux-proxy/tests/target_connector.rs b/crates/jmux-proxy/tests/target_connector.rs index ea2d5c6c4..9f7f36655 100644 --- a/crates/jmux-proxy/tests/target_connector.rs +++ b/crates/jmux-proxy/tests/target_connector.rs @@ -1,11 +1,13 @@ use std::time::Duration; use jmux_proto::{Bytes, BytesMut, DistantChannelId, Header, LocalChannelId, Message, ReasonCode}; -use jmux_proxy::{DestinationUrl, JmuxConfig, JmuxProxy}; +use jmux_proxy::{DestinationUrl, JmuxConfig, JmuxProxy, TrafficEvent}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::sync::mpsc; use tokio::time::timeout; const TEST_TIMEOUT: Duration = Duration::from_secs(5); +const NO_EVENT_TIMEOUT: Duration = Duration::from_millis(100); async fn send_message(writer: &mut (impl AsyncWrite + Unpin), message: Message) { let mut bytes = BytesMut::new(); @@ -30,13 +32,25 @@ async fn receive_message(reader: &mut (impl AsyncRead + Unpin)) -> Message { .expect("JMUX response timed out") } +async fn assert_no_traffic_event(receiver: &mut mpsc::UnboundedReceiver) { + match timeout(NO_EVENT_TIMEOUT, receiver.recv()).await { + Err(_) => {} + Ok(Some(event)) => panic!("unexpected traffic event: {event:?}"), + Ok(None) => panic!("traffic event callback dropped"), + } +} + #[tokio::test] async fn override_stream_carries_channel_data() { let (proxy_stream, peer_stream) = tokio::io::duplex(8192); let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); + let (traffic_tx, mut traffic_rx) = mpsc::unbounded_channel(); let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) .with_config(JmuxConfig::permissive()) + .with_outgoing_traffic_event_callback(move |event| { + traffic_tx.send(event).expect("capture traffic event"); + }) .with_target_connector_override(|_| async move { let (target_stream, mut target_peer) = tokio::io::duplex(64); tokio::spawn(async move { @@ -72,6 +86,13 @@ async fn override_stream_carries_channel_data() { }; assert_eq!(data.recipient_channel_id, 7); assert_eq!(data.transfer_data, b"ping"[..]); + + let Message::Close(close) = receive_message(&mut peer_reader).await else { + panic!("expected CHANNEL CLOSE"); + }; + assert_eq!(close.recipient_channel_id, 7); + send_message(&mut peer_writer, Message::close(local_id)).await; + assert_no_traffic_event(&mut traffic_rx).await; } #[tokio::test] @@ -79,8 +100,12 @@ async fn resolution_failures_free_id_and_keep_direct_fallback() { let (proxy_stream, peer_stream) = tokio::io::duplex(8192); let (proxy_reader, proxy_writer) = tokio::io::split(proxy_stream); let (mut peer_reader, mut peer_writer) = tokio::io::split(peer_stream); + let (traffic_tx, mut traffic_rx) = mpsc::unbounded_channel(); let proxy = JmuxProxy::new(Box::new(proxy_reader), Box::new(proxy_writer)) .with_config(JmuxConfig::permissive()) + .with_outgoing_traffic_event_callback(move |event| { + traffic_tx.send(event).expect("capture traffic event"); + }) .with_target_connector_override(|destination| async move { if destination.host() == "fail.example" { anyhow::bail!("agent error"); @@ -104,6 +129,7 @@ async fn resolution_failures_free_id_and_keep_direct_fallback() { }; assert_eq!(open_failure.reason_code, ReasonCode::GENERAL_FAILURE); assert_eq!(open_failure.description, "target connection failed"); + assert_no_traffic_event(&mut traffic_rx).await; send_message( &mut peer_writer, From 065dcff11c1acaa0c87b620c3b2113f35649907f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Beno=C3=AEt=20CORTIER?= Date: Thu, 10 Sep 2026 13:57:17 -0400 Subject: [PATCH 17/17] chore: drop unrelated PSTools checksum update Keep the third-party executable pin out of the JMUX feature review. The verified checksum update is moving to a dedicated CI pull request with independent provenance. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4d2952c0e..f6d8729c9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1236,7 +1236,7 @@ jobs: - name: Install PsExec shell: pwsh run: | - $expectedHash = '2B10B3D9DAE0403B06D90B13BFB53E723A8B14A788F78CDFD43A445D8991415E' + $expectedHash = '4F49964CC9CBAC2B5D87BDC8F9526012E9C4B243D8B7D0C0BB51F254A721CA2E' $zipPath = Join-Path $env:RUNNER_TEMP 'PSTools.zip' $toolsDir = Join-Path $env:RUNNER_TEMP 'PSTools' Invoke-WebRequest -Uri 'https://download.sysinternals.com/files/PSTools.zip' -OutFile $zipPath @@ -1290,7 +1290,7 @@ jobs: - name: Install PsExec shell: pwsh run: | - $expectedHash = '2B10B3D9DAE0403B06D90B13BFB53E723A8B14A788F78CDFD43A445D8991415E' + $expectedHash = '4F49964CC9CBAC2B5D87BDC8F9526012E9C4B243D8B7D0C0BB51F254A721CA2E' $zipPath = Join-Path $env:RUNNER_TEMP 'PSTools.zip' $toolsDir = Join-Path $env:RUNNER_TEMP 'PSTools' Invoke-WebRequest -Uri 'https://download.sysinternals.com/files/PSTools.zip' -OutFile $zipPath