diff --git a/Cargo.lock b/Cargo.lock index ebf70b9c..f1de027e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1383,6 +1383,9 @@ dependencies = [ "futures-util", "hive-claude", "hive-sh4re", + "http-body-util", + "hyper", + "hyper-util", "reqwest", "rmcp", "rusqlite", diff --git a/Cargo.toml b/Cargo.toml index 10737252..8208a293 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,9 @@ reqwest = { version = "0.12", default-features = false, features = [ "json", "rustls-tls", ] } +hyper = { version = "1", features = ["client", "http1"] } +hyper-util = { version = "0.1", features = ["tokio"] } +http-body-util = "0.1" forgejo-api = { version = "0.11", default-features = false, features = [ "rustls-tls", ] } diff --git a/docs/web-ui/agent.md b/docs/web-ui/agent.md index fa35f39a..3a846dd7 100644 --- a/docs/web-ui/agent.md +++ b/docs/web-ui/agent.md @@ -411,7 +411,11 @@ shaped). on both sides, and buffers the full response body (MVP — SSE connections will appear as one large response rather than streaming). The `/extra/` namespace ensures user-declared proxies can never - conflict with native agent endpoints. Implemented in `web_ui/proxy.rs`. + conflict with native agent endpoints. Upstream values are either an + `http(s)://` URL (forwarded via `reqwest`) or a Unix domain socket, + spelled `unix:` (e.g. `unix:/run/myapp/http.sock`) — dialed + directly with a raw HTTP/1.1 client per request, since `reqwest` has + no UDS transport. Implemented in `web_ui/proxy.rs`. Bus events (new vocabulary on `/events/stream`): diff --git a/hive-ag3nt/Cargo.toml b/hive-ag3nt/Cargo.toml index 6c182bfb..a76f5f31 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -10,6 +10,9 @@ workspace = true anyhow.workspace = true axum.workspace = true reqwest.workspace = true +hyper.workspace = true +hyper-util.workspace = true +http-body-util.workspace = true forgejo-api.workspace = true url.workspace = true time.workspace = true diff --git a/hive-ag3nt/src/web_ui/proxy.rs b/hive-ag3nt/src/web_ui/proxy.rs index 79c752de..79115626 100644 --- a/hive-ag3nt/src/web_ui/proxy.rs +++ b/hive-ag3nt/src/web_ui/proxy.rs @@ -6,8 +6,14 @@ //! proxies from ever colliding with the native agent endpoints (`/api/*`, //! `/events/*`, …). Response bodies are buffered whole (MVP): an SSE upstream //! appears as one large response rather than streaming. +//! +//! Upstreams are either a regular `http(s)://` URL (forwarded via `reqwest`) +//! or a Unix domain socket, spelled `unix:` (e.g. +//! `unix:/run/myapp/http.sock`) — forwarded via a raw hyper/1.1 client +//! dialing the socket directly, since `reqwest` has no UDS transport. use std::collections::HashMap; +use std::path::{Path, PathBuf}; use std::sync::Arc; use axum::{ @@ -17,6 +23,8 @@ use axum::{ http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri}, response::{IntoResponse, Response}, }; +use http_body_util::{BodyExt, Full}; +use hyper_util::rt::TokioIo; use super::AppState; @@ -55,14 +63,28 @@ pub(super) fn mount_extra_proxies(mut app: Router) -> Router } /// Build a transparent reverse-proxy service forwarding every request to -/// `upstream` (e.g. `http://127.0.0.1:3737`). The caller nests it at a path -/// prefix; axum strips the prefix before the service sees the request. -/// Returns `None` when `upstream` is blank or the reqwest client won't build. +/// `upstream` — either an `http(s)://` URL or a `unix:` Unix domain +/// socket. The caller nests it at a path prefix; axum strips the prefix +/// before the service sees the request. Returns `None` when `upstream` is +/// blank or the underlying client won't build. fn extra_proxy_service(upstream: &str) -> Option> { let upstream = upstream.trim(); if upstream.is_empty() { return None; } + if let Some(sock_path) = upstream.strip_prefix("unix:") { + let sock_path = sock_path.trim(); + if sock_path.is_empty() { + tracing::warn!("extra proxy: `unix:` upstream has an empty socket path — ignoring"); + return None; + } + return Some( + Router::new() + .fallback(unix_proxy_handler) + .with_state(Arc::new(PathBuf::from(sock_path))), + ); + } + let client = match reqwest::Client::builder() .timeout(std::time::Duration::from_secs(30)) .build() @@ -148,3 +170,86 @@ async fn proxy_handler( } } } + +/// Forward the request to `sock_path` over a Unix domain socket, dialing a +/// fresh connection per request (MVP — no connection pooling, matching the +/// simplicity of the HTTP path above). `reqwest` has no UDS transport, so +/// this speaks raw HTTP/1.1 via `hyper`'s low-level client directly over a +/// `tokio::net::UnixStream`. +async fn unix_proxy_handler( + State(sock_path): State>, + method: Method, + uri: Uri, + headers: HeaderMap, + body: Bytes, +) -> Response { + let path_and_query = uri + .path_and_query() + .map_or("/", axum::http::uri::PathAndQuery::as_str) + .to_owned(); + + match forward_over_unix_socket(&sock_path, method, &path_and_query, headers, body).await { + Ok(resp) => resp, + Err(e) => { + tracing::warn!( + socket = %sock_path.display(), + path = path_and_query, + "extra proxy: unix upstream request failed: {e}" + ); + StatusCode::BAD_GATEWAY.into_response() + } + } +} + +/// Dial `sock_path`, send one HTTP/1.1 request built from the given parts, +/// and return the axum `Response` built from whatever comes back. Split out +/// of [`unix_proxy_handler`] so the handler can map every failure mode +/// (connect, handshake, send, body read) to the same `BAD_GATEWAY` fallback +/// with one `?`-chain instead of a matching arm per step. +async fn forward_over_unix_socket( + sock_path: &Path, + method: Method, + path_and_query: &str, + mut headers: HeaderMap, + body: Bytes, +) -> anyhow::Result { + let stream = tokio::net::UnixStream::connect(sock_path).await?; + let io = TokioIo::new(stream); + + let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await?; + // The connection future drives I/O in the background; drop-and-forget is + // fine here since we only ever send one request per dialed socket. + tokio::spawn(async move { + if let Err(e) = conn.await { + tracing::debug!("extra proxy: unix connection closed: {e}"); + } + }); + + for h in HOP_BY_HOP { + headers.remove(h); + } + // HTTP/1.1 requires a Host header; a UDS peer has no meaningful + // hostname, so `localhost` is the conventional placeholder (matches + // what tools like `curl --unix-socket` send by default). + if !headers.contains_key(hyper::header::HOST) { + headers.insert(hyper::header::HOST, HeaderValue::from_static("localhost")); + } + + let mut req_builder = hyper::Request::builder().method(method).uri(path_and_query); + if let Some(h) = req_builder.headers_mut() { + *h = headers; + } + let req = req_builder.body(Full::new(body))?; + + let upstream_resp = sender.send_request(req).await?; + let status = StatusCode::from_u16(upstream_resp.status().as_u16())?; + let mut resp_headers = HeaderMap::new(); + for (name, value) in upstream_resp.headers() { + if HOP_BY_HOP.contains(&name.as_str()) { + continue; + } + resp_headers.insert(name.clone(), value.clone()); + } + let body_bytes = upstream_resp.into_body().collect().await?.to_bytes(); + Ok((status, resp_headers, body_bytes).into_response()) +} diff --git a/nix/templates/harness-base.nix b/nix/templates/harness-base.nix index 6b238238..3c2ba44e 100644 --- a/nix/templates/harness-base.nix +++ b/nix/templates/harness-base.nix @@ -819,12 +819,16 @@ in example = lib.literalExpression ''{ "stats" = "http://127.0.0.1:3737"; }''; description = '' Transparent reverse-proxies mounted under `/extra/` in the per-agent web UI. - Each attribute name becomes the sub-path and the value is the upstream URL. + Each attribute name becomes the sub-path and the value is the upstream. E.g. `{ "stats" = "http://127.0.0.1:3737"; }` mounts a proxy at `/agent//extra/stats/` that forwards to port 3737 with the prefix stripped. All user-declared proxies live under `/extra/` so they can never conflict with native agent endpoints (`/api/*`, `/events/*`, etc.). + The upstream value is either an `http(s)://` URL or a Unix domain + socket, spelled `unix:` (e.g. `unix:/run/myapp/http.sock`) — for + agents whose secondary web server only listens on a UDS. + Intended for agents that run secondary web servers in the same container. Static assets served by the secondary app must use relative paths to resolve correctly under the sub-path prefix.