feat(server): IPv6 dual-stack, SO_REUSEADDR, and ConnectionHandler for run()

Replace TcpListener::bind() with TcpSocket for control over socket
options before binding:

- SO_REUSEADDR: prevents EADDRINUSE on server restart (TIME_WAIT)
- IPv6 dual-stack: when bound to [::], accepts both IPv4 and IPv6
  on a single socket. Address family is detected from SocketAddr.
- Backlog increased from default to 1024

Add ConnectionHandler trait for connection lifecycle hooks:

- on_accept(peer) -> bool: pre-handshake hook (rate limiting, logging)
- on_disconnected(peer, duration, error): post-connection cleanup

The handler is optional (set via set_connection_handler) and the
default implementation is a no-op, so this is fully backward compatible.
Callers using run_connection() directly are unaffected.

Author: lamco-office

crates/ironrdp-server/src/server.rs

@@ -1,6 +1,8 @@
 use core::net::SocketAddr;
+use core::time::Duration;
 use std::rc::Rc;
 use std::sync::Arc;
+use std::time::Instant;
 
 use anyhow::{Context as _, Result, bail};
 use ironrdp_acceptor::{Acceptor, AcceptorResult, BeginResult, DesktopSize};
@@ -22,13 +24,47 @@ use ironrdp_svc::{ChannelFlags, StaticChannelId, StaticChannelSet, SvcProcessor,
 use ironrdp_tokio::{FramedRead, FramedWrite, TokioFramed, split_tokio_framed, unsplit_tokio_framed};
 use rdpsnd::server::{RdpsndServer, RdpsndServerMessage};
 use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _};
-use tokio::net::TcpListener;
+use tokio::net::TcpSocket;
 use tokio::sync::{Mutex, mpsc, oneshot};
 use tokio::task;
 use tokio_rustls::TlsAcceptor;
 use tracing::{debug, error, trace, warn};
 use {ironrdp_dvc as dvc, ironrdp_rdpsnd as rdpsnd};
 
+/// Connection lifecycle handler for `RdpServer::run()`.
+///
+/// Implement this trait to hook into the TCP accept loop. The default
+/// implementation is a no-op, so existing callers are unaffected.
+///
+/// # Example
+///
+/// ```rust,ignore
+/// use ironrdp_server::ConnectionHandler;
+/// use std::net::SocketAddr;
+/// use std::time::Duration;
+///
+/// struct MyHandler;
+///
+/// impl ConnectionHandler for MyHandler {
+///     fn on_accept(&self, peer: SocketAddr) -> bool {
+///         println!("Connection from {peer}");
+///         true // accept all connections
+///     }
+/// }
+/// ```
+pub trait ConnectionHandler: Send + Sync {
+    /// Called after TCP accept, before the RDP handshake begins.
+    ///
+    /// Return `true` to proceed with the connection, or `false` to
+    /// drop it immediately (e.g. for rate limiting).
+    fn on_accept(&self, _peer: SocketAddr) -> bool {
+        true
+    }
+
+    /// Called after a connection ends, whether successfully or with an error.
+    fn on_disconnected(&self, _peer: SocketAddr, _duration: Duration, _error: Option<&anyhow::Error>) {}
+}
+
 use crate::clipboard::CliprdrServerFactory;
 use crate::display::{DisplayUpdate, RdpServerDisplay};
 use crate::echo::{EchoDvcBridge, EchoServerHandle, EchoServerMessage, build_echo_request};
@@ -240,6 +276,7 @@ pub struct RdpServer {
     ev_receiver: Arc<Mutex<mpsc::UnboundedReceiver<ServerEvent>>>,
     creds: Option<Credentials>,
     local_addr: Option<SocketAddr>,
+    connection_handler: Option<Box<dyn ConnectionHandler>>,
 }
 
 #[derive(Debug)]
@@ -307,9 +344,18 @@ impl RdpServer {
             ev_receiver: Arc::new(Mutex::new(ev_receiver)),
             creds: None,
             local_addr: None,
+            connection_handler: None,
         }
     }
 
+    /// Set an optional connection lifecycle handler.
+    ///
+    /// The handler is called on TCP accept and disconnect events during
+    /// `run()`. Has no effect when using `run_connection()` directly.
+    pub fn set_connection_handler(&mut self, handler: Box<dyn ConnectionHandler>) {
+        self.connection_handler = Some(handler);
+    }
+
     pub fn builder() -> builder::RdpServerBuilder<builder::WantsAddr> {
         builder::RdpServerBuilder::new()
     }
@@ -448,7 +494,25 @@ impl RdpServer {
     }
 
     pub async fn run(&mut self) -> Result<()> {
-        let listener = TcpListener::bind(self.opts.addr).await?;
+        // Create socket with SO_REUSEADDR and IPv6 dual-stack support.
+        // Using TcpSocket instead of TcpListener::bind() gives control
+        // over socket options before binding.
+        let socket = match self.opts.addr {
+            SocketAddr::V4(_) => TcpSocket::new_v4().context("Failed to create IPv4 socket")?,
+            SocketAddr::V6(_) => {
+                // IPv6 socket with dual-stack: accepts both IPv4 and IPv6
+                // on a single socket. IPv4 clients appear as IPv4-mapped
+                // addresses (::ffff:x.x.x.x). Dual-stack is the Linux
+                // default (net.ipv6.bindv6only=0).
+                TcpSocket::new_v6().context("Failed to create IPv6 socket")?
+            }
+        };
+
+        socket.set_reuseaddr(true).context("Failed to set SO_REUSEADDR")?;
+
+        socket.bind(self.opts.addr).context("Failed to bind listen address")?;
+
+        let listener = socket.listen(1024).context("Failed to start listener")?;
         let local_addr = listener.local_addr()?;
 
         debug!("Listening for connections on {local_addr}");
@@ -476,11 +540,36 @@ impl RdpServer {
                     }
                 },
                 Ok((stream, peer)) = listener.accept() => {
+                    // Connection lifecycle: on_accept hook
+                    if let Some(ref handler) = self.connection_handler {
+                        if !handler.on_accept(peer) {
+                            debug!(?peer, "Connection rejected by handler");
+                            continue;
+                        }
+                    }
+
                     debug!(?peer, "Received connection");
                     drop(ev_receiver);
-                    if let Err(error) = self.run_connection(stream).await {
-                        error!(?error, "Connection error");
+
+                    let start = Instant::now();
+                    let result = self.run_connection(stream).await;
+                    let duration = start.elapsed();
+
+                    // Connection lifecycle: on_disconnected hook
+                    match &result {
+                        Ok(()) => {
+                            if let Some(ref handler) = self.connection_handler {
+                                handler.on_disconnected(peer, duration, None);
+                            }
+                        }
+                        Err(error) => {
+                            if let Some(ref handler) = self.connection_handler {
+                                handler.on_disconnected(peer, duration, Some(error));
+                            }
+                            error!(?error, "Connection error");
+                        }
                     }
+
                     self.static_channels = StaticChannelSet::new();
                 }
                 else => break,