Channel mixup when concurrently opening multiple direct-tcpip channels

[kyrias]

When setting up and using multiple direct-tcpip channels concurrently there seem to be some kind of bug where the the different channels end up being confused and the wrong one used.

Reproduction steps

Server

On the server:

$ mkdir a b
$ echo -n a >a/index.html
$ echo -n b >b/index.html
$ python3 -m http.server 8001 --directory a &
$ python3 -m http.server 8002 --directory b &

Now when we issue requests at http://localhost:8001 we should expect a single a in response and b for :8002.

Client

On the client side set up a project with the following dependencies and code:

async-ssh2-lite = { version = "0.4.7", features = ["tokio"] }
bytes = "1.6.1"
http-body-util = "0.1.2"
hyper = { version = "1.4.1", features = ["client", "http1"] }
hyper-util = { version = "0.1.6", features = ["tokio"] }
tokio = { version = "1.38.1", features = ["macros", "rt", "rt-multi-thread"] }

Client code

use std::{net::Ipv4Addr, path::PathBuf, str::FromStr, sync::Arc, time::Duration};

use async_ssh2_lite::AsyncSession;
use bytes::Bytes;
use http_body_util::{BodyExt, Empty};
use hyper::Uri;
use hyper_util::rt::TokioIo;
use tokio::{
    net::{TcpStream, UnixListener, UnixStream},
    time::sleep,
};

#[tokio::main]
async fn main() {
    let mut session =
        AsyncSession::<TcpStream>::connect((Ipv4Addr::from_str("<REMOTE-IP>").unwrap(), 22), None)
            .await
            .unwrap();
    session.handshake().await.unwrap();

    let private_key = PathBuf::from("/correct/path/to/key");
    session
        .userauth_pubkey_file("<REMOTE-USER>", None, &private_key, None)
        .await
        .unwrap();

    let session = Arc::new(session);
    spawn_uds_listener(Arc::clone(&session), "/tmp/a.sock", 8001);
    spawn_uds_listener(Arc::clone(&session), "/tmp/b.sock", 8002);
    sleep(Duration::from_secs(1)).await;

    loop {
        println!();
        tokio::join!(
            send_request("/tmp/a.sock", "A"),
            send_request("/tmp/a.sock", "A"),
            send_request("/tmp/b.sock", "B"),
        );
        sleep(Duration::from_secs(5)).await
    }
}

fn spawn_uds_listener(session: Arc<AsyncSession<TcpStream>>, socket: &'static str, port: u16) {
    tokio::spawn({
        async move {
            std::fs::remove_file(socket).ok();
            let listener = UnixListener::bind(socket).unwrap();
            loop {
                let (mut local_stream, _addr) = listener.accept().await.unwrap();

                let mut remote_channel = session
                    .channel_direct_tcpip("127.0.0.1", port, None)
                    .await
                    .unwrap();

                tokio::spawn(async move {
                    if let Err(err) =
                        tokio::io::copy_bidirectional(&mut remote_channel, &mut local_stream).await
                    {
                        eprintln!(
                        "Copying data between Unix domain socket A and SSH tunnel failed: {err:?}"
                    );
                    }
                });
            }
        }
    });
}

async fn send_request(socket: &'static str, name: &'static str) {
    let stream = UnixStream::connect(socket).await.unwrap();
    let (mut sender, conn) =
        hyper::client::conn::http1::handshake::<_, Empty<Bytes>>(TokioIo::new(stream))
            .await
            .unwrap();
    tokio::task::spawn(async move {
        if let Err(err) = conn.await {
            panic!("{name} connection failed: {err:?}");
        }
    });

    let request = hyper::Request::builder()
        .method(hyper::Method::GET)
        .uri(Uri::from_static("/"))
        .body(Empty::<Bytes>::new())
        .unwrap();
    let response = sender.send_request(request).await.unwrap();
    let body = response.collect().await.unwrap().to_bytes();
    let body = String::from_utf8_lossy(&body);
    eprintln!("{name}: {body}");
}

Expected

You get a stream of output where each A line has a response of a and each B line has a response of b.

Actual

A: a
A: b
B: b

A: a
A: b
B: b

B: a
A: a
A: a

[vkill]

Hi @kyrias
In this scenario, I recommend establishing multiple connections/sessions.