Harden native buffering and encrypted cache state

Author: eldryoth

README.md

@@ -75,7 +75,7 @@ Fluxheim is licensed under the European Union Public Licence 1.2.
 | Memory cache | ✅ | Bounded in-memory cache tier. |
 | Disk cache | ✅ | Filesystem and storage-bin disk backends with ordered eviction, one process-wide persistence worker, pre-initialization advisory lease enforcement, live inspection, fail-closed lookup admission, and coalesced durable index writes. |
 | Tiered cache | ✅ | Memory plus disk storage plans. |
-| Encrypted disk cache | ✅ | Optional local-key and OpenBao transit encryption paths. |
+| Encrypted disk cache | ✅ | Optional local-key and OpenBao transit encryption paths; confidential v2 object/index metadata and durable fail-closed local AES-GCM invocation accounting. |
 | Static-file cache | ✅ | Optional local static-file caching. |
 | Range and slice cache | ✅ | Bounded range caching and fixed-slice composition for large objects. |
 | Peer fill | ✅ | Optional peer-assisted cache fill for cache-edge deployments. |

crates/fluxheim-config/src/config_error_display.rs

@@ -53,6 +53,10 @@ impl Display for ConfigError {
             }
             Self::EmptyProcessPath { field } => write!(formatter, "{field} cannot be empty"),
             Self::InvalidLimit { field } => write!(formatter, "{field} must be greater than zero"),
+            Self::InvalidLimitRelationship {
+                field,
+                minimum_field,
+            } => write!(formatter, "{field} must be at least {minimum_field}"),
             Self::InvalidAdminListenAddress { address } => write!(
                 formatter,
                 "admin.listen must be an ip:port listener address, got {address:?}"

crates/fluxheim-config/src/config_error_kind.rs

@@ -41,6 +41,10 @@ pub enum ConfigError {
     InvalidLimit {
         field: &'static str,
     },
+    InvalidLimitRelationship {
+        field: &'static str,
+        minimum_field: &'static str,
+    },
     InvalidAdminListenAddress {
         address: String,
     },

crates/fluxheim-config/src/config_root_validate.rs

@@ -39,12 +39,34 @@ impl Config {
         self.udp.validate()?;
         self.wasm.validate()?;
         self.validate_vhosts()?;
+        self.validate_request_body_budget()?;
         self.validate_wasm_attachments()?;
         self.validate_geoip_policy()?;
         self.validate_compliance_internal_crypto()?;
         Ok(())
     }
 
+    fn validate_request_body_budget(&self) -> Result<(), ConfigError> {
+        let budget = self.server.limits.max_buffered_request_body_bytes.as_u64();
+        let exceeds_budget = self.vhosts.iter().any(|vhost| {
+            vhost
+                .max_request_body_bytes
+                .is_some_and(|limit| limit.as_u64() > budget)
+                || vhost.routes.iter().any(|route| {
+                    route
+                        .max_request_body_bytes
+                        .is_some_and(|limit| limit.as_u64() > budget)
+                })
+        });
+        if exceeds_budget {
+            return Err(ConfigError::InvalidLimitRelationship {
+                field: "server.limits.max_buffered_request_body_bytes",
+                minimum_field: "every vhost and route max_request_body_bytes",
+            });
+        }
+        Ok(())
+    }
+
     fn validate_acme_challenge_runtime(&self) -> Result<(), ConfigError> {
         if !self.tls.acme.enabled || self.tls.acme.challenge != AcmeChallenge::TlsAlpn01 {
             return Ok(());

crates/fluxheim-config/src/config_server.rs

@@ -7,12 +7,13 @@ use crate::config::{ByteSize, ConfigError, validate_config_list_len};
 use crate::config_net::{trusted_proxy_ipv6_prefix_broader_than_32, valid_trusted_proxy};
 use crate::config_path::{validate_optional_process_path, validate_required_process_path};
 use crate::config_server_defaults::{
-    default_https_redirect_status, default_listen, default_max_request_body_bytes,
-    default_max_request_header_bytes, default_max_request_headers, default_max_uri_bytes,
-    default_process_certificate_reload_sock, default_process_listener_tasks_per_fd,
-    default_process_max_retries, default_process_pid_file, default_process_threads,
-    default_process_upgrade_sock, default_process_upstream_keepalive_pool_size, default_true,
-    validate_process_optional_duration, validate_process_usize,
+    default_https_redirect_status, default_listen, default_max_buffered_request_body_bytes,
+    default_max_request_body_bytes, default_max_request_header_bytes, default_max_request_headers,
+    default_max_uri_bytes, default_process_certificate_reload_sock,
+    default_process_listener_tasks_per_fd, default_process_max_retries, default_process_pid_file,
+    default_process_threads, default_process_upgrade_sock,
+    default_process_upstream_keepalive_pool_size, default_true, validate_process_optional_duration,
+    validate_process_usize,
 };
 
 pub const MAX_SERVER_LISTENERS: usize = 64;
@@ -370,6 +371,8 @@ pub struct ServerLimitsConfig {
     pub max_request_headers: usize,
     #[serde(default = "default_max_request_body_bytes")]
     pub max_request_body_bytes: ByteSize,
+    #[serde(default = "default_max_buffered_request_body_bytes")]
+    pub max_buffered_request_body_bytes: ByteSize,
 }
 
 impl Default for ServerLimitsConfig {
@@ -379,6 +382,7 @@ impl Default for ServerLimitsConfig {
             max_uri_bytes: default_max_uri_bytes(),
             max_request_headers: default_max_request_headers(),
             max_request_body_bytes: default_max_request_body_bytes(),
+            max_buffered_request_body_bytes: default_max_buffered_request_body_bytes(),
         }
     }
 }
@@ -405,6 +409,12 @@ impl ServerLimitsConfig {
                 field: "server.limits.max_request_body_bytes",
             });
         }
+        if self.max_buffered_request_body_bytes.as_u64() < self.max_request_body_bytes.as_u64() {
+            return Err(ConfigError::InvalidLimitRelationship {
+                field: "server.limits.max_buffered_request_body_bytes",
+                minimum_field: "server.limits.max_request_body_bytes",
+            });
+        }
 
         Ok(())
     }

crates/fluxheim-config/src/config_server_defaults.rs

@@ -26,6 +26,10 @@ pub(crate) fn default_max_request_body_bytes() -> ByteSize {
     ByteSize::from_bytes(16 * 1024 * 1024)
 }
 
+pub(crate) fn default_max_buffered_request_body_bytes() -> ByteSize {
+    ByteSize::from_bytes(1024 * 1024 * 1024)
+}
+
 pub(crate) fn default_process_pid_file() -> PathBuf {
     default_process_runtime_path("fluxheim.pid")
 }

crates/fluxheim-config/src/config_tests_server.rs

@@ -19,6 +19,7 @@ fn parses_server_limits() {
             max_uri_bytes = 4096
             max_request_headers = 32
             max_request_body_bytes = "2MiB"
+            max_buffered_request_body_bytes = "8MiB"
             "#,
     )
     .unwrap();
@@ -36,6 +37,10 @@ fn parses_server_limits() {
         config.server.limits.max_request_body_bytes,
         ByteSize::from_bytes(2 * 1024 * 1024)
     );
+    assert_eq!(
+        config.server.limits.max_buffered_request_body_bytes,
+        ByteSize::from_bytes(8 * 1024 * 1024)
+    );
     assert_eq!(
         config.server.trusted_proxies,
         ["127.0.0.1", "10.0.0.0/8", "2001:db8::/32", "2a06:98c0::/29"]
@@ -125,6 +130,26 @@ fn rejects_zero_server_limits() {
     );
 }
 
+#[test]
+fn rejects_aggregate_body_budget_below_per_request_limit() {
+    let config: Config = toml::from_str(
+        r#"
+            [server.limits]
+            max_request_body_bytes = "8MiB"
+            max_buffered_request_body_bytes = "4MiB"
+            "#,
+    )
+    .unwrap();
+
+    assert_eq!(
+        config.validate(),
+        Err(ConfigError::InvalidLimitRelationship {
+            field: "server.limits.max_buffered_request_body_bytes",
+            minimum_field: "server.limits.max_request_body_bytes",
+        })
+    );
+}
+
 #[test]
 fn rejects_empty_listeners() {
     let config = Config {

crates/fluxheim-server/src/blocking_work.rs

@@ -11,6 +11,7 @@ const MAX_WASM_BLOCKING_WORK: usize = 96;
 #[cfg(any(test, feature = "wasm"))]
 const MAX_WASM_PREVIEW_BLOCKING_WORK: usize = 32;
 const MAX_DISK_CACHE_BLOCKING_WORK: usize = 32;
+const MAX_STATIC_BLOCKING_WORK: usize = 8;
 #[cfg(any(test, all(feature = "traffic-mirror", not(feature = "privacy-mode"))))]
 const MAX_MIRROR_BLOCKING_WORK: usize = 8;
 const MAX_CRITICAL_BLOCKING_WORK: usize = 32;
@@ -26,6 +27,7 @@ pub(crate) enum NativeBlockingWorkClass {
     #[cfg(any(test, feature = "wasm"))]
     WasmPreview,
     DiskCache,
+    Static,
     #[cfg(any(test, all(feature = "traffic-mirror", not(feature = "privacy-mode"))))]
     Mirror,
     Critical,
@@ -48,6 +50,7 @@ struct NativeRequestBlockingWorkBudgets {
     #[cfg(any(test, feature = "wasm"))]
     wasm_preview: Arc<Semaphore>,
     disk_cache: Arc<Semaphore>,
+    static_web: Arc<Semaphore>,
     #[cfg(any(test, all(feature = "traffic-mirror", not(feature = "privacy-mode"))))]
     mirror: Arc<Semaphore>,
     critical: Arc<Semaphore>,
@@ -65,6 +68,7 @@ impl NativeRequestBlockingWorkBudgets {
             #[cfg(any(test, feature = "wasm"))]
             wasm_preview: Arc::new(Semaphore::new(MAX_WASM_PREVIEW_BLOCKING_WORK)),
             disk_cache: Arc::new(Semaphore::new(MAX_DISK_CACHE_BLOCKING_WORK)),
+            static_web: Arc::new(Semaphore::new(MAX_STATIC_BLOCKING_WORK)),
             #[cfg(any(test, all(feature = "traffic-mirror", not(feature = "privacy-mode"))))]
             mirror: Arc::new(Semaphore::new(MAX_MIRROR_BLOCKING_WORK)),
             critical: Arc::new(Semaphore::new(MAX_CRITICAL_BLOCKING_WORK)),
@@ -80,6 +84,7 @@ impl NativeRequestBlockingWorkBudgets {
             #[cfg(any(test, feature = "wasm"))]
             NativeBlockingWorkClass::WasmPreview => &self.wasm_preview,
             NativeBlockingWorkClass::DiskCache => &self.disk_cache,
+            NativeBlockingWorkClass::Static => &self.static_web,
             #[cfg(any(test, all(feature = "traffic-mirror", not(feature = "privacy-mode"))))]
             NativeBlockingWorkClass::Mirror => &self.mirror,
             NativeBlockingWorkClass::Critical => &self.critical,
@@ -162,6 +167,7 @@ mod tests {
                 .try_acquire(NativeBlockingWorkClass::DiskCache)
                 .is_ok()
         );
+        assert!(budgets.try_acquire(NativeBlockingWorkClass::Static).is_ok());
         assert!(
             budgets
                 .try_acquire(NativeBlockingWorkClass::Critical)

crates/fluxheim-server/src/lib.rs

@@ -101,6 +101,8 @@ mod native_runtime_manifest;
 mod plan;
 mod process;
 mod proxy_protocol;
+mod request_body_budget;
+mod response_retention;
 mod service;
 #[cfg(unix)]
 mod unix_listener;
@@ -187,6 +189,7 @@ pub use plan::{
 };
 pub use process::ProcessSpec;
 pub use proxy_protocol::{ProxyProtocolPolicy, ProxyProtocolTrustedSource};
+pub use request_body_budget::NativeRequestBodyBudget;
 pub use service::{AdminOpsSocketPlan, ServiceKind, ServiceSpec};
 #[cfg(unix)]
 pub use unix_listener::replace_private_unix_listener;

crates/fluxheim-server/src/native_http1.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use fluxheim_protocol::{
-    Http1ConnectionDirective, Http1HeadLimits, Http1Header, Http1ParseError,
+    Http1BodyFraming, Http1ConnectionDirective, Http1HeadLimits, Http1Header, Http1ParseError,
     parse_http1_request_head,
 };
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
@@ -96,6 +96,10 @@ pub trait NativeHttp1Handler: Send + Sync + 'static {
         None
     }
 
+    fn request_body_budget(&self) -> Option<crate::NativeRequestBodyBudget> {
+        None
+    }
+
     fn handle<'a>(
         &'a self,
         request: NativeHttp1Request,
@@ -214,13 +218,53 @@ where
                 owned_request_from_head(&head, peer_addr, &request_context),
             )
         };
+        let request_body_budget = handler.request_body_budget();
         let pinned_handler = handler.pin_request_handler();
         let request_handler = pinned_handler
             .as_deref()
             .unwrap_or_else(|| handler.as_ref());
         let request_body_timeout = request_handler
             .request_body_timeout(&request)
             .unwrap_or(policy.request_body_timeout());
+        let reservation_bytes = match body_framing {
+            Http1BodyFraming::NoBody => 0,
+            Http1BodyFraming::ContentLength(length) => {
+                let Ok(length) = usize::try_from(length) else {
+                    write_response(
+                        &mut stream,
+                        NativeHttp1Response::new(413, "Payload Too Large", b"payload too large\n")
+                            .close_connection(),
+                        true,
+                    )
+                    .await?;
+                    return Ok(());
+                };
+                length
+            }
+            Http1BodyFraming::Chunked => policy.max_body_bytes(),
+        };
+        let _request_body_permit = if let Some(budget) = request_body_budget {
+            match budget.reserve(reservation_bytes).await {
+                Ok(permit) => permit,
+                Err(_) => {
+                    write_response(
+                        &mut stream,
+                        NativeHttp1Response::new(
+                            503,
+                            "Service Unavailable",
+                            b"request body capacity unavailable\n",
+                        )
+                        .with_retry_after_secs(1)
+                        .close_connection(),
+                        true,
+                    )
+                    .await?;
+                    return Ok(());
+                }
+            }
+        } else {
+            None
+        };
         let body = match read_body(
             policy,
             request_body_timeout,