Harden wasm route decision ordering

Author: eldryoth

crates/fluxheim-server/src/native_http1_route_proxy_handler.rs

@@ -60,17 +60,6 @@ impl NativeHttp1Handler for NativeHttp1RouteProxy {
             let client_ip = self.access_client_ip(&request);
             let tls_identity = request.tls_identity.as_ref();
             let geo_context = request.geo_context.as_ref();
-            #[cfg(feature = "wasm")]
-            let selected_route = {
-                match self
-                    .wasm_route_decision(&request, &path, selected_route)
-                    .await
-                {
-                    NativeWasmRouteResolution::Continue => selected_route,
-                    NativeWasmRouteResolution::Select(route) => Some(route),
-                    NativeWasmRouteResolution::Reject(response) => return response,
-                }
-            };
             if !self.access.allows(client_ip, tls_identity, geo_context)
                 || selected_route
                     .is_some_and(|route| !route.access.allows(client_ip, tls_identity, geo_context))
@@ -80,22 +69,6 @@ impl NativeHttp1Handler for NativeHttp1RouteProxy {
                 return NativeHttp1Response::new(403, "Forbidden", b"forbidden\n")
                     .close_connection();
             }
-            #[cfg(feature = "wasm")]
-            if let Some(response) =
-                wasm_access_rejection(self, decoded_policy_route.or(selected_route)).await
-            {
-                return response;
-            }
-            if !selected_route
-                .is_some_and(NativeHttp1RouteProxyRoute::https_redirect_exempt_or_redirect)
-                && let Some(response) = https_redirect_response(
-                    &request,
-                    &self.https_redirect,
-                    &self.fallback_response_headers,
-                )
-            {
-                return response;
-            }
             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
@@ -126,6 +99,39 @@ impl NativeHttp1Handler for NativeHttp1RouteProxy {
                     .close_connection();
                 }
             }
+            #[cfg(feature = "wasm")]
+            let selected_route = {
+                match self
+                    .wasm_route_decision(&request, &path, selected_route)
+                    .await
+                {
+                    NativeWasmRouteResolution::Continue => selected_route,
+                    NativeWasmRouteResolution::Select(route) => {
+                        if !route.access.allows(client_ip, tls_identity, geo_context) {
+                            return NativeHttp1Response::new(403, "Forbidden", b"forbidden\n")
+                                .close_connection();
+                        }
+                        Some(route)
+                    }
+                    NativeWasmRouteResolution::Reject(response) => return response,
+                }
+            };
+            #[cfg(feature = "wasm")]
+            if let Some(response) =
+                wasm_access_rejection(self, decoded_policy_route.or(selected_route)).await
+            {
+                return response;
+            }
+            if !selected_route
+                .is_some_and(NativeHttp1RouteProxyRoute::https_redirect_exempt_or_redirect)
+                && let Some(response) = https_redirect_response(
+                    &request,
+                    &self.https_redirect,
+                    &self.fallback_response_headers,
+                )
+            {
+                return response;
+            }
             if let Some(route) = selected_route {
                 if let Some(response) = native_grpc_rejection_response(&route.grpc, &request) {
                     return response;

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

@@ -344,6 +344,36 @@ async fn native_wasm_route_decision_fails_closed_for_unconfigured_branch() {
     assert!(response.ends_with("wasm route decision unavailable\n"));
 }
 
+#[tokio::test]
+async fn native_wasm_route_decision_does_not_run_before_route_acl() {
+    let fixture = WasmRouteFixture::new(&[("router", WasmPluginBody::RouteDecision)]);
+    let upstream = super::upstream_expect_path("/never", "unexpected").await;
+    let mut config = fixture.config_with_attachments(
+        upstream,
+        vec![wasm_vhost_attachment_phase(
+            "router",
+            100,
+            fluxheim_config::WasmPluginPhase::RouteDecision,
+        )],
+    );
+    config.vhosts[0].routes[0].access = fluxheim_config::AccessPolicyConfig {
+        deny: vec!["127.0.0.1".to_owned()],
+        ..Default::default()
+    };
+    let router =
+        NativeHttp1HostRouter::from_config(&config, DownstreamHttp1Policy::default(), 0).unwrap();
+    let proxy = router_listener(router).await;
+
+    let response = downstream_request(
+        proxy,
+        "GET /route HTTP/1.1\r\nHost: route.test\r\nX-Canary: 1\r\nConnection: close\r\n\r\n",
+    )
+    .await;
+
+    assert!(response.starts_with("HTTP/1.1 403 Forbidden\r\n"));
+    assert!(response.ends_with("forbidden\n"));
+}
+
 #[cfg(feature = "load-balancer")]
 #[tokio::test]
 async fn native_wasm_route_decision_selects_configured_load_balanced_route() {

crates/fluxheim-wasm/src/runtime.rs

@@ -1,5 +1,5 @@
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::{Arc, OnceLock, mpsc};
+use std::sync::{Arc, Condvar, Mutex, OnceLock, mpsc};
 use std::thread;
 use std::time::{Duration, Instant};
 
@@ -142,22 +142,54 @@ struct RuntimeStoreState {
     limits: StoreLimits,
 }
 
+#[cfg(test)]
 #[derive(Debug)]
 struct CounterPermit {
     counter: &'static AtomicUsize,
 }
 
+struct CompileSlotPool {
+    active: Mutex<usize>,
+    available: Condvar,
+}
+
+struct CompileSlotPermit {
+    pool: &'static CompileSlotPool,
+}
+
+#[cfg(test)]
 impl Drop for CounterPermit {
     fn drop(&mut self) {
         self.counter.fetch_sub(1, Ordering::AcqRel);
     }
 }
 
+impl Drop for CompileSlotPermit {
+    fn drop(&mut self) {
+        let mut active = self.pool.active.lock().unwrap_or_else(|poisoned| {
+            let _ = poisoned;
+            std::process::abort();
+        });
+        *active = active.saturating_sub(1);
+        self.pool.available.notify_one();
+    }
+}
+
+#[cfg(test)]
 fn compile_slots() -> &'static AtomicUsize {
     static SLOTS: OnceLock<AtomicUsize> = OnceLock::new();
     SLOTS.get_or_init(|| AtomicUsize::new(0))
 }
 
+fn compile_slot_pool() -> &'static CompileSlotPool {
+    static POOL: OnceLock<CompileSlotPool> = OnceLock::new();
+    POOL.get_or_init(|| CompileSlotPool {
+        active: Mutex::new(0),
+        available: Condvar::new(),
+    })
+}
+
+#[cfg(test)]
 fn acquire_counter_permit(
     counter: &'static AtomicUsize,
     limit: usize,
@@ -180,18 +212,35 @@ fn acquire_counter_permit(
 }
 
 fn acquire_counter_permit_with_timeout(
-    counter: &'static AtomicUsize,
+    pool: &'static CompileSlotPool,
     limit: usize,
     timeout: Duration,
-) -> Result<CounterPermit, WasmExecutionError> {
+) -> Result<CompileSlotPermit, WasmExecutionError> {
     let started = Instant::now();
+    let mut active = pool.active.lock().unwrap_or_else(|poisoned| {
+        let _ = poisoned;
+        std::process::abort();
+    });
     loop {
-        match acquire_counter_permit(counter, limit) {
-            Ok(permit) => return Ok(permit),
-            Err(WasmExecutionError::CompileConcurrencyLimit) if started.elapsed() < timeout => {
-                thread::sleep(Duration::from_millis(1));
-            }
-            Err(error) => return Err(error),
+        if *active < limit {
+            *active += 1;
+            return Ok(CompileSlotPermit { pool });
+        }
+        let Some(remaining) = timeout.checked_sub(started.elapsed()) else {
+            return Err(WasmExecutionError::CompileConcurrencyLimit);
+        };
+        if remaining.is_zero() {
+            return Err(WasmExecutionError::CompileConcurrencyLimit);
+        }
+        let (next_active, wait) = pool
+            .available
+            .wait_timeout(active, remaining)
+            .unwrap_or_else(|_| {
+                std::process::abort();
+            });
+        active = next_active;
+        if wait.timed_out() && *active >= limit {
+            return Err(WasmExecutionError::CompileConcurrencyLimit);
         }
     }
 }
@@ -331,23 +380,46 @@ impl FluxWasmRuntime {
     }
 
     fn compile_module(&self, plugin: &WasmPluginFile) -> Result<Module, WasmExecutionError> {
-        self.compile_module_with_counter(plugin, compile_slots(), MAX_CONCURRENT_COMPILES)
+        self.compile_module_with_slot_pool(plugin, compile_slot_pool(), MAX_CONCURRENT_COMPILES)
     }
 
-    fn compile_module_with_counter(
+    fn compile_module_with_slot_pool(
         &self,
         plugin: &WasmPluginFile,
-        counter: &'static AtomicUsize,
+        pool: &'static CompileSlotPool,
         limit: usize,
     ) -> Result<Module, WasmExecutionError> {
         let started = Instant::now();
         let compile_permit =
-            acquire_counter_permit_with_timeout(counter, limit, self.limits.compile_timeout)?;
+            acquire_counter_permit_with_timeout(pool, limit, self.limits.compile_timeout)?;
         let remaining_timeout = self
             .limits
             .compile_timeout
             .checked_sub(started.elapsed())
             .unwrap_or(Duration::ZERO);
+        self.compile_module_with_permit(plugin, compile_permit, remaining_timeout)
+    }
+
+    #[cfg(test)]
+    fn compile_module_with_counter(
+        &self,
+        plugin: &WasmPluginFile,
+        counter: &'static AtomicUsize,
+        limit: usize,
+    ) -> Result<Module, WasmExecutionError> {
+        let compile_permit = acquire_counter_permit(counter, limit)?;
+        self.compile_module_with_permit(plugin, compile_permit, self.limits.compile_timeout)
+    }
+
+    fn compile_module_with_permit<P>(
+        &self,
+        plugin: &WasmPluginFile,
+        compile_permit: P,
+        timeout: Duration,
+    ) -> Result<Module, WasmExecutionError>
+    where
+        P: Send + 'static,
+    {
         let engine = self.engine.clone();
         let bytes = plugin.bytes().to_vec();
         let (result_sender, result_receiver) = mpsc::sync_channel(1);
@@ -358,7 +430,7 @@ impl FluxWasmRuntime {
             let _ = result_sender.send(result);
         });
 
-        match result_receiver.recv_timeout(remaining_timeout) {
+        match result_receiver.recv_timeout(timeout) {
             Ok(result) => result,
             Err(mpsc::RecvTimeoutError::Timeout) => Err(WasmExecutionError::CompileTimeout {
                 timeout_ms: self.limits.compile_timeout.as_millis(),

docs/wasm-extensibility.md

@@ -142,6 +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.
+
 Allowed hooks:
 
 - request headers before upstream selection;

release-notes/RELEASE_NOTES_1.7.3.md

@@ -36,11 +36,16 @@ 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 route ACLs, rate limits, concurrency limits, body limits,
-  redirect policy, and request/response header policy still apply after the
-  Wasm decision selects a route.
+- 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.
 - 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
+  variable inside the configured compile timeout instead of polling in 1 ms
+  sleeps under startup/reload contention.
 - The `wasm` feature remains optional and is still rejected with
   `privacy-mode`.