Enforce selected wasm route limits

Author: eldryoth

crates/fluxheim-server/src/native_http1_route_proxy_handler.rs

@@ -69,23 +69,21 @@ impl NativeHttp1Handler for NativeHttp1RouteProxy {
                 return NativeHttp1Response::new(403, "Forbidden", b"forbidden\n")
                     .close_connection();
             }
-            let concurrency_route = decoded_policy_route.or(selected_route);
             // Delay-mode rate limiting sleeps are still live downstream work.
             // Count them against concurrency so an attacker cannot park
             // unlimited delayed tasks outside the configured vhost/route cap.
-            let _concurrency_permits =
-                match self.acquire_concurrency_permits(concurrency_route).await {
-                    Ok(permits) => permits,
-                    Err(status) => {
-                        return NativeHttp1Response::new(
-                            status,
-                            "Too Many Requests",
-                            b"too many requests\n",
-                        )
-                        .close_connection();
-                    }
-                };
-            match self.check_rate_limits(concurrency_route, client_ip) {
+            let _vhost_concurrency_permit = match self.acquire_vhost_concurrency_permit().await {
+                Ok(permit) => permit,
+                Err(status) => {
+                    return NativeHttp1Response::new(
+                        status,
+                        "Too Many Requests",
+                        b"too many requests\n",
+                    )
+                    .close_connection();
+                }
+            };
+            match self.check_vhost_rate_limit(client_ip) {
                 NativeRateLimitDecision::Allow => {}
                 NativeRateLimitDecision::Delay(delay) => {
                     tokio::time::sleep(delay).await;
@@ -116,6 +114,34 @@ impl NativeHttp1Handler for NativeHttp1RouteProxy {
                     NativeWasmRouteResolution::Reject(response) => return response,
                 }
             };
+            let _route_concurrency_permits = match self
+                .acquire_route_concurrency_permits(decoded_policy_route, selected_route)
+                .await
+            {
+                Ok(permits) => permits,
+                Err(status) => {
+                    return NativeHttp1Response::new(
+                        status,
+                        "Too Many Requests",
+                        b"too many requests\n",
+                    )
+                    .close_connection();
+                }
+            };
+            match self.check_route_rate_limits(decoded_policy_route, selected_route, client_ip) {
+                NativeRateLimitDecision::Allow => {}
+                NativeRateLimitDecision::Delay(delay) => {
+                    tokio::time::sleep(delay).await;
+                }
+                NativeRateLimitDecision::Reject(status) => {
+                    return NativeHttp1Response::new(
+                        status,
+                        "Too Many Requests",
+                        b"rate limited\n",
+                    )
+                    .close_connection();
+                }
+            }
             #[cfg(feature = "wasm")]
             if let Some(response) =
                 wasm_access_rejection(self, decoded_policy_route.or(selected_route)).await

crates/fluxheim-server/src/native_http1_route_proxy_policy.rs

@@ -178,6 +178,36 @@ impl NativeHttp1RouteProxy {
             .unwrap_or(NativeRateLimitDecision::Allow)
     }
 
+    pub(crate) fn check_vhost_rate_limit(
+        &self,
+        client_ip: Option<IpAddr>,
+    ) -> NativeRateLimitDecision {
+        self.rate_limit.check(client_ip)
+    }
+
+    pub(crate) fn check_route_rate_limits(
+        &self,
+        first: Option<&NativeHttp1RouteProxyRoute>,
+        second: Option<&NativeHttp1RouteProxyRoute>,
+        client_ip: Option<IpAddr>,
+    ) -> NativeRateLimitDecision {
+        let mut delay = None;
+        for route in unique_route_pair(first, second) {
+            match route.rate_limit.check(client_ip) {
+                NativeRateLimitDecision::Allow => {}
+                NativeRateLimitDecision::Delay(route_delay) => {
+                    delay = Some(
+                        delay.map_or(route_delay, |current: Duration| current.max(route_delay)),
+                    );
+                }
+                decision => return decision,
+            }
+        }
+        delay
+            .map(NativeRateLimitDecision::Delay)
+            .unwrap_or(NativeRateLimitDecision::Allow)
+    }
+
     pub(crate) async fn acquire_concurrency_permits(
         &self,
         route: Option<&NativeHttp1RouteProxyRoute>,
@@ -193,6 +223,37 @@ impl NativeHttp1RouteProxy {
         }
         Ok(permits)
     }
+
+    pub(crate) async fn acquire_vhost_concurrency_permit(
+        &self,
+    ) -> Result<Option<NativeConcurrencyPermit>, u16> {
+        self.concurrency.acquire().await
+    }
+
+    pub(crate) async fn acquire_route_concurrency_permits(
+        &self,
+        first: Option<&NativeHttp1RouteProxyRoute>,
+        second: Option<&NativeHttp1RouteProxyRoute>,
+    ) -> Result<Vec<NativeConcurrencyPermit>, u16> {
+        let mut permits = Vec::with_capacity(2);
+        for route in unique_route_pair(first, second) {
+            if let Some(permit) = route.concurrency.acquire().await? {
+                permits.push(permit);
+            }
+        }
+        Ok(permits)
+    }
+}
+
+fn unique_route_pair<'a>(
+    first: Option<&'a NativeHttp1RouteProxyRoute>,
+    second: Option<&'a NativeHttp1RouteProxyRoute>,
+) -> impl Iterator<Item = &'a NativeHttp1RouteProxyRoute> {
+    first.into_iter().chain(
+        second
+            .into_iter()
+            .filter(move |route| first.is_none_or(|first| !std::ptr::eq(first, *route))),
+    )
 }
 
 impl NativeHttp1RouteProxyRoute {

crates/fluxheim-server/src/native_http1_route_proxy_tests/wasm.rs

@@ -374,6 +374,52 @@ async fn native_wasm_route_decision_does_not_run_before_route_acl() {
     assert!(response.ends_with("forbidden\n"));
 }
 
+#[tokio::test]
+async fn native_wasm_route_decision_enforces_selected_route_rate_limit() {
+    let fixture = WasmRouteFixture::new(&[("router", WasmPluginBody::RouteDecision)]);
+    let stable = upstream_expect_body("/limit/item", "stable").await;
+    let canary = upstream_expect_body("/limit/item", "canary").await;
+    let mut config = fixture.config_with_attachments(
+        stable,
+        vec![wasm_vhost_attachment_phase(
+            "router",
+            100,
+            fluxheim_config::WasmPluginPhase::RouteDecision,
+        )],
+    );
+    let mut canary_route = named_proxy_route("canary", "/limit", canary);
+    canary_route.rate_limit = fluxheim_config::RateLimitConfig {
+        enabled: true,
+        requests_per_second: 1,
+        burst: 1,
+        status: 429,
+        ..Default::default()
+    };
+    config.vhosts[0].routes = vec![
+        named_proxy_route("standard", "/limit", stable),
+        canary_route,
+    ];
+    let router =
+        NativeHttp1HostRouter::from_config(&config, DownstreamHttp1Policy::default(), 0).unwrap();
+    let proxy = router_listener(router).await;
+
+    let first = downstream_request(
+        proxy,
+        "GET /limit/item HTTP/1.1\r\nHost: route.test\r\nX-Canary: 1\r\nConnection: close\r\n\r\n",
+    )
+    .await;
+    let second = downstream_request(
+        proxy,
+        "GET /limit/item HTTP/1.1\r\nHost: route.test\r\nX-Canary: 1\r\nConnection: close\r\n\r\n",
+    )
+    .await;
+
+    assert!(first.starts_with("HTTP/1.1 200 OK\r\n"));
+    assert!(first.ends_with("canary"));
+    assert!(second.starts_with("HTTP/1.1 429 Too Many Requests\r\n"));
+    assert!(second.ends_with("rate limited\n"));
+}
+
 #[cfg(feature = "load-balancer")]
 #[tokio::test]
 async fn native_wasm_route_decision_selects_configured_load_balanced_route() {

docs/wasm-extensibility.md

@@ -142,9 +142,10 @@ Configured persistence on a selected route also remains Fluxheim-owned. The
 backend after a Wasm route decision. Plugins do not provide arbitrary
 persistence keys in this preview.
 
-Route decisions run only after the built-in vhost/preselected-route ACL,
-rate-limit, and concurrency gates. If the plugin selects a different configured
-route, that selected route's ACL is checked before Fluxheim proceeds.
+Route decisions run only after the built-in vhost ACL, vhost rate-limit, vhost
+concurrency, and preselected/decoded-route ACL gates. If the plugin selects a
+different configured route, that selected route's ACL and route-specific
+rate/concurrency limits are checked before Fluxheim proceeds.
 
 Allowed hooks:
 

release-notes/RELEASE_NOTES_1.7.3.md

@@ -36,11 +36,15 @@ route matching, or override built-in Fluxheim access policy.
 - `route-decision` hooks cannot create destinations or bypass route matchers.
   A selected branch must map to an existing configured route with a matching
   method and path.
-- Built-in vhost and preselected-route ACLs, rate limits, and concurrency
-  limits run before `route-decision` execution, so denied or shaped clients
-  cannot spend the process-wide Wasm admission budget first.
-- Selected-route ACLs, body limits, redirect policy, and request/response
-  header policy still apply after the Wasm decision selects a route.
+- Built-in vhost ACLs, vhost rate limits, and vhost concurrency limits run
+  before `route-decision` execution, so denied or shaped clients cannot spend
+  the process-wide Wasm admission budget first.
+- Built-in preselected/decoded route ACLs run before `route-decision`
+  execution, while selected-route ACLs and route-specific rate/concurrency
+  limits run after the final route decision. A plugin-selected route cannot
+  bypass its own configured route policy.
+- Selected-route body limits, redirect policy, and request/response header
+  policy still apply after the Wasm decision selects a route.
 - If a plugin selects an unavailable branch, Fluxheim returns `503` rather than
   falling back silently.
 - Wasm module compilation now waits for a bounded compile slot with a condition