v0.31.0

src/proxy/cache_http_invalidation.rs

@@ -0,0 +1,162 @@
+// Varnish-like HTTP BAN/PURGE on public vhost listeners. No new admin
+// endpoint or session token: security rests on a strict, opt-in client IP/CIDR
+// allowlist plus normal frontend/host ACLs and rate limits. Never forward.
+fn cache_http_invalidation_method(method: &Method) -> bool {
+    method.as_str() == "BAN" || method.as_str() == "PURGE"
+}
+
+fn cache_http_invalidation_request(
+    method: &Method,
+    host: &str,
+    vhost: &str,
+    scheme: &str,
+    request: &Request,
+) -> Result<crate::cache::InvalidationRequest, &'static str> {
+    // Allow a single explicit Content-Length: 0 (used by some PURGE clients),
+    // but reject ambiguous lengths and all transfer-encoded bodies.
+    let mut lengths = request.headers().get_all(http::header::CONTENT_LENGTH).iter();
+    if let Some(length) = lengths.next() {
+        if lengths.next().is_some() || length.as_bytes() != b"0" {
+            return Err("BAN/PURGE must not contain a request body");
+        }
+    }
+    if request.headers().contains_key("upgrade")
+        || request.headers().contains_key(http::header::TRANSFER_ENCODING)
+        || !request.body().is_end_stream()
+    {
+        return Err("BAN/PURGE must not contain a request body or upgrade");
+    }
+    // Only this vhost and Host are permitted; no client-supplied cross-vhost
+    // selectors, regex, or arbitrary Cache-Tag from the public listener.
+    let target = request.uri().path_and_query().map(|pq| pq.as_str()).unwrap_or("/");
+    if target.len() > 4096 || !target.starts_with('/') {
+        return Err("invalid BAN/PURGE target");
+    }
+    let operation = if method.as_str() == "BAN" {
+        crate::cache::InvalidationOperation::Ban
+    } else {
+        crate::cache::InvalidationOperation::Purge
+    };
+    let selector = if method.as_str() == "BAN" {
+        crate::cache::InvalidationSelector::Prefix
+    } else {
+        crate::cache::InvalidationSelector::Exact
+    };
+    // Never resolve ambiguous duplicate maintenance headers differently from
+    // intermediaries or audit logs. Reject them before selecting a target.
+    if request.headers().get_all("x-rustproxy-ban-ttl").iter().count() > 1
+        || request.headers().get_all("x-rustproxy-confirm-wide").iter().count() > 1
+    {
+        return Err("duplicate BAN/PURGE control header");
+    }
+    let mut ttl_seconds = None;
+    if method.as_str() == "BAN" {
+        if let Some(value) = request.headers().get("x-rustproxy-ban-ttl") {
+            let seconds = value.to_str().ok().and_then(|v| v.parse::<u64>().ok())
+                .ok_or("invalid BAN TTL")?;
+            if !(1..=604_800).contains(&seconds) { return Err("BAN TTL must be 1..604800"); }
+            ttl_seconds = Some(seconds);
+        }
+    } else if request.headers().contains_key("x-rustproxy-ban-ttl") {
+        return Err("BAN TTL is only supported for BAN");
+    }
+    let confirm_wide = request.headers().get("x-rustproxy-confirm-wide")
+        .is_some_and(|value| value == "1");
+    if target == "/" && method.as_str() == "BAN" && !confirm_wide {
+        return Err("BAN / requires X-RustProxy-Confirm-Wide: 1");
+    }
+    if request.headers().contains_key("x-rustproxy-confirm-wide") && !confirm_wide {
+        return Err("invalid wide confirmation header");
+    }
+    if method.as_str() != "BAN" && request.headers().contains_key("x-rustproxy-confirm-wide") {
+        return Err("wide confirmation is only supported for BAN");
+    }
+    // Explicitly scope maintenance to the externally visible scheme. The
+    // ordinary /path matcher would also invalidate the opposite scheme on
+    // a vhost shared between HTTP and HTTPS.
+    if !matches!(scheme, "http" | "https") {
+        return Err("invalid request scheme");
+    }
+    let authority = if host.parse::<std::net::Ipv6Addr>().is_ok() {
+        format!("[{host}]")
+    } else {
+        host.to_owned()
+    };
+    Ok(crate::cache::InvalidationRequest {
+        operation, selector, value: format!("{scheme}://{authority}{target}"),
+        host: Some(host.to_owned()), vhost: Some(vhost.to_owned()),
+        ttl_seconds, confirm_wide,
+    })
+}
+
+async fn cache_http_invalidation(
+    state: &Arc<App>,
+    snapshot: &crate::config::Snapshot,
+    frontend: Option<&Arc<crate::config::CompiledFrontend>>,
+    host: &str,
+    request: Request,
+    client_ip: std::net::IpAddr,
+    peer_ip: std::net::IpAddr,
+    scheme: &str,
+) -> Response {
+    // Fail closed even when the vhost cache is disabled; never pass commands
+    // through to the backend, and never trust arbitrary X-Forwarded-For.
+    let Some(frontend) = frontend else {
+        return error(StatusCode::NOT_FOUND, "RustProxy: no matching vhost for cache invalidation\n");
+    };
+    // A trusted proxy missing a usable client chain must not inherit its own
+    // allowlist permission for every external visitor.
+    if snapshot.forwarding.trusted(peer_ip)
+        && crate::vhost::canonical_ip(peer_ip) == client_ip
+    {
+        return error(StatusCode::FORBIDDEN, "RustProxy: missing trusted proxy client identity\n");
+    }
+    if !snapshot.envelope.config.cache.permits_http_invalidation(client_ip) {
+        return error(StatusCode::FORBIDDEN, "RustProxy: BAN/PURGE client IP is not allowed\n");
+    }
+    let requested = match cache_http_invalidation_request(request.method(), host, &frontend.id, scheme, &request) {
+        Ok(requested) => requested,
+        Err(reason) => return error(StatusCode::BAD_REQUEST, reason),
+    };
+    // Axum's Request<Body> is Send but its Body is not Sync. Never keep
+    // a shared `&Request` inside the handler future across an .await.
+    // Release the unused body before the asynchronous audit/cache work.
+    let method = request.method().clone();
+    drop(request);
+    // Allocate the independent maintenance slot before even writing an audit
+    // row, so a burst from an allowed IP cannot pile up DB writes and workers.
+    let Ok(permit) = state.cache_invalidation_slots.clone().try_acquire_owned() else {
+        return error(StatusCode::TOO_MANY_REQUESTS, "RustProxy: cache maintenance busy\n");
+    };
+    let detail = format!(
+        "client_ip={client_ip}; method={}; host={host}; vhost={}; target={:?}",
+        method, frontend.id, requested.value
+    );
+    if let Err(audit_error) = state.store.audit(&format!("http:{client_ip}"), "cache.http_invalidate", &detail).await {
+        tracing::error!(error = %audit_error, "cache HTTP invalidation audit failed");
+        return error(StatusCode::SERVICE_UNAVAILABLE, "RustProxy: cache maintenance unavailable\n");
+    }
+    // Rate-limit the expensive maintenance path independently of the optional
+    // per-vhost traffic limiter. The permit lives inside the blocking worker
+    // even if a client disconnects while the cache scan is running.
+    let cache = state.cache.clone();
+    let result = tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        cache.invalidate(requested)
+    }).await;
+    let result = match result {
+        Ok(Ok(result)) => result,
+        Ok(Err(invalid)) => {
+            tracing::warn!(reason = %invalid, "cache HTTP invalidation rejected");
+            return error(StatusCode::UNPROCESSABLE_ENTITY, "RustProxy: invalid cache maintenance request\n");
+        },
+        Err(_) => return error(StatusCode::INTERNAL_SERVER_ERROR, "RustProxy: cache maintenance failed\n"),
+    };
+    let code = if result.operation == crate::cache::InvalidationOperation::Ban {
+        StatusCode::ACCEPTED
+    } else { StatusCode::OK };
+    let mut response = (code, axum::Json(serde_json::json!({"result": result}))).into_response();
+    response.headers_mut().insert("cache-control", HeaderValue::from_static("no-store"));
+    response
+}
+