Test authorized TCP pipe forwarding

crates/geth-node/src/lib.rs

@@ -6738,6 +6738,7 @@ fn ssh_revocation_from_stored(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use tokio::io::AsyncReadExt;
 
     fn write_offline_iroh_config(paths: &GethPaths) {
         std::fs::write(
@@ -7655,6 +7656,81 @@ mod tests {
         )
         .expect("grant left db sync");
 
+        let tcp_echo = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind tcp echo");
+        let tcp_target = tcp_echo.local_addr().expect("tcp echo addr").to_string();
+        handle_request(
+            &right,
+            ControlRequest::AuthGrant {
+                subject: left.node_id.clone(),
+                resource: format!("resource:pipe-tcp:{tcp_target}"),
+                capability: "pipe.forward".to_owned(),
+                grant_id: Some("grant:left-pipe-tcp-forward".to_owned()),
+            },
+        )
+        .expect("grant left pipe tcp forward");
+        let echo_task = tokio::spawn(async move {
+            let (mut socket, _) = tcp_echo.accept().await.expect("accept tcp echo");
+            let mut buf = [0_u8; 4];
+            socket.read_exact(&mut buf).await.expect("read tcp echo");
+            assert_eq!(&buf, b"ping");
+            socket.write_all(b"pong").await.expect("write tcp echo");
+            socket.shutdown().await.expect("shutdown tcp echo");
+        });
+        let (forward_client, forward_daemon) = UnixStream::pair().expect("unix stream pair");
+        let right_node_id = right_card.node_id.to_string();
+        let left_for_forward = left.clone();
+        let tcp_target_for_forward = tcp_target.clone();
+        let forward_task = tokio::spawn(async move {
+            handle_local_pipe_tcp_stream(
+                left_for_forward,
+                &right_node_id,
+                tcp_target_for_forward,
+                None,
+                forward_daemon,
+            )
+            .await
+        });
+        let mut forward_reader = BufReader::new(forward_client);
+        let mut line = String::new();
+        forward_reader
+            .read_line(&mut line)
+            .await
+            .expect("read pipe tcp response");
+        match geth_control::decode_response(&line).expect("decode pipe tcp response") {
+            ControlResponse::PipeRemoteConnected {
+                allowed,
+                connection,
+                ..
+            } => {
+                assert!(allowed);
+                assert_eq!(connection.target, tcp_target);
+            }
+            other => panic!("unexpected pipe TCP stream response: {other:?}"),
+        }
+        let stream = forward_reader.into_inner();
+        let (mut forward_read, mut forward_write) = stream.into_split();
+        forward_write
+            .write_all(b"ping")
+            .await
+            .expect("write through pipe tcp forward");
+        forward_write
+            .shutdown()
+            .await
+            .expect("shutdown pipe tcp forward write");
+        let mut forwarded = Vec::new();
+        forward_read
+            .read_to_end(&mut forwarded)
+            .await
+            .expect("read through pipe tcp forward");
+        assert_eq!(forwarded, b"pong");
+        forward_task
+            .await
+            .expect("join pipe tcp forward")
+            .expect("pipe tcp forward");
+        echo_task.await.expect("join tcp echo");
+
         let allowed = handle_request_async(
             &left,
             ControlRequest::PeerAuthCheck {