Example showing how to implement shell_request -> Command -> stdin/stdout/stderr? #215
Comments
|
Ah, I just needed to use // Handling stdout and stderr data
let session_handle = session.handle();
tokio::spawn(async move {
loop {
tokio::select! {
Some(data) = stdout_rx.recv() => {
let _ =session_handle.data(channel_id, CryptoVec::from_slice(&data)).await;
},
Some(data) = stderr_rx.recv() => {
let _ = session_handle.data(channel_id, CryptoVec::from_slice(&data)).await;
},
else => break,
}
}
});
// mark request success
session.request_success();However, it doesn't quite work how I'd expect... still working through it. |
|
I'm trying to do the same thing! Still working on figuring it out... |
|
I think the problem may be that /bin/bash is not running in "repl mode" (or whatever it's called), and so it's not outputting anything to stdout or stderr. I tried replacing /bin/bash with 
There's this issue where stdout characters need to be mapped to terminal commands, which puts the cursor in a weird place. But o/w it seems to work! |
|
The missing part is the PTY. When running from an interactive terminal, the OpenSSH client by default will ask your server for a PTY (Handler::request_pty) and will expect you to return PTY output - see On the bash side, it needs a PTY to run in the "interactive mode". You can create one with the |
|
The OpenSSH client receiving raw (non-PTY) output after having requested the server to run the command in a PTY is what's causing this cursor behaviour. |
|
I've considered the same issue in Warpgate actually. The main problem here is that most Rust TUI crates don't support running in an "emulated" PTY - which has to be the case if you want to keep it all in the same process (PTYs can only be created via In comparison, having a separate TUI process running in a real PTY is much simpler and is guaranteed to work - but obviously uses more resources. |
|
Gotcha, thanks! I opened ratatui/ratatui#643 for discussion as to whether ratatui can handle an emulated PTY. In the meantime, sounds like a separate process is the way to go for now. Perhaps it's wishful thinking but would you happen to have a code example for that by any chance? I'm esp. unclear as to how to get the window dimensions aligned between the child process and the SSH client. |
|
I don't have an example handy, but you really just need to use the The SSH client will track window size changes and report it to the server, which you'll see as |
|
I'm currently working on the PTY solution, and planning on pursuing the in-process solution next. PTY solution is currently blocked on doy/pty-process#7 (comment) and pkgw/stund#305. |
|
Ok I made some progress and now have something that can output command to a PTY which then pipes things to the SSH client. However, I haven't yet figured out a way to get stdin to work. For some reason (bug?), russh is not propagating key press events via either I've put my latest progress into a gist. The key part being // Watch the shell Channel for data from the client and send it to the pty
// NOTE: I'm reading off of the Channel, but it's not clear if this is the correct thing to do in order to receive
// data from the client.
let channel_streams_arc = Arc::clone(&self.channel_streams);
tokio::spawn(async move {
let mut channel_streams = channel_streams_arc.lock().await;
let mut stream = channel_streams.get_mut(&channel_id).unwrap().lock().await;
let mut buffer = vec![0; 1024];
while let Ok(n) = stream.read(&mut buffer).await {
panic!("this never happens :(")
// TODO:
// if n == 0 {
// break;
// }
// pty.write_all(&buffer[..n]).unwrap();
}
});but sadly we're not hitting that @Eugeny am I missing something here? Is this a bug? |
|
Do you get data only after the Enter keypress? Then the TUI app hasn't switched the terminal to raw mode. Also, the client transmits its initial terminal mode in |
I'm not sure I understand the question 100%, but I am not receiving data even after pressing Enter on the client. I am fairly confident that the client terminal is getting set to raw mode since after killing the server I have no cursor and have to run
Yeah, I've been meaning to get around to implementing that, but could that affect this behavior? The server doesn't seem to be receiving any events at all. |
|
No, that could prevent input from being sent immediately but it would still arrive after an Enter key. Do you see the input packets arriving when running with Running the SSH client with |
|
I see it sending characters (key presses) without I did it slightly differently than @samuela Cargo.tomlpty-process = { git = "https://github.com/mobusoperandi/pty-process.git", branch = "macos_draft_pr", features = ["async"] }I used async/non-blocking, he used blocking. I'm not sure if this still makes sense with I'm also not sure if like Sam, being on MacOS makes this a non-starter since we can't accurately use But, like Sam, even when I see the network traffic: I don't see any output from the The ssh client doesn't log anything when I type something, but russh use std::collections::HashMap;
use std::process::Stdio;
use std::sync::Arc;
use async_trait::async_trait;
use log::*;
use russh::server::{Auth, Msg, Session, Response};
use russh::*;
use russh_keys::*;
use tokio::io::{BufReader, AsyncReadExt, AsyncWriteExt};
use tokio::process::{Child, Command, ChildStdin};
use tokio::sync::{mpsc, Mutex};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
env_logger::init();
let mut config = russh::server::Config::default();
config.auth_rejection_time = std::time::Duration::from_secs(3);
config
.keys
.push(russh_keys::key::KeyPair::generate_ed25519().unwrap());
let config = Arc::new(config);
let sh = Server {
clients: Arc::new(Mutex::new(HashMap::new())),
shell_stdins: Arc::new(Mutex::new(HashMap::new())),
id: 0,
};
russh::server::run(config, ("0.0.0.0", 2222), sh).await?;
Ok(())
}
#[derive(Clone)]
struct Server {
clients: Arc<Mutex<HashMap<(usize, ChannelId), Channel<Msg>>>>,
shell_stdins: Arc<Mutex<HashMap<ChannelId, ChildStdin>>>,
id: usize,
}
impl server::Server for Server {
type Handler = Self;
fn new_client(&mut self, _: Option<std::net::SocketAddr>) -> Self {
debug!("new client");
let s = self.clone();
self.id += 1;
s
}
}
#[async_trait]
impl server::Handler for Server {
type Error = anyhow::Error;
async fn channel_open_session(
self,
channel: Channel<Msg>,
session: Session,
) -> Result<(Self, bool, Session), Self::Error> {
{
debug!("channel open session");
let mut clients = self.clients.lock().await;
clients.insert((self.id, channel.id()), channel);
}
Ok((self, true, session))
}
/// The client requests a shell.
#[allow(unused_variables)]
async fn shell_request(
self,
channel_id: ChannelId,
mut session: Session,
) -> Result<(Self, Session), Self::Error> {
debug!("shell_request");
let pty = pty_process::Pty::new().unwrap();
if let Err(e) = pty.resize(pty_process::Size::new(24, 80)) {
// See https://github.com/doy/pty-process/issues/7#issuecomment-1826196215.
log::error!("pty.resize failed: {:?}", e);
}
// Spawn a new /bin/bash process
let mut child = pty_process::Command::new("/bin/bash")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn(&pty.pts().unwrap())
.map_err(anyhow::Error::new)?;
let stderr = child.stderr.take().unwrap();
let stdout = child.stdout.take().unwrap();
let stdin = child.stdin.take().unwrap();
// Store the child process handle
let mut shell_stdins = self.shell_stdins.lock().await;
shell_stdins.insert(channel_id, stdin);
drop(shell_stdins);
// Create channels for stdout and stderr
let (stdout_tx, mut stdout_rx) = mpsc::channel(1024);
let (stderr_tx, mut stderr_rx) = mpsc::channel(1024);
// Spawn task for stdout
tokio::spawn(async move {
let mut stdout = BufReader::new(stdout);
let mut buffer = vec![0; 1024];
while let Ok(size) = stdout.read(&mut buffer).await {
if size == 0 { break; }
log::info!("stdout: {:02x?}", &buffer[..size]);
if stdout_tx.send(buffer[..size].to_vec()).await.is_err() {
break; // Receiver has dropped
}
}
});
// Spawn task for stderr
tokio::spawn(async move {
let mut stderr = BufReader::new(stderr);
let mut buffer = vec![0; 1024];
while let Ok(size) = stderr.read(&mut buffer).await {
if size == 0 { break; }
log::info!("stderr: {:02x?}", &buffer[..size]);
if stderr_tx.send(buffer[..size].to_vec()).await.is_err() {
break; // Receiver has dropped
}
}
});
// Handling stdout and stderr data
let session_handle = session.handle();
tokio::spawn(async move {
loop {
tokio::select! {
Some(data) = stdout_rx.recv() => {
log::info!("stdout_rx: {data:02x?}");
let _ = session_handle.data(channel_id, CryptoVec::from_slice(&data)).await;
},
Some(data) = stderr_rx.recv() => {
log::info!("stderr_rx: {data:02x?}");
let _ = session_handle.data(channel_id, CryptoVec::from_slice(&data)).await;
},
else => break,
}
}
});
// mark request success
session.request_success();
Ok((self, session))
}
#[allow(unused_variables)]
async fn auth_publickey(
self,
user: &str,
public_key: &key::PublicKey,
) -> Result<(Self, Auth), Self::Error> {
debug!("auth_publickey: user: {user} public_key: {public_key:?}");
Ok((self, server::Auth::Accept))
}
#[allow(unused_variables)]
async fn auth_keyboard_interactive(
self,
user: &str,
submethods: &str,
response: Option<Response<'async_trait>>,
) -> Result<(Self, Auth), Self::Error> {
debug!("auth_keyboard_interactive: user: {user}");
Ok((
self,
Auth::Reject {
proceed_with_methods: Some(MethodSet::PUBLICKEY | MethodSet::PASSWORD),
},
))
}
#[allow(unused_variables)]
async fn auth_none(self, user: &str) -> Result<(Self, Auth), Self::Error> {
debug!("auth_none: user: {user}");
Ok((
self,
Auth::Reject {
proceed_with_methods: Some(MethodSet::PUBLICKEY | MethodSet::PASSWORD),
},
))
}
async fn auth_password(self, user: &str, password: &str) -> Result<(Self, Auth), Self::Error> {
debug!("auth_password: credentials: {}, {}", user, password);
Ok((self, Auth::Accept))
}
async fn data(
self,
channel_id: ChannelId,
data: &[u8],
session: Session,
) -> Result<(Self, Session), Self::Error> {
// pipe to shell stdin
let mut shell_stdins = self.shell_stdins.lock().await;
if let Some(stdin) = shell_stdins.get_mut(&channel_id) {
debug!("stdin: {data:02x?}");
stdin.write_all(data).await.map_err(anyhow::Error::new).unwrap();
}
drop(shell_stdins);
Ok((self, session))
}
} |
|
Edit: I think I got it working... use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;
use log::*;
use pty_process::OwnedWritePty;
use russh::server::{Auth, Msg, Session, Response};
use russh::*;
use russh_keys::*;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::Mutex;
#[derive(Clone)]
struct Server {
clients: Arc<Mutex<HashMap<(usize, ChannelId), Channel<Msg>>>>,
channel_pty_writers: Arc<Mutex<HashMap<ChannelId, OwnedWritePty>>>,
id: usize,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
env_logger::init();
let mut config = russh::server::Config::default();
config.auth_rejection_time = std::time::Duration::from_secs(3);
config
.keys
.push(russh_keys::key::KeyPair::generate_ed25519().unwrap());
let config = Arc::new(config);
let sh = Server {
clients: Arc::new(Mutex::new(HashMap::new())),
channel_pty_writers: Arc::new(Mutex::new(HashMap::new())),
id: 0,
};
russh::server::run(config, ("0.0.0.0", 2222), sh).await?;
Ok(())
}
impl server::Server for Server {
type Handler = Self;
fn new_client(&mut self, _: Option<std::net::SocketAddr>) -> Self {
debug!("new client");
let s = self.clone();
self.id += 1;
s
}
}
#[async_trait]
impl server::Handler for Server {
type Error = anyhow::Error;
async fn channel_open_session(
self,
channel: Channel<Msg>,
session: Session,
) -> Result<(Self, bool, Session), Self::Error> {
{
debug!("channel open session");
let mut clients = self.clients.lock().await;
clients.insert((self.id, channel.id()), channel);
}
Ok((self, true, session))
}
/// The client requests a shell.
#[allow(unused_variables)]
async fn shell_request(
self,
channel_id: ChannelId,
mut session: Session,
) -> Result<(Self, Session), Self::Error> {
debug!("shell_request");
// create pty
let pty = pty_process::Pty::new().unwrap();
if let Err(e) = pty.resize(pty_process::Size::new(24, 80)) {
log::error!("pty.resize failed: {:?}", e);
}
// get pts from pty
let pts = pty.pts()?;
// split pty into reader + writer
let (mut pty_reader, pty_writer) = pty.into_split();
// insert pty_reader
self.channel_pty_writers.lock().await.insert(channel_id, pty_writer);
// pty_reader -> session_handle
let session_handle = session.handle();
tokio::spawn(async move {
let mut buffer = vec![0; 1024];
while let Ok(size) = pty_reader.read(&mut buffer).await {
if size == 0 { break; }
let _ = session_handle.data(channel_id, CryptoVec::from_slice(&buffer[0..size])).await;
}
});
// Spawn a new /bin/bash process in pty
let child = pty_process::Command::new("/bin/bash")
.spawn(&pts)
.map_err(anyhow::Error::new)?;
// mark request success
session.request_success();
Ok((self, session))
}
#[allow(unused_variables)]
async fn auth_publickey(
self,
user: &str,
public_key: &key::PublicKey,
) -> Result<(Self, Auth), Self::Error> {
debug!("auth_publickey: user: {user} public_key: {public_key:?}");
Ok((self, server::Auth::Accept))
}
#[allow(unused_variables)]
async fn auth_keyboard_interactive(
self,
user: &str,
submethods: &str,
response: Option<Response<'async_trait>>,
) -> Result<(Self, Auth), Self::Error> {
debug!("auth_keyboard_interactive: user: {user}");
Ok((
self,
Auth::Reject {
proceed_with_methods: Some(MethodSet::PUBLICKEY | MethodSet::PASSWORD),
},
))
}
#[allow(unused_variables)]
async fn auth_none(self, user: &str) -> Result<(Self, Auth), Self::Error> {
debug!("auth_none: user: {user}");
Ok((
self,
Auth::Reject {
proceed_with_methods: Some(MethodSet::PUBLICKEY | MethodSet::PASSWORD),
},
))
}
async fn auth_password(self, user: &str, password: &str) -> Result<(Self, Auth), Self::Error> {
debug!("auth_password: credentials: {}, {}", user, password);
Ok((self, Auth::Accept))
}
async fn data(
self,
channel_id: ChannelId,
data: &[u8],
session: Session,
) -> Result<(Self, Session), Self::Error> {
// session -> pty_writer
let mut channel_pty_writers = self.channel_pty_writers.lock().await;
if let Some(pty_writer) = channel_pty_writers.get_mut(&channel_id) {
log::info!("pty_writer: data = {data:02x?}");
pty_writer.write_all(data).await.map_err(anyhow::Error::new).unwrap();
}
drop(channel_pty_writers);
Ok((self, session))
}
}@samuela can you confirm I'm not crazy/it's working for you too? |
|
Awesome stuff, @brandonros ! I made a few changes to suit my preferences/needs and ended up with this. Notable changes:
|
|
are we at a point where we can replace Dropbear yet for a portable sshd on Windows? what are next best steps, pull your code into my PR? i might be overstating demand for “rust ssh servers” but this functionality seems pretty cool! |
I don't know what Dropbear is, but this sounds cool! :)
totally agree! i think this could even be a fun blog post or something. i imagine a lot of people might be curious about implementing their own SSH server Btw, one thing I am still unsure about is what happens to the processes after a client disconnects. Ie, let's say I connect to the server, the server spins up a PTY and |
https://matt.ucc.asn.au/dropbear/dropbear.html
I've just never had any luck with this: https://learn.microsoft.com/en-us/windows-server/administration/openssh/openssh_install_firstuse?tabs=gui If we break this out from an example to its own repo and let it become a
Good callout. On client close we need to close the PTY and kill the process. Where would that be added in your gist: https://gist.github.com/samuela/a7b20be5019b359e358457bfcc4a3df1 ? Want to make a repo called I'm also curious how |
This is a separate protocol feature and needs to be handled by the server ( |
|
@samuela do you want to PR your gist into https://github.com/brandonros/rustbear and add yourself as a codeowner/maintainer or whatever to give credit? Thanks Eugeny for your help as well. If you don't want to merge this example into the repo that's fine. I'll close this for now I guess. |
|
I want an example like that, I just didn't get to looking at it yet and cleaning it up to not use the mac-specific pty-process crate |
|
Hey @brandonros and @Eugeny, thanks so much for all your help and collaboration so far!
Thank you! That's very generous of you, @brandonros. As it happens, I ended up needing to take this a slightly different direction and published my work as https://github.com/samuela/sshenanigans. rustbear looks quite exciting as well, esp. for folks on Windows! |
This isn't right because
shell_requestneeds to return and can't loop, but it needs to returnsession, and you need to callsession.data()to TX to the client when the shell has output from stdout/stderr.Not sure if I am missing something?
The text was updated successfully, but these errors were encountered: