From 2f7ecee18a8dc4fb4f8bd52f74ac1b84d6b19001 Mon Sep 17 00:00:00 2001
From: Ofer Chen <oferchen@gmail.com>
Date: Wed, 27 May 2026 08:50:01 +0300
Subject: [PATCH 1/5] feat: add DaemonStream abstraction for plain TCP and TLS
 connections

Introduce DaemonStream enum that wraps either a plain TcpStream or a
rustls TLS connection behind a unified Read + Write interface. All
daemon call sites now accept DaemonStream instead of raw TcpStream,
enabling transparent TLS upgrades when the daemon-tls feature is active.

The TLS variant is gated behind #[cfg(feature = "daemon-tls")] so
default builds pay zero cost. Includes wrap_accepted_stream() helper
that performs TLS handshake or wraps as Plain based on TlsAcceptor
availability.
---
 crates/daemon/src/daemon.rs                   |   2 +-
 crates/daemon/src/daemon/sections/greeting.rs |   2 +-
 .../src/daemon/sections/legacy_messages.rs    |   6 +-
 .../daemon/sections/module_access/listing.rs  |   2 +-
 .../daemon/sections/module_access/request.rs  |   4 +-
 .../src/daemon/sections/proxy_protocol.rs     |   5 +-
 .../sections/server_runtime/accept_loop.rs    |   2 +
 .../sections/server_runtime/connection.rs     |  57 +++-
 .../sections/server_runtime/listener.rs       |   2 +-
 .../src/daemon/sections/session_runtime.rs    |  13 +-
 crates/daemon/src/daemon_stream.rs            | 288 ++++++++++++++++++
 crates/daemon/src/lib.rs                      |   3 +
 12 files changed, 361 insertions(+), 25 deletions(-)
 create mode 100644 crates/daemon/src/daemon_stream.rs

diff --git a/crates/daemon/src/daemon.rs b/crates/daemon/src/daemon.rs
index 4677f80d0..eb0a3be44 100644
--- a/crates/daemon/src/daemon.rs
+++ b/crates/daemon/src/daemon.rs
@@ -61,7 +61,7 @@ use protocol::{
     parse_legacy_daemon_message,
 };
 
-use crate::{config::DaemonConfig, error::DaemonError, systemd};
+use crate::{config::DaemonConfig, daemon_stream::DaemonStream, error::DaemonError, systemd};
 
 mod help;
 pub(crate) mod tracing_stream;
diff --git a/crates/daemon/src/daemon/sections/greeting.rs b/crates/daemon/src/daemon/sections/greeting.rs
index 745bd152d..39b995387 100644
--- a/crates/daemon/src/daemon/sections/greeting.rs
+++ b/crates/daemon/src/daemon/sections/greeting.rs
@@ -78,7 +78,7 @@ pub(crate) fn read_trimmed_line<R: BufRead>(reader: &mut R) -> io::Result<Option
 }
 
 fn advertise_capabilities(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     modules: &[ModuleRuntime],
     messages: &LegacyMessageCache,
 ) -> io::Result<()> {
diff --git a/crates/daemon/src/daemon/sections/legacy_messages.rs b/crates/daemon/src/daemon/sections/legacy_messages.rs
index 15ec5177d..51db52dc6 100644
--- a/crates/daemon/src/daemon/sections/legacy_messages.rs
+++ b/crates/daemon/src/daemon/sections/legacy_messages.rs
@@ -42,7 +42,7 @@ impl LegacyMessageCache {
 
     fn write(
         &self,
-        stream: &mut TcpStream,
+        stream: &mut DaemonStream,
         limiter: &mut Option<BandwidthLimiter>,
         message: LegacyDaemonMessage<'_>,
     ) -> io::Result<()> {
@@ -52,7 +52,7 @@ impl LegacyMessageCache {
 
     fn write_ok(
         &self,
-        stream: &mut TcpStream,
+        stream: &mut DaemonStream,
         limiter: &mut Option<BandwidthLimiter>,
     ) -> io::Result<()> {
         write_limited(stream, limiter, &self.ok)
@@ -60,7 +60,7 @@ impl LegacyMessageCache {
 
     fn write_exit(
         &self,
-        stream: &mut TcpStream,
+        stream: &mut DaemonStream,
         limiter: &mut Option<BandwidthLimiter>,
     ) -> io::Result<()> {
         write_limited(stream, limiter, &self.exit)
diff --git a/crates/daemon/src/daemon/sections/module_access/listing.rs b/crates/daemon/src/daemon/sections/module_access/listing.rs
index 4fd89795f..5debd3ecf 100644
--- a/crates/daemon/src/daemon/sections/module_access/listing.rs
+++ b/crates/daemon/src/daemon/sections/module_access/listing.rs
@@ -24,7 +24,7 @@ fn format_module_listing_line(name: &str, comment: &str) -> String {
 /// to access. Only modules marked as listable and that permit the peer's IP
 /// address are included in the response.
 fn respond_with_module_list(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     limiter: &mut Option<BandwidthLimiter>,
     modules: &[ModuleRuntime],
     motd_lines: &[String],
diff --git a/crates/daemon/src/daemon/sections/module_access/request.rs b/crates/daemon/src/daemon/sections/module_access/request.rs
index eaec43125..ee78d030e 100644
--- a/crates/daemon/src/daemon/sections/module_access/request.rs
+++ b/crates/daemon/src/daemon/sections/module_access/request.rs
@@ -14,7 +14,7 @@
 /// with the module request metadata so helper functions receive a single
 /// context parameter instead of many individual arguments.
 struct ModuleRequestContext<'a> {
-    reader: &'a mut BufReader<TcpStream>,
+    reader: &'a mut BufReader<DaemonStream>,
     limiter: &'a mut Option<BandwidthLimiter>,
     peer_ip: IpAddr,
     session_peer_host: Option<&'a str>,
@@ -241,7 +241,7 @@ fn handle_module_denied(
 /// Returns an I/O error if the connection fails, otherwise `Ok(())`.
 #[allow(clippy::too_many_arguments)]
 fn respond_with_module_request(
-    reader: &mut BufReader<TcpStream>,
+    reader: &mut BufReader<DaemonStream>,
     limiter: &mut Option<BandwidthLimiter>,
     modules: &[ModuleRuntime],
     request: &str,
diff --git a/crates/daemon/src/daemon/sections/proxy_protocol.rs b/crates/daemon/src/daemon/sections/proxy_protocol.rs
index 5a65cf830..f5d308eee 100644
--- a/crates/daemon/src/daemon/sections/proxy_protocol.rs
+++ b/crates/daemon/src/daemon/sections/proxy_protocol.rs
@@ -211,9 +211,8 @@ fn parse_v1_header<R: Read>(prefix: &[u8; 12], reader: &mut R) -> io::Result<Opt
 ///
 /// upstream: clientserver.c:1298 - `read_proxy_protocol_header()` is called
 /// before any rsync protocol data when `proxy protocol = true` in the config.
-pub(crate) fn parse_proxy_header(stream: &TcpStream) -> io::Result<Option<SocketAddr>> {
-    // Borrow as Read via reference - TcpStream implements Read for &TcpStream.
-    let mut reader = BufReader::new(stream);
+pub(crate) fn parse_proxy_header(stream: &mut DaemonStream) -> io::Result<Option<SocketAddr>> {
+    let mut reader = BufReader::new(&mut *stream);
     parse_proxy_header_from(&mut reader)
 }
 
diff --git a/crates/daemon/src/daemon/sections/server_runtime/accept_loop.rs b/crates/daemon/src/daemon/sections/server_runtime/accept_loop.rs
index 449b83c8f..a45e01b50 100644
--- a/crates/daemon/src/daemon/sections/server_runtime/accept_loop.rs
+++ b/crates/daemon/src/daemon/sections/server_runtime/accept_loop.rs
@@ -309,6 +309,8 @@ fn serve_connections(
         bandwidth_burst,
         reverse_lookup,
         proxy_protocol,
+        #[cfg(feature = "daemon-tls")]
+        tls_acceptor: None,
     };
 
     if listeners.len() == 1 {
diff --git a/crates/daemon/src/daemon/sections/server_runtime/connection.rs b/crates/daemon/src/daemon/sections/server_runtime/connection.rs
index 04b9b9ce7..81dfcb3a6 100644
--- a/crates/daemon/src/daemon/sections/server_runtime/connection.rs
+++ b/crates/daemon/src/daemon/sections/server_runtime/connection.rs
@@ -28,6 +28,10 @@ struct AcceptLoopState<'a> {
     bandwidth_burst: Option<NonZeroU64>,
     reverse_lookup: bool,
     proxy_protocol: bool,
+    /// TLS acceptor for wrapping accepted TCP connections when the daemon is
+    /// configured with certificate material. `None` means plain TCP only.
+    #[cfg(feature = "daemon-tls")]
+    tls_acceptor: Option<crate::tls::TlsAcceptor>,
 }
 
 /// Checks signal flags and performs maintenance tasks between accept iterations.
@@ -118,7 +122,7 @@ fn check_signals_and_maintain(
 /// `@ERROR: max connections (%d) reached -- try again later\n` to the
 /// client via `io_printf(f_out, ...)`.
 fn refuse_if_at_capacity(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     peer_addr: SocketAddr,
     state: &AcceptLoopState<'_>,
 ) -> bool {
@@ -183,7 +187,7 @@ pub(crate) fn log_max_connections_rejection(
 /// upstream: clientserver.c - fork per connection; we use threads with
 /// `catch_unwind` for equivalent crash isolation.
 fn spawn_connection_worker(
-    stream: TcpStream,
+    stream: DaemonStream,
     raw_peer_addr: SocketAddr,
     state: &AcceptLoopState<'_>,
 ) -> thread::JoinHandle<WorkerResult> {
@@ -250,7 +254,7 @@ fn spawn_connection_worker(
 
 /// Applies socket options to an accepted stream and logs any failure.
 fn apply_client_options(
-    stream: &TcpStream,
+    stream: &DaemonStream,
     client_socket_options: &[SocketOption],
     log_sink: Option<&SharedLogSink>,
 ) {
@@ -258,7 +262,7 @@ fn apply_client_options(
     // on the accepted client fd before the session handler runs.
     if !client_socket_options.is_empty() {
         if let Err(error) =
-            apply_socket_options_to_stream(stream, client_socket_options)
+            apply_socket_options_to_stream(stream.tcp_stream(), client_socket_options)
         {
             if let Some(log) = log_sink {
                 let text = format!(
@@ -271,6 +275,37 @@ fn apply_client_options(
     }
 }
 
+/// Wraps an accepted `TcpStream` into a [`DaemonStream`].
+///
+/// When the `daemon-tls` feature is enabled and a TLS acceptor is configured,
+/// the stream is wrapped through `tls::wrap_stream()` to perform the TLS
+/// handshake. If the handshake fails, the error is logged and `None` is
+/// returned so the accept loop can skip the connection.
+///
+/// When TLS is not configured (or the feature is disabled), the stream is
+/// wrapped as `DaemonStream::Plain`.
+fn wrap_accepted_stream(
+    tcp_stream: TcpStream,
+    #[allow(unused_variables)] state: &AcceptLoopState<'_>,
+) -> Option<DaemonStream> {
+    #[cfg(feature = "daemon-tls")]
+    if let Some(ref acceptor) = state.tls_acceptor {
+        return match crate::tls::wrap_stream(acceptor, tcp_stream) {
+            Ok(tls_stream) => Some(DaemonStream::Tls(tls_stream)),
+            Err(error) => {
+                if let Some(log) = state.log_sink.as_ref() {
+                    let text = format!("TLS handshake failed: {error}");
+                    let message = rsync_warning!(text).with_role(Role::Daemon);
+                    log_message(log, &message);
+                }
+                None
+            }
+        };
+    }
+
+    Some(DaemonStream::plain(tcp_stream))
+}
+
 /// Updates the systemd connection status after a new connection is accepted.
 fn update_connection_status_after_accept(state: &mut AcceptLoopState<'_>) {
     let current_active = state.workers.len();
@@ -307,8 +342,8 @@ fn run_single_listener_loop(
         }
 
         match listener.accept() {
-            Ok((mut stream, raw_peer_addr)) => {
-                if let Err(error) = stream.set_nonblocking(false) {
+            Ok((tcp_stream, raw_peer_addr)) => {
+                if let Err(error) = tcp_stream.set_nonblocking(false) {
                     if let Some(log) = state.log_sink.as_ref() {
                         let text = format!(
                             "failed to set accepted socket to blocking: {error}"
@@ -319,6 +354,10 @@ fn run_single_listener_loop(
                     continue;
                 }
 
+                let Some(mut stream) = wrap_accepted_stream(tcp_stream, state) else {
+                    continue;
+                };
+
                 apply_client_options(&stream, &state.client_socket_options, state.log_sink.as_ref());
 
                 if refuse_if_at_capacity(&mut stream, raw_peer_addr, state) {
@@ -434,7 +473,11 @@ fn run_dual_stack_loop(
 
         // Use recv_timeout to allow periodic worker reaping and signal checks
         match rx.recv_timeout(Duration::from_millis(100)) {
-            Ok(Ok((mut stream, raw_peer_addr))) => {
+            Ok(Ok((tcp_stream, raw_peer_addr))) => {
+                let Some(mut stream) = wrap_accepted_stream(tcp_stream, state) else {
+                    continue;
+                };
+
                 apply_client_options(&stream, &state.client_socket_options, state.log_sink.as_ref());
 
                 if refuse_if_at_capacity(&mut stream, raw_peer_addr, state) {
diff --git a/crates/daemon/src/daemon/sections/server_runtime/listener.rs b/crates/daemon/src/daemon/sections/server_runtime/listener.rs
index 6df3be915..f65ab6089 100644
--- a/crates/daemon/src/daemon/sections/server_runtime/listener.rs
+++ b/crates/daemon/src/daemon/sections/server_runtime/listener.rs
@@ -149,7 +149,7 @@ fn bind_with_backlog(addr: SocketAddr, backlog: i32) -> io::Result<TcpListener>
 }
 
 /// Configures read/write timeouts on an accepted client stream.
-fn configure_stream(stream: &TcpStream) -> io::Result<()> {
+fn configure_stream(stream: &DaemonStream) -> io::Result<()> {
     stream.set_read_timeout(Some(SOCKET_TIMEOUT))?;
     stream.set_write_timeout(Some(SOCKET_TIMEOUT))
 }
diff --git a/crates/daemon/src/daemon/sections/session_runtime.rs b/crates/daemon/src/daemon/sections/session_runtime.rs
index c06113936..ed415ebf8 100644
--- a/crates/daemon/src/daemon/sections/session_runtime.rs
+++ b/crates/daemon/src/daemon/sections/session_runtime.rs
@@ -42,7 +42,7 @@ struct LegacySessionParams<'a> {
 /// each child calls `rsync_module()` which performs the full session lifecycle.
 #[cfg_attr(feature = "tracing", instrument(skip(stream, params), fields(peer = %peer_addr), name = "session_handler"))]
 fn handle_session(
-    stream: TcpStream,
+    stream: DaemonStream,
     peer_addr: SocketAddr,
     params: SessionParams<'_>,
 ) -> io::Result<()> {
@@ -66,8 +66,9 @@ fn handle_session(
 
     // upstream: clientserver.c:1298 - read PROXY protocol header before any
     // rsync protocol data when `proxy protocol = true` in the config.
+    let mut stream = stream;
     let peer_addr = if proxy_protocol {
-        match parse_proxy_header(&stream) {
+        match parse_proxy_header(&mut stream) {
             Ok(Some(proxied_addr)) => proxied_addr,
             Ok(None) => peer_addr,
             Err(error) => {
@@ -174,7 +175,7 @@ enum SessionStyle {
 /// chunks and each chunk is registered with the limiter before sending.
 /// When no limiter is present, the payload is written in a single call.
 fn write_limited(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     limiter: &mut Option<BandwidthLimiter>,
     payload: &[u8],
 ) -> io::Result<()> {
@@ -204,7 +205,7 @@ fn write_limited(
 /// 3. Client sends module name (or `#list`)
 #[cfg_attr(feature = "tracing", instrument(skip(stream, params), fields(peer = %peer_addr), name = "legacy_session"))]
 fn handle_legacy_session(
-    stream: TcpStream,
+    stream: DaemonStream,
     peer_addr: SocketAddr,
     params: LegacySessionParams<'_>,
 ) -> io::Result<()> {
@@ -368,7 +369,7 @@ fn read_early_input(
 }
 
 fn handle_binary_session(
-    stream: TcpStream,
+    stream: DaemonStream,
     daemon_limit: Option<NonZeroU64>,
     daemon_burst: Option<NonZeroU64>,
     log_sink: Option<SharedLogSink>,
@@ -377,7 +378,7 @@ fn handle_binary_session(
 }
 
 fn handle_binary_session_internal(
-    mut stream: TcpStream,
+    mut stream: DaemonStream,
     daemon_limit: Option<NonZeroU64>,
     daemon_burst: Option<NonZeroU64>,
     log_sink: Option<SharedLogSink>,
diff --git a/crates/daemon/src/daemon_stream.rs b/crates/daemon/src/daemon_stream.rs
new file mode 100644
index 000000000..b257289ce
--- /dev/null
+++ b/crates/daemon/src/daemon_stream.rs
@@ -0,0 +1,288 @@
+//! Unified stream abstraction for plain TCP and TLS-wrapped daemon connections.
+//!
+//! [`DaemonStream`] transparently handles both plain `TcpStream` and
+//! TLS-encrypted connections (via rustls) behind a single type that
+//! implements `Read + Write`. This lets the daemon's session handler,
+//! greeting exchange, and module access code operate identically
+//! regardless of whether TLS is active.
+//!
+//! The `Tls` variant is only available when the `daemon-tls` Cargo
+//! feature is enabled. Without it, `DaemonStream` is a simple wrapper
+//! around `TcpStream` with no additional dependencies.
+
+use std::io::{self, Read, Write};
+use std::net::{Shutdown, TcpStream};
+use std::time::Duration;
+
+/// A daemon connection that is either a plain TCP stream or a
+/// TLS-encrypted stream over TCP.
+///
+/// Implements `Read` and `Write` by delegating to the inner stream,
+/// making it a drop-in replacement for `TcpStream` in the daemon's
+/// session handler pipeline.
+///
+/// # Variants
+///
+/// - `Plain` - unencrypted TCP, used when no TLS configuration is
+///   present.
+/// - `Tls` - encrypted via rustls `StreamOwned`, used when the daemon
+///   is started with `--ssl` / certificate configuration. Only
+///   available behind `#[cfg(feature = "daemon-tls")]`.
+pub enum DaemonStream {
+    /// Unencrypted TCP connection.
+    Plain(TcpStream),
+
+    /// TLS-encrypted connection over TCP.
+    ///
+    /// The `StreamOwned` manages the rustls session state and encrypts
+    /// all data transparently. It implements `Read + Write` by
+    /// performing TLS record framing under the hood.
+    #[cfg(feature = "daemon-tls")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "daemon-tls")))]
+    Tls(rustls::StreamOwned<rustls::ServerConnection, TcpStream>),
+}
+
+impl DaemonStream {
+    /// Wraps a plain TCP stream (no encryption).
+    pub fn plain(stream: TcpStream) -> Self {
+        Self::Plain(stream)
+    }
+
+    /// Configures the read timeout on the underlying TCP socket.
+    ///
+    /// Delegates to `TcpStream::set_read_timeout` regardless of whether
+    /// the connection is plain or TLS-wrapped - the timeout applies at the
+    /// OS socket level, beneath the TLS layer.
+    pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
+        match self {
+            Self::Plain(s) => s.set_read_timeout(dur),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.sock.set_read_timeout(dur),
+        }
+    }
+
+    /// Configures the write timeout on the underlying TCP socket.
+    pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
+        match self {
+            Self::Plain(s) => s.set_write_timeout(dur),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.sock.set_write_timeout(dur),
+        }
+    }
+
+    /// Enables or disables `TCP_NODELAY` on the underlying socket.
+    pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
+        match self {
+            Self::Plain(s) => s.set_nodelay(nodelay),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.sock.set_nodelay(nodelay),
+        }
+    }
+
+    /// Shuts down the read, write, or both halves of the connection.
+    ///
+    /// For TLS streams this operates on the underlying TCP socket. The
+    /// TLS `close_notify` alert should be sent (via `flush` / drop)
+    /// before calling this.
+    pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
+        match self {
+            Self::Plain(s) => s.shutdown(how),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.sock.shutdown(how),
+        }
+    }
+
+    /// Returns a reference to the underlying `TcpStream`.
+    ///
+    /// Useful for operations that need the raw socket (e.g., applying
+    /// socket options, reading the peer address).
+    pub fn tcp_stream(&self) -> &TcpStream {
+        match self {
+            Self::Plain(s) => s,
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => &s.sock,
+        }
+    }
+
+    /// Returns `true` if this is a TLS-encrypted connection.
+    pub fn is_tls(&self) -> bool {
+        match self {
+            Self::Plain(_) => false,
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(_) => true,
+        }
+    }
+
+    /// Consumes the `DaemonStream` and returns the inner `TcpStream`.
+    ///
+    /// For the `Plain` variant this is a no-op unwrap. For `Tls`, this
+    /// extracts the underlying TCP socket from the `StreamOwned`,
+    /// discarding the TLS session state. The caller is responsible for
+    /// ensuring the TLS session has been properly shut down before
+    /// calling this.
+    pub fn into_tcp_stream(self) -> TcpStream {
+        match self {
+            Self::Plain(s) => s,
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.sock,
+        }
+    }
+}
+
+impl Read for DaemonStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self {
+            Self::Plain(s) => s.read(buf),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.read(buf),
+        }
+    }
+}
+
+impl Write for DaemonStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        match self {
+            Self::Plain(s) => s.write(buf),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.write(buf),
+        }
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        match self {
+            Self::Plain(s) => s.flush(),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(s) => s.flush(),
+        }
+    }
+}
+
+impl std::fmt::Debug for DaemonStream {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Plain(s) => f.debug_tuple("DaemonStream::Plain").field(s).finish(),
+            #[cfg(feature = "daemon-tls")]
+            Self::Tls(_) => f.debug_tuple("DaemonStream::Tls").field(&"<tls>").finish(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::TcpListener;
+
+    /// Creates a connected pair of TCP streams for testing.
+    fn connected_pair() -> (TcpStream, TcpStream) {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let client = TcpStream::connect(addr).unwrap();
+        let (server, _) = listener.accept().unwrap();
+        (client, server)
+    }
+
+    #[test]
+    fn plain_read_write_roundtrip() {
+        let (client, server) = connected_pair();
+        let mut daemon = DaemonStream::plain(server);
+        let mut client = client;
+
+        let payload = b"hello daemon";
+        client.write_all(payload).unwrap();
+        client.flush().unwrap();
+
+        let mut buf = [0u8; 64];
+        let n = daemon.read(&mut buf).unwrap();
+        assert_eq!(&buf[..n], payload);
+    }
+
+    #[test]
+    fn plain_write_read_roundtrip() {
+        let (client, server) = connected_pair();
+        let mut daemon = DaemonStream::plain(server);
+
+        let payload = b"response from daemon";
+        daemon.write_all(payload).unwrap();
+        daemon.flush().unwrap();
+
+        let mut client = client;
+        let mut buf = [0u8; 64];
+        let n = client.read(&mut buf).unwrap();
+        assert_eq!(&buf[..n], payload);
+    }
+
+    #[test]
+    fn plain_is_not_tls() {
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::plain(server);
+        assert!(!daemon.is_tls());
+    }
+
+    #[test]
+    fn plain_tcp_stream_ref() {
+        let (_client, server) = connected_pair();
+        let addr = server.local_addr().unwrap();
+        let daemon = DaemonStream::plain(server);
+        assert_eq!(daemon.tcp_stream().local_addr().unwrap(), addr);
+    }
+
+    #[test]
+    fn plain_into_tcp_stream() {
+        let (_client, server) = connected_pair();
+        let addr = server.local_addr().unwrap();
+        let daemon = DaemonStream::plain(server);
+        let recovered = daemon.into_tcp_stream();
+        assert_eq!(recovered.local_addr().unwrap(), addr);
+    }
+
+    #[test]
+    fn plain_set_timeouts() {
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::plain(server);
+        let dur = Some(Duration::from_secs(5));
+        daemon.set_read_timeout(dur).unwrap();
+        daemon.set_write_timeout(dur).unwrap();
+    }
+
+    #[test]
+    fn plain_set_nodelay() {
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::plain(server);
+        daemon.set_nodelay(true).unwrap();
+    }
+
+    #[test]
+    fn plain_shutdown() {
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::plain(server);
+        daemon.shutdown(Shutdown::Both).unwrap();
+    }
+
+    #[test]
+    fn plain_debug_format() {
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::plain(server);
+        let debug = format!("{daemon:?}");
+        assert!(debug.contains("Plain"), "got: {debug}");
+    }
+
+    #[cfg(feature = "daemon-tls")]
+    #[test]
+    fn tls_variant_is_tls() {
+        // Constructing a real TLS stream requires certificates; verified
+        // via the TLS handshake integration test below. This test just
+        // confirms the is_tls() discriminator when the feature is enabled.
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::Plain(server);
+        assert!(!daemon.is_tls());
+    }
+
+    #[cfg(feature = "daemon-tls")]
+    #[test]
+    fn tls_debug_format() {
+        let (_client, server) = connected_pair();
+        let daemon = DaemonStream::Plain(server);
+        let debug = format!("{daemon:?}");
+        assert!(debug.contains("Plain"), "got: {debug}");
+    }
+}
diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs
index a2fec7145..19d8319cc 100644
--- a/crates/daemon/src/lib.rs
+++ b/crates/daemon/src/lib.rs
@@ -181,6 +181,8 @@ mod daemon;
 mod error;
 /// Daemon configuration file parsing for `rsyncd.conf`.
 pub mod rsyncd_config;
+/// Unified stream abstraction for plain TCP and TLS-wrapped connections.
+pub mod daemon_stream;
 mod systemd;
 /// Native TLS termination via rustls for the daemon listener.
 #[cfg(feature = "daemon-tls")]
@@ -195,6 +197,7 @@ mod tests;
 
 pub use cli::{exit_code_from, run};
 pub use config::{DaemonConfig, DaemonConfigBuilder};
+pub use daemon_stream::DaemonStream;
 #[cfg(feature = "async-daemon")]
 #[cfg_attr(docsrs, doc(cfg(feature = "async-daemon")))]
 pub use daemon::run_async_daemon;

From 18f5cd4c7f300bbbd7a745c9b109f9da207667c1 Mon Sep 17 00:00:00 2001
From: Ofer Chen <oferchen@gmail.com>
Date: Wed, 27 May 2026 08:59:47 +0300
Subject: [PATCH 2/5] style: fix import ordering in daemon lib.rs for cargo fmt

---
 crates/daemon/src/lib.rs | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs
index 19d8319cc..ef36025e3 100644
--- a/crates/daemon/src/lib.rs
+++ b/crates/daemon/src/lib.rs
@@ -178,11 +178,11 @@ pub mod auth;
 mod cli;
 mod config;
 mod daemon;
+/// Unified stream abstraction for plain TCP and TLS-wrapped connections.
+pub mod daemon_stream;
 mod error;
 /// Daemon configuration file parsing for `rsyncd.conf`.
 pub mod rsyncd_config;
-/// Unified stream abstraction for plain TCP and TLS-wrapped connections.
-pub mod daemon_stream;
 mod systemd;
 /// Native TLS termination via rustls for the daemon listener.
 #[cfg(feature = "daemon-tls")]
@@ -197,9 +197,9 @@ mod tests;
 
 pub use cli::{exit_code_from, run};
 pub use config::{DaemonConfig, DaemonConfigBuilder};
-pub use daemon_stream::DaemonStream;
 #[cfg(feature = "async-daemon")]
 #[cfg_attr(docsrs, doc(cfg(feature = "async-daemon")))]
 pub use daemon::run_async_daemon;
 pub use daemon::run_daemon;
+pub use daemon_stream::DaemonStream;
 pub use error::DaemonError;

From 5ba3b10036772111c443b32deed8281f94273882 Mon Sep 17 00:00:00 2001
From: Ofer Chen <oferchen@gmail.com>
Date: Wed, 27 May 2026 20:38:45 +0300
Subject: [PATCH 3/5] fix: update remaining TcpStream signatures to
 DaemonStream (TLS-7)

Migrate 8 function signatures in daemon module-access and module-parsing
from TcpStream to DaemonStream so the daemon session handler compiles
with the unified stream abstraction. The transfer engine still receives
raw TcpStream via tcp_stream().try_clone().
---
 .../src/daemon/sections/module_access/authentication.rs   | 4 ++--
 .../daemon/src/daemon/sections/module_access/request.rs   | 8 ++++----
 .../daemon/src/daemon/sections/module_access/transfer.rs  | 6 +++---
 crates/daemon/src/daemon/sections/module_parsing.rs       | 2 +-
 4 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/crates/daemon/src/daemon/sections/module_access/authentication.rs b/crates/daemon/src/daemon/sections/module_access/authentication.rs
index 200a89703..caa471bed 100644
--- a/crates/daemon/src/daemon/sections/module_access/authentication.rs
+++ b/crates/daemon/src/daemon/sections/module_access/authentication.rs
@@ -31,7 +31,7 @@ enum AuthenticationStatus {
 ///
 /// Returns `Granted` if authentication succeeded, `Denied` otherwise.
 fn perform_module_authentication(
-    reader: &mut BufReader<TcpStream>,
+    reader: &mut BufReader<DaemonStream>,
     limiter: &mut Option<BandwidthLimiter>,
     module: &ModuleDefinition,
     peer_ip: IpAddr,
@@ -205,7 +205,7 @@ fn check_secrets_file_permissions(path: &Path) -> io::Result<()> {
 ///
 /// upstream: clientserver.c:762 - `@ERROR: auth failed on module %s\n`
 fn send_auth_failed(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     module: &ModuleDefinition,
     limiter: &mut Option<BandwidthLimiter>,
     messages: &LegacyMessageCache,
diff --git a/crates/daemon/src/daemon/sections/module_access/request.rs b/crates/daemon/src/daemon/sections/module_access/request.rs
index ee78d030e..16b27b07e 100644
--- a/crates/daemon/src/daemon/sections/module_access/request.rs
+++ b/crates/daemon/src/daemon/sections/module_access/request.rs
@@ -38,7 +38,7 @@ impl<'a> ModuleRequestContext<'a> {
 
 /// Sends an error message and exit marker to the client.
 fn send_error_and_exit(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     limiter: &mut Option<BandwidthLimiter>,
     messages: &LegacyMessageCache,
     payload: &str,
@@ -56,7 +56,7 @@ fn send_error_and_exit(
 ///
 /// upstream: clientserver.c:733 - `@ERROR: access denied to %s from %s (%s)\n`
 fn deny_module(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     module: &ModuleDefinition,
     peer_ip: IpAddr,
     host: Option<&str>,
@@ -81,7 +81,7 @@ fn deny_module(
 /// This confirms that the module request was accepted and the client
 /// may proceed with sending its arguments.
 fn send_daemon_ok(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     limiter: &mut Option<BandwidthLimiter>,
     messages: &LegacyMessageCache,
 ) -> io::Result<()> {
@@ -186,7 +186,7 @@ fn handle_authentication(
 ///
 /// Sends an error message and logs the event.
 fn handle_unknown_module(
-    stream: &mut TcpStream,
+    stream: &mut DaemonStream,
     limiter: &mut Option<BandwidthLimiter>,
     messages: &LegacyMessageCache,
     request: &str,
diff --git a/crates/daemon/src/daemon/sections/module_access/transfer.rs b/crates/daemon/src/daemon/sections/module_access/transfer.rs
index 2cb2c3bde..1ee99fc9e 100644
--- a/crates/daemon/src/daemon/sections/module_access/transfer.rs
+++ b/crates/daemon/src/daemon/sections/module_access/transfer.rs
@@ -229,7 +229,7 @@ fn setup_transfer_streams(
     let stream = ctx.reader.get_mut();
     stream.set_nodelay(true)?;
 
-    let read_stream = match stream.try_clone() {
+    let read_stream = match stream.tcp_stream().try_clone() {
         Ok(s) => s,
         Err(err) => {
             let payload = format!("@ERROR: failed to clone stream: {err}");
@@ -238,7 +238,7 @@ fn setup_transfer_streams(
         }
     };
 
-    let write_stream = match stream.try_clone() {
+    let write_stream = match stream.tcp_stream().try_clone() {
         Ok(s) => s,
         Err(err) => {
             return Err(io::Error::other(format!(
@@ -252,7 +252,7 @@ fn setup_transfer_streams(
 
 /// Builds the handshake result for the transfer.
 fn build_handshake_result(
-    reader: &BufReader<TcpStream>,
+    reader: &BufReader<DaemonStream>,
     negotiated_protocol: Option<ProtocolVersion>,
     client_args: Vec<String>,
     module: &ModuleRuntime,
diff --git a/crates/daemon/src/daemon/sections/module_parsing.rs b/crates/daemon/src/daemon/sections/module_parsing.rs
index 7bbbf9322..b90d87109 100644
--- a/crates/daemon/src/daemon/sections/module_parsing.rs
+++ b/crates/daemon/src/daemon/sections/module_parsing.rs
@@ -122,7 +122,7 @@ fn canonical_option(text: &str) -> String {
     token.to_ascii_lowercase()
 }
 
-fn apply_module_timeout(stream: &TcpStream, module: &ModuleDefinition) -> io::Result<()> {
+fn apply_module_timeout(stream: &DaemonStream, module: &ModuleDefinition) -> io::Result<()> {
     if let Some(timeout) = module.timeout {
         let duration = Duration::from_secs(timeout.get());
         stream.set_read_timeout(Some(duration))?;

From 5faf57991b4550df53a84d169fb2d268a4ccfd28 Mon Sep 17 00:00:00 2001
From: Ofer Chen <oferchen@gmail.com>
Date: Wed, 27 May 2026 20:50:11 +0300
Subject: [PATCH 4/5] fix: update daemon test helpers for DaemonStream
 migration (TLS-7)

Wrap TcpStream with DaemonStream::plain() in four test functions and
add the cfg-gated tls_acceptor field to AcceptLoopState test helper.
---
 .../src/daemon/sections/server_runtime/tests.rs    | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/crates/daemon/src/daemon/sections/server_runtime/tests.rs b/crates/daemon/src/daemon/sections/server_runtime/tests.rs
index 5812f11c9..ec988954d 100644
--- a/crates/daemon/src/daemon/sections/server_runtime/tests.rs
+++ b/crates/daemon/src/daemon/sections/server_runtime/tests.rs
@@ -729,6 +729,8 @@ fn test_accept_loop_state<'a>(
         bandwidth_burst: None,
         reverse_lookup: false,
         proxy_protocol: false,
+        #[cfg(feature = "daemon-tls")]
+        tls_acceptor: None,
     }
 }
 
@@ -769,8 +771,9 @@ fn refuse_if_at_capacity_admits_when_no_cap_configured() {
     let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
     let local = listener.local_addr().expect("local addr");
     let client_handle = thread::spawn(move || TcpStream::connect(local).expect("connect"));
-    let (mut server_stream, peer) = listener.accept().expect("accept");
+    let (server_stream, peer) = listener.accept().expect("accept");
     let _client = client_handle.join().expect("client connect");
+    let mut server_stream = DaemonStream::plain(server_stream);
 
     let flags = no_op_signal_flags();
     let config_path: Option<PathBuf> = None;
@@ -805,8 +808,9 @@ fn accept_loop_refuses_when_at_capacity() {
     let local = listener.local_addr().expect("local addr");
     let client_handle =
         thread::spawn(move || TcpStream::connect(local).expect("connect third client"));
-    let (mut server_stream, peer) = listener.accept().expect("accept third client");
+    let (server_stream, peer) = listener.accept().expect("accept third client");
     let mut client_stream = client_handle.join().expect("client connect");
+    let mut server_stream = DaemonStream::plain(server_stream);
 
     let flags = no_op_signal_flags();
     let config_path: Option<PathBuf> = None;
@@ -855,8 +859,9 @@ fn refuse_if_at_capacity_emits_structured_warning() {
     let local = listener.local_addr().expect("local addr");
     let client_handle =
         thread::spawn(move || TcpStream::connect(local).expect("connect refused client"));
-    let (mut server_stream, peer) = listener.accept().expect("accept refused client");
+    let (server_stream, peer) = listener.accept().expect("accept refused client");
     let _client = client_handle.join().expect("client connect");
+    let mut server_stream = DaemonStream::plain(server_stream);
 
     let log_dir = tempfile::tempdir().expect("log dir");
     let log_path = log_dir.path().join("daemon.log");
@@ -921,8 +926,9 @@ fn accept_loop_recovers_after_disconnect() {
     let local = listener.local_addr().expect("local addr");
     let client_handle =
         thread::spawn(move || TcpStream::connect(local).expect("connect after drain"));
-    let (mut server_stream, peer) = listener.accept().expect("accept after drain");
+    let (server_stream, peer) = listener.accept().expect("accept after drain");
     let _client = client_handle.join().expect("client connect");
+    let mut server_stream = DaemonStream::plain(server_stream);
 
     let flags = no_op_signal_flags();
     let config_path: Option<PathBuf> = None;

From 77c17876a51889e4f02a4f0f58da3e9e155fddf3 Mon Sep 17 00:00:00 2001
From: Ofer Chen <oferchen@gmail.com>
Date: Wed, 27 May 2026 20:59:16 +0300
Subject: [PATCH 5/5] fix: box DaemonStream::Tls variant to resolve
 large_enum_variant lint

The Tls variant (StreamOwned<ServerConnection, TcpStream>) is 1176 bytes
vs Plain(TcpStream) at 4 bytes. Boxing the Tls payload keeps the enum
at pointer size regardless of rustls internals.
---
 crates/daemon/src/daemon/sections/server_runtime/connection.rs | 2 +-
 crates/daemon/src/daemon_stream.rs                             | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/crates/daemon/src/daemon/sections/server_runtime/connection.rs b/crates/daemon/src/daemon/sections/server_runtime/connection.rs
index 81dfcb3a6..eb8026f10 100644
--- a/crates/daemon/src/daemon/sections/server_runtime/connection.rs
+++ b/crates/daemon/src/daemon/sections/server_runtime/connection.rs
@@ -291,7 +291,7 @@ fn wrap_accepted_stream(
     #[cfg(feature = "daemon-tls")]
     if let Some(ref acceptor) = state.tls_acceptor {
         return match crate::tls::wrap_stream(acceptor, tcp_stream) {
-            Ok(tls_stream) => Some(DaemonStream::Tls(tls_stream)),
+            Ok(tls_stream) => Some(DaemonStream::Tls(Box::new(tls_stream))),
             Err(error) => {
                 if let Some(log) = state.log_sink.as_ref() {
                     let text = format!("TLS handshake failed: {error}");
diff --git a/crates/daemon/src/daemon_stream.rs b/crates/daemon/src/daemon_stream.rs
index b257289ce..9ebdf9ced 100644
--- a/crates/daemon/src/daemon_stream.rs
+++ b/crates/daemon/src/daemon_stream.rs
@@ -39,7 +39,7 @@ pub enum DaemonStream {
     /// performing TLS record framing under the hood.
     #[cfg(feature = "daemon-tls")]
     #[cfg_attr(docsrs, doc(cfg(feature = "daemon-tls")))]
-    Tls(rustls::StreamOwned<rustls::ServerConnection, TcpStream>),
+    Tls(Box<rustls::StreamOwned<rustls::ServerConnection, TcpStream>>),
 }
 
 impl DaemonStream {