From e4331d13f313f439aff9cef1ea221c8adeeebb3a Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 2 Nov 2026 22:46:48 +0100 Subject: api: allow listening to Unix sockets --- src/api/auth.rs | 47 +++++++++--- src/api/listen.rs | 220 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/api/mod.rs | 1 + src/config/mod.rs | 26 ++++--- src/main.rs | 43 ++++------- 5 files changed, 287 insertions(+), 50 deletions(-) create mode 100644 src/api/listen.rs (limited to 'src') 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, + trusted: Vec, +} + +/// 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 { + 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::>>()?; 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) -> Option { +fn proxied_user(state: &AuthState, headers: &HeaderMap, from: Option) -> Option { 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::>() - .map(|ConnectInfo(address)| address.ip()); + .get::>() + .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 { + 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(&self, serializer: S) -> Result { + serializer.collect_str(self) + } +} + +impl<'de> Deserialize<'de> for Bind { + fn deserialize>(deserializer: D) -> Result { + 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, D::Error> { + #[derive(Deserialize)] + #[serde(untagged)] + enum OneOrMore { + One(Bind), + More(Vec), + } + 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> for Peer { + fn connect_info(stream: IncomingStream<'_, tokio::net::TcpListener>) -> Self { + Self::Tcp(*stream.remote_addr()) + } +} + +#[cfg(unix)] +impl Connected> 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( + binds: &[Bind], + app: Router, + stop: impl Fn() -> F, +) -> Result> + Send>>> +where + F: Future + Send + 'static, +{ + let mut servers = tokio::task::JoinSet::new(); + for bind in binds { + let service = app.clone().into_make_service_with_connect_info::(); + 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 { + 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::().unwrap(), + Bind::Tcp("127.0.0.1:3000".parse().unwrap()) + ); + assert_eq!( + "unix:/run/continuum/api.sock".parse::().unwrap(), + Bind::Unix("/run/continuum/api.sock".into()) + ); + assert_eq!( + "unix:/run/continuum/api.sock" + .parse::() + .unwrap() + .to_string(), + "unix:/run/continuum/api.sock" + ); + for bad in ["unix:", "localhost", "3000"] { + assert!(bad.parse::().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, /// Required (`check`), but with `user_header` or `relay_only`. pub(crate) htpasswd: Option, pub(crate) user_header: Option, @@ -643,11 +646,14 @@ pub(crate) struct Settings { pub(crate) debug_epoch: Option, } -fn default_bind_addr() -> SocketAddr { - SocketAddr::from(([127, 0, 0, 1], 3000)) +fn default_bind_addr() -> Vec { + vec![crate::api::listen::Bind::Tcp(SocketAddr::from(( + [127, 0, 0, 1], + 3000, + )))] } fn default_trusted_proxies() -> Vec { - 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::>(), 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::(); - 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::(), - ) - .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"); -- cgit v1.3.1