feat: Initial accept loop

src/cli.rs

@@ -0,0 +1,9 @@
+use argh::FromArgs;
+
+/// Run the llldap server
+#[derive(FromArgs)]
+pub struct CliArgs {
+    /// address or socket to listen on
+    #[argh(positional, default = "String::from(\"127.0.0.1:3389\")")]
+    pub listen: String,
+}

src/error.rs

@@ -0,0 +1,35 @@
+use tracing::dispatcher::SetGlobalDefaultError;
+
+use std::fmt;
+
+use crate::listener::{AcceptError, InvalidListenerError, ListenError};
+
+#[derive(Debug)]
+pub enum GlobalError {
+    /// The provided listen config is invalid
+    Listener(InvalidListenerError),
+    /// Failed to listen on the provided addr/socket
+    Listen(ListenError),
+    /// Failed to receive a new connection on the provided addr/socket
+    Accept(AcceptError),
+    Tracing(SetGlobalDefaultError),
+}
+
+impl fmt::Display for GlobalError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Listener(e) => e.fmt(f),
+            Self::Listen(e) => e.fmt(f),
+            Self::Accept(e) => e.fmt(f),
+            Self::Tracing(e) => write!(f, "Failed to setup logging: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for GlobalError {}
+
+impl From<SetGlobalDefaultError> for GlobalError {
+    fn from(e: SetGlobalDefaultError) -> Self {
+        Self::Tracing(e)
+    }
+}

src/listener.rs

@@ -0,0 +1,259 @@
+use camino::Utf8PathBuf;
+use tokio::net::{TcpListener, UnixListener};
+
+use std::fmt;
+use std::net::SocketAddr;
+use std::str::FromStr;
+
+use crate::error::GlobalError;
+use crate::stream::AbstractStream;
+
+#[derive(Debug)]
+pub struct AcceptError {
+    pub listener: ListenerPath,
+    pub error: std::io::Error,
+}
+
+impl From<AcceptError> for GlobalError {
+    fn from(e: AcceptError) -> Self {
+        Self::Accept(e)
+    }
+}
+
+impl fmt::Display for AcceptError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "Failed to accept new connection on {}: {:?}",
+            self.listener, self.error
+        )
+    }
+}
+
+impl std::error::Error for AcceptError {}
+
+#[derive(Debug)]
+pub struct ListenError {
+    pub listener: ListenerPath,
+    pub error: std::io::Error,
+}
+
+impl From<ListenError> for GlobalError {
+    fn from(e: ListenError) -> Self {
+        Self::Listen(e)
+    }
+}
+
+impl fmt::Display for ListenError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Failed to listen to {}: {:?}", self.listener, self.error)
+    }
+}
+
+impl std::error::Error for ListenError {}
+
+#[derive(Debug)]
+pub struct InvalidListenerError(pub String);
+
+impl From<InvalidListenerError> for GlobalError {
+    fn from(e: InvalidListenerError) -> Self {
+        Self::Listener(e)
+    }
+}
+
+impl fmt::Display for InvalidListenerError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "Invalid listener (use ./foo.sock or /foo.sock for unix domain sockets) : {})",
+            self.0
+        )
+    }
+}
+
+#[derive(Debug)]
+pub enum ListenerKind {
+    Tcp(TcpListener),
+    Uds(UnixListener),
+}
+
+#[derive(Debug)]
+pub struct Listener {
+    pub kind: ListenerKind,
+    pub path: ListenerPath,
+}
+
+impl Listener {
+    pub fn new_tcp(path: ListenerPath, listener: TcpListener) -> Self {
+        Self {
+            kind: ListenerKind::Tcp(listener),
+            path,
+        }
+    }
+
+    pub fn new_uds(path: ListenerPath, listener: UnixListener) -> Self {
+        Self {
+            kind: ListenerKind::Uds(listener),
+            path,
+        }
+    }
+
+    pub async fn accept(&self) -> Result<Option<AbstractStream>, AcceptError> {
+        let res = match &self.kind {
+            ListenerKind::Tcp(l) => l.accept().await.map(AbstractStream::from),
+            ListenerKind::Uds(l) => l.accept().await.map(AbstractStream::from),
+        };
+
+        match res {
+            Ok(stream) => Ok(Some(stream)),
+            Err(e) => {
+                // Here the error could be fatal, or could simply be that a client aborted the connected,
+                // in which case we don't want to crash the server, simply skip this client connection.
+                // https://doc.rust-lang.org/stable/std/net/struct.TcpListener.html#errors
+                match e.kind() {
+                    std::io::ErrorKind::ConnectionAborted => Ok(None),
+                    _ => Err(AcceptError {
+                        listener: self.path.clone(),
+                        error: e,
+                    }),
+                }
+            }
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub enum ListenerPath {
+    Tcp(SocketAddr),
+    Uds(Utf8PathBuf),
+}
+
+impl ListenerPath {
+    pub fn new(s: &str) -> Result<Self, InvalidListenerError> {
+        Self::from_str(s)
+    }
+
+    pub async fn listener(self) -> Result<Listener, ListenError> {
+        match &self {
+            Self::Tcp(p) => TcpListener::bind(p)
+                .await
+                .map(|l| Listener::new_tcp(self.clone(), l)),
+            Self::Uds(p) => UnixListener::bind(p).map(|l| Listener::new_uds(self.clone(), l)),
+        }
+        .map_err(|e| ListenError {
+            listener: self.clone(),
+            error: e,
+        })
+    }
+}
+
+impl fmt::Display for ListenerPath {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Tcp(s) => s.fmt(f),
+            Self::Uds(s) => s.fmt(f),
+        }
+    }
+}
+
+impl FromStr for ListenerPath {
+    type Err = InvalidListenerError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if s.starts_with('.') || s.starts_with('/') {
+            Ok(Self::Uds(Utf8PathBuf::from(s)))
+        } else {
+            let Ok(addr) = SocketAddr::from_str(s) else {
+                return Err(InvalidListenerError(s.to_string()));
+            };
+
+            Ok(Self::Tcp(addr))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
+
+    use super::*;
+
+    #[test]
+    fn valid_relative_socket() {
+        let s = "./foo.sock";
+        let res = ListenerPath::from_str(s);
+        println!("{:?}", res);
+        let p = res.unwrap();
+        assert_eq!(p, ListenerPath::Uds(Utf8PathBuf::from(s)));
+    }
+
+    #[test]
+    fn valid_absolute_socket() {
+        let s = "/foo.sock";
+        let res = ListenerPath::from_str(s);
+        println!("{:?}", res);
+        let p = res.unwrap();
+        assert_eq!(p, ListenerPath::Uds(Utf8PathBuf::from(s)));
+    }
+
+    #[test]
+    fn valid_ipv4_localhost() {
+        let s = "127.0.0.1:3389";
+        let res = ListenerPath::from_str(s);
+        println!("{:?}", res);
+        let p = res.unwrap();
+        assert_eq!(
+            p,
+            ListenerPath::Tcp(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 3389),))
+        );
+    }
+
+    #[test]
+    fn valid_ipv4_any() {
+        let s = "0.0.0.0:3389";
+        let res = ListenerPath::from_str(s);
+        println!("{:?}", res);
+        let p = res.unwrap();
+        assert_eq!(
+            p,
+            ListenerPath::Tcp(SocketAddr::V4(SocketAddrV4::new(
+                Ipv4Addr::UNSPECIFIED,
+                3389
+            )))
+        );
+    }
+
+    #[test]
+    fn valid_ipv6_localhost() {
+        let s = "[::1]:3389";
+        let res = ListenerPath::from_str(s);
+        println!("{:?}", res);
+        let p = res.unwrap();
+        assert_eq!(
+            p,
+            ListenerPath::Tcp(SocketAddr::V6(SocketAddrV6::new(
+                Ipv6Addr::LOCALHOST,
+                3389,
+                0,
+                0
+            ),))
+        );
+    }
+
+    #[test]
+    fn valid_ipv6_any() {
+        let s = "[::]:3389";
+        let res = ListenerPath::from_str(s);
+        println!("{:?}", res);
+        let p = res.unwrap();
+        assert_eq!(
+            p,
+            ListenerPath::Tcp(SocketAddr::V6(SocketAddrV6::new(
+                Ipv6Addr::UNSPECIFIED,
+                3389,
+                0,
+                0
+            ),))
+        );
+    }
+}

src/main.rs

@@ -0,0 +1,42 @@
+// #![deny(warnings)]
+
+mod cli;
+mod error;
+mod listener;
+mod stream;
+
+use cli::CliArgs;
+use error::GlobalError;
+use listener::ListenerPath;
+use stream::AbstractStream;
+
+#[tracing::instrument(name = "client", skip(s), fields(session = %s.session))]
+async fn client(s: AbstractStream) {
+    tracing::info! {
+        remote_addr = ?s.remote_addr,
+        "New client connection"
+    };
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), GlobalError> {
+    let cli: CliArgs = argh::from_env();
+
+    tracing_subscriber::fmt::init();
+    // let subscriber = tracing_subscriber::FmtSubscriber::new();
+    //     .with_file(true)
+    //     .with_line_number(true)
+    //     .finish()
+    //     .init();
+    // tracing::subscriber::set_global_default(subscriber)?;
+
+    let listener = ListenerPath::new(&cli.listen)?.listener().await?;
+
+    // If the connection is None, it's because the client aborted early
+    // so there's nothing to do about it.
+    while let Some(stream) = listener.accept().await? {
+        client(stream).await;
+    }
+
+    Ok(())
+}

src/stream.rs

@@ -0,0 +1,118 @@
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::net::unix::SocketAddr as UnixSocketAddr;
+use tokio::net::{TcpStream, UnixStream};
+use uuid::Uuid;
+
+use std::io::Result;
+use std::marker::Unpin;
+use std::net::SocketAddr as TcpSocketAddr;
+use std::pin::{Pin, pin};
+use std::task::{Context, Poll};
+
+#[derive(Debug)]
+pub enum AbstractSocketAddr {
+    #[expect(unused)]
+    Tcp(TcpSocketAddr),
+    #[expect(unused)]
+    Uds(UnixSocketAddr),
+}
+
+impl From<TcpSocketAddr> for AbstractSocketAddr {
+    fn from(a: TcpSocketAddr) -> Self {
+        Self::Tcp(a)
+    }
+}
+
+impl From<UnixSocketAddr> for AbstractSocketAddr {
+    fn from(a: UnixSocketAddr) -> Self {
+        Self::Uds(a)
+    }
+}
+
+#[derive(Debug)]
+pub enum AbstractStreamKind {
+    Tcp(TcpStream),
+    Uds(UnixStream),
+}
+
+impl From<TcpStream> for AbstractStreamKind {
+    fn from(stream: TcpStream) -> Self {
+        Self::Tcp(stream)
+    }
+}
+
+impl From<UnixStream> for AbstractStreamKind {
+    fn from(stream: UnixStream) -> Self {
+        Self::Uds(stream)
+    }
+}
+
+#[derive(Debug)]
+pub struct AbstractStream {
+    pub remote_addr: AbstractSocketAddr,
+    pub kind: AbstractStreamKind,
+    /// An arbitrary ID to correlate logs with a specific TCP session
+    pub session: Uuid,
+}
+
+impl From<(UnixStream, UnixSocketAddr)> for AbstractStream {
+    fn from(res: (UnixStream, UnixSocketAddr)) -> Self {
+        Self {
+            remote_addr: res.1.into(),
+            kind: res.0.into(),
+            session: Uuid::new_v4(),
+        }
+    }
+}
+
+impl From<(TcpStream, TcpSocketAddr)> for AbstractStream {
+    fn from(res: (TcpStream, std::net::SocketAddr)) -> Self {
+        Self {
+            remote_addr: res.1.into(),
+            kind: res.0.into(),
+            session: Uuid::new_v4(),
+        }
+    }
+}
+
+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.kind {
+            AbstractStreamKind::Tcp(stream) => pin!(stream).poll_read(cx, buf),
+            AbstractStreamKind::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.kind {
+            AbstractStreamKind::Tcp(stream) => pin!(stream).poll_write(cx, buf),
+            AbstractStreamKind::Uds(stream) => pin!(stream).poll_write(cx, buf),
+        }
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        match &mut self.kind {
+            AbstractStreamKind::Tcp(stream) => pin!(stream).poll_flush(cx),
+            AbstractStreamKind::Uds(stream) => pin!(stream).poll_flush(cx),
+        }
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+        match &mut self.kind {
+            AbstractStreamKind::Tcp(stream) => pin!(stream).poll_shutdown(cx),
+            AbstractStreamKind::Uds(stream) => pin!(stream).poll_shutdown(cx),
+        }
+    }
+}