spebern / async-ssh2 Goto Github PK

View Code? Open in Web Editor NEW

43.0 43.0 10.0 81 KB

Rust 97.46% Shell 2.54%

async-ssh2's People

Contributors

Stargazers

Watchers

Forkers

ubnt-intrepid oleid 0xdeafbeef timbotambo spacemeowx2 xudesheng provost-iac ousado ayourtch pbar1

async-ssh2's Issues

Build fails

error: failed to select a version for the requirement `ssh2 = "^0.7"`
  candidate versions found which didn't match: 0.6.0, 0.5.0, 0.4.0, ...
  location searched: crates.io index

How to obtain the exit_signal structure?

Hello,

I'm running a remote command and I'm wondering how to obtain the exit_signal and error_message when calling channel.exit_signal().

Would you have any examples?

libssh2 multiple channels concurrent issue

I had created an issue in the ssh2-rs repo. Mirror it here for more guys to know it.

I'm trying to implement a remote port forwarding proxy using async-ssh2. The core idea can be expressed like bellow code:

use async_ssh2::{Channel, Session};
use futures_util::io::AsyncReadExt;
use smol::io::copy;
use smol::Async;
use std::net::SocketAddr;
use std::net::TcpStream;

async fn do_proxy(tcp_conn: Async<TcpStream>, ch: Channel) -> Result<(), anyhow::Error> {
    let (ch_rh, ch_wh) = ch.split();
    let (conn_rh, conn_wh) = tcp_conn.split();
    smol::future::or(copy(ch_rh, conn_wh), copy(conn_rh, ch_wh))
        .await
        .map_err(|e| {
            println!("copy error: {:?}", e.raw_os_error());
        });
    Ok(())
}

Then I find concurrent connections to the remote port cause ssh2 channel read error at the copy point, and "Channel not found" error at the accept point.

After some investigation, I find a mail about the limitation of libssh2 multiple channels concurrent issue: https://www.libssh2.org/mail/libssh2-devel-archive-2014-06/0028.shtml

Maybe I need to ask in that mail thread directly. But before that, I would like to know are there anyone else have tried ssh2-rs remote port forwarding in a concurrent way?

session.set_tcp_stream

async-ssh2::Session::set_tcp_stream
Why not to use tokio::net::TcpStream instead of std::net::TcpStream?

Blocking on stream.read_to_string()

Looks like there is no way to read stream.

channel
        .stream(0)
        .read_to_string(&mut channel_buffer)
        .map_err(|e| Error::msg(format!("Error reading result of work: {}", e)))?;

"Error reading result of work: would block"

Q: How to forward data to a local tcp server?

Hey there,

to access a host behind a ssh server, I need to to SSH forwarding (or set a proxy connection in my ssh config). Naturally, I can do the forwarding manually. But it would be nice to be able to do it automatically. Thus, I tried something like:

async fn forward_connection(sess: &mut async_ssh2::Session, w: &str, port: u16) -> Result<(), Box<dyn std::error::Error>>
{
    use futures::join;

    let listener = async_std::net::TcpListener::bind("localhost:0").await?;

    let mut incoming = listener.incoming();

    while let Some(stream) = incoming.next().await {
        let stream = stream?;
        let ssh = sess.channel_direct_tcpip(w, port, None).await?;

        let (tcp_rx, tcp_tx) = &mut (&stream, &stream);
        let (ssh_rx, ssh_tx, ) =&mut (ssh, &ssh);

        let a_to_b = async_std::io::copy(tcp_rx, ssh_tx);
        let b_to_a = async_std::io::copy(ssh_rx, tcp_tx);

        join!(a_to_b, b_to_a).await
    }
    Ok(())
}

This is a modified version of async_std's TcpListener example (https://docs.rs/async-std/1.6.0-beta.1/async_std/net/struct.TcpListener.html). Yet, it doesn't work :

error[E0277]: the trait bound `&async_ssh2::channel::Channel: futures_io::if_std::AsyncWrite` is not satisfied
   --> src/main.rs:120:50
    |
120 |         let a_to_b = async_std::io::copy(tcp_rx, ssh_tx);
    |                                                  ^^^^^^ the trait `futures_io::if_std::AsyncWrite` is not implemented for `&async_ssh2::channel::Channel`
    |

What would be the recommended way to split a channel to forward it to a socket (unix or tcp)?

I have a feeling that it might be not possible at the moment. Or I missed something.

Or related - what would be the recommended way to dynamically access SSH servers behind another ssh server?

Enhance Session to support Jumpbox

Please don't deprecate this crate. This one is more suitable for me, compared with async-ssh2-lite.

Can I ask for an enhancement to support ssh via Jumpbox? This crate is already async and supports tokio, it should be straightforward to add a new API in Session, may be like: async fn proxy_session(&self, target_ip:&str, target_port:u16, src:Option<(&str,u16)>)->Result<Session>;

async-ssh2-lite has an example: https://github.com/bk-rs/async-ssh2-lite/blob/master/demos/smol/src/proxy_jump.rs

I tested it with async-ssh2 and it works. I hope it can be built in this crate.

use std::net::{TcpListener, TcpStream, ToSocketAddrs};
use std::path::Path;
use anyhow::Result;
use async_io::Async;
use async_ssh2::Session;
use futures::future::FutureExt;
use futures::select;
use futures::{AsyncReadExt, AsyncWriteExt};


#[tokio::main]
async fn main()->Result<()>{
    let private_key = "output\\dxu-script";
    let private_key = Path::new(private_key);
    let public_key = "output\\dxu-script.pub";
    let public_key = Path::new(public_key);
    

    let bastion_addr = "20.85.xxx.xxx:22".to_socket_addrs().unwrap().next().unwrap();
    let stream = Async::<TcpStream>::connect(bastion_addr).await?;
    let mut sess = Session::new()?;
    sess.set_tcp_stream(stream)?;
    sess.handshake().await?;

    sess.userauth_pubkey_file("dxu",Some(public_key),private_key,None).await?;
    let mut channel = sess.channel_session().await?;
    let cmd = "ls -l";
    channel.exec(cmd).await?;
    let mut s = String::new();
    channel.read_to_string(&mut s).await?;
    println!("{}",s);
    channel.wait_close().await?;
    println!("{}",channel.exit_status()?);

    //10.133.1.5 is the real remote target vm
    let host = "10.133.1.5:22".to_socket_addrs().unwrap().next().unwrap();

    //10.133.1.12 is the private IP of the bastion machine.
    let mut bastion_channel = sess.channel_direct_tcpip(
        host.ip().to_string().as_ref(), host.port(), Some( ("10.133.1.12",22))).await?;
    
    let (forward_stream_s,mut forward_stream_r) = {
        cfg_if::cfg_if! {
            if #[cfg(unix)] {
                let dir = tempdir()?;
                let path = dir.path().join("ssh_channel_direct_tcpip");
                let listener = Async::<UnixListener>::bind(&path)?;
                let stream_s = Async::<UnixStream>::accept(&path).await?;
            }else{
                let listen_addr = TcpListener::bind("localhost:0").unwrap().local_addr().unwrap();
                let listener = Async::<TcpListener>::bind(listen_addr)?;
                let stream_s = Async::<TcpStream>::connect(listen_addr).await?;
            }
        }

        let (stream_r, _) = listener.accept().await?;
        (stream_s, stream_r)
    };

    let backend_task = tokio::spawn(async move {
        let mut buf_bastion_channel = vec![0; 2048];
        let mut buf_forward_stream_r = vec![0; 2048];
        
        loop {
            select! {
                ret_forward_stream_r = forward_stream_r.read(&mut buf_forward_stream_r).fuse() => match ret_forward_stream_r{
                    Ok(n) if n == 0 => {
                        println!("forward_stream_r closed.");
                        break;
                    },
                    Ok(n) => {
                        bastion_channel.write_all(&buf_forward_stream_r[..n]).await.map(|_| ()).map_err(|err| {
                            eprintln!("bastion_channel write failed, err: {:?}",err);
                            err
                        })?;
                    },
                    Err(err) => {
                        eprintln!("forward_stream_r read failed, err: {:?}",err);
                        return Err(err);
                    }
                },
                ret_bastion_channel = bastion_channel.read(&mut buf_bastion_channel).fuse() => match ret_bastion_channel {
                    Ok(n) if n == 0 => {
                        println!("bastion_channel closed.");
                        break;
                    },
                    Ok(n) => {
                        forward_stream_r.write_all(&buf_bastion_channel[..n]).await.map(|_| ()).map_err(|err| {
                            eprintln!("forward_stream_s write failed, err: {:?}",err);
                            err
                        })?;
                    },
                    Err(err) => {
                        eprintln!("bastion_channel read failed, err: {:?}",err);
                        return Err(err);
                    }
                },
            }
        }
        
        println!("byebye.");
        Ok(())
    });

    let mut child_sess = Session::new()?;
    child_sess.set_tcp_stream(forward_stream_s)?;
    child_sess.handshake().await?;
    child_sess.userauth_pubkey_file("dxu",Some(public_key),private_key,None).await?;
    let mut child_channel = child_sess.channel_session().await?;
    let cmd = "ls -l /";
    child_channel.exec(cmd).await?;
    let mut s = String::new();
    child_channel.read_to_string(&mut s).await?;
    println!("{}",s);
    child_channel.wait_close().await?;
    println!("{}",child_channel.exit_status()?);

    child_sess.disconnect(None, "fun",None).await?;
    let ret = backend_task.await?;
    println!("{:?}",ret);

    Ok(())
}

Listener::accept() tight loops

I've been porting one of my projects from ssh2-rs to async-ssh2. This has mostly been fine, and the async syntax is much nicer, but I've found that performance is much worse with async-ssh2 (I hit 100% CPU). I've done some investigation and I think there is a bug in this library that is causing it.

Basically, the problem is that all the "asyncness" in this library is associated with the session's underlying TCP stream. But there are instances where operations can return WOULDBLOCK when the underlying TCP stream is unblocked. One example of this is Listener::accept(). This function returns WOULDBLOCK when the TCP Listener on the remote server is blocked on accept() which has nothing to do with the TCP stream for this session.

Then when smol::Async::with() is called, what happens is:

The operation returns WOULDBLOCK.
The function waits for the session TCP stream to be readable or writeable. This is true immediately since it's the wrong I/O object.
The function tight loops and eats CPU.

I created a simple example to show this happening - see https://github.com/sathiyan-sivathas/ssh2-rs-bench. The ssh2 version uses <1% CPU. The async-ssh2 version uses pretty much 100% CPU.

Build fails

Build fails on rustc 1.40.0 (73528e339 2019-12-16)

error[E0432]: unresolved import `ssh2::BlockDirections`
  --> src/lib.rs:18:5
   |
18 |     BlockDirections, ExitSignal, FileStat, FileType, Host, KnownHostFileKind, KnownHosts,
   |     ^^^^^^^^^^^^^^^ no `BlockDirections` in the root

error[E0726]: implicit elided lifetime not allowed here
  --> src/agent.rs:46:61
   |
46 |     pub async fn userauth(&self, username: &str, identity: &PublicKey) -> Result<(), Error> {
   |                                                             ^^^^^^^^^- help: indicate the anonymous lifetime: `<'_>`

error[E0106]: missing lifetime specifier
  --> src/sftp.rs:22:12
   |
22 |     inner: ssh2::File,
   |            ^^^^^^^^^^ expected lifetime parameter

error[E0599]: no method named `clone` found for type `std::option::Option<ssh2::PtyModes>` in the current scope
  --> src/channel.rs:42:75
   |
42 |         into_the_future!(aio; &mut || { self.inner.request_pty(term, mode.clone(), dim) })
   |                                                                           ^^^^^ method not found in `std::option::Option<ssh2::PtyModes>`
   |
   = note: the method `clone` exists but the following trait bounds were not satisfied:
           `std::option::Option<ssh2::PtyModes> : std::clone::Clone`

error[E0599]: no method named `shutdown` found for type `ssh2::sftp::Sftp` in the current scope
   --> src/sftp.rs:159:52
    |
159 |         into_the_future!(aio; &mut || { self.inner.shutdown() })
    |                                                    ^^^^^^^^ method not found in `ssh2::sftp::Sftp`

error[E0599]: no method named `close` found for type `ssh2::sftp::File<'static>` in the current scope
   --> src/sftp.rs:205:52
    |
205 |         into_the_future!(aio; &mut || { self.inner.close() })
    |                                                    ^^^^^ method not found in `ssh2::sftp::File<'static>`

error: aborting due to 6 previous errors

Rationale for deprecation?

I saw that this package was deprecated a few days ago in favor of async-ssh2-lite. Since I am using async-ssh2 I thought I had better check the other package out - but I found it distinctly under-cooked.

Some examples:

The AsyncFile implementation merely implements AsyncRead/AsyncWrite, but does not expose any of the methods of the underlying libssh2 file, such as stat, readdir, etc.
The sftp readdir method does not work at all for me - it either hangs or returns an empty list. When I inspected the source, it just calls the inner readdir within the Async write_with callback - I can't see how that is ever going to work.
In fact almost all the methods do something similar, even though many of them involve writing a command to the ssh connection, then waiting for a response which has to be read. The 'lite' package seems simplistically to assume that once a command can be written, the response will be available for reading.

At least for me this package seems a lot more capable than the one you are deprecating it in favor of, so I am wondering what the rationale for deprecation is?

(I was also confused as to why/how the commit which deprecated this package came from some user called 'bold' who is not a contributor and I can't even click on their name to see who they are ... that might just be my lack of experience with github though).

Thread blocks on sess.handshake()

runing such code leads to such output:

[src\main.rs:67] format!("Tcp connection intialized {}", & hostname) = "Tcp connection intialized 10.36.78.43:22"
[src\main.rs:74] format!("Session is built {}", & hostname) = "Session is built 10.36.78.43:22"
[src\main.rs:78] format!("Tcp stream is set : {}", & hostname) = "Tcp stream is set : 10.36.78.43:22"

Maybe i'm bad cooking async code
And waiting doesn't help at all.
Thats all packets, which are sent to the host.

LIBSSH2_AGENT doesn't implement Send

error[E0277]: `*mut libssh2_sys::LIBSSH2_AGENT` cannot be sent between threads safely
   --> src/main.rs:311:17
    |
311 |         runtime.spawn(process_host(&host, &command, tx.clone(), num_of_threads));
    |                 ^^^^^ `*mut libssh2_sys::LIBSSH2_AGENT` cannot be sent between threads safely
    |
    = help: within `impl std::future::Future`, the trait `std::marker::Send` is not implemented for `*mut libssh2_sys::LIBSSH2_AGENT`
    = note: required because it appears within the type `ssh2::agent::Agent`
    = note: required because it appears within the type `async_ssh2::agent::Agent`
    = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14, 't15, 't16, 't17, 't18, 't19, 't20, 't21, 't22, 't23, 't24, 't25, 't26, 't27, 't28, 't29, 't30, 't31> {&std::string::String, &'r str, std::sync::mpsc::SyncSender<Response>, tokio::sync::semaphore::Semaphore, &'s tokio::sync::semaphore::Semaphore, tokio::sync::semaphore::Semaphore, impl std::future::Future, impl std::future::Future, (), std::time::Instant, std::net::TcpStream, async_ssh2::session::Session, &'t4 mut async_ssh2::session::Session, async_ssh2::session::Session, impl std::future::Future, impl std::future::Future, (), async_ssh2::agent::Agent, &'t7 mut async_ssh2::agent::Agent, async_ssh2::agent::Agent, impl std::future::Future, impl std::future::Future, (), &'t10 async_ssh2::session::Session, async_ssh2::session::Session, &'t11 str, &'t12 str, impl std::future::Future, impl std::future::Future, (), &'t17 async_ssh2::session::Session, async_ssh2::session::Session, impl std::future::Future, impl std::future::Future, (), async_ssh2::channel::Channel, &'t20 mut async_ssh2::channel::Channel, async_ssh2::channel::Channel, &'t21 str, &'t22 str, impl std::future::Future, impl std::future::Future, (), std::string::String, &'t27 mut async_ssh2::channel::Channel, async_ssh2::channel::Channel, std::string::String, &'t28 mut std::string::String, &'t29 mut std::string::String, tokio::io::util::read_to_string::ReadToString<'t30, async_ssh2::channel::Channel>, tokio::io::util::read_to_string::ReadToString<'t31, async_ssh2::channel::Channel>, ()}`
    = note: required because it appears within the type `[static generator@src/main.rs:58:1: 126:2 hostname:&std::string::String, command:&str, tx:std::sync::mpsc::SyncSender<Response>, connection_pool:tokio::sync::semaphore::Semaphore for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14, 't15, 't16, 't17, 't18, 't19, 't20, 't21, 't22, 't23, 't24, 't25, 't26, 't27, 't28, 't29, 't30, 't31> {&std::string::String, &'r str, std::sync::mpsc::SyncSender<Response>, tokio::sync::semaphore::Semaphore, &'s tokio::sync::semaphore::Semaphore, tokio::sync::semaphore::Semaphore, impl std::future::Future, impl std::future::Future, (), std::time::Instant, std::net::TcpStream, async_ssh2::session::Session, &'t4 mut async_ssh2::session::Session, async_ssh2::session::Session, impl std::future::Future, impl std::future::Future, (), async_ssh2::agent::Agent, &'t7 mut async_ssh2::agent::Agent, async_ssh2::agent::Agent, impl std::future::Future, impl std::future::Future, (), &'t10 async_ssh2::session::Session, async_ssh2::session::Session, &'t11 str, &'t12 str, impl std::future::Future, impl std::future::Future, (), &'t17 async_ssh2::session::Session, async_ssh2::session::Session, impl std::future::Future, impl std::future::Future, (), async_ssh2::channel::Channel, &'t20 mut async_ssh2::channel::Channel, async_ssh2::channel::Channel, &'t21 str, &'t22 str, impl std::future::Future, impl std::future::Future, (), std::string::String, &'t27 mut async_ssh2::channel::Channel, async_ssh2::channel::Channel, std::string::String, &'t28 mut std::string::String, &'t29 mut std::string::String, tokio::io::util::read_to_string::ReadToString<'t30, async_ssh2::channel::Channel>, tokio::io::util::read_to_string::ReadToString<'t31, async_ssh2::channel::Channel>, ()}]`
    = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:58:1: 126:2 hostname:&std::string::String, command:&str, tx:std::sync::mpsc::SyncSender<Response>, connection_pool:tokio::sync::semaphore::Semaphore for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11, 't12, 't13, 't14, 't15, 't16, 't17, 't18, 't19, 't20, 't21, 't22, 't23, 't24, 't25, 't26, 't27, 't28, 't29, 't30, 't31> {&std::string::String, &'r str, std::sync::mpsc::SyncSender<Response>, tokio::sync::semaphore::Semaphore, &'s tokio::sync::semaphore::Semaphore, tokio::sync::semaphore::Semaphore, impl std::future::Future, impl std::future::Future, (), std::time::Instant, std::net::TcpStream, async_ssh2::session::Session, &'t4 mut async_ssh2::session::Session, async_ssh2::session::Session, impl std::future::Future, impl std::future::Future, (), async_ssh2::agent::Agent, &'t7 mut async_ssh2::agent::Agent, async_ssh2::agent::Agent, impl std::future::Future, impl std::future::Future, (), &'t10 async_ssh2::session::Session, async_ssh2::session::Session, &'t11 str, &'t12 str, impl std::future::Future, impl std::future::Future, (), &'t17 async_ssh2::session::Session, async_ssh2::session::Session, impl std::future::Future, impl std::future::Future, (), async_ssh2::channel::Channel, &'t20 mut async_ssh2::channel::Channel, async_ssh2::channel::Channel, &'t21 str, &'t22 str, impl std::future::Future, impl std::future::Future, (), std::string::String, &'t27 mut async_ssh2::channel::Channel, async_ssh2::channel::Channel, std::string::String, &'t28 mut std::string::String, &'t29 mut std::string::String, tokio::io::util::read_to_string::ReadToString<'t30, async_ssh2::channel::Channel>, tokio::io::util::read_to_string::ReadToString<'t31, async_ssh2::channel::Channel>, ()}]>`
    = note: required because it appears within the type `impl std::future::Future`
    = note: required because it appears within the type `impl std::future::Future`

Segfault in Session::userauth_agent()

I cannot authenticate a session without getting a segfault here. This code is from a branch of rftp where I'm trying to use async_ssh2 and using the original ssh2 works fine.

Here is my backtrace

Thread 1 "rftp" received signal SIGSEGV, Segmentation fault.
0x0000555555713ee4 in agent_sign (session=0x555555b4a1e0, sig=0x7fffffff56d0, 
    sig_len=0x7fffffff56d8, data=0x555555b9eb60 "", data_len=628, 
    abstract=0x7fffffff5758) at libssh2/src/agent.c:387
387	    ssize_t len = 1 + 4 + identity->external.blob_len + 4 + data_len + 4;
(gdb) bt
#0  0x0000555555713ee4 in agent_sign (session=0x555555b4a1e0, sig=0x7fffffff56d0, 
    sig_len=0x7fffffff56d8, data=0x555555b9eb60 "", data_len=628, 
    abstract=0x7fffffff5758) at libssh2/src/agent.c:387
#1  0x0000555555731357 in _libssh2_userauth_publickey (session=0x555555b4a1e0, 
    username=0x555555b58760 "ellis", username_len=5, pubkeydata=0x555555b58400 "", 
    pubkeydata_len=535, sign_callback=0x555555713e8e <agent_sign>, 
    abstract=0x7fffffff5758) at libssh2/src/userauth.c:1298
#2  0x0000555555714afe in libssh2_agent_userauth (agent=0x555555b46e10, 
    username=0x555555b58760 "ellis", identity=0x555555b48758)
    at libssh2/src/agent.c:782
#3  0x00005555556fbe24 in ssh2::agent::Agent::userauth (self=0x7fffffff5b30, 
    username=..., identity=0x555555b59df0)
    at /home/ellis/.cargo/git/checkouts/ssh2-rs-ef60f55ca3a50813/5c5bf32/src/agent.rs:117
#4  0x00005555556f3f29 in ssh2::session::Session::userauth_agent (
    self=0x7fffffffc0e8, username=...)
    at /home/ellis/.cargo/git/checkouts/ssh2-rs-ef60f55ca3a50813/5c5bf32/src/session.rs:455
#5  0x00005555555e142c in async_ssh2::session::Session::userauth_agent::{{closure}}::{{closure}} ()
    at /home/ellis/.cargo/git/checkouts/async-ssh2-036534fec3c85a4f/9f6ce5d/src/session.rs:129
...

waking up concurrent reads on the same session

I don't know for sure if this is a problem in reality, but in thinking about this problem space in wezterm, I think it might be.

In wezterm we create a session to talk to the remote host with a channel at initial connection. Then the user can spawn additional tabs, each one is a separate channel with its own pty, but all multiplexed on the same session.

The situation I'm unsure of is if we have 2 or more channels on the same session that are waiting for read. When the underlying fd becomes ready, does the tokio plumbing cause both read futures to be polled, or only one of them? If the latter then I think we'll need some additional logic to fan out the wakeup.