From 4d062d83479d2cffead09416d31319254aaef418 Mon Sep 17 00:00:00 2001 From: Sebastian Martinez Date: Tue, 23 Dec 2025 10:54:24 +0100 Subject: [PATCH] httpd: Add `axum-listener` and use axum for unix-sockets directly --- radicle-httpd/Cargo.lock | 72 ++++++++++++- radicle-httpd/Cargo.toml | 5 +- radicle-httpd/src/lib.rs | 158 +++++------------------------ radicle-httpd/src/main.rs | 45 ++++---- radicle-httpd/src/test.rs | 2 +- radicle-httpd/src/tracing_extra.rs | 9 +- 6 files changed, 126 insertions(+), 165 deletions(-) diff --git a/radicle-httpd/Cargo.lock b/radicle-httpd/Cargo.lock index 6587f72a4..2397f3b23 100644 --- a/radicle-httpd/Cargo.lock +++ b/radicle-httpd/Cargo.lock @@ -211,6 +211,18 @@ dependencies = [ "tower-service", ] +[[package]] +name = "axum-listener" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0cd6458a1428aec99bace599e09140ecd60479f81a33f866ec628b2f5d3a793" +dependencies = [ + "axum", + "futures", + "tokio", + "tracing", +] + [[package]] name = "backtrace" version = "0.3.73" @@ -837,6 +849,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "645c6916888f6cb6350d2550b80fb63e734897a8498abe35cfb732b6487804b0" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.30" @@ -844,6 +871,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eac8f7d7865dcb88bd4373ab671c8cf4508703796caa2b1985a9ca867b3fcb78" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -852,6 +880,40 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dfc6580bb841c5a68e9ef15c77ccc837b40a7504914d52e47b8b0e9bbda25a1d" +[[package]] +name = "futures-executor" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a576fc72ae164fca6b9db127eaa9a9dda0d61316034f33a0a0d4eda41f02b01d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-macro" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.87", +] + +[[package]] +name = "futures-sink" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" + [[package]] name = "futures-task" version = "0.3.30" @@ -864,10 +926,16 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d6401deb83407ab3da39eba7e33987a73c3df0c82b4bb5813ee871c19c41d48" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "pin-utils", + "slab", ] [[package]] @@ -2022,11 +2090,11 @@ version = "0.20.0" dependencies = [ "anyhow", "axum", + "axum-listener", "base64 0.22.1", "chrono", "flate2", "hyper", - "hyper-util", "infer", "lexopt", "lru", @@ -2042,7 +2110,6 @@ dependencies = [ "tempfile", "thiserror 2.0.14", "tokio", - "tower 0.4.13", "tower 0.5.2", "tower-http", "tracing", @@ -2830,7 +2897,6 @@ dependencies = [ "tokio", "tower-layer", "tower-service", - "tracing", ] [[package]] diff --git a/radicle-httpd/Cargo.toml b/radicle-httpd/Cargo.toml index a559669b3..6c7dd98a5 100644 --- a/radicle-httpd/Cargo.toml +++ b/radicle-httpd/Cargo.toml @@ -23,11 +23,11 @@ path = "src/main.rs" [dependencies] anyhow = { version = "1" } axum = { version = "0.8.4", default-features = false, features = ["json", "query", "tokio", "http1"] } +axum-listener = { version = "0.2.0" } base64 = { version = "0.22.1" } chrono = { version = "0.4.41", default-features = false, features = ["clock"] } flate2 = { version = "1" } -hyper = { version = "1.6.0", default-features = false, features = ["server", "http1"] } -hyper-util = { version = "0.1.7", features = ["tokio", "server", "service"] } +hyper = { version = "1.6.0", default-features = false } infer = { version = "0.19.0" } lexopt = { version = "0.3.1" } lru = { version = "0.16.0" } @@ -40,7 +40,6 @@ serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } thiserror = { version = "2" } tokio = { version = "1.47.1", default-features = false, features = ["macros", "rt-multi-thread"] } -tower = { version = "0.4.13", features = ["util"] } tower-http = { version = "0.6.6", default-features = false, features = ["trace", "cors", "set-header"] } tracing = { version = "0.1.41", default-features = false, features = ["std", "log"] } tracing-logfmt = { version = "0.3.5", optional = true } diff --git a/radicle-httpd/src/lib.rs b/radicle-httpd/src/lib.rs index 8450ca40e..3b9c2bb77 100644 --- a/radicle-httpd/src/lib.rs +++ b/radicle-httpd/src/lib.rs @@ -4,26 +4,22 @@ pub mod error; use std::collections::HashMap; -use std::net::SocketAddr; use std::num::NonZeroUsize; -use std::path::PathBuf; use std::process::Command; use std::str; use std::sync::Arc; use std::time::Duration; use anyhow::Context as _; -use axum::body::{Body, HttpBody}; -use axum::http::{Request, Response}; +use axum::body::Body; +use axum::http::Request; use axum::response::IntoResponse; use axum::routing::get; use axum::{middleware, Json, Router}; -use hyper::header::CONTENT_TYPE; +use axum_listener::{DualAddr, DualListener}; +use hyper::body::Body as _; +use hyper::header::{CONTENT_TYPE, FORWARDED}; use hyper::Method; -use hyper_util::rt::TokioIo; -use hyper_util::service::TowerToHyperService; -use tokio::net::{TcpListener, UnixListener}; -use tower::Service; use tower_http::cors; use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; @@ -32,9 +28,8 @@ use tracing::Span; use radicle::identity::RepoId; use radicle::Profile; -use tracing_extra::{tracing_middleware, ColoredStatus, Paint, RequestId, TracingInfo}; - use crate::api::RADICLE_VERSION; +use crate::tracing_extra::{tracing_middleware, ColoredStatus, Paint, RequestId, TracingInfo}; mod api; mod axum_extra; @@ -48,25 +43,10 @@ mod tracing_extra; /// Default cache HTTP size. pub const DEFAULT_CACHE_SIZE: NonZeroUsize = NonZeroUsize::new(100).unwrap(); -#[derive(Debug, Clone)] -pub enum ListenAddress { - Tcp(SocketAddr), - Unix(PathBuf), -} - -impl std::fmt::Display for ListenAddress { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - ListenAddress::Tcp(addr) => write!(f, "http://{}", addr), - ListenAddress::Unix(path) => write!(f, "unix://{}", path.display()), - } - } -} - #[derive(Debug, Clone)] pub struct Options { pub aliases: HashMap, - pub listen: ListenAddress, + pub listen: DualAddr, pub cache: Option, } @@ -80,47 +60,34 @@ pub async fn run(options: Options) -> anyhow::Result<()> { tracing::info!("{}", str::from_utf8(&git_version)?.trim()); - match &options.listen { - ListenAddress::Tcp(addr) => { - let listener = TcpListener::bind(addr).await?; - tracing::info!("listening on {}", options.listen); - run_tcp_server(listener, options).await - } - ListenAddress::Unix(path) => { - // Remove existing socket file if it exists - if path.exists() { - std::fs::remove_file(path)?; - } - let listener = UnixListener::bind(path)?; - tracing::info!("listening on {}", options.listen); - run_unix_server(listener, options).await - } - } -} + let listener = DualListener::bind(&options.listen).await?; + tracing::info!("listening on {:?}", &options.listen); -async fn run_tcp_server(listener: TcpListener, options: Options) -> anyhow::Result<()> { let profile = Profile::load()?; let request_id = RequestId::new(); tracing::info!("using radicle home at {}", profile.home().path().display()); - let app = - router(options, profile)? + let app = router(options, profile)? .layer(middleware::from_fn(tracing_middleware)) .layer( TraceLayer::new_for_http() - .make_span_with(move |_request: &Request| { - tracing::info_span!("request", id = %request_id.clone().next()) + .make_span_with(move |request: &Request| { + if let Some(forwarded) = request.headers().get("X-Forwarded-For").and_then(|s| s.to_str().ok()) { + tracing::info_span!("request", id = %request_id.clone().next(), "X-Forwarded-For" = forwarded) + } else { + tracing::info_span!("request", id = %request_id.clone().next()) + } }) .on_response( - |response: &Response, latency: Duration, _span: &Span| { + |response: &hyper::Response, latency: Duration, _span: &Span| { if let Some(info) = response.extensions().get::() { - let client_addr = info.connect_info - .map(|ci| ci.0.to_string()) - .unwrap_or_else(|| "unknown".to_string()); tracing::info!( "{} \"{} {} {:?}\" {} {:?} {}", - client_addr, + match info.connect_info.0 { + DualAddr::Tcp(c) => c.to_string(), + DualAddr::Uds(_) => "unix-socket".into() + }, info.method, info.uri, info.version, @@ -140,86 +107,14 @@ async fn run_tcp_server(listener: TcpListener, options: Options) -> anyhow::Resu tracing::info!("Processed"); } }, - ), - ) - .into_make_service_with_connect_info::(); + ) + ).into_make_service_with_connect_info::(); axum::serve(listener, app) .await .map_err(anyhow::Error::from) } -async fn run_unix_server(listener: UnixListener, options: Options) -> anyhow::Result<()> { - let profile = Profile::load()?; - let request_id = RequestId::new(); - - tracing::info!("using radicle home at {}", profile.home().path().display()); - - let app = - router(options, profile)? - .layer(middleware::from_fn(tracing_middleware)) - .layer( - TraceLayer::new_for_http() - .make_span_with(move |_request: &Request| { - tracing::info_span!("request", id = %request_id.clone().next()) - }) - .on_response( - |response: &Response, latency: Duration, _span: &Span| { - tracing::info!( - "unix-socket \"{} {} {:?}\" {} {:?} {}", - response.extensions().get::() - .map(|info| info.method.as_str()) - .unwrap_or("UNKNOWN"), - response.extensions().get::() - .map(|info| info.uri.path()) - .unwrap_or("/"), - response.extensions().get::() - .map(|info| info.version) - .unwrap_or(hyper::Version::HTTP_11), - ColoredStatus(response.status()), - latency, - Paint::dim( - response - .body() - .size_hint() - .exact() - .map(|n| n.to_string()) - .unwrap_or("0".to_string()) - .into() - ), - ); - }, - ), - ) - .into_make_service(); - - // Manual Unix socket serving since axum doesn't directly support Unix listeners - loop { - let (stream, _) = listener.accept().await?; - let mut app_service = app.clone(); - - tokio::spawn(async move { - let stream = TokioIo::new(stream); - - // Create a service from the service factory for this connection - match app_service.call(()).await { - Ok(service) => { - let hyper_service = TowerToHyperService::new(service); - if let Err(err) = hyper::server::conn::http1::Builder::new() - .serve_connection(stream, hyper_service) - .await - { - tracing::error!("Error serving Unix socket connection: {}", err); - } - } - Err(err) => { - tracing::error!("Failed to create service: {:?}", err); - } - } - }); - } -} - /// Create a router consisting of other sub-routers. fn router(options: Options, profile: Profile) -> anyhow::Result { let profile = Arc::new(profile); @@ -315,8 +210,9 @@ mod routes { use axum::extract::connect_info::MockConnectInfo; use axum::http::StatusCode; + use axum_listener::DualAddr; - use crate::test::{self, get}; + use crate::test; #[tokio::test] async fn test_invalid_route_returns_404() { @@ -324,7 +220,7 @@ mod routes { let app = super::router( super::Options { aliases: HashMap::new(), - listen: super::ListenAddress::Tcp(SocketAddr::from(([0, 0, 0, 0], 8080))), + listen: DualAddr::Tcp(SocketAddr::from(([0, 0, 0, 0], 8080))), cache: None, }, test::profile(tmp.path(), [0xff; 32]), @@ -332,7 +228,7 @@ mod routes { .unwrap() .layer(MockConnectInfo(SocketAddr::from(([0, 0, 0, 0], 8080)))); - let response = get(&app, "/aa/a").await; + let response = test::get(&app, "/aa/a").await; assert_eq!(response.status(), StatusCode::NOT_FOUND); } diff --git a/radicle-httpd/src/main.rs b/radicle-httpd/src/main.rs index fac9e34d5..81190f246 100644 --- a/radicle-httpd/src/main.rs +++ b/radicle-httpd/src/main.rs @@ -1,7 +1,7 @@ use std::num::NonZeroUsize; -use std::path::PathBuf; use std::{collections::HashMap, process}; +use axum_listener::DualAddr; use radicle::prelude::RepoId; use radicle::version::Version; use radicle_httpd as httpd; @@ -32,13 +32,13 @@ Options #[tokio::main] async fn main() -> anyhow::Result<()> { - let options = parse_options()?; - // SAFETY: The logger is only initialized once. httpd::logger::init().unwrap(); tracing::info!("starting http daemon.."); tracing::info!("version {} ({})", env!("RADICLE_VERSION"), env!("GIT_HEAD")); + let options = parse_options()?; + match httpd::run(options).await { Ok(()) => {} Err(err) => { @@ -61,9 +61,23 @@ fn parse_options() -> Result { while let Some(arg) = parser.next()? { match arg { Long("listen") => { - let value = parser.value()?; - let addr = parse_listen_address(&value.to_string_lossy()) - .map_err(|e| lexopt::Error::from(e.to_string()))?; + let addr: DualAddr = parser.value()?.parse()?; + + // Get socket path and remove it if existing + if let DualAddr::Uds(socket_path) = &addr { + if let Some(path) = socket_path.as_pathname() { + if path.exists() { + tracing::info!("Removing existing socket path at {}", path.display()); + if let Err(e) = std::fs::remove_file(path) { + tracing::error!("{e}"); + } + } + } else { + tracing::error!("Provided socket address isn't a valid path."); + process::exit(0); + } + } + listen = Some(addr); } Long("alias") | Short('a') => { @@ -74,7 +88,7 @@ fn parse_options() -> Result { } Long("version") | Short('v') => { if let Err(e) = VERSION.write(std::io::stdout()) { - eprintln!("error: {e}"); + tracing::error!("{e}"); process::exit(1); }; process::exit(0); @@ -92,22 +106,7 @@ fn parse_options() -> Result { } Ok(httpd::Options { aliases, - listen: listen.unwrap_or_else(|| httpd::ListenAddress::Tcp(([0, 0, 0, 0], 8080).into())), + listen: listen.unwrap_or_else(|| DualAddr::Tcp(([0, 0, 0, 0], 8080).into())), cache, }) } - -fn parse_listen_address(value: &str) -> Result> { - // Check if it's a Unix socket path (contains '/' and no ':' port separator) - if value.contains('/') && !value.contains(':') { - Ok(httpd::ListenAddress::Unix(PathBuf::from(value))) - } else if value.starts_with("unix:") { - // Support explicit unix: prefix - let path = value.strip_prefix("unix:").unwrap(); - Ok(httpd::ListenAddress::Unix(PathBuf::from(path))) - } else { - // Try to parse as TCP address - let addr = value.parse()?; - Ok(httpd::ListenAddress::Tcp(addr)) - } -} diff --git a/radicle-httpd/src/test.rs b/radicle-httpd/src/test.rs index 9ce451dfe..b8f401d10 100644 --- a/radicle-httpd/src/test.rs +++ b/radicle-httpd/src/test.rs @@ -326,7 +326,7 @@ fn seed_with_signer>( let options = crate::Options { aliases: std::collections::HashMap::new(), - listen: std::net::SocketAddr::from(([0, 0, 0, 0], 8080)), + listen: axum_listener::DualAddr::Tcp(std::net::SocketAddr::from(([0, 0, 0, 0], 8080))), cache: Some(crate::DEFAULT_CACHE_SIZE), }; diff --git a/radicle-httpd/src/tracing_extra.rs b/radicle-httpd/src/tracing_extra.rs index bab41f8a5..af2f094cb 100644 --- a/radicle-httpd/src/tracing_extra.rs +++ b/radicle-httpd/src/tracing_extra.rs @@ -1,5 +1,4 @@ use std::fmt; -use std::net::SocketAddr; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -9,6 +8,7 @@ use axum::http::Request; use axum::middleware::Next; use axum::response::IntoResponse; use axum::Extension; +use axum_listener::DualAddr; use hyper::{Method, StatusCode, Uri, Version}; pub use radicle_term::ansi::Paint; @@ -28,7 +28,7 @@ impl RequestId { #[derive(Clone)] pub struct TracingInfo { - pub connect_info: Option>, + pub connect_info: ConnectInfo, pub method: Method, pub version: Version, pub uri: Uri, @@ -50,8 +50,9 @@ impl fmt::Display for ColoredStatus { pub async fn tracing_middleware(request: Request, next: Next) -> impl IntoResponse { let connect_info = request .extensions() - .get::>() - .copied(); + .get::>() + .unwrap() + .clone(); let method = request.method().clone(); let version = request.version();