//! 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}"); } } }