api: allow listening to Unix sockets

3 files changed, 258 insertions, 10 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;