path: root/src/api/listen.rs

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