use futures::FutureExt;
use futures::future::BoxFuture;
use std::collections::HashMap;
use std::os::fd::{FromRawFd, IntoRawFd, RawFd};
use std::path::Path;
use std::path::{Path, PathBuf};
use std::sync::{LazyLock, Mutex};
use std::{net::SocketAddr, sync::Arc};
use tower_http::cors::CorsLayer;
use tower_http::trace::TraceLayer;
141 unmodified lines
Expand all
.into_future()
.boxed()
}
ApiBind::Unix(path) => {
let listener = bind_unix_listener(path, binds.unix_mode)
let listener = bind_unix(path, binds.unix_mode)
.map_err(|e| miette::miette!("failed to bind to {bind}: {e}"))?;
tracing::info!("API server listening on {bind}");
axum::serve(listener, full.clone().into_make_service())
.into_future()
}
const LISTEN_FDS_START: RawFd = 3;
#[derive(Default)]
struct Inherited {
tcp: HashMap<SocketAddr, std::net::TcpListener>,
unix: HashMap<PathBuf, std::os::unix::net::UnixListener>,
}
/// systemd holds these sockets across restarts and connects queue until hydrant is back
fn inherited_listeners(
lookup: impl Fn(&str) -> Option<String>,
pid: u32,
first: RawFd,
) -> HashMap<SocketAddr, std::net::TcpListener> {
) -> Inherited {
let ours = lookup("LISTEN_PID").and_then(|p| p.parse::<u32>().ok()) == Some(pid);
let count = lookup("LISTEN_FDS")
.and_then(|n| n.parse::<RawFd>().ok())
.filter(|_| ours)
.unwrap_or(0);
(first..first.saturating_add(count))
.filter_map(|fd| {
// safety: systemd passed this fd to this pid, and it's taken only once
let socket = unsafe { socket2::Socket::from_raw_fd(fd) };
let stream = socket.r#type().ok() == Some(socket2::Type::STREAM);
let addr = socket.local_addr().ok().and_then(|addr| addr.as_socket());
match addr.filter(|_| stream) {
Some(addr) => Some((addr, socket.into())),
None => {
// dropping the socket would close an fd that isn't ours
let _ = socket.into_raw_fd();
None
}
}
})
.collect()
let mut inherited = Inherited::default();
for fd in first..first.saturating_add(count) {
// safety: systemd passed this fd to this pid, and it's taken only once
let socket = unsafe { socket2::Socket::from_raw_fd(fd) };
let stream = socket.r#type().ok() == Some(socket2::Type::STREAM);
let local = socket.local_addr().ok().filter(|_| stream);
if let Some(addr) = local.as_ref().and_then(|addr| addr.as_socket()) {
inherited.tcp.insert(addr, socket.into());
} else if let Some(path) = local.as_ref().and_then(|addr| addr.as_pathname()) {
inherited.unix.insert(path.to_path_buf(), socket.into());
} else {
// dropping the socket would close an fd that isn't ours
let _ = socket.into_raw_fd();
}
}
inherited
}
static INHERITED_LISTENERS: LazyLock<Mutex<HashMap<SocketAddr, std::net::TcpListener>>> =
LazyLock::new(|| {
let env = |key: &str| std::env::var(key).ok();
Mutex::new(inherited_listeners(env, std::process::id(), LISTEN_FDS_START))
});
static INHERITED_LISTENERS: LazyLock<Mutex<Inherited>> = LazyLock::new(|| {
let env = |key: &str| std::env::var(key).ok();
Mutex::new(inherited_listeners(
env,
std::process::id(),
LISTEN_FDS_START,
))
});
fn bind_listener(
addr: SocketAddr,
v6only: Option<bool>,
) -> std::io::Result<tokio::net::TcpListener> {
let inherited = INHERITED_LISTENERS.lock().unwrap().remove(&addr);
let inherited = INHERITED_LISTENERS.lock().unwrap().tcp.remove(&addr);
if let Some(listener) = inherited {
tracing::info!("using the listener systemd passed in for {addr}");
listener.set_nonblocking(true)?;
return tokio::net::TcpListener::from_std(listener);
let std_listener: std::net::TcpListener = socket.into();
tokio::net::TcpListener::from_std(std_listener)
}
/// a socket systemd passed in keeps the mode its unit gave it. `mode` is for sockets hydrant binds
fn bind_unix(path: &Path, mode: u32) -> std::io::Result<tokio::net::UnixListener> {
let inherited = INHERITED_LISTENERS.lock().unwrap().unix.remove(path);
if let Some(listener) = inherited {
tracing::info!(
"using the listener systemd passed in for {}",
path.display()
);
listener.set_nonblocking(true)?;
return tokio::net::UnixListener::from_std(listener);
}
bind_unix_listener(path, mode)
}
/// replaces a socket left behind by a hydrant that died, but never one something is still
/// listening on, and never a file that isn't a socket.
fn bind_unix_listener(path: &Path, mode: u32) -> std::io::Result<tokio::net::UnixListener> {
use std::io::{Error, ErrorKind};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let fd = listener.into_raw_fd();
let mut found = inherited_listeners(listen_env(std::process::id(), 1), std::process::id(), fd);
let listener = found.remove(&addr).expect("the passed listener is keyed by its address");
assert!(found.is_empty());
let mut found =
inherited_listeners(listen_env(std::process::id(), 1), std::process::id(), fd);
let listener = found
.tcp
.remove(&addr)
.expect("the passed listener is keyed by its address");
assert!(found.tcp.is_empty() && found.unix.is_empty());
let client = std::net::TcpStream::connect(addr).unwrap();
let (accepted, _) = listener.accept().unwrap();
assert_eq!(accepted.peer_addr().unwrap(), client.local_addr().unwrap());
}
use std::os::fd::AsRawFd;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let fd = listener.as_raw_fd();
let pid = std::process::id();
assert!(inherited_listeners(listen_env(pid + 1, 1), pid, fd).is_empty());
assert!(inherited_listeners(|_| None, pid, fd).is_empty());
for found in [
inherited_listeners(listen_env(pid + 1, 1), pid, fd),
inherited_listeners(|_| None, pid, fd),
] {
assert!(found.tcp.is_empty() && found.unix.is_empty());
}
listener.local_addr().unwrap();
}
#[test]
fn inherited_unix_sockets_stay_open() {
use std::os::fd::AsRawFd;
fn inherited_unix_listeners_are_found_by_path() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("s");
let listener = std::os::unix::net::UnixListener::bind(&path).unwrap();
let fd = listener.into_raw_fd();
let pid = std::process::id();
assert!(inherited_listeners(listen_env(pid, 1), pid, listener.as_raw_fd()).is_empty());
let mut found = inherited_listeners(listen_env(pid, 1), pid, fd);
let listener = found
.unix
.remove(&path)
.expect("the passed listener is keyed by its path");
assert!(found.tcp.is_empty() && found.unix.is_empty());
std::os::unix::net::UnixStream::connect(&path).unwrap();
listener.accept().unwrap();
}
#[test]
fn inherited_sockets_that_are_not_streams_stay_open() {
use std::os::fd::AsRawFd;
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("s");
let socket = std::os::unix::net::UnixDatagram::bind(&path).unwrap();
let pid = std::process::id();
let found = inherited_listeners(listen_env(pid, 1), pid, socket.as_raw_fd());
assert!(found.tcp.is_empty() && found.unix.is_empty());
let client = std::os::unix::net::UnixDatagram::unbound().unwrap();
client.send_to(b"x", &path).unwrap();
socket.recv(&mut [0; 1]).unwrap();
}
#[tokio::test]
async fn bind_unix_takes_the_listener_systemd_passed_in() {
use std::os::unix::fs::PermissionsExt;
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("api.sock");
let passed = std::os::unix::net::UnixListener::bind(&path).unwrap();
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o660)).unwrap();
INHERITED_LISTENERS
.lock()
.unwrap()
.unix
.insert(path.clone(), passed);
// a live socket would make a fresh bind fail, and a different mode would show a rebind
let listener = bind_unix(&path, 0o600).unwrap();
assert_eq!(
std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
0o660
);
let client = tokio::net::UnixStream::connect(&path);
let (accepted, connected) = tokio::join!(listener.accept(), client);
accepted.unwrap();
connected.unwrap();
}
#[tokio::test]
async fn unix_bind_uses_the_configured_mode() {
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("api.sock");
let listener = bind_unix_listener(&path, 0o660).unwrap();