Harden native stream forwarding

Author: eldryoth

CHANGELOG.md

@@ -23,6 +23,18 @@ behavior when the change improves security or project direction.
 
 ### Security
 
+- Normalize IPv4-mapped and legacy IPv4-compatible IPv6 stream DNS answers
+  before rebinding policy checks, preventing loopback, private, link-local,
+  carrier-grade NAT, benchmark, and other reserved IPv4 targets from bypassing
+  the stream SSRF guard.
+- Replace cancellation-unsafe inline bidirectional stream copies with two
+  persistent pinned direction futures and a shared last-activity idle timer,
+  preventing partial writes from losing or duplicating plaintext under
+  simultaneous traffic and backpressure.
+- Clear forwarded stream plaintext immediately and clear each full 16 KiB copy
+  buffer on drop with `sanitization`.
+- Enforce positive per-upstream and checked aggregate stream weights inside the
+  public runtime selector boundary, independently of config validation.
 - Make OpenSSL cipher allow-lists deterministic across protocol families:
   omitting all TLS 1.2 or TLS 1.3 suites now disables that protocol version
   instead of retaining Mozilla acceptor defaults. Use the current Mozilla v5
@@ -39,6 +51,8 @@ behavior when the change improves security or project direction.
 - Disable rustls and tokio-rustls default provider features so normal Ring
   builds no longer compile AWS-LC; FIPS profiles continue to select AWS-LC
   explicitly.
+- Make `base64-ng` optional in `fluxheim-tls` and activate it only for Rustls
+  key parsing, keeping it out of the crate's default and OpenSSL-only graphs.
 - Validate every `wasi_snapshot_preview1` import against the plugin's explicit
   capability grants before instantiation. Environment, arguments, inherited
   stdio, filesystem, sockets/network, polling, and process state remain denied.

Cargo.lock

@@ -19,8 +19,8 @@ pub const MAX_STREAM_LISTENERS: usize = 64;
 pub const MAX_STREAM_UPSTREAMS: usize = 64;
 pub const MAX_STREAM_SOURCE_MATCHERS: usize = 256;
 pub const MAX_STREAM_MAX_CONNECTIONS: usize = 1_000_000;
-const MAX_STREAM_UPSTREAM_WEIGHT: usize = 1000;
-const MAX_STREAM_UPSTREAM_TOTAL_WEIGHT: usize = u16::MAX as usize;
+pub const MAX_STREAM_UPSTREAM_WEIGHT: usize = 1000;
+pub const MAX_STREAM_UPSTREAM_TOTAL_WEIGHT: usize = u16::MAX as usize;
 
 #[derive(Debug, Clone, Default, Eq, PartialEq, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]

crates/fluxheim-config/src/config_stream.rs

@@ -11,4 +11,5 @@ description = "Internal Fluxheim TCP stream proxy runtime boundary."
 fluxheim-common = { path = "../fluxheim-common" }
 fluxheim-config = { path = "../fluxheim-config", features = ["stream-proxy"] }
 fluxheim-protocol = { path = "../fluxheim-protocol" }
-tokio = { version = "1.52.3", features = ["io-util", "macros", "time"] }
+sanitization = { version = "1.2.4" }
+tokio = { version = "1.52.3", features = ["io-util", "macros", "rt", "sync", "time"] }

crates/fluxheim-stream/Cargo.toml

@@ -0,0 +1,160 @@
+use std::io;
+use std::time::Duration;
+
+use fluxheim_common::{FluxError, FluxResult};
+use sanitization::sanitize_bytes;
+use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
+use tokio::sync::watch;
+
+const STREAM_COPY_BUFFER_BYTES: usize = 16 * 1024;
+
+struct StreamCopyBuffer([u8; STREAM_COPY_BUFFER_BYTES]);
+
+impl StreamCopyBuffer {
+    fn new() -> Self {
+        Self([0; STREAM_COPY_BUFFER_BYTES])
+    }
+
+    fn clear_forwarded(&mut self, bytes: usize) {
+        sanitize_bytes(&mut self.0[..bytes]);
+    }
+}
+
+impl Drop for StreamCopyBuffer {
+    fn drop(&mut self) {
+        sanitize_bytes(&mut self.0);
+    }
+}
+
+pub fn checked_stream_byte_count(
+    current: u64,
+    additional: u64,
+    max_connection_bytes: Option<u64>,
+) -> FluxResult<u64> {
+    let next = current.checked_add(additional).ok_or_else(|| {
+        FluxError::io(
+            "count stream bytes",
+            io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stream copied byte counter overflowed",
+            ),
+        )
+    })?;
+    if max_connection_bytes.is_some_and(|limit| next > limit) {
+        return Err(FluxError::io(
+            "enforce stream byte limit",
+            io::Error::new(
+                io::ErrorKind::PermissionDenied,
+                "stream max connection bytes exceeded",
+            ),
+        ));
+    }
+    Ok(next)
+}
+
+pub async fn copy_bidirectional_with_limits(
+    downstream: &mut (impl AsyncRead + AsyncWrite + Unpin),
+    upstream: &mut (impl AsyncRead + AsyncWrite + Unpin),
+    idle_timeout: Duration,
+    max_connection_bytes: Option<u64>,
+) -> FluxResult<(u64, u64)> {
+    let (downstream_reader, downstream_writer) = tokio::io::split(downstream);
+    let (upstream_reader, upstream_writer) = tokio::io::split(upstream);
+    let (activity_tx, mut activity_rx) = watch::channel(tokio::time::Instant::now());
+    let downstream_copy = copy_direction(
+        downstream_reader,
+        upstream_writer,
+        max_connection_bytes,
+        activity_tx.clone(),
+    );
+    let upstream_copy = copy_direction(
+        upstream_reader,
+        downstream_writer,
+        max_connection_bytes,
+        activity_tx,
+    );
+    tokio::pin!(downstream_copy);
+    tokio::pin!(upstream_copy);
+
+    let idle_timer = tokio::time::sleep(idle_timeout);
+    tokio::pin!(idle_timer);
+    let mut downstream_total = None;
+    let mut upstream_total = None;
+    let mut activity_open = true;
+
+    loop {
+        tokio::select! {
+            biased;
+            result = &mut downstream_copy, if downstream_total.is_none() => {
+                downstream_total = Some(result?);
+            }
+            result = &mut upstream_copy, if upstream_total.is_none() => {
+                upstream_total = Some(result?);
+            }
+            activity = activity_rx.changed(), if activity_open => {
+                if activity.is_ok() {
+                    let last_activity = *activity_rx.borrow_and_update();
+                    idle_timer.as_mut().reset(last_activity + idle_timeout);
+                } else {
+                    activity_open = false;
+                }
+            }
+            _ = &mut idle_timer => {
+                return Err(FluxError::timeout(
+                    "stream idle timeout",
+                    "stream idle timeout elapsed",
+                ));
+            }
+        }
+        if let (Some(downstream_total), Some(upstream_total)) = (downstream_total, upstream_total) {
+            return Ok((downstream_total, upstream_total));
+        }
+    }
+}
+
+async fn copy_direction<R, W>(
+    mut reader: R,
+    mut writer: W,
+    max_connection_bytes: Option<u64>,
+    activity: watch::Sender<tokio::time::Instant>,
+) -> FluxResult<u64>
+where
+    R: AsyncRead + Unpin,
+    W: AsyncWrite + Unpin,
+{
+    let mut buffer = StreamCopyBuffer::new();
+    let mut total = 0u64;
+    loop {
+        let bytes = reader
+            .read(&mut buffer.0)
+            .await
+            .map_err(|error| FluxError::io("read stream", error))?;
+        if bytes == 0 {
+            writer
+                .shutdown()
+                .await
+                .map_err(|error| FluxError::io("shutdown stream", error))?;
+            return Ok(total);
+        }
+        total = checked_stream_byte_count(total, bytes as u64, max_connection_bytes)?;
+        writer
+            .write_all(&buffer.0[..bytes])
+            .await
+            .map_err(|error| FluxError::io("write stream", error))?;
+        buffer.clear_forwarded(bytes);
+        activity.send_replace(tokio::time::Instant::now());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn stream_copy_buffer_clears_forwarded_prefix() {
+        let mut buffer = StreamCopyBuffer([7; STREAM_COPY_BUFFER_BYTES]);
+        buffer.clear_forwarded(3);
+        assert_eq!(&buffer.0[..3], &[0, 0, 0]);
+        assert_eq!(buffer.0[3], 7);
+    }
+}

crates/fluxheim-stream/src/copy.rs

@@ -19,8 +19,10 @@ use fluxheim_protocol::{
 };
 use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
 
+mod copy;
 mod selector;
 
+pub use copy::{checked_stream_byte_count, copy_bidirectional_with_limits};
 pub use selector::{StreamSelectedUpstream, StreamUpstreamSelector};
 
 #[derive(Debug, Clone, Eq, PartialEq)]
@@ -140,6 +142,9 @@ fn stream_dns_resolved_ipv4_address_allowed(address: Ipv4Addr) -> bool {
 }
 
 fn stream_dns_resolved_ipv6_address_allowed(address: Ipv6Addr) -> bool {
+    if let Some(address) = address.to_ipv4() {
+        return stream_dns_resolved_ipv4_address_allowed(address);
+    }
     let segments = address.segments();
     !(address.is_unspecified()
         || address.is_loopback()
@@ -172,115 +177,6 @@ pub fn stream_error_outcome(error: &FluxError) -> &'static str {
     }
 }
 
-pub fn checked_stream_byte_count(
-    current: u64,
-    additional: u64,
-    max_connection_bytes: Option<u64>,
-) -> FluxResult<u64> {
-    let next = current.checked_add(additional).ok_or_else(|| {
-        FluxError::io(
-            "count stream bytes",
-            io::Error::new(
-                io::ErrorKind::InvalidData,
-                "stream copied byte counter overflowed",
-            ),
-        )
-    })?;
-    if max_connection_bytes.is_some_and(|limit| next > limit) {
-        return Err(FluxError::io(
-            "enforce stream byte limit",
-            io::Error::new(
-                io::ErrorKind::PermissionDenied,
-                "stream max connection bytes exceeded",
-            ),
-        ));
-    }
-    Ok(next)
-}
-
-enum StreamCopyEvent {
-    DownstreamTotal(u64),
-    UpstreamTotal(u64),
-    DownstreamEof,
-    UpstreamEof,
-}
-
-pub async fn copy_bidirectional_with_limits(
-    downstream: &mut (impl AsyncRead + AsyncWrite + Unpin),
-    upstream: &mut (impl AsyncRead + AsyncWrite + Unpin),
-    idle_timeout: Duration,
-    max_connection_bytes: Option<u64>,
-) -> FluxResult<(u64, u64)> {
-    let (mut downstream_reader, mut downstream_writer) = tokio::io::split(downstream);
-    let (mut upstream_reader, mut upstream_writer) = tokio::io::split(upstream);
-    let mut downstream_buffer = [0u8; 16 * 1024];
-    let mut upstream_buffer = [0u8; 16 * 1024];
-    let mut downstream_to_upstream = 0u64;
-    let mut upstream_to_downstream = 0u64;
-    let mut downstream_eof = false;
-    let mut upstream_eof = false;
-
-    while !downstream_eof || !upstream_eof {
-        let event = tokio::select! {
-            result = async {
-                let bytes = read_with_idle_timeout(
-                    &mut downstream_reader,
-                    &mut downstream_buffer,
-                    idle_timeout,
-                ).await?;
-                if bytes == 0 {
-                    shutdown_with_idle_timeout(&mut upstream_writer, idle_timeout).await?;
-                    Ok::<_, FluxError>(StreamCopyEvent::DownstreamEof)
-                } else {
-                    let next = checked_stream_byte_count(
-                        downstream_to_upstream,
-                        bytes as u64,
-                        max_connection_bytes,
-                    )?;
-                    write_with_idle_timeout(
-                        &mut upstream_writer,
-                        &downstream_buffer[..bytes],
-                        idle_timeout,
-                    ).await?;
-                    Ok::<_, FluxError>(StreamCopyEvent::DownstreamTotal(next))
-                }
-            }, if !downstream_eof => result,
-            result = async {
-                let bytes = read_with_idle_timeout(
-                    &mut upstream_reader,
-                    &mut upstream_buffer,
-                    idle_timeout,
-                ).await?;
-                if bytes == 0 {
-                    shutdown_with_idle_timeout(&mut downstream_writer, idle_timeout).await?;
-                    Ok::<_, FluxError>(StreamCopyEvent::UpstreamEof)
-                } else {
-                    let next = checked_stream_byte_count(
-                        upstream_to_downstream,
-                        bytes as u64,
-                        max_connection_bytes,
-                    )?;
-                    write_with_idle_timeout(
-                        &mut downstream_writer,
-                        &upstream_buffer[..bytes],
-                        idle_timeout,
-                    ).await?;
-                    Ok::<_, FluxError>(StreamCopyEvent::UpstreamTotal(next))
-                }
-            }, if !upstream_eof => result,
-        }?;
-
-        match event {
-            StreamCopyEvent::DownstreamTotal(total) => downstream_to_upstream = total,
-            StreamCopyEvent::UpstreamTotal(total) => upstream_to_downstream = total,
-            StreamCopyEvent::DownstreamEof => downstream_eof = true,
-            StreamCopyEvent::UpstreamEof => upstream_eof = true,
-        }
-    }
-
-    Ok((downstream_to_upstream, upstream_to_downstream))
-}
-
 pub async fn write_upstream_proxy_protocol(
     upstream: &mut (impl AsyncWrite + Unpin),
     protocol: UpstreamProxyProtocol,
@@ -414,19 +310,6 @@ where
     }
 }
 
-async fn shutdown_with_idle_timeout<W>(writer: &mut W, idle_timeout: Duration) -> FluxResult<()>
-where
-    W: AsyncWrite + Unpin,
-{
-    match tokio::time::timeout(idle_timeout, writer.shutdown()).await {
-        Ok(result) => result.map_err(|error| FluxError::io("shutdown stream", error)),
-        Err(_) => Err(FluxError::timeout(
-            "stream shutdown timeout",
-            "stream shutdown timeout elapsed",
-        )),
-    }
-}
-
 async fn read_exact_with_idle_timeout<R>(
     reader: &mut R,
     buffer: &mut [u8],