web_ui: support unix:<path> upstreams in extraWebProxies
reqwest has no UDS transport, so unix: upstreams dial the socket directly with a raw hyper/1.1 client per request instead. http(s):// upstreams are unaffected (still go through the existing reqwest path). Adds hyper (client, http1), hyper-util (tokio IO adapter), and http-body-util as direct hive-ag3nt dependencies - all three were already present transitively via reqwest, this just uses them directly for the new code path.
This commit is contained in:
parent
cfeb4f0170
commit
476a7a3c9f
6 changed files with 127 additions and 5 deletions
3
Cargo.lock
generated
3
Cargo.lock
generated
|
|
@ -1383,6 +1383,9 @@ dependencies = [
|
||||||
"futures-util",
|
"futures-util",
|
||||||
"hive-claude",
|
"hive-claude",
|
||||||
"hive-sh4re",
|
"hive-sh4re",
|
||||||
|
"http-body-util",
|
||||||
|
"hyper",
|
||||||
|
"hyper-util",
|
||||||
"reqwest",
|
"reqwest",
|
||||||
"rmcp",
|
"rmcp",
|
||||||
"rusqlite",
|
"rusqlite",
|
||||||
|
|
|
||||||
|
|
@ -68,6 +68,9 @@ reqwest = { version = "0.12", default-features = false, features = [
|
||||||
"json",
|
"json",
|
||||||
"rustls-tls",
|
"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 = [
|
forgejo-api = { version = "0.11", default-features = false, features = [
|
||||||
"rustls-tls",
|
"rustls-tls",
|
||||||
] }
|
] }
|
||||||
|
|
|
||||||
|
|
@ -411,7 +411,11 @@ shaped).
|
||||||
on both sides, and buffers the full response body (MVP — SSE
|
on both sides, and buffers the full response body (MVP — SSE
|
||||||
connections will appear as one large response rather than streaming).
|
connections will appear as one large response rather than streaming).
|
||||||
The `/extra/` namespace ensures user-declared proxies can never
|
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:<path>` (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`):
|
Bus events (new vocabulary on `/events/stream`):
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -10,6 +10,9 @@ workspace = true
|
||||||
anyhow.workspace = true
|
anyhow.workspace = true
|
||||||
axum.workspace = true
|
axum.workspace = true
|
||||||
reqwest.workspace = true
|
reqwest.workspace = true
|
||||||
|
hyper.workspace = true
|
||||||
|
hyper-util.workspace = true
|
||||||
|
http-body-util.workspace = true
|
||||||
forgejo-api.workspace = true
|
forgejo-api.workspace = true
|
||||||
url.workspace = true
|
url.workspace = true
|
||||||
time.workspace = true
|
time.workspace = true
|
||||||
|
|
|
||||||
|
|
@ -6,8 +6,14 @@
|
||||||
//! proxies from ever colliding with the native agent endpoints (`/api/*`,
|
//! proxies from ever colliding with the native agent endpoints (`/api/*`,
|
||||||
//! `/events/*`, …). Response bodies are buffered whole (MVP): an SSE upstream
|
//! `/events/*`, …). Response bodies are buffered whole (MVP): an SSE upstream
|
||||||
//! appears as one large response rather than streaming.
|
//! 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:<path>` (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::collections::HashMap;
|
||||||
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use axum::{
|
use axum::{
|
||||||
|
|
@ -17,6 +23,8 @@ use axum::{
|
||||||
http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri},
|
http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri},
|
||||||
response::{IntoResponse, Response},
|
response::{IntoResponse, Response},
|
||||||
};
|
};
|
||||||
|
use http_body_util::{BodyExt, Full};
|
||||||
|
use hyper_util::rt::TokioIo;
|
||||||
|
|
||||||
use super::AppState;
|
use super::AppState;
|
||||||
|
|
||||||
|
|
@ -55,14 +63,28 @@ pub(super) fn mount_extra_proxies(mut app: Router<AppState>) -> Router<AppState>
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Build a transparent reverse-proxy service forwarding every request to
|
/// 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
|
/// `upstream` — either an `http(s)://` URL or a `unix:<path>` Unix domain
|
||||||
/// prefix; axum strips the prefix before the service sees the request.
|
/// socket. The caller nests it at a path prefix; axum strips the prefix
|
||||||
/// Returns `None` when `upstream` is blank or the reqwest client won't build.
|
/// 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<Router<()>> {
|
fn extra_proxy_service(upstream: &str) -> Option<Router<()>> {
|
||||||
let upstream = upstream.trim();
|
let upstream = upstream.trim();
|
||||||
if upstream.is_empty() {
|
if upstream.is_empty() {
|
||||||
return None;
|
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()
|
let client = match reqwest::Client::builder()
|
||||||
.timeout(std::time::Duration::from_secs(30))
|
.timeout(std::time::Duration::from_secs(30))
|
||||||
.build()
|
.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<Arc<PathBuf>>,
|
||||||
|
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<Response> {
|
||||||
|
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())
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -819,12 +819,16 @@ in
|
||||||
example = lib.literalExpression ''{ "stats" = "http://127.0.0.1:3737"; }'';
|
example = lib.literalExpression ''{ "stats" = "http://127.0.0.1:3737"; }'';
|
||||||
description = ''
|
description = ''
|
||||||
Transparent reverse-proxies mounted under `/extra/` in the per-agent web UI.
|
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
|
E.g. `{ "stats" = "http://127.0.0.1:3737"; }` mounts a proxy at
|
||||||
`/agent/<name>/extra/stats/` that forwards to port 3737 with the prefix
|
`/agent/<name>/extra/stats/` that forwards to port 3737 with the prefix
|
||||||
stripped. All user-declared proxies live under `/extra/` so they can
|
stripped. All user-declared proxies live under `/extra/` so they can
|
||||||
never conflict with native agent endpoints (`/api/*`, `/events/*`, etc.).
|
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:<path>` (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.
|
Intended for agents that run secondary web servers in the same container.
|
||||||
Static assets served by the secondary app must use relative paths to
|
Static assets served by the secondary app must use relative paths to
|
||||||
resolve correctly under the sub-path prefix.
|
resolve correctly under the sub-path prefix.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue