gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.27.3

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-07 22:22:54 +02:00
1 parent e84df445c6
commit e87a382a7e
109 files changed
+1092 -2925
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/acme/runtime_bootstrap.rs
+128
View File
@@ -0,0 +1,128 @@
async fn create_virtual_environment(
python: &Path,
staging: &Path,
system_pip: bool,
shutdown: &tokio_util::sync::CancellationToken,
) -> bool {
let mut create = Command::new(python);
create.arg("-m").arg("venv").arg(staging);
if command_success(create, Duration::from_secs(180), shutdown).await {
return true;
}
if !system_pip {
return false;
}
let _ = tokio::fs::remove_dir_all(staging).await;
// Fall back to an application-local virtualenv bootstrap.
let bootstrap = staging.with_extension("virtualenv-bootstrap");
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
let install_bootstrap = |break_system_packages: bool| {
let mut install = Command::new(python);
install.args([
"-m",
"pip",
"install",
"--disable-pip-version-check",
"--no-input",
"--no-cache-dir",
"--retries",
"3",
"--timeout",
"30",
]);
// Some distro pip builds enforce PEP 668 even for --target. Retrying with
// --break-system-packages is safe here because the destination is the
// private bootstrap directory; distribution site-packages are untouched.
if break_system_packages {
install.arg("--break-system-packages");
}
install.arg("--target").arg(&bootstrap).arg("virtualenv>=20,<21");
install
};
let installed = command_success(
install_bootstrap(false),
Duration::from_secs(300),
shutdown,
)
.await;
if !installed {
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
if !command_success(
install_bootstrap(true),
Duration::from_secs(300),
shutdown,
)
.await
{
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
return false;
}
}
let mut create = Command::new(python);
create
.arg("-c")
.arg("import runpy,sys; sys.path.insert(0,sys.argv[1]); sys.argv=['virtualenv',sys.argv[2]]; runpy.run_module('virtualenv',run_name='__main__')")
.arg(&bootstrap)
.arg(staging);
let created = command_success(create, Duration::from_secs(180), shutdown).await;
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
created
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum BuildFailure {
CreateEnvironment,
PreparePip,
InstallPackages,
Validate,
}
async fn build_managed_environment(
python: &Path,
target: &Path,
system_pip: bool,
shutdown: &tokio_util::sync::CancellationToken,
) -> Result<(), BuildFailure> {
let legacy_stage = target.with_extension("installing");
let _ = tokio::fs::remove_dir_all(&legacy_stage).await;
let _ = tokio::fs::remove_dir_all(target).await;
let _ = tokio::fs::remove_file(target).await;
if !create_virtual_environment(python, target, system_pip, shutdown).await {
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::CreateEnvironment);
}
let target_python = venv_python(target);
let mut pip_check = Command::new(&target_python);
pip_check.args(["-m", "pip", "--version"]);
if !command_success(pip_check, Duration::from_secs(30), shutdown).await {
let mut ensurepip = Command::new(&target_python);
ensurepip.args(["-m", "ensurepip", "--upgrade"]);
if !command_success(ensurepip, Duration::from_secs(120), shutdown).await {
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::PreparePip);
}
}
let mut pip = Command::new(&target_python);
pip.args([
"-m",
"pip",
"install",
"--disable-pip-version-check",
"--no-input",
"--no-cache-dir",
"--retries",
"3",
"--timeout",
"30",
]);
pip.args(CERTBOT_PACKAGES.iter().copied());
if !command_success(pip, Duration::from_secs(600), shutdown).await {
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::InstallPackages);
}
if !managed_environment_valid(target, shutdown).await {
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::Validate);
}
Ok(())
}
src/acme/runtime_env.rs
+38 -115
View File
@@ -60,7 +60,7 @@ impl Manager {
struct RuntimeLaunch {
python: PathBuf,
prepend_path: Option<PathBuf>,
managed_certbot_python: Option<PathBuf>,
managed_certbot_executable: Option<PathBuf>,
}
fn find_on_path(name: &str) -> Option<PathBuf> {
@@ -107,6 +107,15 @@ fn venv_python(directory: &Path) -> PathBuf {
venv_bin(directory).join("python")
}
#[cfg(windows)]
fn venv_certbot(directory: &Path) -> PathBuf {
venv_bin(directory).join("certbot.exe")
}
#[cfg(not(windows))]
fn venv_certbot(directory: &Path) -> PathBuf {
venv_bin(directory).join("certbot")
}
async fn command_success(
mut command: Command,
timeout: Duration,
@@ -145,9 +154,12 @@ async fn managed_environment_valid(
if !command_success(imports, Duration::from_secs(20), shutdown).await {
return false;
}
let certbot = venv_bin(directory).join("certbot");
let mut version = Command::new(&python);
version.arg("-I").arg(certbot).arg("--version");
let certbot = venv_certbot(directory);
if !certbot.is_file() {
return false;
}
let mut version = Command::new(&certbot);
version.arg("--version");
command_success(version, Duration::from_secs(20), shutdown).await
}
@@ -172,107 +184,7 @@ async fn system_has_pip(python: &Path, shutdown: &tokio_util::sync::Cancellation
command_success(command, Duration::from_secs(20), shutdown).await
}
async fn create_virtual_environment(
python: &Path,
staging: &Path,
system_pip: bool,
shutdown: &tokio_util::sync::CancellationToken,
) -> bool {
let mut create = Command::new(python);
create.arg("-m").arg("venv").arg(staging);
if command_success(create, Duration::from_secs(180), shutdown).await {
return true;
}
if !system_pip {
return false;
}
let _ = tokio::fs::remove_dir_all(staging).await;
// Fall back to an application-local virtualenv bootstrap.
let bootstrap = staging.with_extension("virtualenv-bootstrap");
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
let mut install = Command::new(python);
install.args([
"-m",
"pip",
"install",
"--disable-pip-version-check",
"--no-input",
"--no-cache-dir",
"--retries",
"3",
"--timeout",
"30",
"--target",
]);
install.arg(&bootstrap).arg("virtualenv>=20,<21");
if !command_success(install, Duration::from_secs(300), shutdown).await {
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
return false;
}
let mut create = Command::new(python);
create
.arg("-c")
.arg("import runpy,sys; sys.path.insert(0,sys.argv[1]); sys.argv=['virtualenv',sys.argv[2]]; runpy.run_module('virtualenv',run_name='__main__')")
.arg(&bootstrap)
.arg(staging);
let created = command_success(create, Duration::from_secs(180), shutdown).await;
let _ = tokio::fs::remove_dir_all(&bootstrap).await;
created
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum BuildFailure {
Bootstrap,
Install,
}
async fn build_managed_environment(
python: &Path,
target: &Path,
system_pip: bool,
shutdown: &tokio_util::sync::CancellationToken,
) -> Result<(), BuildFailure> {
let legacy_stage = target.with_extension("installing");
let _ = tokio::fs::remove_dir_all(&legacy_stage).await;
let _ = tokio::fs::remove_dir_all(target).await;
let _ = tokio::fs::remove_file(target).await;
if !create_virtual_environment(python, target, system_pip, shutdown).await {
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::Bootstrap);
}
let target_python = venv_python(target);
let mut pip_check = Command::new(&target_python);
pip_check.args(["-m", "pip", "--version"]);
if !command_success(pip_check, Duration::from_secs(30), shutdown).await {
let mut ensurepip = Command::new(&target_python);
ensurepip.args(["-m", "ensurepip", "--upgrade"]);
if !command_success(ensurepip, Duration::from_secs(120), shutdown).await {
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::Bootstrap);
}
}
let mut pip = Command::new(&target_python);
pip.args([
"-m",
"pip",
"install",
"--disable-pip-version-check",
"--no-input",
"--no-cache-dir",
"--retries",
"3",
"--timeout",
"30",
]);
pip.args(CERTBOT_PACKAGES.iter().copied());
if !command_success(pip, Duration::from_secs(600), shutdown).await
|| !managed_environment_valid(target, shutdown).await
{
let _ = tokio::fs::remove_dir_all(target).await;
return Err(BuildFailure::Install);
}
Ok(())
}
include!("runtime_bootstrap.rs");
async fn ensure_runtime(state: &App) -> Result<RuntimeLaunch, String> {
let _permit = state
@@ -309,7 +221,7 @@ async fn ensure_runtime(state: &App) -> Result<RuntimeLaunch, String> {
return Ok(RuntimeLaunch {
python: venv_python(&managed),
prepend_path: Some(venv_bin(&managed)),
managed_certbot_python: Some(venv_python(&managed)),
managed_certbot_executable: Some(venv_certbot(&managed)),
});
}
let distro_certbot = find_on_path("certbot");
@@ -357,7 +269,7 @@ async fn ensure_runtime(state: &App) -> Result<RuntimeLaunch, String> {
return Ok(RuntimeLaunch {
python: venv_python(&managed),
prepend_path: Some(venv_bin(&managed)),
managed_certbot_python: Some(venv_python(&managed)),
managed_certbot_executable: Some(venv_certbot(&managed)),
});
}
if distro_certbot_available {
@@ -374,13 +286,23 @@ async fn ensure_runtime(state: &App) -> Result<RuntimeLaunch, String> {
return Ok(RuntimeLaunch {
python,
prepend_path: None,
managed_certbot_python: None,
managed_certbot_executable: None,
});
}
let message = if build == Err(BuildFailure::Bootstrap) && !system_pip {
"acme.runtime_pip_missing"
} else {
"acme.runtime_setup_failed"
tracing::warn!(
python = %python.display(),
managed = %managed.display(),
system_pip,
failure = ?build,
"managed ACME runtime setup failed; no system Certbot fallback available"
);
let message = match build {
Err(BuildFailure::CreateEnvironment) if !system_pip => "acme.runtime_pip_missing",
Err(BuildFailure::CreateEnvironment) => "acme.runtime_venv_failed",
Err(BuildFailure::PreparePip) => "acme.runtime_venv_pip_failed",
Err(BuildFailure::InstallPackages) => "acme.runtime_certbot_install_failed",
Err(BuildFailure::Validate) => "acme.runtime_certbot_validation_failed",
Ok(()) => "acme.runtime_setup_failed",
};
state.acme.set_runtime(RuntimeStatus {
status: "unavailable",
@@ -428,19 +350,20 @@ async fn worker(state: &App, directory: PathBuf, raw: Zeroizing<String>) -> Resu
command
.arg("-I")
.arg("-c")
.arg(include_str!("../../scripts/acme_provider.py"))
.arg(include_str!("../../scripts/runtime/acme_provider.py"))
.arg("--cert-dir")
.arg(&directory)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.kill_on_drop(true);
command.env_remove("RUSTPROXY_CERTBOT_EXECUTABLE");
command.env_remove("RUSTPROXY_CERTBOT_PYTHON");
if let Some(prepend) = launch.prepend_path.as_deref() {
command.env("PATH", merged_path(prepend)?);
}
if let Some(python) = launch.managed_certbot_python.as_deref() {
command.env("RUSTPROXY_CERTBOT_PYTHON", python);
if let Some(certbot) = launch.managed_certbot_executable.as_deref() {
command.env("RUSTPROXY_CERTBOT_EXECUTABLE", certbot);
}
#[cfg(unix)]
{
src/admin/documentation/documents.rs
+1
View File
@@ -409,6 +409,7 @@ pub(super) static READER_DOCUMENTS: &[(&str, &str, &str)] = &[
"/docs/V0.25.1_AUDIT.md"
)),
),
reader_doc!("docs/V0.27.3_IMPLEMENTATION.md", "technical"),
reader_doc!("docs/V0.27.2_IMPLEMENTATION.md", "technical"),
reader_doc!("docs/V0.27.1_IMPLEMENTATION.md", "historical"),
reader_doc!("docs/V0.27.0_IMPLEMENTATION.md", "technical"),
src/admin/openapi.json
+1 -1
View File
@@ -2,7 +2,7 @@
"openapi": "3.0.3",
"info": {
"title": "RustProxy Administration API",
"version": "0.27.2",
"version": "0.27.3",
"description": "Administration API for the active panel origin. Sign in through the panel; the same-origin HttpOnly cookie authenticates all private operations. Mutations also require the exact Origin, X-RustProxy-Request: 1 and a session-bound X-CSRF-Token. Loopback HTTP is the default; Settings can enable HTTPS before using a non-loopback administration bind. This document contains no live configuration or credentials. Object schemas marked extensible are summaries: full configuration validation is authoritative in Rust."
},
"servers": [
src/admin/web/index.html
+2 -2
View File
@@ -2823,7 +2823,7 @@
</section>
</div>
</section>
<footer class="page-footer"><span>RustProxy <b>0.27.2</b></span><span id="footer-status">API - /
<footer class="page-footer"><span>RustProxy <b>0.27.3</b></span><span id="footer-status">API - /
dropped: 0</span></footer>
</div>
</main>
@@ -3567,7 +3567,7 @@ max_execution_time = 30</code></pre>
<article class="card about-card">
<div class="about-product">
<span aria-hidden="true" class="brand-mark">r<span>/</span>p</span>
<div><span class="small-label">RustProxy</span><strong>0.27.2</strong></div>
<div><span class="small-label">RustProxy</span><strong>0.27.3</strong></div>
</div>
<dl class="about-meta">
<div>
src/backup/archive.rs
+8 -1
View File
@@ -462,6 +462,13 @@ fn hash_bytes(bytes: &[u8]) -> String {
fn excluded_directory(name: &str) -> bool {
matches!(
name,
"cache" | "logs" | "tmp" | "temp" | "work" | "acme-http01"
"cache"
| "logs"
| "tmp"
| "temp"
| "work"
| "acme-http01"
| "acme-python"
| "acme_python"
)
}
src/backup/tests.rs
+7
View File
@@ -29,6 +29,13 @@ mod tests {
assert!(validate_backup_version("999.0.0").is_err());
}
#[test]
fn backup_excludes_rebuildable_managed_python_runtime() {
assert!(excluded_directory("acme-python"));
assert!(excluded_directory("acme_python"));
assert!(!excluded_directory("certs"));
}
#[cfg(unix)]
#[tokio::test]
async fn live_backup_round_trips_internal_certbot_symlink() {
src/cache/cold/demotion.rs
Vendored
+1 -2
View File
@@ -243,8 +243,7 @@ impl Cache {
entry.ttl,
entry.stale_while_revalidate,
entry.stale_if_error,
entry.origin_etag.is_some()
|| entry.origin_last_modified.is_some(),
entry.has_validator,
)
});
if should_remove {
src/cache/cold/startup.rs
Vendored
+15 -18
View File
@@ -99,19 +99,19 @@ impl Cache {
if path.file_stem().and_then(|v| v.to_str()) != Some(file.key.as_str()) {
continue;
}
let Some(headers) = pairs_to_headers(&file.headers) else {
if pairs_to_headers(&file.headers).is_none() {
continue;
};
}
if body_path.parent() != Some(shard.as_path()) {
continue;
}
let Ok(body_meta) = std::fs::metadata(&body_path) else {
continue;
};
if body_meta.len() != file.body_bytes as u64
|| file.body_sha256.len() != 64
|| !file.body_sha256.bytes().all(|b| b.is_ascii_hexdigit())
{
let Some(body_sha256) = parse_key_hex(&file.body_sha256) else {
continue;
};
if body_meta.len() != file.body_bytes as u64 {
continue;
}
let origin_etag = file
@@ -138,8 +138,7 @@ impl Cache {
let _ = std::fs::remove_file(&body_path);
continue;
}
loaded.push(ColdEntry {
key,
loaded.push((key, ColdEntry {
base_key,
vary,
revision,
@@ -163,11 +162,10 @@ impl Cache {
} else {
system_time_ms(file.created_at_ms)
},
headers,
body_path,
meta_path: path,
body_bytes: file.body_bytes,
body_sha256: file.body_sha256,
body_sha256,
disk_bytes: body_meta.len().saturating_add(bytes.len() as u64),
stored_at,
last_access: if file.last_access_ms == 0 {
@@ -182,8 +180,7 @@ impl Cache {
stale_if_error: file.stale_if_error,
must_revalidate: file.must_revalidate,
proxy_revalidate: file.proxy_revalidate,
origin_etag,
origin_last_modified,
has_validator: origin_etag.is_some() || origin_last_modified.is_some(),
hits: file.hits,
transformed: file.transformed,
sequence: 0,
@@ -191,12 +188,12 @@ impl Cache {
skip_compression: file.skip_compression,
requires_revalidation: file.requires_revalidation,
pinned_reads: 0,
});
}));
}
}
loaded.sort_by_key(|entry| entry.last_access);
loaded.sort_by_key(|(_, entry)| entry.last_access);
let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner());
for mut entry in loaded {
for (key, mut entry) in loaded {
for key in incompatible_cold_variants(&cold, &entry.base_key, &entry.vary) {
if let Some(old) = cold.remove(&key) {
let _ = std::fs::remove_file(old.meta_path);
@@ -218,9 +215,9 @@ impl Cache {
entry.sequence = cold.next_sequence();
entry.generation = entry.sequence;
cold.bytes = cold.bytes.saturating_add(entry.disk_bytes);
cold.lru.insert(entry.sequence, entry.key);
cold.index_variant(entry.base_key, entry.key);
cold.entries.insert(entry.key, entry);
cold.lru.insert(entry.sequence, key);
cold.index_variant(entry.base_key, key);
cold.entries.insert(key, entry);
}
}
}
src/cache/cold/worker.rs
Vendored
+6 -6
View File
@@ -114,6 +114,7 @@ impl Cache {
.saturating_add(draft.entry.received.elapsed())
.as_secs();
let resident_age_at_store = hot_resident_age(&draft.entry).as_secs();
let body_sha256 = body_sha256_bytes(&draft.entry.body);
let file = ColdFile {
format: COLD_FORMAT_VERSION,
key: key.clone(),
@@ -137,7 +138,7 @@ impl Cache {
headers,
body_file: body_name,
body_bytes: draft.entry.body.len(),
body_sha256: body_sha256(&draft.entry.body),
body_sha256: key_hex(&body_sha256),
stored_at_ms: now_ms(),
age_at_store_seconds: age_at_store,
resident_age_at_store_seconds: resident_age_at_store,
@@ -236,8 +237,9 @@ impl Cache {
delete.extend([meta_path.clone(), body_path.clone()]);
} else {
let sequence = cold.next_sequence();
let has_validator =
draft.entry.origin_etag.is_some() || draft.entry.origin_last_modified.is_some();
let entry = ColdEntry {
key: draft.key,
base_key: draft.entry.base_key,
vary: draft.entry.vary,
revision: draft.revision,
@@ -249,11 +251,10 @@ impl Cache {
target: draft.entry.target,
tags: draft.entry.tags,
created_at: draft.entry.created_at,
headers: draft.entry.headers,
body_path: body_path.clone(),
meta_path: meta_path.clone(),
body_bytes: draft.entry.body.len(),
body_sha256: file.body_sha256,
body_sha256,
disk_bytes,
stored_at: SystemTime::now(),
age_at_store_seconds: age_at_store,
@@ -263,8 +264,7 @@ impl Cache {
stale_if_error: draft.entry.stale_if_error,
must_revalidate: draft.entry.must_revalidate,
proxy_revalidate: draft.entry.proxy_revalidate,
origin_etag: draft.entry.origin_etag,
origin_last_modified: draft.entry.origin_last_modified,
has_validator,
last_access: SystemTime::now(),
hits: draft.entry.hits,
transformed: draft.entry.transformed,
src/cache/lookup/hot.rs
Vendored
+14 -7
View File
@@ -150,6 +150,13 @@ impl Cache {
key: storage_key,
generation: entry.generation,
};
let Some(metadata) = read_cold_metadata(&storage_key, &entry).await else {
self.counters.cold_errors.fetch_add(1, Ordering::Relaxed);
if let Some(entry) = self.take_cold(&storage_key) {
self.delete_cold_entry(entry);
}
return None;
};
if entry.body_bytes > settings.max_object_bytes {
self.counters
.memory_pressure_rejections
@@ -203,7 +210,7 @@ impl Cache {
let hits = entry.hits.saturating_add(1);
let cost = hot_entry_cost(
body_allocated,
&entry.headers,
&metadata.headers,
&entry.vary,
&entry.scope,
&entry.scheme,
@@ -212,8 +219,8 @@ impl Cache {
&entry.path,
&entry.target,
&entry.tags,
entry.origin_etag.as_ref(),
entry.origin_last_modified.as_ref(),
metadata.origin_etag.as_ref(),
metadata.origin_last_modified.as_ref(),
);
let resident_age_seconds = cold_resident_age(&entry);
let hot_entry = Entry {
@@ -227,7 +234,7 @@ impl Cache {
target: entry.target.clone(),
tags: entry.tags.clone(),
created_at: entry.created_at,
headers: entry.headers.clone(),
headers: metadata.headers.clone(),
body: body.clone(),
body_allocated,
received: Instant::now(),
@@ -240,8 +247,8 @@ impl Cache {
stale_if_error: entry.stale_if_error,
must_revalidate: entry.must_revalidate,
proxy_revalidate: entry.proxy_revalidate,
origin_etag: entry.origin_etag.clone(),
origin_last_modified: entry.origin_last_modified.clone(),
origin_etag: metadata.origin_etag.clone(),
origin_last_modified: metadata.origin_last_modified.clone(),
cost,
sequence: 0,
hits,
@@ -253,7 +260,7 @@ impl Cache {
return None;
}
let mut response = Response::new(Body::from(body.clone()));
*response.headers_mut() = entry.headers;
*response.headers_mut() = metadata.headers;
if entry.transformed {
response
.extensions_mut()
src/cache/model.rs
Vendored
+1
View File
@@ -2,4 +2,5 @@ include!("model/settings.rs");
include!("model/entry.rs");
include!("model/cache.rs");
include!("model/maintenance.rs");
include!("model/cold_metadata.rs");
include!("model/views.rs");
src/cache/model/cache.rs
Vendored
+2 -5
View File
@@ -246,7 +246,6 @@ struct ColdFile {
#[derive(Clone)]
struct ColdEntry {
key: Key,
base_key: Key,
vary: Vec<String>,
revision: u64,
@@ -258,11 +257,10 @@ struct ColdEntry {
target: String,
tags: Vec<String>,
created_at: SystemTime,
headers: HeaderMap,
body_path: PathBuf,
meta_path: PathBuf,
body_bytes: usize,
body_sha256: String,
body_sha256: Key,
disk_bytes: u64,
stored_at: SystemTime,
last_access: SystemTime,
@@ -273,8 +271,7 @@ struct ColdEntry {
stale_if_error: u64,
must_revalidate: bool,
proxy_revalidate: bool,
origin_etag: Option<HeaderValue>,
origin_last_modified: Option<HeaderValue>,
has_validator: bool,
hits: u64,
transformed: bool,
sequence: u64,
src/cache/model/cold_metadata.rs
Vendored
+69
View File
@@ -0,0 +1,69 @@
struct ColdMetadata {
headers: HeaderMap,
origin_etag: Option<HeaderValue>,
origin_last_modified: Option<HeaderValue>,
}
async fn read_cold_metadata(storage_key: &Key, entry: &ColdEntry) -> Option<ColdMetadata> {
let bytes = tokio::fs::read(&entry.meta_path).await.ok()?;
let file: ColdFile = serde_json::from_slice(&bytes).ok()?;
let key = parse_key_hex(&file.key)?;
let base_key = parse_key_hex(&file.base_key)?;
let body_sha256 = parse_key_hex(&file.body_sha256)?;
let mut vary = file.vary.clone();
vary.sort();
vary.dedup();
let authority = if file.authority.is_empty() {
file.domain.as_str()
} else {
file.authority.as_str()
};
let target = if file.target.is_empty() {
file.path.as_str()
} else {
file.target.as_str()
};
if file.format != COLD_FORMAT_VERSION
|| key != *storage_key
|| base_key != entry.base_key
|| vary != entry.vary
|| file.revision != entry.revision
|| file.scope != entry.scope
|| file.scheme != entry.scheme
|| authority != entry.authority
|| file.domain != entry.domain
|| file.path != entry.path
|| target != entry.target
|| file.tags != entry.tags
|| file.body_bytes != entry.body_bytes
|| body_sha256 != entry.body_sha256
|| file.ttl != entry.ttl
|| file.stale_while_revalidate != entry.stale_while_revalidate
|| file.stale_if_error != entry.stale_if_error
|| file.must_revalidate != entry.must_revalidate
|| file.proxy_revalidate != entry.proxy_revalidate
|| file.transformed != entry.transformed
|| file.skip_compression != entry.skip_compression
|| file.requires_revalidation != entry.requires_revalidation
|| entry.body_path.file_name().and_then(|name| name.to_str()) != Some(file.body_file.as_str())
{
return None;
}
let headers = pairs_to_headers(&file.headers)?;
let origin_etag = file
.origin_etag
.as_deref()
.and_then(|value| HeaderValue::from_str(value).ok());
let origin_last_modified = file
.origin_last_modified
.as_deref()
.and_then(|value| HeaderValue::from_str(value).ok());
if entry.has_validator != (origin_etag.is_some() || origin_last_modified.is_some()) {
return None;
}
Some(ColdMetadata {
headers,
origin_etag,
origin_last_modified,
})
}
src/cache/model/maintenance.rs
Vendored
+4 -7
View File
@@ -404,18 +404,15 @@ fn incompatible_cold_variants(store: &ColdStore, base_key: &Key, vary: &[String]
.collect()
}
fn body_sha256(body: &[u8]) -> String {
fn body_sha256_bytes(body: &[u8]) -> Key {
let digest = Sha256::digest(body);
let mut out = String::with_capacity(64);
for byte in digest {
use std::fmt::Write as _;
let _ = write!(&mut out, "{byte:02x}");
}
let mut out = [0u8; 32];
out.copy_from_slice(&digest);
out
}
fn cold_body_matches(entry: &ColdEntry, body: &[u8]) -> bool {
body.len() == entry.body_bytes && body_sha256(body).as_str() == entry.body_sha256.as_str()
body.len() == entry.body_bytes && body_sha256_bytes(body) == entry.body_sha256
}
fn key_hex(key: &Key) -> String {
src/cache/tests/cold_tier_tests/part_02.rs
Vendored
+2 -5
View File
@@ -256,7 +256,6 @@ fn old_cold_read_guard_does_not_unpin_a_replacement_with_the_same_key() {
let cache = Cache::new(1);
let key = [7_u8; 32];
let template = ColdEntry {
key,
base_key: key,
vary: Vec::new(),
revision: 1,
@@ -268,11 +267,10 @@ fn old_cold_read_guard_does_not_unpin_a_replacement_with_the_same_key() {
target: "https://example.test/asset".to_owned(),
tags: Vec::new(),
created_at: SystemTime::now(),
headers: HeaderMap::new(),
body_path: PathBuf::from("old.body"),
meta_path: PathBuf::from("old.json"),
body_bytes: 1,
body_sha256: "00".repeat(32),
body_sha256: [0u8; 32],
disk_bytes: 2,
stored_at: SystemTime::now(),
last_access: SystemTime::now(),
@@ -283,8 +281,7 @@ fn old_cold_read_guard_does_not_unpin_a_replacement_with_the_same_key() {
stale_if_error: 0,
must_revalidate: false,
proxy_revalidate: false,
origin_etag: None,
origin_last_modified: None,
has_validator: false,
hits: 0,
transformed: false,
sequence: 1,
src/cache/tickets.rs
Vendored
+11 -4
View File
@@ -158,7 +158,7 @@ impl Cache {
entry.ttl,
entry.stale_while_revalidate,
entry.stale_if_error,
entry.origin_etag.is_some() || entry.origin_last_modified.is_some(),
entry.has_validator,
)
{
return None;
@@ -179,6 +179,13 @@ impl Cache {
key: storage_key,
generation: entry.generation,
};
let Some(metadata) = read_cold_metadata(&storage_key, &entry).await else {
self.counters.cold_errors.fetch_add(1, Ordering::Relaxed);
if let Some(entry) = self.take_cold(&storage_key) {
self.delete_cold_entry(entry);
}
return None;
};
let settings = self
.tier_settings
.lock()
@@ -219,7 +226,7 @@ impl Cache {
let age = cold_age(&entry);
let resident_age_at_lookup = Duration::from_secs(cold_resident_age(&entry));
Some(Revalidation {
headers: entry.headers,
headers: metadata.headers,
body,
body_allocated,
_transient_leases: extra_lease
@@ -237,8 +244,8 @@ impl Cache {
stale_if_error: entry.stale_if_error,
must_revalidate: entry.must_revalidate,
proxy_revalidate: entry.proxy_revalidate,
origin_etag: entry.origin_etag,
origin_last_modified: entry.origin_last_modified,
origin_etag: metadata.origin_etag,
origin_last_modified: metadata.origin_last_modified,
})
}
pub async fn revalidation_tiered(&self, ticket: &Ticket) -> Option<Revalidation> {
src/main/bootstrap.rs
+1 -1
View File
@@ -41,7 +41,7 @@ use zeroize::Zeroizing;
#[command(
name = "rust-proxy",
version,
about = "RustProxy - independent Rust reverse proxy and static Web Server 0.27.2"
about = "RustProxy - independent Rust reverse proxy and static Web Server 0.27.3"
)]
struct Cli {
#[arg(
src/main/tasks.rs
+1 -1
View File
@@ -239,7 +239,7 @@ async fn live_logs_command(dir: PathBuf, addr: SocketAddr) -> Result<()> {
}
});
let interactive = io::stdin().is_terminal() && io::stdout().is_terminal();
println!("RustProxy live logs 0.27.2 - all vhosts. Type 'help' for interactive filters; Ctrl-C or 'quit' exits.");
println!("RustProxy live logs 0.27.3 - all vhosts. Type 'help' for interactive filters; Ctrl-C or 'quit' exits.");
if interactive { print!("> "); io::stdout().flush()?; }
let mut filters = LiveLogFilters::default();
let mut history = std::collections::BTreeMap::<u64, serde_json::Value>::new();
src/proxy/forward/route_cache.rs
+4 -1
View File
@@ -162,7 +162,10 @@ let policy_reason = if admin_control_plane {
None
};
let cache_enabled = policy_reason.is_none();
let body_empty = if cache_enabled && handshake.is_none() && request.method() == Method::GET {
let body_empty = if cache_enabled
&& handshake.is_none()
&& (request.method() == Method::GET || request.method() == Method::HEAD)
{
match empty_for_cache(&mut request).await {
Ok(empty) => empty,
Err(_) => {
src/rate_limit.rs
+79 -27
View File
@@ -209,32 +209,76 @@ struct Limit {
window_seconds: u64,
burst: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct Key {
revision: u64,
vhost: String,
client: IpAddr,
}
#[derive(Debug)]
struct Bucket {
started: Instant,
count: u32,
window_seconds: u64,
}
#[derive(Debug, Default)]
struct LimiterState {
revision: Option<u64>,
// Store the vhost String once per vhost instead of once per client IP.
buckets: HashMap<String, HashMap<IpAddr, Bucket>>,
bucket_count: usize,
last_sweep: Option<Instant>,
}
impl LimiterState {
fn clear(&mut self) {
self.buckets.clear();
self.bucket_count = 0;
self.last_sweep = None;
}
fn sweep_expired(&mut self, now: Instant, force: bool) {
if !force
&& self
.last_sweep
.is_some_and(|last| now.duration_since(last) < Duration::from_secs(5))
{
return;
}
self.last_sweep = Some(now);
let mut removed = 0usize;
self.buckets.retain(|_, clients| {
let before = clients.len();
clients.retain(|_, bucket| {
now.duration_since(bucket.started) < Duration::from_secs(bucket.window_seconds)
});
let removed_here = before.saturating_sub(clients.len());
removed = removed.saturating_add(removed_here);
if removed_here > 0
&& !clients.is_empty()
&& clients.capacity() > clients.len().saturating_mul(2).max(64)
{
clients.shrink_to_fit();
}
!clients.is_empty()
});
self.bucket_count = self.bucket_count.saturating_sub(removed);
if removed > 0
&& self.buckets.capacity() > self.buckets.len().saturating_mul(2).max(16)
{
self.buckets.shrink_to_fit();
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Limited {
pub retry_after_seconds: u64,
}
#[derive(Debug, Default)]
pub struct Limiter {
buckets: Mutex<HashMap<Key, Bucket>>,
state: Mutex<LimiterState>,
}
impl Limiter {
pub fn new() -> Self {
Self::default()
}
pub fn reset(&self) {
lock(&self.buckets).clear();
lock(&self.state).clear();
}
pub fn maintenance(&self) {
lock(&self.state).sweep_expired(Instant::now(), true);
}
pub fn check(
&self,
@@ -247,30 +291,38 @@ impl Limiter {
let Some(limit) = settings.effective(country) else {
return Ok(());
};
let key = Key {
revision,
vhost: vhost.to_owned(),
client,
};
let mut buckets = lock(&self.buckets);
if !buckets.contains_key(&key) && buckets.len() >= MAX_BUCKETS {
buckets.retain(|key, bucket| {
key.revision == revision
&& bucket.started.elapsed()
< Duration::from_secs(bucket.window_seconds.saturating_mul(2).max(2))
});
if buckets.len() >= MAX_BUCKETS {
let now = Instant::now();
let mut state = lock(&self.state);
if state.revision != Some(revision) {
state.clear();
state.revision = Some(revision);
}
state.sweep_expired(now, false);
let new_client = !state
.buckets
.get(vhost)
.is_some_and(|clients| clients.contains_key(&client));
if new_client && state.bucket_count >= MAX_BUCKETS {
state.sweep_expired(now, true);
if state.bucket_count >= MAX_BUCKETS {
return Err(Limited {
retry_after_seconds: 1,
});
}
}
let now = Instant::now();
let bucket = buckets.entry(key).or_insert(Bucket {
started: now,
count: 0,
window_seconds: limit.window_seconds,
});
if new_client {
state.bucket_count = state.bucket_count.saturating_add(1);
}
let bucket = state
.buckets
.entry(vhost.to_owned())
.or_default()
.entry(client)
.or_insert(Bucket {
started: now,
count: 0,
window_seconds: limit.window_seconds,
});
if bucket.window_seconds != limit.window_seconds
|| now.duration_since(bucket.started) >= Duration::from_secs(limit.window_seconds)
{
src/rate_limit/tests.rs
+76
View File
@@ -33,4 +33,80 @@ mod tests {
assert!(limiter.check(1, "site", ip, Some("PL"), &settings).is_ok());
assert!(limiter.check(1, "site", ip, Some("PL"), &settings).is_err());
}
#[test]
fn expired_client_buckets_are_reclaimed_without_waiting_for_capacity_pressure() {
let settings = Settings {
enabled: true,
requests: 10,
window_seconds: 1,
burst: 0,
..Default::default()
};
let limiter = Limiter::new();
let first = "192.0.2.1".parse().unwrap();
let second = "192.0.2.2".parse().unwrap();
assert!(limiter.check(7, "site", first, None, &settings).is_ok());
{
let mut state = lock(&limiter.state);
let bucket = state.buckets.get_mut("site").unwrap().get_mut(&first).unwrap();
bucket.started = Instant::now()
.checked_sub(Duration::from_secs(2))
.unwrap();
state.last_sweep = None;
}
assert!(limiter.check(7, "site", second, None, &settings).is_ok());
let state = lock(&limiter.state);
assert_eq!(state.bucket_count, 1);
assert!(!state.buckets.get("site").unwrap().contains_key(&first));
assert!(state.buckets.get("site").unwrap().contains_key(&second));
}
#[test]
fn maintenance_reclaims_expired_clients_without_new_traffic() {
let settings = Settings {
enabled: true,
requests: 10,
window_seconds: 1,
burst: 0,
..Default::default()
};
let limiter = Limiter::new();
let ip = "192.0.2.10".parse().unwrap();
assert!(limiter.check(9, "idle-site", ip, None, &settings).is_ok());
{
let mut state = lock(&limiter.state);
let bucket = state
.buckets
.get_mut("idle-site")
.unwrap()
.get_mut(&ip)
.unwrap();
bucket.started = Instant::now()
.checked_sub(Duration::from_secs(2))
.unwrap();
}
limiter.maintenance();
let state = lock(&limiter.state);
assert_eq!(state.bucket_count, 0);
assert!(state.buckets.is_empty());
}
#[test]
fn vhost_name_is_stored_once_for_multiple_clients() {
let settings = Settings {
enabled: true,
requests: 10,
..Default::default()
};
let limiter = Limiter::new();
for ip in ["192.0.2.1", "192.0.2.2"] {
assert!(limiter
.check(1, "shared-vhost", ip.parse().unwrap(), None, &settings)
.is_ok());
}
let state = lock(&limiter.state);
assert_eq!(state.buckets.len(), 1);
assert_eq!(state.bucket_count, 2);
}
}
src/runtime/constructors.rs
+4
View File
@@ -212,6 +212,7 @@ impl App {
// cache fill paths. glibc may retain already-free arenas as RSS. Trim
// them periodically without evicting live cache objects or connections.
let trim_shutdown = app.shutdown.clone();
let trim_app = Arc::downgrade(&app);
app.tunnels.spawn(async move {
let mut tick = tokio::time::interval(Duration::from_secs(60));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -220,6 +221,9 @@ impl App {
tokio::select! {
_ = trim_shutdown.cancelled() => break,
_ = tick.tick() => {
if let Some(app) = trim_app.upgrade() {
app.rate_limiter.maintenance();
}
let _ = tokio::task::spawn_blocking(crate::diagnostics::trim_allocator).await;
}
}
src/web_server/fastcgi/protocol.rs
+1 -1
View File
@@ -159,7 +159,7 @@ pub(super) fn fastcgi_params(
if context.tls { "on" } else { "off" }.into(),
);
params.insert("GATEWAY_INTERFACE".into(), "CGI/1.1".into());
params.insert("SERVER_SOFTWARE".into(), "RustProxy/0.27.2".into());
params.insert("SERVER_SOFTWARE".into(), "RustProxy/0.27.3".into());
params.insert("REDIRECT_STATUS".into(), "200".into());
if let Some(value) = headers
.get(CONTENT_TYPE)
src/web_server/response/file.rs
+1 -1
View File
@@ -185,7 +185,7 @@ pub(super) fn directory_listing(path: &str, entries: &[DirectoryEntry]) -> Vec<u
"<a class=\"{class}\" href=\"{href}\"><span class=\"type\">{type_label}</span><span class=\"name\">{label}{slash}</span><span class=\"kind\">{kind}</span></a>"
));
}
html.push_str(r#"</div></section><footer class="page-footer"><span>RustProxy <b>0.27.2</b></span><span>Web Server · Directory listing</span></footer></main></body></html>"#);
html.push_str(r#"</div></section><footer class="page-footer"><span>RustProxy <b>0.27.3</b></span><span>Web Server · Directory listing</span></footer></main></body></html>"#);
html.into_bytes()
}
src/web_server/tests/part_02.rs
+1 -1
View File
@@ -202,7 +202,7 @@ async fn empty_index_can_go_directly_to_directory_listing() {
let body = std::str::from_utf8(&body).unwrap();
assert!(body.contains("file.txt"));
assert!(body.contains("DIRECTORY LISTING"));
assert!(body.contains("RustProxy <b>0.27.2</b>"));
assert!(body.contains("RustProxy <b>0.27.3</b>"));
}
#[tokio::test]