mirror of
https://github.com/fawney19/Aether.git
synced 2026-09-02 01:10:23 +08:00
refactor(gateway): 将 CF 头剥离中间件下移至各 router 构建函数并重构为前缀匹配
This commit is contained in:
@@ -20,6 +20,7 @@ use thiserror::Error;
|
|||||||
use crate::execution_runtime::{
|
use crate::execution_runtime::{
|
||||||
build_direct_execution_frame_stream, DirectSyncExecutionRuntime, ExecutionRuntimeTransportError,
|
build_direct_execution_frame_stream, DirectSyncExecutionRuntime, ExecutionRuntimeTransportError,
|
||||||
};
|
};
|
||||||
|
use crate::middleware;
|
||||||
|
|
||||||
const EXECUTION_RUNTIME_COMPONENT: &str = "aether-gateway-execution-runtime";
|
const EXECUTION_RUNTIME_COMPONENT: &str = "aether-gateway-execution-runtime";
|
||||||
const REQUEST_GATE_NAME: &str = "execution_runtime_requests";
|
const REQUEST_GATE_NAME: &str = "execution_runtime_requests";
|
||||||
@@ -128,12 +129,14 @@ pub fn build_execution_runtime_router_with_request_gates(
|
|||||||
.with_distributed_request_gate(gate),
|
.with_distributed_request_gate(gate),
|
||||||
None => ExecutionRuntimeAppState::with_request_concurrency_limit(limit),
|
None => ExecutionRuntimeAppState::with_request_concurrency_limit(limit),
|
||||||
};
|
};
|
||||||
Router::new()
|
middleware::apply_cf_header_stripping(
|
||||||
.route("/health", get(health))
|
Router::new()
|
||||||
.route("/metrics", get(metrics))
|
.route("/health", get(health))
|
||||||
.route("/v1/execute/sync", post(execute_sync))
|
.route("/metrics", get(metrics))
|
||||||
.route("/v1/execute/stream", post(execute_stream))
|
.route("/v1/execute/sync", post(execute_sync))
|
||||||
.with_state(state)
|
.route("/v1/execute/stream", post(execute_stream))
|
||||||
|
.with_state(state),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn serve_execution_runtime_tcp(
|
pub async fn serve_execution_runtime_tcp(
|
||||||
|
|||||||
@@ -959,20 +959,14 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
|||||||
let frontdoor_health_url = format!("{public_base_url}/_gateway/health");
|
let frontdoor_health_url = format!("{public_base_url}/_gateway/health");
|
||||||
let api_router = build_router_with_state(state);
|
let api_router = build_router_with_state(state);
|
||||||
|
|
||||||
// Compose the final router: API routes + optional static file serving + CF header stripping
|
// Compose the final router: API routes + optional static file serving.
|
||||||
let router = if let Some(ref static_dir) = args.static_dir {
|
let router = if let Some(ref static_dir) = args.static_dir {
|
||||||
use tower_http::compression::CompressionLayer;
|
use tower_http::compression::CompressionLayer;
|
||||||
info!(static_dir = %static_dir, "serving frontend static files");
|
info!(static_dir = %static_dir, "serving frontend static files");
|
||||||
|
|
||||||
attach_static_frontend(api_router, static_dir)
|
attach_static_frontend(api_router, static_dir).layer(CompressionLayer::new())
|
||||||
.layer(CompressionLayer::new())
|
|
||||||
.layer(axum::middleware::from_fn(
|
|
||||||
aether_gateway::strip_cf_headers_middleware,
|
|
||||||
))
|
|
||||||
} else {
|
} else {
|
||||||
api_router.layer(axum::middleware::from_fn(
|
api_router
|
||||||
aether_gateway::strip_cf_headers_middleware,
|
|
||||||
))
|
|
||||||
};
|
};
|
||||||
|
|
||||||
info!(
|
info!(
|
||||||
|
|||||||
@@ -6,4 +6,5 @@ pub(crate) use access_log::{
|
|||||||
access_log_middleware, should_downgrade_access_log, RequestLogEmitted,
|
access_log_middleware, should_downgrade_access_log, RequestLogEmitted,
|
||||||
};
|
};
|
||||||
pub(crate) use frontdoor_cors::frontdoor_cors_middleware;
|
pub(crate) use frontdoor_cors::frontdoor_cors_middleware;
|
||||||
|
pub(crate) use strip_cf_headers::apply_cf_header_stripping;
|
||||||
pub use strip_cf_headers::strip_cf_headers_middleware;
|
pub use strip_cf_headers::strip_cf_headers_middleware;
|
||||||
|
|||||||
@@ -1,36 +1,146 @@
|
|||||||
use axum::{extract::Request, middleware::Next, response::Response};
|
use axum::{extract::Request, middleware::Next, response::Response, Router};
|
||||||
use http::header::HeaderName;
|
use http::{header::HeaderName, HeaderMap};
|
||||||
|
|
||||||
/// Cloudflare headers to strip from incoming requests and outgoing responses.
|
/// Cloudflare-specific headers that are not part of the `cf-*` prefix family.
|
||||||
/// Prevents leaking CF metadata to upstream providers or back to clients.
|
const CF_EXACT_HEADERS: &[&str] = &["cdn-loop", "true-client-ip"];
|
||||||
static CF_HEADERS: &[&str] = &[
|
|
||||||
"cf-connecting-ip",
|
fn should_strip_cf_header(name: &HeaderName) -> bool {
|
||||||
"cf-ipcountry",
|
let normalized = name.as_str();
|
||||||
"cf-ray",
|
normalized.starts_with("cf-") || CF_EXACT_HEADERS.contains(&normalized)
|
||||||
"cf-visitor",
|
}
|
||||||
"cdn-loop",
|
|
||||||
"true-client-ip",
|
fn strip_cf_headers(headers: &mut HeaderMap) {
|
||||||
"cf-worker",
|
let to_remove: Vec<_> = headers
|
||||||
"cf-ew-via",
|
.keys()
|
||||||
"cf-warp-tag-id",
|
.filter(|name| should_strip_cf_header(name))
|
||||||
];
|
.cloned()
|
||||||
|
.collect();
|
||||||
|
for name in to_remove {
|
||||||
|
headers.remove(name);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn apply_cf_header_stripping(router: Router) -> Router {
|
||||||
|
router.layer(axum::middleware::from_fn(strip_cf_headers_middleware))
|
||||||
|
}
|
||||||
|
|
||||||
pub async fn strip_cf_headers_middleware(mut request: Request, next: Next) -> Response {
|
pub async fn strip_cf_headers_middleware(mut request: Request, next: Next) -> Response {
|
||||||
// Strip CF headers from the incoming request
|
strip_cf_headers(request.headers_mut());
|
||||||
for name in CF_HEADERS {
|
|
||||||
if let Ok(header) = HeaderName::from_bytes(name.as_bytes()) {
|
|
||||||
request.headers_mut().remove(&header);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
let mut response = next.run(request).await;
|
let mut response = next.run(request).await;
|
||||||
|
|
||||||
// Strip CF headers from the outgoing response
|
strip_cf_headers(response.headers_mut());
|
||||||
for name in CF_HEADERS {
|
|
||||||
if let Ok(header) = HeaderName::from_bytes(name.as_bytes()) {
|
|
||||||
response.headers_mut().remove(&header);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
response
|
response
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use axum::body::{to_bytes, Body};
|
||||||
|
use axum::routing::any;
|
||||||
|
use axum::Router;
|
||||||
|
use http::{HeaderValue, Request, Response};
|
||||||
|
use tower::ServiceExt;
|
||||||
|
|
||||||
|
use super::apply_cf_header_stripping;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn strips_cf_prefixed_and_exact_headers_from_request_and_response() {
|
||||||
|
let app = apply_cf_header_stripping(Router::new().route(
|
||||||
|
"/",
|
||||||
|
any(|headers: http::HeaderMap| async move {
|
||||||
|
let leaked = headers.contains_key("cf-ipcity")
|
||||||
|
|| headers.contains_key("cf-ray")
|
||||||
|
|| headers.contains_key("true-client-ip")
|
||||||
|
|| headers.contains_key("cdn-loop");
|
||||||
|
let mut response =
|
||||||
|
Response::new(Body::from(if leaked { "leaked" } else { "clean" }));
|
||||||
|
response.headers_mut().insert(
|
||||||
|
http::header::HeaderName::from_static("cf-ipcity"),
|
||||||
|
HeaderValue::from_static("Shanghai"),
|
||||||
|
);
|
||||||
|
response.headers_mut().insert(
|
||||||
|
http::header::HeaderName::from_static("cf-cache-status"),
|
||||||
|
HeaderValue::from_static("HIT"),
|
||||||
|
);
|
||||||
|
response.headers_mut().insert(
|
||||||
|
http::header::HeaderName::from_static("true-client-ip"),
|
||||||
|
HeaderValue::from_static("1.1.1.1"),
|
||||||
|
);
|
||||||
|
response.headers_mut().insert(
|
||||||
|
http::header::HeaderName::from_static("cdn-loop"),
|
||||||
|
HeaderValue::from_static("cloudflare"),
|
||||||
|
);
|
||||||
|
response
|
||||||
|
}),
|
||||||
|
));
|
||||||
|
|
||||||
|
let response = app
|
||||||
|
.oneshot(
|
||||||
|
Request::builder()
|
||||||
|
.uri("/")
|
||||||
|
.header("cf-ipcity", "Shanghai")
|
||||||
|
.header("cf-ray", "abc123")
|
||||||
|
.header("true-client-ip", "1.1.1.1")
|
||||||
|
.header("cdn-loop", "cloudflare")
|
||||||
|
.body(Body::empty())
|
||||||
|
.expect("request should build"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("request should succeed");
|
||||||
|
|
||||||
|
assert!(response.headers().get("cf-ipcity").is_none());
|
||||||
|
assert!(response.headers().get("cf-cache-status").is_none());
|
||||||
|
assert!(response.headers().get("true-client-ip").is_none());
|
||||||
|
assert!(response.headers().get("cdn-loop").is_none());
|
||||||
|
|
||||||
|
let body = to_bytes(response.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.expect("body should be readable");
|
||||||
|
assert_eq!(body.as_ref(), b"clean");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn preserves_non_cf_headers() {
|
||||||
|
let app = apply_cf_header_stripping(Router::new().route(
|
||||||
|
"/",
|
||||||
|
any(|headers: http::HeaderMap| async move {
|
||||||
|
let mut response = Response::new(Body::from(
|
||||||
|
headers
|
||||||
|
.get("x-custom-header")
|
||||||
|
.and_then(|value| value.to_str().ok())
|
||||||
|
.unwrap_or_default()
|
||||||
|
.to_string(),
|
||||||
|
));
|
||||||
|
response.headers_mut().insert(
|
||||||
|
http::header::HeaderName::from_static("x-custom-response"),
|
||||||
|
HeaderValue::from_static("kept"),
|
||||||
|
);
|
||||||
|
response
|
||||||
|
}),
|
||||||
|
));
|
||||||
|
|
||||||
|
let response = app
|
||||||
|
.oneshot(
|
||||||
|
Request::builder()
|
||||||
|
.uri("/")
|
||||||
|
.header("x-custom-header", "kept")
|
||||||
|
.body(Body::empty())
|
||||||
|
.expect("request should build"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("request should succeed");
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
response
|
||||||
|
.headers()
|
||||||
|
.get("x-custom-response")
|
||||||
|
.and_then(|value| value.to_str().ok()),
|
||||||
|
Some("kept")
|
||||||
|
);
|
||||||
|
let body = to_bytes(response.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.expect("body should be readable");
|
||||||
|
assert_eq!(body.as_ref(), b"kept");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -43,19 +43,19 @@ pub fn build_router_with_state(state: AppState) -> Router {
|
|||||||
middleware::frontdoor_cors_middleware,
|
middleware::frontdoor_cors_middleware,
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
router
|
middleware::apply_cf_header_stripping(router)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn attach_static_frontend(router: Router, static_dir: impl Into<PathBuf>) -> Router {
|
pub fn attach_static_frontend(router: Router, static_dir: impl Into<PathBuf>) -> Router {
|
||||||
let static_dir = static_dir.into();
|
let static_dir = static_dir.into();
|
||||||
let index_html = static_dir.join("index.html");
|
let index_html = static_dir.join("index.html");
|
||||||
router.layer(axum::middleware::from_fn_with_state(
|
middleware::apply_cf_header_stripping(router.layer(axum::middleware::from_fn_with_state(
|
||||||
FrontendStaticState {
|
FrontendStaticState {
|
||||||
static_dir,
|
static_dir,
|
||||||
index_html,
|
index_html,
|
||||||
},
|
},
|
||||||
frontend_static_middleware,
|
frontend_static_middleware,
|
||||||
))
|
)))
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn frontend_static_middleware(
|
async fn frontend_static_middleware(
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ use axum::routing::{get, post};
|
|||||||
use axum::Router;
|
use axum::Router;
|
||||||
use tracing::warn;
|
use tracing::warn;
|
||||||
|
|
||||||
use crate::data::GatewayDataState;
|
use crate::{data::GatewayDataState, middleware};
|
||||||
|
|
||||||
pub use control_plane::ControlPlaneClient;
|
pub use control_plane::ControlPlaneClient;
|
||||||
pub use hub::{ConnConfig, HubRouter, LocalBodyEvent, ProxyConn};
|
pub use hub::{ConnConfig, HubRouter, LocalBodyEvent, ProxyConn};
|
||||||
@@ -138,16 +138,18 @@ impl AppState {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn build_router_with_state(state: AppState) -> Router {
|
pub fn build_router_with_state(state: AppState) -> Router {
|
||||||
Router::new()
|
middleware::apply_cf_header_stripping(
|
||||||
.route("/health", get(health))
|
Router::new()
|
||||||
.route("/metrics", get(metrics))
|
.route("/health", get(health))
|
||||||
.route("/stats", get(stats))
|
.route("/metrics", get(metrics))
|
||||||
.route("/api/internal/proxy-tunnel", get(ws_proxy))
|
.route("/stats", get(stats))
|
||||||
.route(
|
.route("/api/internal/proxy-tunnel", get(ws_proxy))
|
||||||
"/api/internal/tunnel/relay/{node_id}",
|
.route(
|
||||||
post(local_relay::relay_request),
|
"/api/internal/tunnel/relay/{node_id}",
|
||||||
)
|
post(local_relay::relay_request),
|
||||||
.with_state(state)
|
)
|
||||||
|
.with_state(state),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn health(State(state): State<AppState>) -> impl IntoResponse {
|
async fn health(State(state): State<AppState>) -> impl IntoResponse {
|
||||||
|
|||||||
Reference in New Issue
Block a user