diff --git a/fluxer_app_proxy/src/config.rs b/fluxer_app_proxy/src/config.rs index bae9894c2..f69c67675 100644 --- a/fluxer_app_proxy/src/config.rs +++ b/fluxer_app_proxy/src/config.rs @@ -2,10 +2,340 @@ use fluxer_common::config::{self as cfg, GeoipS3Config, GeoipSourceConfig}; use fluxer_svc::config::{DatabaseBackend, normalize_host, parse_hosts}; +use reqwest::Url; use std::env; +use std::fmt; const DEFAULT_DISCOVERY_UPSTREAM_URL: &str = "http://localhost:8088/api/.well-known/fluxer"; +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum InvalidAppProxyEnvironmentError { + InvalidValue { + name: &'static str, + value: String, + expected: &'static str, + }, +} + +impl InvalidAppProxyEnvironmentError { + fn new(name: &'static str, value: &str, expected: &'static str) -> Self { + Self::InvalidValue { + name, + value: value.to_owned(), + expected, + } + } +} + +impl fmt::Display for InvalidAppProxyEnvironmentError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidValue { + name, + value, + expected, + } => write!(formatter, "{name} must be {expected}, got {value:?}"), + } + } +} + +impl std::error::Error for InvalidAppProxyEnvironmentError {} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct HttpUrl(Url); + +impl HttpUrl { + pub fn parse(name: &'static str, value: &str) -> Result { + let url = Url::parse(value.trim()).map_err(|_| { + InvalidAppProxyEnvironmentError::new(name, value, "a valid HTTP or HTTPS URL") + })?; + if !matches!(url.scheme(), "http" | "https") + || url.host_str().is_none() + || !url.username().is_empty() + || url.password().is_some() + || url.fragment().is_some() + { + return Err(InvalidAppProxyEnvironmentError::new( + name, + value, + "an HTTP or HTTPS URL with a host and no credentials or fragment", + )); + } + Ok(Self(url)) + } + + pub fn as_url(&self) -> &Url { + &self.0 + } +} + +impl fmt::Display for HttpUrl { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(formatter) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct HttpEndpoint { + url: Url, + csp_origin: String, +} + +impl HttpEndpoint { + pub fn parse(name: &'static str, value: &str) -> Result { + let mut url = HttpUrl::parse(name, value)?.0; + if url.query().is_some() { + return Err(InvalidAppProxyEnvironmentError::new( + name, + value, + "an HTTP or HTTPS endpoint without a query or fragment", + )); + } + if !url.path().ends_with('/') { + let mut path = url.path().to_owned(); + path.push('/'); + url.set_path(&path); + } + let csp_origin = url.origin().ascii_serialization(); + if csp_origin == "null" { + return Err(InvalidAppProxyEnvironmentError::new( + name, + value, + "an HTTP or HTTPS endpoint with a tuple origin", + )); + } + Ok(Self { url, csp_origin }) + } + + pub fn with_host_prefix( + &self, + name: &'static str, + prefix: &str, + ) -> Result { + if !is_dns_bucket_name(prefix) { + return Err(InvalidAppProxyEnvironmentError::new( + name, + prefix, + "a DNS-compatible bucket name", + )); + } + let host = self + .url + .host_str() + .expect("validated HTTP endpoint must have a host"); + let prefixed_host = if host.starts_with(&format!("{prefix}.")) { + host.to_owned() + } else { + format!("{prefix}.{host}") + }; + let mut url = self.url.clone(); + url.set_host(Some(&prefixed_host)).map_err(|_| { + InvalidAppProxyEnvironmentError::new(name, prefix, "a DNS-compatible bucket name") + })?; + let csp_origin = url.origin().ascii_serialization(); + Ok(Self { url, csp_origin }) + } + + pub fn as_url(&self) -> &Url { + &self.url + } + + pub fn as_str(&self) -> &str { + self.url.as_str().trim_end_matches('/') + } + + pub fn csp_origin(&self) -> &str { + &self.csp_origin + } +} + +fn is_dns_bucket_name(value: &str) -> bool { + if value.is_empty() || value.len() > 253 { + return false; + } + value.split('.').all(|label| { + if label.is_empty() || label.len() > 63 { + return false; + } + let bytes = label.as_bytes(); + if !bytes[0].is_ascii_lowercase() && !bytes[0].is_ascii_digit() { + return false; + } + if !bytes[bytes.len() - 1].is_ascii_lowercase() && !bytes[bytes.len() - 1].is_ascii_digit() + { + return false; + } + bytes + .iter() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || *byte == b'-') + }) +} + +impl fmt::Display for HttpEndpoint { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.as_str()) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CspSource(String); + +impl CspSource { + pub fn parse(name: &'static str, value: &str) -> Result { + if value + .bytes() + .any(|byte| byte.is_ascii_whitespace() || matches!(byte, b';' | b',')) + { + return Err(InvalidAppProxyEnvironmentError::new( + name, + value, + "one CSP source without whitespace or policy delimiters", + )); + } + if value == "*" { + return Ok(Self(value.to_owned())); + } + if is_csp_keyword_source(value) || is_csp_nonce_or_hash_source(value) { + return Ok(Self(value.to_owned())); + } + if matches!( + value, + "http:" | "https:" | "ws:" | "wss:" | "data:" | "blob:" + ) { + return Ok(Self(value.to_owned())); + } + if let Some(source) = parse_csp_network_source(value) { + return Ok(Self(source)); + } + Err(InvalidAppProxyEnvironmentError::new( + name, + value, + "a supported CSP keyword, scheme, wildcard, nonce, hash, or HTTP(S)/WS(S) source", + )) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +fn is_csp_keyword_source(value: &str) -> bool { + matches!( + value, + "'self'" + | "'unsafe-inline'" + | "'unsafe-eval'" + | "'wasm-unsafe-eval'" + | "'strict-dynamic'" + | "'report-sample'" + ) +} + +fn is_csp_nonce_or_hash_source(value: &str) -> bool { + let Some(inner) = value + .strip_prefix('\'') + .and_then(|value| value.strip_suffix('\'')) + else { + return false; + }; + let Some((algorithm, encoded)) = inner.split_once('-') else { + return false; + }; + if !matches!(algorithm, "nonce" | "sha256" | "sha384" | "sha512") || encoded.is_empty() { + return false; + } + encoded.bytes().all(|byte| { + byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'/' | b'_' | b'-' | b'=') + }) +} + +fn parse_csp_network_source(value: &str) -> Option { + let (scheme, authority_and_path) = value.split_once("://")?; + if !matches!(scheme, "http" | "https" | "ws" | "wss") { + return None; + } + let wildcard = authority_and_path.starts_with("*."); + let parse_value = if wildcard { + format!( + "{scheme}://csp-wildcard.invalid.{}", + &authority_and_path[2..] + ) + } else { + value.to_owned() + }; + let url = Url::parse(&parse_value).ok()?; + if url.host_str().is_none() + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + return None; + } + let mut source = url.origin().ascii_serialization(); + if source == "null" { + return None; + } + if wildcard { + source = source.replacen("csp-wildcard.invalid.", "*.", 1); + } + if url.path() != "/" { + source.push_str(url.path()); + } + Some(source) +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CspReportUri(HttpUrl); + +impl CspReportUri { + pub fn parse(name: &'static str, value: &str) -> Result { + if value + .bytes() + .any(|byte| byte.is_ascii_whitespace() || matches!(byte, b';' | b',')) + { + return Err(InvalidAppProxyEnvironmentError::new( + name, + value, + "one HTTP or HTTPS report URI without whitespace or policy delimiters", + )); + } + Ok(Self(HttpUrl::parse(name, value)?)) + } +} + +impl fmt::Display for CspReportUri { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(formatter) + } +} + +fn warn_invalid(error: InvalidAppProxyEnvironmentError) { + tracing::warn!(%error, "ignoring invalid app proxy environment value"); +} + +fn parse_optional_http_url(name: &'static str, value: Option) -> Option { + let value = value?; + match HttpUrl::parse(name, &value) { + Ok(url) => Some(url), + Err(error) => { + warn_invalid(error); + None + } + } +} + +fn parse_optional_http_endpoint(name: &'static str, value: Option) -> Option { + let value = value?; + match HttpEndpoint::parse(name, &value) { + Ok(endpoint) => Some(endpoint), + Err(error) => { + warn_invalid(error); + None + } + } +} + fn parse_env_or_warn(name: &str, raw: &str, default: T) -> T { raw.parse::().unwrap_or_else(|_| { tracing::warn!( @@ -22,10 +352,10 @@ pub struct AppProxyConfig { pub host: String, pub port: u16, pub static_dir: String, - pub index_upstream_url: Option, - pub static_cdn_endpoint: Option, - pub s3_public_endpoint: Option, - pub s3_uploads_bucket: String, + pub index_upstream_url: Option, + pub static_cdn_endpoint: Option, + pub s3_public_endpoint: Option, + pub s3_uploads_endpoint: Option, pub discovery_upstream_url: String, pub discovery_refresh_interval_ms: u64, pub release_channel: ReleaseChannel, @@ -88,17 +418,17 @@ impl ReleaseChannel { #[derive(Clone, Debug, Default)] pub struct CspConfig { - pub extra_default_src: Option>, - pub extra_connect_src: Option>, - pub extra_img_src: Option>, - pub extra_media_src: Option>, - pub extra_font_src: Option>, - pub extra_script_src: Option>, - pub extra_style_src: Option>, - pub extra_frame_src: Option>, - pub extra_worker_src: Option>, - pub extra_manifest_src: Option>, - pub report_uri: Option, + pub extra_default_src: Vec, + pub extra_connect_src: Vec, + pub extra_img_src: Vec, + pub extra_media_src: Vec, + pub extra_font_src: Vec, + pub extra_script_src: Vec, + pub extra_style_src: Vec, + pub extra_frame_src: Vec, + pub extra_worker_src: Vec, + pub extra_manifest_src: Vec, + pub report_uri: Option, } impl CspConfig { @@ -114,23 +444,34 @@ impl CspConfig { extra_frame_src: read_csp_sources("FLUXER_CSP_EXTRA_FRAME_SRC"), extra_worker_src: read_csp_sources("FLUXER_CSP_EXTRA_WORKER_SRC"), extra_manifest_src: read_csp_sources("FLUXER_CSP_EXTRA_MANIFEST_SRC"), - report_uri: cfg::non_empty_env("FLUXER_CSP_REPORT_URI"), + report_uri: read_csp_report_uri("FLUXER_CSP_REPORT_URI"), } } } -fn read_csp_sources(name: &str) -> Option> { - let sources: Vec = cfg::read_env(name, "") +fn read_csp_sources(name: &'static str) -> Vec { + cfg::read_env(name, "") .split([',', ' ', '\t', '\n']) .map(str::trim) .filter(|source| !source.is_empty()) - .map(str::to_owned) - .collect(); + .filter_map(|source| match CspSource::parse(name, source) { + Ok(source) => Some(source), + Err(error) => { + warn_invalid(error); + None + } + }) + .collect() +} - if sources.is_empty() { - None - } else { - Some(sources) +fn read_csp_report_uri(name: &'static str) -> Option { + let value = cfg::non_empty_env(name)?; + match CspReportUri::parse(name, &value) { + Ok(report_uri) => Some(report_uri), + Err(error) => { + warn_invalid(error); + None + } } } @@ -175,6 +516,21 @@ impl AppProxyConfig { ) .max(1); + let s3_public_endpoint = parse_optional_http_endpoint( + "FLUXER_S3_PUBLIC_ENDPOINT", + cfg::non_empty_env("FLUXER_S3_PUBLIC_ENDPOINT"), + ); + let s3_uploads_bucket = cfg::read_env("FLUXER_S3_BUCKET_UPLOADS", "fluxer-uploads"); + let s3_uploads_endpoint = s3_public_endpoint.as_ref().and_then(|endpoint| { + match endpoint.with_host_prefix("FLUXER_S3_BUCKET_UPLOADS", s3_uploads_bucket.trim()) { + Ok(endpoint) => Some(endpoint), + Err(error) => { + warn_invalid(error); + None + } + } + }); + Self { host: cfg::read_env("FLUXER_APP_PROXY_HOST", "0.0.0.0"), port: parse_env_or_warn( @@ -183,10 +539,16 @@ impl AppProxyConfig { 8080u16, ), static_dir: cfg::read_env("FLUXER_STATIC_DIR", "./static"), - index_upstream_url: cfg::non_empty_env("FLUXER_APP_PROXY_INDEX_UPSTREAM_URL"), - static_cdn_endpoint: cfg::non_empty_env("FLUXER_STATIC_CDN_ENDPOINT"), - s3_public_endpoint: cfg::non_empty_env("FLUXER_S3_PUBLIC_ENDPOINT"), - s3_uploads_bucket: cfg::read_env("FLUXER_S3_BUCKET_UPLOADS", "fluxer-uploads"), + index_upstream_url: parse_optional_http_url( + "FLUXER_APP_PROXY_INDEX_UPSTREAM_URL", + cfg::non_empty_env("FLUXER_APP_PROXY_INDEX_UPSTREAM_URL"), + ), + static_cdn_endpoint: parse_optional_http_endpoint( + "FLUXER_STATIC_CDN_ENDPOINT", + cfg::non_empty_env("FLUXER_STATIC_CDN_ENDPOINT"), + ), + s3_public_endpoint: s3_public_endpoint.clone(), + s3_uploads_endpoint, discovery_upstream_url: resolve_discovery_upstream_url_from_env(), discovery_refresh_interval_ms: parse_env_or_warn( "DISCOVERY_REFRESH_INTERVAL_MS", @@ -458,7 +820,9 @@ mod tests { fn csp_config_default_has_no_extra_sources() { let c = CspConfig::default(); assert!( - c.extra_default_src.is_none() && c.extra_script_src.is_none() && c.report_uri.is_none() + c.extra_default_src.is_empty() + && c.extra_script_src.is_empty() + && c.report_uri.is_none() ); } diff --git a/fluxer_app_proxy/src/csp.rs b/fluxer_app_proxy/src/csp.rs index c011265cd..db011142c 100644 --- a/fluxer_app_proxy/src/csp.rs +++ b/fluxer_app_proxy/src/csp.rs @@ -1,16 +1,25 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -use crate::config::CspConfig; +use crate::config::{AppProxyConfig, CspConfig, CspSource, HttpEndpoint}; +use axum::http::HeaderValue; +use axum::http::header::InvalidHeaderValue; use rand::RngExt; -use reqwest::Url; + +const CSP_NONCE_HEX_DIGITS: usize = 32; +const CSP_VALIDATION_NONCE: &str = "00000000000000000000000000000000"; + +const _: () = assert!( + CSP_VALIDATION_NONCE.len() == CSP_NONCE_HEX_DIGITS, + "the nonce a policy is validated with must be shaped like the nonce a request carries" +); #[derive(Clone, Debug, Default)] pub struct RuntimeCspSources { - pub static_cdn_endpoint: Option, - pub media_endpoint: Option, - pub s3_public_endpoint: Option, - pub s3_uploads_bucket: Option, - pub branding_image_origins: Vec, + pub static_cdn_endpoint: Option, + pub media_endpoint: Option, + pub s3_public_endpoint: Option, + pub s3_uploads_endpoint: Option, + pub branding_image_origins: Vec, } const FRAME_SOURCES: &[&str] = &[ @@ -74,16 +83,99 @@ const WORKER_SOURCES: &[&str] = &["https://*.fluxer.app", "blob:"]; const MANIFEST_SOURCES: &[&str] = &["https://*.fluxer.app"]; +#[derive(Debug)] +pub enum CspCompileError { + InvalidAssetPolicy(InvalidHeaderValue), + InvalidSpaPolicy(InvalidHeaderValue), +} + +impl std::fmt::Display for CspCompileError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::InvalidAssetPolicy(_) => { + formatter.write_str("the asset content security policy is not a valid header value") + } + Self::InvalidSpaPolicy(_) => { + formatter.write_str("the SPA content security policy is not a valid header value") + } + } + } +} + +impl std::error::Error for CspCompileError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::InvalidAssetPolicy(source) | Self::InvalidSpaPolicy(source) => Some(source), + } + } +} + +#[derive(Clone, Debug)] +pub struct CompiledCspPolicy { + config: CspConfig, + asset: HeaderValue, +} + +impl CompiledCspPolicy { + pub fn from_config(config: &AppProxyConfig) -> Result { + Self::compile( + config.csp.clone(), + &RuntimeCspSources { + static_cdn_endpoint: config.static_cdn_endpoint.clone(), + media_endpoint: None, + s3_public_endpoint: config.s3_public_endpoint.clone(), + s3_uploads_endpoint: config.s3_uploads_endpoint.clone(), + branding_image_origins: Vec::new(), + }, + ) + } + + pub fn compile( + config: CspConfig, + configured_sources: &RuntimeCspSources, + ) -> Result { + let asset_sources = RuntimeCspSources { + static_cdn_endpoint: configured_sources.static_cdn_endpoint.clone(), + ..RuntimeCspSources::default() + }; + let asset = HeaderValue::from_str(&build_asset_csp(&config, &asset_sources)) + .map_err(CspCompileError::InvalidAssetPolicy)?; + HeaderValue::from_str(&build_csp( + &config, + CSP_VALIDATION_NONCE, + configured_sources, + )) + .map_err(CspCompileError::InvalidSpaPolicy)?; + Ok(Self { config, asset }) + } + + pub fn asset_header(&self) -> HeaderValue { + self.asset.clone() + } + + pub fn spa_header(&self, nonce: &str, runtime_sources: &RuntimeCspSources) -> HeaderValue { + assert!( + nonce.len() == CSP_NONCE_HEX_DIGITS + && nonce.bytes().all(|byte| byte.is_ascii_hexdigit()), + "a CSP nonce must be a 128-bit hexadecimal value" + ); + HeaderValue::from_str(&build_csp(&self.config, nonce, runtime_sources)).expect( + "every CSP source is a validated keyword, scheme, or ASCII origin, so a policy built \ + from them is always a valid header value", + ) + } +} + pub fn generate_nonce() -> String { let bytes: [u8; 16] = rand::rng().random(); hex::encode(bytes) } -pub fn build_csp(config: &CspConfig, nonce: &str, runtime_sources: &RuntimeCspSources) -> String { +fn build_csp(config: &CspConfig, nonce: &str, runtime_sources: &RuntimeCspSources) -> String { build_csp_directives(config, Some(nonce), runtime_sources).join("; ") } -pub fn build_asset_csp(config: &CspConfig, runtime_sources: &RuntimeCspSources) -> String { +fn build_asset_csp(config: &CspConfig, runtime_sources: &RuntimeCspSources) -> String { build_csp_directives(config, None, runtime_sources).join("; ") } @@ -95,7 +187,7 @@ fn build_csp_directives( let mut directives = Vec::with_capacity(14); let mut default = vec!["'self'".to_owned()]; - extend_from(&mut default, config.extra_default_src.as_deref(), &[]); + extend_from(&mut default, &config.extra_default_src, &[]); directives.push(format!("default-src {}", default.join(" "))); let mut script = vec![ @@ -106,21 +198,17 @@ fn build_csp_directives( if let Some(n) = nonce { script.insert(1, format!("'nonce-{n}'")); } - extend_from( - &mut script, - config.extra_script_src.as_deref(), - SCRIPT_SOURCES, - ); + extend_from(&mut script, &config.extra_script_src, SCRIPT_SOURCES); extend_runtime_sources(&mut script, runtime_sources, true, false); directives.push(format!("script-src {}", script.join(" "))); let mut style = vec!["'self'".to_owned(), "'unsafe-inline'".to_owned()]; - extend_from(&mut style, config.extra_style_src.as_deref(), STYLE_SOURCES); + extend_from(&mut style, &config.extra_style_src, STYLE_SOURCES); extend_runtime_sources(&mut style, runtime_sources, true, true); directives.push(format!("style-src {}", style.join(" "))); let mut img = vec!["'self'".to_owned(), "blob:".to_owned(), "data:".to_owned()]; - extend_from(&mut img, config.extra_img_src.as_deref(), IMAGE_SOURCES); + extend_from(&mut img, &config.extra_img_src, IMAGE_SOURCES); extend_runtime_sources(&mut img, runtime_sources, true, true); for origin in &runtime_sources.branding_image_origins { push_endpoint_source(&mut img, Some(origin)); @@ -128,44 +216,32 @@ fn build_csp_directives( directives.push(format!("img-src {}", img.join(" "))); let mut media = vec!["'self'".to_owned(), "blob:".to_owned()]; - extend_from(&mut media, config.extra_media_src.as_deref(), MEDIA_SOURCES); + extend_from(&mut media, &config.extra_media_src, MEDIA_SOURCES); extend_runtime_sources(&mut media, runtime_sources, true, true); directives.push(format!("media-src {}", media.join(" "))); let mut font = vec!["'self'".to_owned(), "data:".to_owned()]; - extend_from(&mut font, config.extra_font_src.as_deref(), FONT_SOURCES); + extend_from(&mut font, &config.extra_font_src, FONT_SOURCES); extend_runtime_sources(&mut font, runtime_sources, true, true); directives.push(format!("font-src {}", font.join(" "))); let mut connect = vec!["'self'".to_owned(), "data:".to_owned()]; - extend_from( - &mut connect, - config.extra_connect_src.as_deref(), - CONNECT_SOURCES, - ); + extend_from(&mut connect, &config.extra_connect_src, CONNECT_SOURCES); extend_runtime_sources(&mut connect, runtime_sources, true, true); extend_runtime_s3_sources(&mut connect, runtime_sources); directives.push(format!("connect-src {}", connect.join(" "))); let mut frame = vec!["'self'".to_owned()]; - extend_from(&mut frame, config.extra_frame_src.as_deref(), FRAME_SOURCES); + extend_from(&mut frame, &config.extra_frame_src, FRAME_SOURCES); directives.push(format!("frame-src {}", frame.join(" "))); let mut worker = vec!["'self'".to_owned(), "blob:".to_owned()]; - extend_from( - &mut worker, - config.extra_worker_src.as_deref(), - WORKER_SOURCES, - ); + extend_from(&mut worker, &config.extra_worker_src, WORKER_SOURCES); extend_runtime_sources(&mut worker, runtime_sources, true, false); directives.push(format!("worker-src {}", worker.join(" "))); let mut manifest = vec!["'self'".to_owned()]; - extend_from( - &mut manifest, - config.extra_manifest_src.as_deref(), - MANIFEST_SOURCES, - ); + extend_from(&mut manifest, &config.extra_manifest_src, MANIFEST_SOURCES); extend_runtime_sources(&mut manifest, runtime_sources, true, false); directives.push(format!("manifest-src {}", manifest.join(" "))); @@ -187,78 +263,40 @@ fn extend_runtime_sources( include_media: bool, ) { if include_static { - push_endpoint_source(target, runtime_sources.static_cdn_endpoint.as_deref()); + push_endpoint_source(target, runtime_sources.static_cdn_endpoint.as_ref()); } if include_media { - push_endpoint_source(target, runtime_sources.media_endpoint.as_deref()); + push_endpoint_source(target, runtime_sources.media_endpoint.as_ref()); } } -pub fn http_origin(raw: &str) -> Option { - let url = Url::parse(raw.trim()).ok()?; - let scheme = url.scheme(); - if scheme != "http" && scheme != "https" { - return None; - } - let host = url.host_str()?; - let port = url - .port() - .map(|port| format!(":{port}")) - .unwrap_or_default(); - Some(format!("{scheme}://{host}{port}")) -} - -fn push_endpoint_source(target: &mut Vec, endpoint: Option<&str>) { +fn push_endpoint_source(target: &mut Vec, endpoint: Option<&HttpEndpoint>) { let Some(endpoint) = endpoint else { return; }; - let source = endpoint.trim().trim_end_matches('/'); - if source.is_empty() || target.iter().any(|existing| existing == source) { + let source = endpoint.csp_origin(); + if target.iter().any(|existing| existing == source) { return; } target.push(source.to_owned()); } fn extend_runtime_s3_sources(target: &mut Vec, runtime_sources: &RuntimeCspSources) { - push_endpoint_source(target, runtime_sources.s3_public_endpoint.as_deref()); - - let Some(source) = s3_uploads_bucket_origin(runtime_sources) else { - return; - }; - push_endpoint_source(target, Some(&source)); + push_endpoint_source(target, runtime_sources.s3_public_endpoint.as_ref()); + push_endpoint_source(target, runtime_sources.s3_uploads_endpoint.as_ref()); } -fn s3_uploads_bucket_origin(runtime_sources: &RuntimeCspSources) -> Option { - let bucket = runtime_sources - .s3_uploads_bucket - .as_deref() - .map(str::trim) - .filter(|value| !value.is_empty())?; - let endpoint = runtime_sources.s3_public_endpoint.as_deref()?.trim(); - let url = Url::parse(endpoint).ok()?; - let scheme = url.scheme(); - if scheme != "http" && scheme != "https" { - return None; +fn extend_from(target: &mut Vec, extra: &[CspSource], defaults: &[&str]) { + for source in defaults { + if target.iter().any(|existing| existing == source) { + continue; + } + target.push((*source).to_owned()); } - let host = url.host_str()?; - let host = if host.starts_with(&format!("{bucket}.")) { - host.to_owned() - } else { - format!("{bucket}.{host}") - }; - let port = url - .port() - .map(|port| format!(":{port}")) - .unwrap_or_default(); - Some(format!("{scheme}://{host}{port}")) -} -fn extend_from(target: &mut Vec, extra: Option<&[String]>, defaults: &[&str]) { - target.extend(defaults.iter().map(|s| (*s).to_owned())); - - for source in extra.into_iter().flatten() { - let source = source.trim(); - if source.is_empty() || target.iter().any(|existing| existing == source) { + for source in extra { + let source = source.as_str(); + if target.iter().any(|existing| existing == source) { continue; } target.push(source.to_owned()); @@ -272,8 +310,11 @@ mod tests { #[test] fn generate_nonce_produces_32_char_hex() { let nonce = generate_nonce(); - assert_eq!(nonce.len(), 32); + assert_eq!(nonce.len(), CSP_NONCE_HEX_DIGITS); assert!(nonce.chars().all(|c| c.is_ascii_hexdigit())); + CompiledCspPolicy::compile(default_csp_config(), &runtime_sources()) + .unwrap() + .spa_header(&nonce, &runtime_sources()); } #[test] @@ -291,6 +332,10 @@ mod tests { RuntimeCspSources::default() } + fn endpoint(value: &str) -> HttpEndpoint { + HttpEndpoint::parse("TEST_ENDPOINT", value).unwrap() + } + #[test] fn build_csp_includes_required_directives() { let config = default_csp_config(); @@ -336,7 +381,13 @@ mod tests { #[test] fn build_csp_includes_report_uri_when_configured() { let config = CspConfig { - report_uri: Some("https://example.com/csp-report".to_owned()), + report_uri: Some( + crate::config::CspReportUri::parse( + "TEST_CSP_REPORT_URI", + "https://example.com/csp-report", + ) + .unwrap(), + ), ..Default::default() }; let csp = build_csp(&config, "nonce1", &runtime_sources()); @@ -354,8 +405,8 @@ mod tests { fn build_csp_includes_configured_runtime_endpoints() { let config = default_csp_config(); let runtime_sources = RuntimeCspSources { - static_cdn_endpoint: Some("https://static.example.test/".to_owned()), - media_endpoint: Some("https://media.example.test".to_owned()), + static_cdn_endpoint: Some(endpoint("https://static.example.test/")), + media_endpoint: Some(endpoint("https://media.example.test")), ..Default::default() }; let csp = build_csp(&config, "nonce1", &runtime_sources); @@ -365,12 +416,38 @@ mod tests { assert!(!csp.contains("https://static.example.test/ ")); } + #[test] + fn a_csp_source_cannot_smuggle_a_second_directive() { + for injected in [ + "https://evil.test; script-src *", + "https://evil.test,https://other.test", + "https://evil.test https://other.test", + "https://evil.test\nscript-src *", + ] { + assert!( + CspSource::parse("TEST_CSP_SOURCE", injected).is_err(), + "{injected:?} must not parse as a single CSP source" + ); + } + } + + #[test] + fn a_report_uri_cannot_smuggle_a_second_directive() { + assert!( + crate::config::CspReportUri::parse( + "TEST_CSP_REPORT_URI", + "https://evil.test/r; script-src *" + ) + .is_err() + ); + } + #[test] fn build_csp_includes_s3_public_and_virtual_hosted_upload_origins() { let config = default_csp_config(); let runtime_sources = RuntimeCspSources { - s3_public_endpoint: Some("http://localhost:3900/".to_owned()), - s3_uploads_bucket: Some("fluxer-uploads".to_owned()), + s3_public_endpoint: Some(endpoint("http://localhost:3900/")), + s3_uploads_endpoint: Some(endpoint("http://fluxer-uploads.localhost:3900/")), ..Default::default() }; @@ -380,4 +457,72 @@ mod tests { assert!(csp.contains("http://fluxer-uploads.localhost:3900")); assert!(!csp.contains("http://localhost:3900/ ")); } + + #[test] + fn a_compiled_asset_header_is_the_policy_every_asset_response_reuses() { + let sources = RuntimeCspSources { + static_cdn_endpoint: Some(endpoint("https://static.example.test/")), + media_endpoint: Some(endpoint("https://media.example.test")), + s3_public_endpoint: Some(endpoint("http://localhost:3900/")), + ..Default::default() + }; + let policy = CompiledCspPolicy::compile(default_csp_config(), &sources).unwrap(); + + assert_eq!(policy.asset_header(), policy.asset_header()); + let asset = policy.asset_header(); + let asset = asset.to_str().unwrap(); + assert!(!asset.contains("nonce-")); + assert!(asset.contains("https://static.example.test")); + assert!( + !asset.contains("https://media.example.test"), + "an asset response must not widen the policy with the endpoints only the document needs" + ); + assert!(!asset.contains("http://localhost:3900")); + } + + #[test] + fn a_compiled_policy_stamps_the_requests_own_nonce_and_discovery_endpoints() { + let policy = CompiledCspPolicy::compile(default_csp_config(), &runtime_sources()).unwrap(); + let discovered = RuntimeCspSources { + static_cdn_endpoint: Some(endpoint("https://cdn.discovered.test")), + branding_image_origins: vec![endpoint("https://branding.discovered.test")], + ..Default::default() + }; + + let header = policy.spa_header("0123456789abcdef0123456789abcdef", &discovered); + let header = header.to_str().unwrap(); + + assert!(header.contains("'nonce-0123456789abcdef0123456789abcdef'")); + assert!(header.contains("https://cdn.discovered.test")); + assert!(header.contains("https://branding.discovered.test")); + } + + #[test] + fn a_compiled_policy_matches_the_directives_it_was_compiled_from() { + let config = default_csp_config(); + let sources = RuntimeCspSources { + static_cdn_endpoint: Some(endpoint("https://static.example.test/")), + ..Default::default() + }; + let policy = CompiledCspPolicy::compile(config.clone(), &sources).unwrap(); + + assert_eq!( + policy.asset_header().to_str().unwrap(), + build_asset_csp(&config, &sources) + ); + assert_eq!( + policy + .spa_header(CSP_VALIDATION_NONCE, &sources) + .to_str() + .unwrap(), + build_csp(&config, CSP_VALIDATION_NONCE, &sources) + ); + } + + #[test] + #[should_panic(expected = "a CSP nonce must be a 128-bit hexadecimal value")] + fn a_compiled_policy_refuses_a_nonce_it_did_not_generate() { + let policy = CompiledCspPolicy::compile(default_csp_config(), &runtime_sources()).unwrap(); + policy.spa_header("not-a-nonce", &runtime_sources()); + } } diff --git a/fluxer_app_proxy/src/discovery_cache.rs b/fluxer_app_proxy/src/discovery_cache.rs index d26a2445b..cc38785a7 100644 --- a/fluxer_app_proxy/src/discovery_cache.rs +++ b/fluxer_app_proxy/src/discovery_cache.rs @@ -1,5 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +use crate::config::HttpEndpoint; use serde::{Deserialize, Serialize}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -14,6 +15,26 @@ pub struct DiscoveryResponse { pub data: serde_json::Value, } +pub fn discovery_endpoint( + discovery: &DiscoveryResponse, + key: &'static str, +) -> Option { + let raw = discovery + .data + .get("endpoints") + .and_then(|endpoints| endpoints.get(key)) + .and_then(|value| value.as_str()) + .map(str::trim) + .filter(|value| !value.is_empty())?; + match HttpEndpoint::parse(key, raw) { + Ok(endpoint) => Some(endpoint), + Err(error) => { + tracing::warn!(%error, "ignoring invalid discovery endpoint"); + None + } + } +} + pub struct DiscoveryCache { cached: RwLock>, cold_start_attempt: Mutex>, diff --git a/fluxer_app_proxy/src/main.rs b/fluxer_app_proxy/src/main.rs index 20b05281e..a743314d1 100644 --- a/fluxer_app_proxy/src/main.rs +++ b/fluxer_app_proxy/src/main.rs @@ -3,10 +3,13 @@ use anyhow::Context; use fluxer_app_proxy::{ config::AppProxyConfig, + csp::CompiledCspPolicy, discovery_cache::DiscoveryCache, geoip, invite_meta, routes::build_router, - state::{AppState, build_http_client}, + state::{ + AppProxyBudgets, AppState, MAX_SPA_INDEX_BYTES, build_http_client, read_bounded_text_file, + }, }; use std::sync::{Arc, OnceLock}; use tokio::{net::TcpListener, runtime::Builder}; @@ -23,6 +26,11 @@ fn main() -> anyhow::Result<()> { let config = Arc::new(AppProxyConfig::from_env()); let addr = format!("{}:{}", config.host, config.port); + let csp = Arc::new( + CompiledCspPolicy::from_config(&config) + .context("failed to compile the Fluxer app proxy content security policy")?, + ); + let geoip = Arc::new(geoip::resolver_from_app_config(&config)); let runtime = Builder::new_multi_thread() @@ -55,7 +63,7 @@ fn main() -> anyhow::Result<()> { let index_html = if config.index_upstream_url.is_none() { let index_path = std::path::Path::new(&config.static_dir).join("index.html"); - match tokio::fs::read_to_string(&index_path).await { + match read_bounded_text_file(&index_path, MAX_SPA_INDEX_BYTES).await { Ok(contents) => Some(Arc::::from(contents)), Err(err) => { tracing::warn!(path = ?index_path, %err, "failed to preload index.html; will read per request"); @@ -68,11 +76,13 @@ fn main() -> anyhow::Result<()> { let state = AppState { config, + csp, http_client, discovery_cache, geoip, invite_meta, index_html, + budgets: AppProxyBudgets::default(), }; let router = build_router(state); diff --git a/fluxer_app_proxy/src/routes/assets_proxy.rs b/fluxer_app_proxy/src/routes/assets_proxy.rs index 23906c4c7..664ed6c43 100644 --- a/fluxer_app_proxy/src/routes/assets_proxy.rs +++ b/fluxer_app_proxy/src/routes/assets_proxy.rs @@ -1,7 +1,6 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -use crate::csp::{RuntimeCspSources, build_asset_csp}; -use crate::state::AppState; +use crate::state::{AppProxyBudgets, AppState}; use axum::{ body::Body, extract::{Path, State}, @@ -10,6 +9,9 @@ use axum::{ }; use std::path::{Path as FsPath, PathBuf}; use std::time::Duration; +use tokio::io::{AsyncWriteExt, DuplexStream}; +use tokio::sync::{OwnedSemaphorePermit, TryAcquireError}; +use tokio_util::io::ReaderStream; use super::file_stream::stream_file; use super::spa_static::{CORS_ALLOW_ANY_VALUE, asset_cache_control, guess_mime, is_font_mime}; @@ -17,6 +19,7 @@ use super::spa_static::{CORS_ALLOW_ANY_VALUE, asset_cache_control, guess_mime, i const ASSET_REQUEST_TIMEOUT: Duration = Duration::from_secs(15); const PRECOMPRESSED_VARIANTS: &[(&str, &str)] = &[("br", "br"), ("gzip", "gz")]; const MAX_ASSET_SIZE_BYTES: u64 = 100 * 1024 * 1024; +const UPSTREAM_ASSET_PUMP_BUFFER_BYTES: usize = 64 * 1024; const UPSTREAM_FAILURE_CACHE_CONTROL: &str = "no-store"; const UPSTREAM_FAILURE_STRIPPED_HEADERS: &[&str] = &[ "cdn-cache-control", @@ -60,21 +63,38 @@ pub async fn proxy_assets( ) -> Response { let Some(cdn_endpoint) = &state.config.static_cdn_endpoint else { return serve_local_asset( + &state.budgets, &state.config.static_dir, &format!("assets/{path}"), request.headers(), + state.csp.asset_header(), ) .await; }; - let target_url = format!("{cdn_endpoint}/assets/{path}"); + let target_url = format!("{}/assets/{path}", cdn_endpoint.as_str()); let upstream_host = cdn_endpoint - .trim_start_matches("https://") - .trim_start_matches("http://") - .split('/') - .next() - .unwrap_or("localhost"); + .as_url() + .host_str() + .map(|host| match cdn_endpoint.as_url().port() { + Some(port) => format!("{host}:{port}"), + None => host.to_owned(), + }) + .unwrap_or_else(|| "localhost".to_owned()); + + let upstream_slot = match state + .budgets + .upstream_asset_slots + .clone() + .try_acquire_owned() + { + Ok(permit) => permit, + Err(TryAcquireError::NoPermits) => return super::capacity_refused_response(), + Err(TryAcquireError::Closed) => { + panic!("upstream asset slot semaphore closed unexpectedly") + } + }; let mut request_builder = state .http_client @@ -88,7 +108,7 @@ pub async fn proxy_assets( } request_builder = request_builder.header(name.clone(), value.clone()); } - request_builder = request_builder.header("host", upstream_host); + request_builder = request_builder.header("host", upstream_host.as_str()); let upstream_response = match request_builder.send().await { Ok(resp) => resp, @@ -125,33 +145,54 @@ pub async fn proxy_assets( set_proxied_cache_control(&mut response_headers, &path, status); set_vary_on_accept_encoding(&mut response_headers); - let asset_csp = build_asset_csp( - &state.config.csp, - &RuntimeCspSources { - static_cdn_endpoint: state.config.static_cdn_endpoint.clone(), - media_endpoint: None, - s3_public_endpoint: None, - s3_uploads_bucket: None, - branding_image_origins: Vec::new(), - }, - ); - if let Ok(value) = HeaderValue::from_str(&asset_csp) { - response_headers.insert(header::CONTENT_SECURITY_POLICY, value); - } + response_headers.insert(header::CONTENT_SECURITY_POLICY, state.csp.asset_header()); response_headers.remove("content-security-policy-report-only"); - let body = Body::from_stream(upstream_response.bytes_stream()); + let body = Body::from_stream(upstream_asset_body(upstream_response, upstream_slot)); let mut response = Response::new(body); *response.status_mut() = status; *response.headers_mut() = response_headers; response } +fn upstream_asset_body( + mut upstream_response: reqwest::Response, + upstream_slot: OwnedSemaphorePermit, +) -> ReaderStream { + let (writer, reader) = tokio::io::duplex(UPSTREAM_ASSET_PUMP_BUFFER_BYTES); + tokio::spawn(async move { + let _upstream_slot = upstream_slot; + let mut writer = writer; + loop { + match upstream_response.chunk().await { + Ok(Some(chunk)) => { + if writer.write_all(&chunk).await.is_err() { + return; + } + } + Ok(None) => break, + Err(err) => { + tracing::warn!(%err, "upstream asset body ended early"); + return; + } + } + } + let _ = writer.shutdown().await; + }); + ReaderStream::new(reader) +} + pub(super) async fn serve_local_asset( + budgets: &AppProxyBudgets, static_dir: &str, relative_path: &str, request_headers: &HeaderMap, + csp_asset_header: HeaderValue, ) -> Response { + let Ok(_read_slot) = budgets.local_read_slots.try_acquire() else { + return super::capacity_refused_response(); + }; + let file_path = FsPath::new(static_dir).join(relative_path); let resolved = match tokio::fs::canonicalize(&file_path).await { @@ -179,7 +220,12 @@ pub(super) async fn serve_local_asset( && if_none_match_matches(request_headers, entity_tag) { let mut response = StatusCode::NOT_MODIFIED.into_response(); - set_local_asset_headers(response.headers_mut(), relative_path, Some(entity_tag)); + set_local_asset_headers( + response.headers_mut(), + relative_path, + Some(entity_tag), + &csp_asset_header, + ); return response; } @@ -205,7 +251,12 @@ pub(super) async fn serve_local_asset( HeaderValue::from_static(content_encoding), ); } - set_local_asset_headers(response.headers_mut(), relative_path, entity_tag.as_deref()); + set_local_asset_headers( + response.headers_mut(), + relative_path, + entity_tag.as_deref(), + &csp_asset_header, + ); response } @@ -269,11 +320,17 @@ fn is_zero_quality(parameter: &str) -> bool { .is_ok_and(|quality| quality <= 0.0) } -fn set_local_asset_headers(headers: &mut HeaderMap, relative_path: &str, entity_tag: Option<&str>) { +fn set_local_asset_headers( + headers: &mut HeaderMap, + relative_path: &str, + entity_tag: Option<&str>, + csp_asset_header: &HeaderValue, +) { headers.insert( header::CACHE_CONTROL, HeaderValue::from_static(asset_cache_control(relative_path)), ); + headers.insert(header::CONTENT_SECURITY_POLICY, csp_asset_header.clone()); set_vary_on_accept_encoding(headers); if is_font_mime(guess_mime(relative_path)) { headers.insert( @@ -403,7 +460,9 @@ mod tests { fn upstream_backed_state(cdn_endpoint: &str) -> AppState { let mut config = AppProxyConfig::from_env(); - config.static_cdn_endpoint = Some(cdn_endpoint.to_owned()); + config.static_cdn_endpoint = Some( + crate::config::HttpEndpoint::parse("TEST_STATIC_CDN_ENDPOINT", cdn_endpoint).unwrap(), + ); state_from_config(config) } @@ -415,8 +474,13 @@ mod tests { } fn state_from_config(config: AppProxyConfig) -> AppState { + let csp = Arc::new( + crate::csp::CompiledCspPolicy::from_config(&config) + .expect("the test configuration must compile to a valid CSP"), + ); AppState { config: Arc::new(config), + csp, http_client: build_http_client().unwrap(), discovery_cache: Arc::new(DiscoveryCache::new()), geoip: Arc::new(GeoipResolver::from_config(&GeoipConfig { @@ -429,6 +493,7 @@ mod tests { })), invite_meta: Arc::new(OnceLock::new()), index_html: None, + budgets: crate::state::AppProxyBudgets::default(), } } @@ -656,6 +721,18 @@ mod tests { } } + fn budgets() -> AppProxyBudgets { + AppProxyBudgets::default() + } + + fn test_asset_csp() -> HeaderValue { + let mut config = AppProxyConfig::from_env(); + config.static_cdn_endpoint = None; + crate::csp::CompiledCspPolicy::from_config(&config) + .expect("the test configuration must compile to a valid CSP") + .asset_header() + } + fn entity_tag_of(response: &Response) -> Option { response .headers() @@ -676,9 +753,11 @@ mod tests { let fixture = LocalAssetDir::with_asset("0018072843a46dc4.woff2", b"wOF2stub"); let first = serve_local_asset( + &budgets(), fixture.dir(), "assets/0018072843a46dc4.woff2", &HeaderMap::new(), + test_asset_csp(), ) .await; assert_eq!(cors_origin_of(&first), Some(CORS_ALLOW_ANY_VALUE)); @@ -689,8 +768,14 @@ mod tests { header::IF_NONE_MATCH, HeaderValue::from_str(&entity_tag).unwrap(), ); - let second = - serve_local_asset(fixture.dir(), "assets/0018072843a46dc4.woff2", &conditional).await; + let second = serve_local_asset( + &budgets(), + fixture.dir(), + "assets/0018072843a46dc4.woff2", + &conditional, + test_asset_csp(), + ) + .await; assert_eq!(second.status(), StatusCode::NOT_MODIFIED); assert_eq!( @@ -705,9 +790,11 @@ mod tests { let fixture = LocalAssetDir::with_asset("356aaade04a117b1.js", b"console.log(1)"); let response = serve_local_asset( + &budgets(), fixture.dir(), "assets/356aaade04a117b1.js", &HeaderMap::new(), + test_asset_csp(), ) .await; @@ -728,7 +815,14 @@ mod tests { let mut resumed = HeaderMap::new(); resumed.insert(header::RANGE, HeaderValue::from_static("bytes=10-")); - let response = serve_local_asset(fixture.dir(), "assets/fluxer-setup.exe", &resumed).await; + let response = serve_local_asset( + &budgets(), + fixture.dir(), + "assets/fluxer-setup.exe", + &resumed, + test_asset_csp(), + ) + .await; assert_eq!( response.status(), @@ -756,9 +850,11 @@ mod tests { let fixture = LocalAssetDir::with_asset("f00dcafe12345678.css", b"body{}"); let first = serve_local_asset( + &budgets(), fixture.dir(), "assets/f00dcafe12345678.css", &HeaderMap::new(), + test_asset_csp(), ) .await; let entity_tag = entity_tag_of(&first).expect("first response carries a validator"); @@ -768,8 +864,14 @@ mod tests { header::IF_NONE_MATCH, HeaderValue::from_str(&entity_tag).unwrap(), ); - let second = - serve_local_asset(fixture.dir(), "assets/f00dcafe12345678.css", &conditional).await; + let second = serve_local_asset( + &budgets(), + fixture.dir(), + "assets/f00dcafe12345678.css", + &conditional, + test_asset_csp(), + ) + .await; assert_eq!(second.status(), StatusCode::NOT_MODIFIED); assert_eq!(entity_tag_of(&second).as_deref(), Some(entity_tag.as_str())); @@ -788,8 +890,14 @@ mod tests { header::IF_NONE_MATCH, HeaderValue::from_static("\"stale-from-a-previous-build\""), ); - let response = - serve_local_asset(fixture.dir(), "assets/voice_engine_bg.wasm", &conditional).await; + let response = serve_local_asset( + &budgets(), + fixture.dir(), + "assets/voice_engine_bg.wasm", + &conditional, + test_asset_csp(), + ) + .await; assert_eq!(response.status(), StatusCode::OK); assert_eq!( @@ -807,9 +915,11 @@ mod tests { let fixture = LocalAssetDir::with_asset("2d715e4730758083.worker.js", b"self.onmessage=0"); let response = serve_local_asset( + &budgets(), fixture.dir(), "assets/2d715e4730758083.worker.js", &HeaderMap::new(), + test_asset_csp(), ) .await; @@ -859,9 +969,11 @@ mod tests { .and_sibling("356aaade04a117b1.js.br", b"brotli-bytes"); let response = serve_local_asset( + &budgets(), fixture.dir(), "assets/356aaade04a117b1.js", &accept_encoding("gzip, deflate, br, zstd"), + test_asset_csp(), ) .await; @@ -887,9 +999,11 @@ mod tests { let fixture = LocalAssetDir::with_asset("469e0b8f10c496a1.css", b"body{color:red}"); let response = serve_local_asset( + &budgets(), fixture.dir(), "assets/469e0b8f10c496a1.css", &accept_encoding("gzip, deflate, br"), + test_asset_csp(), ) .await; @@ -906,18 +1020,22 @@ mod tests { .and_sibling("488b87159423ca35.js.gz", b"gzip-bytes"); let gzip_only = serve_local_asset( + &budgets(), fixture.dir(), "assets/488b87159423ca35.js", &accept_encoding("gzip, deflate"), + test_asset_csp(), ) .await; assert_eq!(content_encoding_of(&gzip_only), Some("gzip")); assert_eq!(body_bytes(gzip_only).await, b"gzip-bytes"); let identity = serve_local_asset( + &budgets(), fixture.dir(), "assets/488b87159423ca35.js", &HeaderMap::new(), + test_asset_csp(), ) .await; assert_eq!( @@ -934,9 +1052,11 @@ mod tests { .and_sibling("2d715e4730758083.worker.js.br", b"brotli-bytes"); let response = serve_local_asset( + &budgets(), fixture.dir(), "assets/2d715e4730758083.worker.js", &accept_encoding("br;q=0, gzip"), + test_asset_csp(), ) .await; @@ -950,17 +1070,21 @@ mod tests { .and_sibling("f00dcafe12345678.css.br", b"brotli-bytes-are-longer"); let brotli = serve_local_asset( + &budgets(), fixture.dir(), "assets/f00dcafe12345678.css", &accept_encoding("br"), + test_asset_csp(), ) .await; let brotli_tag = entity_tag_of(&brotli).expect("the brotli variant carries a validator"); let identity = serve_local_asset( + &budgets(), fixture.dir(), "assets/f00dcafe12345678.css", &HeaderMap::new(), + test_asset_csp(), ) .await; let identity_tag = entity_tag_of(&identity).expect("the raw file carries a validator"); @@ -975,8 +1099,14 @@ mod tests { header::IF_NONE_MATCH, HeaderValue::from_str(&brotli_tag).unwrap(), ); - let revalidated = - serve_local_asset(fixture.dir(), "assets/f00dcafe12345678.css", &conditional).await; + let revalidated = serve_local_asset( + &budgets(), + fixture.dir(), + "assets/f00dcafe12345678.css", + &conditional, + test_asset_csp(), + ) + .await; assert_eq!(revalidated.status(), StatusCode::NOT_MODIFIED); assert!(varies_on_accept_encoding(&revalidated)); } @@ -988,8 +1118,14 @@ mod tests { let mut ranged = accept_encoding("br"); ranged.insert(header::RANGE, HeaderValue::from_static("bytes=4-6")); - let response = - serve_local_asset(fixture.dir(), "assets/356aaade04a117b1.js", &ranged).await; + let response = serve_local_asset( + &budgets(), + fixture.dir(), + "assets/356aaade04a117b1.js", + &ranged, + test_asset_csp(), + ) + .await; assert_eq!(response.status(), StatusCode::PARTIAL_CONTENT); assert_eq!(content_encoding_of(&response), Some("br")); @@ -1305,4 +1441,84 @@ mod tests { Some("application/octet-stream") ); } + + #[tokio::test] + async fn a_local_asset_is_refused_once_the_read_slots_are_gone() { + let fixture = LocalAssetDir::with_asset("356aaade04a117b1.js", b"console.log(1)"); + let budgets = AppProxyBudgets::default(); + let held = budgets + .local_read_slots + .clone() + .try_acquire_many_owned( + u32::try_from(crate::state::LOCAL_FILE_READS_IN_FLIGHT_MAX).unwrap(), + ) + .expect("a fresh budget holds every local read slot"); + + let refused = serve_local_asset( + &budgets, + fixture.dir(), + "assets/356aaade04a117b1.js", + &HeaderMap::new(), + test_asset_csp(), + ) + .await; + assert_eq!(refused.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + refused + .headers() + .get(header::CACHE_CONTROL) + .and_then(|value| value.to_str().ok()), + Some("no-store"), + "a cached refusal would pin the outage for every later reader" + ); + + drop(held); + let served = serve_local_asset( + &budgets, + fixture.dir(), + "assets/356aaade04a117b1.js", + &HeaderMap::new(), + test_asset_csp(), + ) + .await; + assert_eq!(served.status(), StatusCode::OK); + } + + #[tokio::test] + async fn an_open_local_asset_response_never_holds_a_read_slot() { + let fixture = LocalAssetDir::with_asset("356aaade04a117b1.js", b"console.log(1)"); + let budgets = AppProxyBudgets::default(); + + let mut open = Vec::with_capacity(crate::state::LOCAL_FILE_READS_IN_FLIGHT_MAX + 1); + for _ in 0..=crate::state::LOCAL_FILE_READS_IN_FLIGHT_MAX { + open.push( + serve_local_asset( + &budgets, + fixture.dir(), + "assets/356aaade04a117b1.js", + &HeaderMap::new(), + test_asset_csp(), + ) + .await, + ); + } + + let refused = open + .iter() + .filter(|response| response.status() != StatusCode::OK) + .count(); + assert_eq!( + refused, 0, + "{refused} readers were turned away while earlier responses were still open" + ); + + let body = axum::body::to_bytes(open.pop().unwrap().into_body(), usize::MAX) + .await + .unwrap(); + assert_eq!( + body.as_ref(), + b"console.log(1)", + "a response served past the read slot count carried the wrong bytes" + ); + } } diff --git a/fluxer_app_proxy/src/routes/health.rs b/fluxer_app_proxy/src/routes/health.rs index 19a2f108c..12667509a 100644 --- a/fluxer_app_proxy/src/routes/health.rs +++ b/fluxer_app_proxy/src/routes/health.rs @@ -81,8 +81,13 @@ mod tests { let mut config = AppProxyConfig::from_env(); config.invite_meta_enabled = invite_meta_enabled; config.discovery_upstream_url = spawn_discovery_origin().await; + let csp = Arc::new( + crate::csp::CompiledCspPolicy::from_config(&config) + .expect("the test configuration must compile to a valid CSP"), + ); AppState { config: Arc::new(config), + csp, http_client: build_http_client().unwrap(), discovery_cache: Arc::new(DiscoveryCache::new()), geoip: Arc::new(GeoipResolver::from_config(&GeoipConfig { @@ -95,6 +100,7 @@ mod tests { })), invite_meta: Arc::new(OnceLock::new()), index_html: None, + budgets: crate::state::AppProxyBudgets::default(), } } diff --git a/fluxer_app_proxy/src/routes/mod.rs b/fluxer_app_proxy/src/routes/mod.rs index af11ea4a8..29e971eed 100644 --- a/fluxer_app_proxy/src/routes/mod.rs +++ b/fluxer_app_proxy/src/routes/mod.rs @@ -14,7 +14,7 @@ use axum::{ extract::Request, http::{HeaderName, HeaderValue, header}, middleware::{Next, from_fn, from_fn_with_state}, - response::Response, + response::{IntoResponse, Response}, routing::get, }; use rand::RngExt; @@ -127,6 +127,14 @@ fn generate_request_id() -> String { hex::encode(bytes) } +pub(super) fn capacity_refused_response() -> Response { + let mut response = axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); + let headers = response.headers_mut(); + headers.insert(header::RETRY_AFTER, HeaderValue::from_static("1")); + headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + response +} + fn set_static_header(headers: &mut axum::http::HeaderMap, name: HeaderName, value: &'static str) { headers .entry(name) diff --git a/fluxer_app_proxy/src/routes/spa_index.rs b/fluxer_app_proxy/src/routes/spa_index.rs index d10419707..c1fc9a046 100644 --- a/fluxer_app_proxy/src/routes/spa_index.rs +++ b/fluxer_app_proxy/src/routes/spa_index.rs @@ -1,23 +1,29 @@ // SPDX-License-Identifier: AGPL-3.0-or-later use crate::bootstrap::{build_bootstrap_script, inject_bootstrap}; -use crate::csp::{RuntimeCspSources, build_csp, generate_nonce, http_origin}; -use crate::discovery_cache::DiscoveryResponse; +use crate::config::HttpEndpoint; +use crate::csp::{RuntimeCspSources, generate_nonce}; +use crate::discovery_cache::{DiscoveryResponse, discovery_endpoint}; use crate::geoip::build_geoip_response; use crate::invite_meta::{ InviteMetaEndpoints, InvitePageMeta, inject_invite_meta, invite_code_from_path, }; -use crate::state::AppState; +use crate::state::{ + AppProxyBudgets, AppState, MAX_RENDERED_SPA_INDEX_BYTES, MAX_SPA_INDEX_BYTES, + SPA_DOCUMENT_RENDER_RESERVATION_BYTES, read_bounded_text_file, +}; use crate::time_freeze::{ load_time_freeze_config_for_request, should_serve_frozen, time_freeze_debug_header, }; use axum::{ + body::{Body, Bytes}, extract::{Request, State}, http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header}, response::{IntoResponse, Response}, }; use std::path::Path; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::{OwnedSemaphorePermit, TryAcquireError}; use super::assets_proxy::serve_local_asset; use super::file_stream::stream_file; @@ -36,6 +42,7 @@ pub async fn spa_catch_all( if let Some(cache_control) = static_root_file_cache_control(request_path) { return serve_static_file( + &state.budgets, &state.config.static_dir, request_path, cache_control, @@ -45,9 +52,11 @@ pub async fn spa_catch_all( } if is_static_asset_path(request_path) { return serve_local_asset( + &state.budgets, &state.config.static_dir, request_path.trim_start_matches('/'), &headers, + state.csp.asset_header(), ) .await; } @@ -83,11 +92,16 @@ fn is_static_asset_path(request_path: &str) -> bool { } async fn serve_static_file( + budgets: &AppProxyBudgets, static_dir: &str, request_path: &str, cache_control: &'static str, request_headers: &HeaderMap, ) -> Response { + let Ok(_read_slot) = budgets.local_read_slots.try_acquire() else { + return super::capacity_refused_response(); + }; + let file_path = Path::new(static_dir).join(request_path.trim_start_matches('/')); let resolved = match tokio::fs::canonicalize(&file_path).await { @@ -149,35 +163,40 @@ async fn serve_spa_index(state: &AppState, headers: &HeaderMap, request_path: &s let invite_meta = resolve_invite_meta(state, request_path, &runtime_csp_sources).await; let static_cdn_endpoint = runtime_csp_sources .static_cdn_endpoint - .as_deref() - .unwrap_or(""); - let media_endpoint = runtime_csp_sources.media_endpoint.as_deref().unwrap_or(""); - let csp = build_csp(&state.config.csp, &nonce, &runtime_csp_sources); + .as_ref() + .map_or("", HttpEndpoint::as_str); + let media_endpoint = runtime_csp_sources + .media_endpoint + .as_ref() + .map_or("", HttpEndpoint::as_str); + let csp = state.csp.spa_header(&nonce, &runtime_csp_sources); let geoip = build_geoip_response(state.geoip.lookup(headers)); let script_tag = build_bootstrap_script(&state.config, &discovery, &geoip, &nonce); - if let Some(snapshot) = should_serve_frozen(&time_freeze) { - let frozen_html = String::from_utf8_lossy(&snapshot.index_html); - let dev_buster = should_bust_dev_assets.then(current_dev_asset_cache_buster); - let html = render_spa_document( - &frozen_html, - &nonce, - &script_tag, - static_cdn_endpoint, - media_endpoint, - invite_meta.as_ref(), - dev_buster.as_deref(), - ); - return build_spa_response(html, &csp, debug_header.as_deref(), should_bust_dev_assets); - } + let raw_html = if let Some(snapshot) = should_serve_frozen(&time_freeze) { + String::from_utf8_lossy(&snapshot.index_html).into_owned() + } else { + match load_spa_index_html(state).await { + Ok(content) => content, + Err(response) => return response, + } + }; - let raw_html = match load_spa_index_html(state).await { - Ok(content) => content, - Err(response) => return response, + let mut document_budget = match state + .budgets + .spa_document_memory + .clone() + .try_acquire_many_owned(SPA_DOCUMENT_RENDER_RESERVATION_BYTES) + { + Ok(permit) => permit, + Err(TryAcquireError::NoPermits) => return super::capacity_refused_response(), + Err(TryAcquireError::Closed) => { + panic!("SPA document memory budget semaphore closed unexpectedly") + } }; let dev_buster = should_bust_dev_assets.then(current_dev_asset_cache_buster); - let html = render_spa_document( + let html = match render_spa_document( &raw_html, &nonce, &script_tag, @@ -185,8 +204,60 @@ async fn serve_spa_index(state: &AppState, headers: &HeaderMap, request_path: &s media_endpoint, invite_meta.as_ref(), dev_buster.as_deref(), - ); - build_spa_response(html, &csp, debug_header.as_deref(), should_bust_dev_assets) + ) { + Ok(html) => html, + Err(error) => { + tracing::error!(%error, "failed to render SPA document within its size limit"); + return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + } + }; + let html = html.into_boxed_str(); + let retained_bytes = u32::try_from(html.len()).expect("bounded SPA document size must fit u32"); + let released_bytes = SPA_DOCUMENT_RENDER_RESERVATION_BYTES + .checked_sub(retained_bytes) + .expect("rendered SPA document must fit its memory reservation"); + if released_bytes > 0 { + let released_permits = + usize::try_from(released_bytes).expect("SPA document permit count must fit usize"); + drop( + document_budget + .split(released_permits) + .expect("SPA document memory reservation must contain its unused permits"), + ); + } + build_spa_response( + html, + csp, + debug_header.as_deref(), + should_bust_dev_assets, + document_budget, + ) +} + +#[derive(Debug)] +struct SpaDocumentSizeLimitError { + attempted_bytes: usize, +} + +impl std::fmt::Display for SpaDocumentSizeLimitError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "rendered SPA document would be {} bytes, exceeding the {MAX_RENDERED_SPA_INDEX_BYTES} byte limit", + self.attempted_bytes + ) + } +} + +impl std::error::Error for SpaDocumentSizeLimitError {} + +fn bounded_document(document: String) -> Result { + if document.len() > MAX_RENDERED_SPA_INDEX_BYTES { + return Err(SpaDocumentSizeLimitError { + attempted_bytes: document.len(), + }); + } + Ok(document) } fn render_spa_document( @@ -197,16 +268,21 @@ fn render_spa_document( media_endpoint: &str, invite_meta: Option<&InvitePageMeta>, dev_asset_cache_buster: Option<&str>, -) -> String { - let mut document = - inject_bootstrap(html, nonce, script_tag, static_cdn_endpoint, media_endpoint); +) -> Result { + let mut document = bounded_document(inject_bootstrap( + html, + nonce, + script_tag, + static_cdn_endpoint, + media_endpoint, + ))?; if let Some(meta) = invite_meta { - document = inject_invite_meta(&document, meta); + document = bounded_document(inject_invite_meta(&document, meta))?; } if let Some(buster) = dev_asset_cache_buster { - document = append_dev_asset_cache_buster(&document, buster); + document = bounded_document(append_dev_asset_cache_buster(&document, buster))?; } - document + Ok(document) } async fn refresh_discovery_for_spa(state: &AppState) -> Option { @@ -224,8 +300,14 @@ async fn resolve_invite_meta( let code = invite_code_from_path(request_path)?; let resolver = state.invite_meta.get()?; let endpoints = InviteMetaEndpoints { - media_endpoint: runtime_csp_sources.media_endpoint.clone(), - static_cdn_endpoint: runtime_csp_sources.static_cdn_endpoint.clone(), + media_endpoint: runtime_csp_sources + .media_endpoint + .as_ref() + .map(|endpoint| endpoint.as_str().to_owned()), + static_cdn_endpoint: runtime_csp_sources + .static_cdn_endpoint + .as_ref() + .map(|endpoint| endpoint.as_str().to_owned()), }; match resolver.resolve(code, &endpoints).await { @@ -243,7 +325,7 @@ fn build_runtime_csp_sources(state: &AppState, discovery: &DiscoveryResponse) -> .or_else(|| state.config.static_cdn_endpoint.clone()), media_endpoint: discovery_endpoint(discovery, "media"), s3_public_endpoint: state.config.s3_public_endpoint.clone(), - s3_uploads_bucket: Some(state.config.s3_uploads_bucket.clone()), + s3_uploads_endpoint: state.config.s3_uploads_endpoint.clone(), branding_image_origins: branding_image_origins(discovery), } } @@ -256,7 +338,7 @@ const BRANDING_IMAGE_KEYS: &[&str] = &[ "favicon_url", ]; -fn branding_image_origins(discovery: &DiscoveryResponse) -> Vec { +fn branding_image_origins(discovery: &DiscoveryResponse) -> Vec { let Some(branding) = discovery .data .get("app_public") @@ -264,39 +346,39 @@ fn branding_image_origins(discovery: &DiscoveryResponse) -> Vec { else { return Vec::new(); }; - let mut origins: Vec = Vec::new(); + let mut origins: Vec = Vec::new(); for key in BRANDING_IMAGE_KEYS { - let Some(origin) = branding + let Some(raw) = branding .get(*key) .and_then(|value| value.as_str()) - .and_then(http_origin) + .map(str::trim) + .filter(|value| !value.is_empty()) else { continue; }; - if !origins.contains(&origin) { + let origin = match HttpEndpoint::parse(key, raw) { + Ok(origin) => origin, + Err(error) => { + tracing::warn!(%error, "ignoring invalid branding image origin"); + continue; + } + }; + if !origins + .iter() + .any(|existing| existing.csp_origin() == origin.csp_origin()) + { origins.push(origin); } } origins } -fn discovery_endpoint(discovery: &DiscoveryResponse, key: &str) -> Option { - discovery - .data - .get("endpoints") - .and_then(|endpoints| endpoints.get(key)) - .and_then(|value| value.as_str()) - .map(str::trim) - .filter(|value| !value.is_empty()) - .map(ToOwned::to_owned) -} - #[allow(clippy::result_large_err)] async fn load_spa_index_html(state: &AppState) -> Result { if let Some(index_upstream_url) = &state.config.index_upstream_url { let response = state .http_client - .get(index_upstream_url) + .get(index_upstream_url.as_url().clone()) .timeout(Duration::from_secs(10)) .send() .await @@ -309,8 +391,31 @@ async fn load_spa_index_html(state: &AppState) -> Result { tracing::error!(url = %index_upstream_url, %status, "upstream index.html returned non-success status"); return Err(StatusCode::BAD_GATEWAY.into_response()); } - return response.text().await.map_err(|err| { - tracing::error!(url = %index_upstream_url, %err, "failed to read upstream index.html body"); + if response + .content_length() + .is_some_and(|length| length > MAX_SPA_INDEX_BYTES as u64) + { + tracing::error!(url = %index_upstream_url, "upstream index.html exceeds the size limit"); + return Err(StatusCode::BAD_GATEWAY.into_response()); + } + let mut response = response; + let mut bytes: Vec = Vec::new(); + loop { + let chunk = response.chunk().await.map_err(|err| { + tracing::error!(url = %index_upstream_url, %err, "failed to read upstream index.html body"); + StatusCode::BAD_GATEWAY.into_response() + })?; + let Some(chunk) = chunk else { + break; + }; + if chunk.len() > MAX_SPA_INDEX_BYTES - bytes.len() { + tracing::error!(url = %index_upstream_url, "upstream index.html exceeds the size limit"); + return Err(StatusCode::BAD_GATEWAY.into_response()); + } + bytes.extend_from_slice(&chunk); + } + return String::from_utf8(bytes).map_err(|err| { + tracing::error!(url = %index_upstream_url, %err, "upstream index.html is not valid UTF-8"); StatusCode::BAD_GATEWAY.into_response() }); } @@ -319,25 +424,44 @@ async fn load_spa_index_html(state: &AppState) -> Result { return Ok(cached.to_string()); } + let Ok(_read_slot) = state.budgets.local_read_slots.try_acquire() else { + return Err(super::capacity_refused_response()); + }; let index_path = Path::new(&state.config.static_dir).join("index.html"); - tokio::fs::read_to_string(&index_path).await.map_err(|err| { - tracing::error!(path = ?index_path, %err, "failed to read index.html"); - StatusCode::INTERNAL_SERVER_ERROR.into_response() - }) + read_bounded_text_file(&index_path, MAX_SPA_INDEX_BYTES) + .await + .map_err(|err| { + tracing::error!(path = ?index_path, %err, "failed to read index.html"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + }) +} + +struct SpaDocumentBody { + html: Box, + _budget: OwnedSemaphorePermit, +} + +impl AsRef<[u8]> for SpaDocumentBody { + fn as_ref(&self) -> &[u8] { + self.html.as_bytes() + } } fn build_spa_response( - html: String, - csp: &str, + html: Box, + csp: HeaderValue, time_freeze_header: Option<&str>, dev_no_store: bool, + document_budget: OwnedSemaphorePermit, ) -> Response { - let mut response = html.into_response(); + let body = Bytes::from_owner(SpaDocumentBody { + html, + _budget: document_budget, + }); + let mut response = Response::new(Body::from(body)); let headers = response.headers_mut(); - if let Ok(v) = HeaderValue::from_str(csp) { - headers.insert(header::CONTENT_SECURITY_POLICY, v); - } + headers.insert(header::CONTENT_SECURITY_POLICY, csp); headers.insert( header::CONTENT_TYPE, HeaderValue::from_static("text/html; charset=utf-8"), @@ -609,7 +733,8 @@ mod tests { "", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!(!rendered.contains("{{CSP_NONCE_PLACEHOLDER}}")); assert!(rendered.contains(r#"nonce="reqnonce""#)); @@ -627,7 +752,8 @@ mod tests { "", Some(&meta), None, - ); + ) + .expect("test SPA document must render within its size limit"); let without_meta = render_spa_document( SHELL_WITH_A_NONCE_HOLE, "reqnonce", @@ -636,7 +762,8 @@ mod tests { "", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!(with_meta.contains("Join Sample Space")); assert!(with_meta.contains("og:title")); @@ -654,7 +781,8 @@ mod tests { "", None, Some("9911"), - ); + ) + .expect("test SPA document must render within its size limit"); let untouched = render_spa_document( SHELL_WITH_A_NONCE_HOLE, "reqnonce", @@ -663,7 +791,8 @@ mod tests { "", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!(busted.contains(r#"src="/assets/app.js?_=9911""#)); assert!(untouched.contains(r#"src="/assets/app.js""#)); @@ -688,7 +817,8 @@ mod tests { "", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!( !served.contains("{{CSP_NONCE_PLACEHOLDER}}"), @@ -719,7 +849,8 @@ mod tests { "https://media.example.test", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!( rendered @@ -752,7 +883,8 @@ mod tests { "https://media.example.test/", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!( distinct.contains(r#""#), "the media argument never reached the media preconnect" @@ -768,7 +900,8 @@ mod tests { "https://cdn.example.test", None, None, - ); + ) + .expect("test SPA document must render within its size limit"); assert!( shared.contains(r#""#), "the static preconnects must survive a media endpoint that collapses onto them" @@ -844,13 +977,24 @@ mod tests { let mut config = AppProxyConfig::from_env(); config.release_channel = channel; config.time_freeze_enabled = true; - config.index_upstream_url = index_upstream_url; - config.static_cdn_endpoint = static_cdn_fallback.map(ToOwned::to_owned); + config.index_upstream_url = index_upstream_url.map(|url| { + crate::config::HttpUrl::parse("TEST_INDEX_UPSTREAM_URL", &url) + .expect("test index upstream URL must be a valid HTTP URL") + }); + config.static_cdn_endpoint = static_cdn_fallback.map(|endpoint| { + HttpEndpoint::parse("TEST_STATIC_CDN_ENDPOINT", endpoint) + .expect("test static CDN endpoint must be a valid HTTP endpoint") + }); config.trust_client_ip_header = false; config.discovery_upstream_url = discovery_upstream_url; + let csp = Arc::new( + crate::csp::CompiledCspPolicy::from_config(&config) + .expect("the test configuration must compile to a valid CSP"), + ); AppState { config: Arc::new(config), + csp, http_client: reqwest::Client::new(), discovery_cache: Arc::new(DiscoveryCache::new()), geoip: Arc::new(GeoipResolver::from_config(&GeoipConfig { @@ -863,6 +1007,7 @@ mod tests { })), invite_meta: Arc::new(OnceLock::new()), index_html: cached_shell.map(Arc::from), + budgets: crate::state::AppProxyBudgets::default(), } } @@ -1111,8 +1256,14 @@ mod tests { "an application route was mistaken for a static root file" ); - let response = - serve_static_file(static_dir, "/robots.txt", policy, &HeaderMap::new()).await; + let response = serve_static_file( + &AppProxyBudgets::default(), + static_dir, + "/robots.txt", + policy, + &HeaderMap::new(), + ) + .await; assert_eq!(response.status(), StatusCode::OK); let cache_control = response .headers() diff --git a/fluxer_app_proxy/src/routes/spa_static.rs b/fluxer_app_proxy/src/routes/spa_static.rs index 01a7ba870..22f611ece 100644 --- a/fluxer_app_proxy/src/routes/spa_static.rs +++ b/fluxer_app_proxy/src/routes/spa_static.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: AGPL-3.0-or-later -use crate::discovery_cache::DiscoveryResponse; -use crate::state::AppState; +use crate::config::HttpEndpoint; +use crate::discovery_cache::discovery_endpoint; +use crate::state::{AppState, MAX_STATIC_TEXT_FILE_BYTES, read_bounded_file}; use crate::time_freeze::{ TimeFreezeConfig, describe_decision, load_time_freeze_config_for_request, time_freeze_debug_header, @@ -50,8 +51,7 @@ pub async fn version_json(State(state): State, headers: HeaderMap) -> return resp; } - let mut result = - serve_static_text_file(&state.config.static_dir, "version.json", "application/json"); + let mut result = serve_static_text_file(&state, "version.json", "application/json").await; if result.status() == StatusCode::NOT_FOUND && !state.config.build_version.is_empty() { let body = serde_json::json!({ "version": state.config.build_version }); @@ -68,21 +68,23 @@ pub async fn version_json(State(state): State, headers: HeaderMap) -> pub async fn manifest_json(State(state): State) -> Response { let static_cdn_endpoint = runtime_static_cdn_endpoint(&state).await; serve_static_text_file_with_cdn( - &state.config.static_dir, + &state, "manifest.json", "application/manifest+json", - static_cdn_endpoint.as_deref(), + static_cdn_endpoint.as_ref(), ) + .await } pub async fn browserconfig_xml(State(state): State) -> Response { let static_cdn_endpoint = runtime_static_cdn_endpoint(&state).await; serve_static_text_file_with_cdn( - &state.config.static_dir, + &state, "browserconfig.xml", "application/xml; charset=utf-8", - static_cdn_endpoint.as_deref(), + static_cdn_endpoint.as_ref(), ) + .await } pub async fn service_worker(State(state): State, headers: HeaderMap) -> Response { @@ -96,17 +98,14 @@ pub async fn service_worker(State(state): State, headers: HeaderMap) - if let Some(resp) = frozen { return resp; } - let mut result = serve_static_text_file( - &state.config.static_dir, - "sw.js", - "application/javascript; charset=utf-8", - ); + let mut result = + serve_static_text_file(&state, "sw.js", "application/javascript; charset=utf-8").await; set_time_freeze_header(&mut result, debug_header.as_deref()); result } pub async fn service_worker_map(State(state): State) -> Response { - serve_static_text_file(&state.config.static_dir, "sw.js.map", "application/json") + serve_static_text_file(&state, "sw.js.map", "application/json").await } fn set_time_freeze_header(response: &mut Response, value: Option<&str>) { @@ -125,7 +124,7 @@ fn set_time_freeze_header(response: &mut Response, value: Option<&str>) { let _ = (response, value); } -async fn runtime_static_cdn_endpoint(state: &AppState) -> Option { +async fn runtime_static_cdn_endpoint(state: &AppState) -> Option { if let Some(discovery) = state.discovery_cache.get().await && let Some(endpoint) = discovery_endpoint(&discovery, "static_cdn") { @@ -135,34 +134,28 @@ async fn runtime_static_cdn_endpoint(state: &AppState) -> Option { state.config.static_cdn_endpoint.clone() } -fn discovery_endpoint(discovery: &DiscoveryResponse, key: &str) -> Option { - discovery - .data - .get("endpoints") - .and_then(|endpoints| endpoints.get(key)) - .and_then(|value| value.as_str()) - .map(str::trim) - .filter(|value| !value.is_empty()) - .map(ToOwned::to_owned) +async fn serve_static_text_file(state: &AppState, filename: &str, content_type: &str) -> Response { + serve_static_text_file_with_cdn(state, filename, content_type, None).await } -fn serve_static_text_file(static_dir: &str, filename: &str, content_type: &str) -> Response { - serve_static_text_file_with_cdn(static_dir, filename, content_type, None) -} - -fn serve_static_text_file_with_cdn( - static_dir: &str, +async fn serve_static_text_file_with_cdn( + state: &AppState, filename: &str, content_type: &str, - static_cdn_endpoint: Option<&str>, + static_cdn_endpoint: Option<&HttpEndpoint>, ) -> Response { + let static_dir = state.config.static_dir.as_str(); let file_path = Path::new(static_dir).join(filename); - let resolved = match file_path.canonicalize() { + let Ok(_read_slot) = state.budgets.local_read_slots.try_acquire() else { + return super::capacity_refused_response(); + }; + + let resolved = match tokio::fs::canonicalize(&file_path).await { Ok(p) => p, Err(_) => return StatusCode::NOT_FOUND.into_response(), }; - let base = match Path::new(static_dir).canonicalize() { + let base = match tokio::fs::canonicalize(static_dir).await { Ok(p) => p, Err(_) => return StatusCode::NOT_FOUND.into_response(), }; @@ -170,12 +163,16 @@ fn serve_static_text_file_with_cdn( return StatusCode::NOT_FOUND.into_response(); } - let content = match std::fs::read(&resolved) { + let content = match read_bounded_file(&resolved, MAX_STATIC_TEXT_FILE_BYTES).await { Ok(bytes) => bytes, - Err(_) => return StatusCode::NOT_FOUND.into_response(), + Err(error) if error.is_not_found() => return StatusCode::NOT_FOUND.into_response(), + Err(error) => { + tracing::error!(file = filename, %error, "refusing to serve static text file"); + return StatusCode::NOT_FOUND.into_response(); + } }; - let replacement = static_cdn_endpoint.unwrap_or("").trim_end_matches('/'); + let replacement = static_cdn_endpoint.map_or("", HttpEndpoint::as_str); let body: axum::body::Body = match std::str::from_utf8(&content) { Ok(text) => text .replace("{{STATIC_CDN_ENDPOINT}}", replacement) diff --git a/fluxer_app_proxy/src/state.rs b/fluxer_app_proxy/src/state.rs index aa0e29caf..1e94f0d13 100644 --- a/fluxer_app_proxy/src/state.rs +++ b/fluxer_app_proxy/src/state.rs @@ -1,20 +1,136 @@ // SPDX-License-Identifier: AGPL-3.0-or-later use crate::config::AppProxyConfig; +use crate::csp::CompiledCspPolicy; use crate::discovery_cache::DiscoveryCache; use crate::invite_meta::InviteMetaResolver; use fluxer_common::geoip::GeoipResolver; use std::sync::{Arc, OnceLock}; use std::time::Duration; +use tokio::io::AsyncReadExt; +use tokio::sync::Semaphore; + +pub const MAX_SPA_INDEX_BYTES: usize = 4 * 1024 * 1024; +pub const MAX_RENDERED_SPA_INDEX_BYTES: usize = 8 * 1024 * 1024; +pub const MAX_STATIC_TEXT_FILE_BYTES: usize = 4 * 1024 * 1024; +pub const SPA_DOCUMENT_MEMORY_BUDGET_BYTES: usize = 160 * 1024 * 1024; +pub const SPA_DOCUMENT_RENDER_RESERVATION_BYTES: u32 = 40 * 1024 * 1024; +pub const UPSTREAM_ASSET_RESPONSES_IN_FLIGHT_MAX: usize = 32; +pub const LOCAL_FILE_READS_IN_FLIGHT_MAX: usize = 256; + +const _: () = assert!( + MAX_RENDERED_SPA_INDEX_BYTES <= SPA_DOCUMENT_RENDER_RESERVATION_BYTES as usize, + "a rendered SPA document must fit inside the memory reserved to render it" +); + +#[derive(Clone)] +pub struct AppProxyBudgets { + pub spa_document_memory: Arc, + pub upstream_asset_slots: Arc, + pub local_read_slots: Arc, +} + +impl AppProxyBudgets { + pub fn new() -> Self { + Self { + spa_document_memory: Arc::new(Semaphore::new(SPA_DOCUMENT_MEMORY_BUDGET_BYTES)), + upstream_asset_slots: Arc::new(Semaphore::new(UPSTREAM_ASSET_RESPONSES_IN_FLIGHT_MAX)), + local_read_slots: Arc::new(Semaphore::new(LOCAL_FILE_READS_IN_FLIGHT_MAX)), + } + } +} + +impl Default for AppProxyBudgets { + fn default() -> Self { + Self::new() + } +} #[derive(Clone)] pub struct AppState { pub config: Arc, + pub csp: Arc, pub http_client: reqwest::Client, pub discovery_cache: Arc, pub geoip: Arc, pub invite_meta: Arc>, pub index_html: Option>, + pub budgets: AppProxyBudgets, +} + +#[derive(Debug)] +pub enum BoundedFileReadError { + TooLarge { actual: u64, maximum: usize }, + Io(std::io::Error), +} + +impl std::fmt::Display for BoundedFileReadError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::TooLarge { actual, maximum } => write!( + formatter, + "file is {actual} bytes, exceeding the {maximum} byte limit" + ), + Self::Io(source) => source.fmt(formatter), + } + } +} + +impl std::error::Error for BoundedFileReadError {} + +impl BoundedFileReadError { + pub fn is_not_found(&self) -> bool { + matches!(self, Self::Io(source) if source.kind() == std::io::ErrorKind::NotFound) + } +} + +pub async fn read_bounded_file( + path: &std::path::Path, + max_bytes: usize, +) -> Result, BoundedFileReadError> { + let file = tokio::fs::File::open(path) + .await + .map_err(BoundedFileReadError::Io)?; + let metadata = file.metadata().await.map_err(BoundedFileReadError::Io)?; + if !metadata.is_file() { + return Err(BoundedFileReadError::Io(std::io::Error::new( + std::io::ErrorKind::NotFound, + "not a regular file", + ))); + } + let declared_length = metadata.len(); + if declared_length > max_bytes as u64 { + return Err(BoundedFileReadError::TooLarge { + actual: declared_length, + maximum: max_bytes, + }); + } + let expected_bytes = + usize::try_from(declared_length).expect("a length within a usize limit must fit usize"); + let mut bytes = Vec::with_capacity(expected_bytes); + tokio::io::AsyncReadExt::read_to_end(&mut file.take(declared_length + 1), &mut bytes) + .await + .map_err(BoundedFileReadError::Io)?; + if bytes.len() > expected_bytes { + return Err(BoundedFileReadError::TooLarge { + actual: bytes.len() as u64, + maximum: max_bytes, + }); + } + Ok(bytes) +} + +pub async fn read_bounded_text_file( + path: &std::path::Path, + max_bytes: usize, +) -> Result { + let bytes = read_bounded_file(path, max_bytes).await?; + String::from_utf8(bytes).map_err(|error| { + BoundedFileReadError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + error.utf8_error(), + )) + }) } pub fn build_http_client() -> reqwest::Result {