api: allow listening to Unix sockets

5 files changed, 287 insertions, 50 deletions

diff --git a/src/api/auth.rs b/src/api/auth.rs
index 48ec02d..400feed 100644
--- a/src/api/auth.rs
+++ b/src/api/auth.rs
@@ -1,6 +1,8 @@
 use anyhow::Context;
 use argon2::{Argon2, password_hash::PasswordVerifier};
 use axum::extract::{ConnectInfo, State};
+
+use super::listen::Peer;
 use axum::{
     body::Body,
     http::{
@@ -16,7 +18,7 @@ use std::{
     collections::HashMap,
     fs::File,
     io::{BufRead, BufReader},
-    net::{IpAddr, SocketAddr},
+    net::IpAddr,
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -68,7 +70,32 @@ pub(crate) struct AuthState {
 /// A reverse proxy authenticates, and names the user in `header`.
 struct ProxyAuth {
     header: HeaderName,
-    trusted: Vec<Range>,
+    trusted: Vec<Trusted>,
+}
+
+/// Where a proxy is believed from (`--trusted-proxies`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Trusted {
+    Addresses(Range),
+    /// The API's Unix sockets (`unix`).
+    Unix,
+}
+
+impl Trusted {
+    fn parse(text: &str) -> anyhow::Result<Self> {
+        if text.trim() == "unix" {
+            return Ok(Self::Unix);
+        }
+        Range::parse(text).map(Self::Addresses)
+    }
+
+    fn holds(&self, peer: Peer) -> bool {
+        match (self, peer) {
+            (Self::Addresses(range), Peer::Tcp(address)) => range.contains(address.ip()),
+            (Self::Unix, Peer::Unix) => true,
+            _ => false,
+        }
+    }
 }
 
 /// An address, or the addresses sharing its first `bits` bits.
@@ -84,9 +111,9 @@ impl Range {
             Some((address, bits)) => (address, Some(bits)),
             None => (text.trim(), None),
         };
-        let address: IpAddr = address
-            .parse()
-            .with_context(|| format!("trusted-proxies: {text:?} is no IP address"))?;
+        let address: IpAddr = address.parse().with_context(|| {
+            format!("trusted-proxies: {text:?} is neither an IP address nor unix")
+        })?;
         let all = if address.is_ipv4() { 32 } else { 128 };
         let bits = match bits {
             None => all,
@@ -140,7 +167,7 @@ impl AuthState {
                 let trusted = settings
                     .trusted_proxies
                     .iter()
-                    .map(|range| Range::parse(range))
+                    .map(|trusted| Trusted::parse(trusted))
                     .collect::<anyhow::Result<Vec<_>>>()?;
                 Ok::<_, anyhow::Error>(Arc::new(ProxyAuth { header, trusted }))
             })
@@ -174,13 +201,13 @@ fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> {
 pub(crate) struct AuthenticatedUser(pub(crate) String);
 
 /// The user the trusted proxy the request came from names, if it does.
-fn proxied_user(state: &AuthState, headers: &HeaderMap, from: Option<IpAddr>) -> Option<String> {
+fn proxied_user(state: &AuthState, headers: &HeaderMap, from: Option<Peer>) -> Option<String> {
     let proxy = state.proxy.as_ref()?;
     let user = headers.get(&proxy.header)?.to_str().ok()?.trim();
     if user.is_empty() {
         return None;
     }
-    let trusted = from.is_some_and(|from| proxy.trusted.iter().any(|range| range.contains(from)));
+    let trusted = from.is_some_and(|from| proxy.trusted.iter().any(|trusted| trusted.holds(from)));
     if !trusted {
         tracing::debug!(from = ?from, header = %proxy.header, "User header from an address not in trusted-proxies; ignored");
         return None;
@@ -247,8 +274,8 @@ pub(crate) async fn basic_auth(
 ) -> Response {
     let from = request
         .extensions()
-        .get::<ConnectInfo<SocketAddr>>()
-        .map(|ConnectInfo(address)| address.ip());
+        .get::<ConnectInfo<Peer>>()
+        .map(|ConnectInfo(peer)| *peer);
     let user = match proxied_user(&state, &headers, from) {
         Some(user) => Some(user),
         None => authenticate(&state, &headers).await,
diff --git a/src/api/listen.rs b/src/api/listen.rs
new file mode 100644
index 0000000..df2df22
--- /dev/null
+++ b/src/api/listen.rs
@@ -0,0 +1,220 @@
+//! Where the HTTP API listens (`--bind-addr`): TCP addresses, and Unix sockets
+//! (`unix:PATH`), any number of them at once, all serving the same API. A request
+//! carries where it came from (`Peer`), which decides whether a reverse proxy's user
+//! header is believed (`api::auth`, `--trusted-proxies`): on a Unix socket, its file's
+//! permissions decide who may connect at all, which on one host is what an address
+//! can't.
+
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::str::FromStr;
+
+use anyhow::{Context, Result};
+use axum::Router;
+use axum::extract::connect_info::Connected;
+use axum::serve::IncomingStream;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use tracing::{info, warn};
+
+/// One place the API listens.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Bind {
+    Tcp(SocketAddr),
+    /// A Unix socket at this path, made by the node (readable and writable by its user
+    /// and group only), and removed when it stops.
+    Unix(PathBuf),
+}
+
+const UNIX: &str = "unix:";
+
+impl FromStr for Bind {
+    type Err = anyhow::Error;
+
+    fn from_str(text: &str) -> Result<Self> {
+        let text = text.trim();
+        match text.strip_prefix(UNIX) {
+            Some(path) if !path.is_empty() => Ok(Self::Unix(PathBuf::from(path))),
+            Some(_) => {
+                anyhow::bail!("bind-addr: unix: needs a path (unix:/run/continuum/api.sock)")
+            }
+            None => text.parse().map(Self::Tcp).with_context(|| {
+                format!("bind-addr: {text:?} is neither ADDRESS:PORT nor unix:PATH")
+            }),
+        }
+    }
+}
+
+impl std::fmt::Display for Bind {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Tcp(address) => write!(f, "{address}"),
+            Self::Unix(path) => write!(f, "{UNIX}{}", path.display()),
+        }
+    }
+}
+
+impl Serialize for Bind {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        serializer.collect_str(self)
+    }
+}
+
+impl<'de> Deserialize<'de> for Bind {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        let text = String::deserialize(deserializer)?;
+        text.parse()
+            .map_err(|error: anyhow::Error| serde::de::Error::custom(format!("{error:#}")))
+    }
+}
+
+/// `--bind-addr`: one place (as before it took more), or several.
+pub(crate) fn one_or_more<'de, D: Deserializer<'de>>(
+    deserializer: D,
+) -> Result<Vec<Bind>, D::Error> {
+    #[derive(Deserialize)]
+    #[serde(untagged)]
+    enum OneOrMore {
+        One(Bind),
+        More(Vec<Bind>),
+    }
+    Ok(match OneOrMore::deserialize(deserializer)? {
+        OneOrMore::One(bind) => vec![bind],
+        OneOrMore::More(binds) => binds,
+    })
+}
+
+/// Where a request came from (`axum::extract::ConnectInfo`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Peer {
+    Tcp(SocketAddr),
+    /// Over one of the API's Unix sockets.
+    Unix,
+}
+
+impl Connected<IncomingStream<'_, tokio::net::TcpListener>> for Peer {
+    fn connect_info(stream: IncomingStream<'_, tokio::net::TcpListener>) -> Self {
+        Self::Tcp(*stream.remote_addr())
+    }
+}
+
+#[cfg(unix)]
+impl Connected<IncomingStream<'_, tokio::net::UnixListener>> for Peer {
+    fn connect_info(_stream: IncomingStream<'_, tokio::net::UnixListener>) -> Self {
+        Self::Unix
+    }
+}
+
+/// Serves `app` at every one of `binds` until `stop` resolves (each server then drains
+/// its connections); resolves once all have stopped, with the first error if any did.
+/// Binds them all before it returns: one that can't be bound is an error at once.
+pub(crate) async fn serve<F>(
+    binds: &[Bind],
+    app: Router,
+    stop: impl Fn() -> F,
+) -> Result<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>>
+where
+    F: Future<Output = ()> + Send + 'static,
+{
+    let mut servers = tokio::task::JoinSet::new();
+    for bind in binds {
+        let service = app.clone().into_make_service_with_connect_info::<Peer>();
+        let stopped = stop();
+        match bind {
+            Bind::Tcp(address) => {
+                let listener = tokio::net::TcpListener::bind(address)
+                    .await
+                    .with_context(|| format!("Binding the HTTP API to {address}"))?;
+                info!(address = %listener.local_addr()?, "HTTP API listening");
+                servers.spawn(async move {
+                    axum::serve(listener, service)
+                        .with_graceful_shutdown(stopped)
+                        .await
+                        .map_err(anyhow::Error::from)
+                });
+            }
+            #[cfg(unix)]
+            Bind::Unix(path) => {
+                let listener = bind_unix(path)?;
+                info!(socket = %path.display(), "HTTP API listening");
+                let path = path.clone();
+                servers.spawn(async move {
+                    let served = axum::serve(listener, service)
+                        .with_graceful_shutdown(stopped)
+                        .await;
+                    if let Err(error) = std::fs::remove_file(&path) {
+                        warn!(socket = %path.display(), error = %error, "Couldn't remove the HTTP API's socket");
+                    }
+                    served.map_err(anyhow::Error::from)
+                });
+            }
+            #[cfg(not(unix))]
+            Bind::Unix(_) => anyhow::bail!("bind-addr: Unix sockets need a Unix"),
+        }
+    }
+    Ok(Box::pin(async move {
+        let mut first = Ok(());
+        while let Some(served) = servers.join_next().await {
+            let served = served
+                .map_err(anyhow::Error::from)
+                .and_then(|served| served);
+            if let Err(error) = served
+                && first.is_ok()
+            {
+                first = Err(error);
+            }
+        }
+        first
+    }))
+}
+
+/// A Unix socket at `path`: one left there by an earlier run is replaced, anything else
+/// there is an error. Its user and group may connect, nobody else.
+#[cfg(unix)]
+fn bind_unix(path: &std::path::Path) -> Result<tokio::net::UnixListener> {
+    use std::os::unix::fs::{FileTypeExt, PermissionsExt};
+    match std::fs::symlink_metadata(path) {
+        Ok(metadata) if metadata.file_type().is_socket() => std::fs::remove_file(path)
+            .with_context(|| format!("Removing the old socket {}", path.display()))?,
+        Ok(_) => anyhow::bail!(
+            "bind-addr: {} exists and is no socket; not replacing it",
+            path.display()
+        ),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+        Err(error) => {
+            return Err(error).with_context(|| format!("Looking at {}", path.display()));
+        }
+    }
+    let listener = tokio::net::UnixListener::bind(path)
+        .with_context(|| format!("Binding the HTTP API to {}", path.display()))?;
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660))
+        .with_context(|| format!("Setting who may use {}", path.display()))?;
+    Ok(listener)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Bind;
+
+    #[test]
+    fn binds_parse() {
+        assert_eq!(
+            "127.0.0.1:3000".parse::<Bind>().unwrap(),
+            Bind::Tcp("127.0.0.1:3000".parse().unwrap())
+        );
+        assert_eq!(
+            "unix:/run/continuum/api.sock".parse::<Bind>().unwrap(),
+            Bind::Unix("/run/continuum/api.sock".into())
+        );
+        assert_eq!(
+            "unix:/run/continuum/api.sock"
+                .parse::<Bind>()
+                .unwrap()
+                .to_string(),
+            "unix:/run/continuum/api.sock"
+        );
+        for bad in ["unix:", "localhost", "3000"] {
+            assert!(bad.parse::<Bind>().is_err(), "{bad}");
+        }
+    }
+}
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 35b1a06..e4ccee6 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -9,6 +9,7 @@ mod dashboard;
 mod debug;
 mod document;
 mod health;
+pub(crate) mod listen;
 mod log;
 mod peers;
 mod stats;
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 527dced..a1011ee 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -90,10 +90,10 @@ pub(crate) const FORCE: &[&str] = &["swarmId", "badDocumentFile", "replaceDocume
 pub(crate) const OPTIONS: &[OptionSpec] = &[
     option(
         "bind-addr",
-        Kind::Text,
-        "HOSTNAME:PORT",
+        Kind::List,
+        "ADDRESS:PORT|unix:PATH,[...]",
         Some("127.0.0.1:3000"),
-        "Address of the HTTP API.",
+        "Where the HTTP API listens: TCP addresses, and Unix sockets (unix:PATH, made readable and writable by the node's user and group only), as many as given.",
     ),
     option(
         "htpasswd",
@@ -113,8 +113,8 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "trusted-proxies",
         Kind::List,
         "ADDRESS,[ADDRESS...]",
-        Some("127.0.0.1,::1"),
-        "Addresses (an IP address, or a range as ADDRESS/BITS) of the reverse proxies whose user-header is believed.",
+        Some("127.0.0.1,::1,unix"),
+        "Where the reverse proxies whose user-header is believed connect from: IP addresses (or ranges, as ADDRESS/BITS), and unix for the API's Unix sockets (bind-addr). On a host with other users, leave out the addresses and have the proxy use a Unix socket: others can connect over loopback too.",
     ),
     option(
         "protocol-writers",
@@ -501,8 +501,11 @@ pub(crate) fn spec(name: &str) -> Option<&'static OptionSpec> {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "kebab-case", deny_unknown_fields)]
 pub(crate) struct Settings {
-    #[serde(default = "default_bind_addr")]
-    pub(crate) bind_addr: SocketAddr,
+    #[serde(
+        default = "default_bind_addr",
+        deserialize_with = "crate::api::listen::one_or_more"
+    )]
+    pub(crate) bind_addr: Vec<crate::api::listen::Bind>,
     /// Required (`check`), but with `user_header` or `relay_only`.
     pub(crate) htpasswd: Option<String>,
     pub(crate) user_header: Option<String>,
@@ -643,11 +646,14 @@ pub(crate) struct Settings {
     pub(crate) debug_epoch: Option<u64>,
 }
 
-fn default_bind_addr() -> SocketAddr {
-    SocketAddr::from(([127, 0, 0, 1], 3000))
+fn default_bind_addr() -> Vec<crate::api::listen::Bind> {
+    vec![crate::api::listen::Bind::Tcp(SocketAddr::from((
+        [127, 0, 0, 1],
+        3000,
+    )))]
 }
 fn default_trusted_proxies() -> Vec<String> {
-    vec!["127.0.0.1".to_owned(), "::1".to_owned()]
+    vec!["127.0.0.1".to_owned(), "::1".to_owned(), "unix".to_owned()]
 }
 fn default_document_append() -> Duration {
     Duration::from_secs(5)
diff --git a/src/main.rs b/src/main.rs
index c474d45..49971fb 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -4,7 +4,6 @@
 //! main.rs only wires things together: arguments, the starting document, the node
 //! (`node`), bootstrap and the HTTP server.
 
-use std::future::IntoFuture;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
@@ -98,7 +97,7 @@ async fn main() -> Result<()> {
         AuthState::from_settings(&args)?.context("htpasswd or user-header is required")?;
 
     info!(
-        bind_addr = args.bind_addr.to_string(),
+        bind_addr = ?args.bind_addr.iter().map(ToString::to_string).collect::<Vec<_>>(),
         transports = ?args.transports,
         "Starting replication service"
     );
@@ -262,23 +261,16 @@ async fn main() -> Result<()> {
     let joined = AtomicBool::new(false);
     let app = api::router(state.clone(), auth_state, args.debug_api);
 
-    let listener = tokio::net::TcpListener::bind(args.bind_addr)
-        .await
-        .with_context(|| format!("Binding local API to {}", args.bind_addr))?;
-
-    info!(address = %listener.local_addr()?, "HTTP API listening");
-
-    // The node shutting down (`AppState::closing`): the server drains.
+    // The node shutting down (`AppState::closing`): the servers drain.
     let closing = state.closing.subscribe();
-    // With the address each request came from: a proxy's user header is believed from
-    // `--trusted-proxies` only (`api::auth`).
-    let app = app.into_make_service_with_connect_info::<std::net::SocketAddr>();
-    let server_builder = axum::serve(listener, app).with_graceful_shutdown(async move {
-        stopped(closing).await;
-        info!("Shutdown signal received. Draining API connections...")
-    });
-
-    let server_fut = server_builder.into_future();
+    let server_fut = api::listen::serve(&args.bind_addr, app, move || {
+        let closing = closing.clone();
+        async move {
+            stopped(closing).await;
+            info!("Shutdown signal received. Draining API connections...")
+        }
+    })
+    .await?;
     tokio::pin!(server_fut);
 
     // Joins, then runs until stopped; returns only if joining fails.
@@ -350,19 +342,10 @@ async fn relay_only(args: &config::Settings) -> Result<()> {
     let server = match auth_state {
         Some(auth_state) => {
             let app = api::relay_only_router(Arc::clone(&transport), auth_state);
-            let listener = tokio::net::TcpListener::bind(args.bind_addr)
-                .await
-                .with_context(|| format!("Binding local API to {}", args.bind_addr))?;
-            info!(address = %listener.local_addr()?, "HTTP API listening");
             let stop = stop.clone();
-            Some(tokio::spawn(async move {
-                axum::serve(
-                    listener,
-                    app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
-                )
-                .with_graceful_shutdown(stopped(stop))
-                .await
-            }))
+            let served =
+                api::listen::serve(&args.bind_addr, app, move || stopped(stop.clone())).await?;
+            Some(tokio::spawn(served))
         }
         None => {
             info!("No htpasswd or user-header: no HTTP API");