Merge branch 'master' into next

CHANGELOG.md

@@ -101,6 +101,10 @@ with v0.3.x peers.
 
 ### Fixed
 
+- The Tor transport now increments its `connect_refused` statistic (the
+  "Refused" line in fipstop) when a SOCKS5 connection is actively
+  refused, instead of recording every connect failure as a generic
+  SOCKS5 error. The counter previously stayed at zero.
 - MMP sender metrics now ignore duplicate or regressed receiver reports
   before updating RTT, loss, goodput, or ETX. Receiver reports also
   suppress timestamp echo when dwell time overflows, so stale reports

src/transport/tor/mock_socks5.rs

@@ -33,6 +33,11 @@ pub struct MockSocks5Server {
     addr: SocketAddr,
     /// The real target address to connect to (ignores SOCKS5 requested target).
     target_addr: SocketAddr,
+    /// SOCKS5 reply code (REP field) to send in the CONNECT reply. When this
+    /// is `REP_SUCCESS` (0x00) the server proxies to `target_addr`; any other
+    /// value is sent as an error reply and the connection is closed without
+    /// proxying. Use `0x05` (Connection refused) to drive the refused path.
+    reply_code: u8,
     /// Listener handle.
     listener: Option<TcpListener>,
 }
@@ -40,13 +45,23 @@ pub struct MockSocks5Server {
 impl MockSocks5Server {
     /// Create a new mock SOCKS5 server that forwards to the given target.
     ///
-    /// Binds to `127.0.0.1:0` (OS-assigned port).
+    /// Binds to `127.0.0.1:0` (OS-assigned port). Replies with success and
+    /// proxies bytes bidirectionally.
     pub async fn new(target_addr: SocketAddr) -> std::io::Result<Self> {
+        Self::with_reply_code(target_addr, REP_SUCCESS).await
+    }
+
+    /// Create a mock SOCKS5 server that replies to the CONNECT request with
+    /// the given REP code. A non-success code (e.g. `0x05` Connection refused)
+    /// causes the server to send the error reply and close the connection
+    /// without proxying, exercising the client's connect-error path.
+    pub async fn with_reply_code(target_addr: SocketAddr, reply_code: u8) -> std::io::Result<Self> {
         let listener = TcpListener::bind("127.0.0.1:0").await?;
         let addr = listener.local_addr()?;
         Ok(Self {
             addr,
             target_addr,
+            reply_code,
             listener: Some(listener),
         })
     }
@@ -62,6 +77,7 @@ impl MockSocks5Server {
     pub fn spawn(mut self) -> JoinHandle<()> {
         let listener = self.listener.take().expect("listener already consumed");
         let target_addr = self.target_addr;
+        let reply_code = self.reply_code;
 
         tokio::spawn(async move {
             // Accept one SOCKS5 client
@@ -160,6 +176,28 @@ impl MockSocks5Server {
                 other => panic!("unsupported ATYP: {}", other),
             }
 
+            // Error path: reply with the configured REP code and close without
+            // proxying (the client maps the REP code to a tokio_socks::Error).
+            if reply_code != REP_SUCCESS {
+                let reply = [
+                    SOCKS_VERSION,
+                    reply_code,
+                    0x00, // RSV
+                    ATYP_IPV4,
+                    0,
+                    0,
+                    0,
+                    0, // bind addr
+                    0,
+                    0, // bind port
+                ];
+                client
+                    .write_all(&reply)
+                    .await
+                    .expect("write error connect reply");
+                return;
+            }
+
             // Connect to the real target
             let mut target = tokio::net::TcpStream::connect(target_addr)
                 .await

src/transport/tor/mod.rs

@@ -775,7 +775,14 @@ impl TorTransport {
         let stream = match socks_result {
             Ok(Ok(socks_stream)) => socks_stream.into_inner(),
             Ok(Err(e)) => {
-                self.stats.record_socks5_error();
+                // A SOCKS5 REP=0x05 reply is a genuine connection refusal;
+                // every other failure mode (unreachable proxy, ruleset,
+                // protocol/parse, I/O) is a generic SOCKS5 error.
+                if matches!(e, tokio_socks::Error::ConnectionRefused) {
+                    self.stats.record_connect_refused();
+                } else {
+                    self.stats.record_socks5_error();
+                }
                 warn!(
                     transport_id = %self.transport_id,
                     remote_addr = %addr,
@@ -891,6 +898,7 @@ impl TorTransport {
         let transport_id = self.transport_id;
         let remote_addr = addr.clone();
         let config = self.config.clone();
+        let stats = self.stats.clone();
 
         debug!(
             transport_id = %transport_id,
@@ -932,6 +940,13 @@ impl TorTransport {
             let stream = match socks_result {
                 Ok(Ok(socks_stream)) => socks_stream.into_inner(),
                 Ok(Err(e)) => {
+                    // Mirror the synchronous path: count a genuine REP=0x05
+                    // refusal precisely, everything else as a SOCKS5 error.
+                    if matches!(e, tokio_socks::Error::ConnectionRefused) {
+                        stats.record_connect_refused();
+                    } else {
+                        stats.record_socks5_error();
+                    }
                     debug!(
                         transport_id = %transport_id,
                         remote_addr = %remote_addr,
@@ -1718,6 +1733,68 @@ mod tests {
         assert!(result.is_err());
     }
 
+    /// A SOCKS5 server that actively refuses the CONNECT (REP=0x05) must
+    /// increment `connect_refused`, not `socks5_errors`. Drives the
+    /// synchronous connect path via `send_async`.
+    #[tokio::test]
+    async fn test_connect_refused_increments_refused_counter() {
+        // The mock never proxies on the refusal path, so the target addr is
+        // unused; any well-formed address suffices.
+        let dummy_target: SocketAddr = "127.0.0.1:1".parse().unwrap();
+        let mock = MockSocks5Server::with_reply_code(dummy_target, 0x05)
+            .await
+            .unwrap();
+        let proxy_addr = mock.addr();
+        let _proxy_handle = mock.spawn();
+
+        let (tx, _rx) = packet_channel(32);
+        let config = TorConfig {
+            socks5_addr: Some(proxy_addr.to_string()),
+            connect_timeout_ms: Some(2000),
+            ..Default::default()
+        };
+        let mut transport = TorTransport::new(TransportId::new(1), None, config, tx);
+        transport.start_async().await.unwrap();
+
+        let addr = TransportAddr::from_string("192.168.1.1:2121");
+        let result = transport.send_async(&addr, &build_msg1_frame()).await;
+        assert!(result.is_err(), "refused connect should fail");
+
+        let snap = transport.stats().snapshot();
+        assert_eq!(snap.connect_refused, 1, "connect_refused should be 1");
+        assert_eq!(snap.socks5_errors, 0, "socks5_errors should be 0");
+    }
+
+    /// A non-refusal SOCKS5 error reply (e.g. REP=0x01 general failure) must
+    /// still be counted as a generic `socks5_errors`, leaving `connect_refused`
+    /// untouched.
+    #[tokio::test]
+    async fn test_socks5_general_failure_increments_error_counter() {
+        let dummy_target: SocketAddr = "127.0.0.1:1".parse().unwrap();
+        let mock = MockSocks5Server::with_reply_code(dummy_target, 0x01)
+            .await
+            .unwrap();
+        let proxy_addr = mock.addr();
+        let _proxy_handle = mock.spawn();
+
+        let (tx, _rx) = packet_channel(32);
+        let config = TorConfig {
+            socks5_addr: Some(proxy_addr.to_string()),
+            connect_timeout_ms: Some(2000),
+            ..Default::default()
+        };
+        let mut transport = TorTransport::new(TransportId::new(1), None, config, tx);
+        transport.start_async().await.unwrap();
+
+        let addr = TransportAddr::from_string("192.168.1.1:2121");
+        let result = transport.send_async(&addr, &build_msg1_frame()).await;
+        assert!(result.is_err(), "failed connect should fail");
+
+        let snap = transport.stats().snapshot();
+        assert_eq!(snap.socks5_errors, 1, "socks5_errors should be 1");
+        assert_eq!(snap.connect_refused, 0, "connect_refused should be 0");
+    }
+
     #[tokio::test]
     async fn test_connect_timeout() {
         // Use a non-routable address as the SOCKS5 proxy to trigger timeout