Strange behaviour of async SFTP

help
1

Hello Rust community,

I got a strange behaviour when I send files in a SFTP server in async mode: sometimes, one or more files sent are not wrote.

This is my Cargo.toml:

futures = "0.3.15"
async-io = "1.6.0"
async-ssh2-lite = "0.2.1"
clap = "2.33.3"
async-std = "1.9.0"

This is my simple SFTP client in Rust:

use async_io::Async;
use async_ssh2_lite::{AsyncSession, AsyncSftp};
use async_std::fs::read;
use clap::{crate_authors, crate_description, crate_name, crate_version, App, Arg};
use futures::executor::block_on;
use futures::future::join_all;
use futures::AsyncWriteExt;
use std::io;
use std::net::{TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};

async fn send(sftp: &AsyncSftp<TcpStream>, host: &str, input: &str) -> io::Result<()> {
    let basename = PathBuf::from(input).file_name().unwrap().to_os_string();
    let basename = Path::new(basename.as_os_str());
    // Read input content
    let content = read(input).await?;
    let mut f = sftp.create(basename).await?;
    f.write(content.as_ref()).await?;
    f.close().await?;
    println!("[{}] <- {}", host, basename.display());
    Ok(())
}

async fn run(host: &str, inputs: Vec<&str>) -> io::Result<()> {
    let addr = host.to_socket_addrs()?.next().unwrap();
    let stream = Async::<TcpStream>::connect(addr).await?;

    println!("Starting SFTP session at {}", host);
    let mut session = AsyncSession::new(stream, None)?;
    session.handshake().await?;
    // session.userauth_password("user", "password").await?;
    session
        .userauth_pubkey_file(
            "user",
            Some(Path::new("/home/user/.ssh/id_rsa.pub")),
            Path::new("/home/user/.ssh/id_rsa"),
            None,
        )
        .await?;

    let sftp = session.sftp().await?;
    let funcs = inputs.iter().map(|input| send(&sftp, host, *input));
    let result = join_all(funcs).await;

    println!("result = {:?}", result);
    session
        .disconnect(None, "closed by application", None)
        .await?;
    Ok(())
}

fn main() {
    // Arguments and options
    let matches = App::new(crate_name!())
        .version(crate_version!())
        .author(crate_authors!())
        .about(crate_description!())
        .arg(
            Arg::with_name("host")
                .short("h")
                .long("host")
                .default_value("127.0.0.1:22")
                .help("Server address")
                .takes_value(true),
        )
        .arg(Arg::with_name("inputs").required(true).min_values(1))
        .get_matches();

    let hosts = matches.value_of("host").unwrap();
    let inputs: Vec<_> = matches.values_of("inputs").unwrap().collect();
    block_on(run(hosts, inputs)).expect("Error");
}

Inside ~/tmp/to_inject:

$ ls -l ~/tmp/to_inject                                                        
total 20
-rw-r--r--. 1 user user 2 Jul 23 09:58 file_1
-rw-r--r--. 1 user user 3 Jul 23 09:58 file_2
-rw-r--r--. 1 user user 4 Jul 23 09:58 file_3
-rw-r--r--. 1 user user 5 Jul 23 09:58 file_4
-rw-r--r--. 1 user user 6 Jul 23 09:58 file_5

I run cargo run -- -h 127.0.0.1:3373 ~/tmp/to_inject/*:

Starting SFTP session at 127.0.0.1:3373
[127.0.0.1:3373] <- file_1
[127.0.0.1:3373] <- file_2
[127.0.0.1:3373] <- file_3
[127.0.0.1:3373] <- file_4
[127.0.0.1:3373] <- file_5
result = [Ok(()), Ok(()), Ok(()), Ok(()), Ok(())]

At the destination folder, sometimes (nearly 1 time over 5), not all files are present:

-rw-r--r--. 1 user user 4 Jul 23 12:09 file_3
-rw-r--r--. 1 user user 5 Jul 23 12:09 file_4
-rw-r--r--. 1 user user 6 Jul 23 12:09 file_5

This is the log of the Python sftpserver but I got the same behaviour with SSHd:

$ sftpserver --level=DEBUG --port=3373 --keyfile=/home/user/.ssh/id_rsa
DEBUG:paramiko.transport:starting thread (server mode): 0x617e8580
DEBUG:paramiko.transport:Local version/idstring: SSH-2.0-paramiko_2.7.2
DEBUG:paramiko.transport:Remote version/idstring: SSH-2.0-libssh2_1.9.0_DEV
INFO:paramiko.transport:Connected (version 2.0, client libssh2_1.9.0_DEV)
DEBUG:paramiko.transport:kex algos:['ecdh-sha2-nistp256', 'ecdh-sha2-nistp384', 'ecdh-sha2-nistp521', 'diffie-hellman-group-exchange-sha256', 'diffie-hellman-group16-sha512', 'diffie-hellman-group18-sha512', 'diffie-hellman-group14-sha256', 'diffie-hellman-group14-sha1', 'diffie-hellman-group1-sha1', 'diffie-hellman-group-exchange-sha1'] server key:['ecdsa-sha2-nistp256', 'ecdsa-sha2-nistp384', 'ecdsa-sha2-nistp521', 'ssh-rsa', 'ssh-dss'] client encrypt:['aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes256-cbc', 'rijndael-cbc@lysator.liu.se', 'aes192-cbc', 'aes128-cbc', 'blowfish-cbc', 'arcfour128', 'arcfour', 'cast128-cbc', '3des-cbc'] server encrypt:['aes128-ctr', 'aes192-ctr', 'aes256-ctr', 'aes256-cbc', 'rijndael-cbc@lysator.liu.se', 'aes192-cbc', 'aes128-cbc', 'blowfish-cbc', 'arcfour128', 'arcfour', 'cast128-cbc', '3des-cbc'] client mac:['hmac-sha2-256', 'hmac-sha2-512', 'hmac-sha1', 'hmac-sha1-96', 'hmac-md5', 'hmac-md5-96', 'hmac-ripemd160', 'hmac-ripemd160@openssh.com'] server mac:['hmac-sha2-256', 'hmac-sha2-512', 'hmac-sha1', 'hmac-sha1-96', 'hmac-md5', 'hmac-md5-96', 'hmac-ripemd160', 'hmac-ripemd160@openssh.com'] client compress:['none'] server compress:['none'] client lang:[''] server lang:[''] kex follows?False
DEBUG:paramiko.transport:Kex agreed: ecdh-sha2-nistp256
DEBUG:paramiko.transport:HostKey agreed: ssh-rsa
DEBUG:paramiko.transport:Cipher agreed: aes128-ctr
DEBUG:paramiko.transport:MAC agreed: hmac-sha2-256
DEBUG:paramiko.transport:Compression agreed: none
DEBUG:paramiko.transport:kex engine KexNistp256 specified hash_algo <built-in function openssl_sha256>
DEBUG:paramiko.transport:Switch to new keys ...
DEBUG:paramiko.transport:Auth request (type=publickey) service=ssh-connection, username=user
DEBUG:paramiko.transport:Auth request (type=publickey) service=ssh-connection, username=user
INFO:paramiko.transport:Auth granted (publickey).
DEBUG:paramiko.transport:[chan 0] Max packet in: 32768 bytes
DEBUG:paramiko.transport:[chan 0] Max packet out: 32768 bytes
DEBUG:paramiko.transport:Secsh channel 0 (session) opened.
DEBUG:paramiko.transport:Starting handler for subsystem sftp
DEBUG:paramiko.transport.sftp:[chan 0] Started sftp server on channel <paramiko.Channel 0 (open) window=2097152 -> <paramiko.Transport at 0x617e8580 (cipher aes128-ctr, 128 bits) (active; 1 open channel(s))>>
DEBUG:paramiko.transport.sftp:[chan 0] Request: open
DEBUG:paramiko.transport.sftp:[chan 0] Request: write
DEBUG:paramiko.transport.sftp:[chan 0] Request: open
DEBUG:paramiko.transport.sftp:[chan 0] Request: close
DEBUG:paramiko.transport.sftp:[chan 0] Request: write
DEBUG:paramiko.transport.sftp:[chan 0] Request: open
DEBUG:paramiko.transport.sftp:[chan 0] Request: close
DEBUG:paramiko.transport.sftp:[chan 0] Request: write
DEBUG:paramiko.transport.sftp:[chan 0] Request: open
DEBUG:paramiko.transport.sftp:[chan 0] Request: close
DEBUG:paramiko.transport.sftp:[chan 0] Request: write
DEBUG:paramiko.transport.sftp:[chan 0] Request: open
DEBUG:paramiko.transport.sftp:[chan 0] Request: close
DEBUG:paramiko.transport.sftp:[chan 0] Request: write
DEBUG:paramiko.transport.sftp:[chan 0] Request: close
INFO:paramiko.transport:Disconnect (code 11): closed by application
DEBUG:paramiko.transport.sftp:[chan 0] EOF -- end of session

In this log, we can see 5 open, write and close.

The log of pyinotify in the destination folder:

$ pyinotify -v .                                                                                                                                        
[2021-07-23 12:09:16,250 pyinotify DEBUG] Start monitoring ['.'], (press c^c to halt pyinotify)
[2021-07-23 12:09:16,251 pyinotify DEBUG] New <Watch wd=1 path=. mask=4095 proc_fun=None auto_add=None exclude_filter=<function <lambda> at 0x128cb18> dir=True >
[2021-07-23 12:09:23,454 pyinotify DEBUG] Event queue size: 64
[2021-07-23 12:09:23,454 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x100 name=file_4 wd=1 >
[2021-07-23 12:09:23,454 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x20 name=file_4 wd=1 >
<Event dir=False mask=0x100 maskname=IN_CREATE name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
<Event dir=False mask=0x20 maskname=IN_OPEN name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
[2021-07-23 12:09:23,455 pyinotify DEBUG] Event queue size: 32
[2021-07-23 12:09:23,455 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x2 name=file_4 wd=1 >
<Event dir=False mask=0x2 maskname=IN_MODIFY name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
[2021-07-23 12:09:23,456 pyinotify DEBUG] Event queue size: 64
[2021-07-23 12:09:23,456 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x100 name=file_3 wd=1 >
[2021-07-23 12:09:23,456 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x20 name=file_3 wd=1 >
<Event dir=False mask=0x100 maskname=IN_CREATE name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
<Event dir=False mask=0x20 maskname=IN_OPEN name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
[2021-07-23 12:09:23,457 pyinotify DEBUG] Event queue size: 32
[2021-07-23 12:09:23,457 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x8 name=file_4 wd=1 >
<Event dir=False mask=0x8 maskname=IN_CLOSE_WRITE name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
[2021-07-23 12:09:23,458 pyinotify DEBUG] Event queue size: 64
[2021-07-23 12:09:23,458 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x2 name=file_3 wd=1 >
[2021-07-23 12:09:23,458 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x20 name=file_3 wd=1 >
<Event dir=False mask=0x2 maskname=IN_MODIFY name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
<Event dir=False mask=0x20 maskname=IN_OPEN name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
[2021-07-23 12:09:23,459 pyinotify DEBUG] Event queue size: 128
[2021-07-23 12:09:23,459 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x8 name=file_3 wd=1 >
[2021-07-23 12:09:23,459 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x2 name=file_3 wd=1 >
[2021-07-23 12:09:23,459 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x2 name=file_4 wd=1 >
[2021-07-23 12:09:23,460 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x20 name=file_4 wd=1 >
<Event dir=False mask=0x8 maskname=IN_CLOSE_WRITE name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
<Event dir=False mask=0x2 maskname=IN_MODIFY name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
<Event dir=False mask=0x2 maskname=IN_MODIFY name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
<Event dir=False mask=0x20 maskname=IN_OPEN name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
[2021-07-23 12:09:23,461 pyinotify DEBUG] Event queue size: 160
[2021-07-23 12:09:23,461 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x8 name=file_3 wd=1 >
[2021-07-23 12:09:23,461 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x2 name=file_4 wd=1 >
[2021-07-23 12:09:23,462 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x100 name=file_5 wd=1 >
[2021-07-23 12:09:23,462 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x20 name=file_5 wd=1 >
[2021-07-23 12:09:23,462 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x8 name=file_4 wd=1 >
<Event dir=False mask=0x8 maskname=IN_CLOSE_WRITE name=file_3 path=. pathname=/home/user/tmp/dir/file_3 wd=1 >
<Event dir=False mask=0x2 maskname=IN_MODIFY name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
<Event dir=False mask=0x100 maskname=IN_CREATE name=file_5 path=. pathname=/home/user/tmp/dir/file_5 wd=1 >
<Event dir=False mask=0x20 maskname=IN_OPEN name=file_5 path=. pathname=/home/user/tmp/dir/file_5 wd=1 >
<Event dir=False mask=0x8 maskname=IN_CLOSE_WRITE name=file_4 path=. pathname=/home/user/tmp/dir/file_4 wd=1 >
[2021-07-23 12:09:23,463 pyinotify DEBUG] Event queue size: 64
[2021-07-23 12:09:23,463 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x2 name=file_5 wd=1 >
[2021-07-23 12:09:23,463 pyinotify DEBUG] <_RawEvent cookie=0 mask=0x8 name=file_5 wd=1 >
<Event dir=False mask=0x2 maskname=IN_MODIFY name=file_5 path=. pathname=/home/user/tmp/dir/file_5 wd=1 >
<Event dir=False mask=0x8 maskname=IN_CLOSE_WRITE name=file_5 path=. pathname=/home/user/tmp/dir/file_5 wd=1 >

There's only the trace of 3 files.

Any advice ? Is the problem from my Rust code or elsewhere ?

Thank you in advance.

2

I move the SFTP session opening (session.sftp) from the run to the send function and it working fine now.

Is the right thing to do ?

3

I suspect you have found a bug in the async_ssh2_lite crate.

I notice that its dependency ssh2 uses blocking mutexes throughout its SFTP implementation. Perhaps this is causing deadlocks if a lock is held across an await point in an async function. It makes sense that you could work around this bug by creating a separate session for each file, since the lock is per-session.

There is also a known bug in async_ssh2_lite that can cause deadlocks, but I don't know if this one is relevant to your use case.

2 Likes
4

Thank you @mbrubeck for your analyze. I open an issue.

6

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.