support Unix sockets (#133)

server/src/cmds/run/config.rs

@@ -77,9 +77,9 @@ pub enum AddressConfig {
 
     /// IPv6 address such as `[::]:8080` or `[::1]:8080`.
     Ipv6(std::net::SocketAddrV6),
-    // TODO: /// Unix socket path such as `/var/lib/moonfire-nvr/sock`.
-    // Unix(PathBuf),
 
+    /// Unix socket path such as `/var/lib/moonfire-nvr/sock`.
+    Unix(PathBuf),
     // TODO: SystemdFileDescriptorName(String), see
     // https://www.freedesktop.org/software/systemd/man/systemd.socket.html
 }

server/src/cmds/run/mod.rs

@@ -5,6 +5,7 @@
 use crate::cmds::run::config::Permissions;
 use crate::streamer;
 use crate::web;
+use crate::web::accept::Listener;
 use base::clock;
 use db::{dir, writer};
 use failure::{bail, Error, ResultExt};
@@ -12,6 +13,7 @@ use fnv::FnvHashMap;
 use hyper::service::{make_service_fn, service_fn};
 use log::error;
 use log::{info, warn};
+use std::net::SocketAddr;
 use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -182,6 +184,26 @@ async fn async_run(read_only: bool, config: &ConfigFile) -> Result<i32, Error> {
     }
 }
 
+fn make_listener(addr: &config::AddressConfig) -> Result<Listener, Error> {
+    let sa: SocketAddr = match addr {
+        config::AddressConfig::Ipv4(a) => a.clone().into(),
+        config::AddressConfig::Ipv6(a) => a.clone().into(),
+        config::AddressConfig::Unix(p) => {
+            return Ok(Listener::Unix(
+                tokio::net::UnixListener::bind(p)
+                    .with_context(|_| format!("unable bind Unix socket {}", p.display()))?,
+            ));
+        }
+    };
+
+    // Go through std::net::TcpListener to avoid needing async. That's there for DNS resolution,
+    // but it's unnecessary when starting from a SocketAddr.
+    let listener = std::net::TcpListener::bind(&sa)
+        .with_context(|_| format!("unable to bind TCP socket {}", &sa))?;
+    listener.set_nonblocking(true)?;
+    Ok(Listener::Tcp(tokio::net::TcpListener::from_std(listener)?))
+}
+
 async fn inner(
     read_only: bool,
     config: &ConfigFile,
@@ -334,14 +356,8 @@ async fn inner(
                     move |req| Arc::clone(&svc).serve(req)
                 }))
             });
-            let socket_addr = match b.address {
+            let listener = make_listener(&b.address)?;
-                config::AddressConfig::Ipv4(a) => a.into(),
+            let server = ::hyper::Server::builder(listener).serve(make_svc);
-                config::AddressConfig::Ipv6(a) => a.into(),
-            };
-            let server = ::hyper::Server::try_bind(&socket_addr)
-                .with_context(|_| format!("unable to bind to {}", &socket_addr))?
-                .tcp_nodelay(true)
-                .serve(make_svc);
             let server = server.with_graceful_shutdown(shutdown_rx.future());
             Ok(tokio::spawn(server))
         })

server/src/web/accept.rs

@@ -0,0 +1,122 @@
+// This file is part of Moonfire NVR, a security camera network video recorder.
+// Copyright (C) 2021 The Moonfire NVR Authors; see AUTHORS and LICENSE.txt.
+// SPDX-License-Identifier: GPL-v3.0-or-later WITH GPL-3.0-linking-exception.
+
+//! Unified [`hyper::server::accept::Accept`] impl for TCP and Unix sockets.
+
+use std::pin::Pin;
+
+use hyper::server::accept::Accept;
+
+pub enum Listener {
+    Tcp(tokio::net::TcpListener),
+    Unix(tokio::net::UnixListener),
+}
+
+impl Accept for Listener {
+    type Conn = Conn;
+    type Error = std::io::Error;
+
+    fn poll_accept(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Result<Self::Conn, Self::Error>>> {
+        match Pin::into_inner(self) {
+            Listener::Tcp(l) => Pin::new(l).poll_accept(cx)?.map(|(s, a)| {
+                if let Err(e) = s.set_nodelay(true) {
+                    return Some(Err(e));
+                }
+                Some(Ok(Conn {
+                    stream: Stream::Tcp(s),
+                    client_unix_uid: None,
+                    client_addr: Some(a),
+                }))
+            }),
+            Listener::Unix(l) => Pin::new(l).poll_accept(cx)?.map(|(s, _a)| {
+                let ucred = match s.peer_cred() {
+                    Err(e) => return Some(Err(e)),
+                    Ok(ucred) => ucred,
+                };
+                Some(Ok(Conn {
+                    stream: Stream::Unix(s),
+                    client_unix_uid: Some(ucred.uid()),
+                    client_addr: None,
+                }))
+            }),
+        }
+    }
+}
+
+/// An open connection.
+pub struct Conn {
+    stream: Stream,
+    client_unix_uid: Option<libc::uid_t>,
+    client_addr: Option<std::net::SocketAddr>,
+}
+
+impl Conn {
+    #[allow(dead_code)] // TODO: feed this onward.
+    pub fn client_unix_uid(&self) -> Option<libc::uid_t> {
+        self.client_unix_uid
+    }
+
+    #[allow(dead_code)] // TODO: feed this onward.
+    pub fn client_addr(&self) -> Option<&std::net::SocketAddr> {
+        self.client_addr.as_ref()
+    }
+}
+
+impl tokio::io::AsyncRead for Conn {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        match self.stream {
+            Stream::Tcp(ref mut s) => Pin::new(s).poll_read(cx, buf),
+            Stream::Unix(ref mut s) => Pin::new(s).poll_read(cx, buf),
+        }
+    }
+}
+
+impl tokio::io::AsyncWrite for Conn {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<Result<usize, std::io::Error>> {
+        match self.stream {
+            Stream::Tcp(ref mut s) => Pin::new(s).poll_write(cx, buf),
+            Stream::Unix(ref mut s) => Pin::new(s).poll_write(cx, buf),
+        }
+    }
+
+    fn poll_flush(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        match self.stream {
+            Stream::Tcp(ref mut s) => Pin::new(s).poll_flush(cx),
+            Stream::Unix(ref mut s) => Pin::new(s).poll_flush(cx),
+        }
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        match self.stream {
+            Stream::Tcp(ref mut s) => Pin::new(s).poll_shutdown(cx),
+            Stream::Unix(ref mut s) => Pin::new(s).poll_shutdown(cx),
+        }
+    }
+}
+
+/// An open stream.
+///
+/// Ultimately `Tcp` and `Unix` result in the same syscalls, but using an
+/// `enum` seems easier for the moment than fighting the tokio API.
+enum Stream {
+    Tcp(tokio::net::TcpStream),
+    Unix(tokio::net::UnixStream),
+}

server/src/web/mod.rs

@@ -2,6 +2,7 @@
 // Copyright (C) 2021 The Moonfire NVR Authors; see AUTHORS and LICENSE.txt.
 // SPDX-License-Identifier: GPL-v3.0-or-later WITH GPL-3.0-linking-exception.
 
+pub mod accept;
 mod live;
 mod path;
 mod session;