Don’t have the RTR listener fail if a socket fails after accept. (#937)

This PR changes the RTR listener to not fail the accept loop when setting up
a stream after accepting it fails. Instead it will quietly drop the stream
in this case and keep going.

As part of this, the PR also drops tokio-stream as a dependency and
implements its own listener stream.

This PR fixes CVE-2024-1622 reported by Yohei Nishimura, Atsushi Enomoto,
Ruka Miyachi; Internet Multifeed Co., Japan.

Cargo.toml

@@ -37,7 +37,6 @@ serde_json      = "1.0.57"
 tempfile        = "3.1.0"
 tokio           = { version = "1.24", features = [ "io-util", "macros", "process", "rt", "rt-multi-thread", "signal", "sync" ] }
 tokio-rustls    = "0.24.1"
-tokio-stream    = { version = "0.1", features = ["net"] }
 toml_edit       = "0.20"
 uuid            = "1.1"
 routinator-ui   = { version = "0.3.4", optional = true }

src/rtr.rs

@@ -2,20 +2,19 @@
 
 use std::io;
 use std::future::Future;
-use std::net::{TcpListener as StdListener};
+use std::net::{SocketAddr, TcpListener as StdListener};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::time::Duration;
-use futures::{pin_mut, StreamExt, TryStreamExt};
+use futures::{pin_mut, Stream};
 use futures::future::{pending, select_all};
 use log::error;
 use rpki::rtr::server::{NotifySender, Server, Socket};
 use rpki::rtr::state::State;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::{TcpListener, TcpStream};
 use tokio_rustls::TlsAcceptor;
-use tokio_stream::wrappers::TcpListenerStream;
 use crate::config::Config;
 use crate::error::ExitError;
 use crate::metrics::{SharedRtrServerMetrics, RtrClientMetrics};
@@ -118,9 +117,9 @@ async fn single_rtr_listener(
         }
     };
     let tls = tls.map(TlsAcceptor::from);
-    let listener = TcpListenerStream::new(listener).and_then(|sock| async {
-        RtrStream::new(sock, tls.as_ref(), keepalive, server_metrics.clone())
-    }).boxed();
+    let listener = RtrListener {
+        tcp: listener, tls, keepalive, server_metrics
+    };
     if let Err(err) = Server::new(
         listener, sender, origins.clone()
     ).run().await {
@@ -129,6 +128,40 @@ async fn single_rtr_listener(
 }
 
 
+//------------ RtrListener --------------------------------------------------
+
+/// A wrapper around an TCP listener that produces RTR streams.
+struct RtrListener {
+    tcp: TcpListener,
+    tls: Option<TlsAcceptor>,
+    keepalive: Option<Duration>,
+    server_metrics: SharedRtrServerMetrics,
+}
+
+impl Stream for RtrListener {
+    type Item = Result<RtrStream, io::Error>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        ctx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        match self.tcp.poll_accept(ctx) {
+            Poll::Ready(Ok((sock, addr))) => {
+                match RtrStream::new(
+                    sock, addr,
+                    self.tls.as_ref(), self.keepalive,
+                    self.server_metrics.clone()
+                ) {
+                    Ok(stream) => Poll::Ready(Some(Ok(stream))),
+                    Err(_) => Poll::Pending,
+                }
+            }
+            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
 //------------ RtrStream ----------------------------------------------------
 
 /// A wrapper around a stream socket that takes care of updating metrics.
@@ -141,14 +174,15 @@ impl RtrStream {
     #[allow(clippy::redundant_async_block)] // False positive
     fn new(
         sock: TcpStream,
+        addr: SocketAddr,
         tls: Option<&TlsAcceptor>,
         keepalive: Option<Duration>,
         server_metrics: SharedRtrServerMetrics,
     ) -> Result<Self, io::Error> {
         if let Some(duration) = keepalive {
             Self::set_keepalive(&sock, duration)?
         }
-        let metrics = Arc::new(RtrClientMetrics::new(sock.peer_addr()?.ip()));
+        let metrics = Arc::new(RtrClientMetrics::new(addr.ip()));
         let client_metrics = metrics.clone();
         tokio::spawn(async move {
             server_metrics.add_client(client_metrics).await