Skip to content

Commit c1f8141

Browse files
EugenyEugeny
authored
fixed #686 - make channel confirmations truly async (#723)
1 parent 713fc34 commit c1f8141

20 files changed

Lines changed: 556 additions & 311 deletions

‎russh/examples/echoserver.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -80,13 +80,15 @@ impl server::Handler for Server {
8080
async fn channel_open_session(
8181
&mut self,
8282
channel: Channel<Msg>,
83+
reply: server::ChannelOpenHandle,
8384
session: &mut Session,
84-
) -> Result<bool, Self::Error> {
85+
) -> Result<(), Self::Error> {
8586
{
8687
let mut clients = self.clients.lock().await;
8788
clients.insert(self.id, (channel.id(), session.handle()));
8889
}
89-
Ok(true)
90+
reply.accept().await;
91+
Ok(())
9092
}
9193

9294
async fn auth_publickey(

‎russh/examples/ratatui_app.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -147,8 +147,9 @@ impl Handler for AppServer {
147147
async fn channel_open_session(
148148
&mut self,
149149
channel: Channel<Msg>,
150+
reply: ChannelOpenHandle,
150151
session: &mut Session,
151-
) -> Result<bool, Self::Error> {
152+
) -> Result<(), Self::Error> {
152153
let terminal_handle = TerminalHandle::start(session.handle(), channel.id()).await;
153154

154155
let backend = CrosstermBackend::new(terminal_handle);
@@ -164,7 +165,8 @@ impl Handler for AppServer {
164165
let mut clients = self.clients.lock().await;
165166
clients.insert(self.id, (terminal, app));
166167

167-
Ok(true)
168+
reply.accept().await;
169+
Ok(())
168170
}
169171

170172
async fn auth_publickey(&mut self, _: &str, _: &PublicKey) -> Result<Auth, Self::Error> {

‎russh/examples/ratatui_shared_app.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -149,8 +149,9 @@ impl Handler for AppServer {
149149
async fn channel_open_session(
150150
&mut self,
151151
channel: Channel<Msg>,
152+
reply: ChannelOpenHandle,
152153
session: &mut Session,
153-
) -> Result<bool, Self::Error> {
154+
) -> Result<(), Self::Error> {
154155
let terminal_handle = TerminalHandle::start(session.handle(), channel.id()).await;
155156

156157
let backend = CrosstermBackend::new(terminal_handle);
@@ -165,7 +166,8 @@ impl Handler for AppServer {
165166
let mut clients = self.clients.lock().await;
166167
clients.insert(self.id, terminal);
167168

168-
Ok(true)
169+
reply.accept().await;
170+
Ok(())
169171
}
170172

171173
async fn auth_publickey(&mut self, _: &str, _: &PublicKey) -> Result<Auth, Self::Error> {

‎russh/examples/sftp_server.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,13 +59,15 @@ impl russh::server::Handler for SshSession {
5959
async fn channel_open_session(
6060
&mut self,
6161
channel: Channel<Msg>,
62+
reply: russh::server::ChannelOpenHandle,
6263
_session: &mut Session,
63-
) -> Result<bool, Self::Error> {
64+
) -> Result<(), Self::Error> {
6465
{
6566
let mut clients = self.clients.lock().await;
6667
clients.insert(channel.id(), channel);
6768
}
68-
Ok(true)
69+
reply.accept().await;
70+
Ok(())
6971
}
7072

7173
async fn channel_eof(

‎russh/examples/test.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,14 +51,16 @@ impl server::Handler for Server {
5151
async fn channel_open_session(
5252
&mut self,
5353
channel: Channel<Msg>,
54+
reply: server::ChannelOpenHandle,
5455
_session: &mut Session,
55-
) -> Result<bool, Self::Error> {
56+
) -> Result<(), Self::Error> {
5657
{
5758
debug!("channel open session");
5859
let mut clients = self.clients.lock().unwrap();
5960
clients.insert((self.id, channel.id()), channel);
6061
}
61-
Ok(true)
62+
reply.accept().await;
63+
Ok(())
6264
}
6365

6466
/// The client requests a shell.

‎russh/src/client/encrypted.rs‎

Lines changed: 121 additions & 137 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ use ssh_key::Algorithm;
2424
use super::IncomingSshPacket;
2525
use crate::auth::AuthRequest;
2626
use crate::cert::PublicKeyOrCertificate;
27-
use crate::client::{Handler, Msg, Prompt, Reply, Session};
27+
use crate::client::{ChannelOpenHandle, Handler, Msg, Prompt, Reply, Session};
2828
use crate::helpers::{AlgorithmExt, EncodedExt, NameList, sign_with_hash_alg};
2929
use crate::keys::key::parse_public_key;
3030
use crate::parsing::{ChannelOpenConfirmation, ChannelType, OpenChannelMessage, ensure_end};
@@ -432,7 +432,7 @@ impl Session {
432432
debug!("channel_open_failure");
433433
let channel_num = map_err!(ChannelId::decode(&mut r))?;
434434
let reason_code = ChannelOpenFailure::from_u32(map_err!(u32::decode(&mut r))?)
435-
.unwrap_or(ChannelOpenFailure::Unknown);
435+
.unwrap_or(ChannelOpenFailure::AdministrativelyProhibited);
436436
let descr = map_err!(String::decode(&mut r))?;
437437
let language = map_err!(String::decode(&mut r))?;
438438
map_err!(ensure_end(&r))?;
@@ -441,7 +441,7 @@ impl Session {
441441
}
442442

443443
if let Some(sender) = self.channels.remove(&channel_num) {
444-
let _ = sender.send(ChannelMsg::OpenFailure(reason_code)).await;
444+
let _ = sender.send(ChannelMsg::OpenFailure(reason_code.clone())).await;
445445
}
446446

447447
let _ = self.sender.send(Reply::ChannelOpenFailure);
@@ -677,127 +677,129 @@ impl Session {
677677
Some((&msg::CHANNEL_OPEN, mut r)) => {
678678
let msg = OpenChannelMessage::parse(&mut r)?;
679679

680-
if let Some(ref mut enc) = self.common.encrypted {
681-
let id = enc.new_channel_id();
682-
let channel = ChannelParams {
683-
recipient_channel: msg.recipient_channel,
684-
sender_channel: id,
685-
recipient_window_size: msg.recipient_window_size,
686-
sender_window_size: self.common.config.window_size,
687-
recipient_maximum_packet_size: msg.recipient_maximum_packet_size,
688-
sender_maximum_packet_size: self.common.config.maximum_packet_size,
689-
confirmed: true,
690-
wants_reply: false,
691-
pending_data: std::collections::VecDeque::new(),
692-
pending_eof: false,
693-
pending_close: false,
694-
};
680+
let id = if let Some(ref mut enc) = self.common.encrypted {
681+
enc.new_channel_id()
682+
} else {
683+
return Err(crate::Error::Inconsistent.into());
684+
};
695685

696-
let confirm = || {
697-
debug!("confirming channel: {msg:?}");
698-
map_err!(msg.confirm(
699-
&mut enc.write,
700-
id.0,
701-
channel.sender_window_size,
702-
channel.sender_maximum_packet_size,
703-
))?;
704-
enc.channels.insert(id, channel);
705-
Ok(())
706-
};
686+
let channel_params = ChannelParams {
687+
recipient_channel: msg.recipient_channel,
688+
sender_channel: id,
689+
recipient_window_size: msg.recipient_window_size,
690+
sender_window_size: self.common.config.window_size,
691+
recipient_maximum_packet_size: msg.recipient_maximum_packet_size,
692+
sender_maximum_packet_size: self.common.config.maximum_packet_size,
693+
confirmed: true,
694+
wants_reply: false,
695+
pending_data: std::collections::VecDeque::new(),
696+
pending_eof: false,
697+
pending_close: false,
698+
};
707699

708-
match &msg.typ {
709-
ChannelType::Session => {
710-
confirm()?;
711-
let channel = self.accept_server_initiated_channel(id, &msg);
712-
client.server_channel_open_session(channel, self).await?
713-
}
714-
ChannelType::DirectTcpip(d) => {
715-
confirm()?;
716-
let channel = self.accept_server_initiated_channel(id, &msg);
717-
client
718-
.server_channel_open_direct_tcpip(
719-
channel,
720-
&d.host_to_connect,
721-
d.port_to_connect,
722-
&d.originator_address,
723-
d.originator_port,
724-
self,
725-
)
726-
.await?
727-
}
728-
ChannelType::DirectStreamLocal(d) => {
729-
confirm()?;
730-
let channel = self.accept_server_initiated_channel(id, &msg);
731-
client
732-
.server_channel_open_direct_streamlocal(
733-
channel,
734-
&d.socket_path,
735-
self,
736-
)
737-
.await?
738-
}
739-
ChannelType::X11 {
740-
originator_address,
741-
originator_port,
742-
} => {
743-
confirm()?;
744-
let channel = self.accept_server_initiated_channel(id, &msg);
745-
client
746-
.server_channel_open_x11(
747-
channel,
748-
originator_address,
749-
*originator_port,
750-
self,
751-
)
752-
.await?
753-
}
754-
ChannelType::ForwardedTcpIp(d) => {
755-
confirm()?;
756-
let channel = self.accept_server_initiated_channel(id, &msg);
757-
client
758-
.server_channel_open_forwarded_tcpip(
759-
channel,
760-
&d.host_to_connect,
761-
d.port_to_connect,
762-
&d.originator_address,
763-
d.originator_port,
764-
self,
765-
)
766-
.await?
767-
}
768-
ChannelType::ForwardedStreamLocal(d) => {
769-
confirm()?;
770-
let channel = self.accept_server_initiated_channel(id, &msg);
771-
client
772-
.server_channel_open_forwarded_streamlocal(
773-
channel,
774-
&d.socket_path,
775-
self,
776-
)
777-
.await?;
778-
}
779-
ChannelType::AgentForward => {
780-
confirm()?;
781-
let channel = self.accept_server_initiated_channel(id, &msg);
782-
client
783-
.server_channel_open_agent_forward(channel, self)
784-
.await?
785-
}
786-
ChannelType::Unknown { typ } => {
787-
if client.should_accept_unknown_server_channel(id, typ).await {
788-
confirm()?;
789-
let channel = self.accept_server_initiated_channel(id, &msg);
790-
client.server_channel_open_unknown(channel, self).await?;
791-
} else {
792-
debug!("unknown channel type: {typ}");
700+
let (channel, channel_ref) = Channel::new(
701+
id,
702+
self.inbound_channel_sender.clone(),
703+
channel_params.recipient_maximum_packet_size,
704+
channel_params.recipient_window_size,
705+
self.common.config.channel_buffer_size,
706+
);
707+
708+
let pending = crate::PendingChannelOpen {
709+
recipient_channel: msg.recipient_channel,
710+
sender_channel: id,
711+
window_size: self.common.config.window_size,
712+
packet_size: self.common.config.maximum_packet_size,
713+
channel_ref,
714+
channel_params,
715+
};
716+
let reply = ChannelOpenHandle::new(
717+
self.inbound_channel_sender.clone(),
718+
pending,
719+
|pending, result| Msg::ServerChannelOpenReply { pending, result },
720+
);
721+
722+
match &msg.typ {
723+
ChannelType::Session => {
724+
client.server_channel_open_session(channel, reply, self).await?
725+
}
726+
ChannelType::DirectTcpip(d) => {
727+
client
728+
.server_channel_open_direct_tcpip(
729+
channel,
730+
&d.host_to_connect,
731+
d.port_to_connect,
732+
&d.originator_address,
733+
d.originator_port,
734+
reply,
735+
self,
736+
)
737+
.await?
738+
}
739+
ChannelType::DirectStreamLocal(d) => {
740+
client
741+
.server_channel_open_direct_streamlocal(
742+
channel,
743+
&d.socket_path,
744+
reply,
745+
self,
746+
)
747+
.await?
748+
}
749+
ChannelType::X11 {
750+
originator_address,
751+
originator_port,
752+
} => {
753+
client
754+
.server_channel_open_x11(
755+
channel,
756+
originator_address,
757+
*originator_port,
758+
reply,
759+
self,
760+
)
761+
.await?
762+
}
763+
ChannelType::ForwardedTcpIp(d) => {
764+
client
765+
.server_channel_open_forwarded_tcpip(
766+
channel,
767+
&d.host_to_connect,
768+
d.port_to_connect,
769+
&d.originator_address,
770+
d.originator_port,
771+
reply,
772+
self,
773+
)
774+
.await?
775+
}
776+
ChannelType::ForwardedStreamLocal(d) => {
777+
client
778+
.server_channel_open_forwarded_streamlocal(
779+
channel,
780+
&d.socket_path,
781+
reply,
782+
self,
783+
)
784+
.await?
785+
}
786+
ChannelType::AgentForward => {
787+
client
788+
.server_channel_open_agent_forward(channel, reply, self)
789+
.await?
790+
}
791+
ChannelType::Unknown { typ } => {
792+
if client.should_accept_unknown_server_channel(id, typ).await {
793+
client.server_channel_open_unknown(channel, reply, self).await?;
794+
} else {
795+
debug!("unknown channel type: {typ}");
796+
if let Some(ref mut enc) = self.common.encrypted {
793797
msg.unknown_type(&mut enc.write)?;
794798
}
795799
}
796-
};
797-
Ok(())
798-
} else {
799-
Err(crate::Error::Inconsistent.into())
800-
}
800+
}
801+
};
802+
Ok(())
801803
}
802804
Some((&msg::REQUEST_SUCCESS, mut r)) => {
803805
trace!("Global Request Success");
@@ -894,24 +896,6 @@ impl Session {
894896
}
895897
}
896898

897-
fn accept_server_initiated_channel(
898-
&mut self,
899-
id: ChannelId,
900-
msg: &OpenChannelMessage,
901-
) -> Channel<Msg> {
902-
let (channel, channel_ref) = Channel::new(
903-
id,
904-
self.inbound_channel_sender.clone(),
905-
msg.recipient_maximum_packet_size,
906-
msg.recipient_window_size,
907-
self.common.config.channel_buffer_size,
908-
);
909-
910-
self.channels.insert(id, channel_ref);
911-
912-
channel
913-
}
914-
915899
pub(crate) fn write_auth_request_if_needed(
916900
&mut self,
917901
user: &str,

0 commit comments

Comments
 (0)