feat: Support backend socket connections (UDS)

README.md

@@ -11,7 +11,7 @@ Proxy LDAP requests to different LDAP servers based on base DN. Based on code fr
 - [ ] Default fallback to `/etc/ldap-rp/config.toml`
 - [x] Unix Domain Socket support (incoming requests)
 - [ ] Unix Domain Socket garbage collection (incoming requests)
-- [ ] Unix Domain Socket support (outgoing requests)
+- [x] Unix Domain Socket support (outgoing requests)
 - [ ] **not planned:** TLS termination (incoming requests)
 - [ ] **not planned:** TLS backend connections (outgoing requests)
 - [ ] **not planned:** TLS SNI passthrough
@@ -30,6 +30,10 @@ listen = "[::]:3389"
 [[mapping]]
 from = "a.localhost"
 to = "example.com"
+# The LDAP address of the backend server:
+# - start with `./` or `/` for a socket URI
+# - start with anything else for a TCP connection
+# backend = "/run/lldap/example.com.sock"
 backend = "127.0.0.1:4389"
 [[mapping]]
 from = "b.localhost"

src/client.rs

@@ -3,13 +3,13 @@ use futures_util::stream::StreamExt;
 use ldap3_proto::LdapCodec;
 use ldap3_proto::control::LdapControl;
 use ldap3_proto::proto::*;
-use tokio::net::TcpStream;
+use tokio::net::{TcpStream, UnixStream};
 use tokio::time::timeout;
 use tokio_util::codec::{FramedRead, FramedWrite};
 
 use std::time::Duration;
 
-use crate::{CR, CW, LdapError};
+use crate::{AbstractStream, CR, CW, LdapError};
 
 pub struct BasicLdapClient {
     r: FramedRead<CR, LdapCodec>,
@@ -24,24 +24,47 @@ impl BasicLdapClient {
     }
 
     pub async fn build(addr: &str) -> Result<Self, LdapError> {
-        let tcpstream = match timeout(Duration::from_secs(1), TcpStream::connect(addr)).await {
+        // If addr is a relative or absolute path, consider it's a socket
-            Ok(Ok(t)) => {
+        let stream: AbstractStream = if addr.starts_with('.') || addr.starts_with('/') {
-                trace!("connection established to {addr}");
+            let unixstream = match timeout(Duration::from_secs(1), UnixStream::connect(addr)).await
-                t
+            {
-            }
+                Ok(Ok(t)) => {
-            Ok(Err(err)) => {
+                    trace!("connection established to {addr}");
-                // trace!(?addr, ?err, "error");
+                    t
-                error!("error to {addr}: {err}");
+                }
-                panic!();
+                Ok(Err(err)) => {
-            }
+                    // trace!(?addr, ?err, "error");
-            Err(_) => {
+                    error!("error to {addr}: {err}");
-                warn!("timeout to {addr}");
+                    panic!();
-                panic!();
+                }
-                // continue;
+                Err(_) => {
-            }
+                    warn!("timeout to {addr}");
+                    panic!();
+                    // continue;
+                }
+            };
+            unixstream.into()
+        } else {
+            let tcpstream = match timeout(Duration::from_secs(1), TcpStream::connect(addr)).await {
+                Ok(Ok(t)) => {
+                    trace!("connection established to {addr}");
+                    t
+                }
+                Ok(Err(err)) => {
+                    // trace!(?addr, ?err, "error");
+                    error!("error to {addr}: {err}");
+                    panic!();
+                }
+                Err(_) => {
+                    warn!("timeout to {addr}");
+                    panic!();
+                    // continue;
+                }
+            };
+            tcpstream.into()
         };
 
-        let (r, w) = tokio::io::split(tcpstream);
+        let (r, w) = tokio::io::split(stream);
 
         let w = FramedWrite::new(w, LdapCodec::new(None, None));
         let r = FramedRead::new(r, LdapCodec::new(None, None));

src/main.rs

@@ -6,7 +6,6 @@ use futures_util::StreamExt;
 use ldap3_proto::LdapCodec;
 use ldap3_proto::proto::*;
 use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
-use tokio::net::TcpStream;
 use tokio::time::timeout;
 use tokio_util::codec::{FramedRead, FramedWrite};
 
@@ -20,13 +19,15 @@ use crate::client::BasicLdapClient;
 mod config;
 use config::Config;
 mod dn;
-mod op;
 use crate::dn::Dn;
+mod op;
+mod stream;
+use stream::AbstractStream;
 
 const LDAP_CLIENT_IO_TIMEOUT: Duration = Duration::from_secs(1);
 
-type CR = ReadHalf<TcpStream>;
+type CR = ReadHalf<AbstractStream>;
-type CW = WriteHalf<TcpStream>;
+type CW = WriteHalf<AbstractStream>;
 
 #[derive(Debug, Clone)]
 pub enum LdapError {

src/stream.rs

@@ -0,0 +1,66 @@
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::net::{TcpStream, UnixStream};
+
+use std::io::Result;
+use std::marker::Unpin;
+use std::pin::{Pin, pin};
+use std::task::{Context, Poll};
+
+pub enum AbstractStream {
+    Tcp(TcpStream),
+    Uds(UnixStream),
+}
+
+impl From<TcpStream> for AbstractStream {
+    fn from(stream: TcpStream) -> Self {
+        Self::Tcp(stream)
+    }
+}
+
+impl From<UnixStream> for AbstractStream {
+    fn from(stream: UnixStream) -> Self {
+        Self::Uds(stream)
+    }
+}
+
+impl Unpin for AbstractStream {}
+
+impl AsyncRead for AbstractStream {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<Result<()>> {
+        match &mut *self {
+            Self::Tcp(stream) => pin!(stream).poll_read(cx, buf),
+            Self::Uds(stream) => pin!(stream).poll_read(cx, buf),
+        }
+    }
+}
+
+impl AsyncWrite for AbstractStream {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize>> {
+        match &mut *self {
+            Self::Tcp(stream) => pin!(stream).poll_write(cx, buf),
+            Self::Uds(stream) => pin!(stream).poll_write(cx, buf),
+        }
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        match &mut *self {
+            Self::Tcp(stream) => pin!(stream).poll_flush(cx),
+            Self::Uds(stream) => pin!(stream).poll_flush(cx),
+        }
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        match &mut *self {
+            Self::Tcp(stream) => pin!(stream).poll_shutdown(cx),
+            Self::Uds(stream) => pin!(stream).poll_shutdown(cx),
+        }
+    }
+}