From cfa4b8a43706125e9999be8219530abccb7661cc Mon Sep 17 00:00:00 2001 From: selfhoster1312 Date: Fri, 21 Aug 2026 15:14:52 +0200 Subject: [PATCH] feat: Support backend socket connections (UDS) --- README.md | 6 ++++- src/client.rs | 59 +++++++++++++++++++++++++++++++-------------- src/main.rs | 9 +++---- src/stream.rs | 66 +++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 117 insertions(+), 23 deletions(-) create mode 100644 src/stream.rs diff --git a/README.md b/README.md index c990a7e..a5d019e 100644 --- a/README.md +++ b/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" diff --git a/src/client.rs b/src/client.rs index a61f014..870d622 100644 --- a/src/client.rs +++ b/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, @@ -24,24 +24,47 @@ impl BasicLdapClient { } pub async fn build(addr: &str) -> Result { - 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; - } + // If addr is a relative or absolute path, consider it's a socket + let stream: AbstractStream = if addr.starts_with('.') || addr.starts_with('/') { + let unixstream = match timeout(Duration::from_secs(1), UnixStream::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; + } + }; + 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)); diff --git a/src/main.rs b/src/main.rs index a32df7f..403e403 100644 --- a/src/main.rs +++ b/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; -type CW = WriteHalf; +type CR = ReadHalf; +type CW = WriteHalf; #[derive(Debug, Clone)] pub enum LdapError { diff --git a/src/stream.rs b/src/stream.rs new file mode 100644 index 0000000..d5fd782 --- /dev/null +++ b/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 for AbstractStream { + fn from(stream: TcpStream) -> Self { + Self::Tcp(stream) + } +} + +impl From 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> { + 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> { + 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> { + 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> { + match &mut *self { + Self::Tcp(stream) => pin!(stream).poll_shutdown(cx), + Self::Uds(stream) => pin!(stream).poll_shutdown(cx), + } + } +}