gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.29.2

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-08 22:22:46 +02:00
1 parent 76cb4acef0
commit 68678e98be
84 files changed
+1519 -325
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/proxy/error_fallback.rs
+155
View File
@@ -0,0 +1,155 @@
#[derive(Clone)]
struct InternalFallbackRoute {
vhost_id: String,
target: String,
depth: u8,
}
#[derive(Clone)]
struct FinalRouteContext(crate::config::CompiledRoute);
#[derive(Clone)]
struct ErrorFallbackRequest {
method: Method,
uri: Uri,
version: Version,
headers: HeaderMap,
transport: Option<Transport>,
snapshot: Arc<crate::config::Snapshot>,
exchange: Option<Arc<Exchange>>,
connection_lease: Option<Arc<ConnectionLease>>,
stop: Option<ServerShutdown>,
depth: u8,
}
impl ErrorFallbackRequest {
fn capture(request: &Request, snapshot: &Arc<crate::config::Snapshot>) -> Self {
Self {
method: request.method().clone(),
uri: request.uri().clone(),
version: request.version(),
headers: request.headers().clone(),
transport: request.extensions().get::<Transport>().copied(),
snapshot: snapshot.clone(),
exchange: request.extensions().get::<Arc<Exchange>>().cloned(),
connection_lease: request
.extensions()
.get::<Arc<ConnectionLease>>()
.cloned(),
stop: request.extensions().get::<ServerShutdown>().cloned(),
depth: request
.extensions()
.get::<InternalFallbackRoute>()
.map(|dispatch| dispatch.depth)
.unwrap_or(0),
}
}
fn rebuild(&self, body: Bytes, peer: PeerAddr, local: LocalAddr, dispatch: InternalFallbackRoute) -> Request {
let mut request = Request::new(Body::from(body));
*request.method_mut() = self.method.clone();
*request.uri_mut() = self.uri.clone();
*request.version_mut() = self.version;
*request.headers_mut() = self.headers.clone();
request.extensions_mut().insert(peer);
request.extensions_mut().insert(local);
if let Some(transport) = self.transport {
request.extensions_mut().insert(transport);
}
request.extensions_mut().insert(self.snapshot.clone());
if let Some(exchange) = &self.exchange {
request.extensions_mut().insert(exchange.clone());
}
if let Some(lease) = &self.connection_lease {
request.extensions_mut().insert(lease.clone());
}
if let Some(stop) = &self.stop {
request.extensions_mut().insert(stop.clone());
}
request.extensions_mut().insert(dispatch);
request
}
}
enum ErrorFallbackBody {
Disabled,
Replayable(Bytes),
Unavailable,
}
const MAX_ERROR_FALLBACK_DEPTH: u8 = 8;
const MAX_ERROR_FALLBACK_REPLAY_BODY_BYTES: u64 = 8 * 1024 * 1024;
async fn prepare_error_fallback_body(request: &mut Request, enabled: bool) -> Result<ErrorFallbackBody, Response> {
if !enabled {
return Ok(ErrorFallbackBody::Disabled);
}
if request.body().is_end_stream() {
return Ok(ErrorFallbackBody::Replayable(Bytes::new()));
}
let Some(length) = request.body().size_hint().exact() else {
return Ok(ErrorFallbackBody::Unavailable);
};
if length > MAX_ERROR_FALLBACK_REPLAY_BODY_BYTES {
return Ok(ErrorFallbackBody::Unavailable);
}
let body = std::mem::replace(request.body_mut(), Body::empty());
let collected = body.collect().await.map_err(|_| {
error(
StatusCode::BAD_REQUEST,
"RustProxy: invalid or incomplete request body\n",
)
})?;
let bytes = collected.to_bytes();
*request.body_mut() = Body::from(bytes.clone());
Ok(ErrorFallbackBody::Replayable(bytes))
}
#[allow(clippy::too_many_arguments)]
async fn execute_error_fallback(
state: Arc<App>,
peer: PeerAddr,
local: LocalAddr,
snapshot: &Arc<crate::config::Snapshot>,
host: &str,
tls: bool,
route: &crate::config::CompiledRoute,
status: StatusCode,
template: &ErrorFallbackRequest,
body: &ErrorFallbackBody,
) -> Option<Response> {
let fallback = route
.error_fallback
.as_ref()
.filter(|fallback| fallback.matches(status))?;
let frontend = route.frontend.as_ref()?;
if template.depth >= MAX_ERROR_FALLBACK_DEPTH {
tracing::warn!(vhost = %frontend.id, route = %route.id, target = %fallback.target, "error fallback depth limit reached");
return None;
}
let Some(target) = snapshot
.select_internal_on(host, &frontend.id, &fallback.target, local.0, tls)
.cloned()
else {
return None;
};
let ErrorFallbackBody::Replayable(body) = body else {
tracing::warn!(vhost = %frontend.id, route = %route.id, target = %fallback.target, "error fallback request body is not replayable");
return None;
};
let request = template.rebuild(
body.clone(),
peer,
local,
InternalFallbackRoute {
vhost_id: frontend.id.clone(),
target: fallback.target.clone(),
depth: template.depth.saturating_add(1),
},
);
let mut response = Box::pin(forward_request(state, peer, local, request)).await;
if response.extensions().get::<FinalRouteContext>().is_none() {
response.extensions_mut().insert(FinalRouteContext(target));
}
Some(response)
}
src/proxy/forward/response.rs
+18
View File
@@ -1,6 +1,12 @@
{
match upstream_result {
Ok(Ok(mut response)) => {
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
let (origin_validation_headers, origin_plan, origin_reason, mut response) =
include!("success_prepare.rs");
include!("success_finish.rs")
@@ -53,6 +59,12 @@
return stale;
}
}
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
status, &fallback_request, &fallback_body,
).await {
return fallback;
}
let mut response = error(
status,
if status == StatusCode::SERVICE_UNAVAILABLE {
@@ -95,6 +107,12 @@
return stale;
}
}
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
StatusCode::GATEWAY_TIMEOUT, &fallback_request, &fallback_body,
).await {
return fallback;
}
error(
StatusCode::GATEWAY_TIMEOUT,
"RustProxy: upstream response deadline exceeded\n",
src/proxy/forward/route_cache.rs
+32 -1
View File
@@ -39,6 +39,12 @@ if route.unavailable {
response
.headers_mut()
.insert("retry-after", HeaderValue::from_static("1"));
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
return response;
}
if let crate::config::CompiledSource::WebServer(service) = &route.source {
@@ -88,6 +94,12 @@ if let crate::config::CompiledSource::WebServer(service) = &route.source {
},
)
.await;
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
response.extensions_mut().insert(StaticSourceResponse);
return response;
}
@@ -108,6 +120,12 @@ let (backend, backend_lease, affinity_cookie) = if let Some(pool) = &route.pool
response
.headers_mut()
.insert("retry-after", HeaderValue::from_static("1"));
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
return response;
}
Err(crate::balance::SelectionFailure::Capacity) => {
@@ -118,6 +136,12 @@ let (backend, backend_lease, affinity_cookie) = if let Some(pool) = &route.pool
response
.headers_mut()
.insert("retry-after", HeaderValue::from_static("1"));
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
return response;
}
};
@@ -128,10 +152,17 @@ let (backend, backend_lease, affinity_cookie) = if let Some(pool) = &route.pool
)
} else {
let Some(backend) = route.backend.clone() else {
return error(
let response = error(
StatusCode::INTERNAL_SERVER_ERROR,
"RustProxy: Reverse Proxy route has no backend\n",
);
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
return response;
};
(backend, None, None)
};
src/proxy/forward_entry.rs
+43 -15
View File
@@ -2,6 +2,15 @@ async fn forward(
State(state): State<Arc<App>>,
Extension(peer): Extension<PeerAddr>,
Extension(local): Extension<LocalAddr>,
request: Request,
) -> Response {
forward_request(state, peer, local, request).await
}
async fn forward_request(
state: Arc<App>,
peer: PeerAddr,
local: LocalAddr,
mut request: Request,
) -> Response {
if request.method() == Method::CONNECT
@@ -88,6 +97,10 @@ async fn forward(
.get::<Arc<crate::config::Snapshot>>()
.cloned()
.unwrap_or_else(|| state.config.load_full());
let internal_dispatch = request
.extensions()
.get::<InternalFallbackRoute>()
.cloned();
let Ok(path) = crate::acl::routing_path(request.uri().path()) else {
return error(
StatusCode::BAD_REQUEST,
@@ -226,21 +239,23 @@ async fn forward(
"RustProxy: denied by access policy\n",
);
}
if let Err(limited) = state.rate_limiter.check(
snapshot.envelope.revision,
&frontend.id,
context.client_ip,
country.as_deref(),
&frontend.rate_limit,
) {
let mut response = error(
StatusCode::TOO_MANY_REQUESTS,
"RustProxy: request rate limit exceeded\n",
);
if let Ok(value) = HeaderValue::from_str(&limited.retry_after_seconds.to_string()) {
response.headers_mut().insert("retry-after", value);
if internal_dispatch.is_none() {
if let Err(limited) = state.rate_limiter.check(
snapshot.envelope.revision,
&frontend.id,
context.client_ip,
country.as_deref(),
&frontend.rate_limit,
) {
let mut response = error(
StatusCode::TOO_MANY_REQUESTS,
"RustProxy: request rate limit exceeded\n",
);
if let Ok(value) = HeaderValue::from_str(&limited.retry_after_seconds.to_string()) {
response.headers_mut().insert("retry-after", value);
}
return response;
}
return response;
}
}
// Global host ACLs remain an independent restrictive scope and also apply to
@@ -272,7 +287,15 @@ async fn forward(
});
return response;
}
let selected = snapshot.select_on(&host, &path, local.0, tls);
let selected = if let Some(dispatch) = &internal_dispatch {
if frontend.is_none_or(|frontend| frontend.id != dispatch.vhost_id) {
None
} else {
snapshot.select_internal_on(&host, &dispatch.vhost_id, &dispatch.target, local.0, tls)
}
} else {
snapshot.select_on(&host, &path, local.0, tls)
};
if let Some(route) = selected {
if let Some(response) = method_policy_denial(&route.method_policy, request.method()) {
return response;
@@ -306,6 +329,7 @@ async fn forward(
);
}
}
let fallback_request = ErrorFallbackRequest::capture(&request, &snapshot);
// The admin cookie is never ordinary application state. Strip it BEFORE
// auth/cache and keep only the removed fragment locally. A verified self-admin
// backend may restore it after the actual upstream has been selected.
@@ -374,6 +398,10 @@ async fn forward(
)
});
};
let fallback_body = match prepare_error_fallback_body(&mut request, route.error_fallback.is_some()).await {
Ok(body) => body,
Err(response) => return response,
};
let (
backend,
backend_lease,
src/proxy/mod.rs
+1
View File
@@ -1,4 +1,5 @@
include!("types.rs");
include!("error_fallback.rs");
include!("cache_helpers.rs");
include!("upstream_helpers.rs");
include!("redirect_helpers.rs");
src/proxy/types.rs
+2
View File
@@ -20,6 +20,7 @@ use http::{
};
use http_body::Body as HttpBody;
use http_body_util::BodyExt;
use bytes::Bytes;
#[derive(Clone)]
struct ResponseRulesApplied;
#[derive(Clone)]
@@ -38,6 +39,7 @@ struct LocalHostResponse {
struct OriginTiming(Duration);
#[derive(Clone)]
struct OriginBackend(String);
use std::{
sync::{
atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering},
src/proxy/upstream_helpers.rs
+8 -3
View File
@@ -191,6 +191,11 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
))
});
let mut response = next.run(request).await;
let final_route_context = response
.extensions()
.get::<FinalRouteContext>()
.map(|context| context.0.clone());
let effective_route_context = final_route_context.as_ref().or(route_context.as_ref());
let minimal_welcome = response
.extensions()
.get::<crate::pages::MinimalWelcome>()
@@ -255,7 +260,7 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
response.headers_mut(),
&snapshot,
frontend_context.as_deref(),
route_context.as_ref(),
effective_route_context,
false,
);
}
@@ -289,7 +294,7 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
if let Some(frontend) = frontend_context.as_ref() {
compression = frontend.compression.resolve(global_compression, &compression);
}
if let Some(route) = route_context.as_ref() {
if let Some(route) = effective_route_context {
if let crate::config::CompiledSource::WebServer(service) = &route.source {
compression = service.compression.resolve(global_compression, &compression);
}
@@ -355,7 +360,7 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
response.headers_mut(),
&snapshot,
frontend_context.as_deref(),
route_context.as_ref(),
effective_route_context,
);
}
if response