fluxerapp/fluxer
1
0
Fork 0
mirror of https://github.com/fluxerapp/fluxer.git synced 2026-09-02 21:04:06 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(app-proxy): validate config, csp sources, cap resource use (#2354)

Browse Source
This commit is contained in:
Hampus
2026-09-02 17:25:10 +02:00
committed by GitHub
parent 9f33177eab
commit 73a2345c26
10 changed files with 1313 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fluxer_app_proxy/src/config.rs
+393 -29
View File
@@ -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<Self, InvalidAppProxyEnvironmentError> {
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<Self, InvalidAppProxyEnvironmentError> {
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<Self, InvalidAppProxyEnvironmentError> {
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<Self, InvalidAppProxyEnvironmentError> {
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<String> {
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<Self, InvalidAppProxyEnvironmentError> {
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<String>) -> Option<HttpUrl> {
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<String>) -> Option<HttpEndpoint> {
let value = value?;
match HttpEndpoint::parse(name, &value) {
Ok(endpoint) => Some(endpoint),
Err(error) => {
warn_invalid(error);
None
}
}
}
fn parse_env_or_warn<T: std::str::FromStr>(name: &str, raw: &str, default: T) -> T {
raw.parse::<T>().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<String>,
pub static_cdn_endpoint: Option<String>,
pub s3_public_endpoint: Option<String>,
pub s3_uploads_bucket: String,
pub index_upstream_url: Option<HttpUrl>,
pub static_cdn_endpoint: Option<HttpEndpoint>,
pub s3_public_endpoint: Option<HttpEndpoint>,
pub s3_uploads_endpoint: Option<HttpEndpoint>,
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<Vec<String>>,
pub extra_connect_src: Option<Vec<String>>,
pub extra_img_src: Option<Vec<String>>,
pub extra_media_src: Option<Vec<String>>,
pub extra_font_src: Option<Vec<String>>,
pub extra_script_src: Option<Vec<String>>,
pub extra_style_src: Option<Vec<String>>,
pub extra_frame_src: Option<Vec<String>>,
pub extra_worker_src: Option<Vec<String>>,
pub extra_manifest_src: Option<Vec<String>>,
pub report_uri: Option<String>,
pub extra_default_src: Vec<CspSource>,
pub extra_connect_src: Vec<CspSource>,
pub extra_img_src: Vec<CspSource>,
pub extra_media_src: Vec<CspSource>,
pub extra_font_src: Vec<CspSource>,
pub extra_script_src: Vec<CspSource>,
pub extra_style_src: Vec<CspSource>,
pub extra_frame_src: Vec<CspSource>,
pub extra_worker_src: Vec<CspSource>,
pub extra_manifest_src: Vec<CspSource>,
pub report_uri: Option<CspReportUri>,
}
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<Vec<String>> {
let sources: Vec<String> = cfg::read_env(name, "")
fn read_csp_sources(name: &'static str) -> Vec<CspSource> {
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<CspReportUri> {
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()
);
}
fluxer_app_proxy/src/csp.rs
+240 -95
View File
@@ -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<String>,
pub media_endpoint: Option<String>,
pub s3_public_endpoint: Option<String>,
pub s3_uploads_bucket: Option<String>,
pub branding_image_origins: Vec<String>,
pub static_cdn_endpoint: Option<HttpEndpoint>,
pub media_endpoint: Option<HttpEndpoint>,
pub s3_public_endpoint: Option<HttpEndpoint>,
pub s3_uploads_endpoint: Option<HttpEndpoint>,
pub branding_image_origins: Vec<HttpEndpoint>,
}
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, CspCompileError> {
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<Self, CspCompileError> {
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<String> {
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<String>, endpoint: Option<&str>) {
fn push_endpoint_source(target: &mut Vec<String>, 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<String>, 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<String> {
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<String>, 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<String>, 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());
}
}
fluxer_app_proxy/src/discovery_cache.rs
+21
View File
@@ -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<HttpEndpoint> {
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<Option<DiscoveryResponse>>,
cold_start_attempt: Mutex<Option<Instant>>,
fluxer_app_proxy/src/main.rs
+12 -2
View File
@@ -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::<str>::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);
fluxer_app_proxy/src/routes/assets_proxy.rs
+254 -38
View File
@@ -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<DuplexStream> {
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<String> {
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"
);
}
}
fluxer_app_proxy/src/routes/health.rs
+6
View File
@@ -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(),
}
}
fluxer_app_proxy/src/routes/mod.rs
+9 -1
View File
@@ -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)
fluxer_app_proxy/src/routes/spa_index.rs
+229 -78
View File
@@ -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<String, SpaDocumentSizeLimitError> {
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<String, SpaDocumentSizeLimitError> {
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<DiscoveryResponse> {
@@ -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<String> {
fn branding_image_origins(discovery: &DiscoveryResponse) -> Vec<HttpEndpoint> {
let Some(branding) = discovery
.data
.get("app_public")
@@ -264,39 +346,39 @@ fn branding_image_origins(discovery: &DiscoveryResponse) -> Vec<String> {
else {
return Vec::new();
};
let mut origins: Vec<String> = Vec::new();
let mut origins: Vec<HttpEndpoint> = 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<String> {
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<String, Response> {
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<String, Response> {
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<u8> = 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<String, Response> {
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<str>,
_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<str>,
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#"<link rel="preconnect" href="https://media.example.test">"#),
"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#"<link rel="preconnect" href="https://cdn.example.test">"#),
"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()
fluxer_app_proxy/src/routes/spa_static.rs
+33 -36
View File
@@ -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<AppState>, 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<AppState>, headers: HeaderMap) ->
pub async fn manifest_json(State(state): State<AppState>) -> 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<AppState>) -> 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<AppState>, headers: HeaderMap) -> Response {
@@ -96,17 +98,14 @@ pub async fn service_worker(State(state): State<AppState>, 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<AppState>) -> 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<String> {
async fn runtime_static_cdn_endpoint(state: &AppState) -> Option<HttpEndpoint> {
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<String> {
state.config.static_cdn_endpoint.clone()
}
fn discovery_endpoint(discovery: &DiscoveryResponse, key: &str) -> Option<String> {
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)
fluxer_app_proxy/src/state.rs
+116
View File
@@ -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<Semaphore>,
pub upstream_asset_slots: Arc<Semaphore>,
pub local_read_slots: Arc<Semaphore>,
}
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<AppProxyConfig>,
pub csp: Arc<CompiledCspPolicy>,
pub http_client: reqwest::Client,
pub discovery_cache: Arc<DiscoveryCache>,
pub geoip: Arc<GeoipResolver>,
pub invite_meta: Arc<OnceLock<InviteMetaResolver>>,
pub index_html: Option<Arc<str>>,
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<Vec<u8>, 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<String, BoundedFileReadError> {
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<reqwest::Client> {