Bound SDK integration test waits

src/bin/dosh-client.rs

@@ -4476,7 +4476,7 @@ fn latest_release_tag(repo: &str) -> Result<Option<String>> {
     }
     let effective = String::from_utf8_lossy(&output.stdout);
     let effective = effective.trim();
-    Ok(release_tag_from_effective_url(web, &effective))
+    Ok(release_tag_from_effective_url(web, effective))
 }
 
 fn release_tag_download_url(repo: &str, tag: &str, artifact: &str) -> Option<String> {
@@ -5523,12 +5523,12 @@ fn resolve_forward_agent_endpoint(forward_agent: bool) -> Result<Option<PathBuf>
 
     #[cfg(unix)]
     {
-        return match std::env::var_os("SSH_AUTH_SOCK") {
+        match std::env::var_os("SSH_AUTH_SOCK") {
             Some(path) if !path.is_empty() => Ok(Some(PathBuf::from(path))),
             _ => Err(anyhow!(
                 "agent forwarding requested but SSH_AUTH_SOCK is not set"
             )),
-        };
+        }
     }
 
     #[cfg(windows)]
@@ -6150,10 +6150,10 @@ fn local_username() -> String {
     local_username_from_env(|name| std::env::var(name).ok())
 }
 
-fn local_username_from_env(mut get: impl FnMut(&str) -> Option<String>) -> String {
+fn local_username_from_env(get: impl FnMut(&str) -> Option<String>) -> String {
     ["USER", "USERNAME"]
         .into_iter()
-        .filter_map(|name| get(name))
+        .filter_map(get)
         .find(|value| !value.is_empty())
         .unwrap_or_else(|| "unknown".to_string())
 }

src/client.rs

@@ -201,7 +201,6 @@ impl DoshClientBuilder {
             match connect_sdk_peer(
                 peer_addr,
                 &self.client.config,
-                &host_config,
                 &self.host,
                 &raw_server,
                 ssh_port,
@@ -244,7 +243,6 @@ impl ConnectedDoshClient {
 async fn connect_sdk_peer(
     peer_addr: SocketAddr,
     config: &ClientConfig,
-    host_config: &HostConfig,
     host: &str,
     server: &str,
     ssh_port: Option<u16>,
@@ -311,18 +309,17 @@ async fn connect_sdk_peer(
         &hello,
         &server_hello.hello,
     )?;
-    let auth = sign_auth(
+    let auth = sign_auth(NativeAuthSignRequest {
         config,
-        host_config,
+        hello: &hello,
-        &hello,
+        server_hello: &server_hello.hello,
-        &server_hello.hello,
+        requested_forwardings: requested_forwardings.to_vec(),
-        requested_forwardings.to_vec(),
+        explicit_identity_files: identity_files.to_vec(),
-        identity_files.to_vec(),
         use_ssh_agent,
         server,
         ssh_port,
         ssh_config,
-    )?;
+    })?;
     let mut pending_id = [0u8; 16];
     pending_id.copy_from_slice(&server_hello.hello.auth_challenge[..16]);
     let auth_packet = protocol::encode_encrypted(
@@ -387,18 +384,30 @@ fn verify_or_trust_host(
     }
 }
 
-fn sign_auth(
+struct NativeAuthSignRequest<'a> {
-    config: &ClientConfig,
+    config: &'a ClientConfig,
-    _host_config: &HostConfig,
+    hello: &'a NativeClientHello,
-    hello: &NativeClientHello,
+    server_hello: &'a native::NativeServerHello,
-    server_hello: &native::NativeServerHello,
     requested_forwardings: Vec<ForwardingRequest>,
     explicit_identity_files: Vec<PathBuf>,
     use_ssh_agent: Option<bool>,
-    server: &str,
+    server: &'a str,
     ssh_port: Option<u16>,
-    ssh_config: &SdkSshConfig,
+    ssh_config: &'a SdkSshConfig,
-) -> Result<native::NativeUserAuth> {
+}
+
+fn sign_auth(request: NativeAuthSignRequest<'_>) -> Result<native::NativeUserAuth> {
+    let NativeAuthSignRequest {
+        config,
+        hello,
+        server_hello,
+        requested_forwardings,
+        explicit_identity_files,
+        use_ssh_agent,
+        server,
+        ssh_port,
+        ssh_config,
+    } = request;
     let use_agent = use_ssh_agent.unwrap_or(config.use_ssh_agent);
     let mut errors = Vec::new();
     if use_agent && !ssh_config.identities_only {
@@ -448,12 +457,6 @@ fn sdk_identity_paths(
     for path in explicit_identity_files {
         push_identity_path(&mut paths, path);
     }
-    for path in &ssh_config.identity_files {
-        push_identity_path(
-            &mut paths,
-            expand_tilde(&expand_ssh_path_tokens(path, token_context)),
-        );
-    }
     if !ssh_config.identities_only {
         for path in &config.identity_files {
             push_identity_path(
@@ -462,6 +465,12 @@ fn sdk_identity_paths(
             );
         }
     }
+    for path in &ssh_config.identity_files {
+        push_identity_path(
+            &mut paths,
+            expand_tilde(&expand_ssh_path_tokens(path, token_context)),
+        );
+    }
     paths
 }
 
@@ -818,10 +827,10 @@ fn local_username() -> Option<String> {
     local_username_from_env(|name| std::env::var(name).ok())
 }
 
-fn local_username_from_env(mut get: impl FnMut(&str) -> Option<String>) -> Option<String> {
+fn local_username_from_env(get: impl FnMut(&str) -> Option<String>) -> Option<String> {
     ["USER", "USERNAME"]
         .into_iter()
-        .filter_map(|name| get(name))
+        .filter_map(get)
         .find(|value| !value.is_empty())
 }
 
@@ -1092,7 +1101,7 @@ mod tests {
 
         let paths = sdk_identity_paths(&config, Vec::new(), &ssh_config, &token_context);
 
-        assert_eq!(paths, vec![dir.path().join("ssh"), config_identity]);
+        assert_eq!(paths, vec![config_identity, dir.path().join("ssh")]);
     }
 
     #[test]

src/server.rs

@@ -578,20 +578,43 @@ mod tests {
             .user("sdk-user")
             .service("echo")
             .connect();
-        let accept = async {
+        let handshake = async {
+            tokio::pin!(connect);
+            let mut connected = None;
+            let mut accepted = None;
             loop {
-                if let DoshServerEvent::Accepted(accepted) = server.recv().await.unwrap() {
+                tokio::select! {
-                    break accepted;
+                    result = &mut connect, if connected.is_none() => {
+                        connected = Some(result?);
+                    }
+                    event = server.recv(), if accepted.is_none() => {
+                        if let DoshServerEvent::Accepted(connection) = event? {
+                            accepted = Some(connection);
+                        }
+                    }
+                }
+                if connected.is_some() && accepted.is_some() {
+                    return Ok::<_, anyhow::Error>((
+                        connected.take().expect("connected result checked"),
+                        accepted.take().expect("accepted result checked"),
+                    ));
                 }
             }
         };
-        let (connected, accepted) = tokio::join!(connect, accept);
+        let (connected, accepted) = tokio::time::timeout(Duration::from_secs(3), handshake)
-        let mut client_transport = connected.unwrap().into_transport();
+            .await
+            .expect("SDK client/server authentication timed out")
+            .unwrap();
+        let mut client_transport = connected.into_transport();
         let conn_id = accepted.conn_id;
         assert_eq!(accepted.services, vec!["echo".to_string()]);
 
         let stream_id = client_transport.open_service("echo").await.unwrap();
-        match server.recv().await.unwrap() {
+        match tokio::time::timeout(Duration::from_secs(3), server.recv())
+            .await
+            .expect("server timed out waiting for stream open")
+            .unwrap()
+        {
             DoshServerEvent::Session {
                 event: SessionEvent::Stream(TransportEvent::Open(open)),
                 ..
@@ -602,7 +625,10 @@ mod tests {
             other => panic!("unexpected event {other:?}"),
         }
         assert!(matches!(
-            client_transport.recv().await.unwrap(),
+            tokio::time::timeout(Duration::from_secs(3), client_transport.recv())
+                .await
+                .expect("client timed out waiting for stream-open acknowledgement")
+                .unwrap(),
             SessionEvent::Stream(TransportEvent::OpenOk { .. })
         ));
 
@@ -611,7 +637,11 @@ mod tests {
             .await
             .unwrap();
         loop {
-            match server.recv().await.unwrap() {
+            match tokio::time::timeout(Duration::from_secs(3), server.recv())
+                .await
+                .expect("server timed out waiting for stream data")
+                .unwrap()
+            {
                 DoshServerEvent::Session {
                     event: SessionEvent::Stream(TransportEvent::Data(data)),
                     ..
@@ -628,7 +658,11 @@ mod tests {
             }
         }
         loop {
-            match client_transport.recv().await.unwrap() {
+            match tokio::time::timeout(Duration::from_secs(3), client_transport.recv())
+                .await
+                .expect("client timed out waiting for stream response")
+                .unwrap()
+            {
                 SessionEvent::Stream(TransportEvent::Data(data)) => {
                     assert_eq!(data.chunks, vec![b"pong".to_vec()]);
                     break;