From 55f3d19ee0156107201ce7a7d363939f14e9c136 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Tue, 15 Sep 2026 20:37:41 +0300 Subject: [PATCH] Direct-since-epoch --- src/api/config_store/persistence.rs | 167 +++++++++++++++++- src/api/runtime_zero.rs | 2 +- src/daemon/pid_file.rs | 140 +++++++++++++-- src/maestro/listeners/bind.rs | 131 +++++++++++++- src/proxy/route_mode.rs | 28 +-- .../route_mode_coherence_adversarial_tests.rs | 39 ++-- src/proxy/tests/route_mode_security_tests.rs | 14 +- src/web/http/tests.rs | 33 +++- src/web/manager/credentials.rs | 2 +- 9 files changed, 483 insertions(+), 73 deletions(-) diff --git a/src/api/config_store/persistence.rs b/src/api/config_store/persistence.rs index 8893e9e..1b6ee39 100644 --- a/src/api/config_store/persistence.rs +++ b/src/api/config_store/persistence.rs @@ -1,7 +1,10 @@ use std::collections::BTreeMap; -use std::io::Write; +use std::io::{Read, Write}; use std::path::{Path, PathBuf}; +#[cfg(unix)] +use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; + use chrono::{DateTime, Utc}; use serde::Serialize; @@ -10,8 +13,8 @@ use crate::config::{ProxyConfig, RateLimitBps}; #[cfg(test)] use super::compute_revision; use super::{ - AccessSection, compute_snapshot_revision, load_candidate_snapshot, load_config_snapshot, - resolve_single_source_owner, toml_path_exists, + AccessSection, compute_snapshot_revision, compute_source_revision, load_candidate_snapshot, + load_config_snapshot, resolve_single_source_owner, toml_path_exists, }; use crate::api::model::ApiFailure; @@ -99,8 +102,22 @@ pub(in crate::api) async fn save_access_sections_to_disk( config_path: &Path, cfg: &ProxyConfig, sections: &[AccessSection], +) -> Result { + save_access_sections_to_disk_if_revision(config_path, cfg, sections, None).await +} + +/// Persists access tables only while the complete source graph remains unchanged. +pub(in crate::api) async fn save_access_sections_to_disk_if_revision( + config_path: &Path, + cfg: &ProxyConfig, + sections: &[AccessSection], + expected_revision: Option<&str>, ) -> Result { let loaded = load_config_snapshot(config_path, false).await?; + let loaded_revision = compute_snapshot_revision(&loaded); + if expected_revision.is_some_and(|expected| expected != loaded_revision) { + return Err(revision_conflict()); + } let mut applied = Vec::new(); for section in sections { if applied.contains(section) { @@ -117,7 +134,7 @@ pub(in crate::api) async fn save_access_sections_to_disk( }) }); if applied.is_empty() { - return Ok(compute_snapshot_revision(&loaded)); + return Ok(loaded_revision); } let targets = applied @@ -130,6 +147,7 @@ pub(in crate::api) async fn save_access_sections_to_disk( .get(&owner_path) .cloned() .ok_or_else(|| ApiFailure::internal("config source owner is missing from snapshot"))?; + let expected_owner_contents = owner_contents.clone(); for section in applied { let rendered = render_access_section(cfg, section)?; owner_contents = upsert_toml_table(&owner_contents, section.table_name(), &rendered); @@ -143,7 +161,14 @@ pub(in crate::api) async fn save_access_sections_to_disk( ) .await?; let revision = compute_snapshot_revision(&candidate); - write_atomic(owner_path, owner_contents).await?; + write_atomic_if_unchanged( + config_path.to_path_buf(), + loaded_revision, + owner_path, + expected_owner_contents, + owner_contents, + ) + .await?; Ok(revision) } @@ -378,15 +403,112 @@ pub(in crate::api) async fn write_atomic( path: PathBuf, contents: String, ) -> Result<(), ApiFailure> { - tokio::task::spawn_blocking(move || write_atomic_sync(&path, &contents)) + tokio::task::spawn_blocking(move || write_atomic_sync(&path, None, &contents)) .await .map_err(|e| ApiFailure::internal(format!("failed to join writer: {}", e)))? .map_err(|e| ApiFailure::internal(format!("failed to write config: {}", e))) } -fn write_atomic_sync(path: &Path, contents: &str) -> std::io::Result<()> { +/// Replaces one source only if both its graph revision and owner contents are unchanged. +pub(in crate::api) async fn write_atomic_if_unchanged( + config_path: PathBuf, + expected_revision: String, + path: PathBuf, + expected_contents: String, + contents: String, +) -> Result<(), ApiFailure> { + tokio::task::spawn_blocking(move || { + let graph = ProxyConfig::read_source_graph(&config_path) + .map_err(|error| AtomicWriteError::ReadGraph(error.to_string()))?; + if compute_source_revision(&graph) != expected_revision { + return Err(AtomicWriteError::Conflict); + } + write_atomic_sync(&path, Some(&expected_contents), &contents) + .map_err(AtomicWriteError::Io) + }) + .await + .map_err(|error| ApiFailure::internal(format!("failed to join writer: {error}")))? + .map_err(|error| match error { + AtomicWriteError::Conflict => revision_conflict(), + AtomicWriteError::ReadGraph(error) => { + ApiFailure::internal(format!("failed to verify config graph: {error}")) + } + AtomicWriteError::Io(error) => { + ApiFailure::internal(format!("failed to write config: {error}")) + } + }) +} + +enum AtomicWriteError { + Conflict, + ReadGraph(String), + Io(std::io::Error), +} + +struct ExistingTarget { + contents: String, + metadata: std::fs::Metadata, +} + +fn revision_conflict() -> ApiFailure { + ApiFailure::new( + hyper::StatusCode::CONFLICT, + "revision_conflict", + "Config revision changed before persistence", + ) +} + +fn open_existing_target(path: &Path) -> std::io::Result> { + let mut options = std::fs::OpenOptions::new(); + options.read(true); + #[cfg(unix)] + options.custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW); + let mut file = match options.open(path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error), + }; + let metadata = file.metadata()?; + if !metadata.is_file() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "config target must be a regular file", + )); + } + let mut contents = String::new(); + file.read_to_string(&mut contents)?; + Ok(Some(ExistingTarget { contents, metadata })) +} + +fn same_target(left: &std::fs::Metadata, right: &std::fs::Metadata) -> bool { + #[cfg(unix)] + { + left.dev() == right.dev() && left.ino() == right.ino() + } + #[cfg(not(unix))] + { + left.len() == right.len() && left.modified().ok() == right.modified().ok() + } +} + +fn write_atomic_sync( + path: &Path, + expected_contents: Option<&str>, + contents: &str, +) -> std::io::Result<()> { let parent = path.parent().unwrap_or_else(|| Path::new(".")); std::fs::create_dir_all(parent)?; + let existing = open_existing_target(path)?; + if expected_contents.is_some_and(|expected| { + existing + .as_ref() + .is_none_or(|target| target.contents != expected) + }) { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "config source changed before persistence", + )); + } let tmp_name = format!( ".{}.tmp-{}", @@ -401,9 +523,40 @@ fn write_atomic_sync(path: &Path, contents: &str) -> std::io::Result<()> { let mut file = std::fs::OpenOptions::new() .create_new(true) .write(true) + #[cfg(unix)] + .mode(0o600) .open(&tmp_path)?; + #[cfg(unix)] + if let Some(existing) = existing.as_ref() { + use nix::unistd::{Gid, Uid, fchown}; + + fchown( + &file, + Some(Uid::from_raw(existing.metadata.uid())), + Some(Gid::from_raw(existing.metadata.gid())), + ) + .map_err(|error| std::io::Error::from_raw_os_error(error as i32))?; + file.set_permissions(std::fs::Permissions::from_mode( + existing.metadata.mode() & 0o7777, + ))?; + } file.write_all(contents.as_bytes())?; file.sync_all()?; + let current = open_existing_target(path)?; + let target_unchanged = match (&existing, ¤t) { + (Some(expected), Some(current)) => { + same_target(&expected.metadata, ¤t.metadata) + && expected.contents == current.contents + } + (None, None) => true, + _ => false, + }; + if !target_unchanged { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "config target changed during persistence", + )); + } std::fs::rename(&tmp_path, path)?; if let Ok(dir) = std::fs::File::open(parent) { let _ = dir.sync_all(); diff --git a/src/api/runtime_zero.rs b/src/api/runtime_zero.rs index e9dce7f..98491af 100644 --- a/src/api/runtime_zero.rs +++ b/src/api/runtime_zero.rs @@ -200,7 +200,7 @@ pub(super) async fn build_runtime_gates_data( && cfg.general.me2dc_fallback && matches!(route_state.mode, RelayRouteMode::Direct); let reroute_to_direct_at_epoch_secs = if reroute_active { - shared.route_runtime.direct_since_epoch_secs() + route_state.direct_since_epoch_secs } else { None }; diff --git a/src/daemon/pid_file.rs b/src/daemon/pid_file.rs index 7f3e95b..ae1c3e1 100644 --- a/src/daemon/pid_file.rs +++ b/src/daemon/pid_file.rs @@ -1,6 +1,6 @@ use std::fs::{self, File, OpenOptions}; use std::io::{ErrorKind, Read, Write}; -use std::os::unix::fs::OpenOptionsExt; +use std::os::unix::fs::{MetadataExt, OpenOptionsExt}; use std::path::{Path, PathBuf}; use nix::fcntl::{Flock, FlockArg}; @@ -14,9 +14,25 @@ pub struct PidFile { path: PathBuf, lock_path: PathBuf, pid_file: Option, + pid_identity: Option, lock_file: Option>, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct FileIdentity { + device: u64, + inode: u64, +} + +impl FileIdentity { + fn from_metadata(metadata: &fs::Metadata) -> Self { + Self { + device: metadata.dev(), + inode: metadata.ino(), + } + } +} + impl PidFile { /// Creates a new PID file manager for the given path. pub fn new>(path: P) -> Self { @@ -26,6 +42,7 @@ impl PidFile { path, lock_path, pid_file: None, + pid_identity: None, lock_file: None, } } @@ -60,6 +77,7 @@ impl PidFile { .create(true) .truncate(false) .mode(0o644) + .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW) .open(&self.lock_path) .map_err(|error| { DaemonError::PidFile(format!( @@ -68,6 +86,7 @@ impl PidFile { error )) })?; + validate_regular_single_link(&lock_file, &self.lock_path)?; let lock_file = Flock::lock(lock_file, FlockArg::LockExclusiveNonblock).map_err(|(_, errno)| { if let Some(pid) = self.check_running().ok().flatten() { @@ -91,10 +110,13 @@ impl PidFile { .create(true) .truncate(true) .mode(0o644) + .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW) .open(&self.path) .map_err(|error| { DaemonError::PidFile(format!("cannot open {}: {}", self.path.display(), error)) })?; + let pid_metadata = validate_regular_single_link(&pid_file, &self.path)?; + let pid_identity = FileIdentity::from_metadata(&pid_metadata); let pid = getpid(); writeln!(pid_file, "{}", pid).map_err(|error| { DaemonError::PidFile(format!( @@ -103,8 +125,16 @@ impl PidFile { error )) })?; + pid_file.sync_data().map_err(|error| { + DaemonError::PidFile(format!( + "cannot sync PID file {}: {}", + self.path.display(), + error + )) + })?; self.pid_file = Some(pid_file); + self.pid_identity = Some(pid_identity); self.lock_file = Some(lock_file); info!(pid = pid.as_raw(), path = %self.path.display(), "PID file created"); Ok(()) @@ -114,19 +144,36 @@ impl PidFile { pub fn release(&mut self) -> Result<(), DaemonError> { if self.lock_file.is_none() { self.pid_file = None; + self.pid_identity = None; return Ok(()); } - let removal = match fs::remove_file(&self.path) { - Ok(()) => Ok(()), + let removal = match fs::symlink_metadata(&self.path) { + Ok(metadata) + if self.pid_identity == Some(FileIdentity::from_metadata(&metadata)) + && metadata.is_file() => + { + fs::remove_file(&self.path).map_err(|error| { + DaemonError::PidFile(format!( + "cannot remove {}: {}", + self.path.display(), + error + )) + }) + } + Ok(_) => Err(DaemonError::PidFile(format!( + "refusing to remove replaced PID file {}", + self.path.display() + ))), Err(error) if error.kind() == ErrorKind::NotFound => Ok(()), Err(error) => Err(DaemonError::PidFile(format!( - "cannot remove {}: {}", + "cannot inspect {} before removal: {}", self.path.display(), error ))), }; self.pid_file = None; + self.pid_identity = None; self.lock_file = None; removal?; debug!(path = %self.path.display(), "PID file removed"); @@ -161,7 +208,11 @@ fn sibling_lock_path(path: &Path) -> PathBuf { } fn read_pid_file_if_exists(path: &Path) -> Result, DaemonError> { - let mut file = match File::open(path) { + let mut file = match OpenOptions::new() + .read(true) + .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW) + .open(path) + { Ok(file) => file, Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None), Err(error) => { @@ -172,17 +223,46 @@ fn read_pid_file_if_exists(path: &Path) -> Result, DaemonError> { ))); } }; + let metadata = validate_regular_single_link(&file, path)?; + if metadata.len() > 64 { + return Err(DaemonError::PidFile(format!( + "invalid PID in {}", + path.display() + ))); + } let mut contents = String::new(); file.read_to_string(&mut contents).map_err(|error| { DaemonError::PidFile(format!("cannot read {}: {}", path.display(), error)) })?; - let pid = contents + let pid: i32 = contents .trim() .parse() .map_err(|_| DaemonError::PidFile(format!("invalid PID in {}", path.display())))?; + if pid <= 1 { + return Err(DaemonError::PidFile(format!( + "invalid PID in {}", + path.display() + ))); + } Ok(Some(pid)) } +fn validate_regular_single_link( + file: &File, + path: &Path, +) -> Result { + let metadata = file.metadata().map_err(|error| { + DaemonError::PidFile(format!("cannot inspect {}: {}", path.display(), error)) + })?; + if !metadata.is_file() || metadata.nlink() != 1 { + return Err(DaemonError::PidFile(format!( + "{} must be a regular file with one directory entry", + path.display() + ))); + } + Ok(metadata) +} + /// Reads a PID from a PID file. #[allow(dead_code)] pub fn read_pid_file>(path: P) -> Result { @@ -202,13 +282,6 @@ pub fn signal_pid_file>( signal: nix::sys::signal::Signal, ) -> Result<(), DaemonError> { let pid = read_pid_file(&path)?; - if !is_process_running(pid) { - return Err(DaemonError::PidFile(format!( - "process {} from {} is not running", - pid, - path.as_ref().display() - ))); - } nix::sys::signal::kill(Pid::from_raw(pid), signal) .map_err(|error| DaemonError::PidFile(format!("cannot signal process {}: {}", pid, error))) } @@ -243,6 +316,7 @@ fn is_process_running(pid: i32) -> bool { #[cfg(test)] mod tests { use std::os::unix::fs::MetadataExt; + use std::os::unix::fs::symlink; use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; @@ -367,6 +441,46 @@ mod tests { assert!(pid_path.exists()); } + #[test] + fn acquire_rejects_pid_symlink_without_truncating_target() { + let directory = tempfile::tempdir().unwrap(); + let pid_path = directory.path().join("telemt.pid"); + let target_path = directory.path().join("target"); + fs::write(&target_path, b"preserve\n").unwrap(); + symlink(&target_path, &pid_path).unwrap(); + let mut pid_file = PidFile::new(&pid_path); + + assert!(pid_file.acquire().is_err()); + assert_eq!(fs::read(&target_path).unwrap(), b"preserve\n"); + } + + #[test] + fn release_does_not_remove_replacement_path() { + let directory = tempfile::tempdir().unwrap(); + let pid_path = directory.path().join("telemt.pid"); + let owned_path = directory.path().join("owned.pid"); + let mut pid_file = PidFile::new(&pid_path); + pid_file.acquire().unwrap(); + fs::rename(&pid_path, &owned_path).unwrap(); + fs::write(&pid_path, b"replacement\n").unwrap(); + + let error = pid_file.release().unwrap_err(); + + assert!(error.to_string().contains("refusing to remove replaced PID file")); + assert_eq!(fs::read(&pid_path).unwrap(), b"replacement\n"); + } + + #[test] + fn pid_parser_rejects_process_group_values() { + let directory = tempfile::tempdir().unwrap(); + let pid_path = directory.path().join("telemt.pid"); + + for value in ["-1\n", "0\n", "1\n"] { + fs::write(&pid_path, value).unwrap(); + assert!(read_pid_file(&pid_path).is_err()); + } + } + #[test] fn pid_file_release_keeps_lock_inode() { let directory = tempfile::tempdir().unwrap(); diff --git a/src/maestro/listeners/bind.rs b/src/maestro/listeners/bind.rs index 4efb527..7b325f8 100644 --- a/src/maestro/listeners/bind.rs +++ b/src/maestro/listeners/bind.rs @@ -1,5 +1,13 @@ use std::error::Error; +#[cfg(unix)] +use std::io::{Error as IoError, ErrorKind}; use std::net::{IpAddr, SocketAddr}; +#[cfg(unix)] +use std::os::unix::fs::{FileTypeExt, MetadataExt}; +#[cfg(unix)] +use std::os::unix::net::UnixStream as StdUnixStream; +#[cfg(unix)] +use std::path::Path; use std::sync::Arc; use socket2::Socket; @@ -187,6 +195,61 @@ fn print_configured_links( print_proxy_links(host, port, config); } +#[cfg(unix)] +fn unix_path_identity(metadata: &std::fs::Metadata) -> (u64, u64) { + (metadata.dev(), metadata.ino()) +} + +#[cfg(unix)] +fn remove_stale_unix_socket(path: &Path) -> std::io::Result<()> { + let metadata = match std::fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error), + }; + if !metadata.file_type().is_socket() { + return Err(IoError::new( + ErrorKind::AlreadyExists, + format!( + "refusing to remove non-socket Unix listener path {}", + path.display() + ), + )); + } + match StdUnixStream::connect(path) { + Ok(_) => { + return Err(IoError::new( + ErrorKind::AddrInUse, + format!("Unix listener {} is already active", path.display()), + )); + } + Err(error) if error.kind() == ErrorKind::ConnectionRefused => {} + Err(error) => return Err(error), + } + let current = std::fs::symlink_metadata(path)?; + if !current.file_type().is_socket() + || unix_path_identity(¤t) != unix_path_identity(&metadata) + { + return Err(IoError::new( + ErrorKind::AlreadyExists, + format!("Unix listener path {} changed during cleanup", path.display()), + )); + } + std::fs::remove_file(path) +} + +#[cfg(unix)] +fn verify_bound_unix_socket(path: &Path, expected: (u64, u64)) -> std::io::Result<()> { + let metadata = std::fs::symlink_metadata(path)?; + if metadata.file_type().is_socket() && unix_path_identity(&metadata) == expected { + return Ok(()); + } + Err(IoError::new( + ErrorKind::AlreadyExists, + format!("Unix listener path {} was replaced", path.display()), + )) +} + /// Binds every eligible configured listener or fails without a partial inventory. pub(crate) async fn bind_listeners( config: &Arc, @@ -217,27 +280,39 @@ pub(crate) async fn bind_listeners( let mut unix_listener_out = None; #[cfg(unix)] if let Some(unix_path) = &config.server.listen_unix_sock { - let _ = tokio::fs::remove_file(unix_path).await; + let unix_path = Path::new(unix_path); + remove_stale_unix_socket(unix_path)?; let unix_listener = UnixListener::bind(unix_path)?; + let socket_metadata = std::fs::symlink_metadata(unix_path)?; + if !socket_metadata.file_type().is_socket() { + return Err(IoError::new( + ErrorKind::AlreadyExists, + format!("Unix listener path {} was replaced", unix_path.display()), + ) + .into()); + } + let socket_identity = unix_path_identity(&socket_metadata); if let Some(perm_str) = &config.server.listen_unix_sock_perm { match u32::from_str_radix(perm_str.trim_start_matches('0'), 8) { Ok(mode) => { use std::os::unix::fs::PermissionsExt; let permissions = std::fs::Permissions::from_mode(mode); + verify_bound_unix_socket(unix_path, socket_identity)?; if let Err(error_value) = std::fs::set_permissions(unix_path, permissions) { error!( - path = %unix_path, + path = %unix_path.display(), permissions = %perm_str, error = %error_value, "Failed to set Unix socket permissions" ); } else { - info!(path = %unix_path, permissions = %perm_str, "Listening on Unix socket"); + verify_bound_unix_socket(unix_path, socket_identity)?; + info!(path = %unix_path.display(), permissions = %perm_str, "Listening on Unix socket"); } } Err(error_value) => { warn!( - path = %unix_path, + path = %unix_path.display(), permissions = %perm_str, error = %error_value, "Invalid Unix socket permissions; keeping umask-derived mode" @@ -245,7 +320,7 @@ pub(crate) async fn bind_listeners( } } } else { - info!(path = %unix_path, "Listening on Unix socket"); + info!(path = %unix_path.display(), "Listening on Unix socket"); } unix_listener_out = Some(unix_listener); } @@ -271,3 +346,49 @@ pub(crate) async fn bind_listeners( unix_listener: unix_listener_out, }) } + +#[cfg(all(test, unix))] +mod tests { + use std::os::unix::fs::symlink; + use std::os::unix::net::UnixListener as StdUnixListener; + + use super::*; + + #[test] + fn unix_socket_cleanup_refuses_regular_file_and_symlink() { + let directory = tempfile::tempdir().unwrap(); + let regular = directory.path().join("regular"); + let link = directory.path().join("listener.sock"); + std::fs::write(®ular, b"preserve").unwrap(); + symlink(®ular, &link).unwrap(); + + assert!(remove_stale_unix_socket(®ular).is_err()); + assert!(remove_stale_unix_socket(&link).is_err()); + assert_eq!(std::fs::read(®ular).unwrap(), b"preserve"); + assert!(std::fs::symlink_metadata(&link).unwrap().file_type().is_symlink()); + } + + #[test] + fn unix_socket_cleanup_removes_only_stale_socket() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("listener.sock"); + let listener = StdUnixListener::bind(&path).unwrap(); + drop(listener); + + remove_stale_unix_socket(&path).unwrap(); + + assert!(!path.exists()); + } + + #[test] + fn unix_socket_cleanup_preserves_live_listener() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("listener.sock"); + let _listener = StdUnixListener::bind(&path).unwrap(); + + let error = remove_stale_unix_socket(&path).unwrap_err(); + + assert_eq!(error.kind(), ErrorKind::AddrInUse); + assert!(path.exists()); + } +} diff --git a/src/proxy/route_mode.rs b/src/proxy/route_mode.rs index a3a5d6c..d5324fd 100644 --- a/src/proxy/route_mode.rs +++ b/src/proxy/route_mode.rs @@ -1,5 +1,4 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::watch; @@ -24,11 +23,11 @@ impl RelayRouteMode { pub(crate) struct RouteCutoverState { pub mode: RelayRouteMode, pub generation: u64, + pub direct_since_epoch_secs: Option, } #[derive(Clone)] pub(crate) struct RouteRuntimeController { - direct_since_epoch_secs: Arc, tx: watch::Sender, } @@ -37,17 +36,11 @@ impl RouteRuntimeController { let initial = RouteCutoverState { mode: initial_mode, generation: 0, + direct_since_epoch_secs: matches!(initial_mode, RelayRouteMode::Direct) + .then(now_epoch_secs), }; let (tx, _rx) = watch::channel(initial); - let direct_since_epoch_secs = if matches!(initial_mode, RelayRouteMode::Direct) { - now_epoch_secs() - } else { - 0 - }; - Self { - direct_since_epoch_secs: Arc::new(AtomicU64::new(direct_since_epoch_secs)), - tx, - } + Self { tx } } pub(crate) fn snapshot(&self) -> RouteCutoverState { @@ -58,25 +51,16 @@ impl RouteRuntimeController { self.tx.subscribe() } - pub(crate) fn direct_since_epoch_secs(&self) -> Option { - let value = self.direct_since_epoch_secs.load(Ordering::Relaxed); - (value > 0).then_some(value) - } - pub(crate) fn set_mode(&self, mode: RelayRouteMode) -> Option { let mut next = None; let changed = self.tx.send_if_modified(|state| { if state.mode == mode { return false; } - if matches!(mode, RelayRouteMode::Direct) { - self.direct_since_epoch_secs - .store(now_epoch_secs(), Ordering::Relaxed); - } else { - self.direct_since_epoch_secs.store(0, Ordering::Relaxed); - } state.mode = mode; state.generation = state.generation.saturating_add(1); + state.direct_since_epoch_secs = + matches!(mode, RelayRouteMode::Direct).then(now_epoch_secs); next = Some(*state); true }); diff --git a/src/proxy/tests/route_mode_coherence_adversarial_tests.rs b/src/proxy/tests/route_mode_coherence_adversarial_tests.rs index b7f816e..b96d44b 100644 --- a/src/proxy/tests/route_mode_coherence_adversarial_tests.rs +++ b/src/proxy/tests/route_mode_coherence_adversarial_tests.rs @@ -9,7 +9,7 @@ fn positive_direct_cutover_sets_timestamp_and_snapshot_coherently() { let rx = runtime.subscribe(); assert!( - runtime.direct_since_epoch_secs().is_none(), + runtime.snapshot().direct_since_epoch_secs.is_none(), "middle startup must not expose direct-since timestamp" ); @@ -24,7 +24,7 @@ fn positive_direct_cutover_sets_timestamp_and_snapshot_coherently() { ); assert_eq!(observed.mode, RelayRouteMode::Direct); assert!( - runtime.direct_since_epoch_secs().is_some(), + observed.direct_since_epoch_secs.is_some(), "direct cutover must publish a non-empty direct-since timestamp" ); } @@ -34,12 +34,12 @@ fn negative_idempotent_set_mode_does_not_mutate_timestamp_or_generation() { let runtime = RouteRuntimeController::new(RelayRouteMode::Direct); let before_state = runtime.snapshot(); - let before_ts = runtime.direct_since_epoch_secs(); + let before_ts = before_state.direct_since_epoch_secs; let changed = runtime.set_mode(RelayRouteMode::Direct); let after_state = runtime.snapshot(); - let after_ts = runtime.direct_since_epoch_secs(); + let after_ts = after_state.direct_since_epoch_secs; assert!(changed.is_none(), "idempotent set_mode must return None"); assert_eq!( @@ -58,7 +58,7 @@ fn edge_middle_cutover_clears_timestamp() { let rx = runtime.subscribe(); assert!( - runtime.direct_since_epoch_secs().is_some(), + runtime.snapshot().direct_since_epoch_secs.is_some(), "direct startup must expose direct-since timestamp" ); @@ -73,7 +73,7 @@ fn edge_middle_cutover_clears_timestamp() { ); assert_eq!(observed.mode, RelayRouteMode::Middle); assert!( - runtime.direct_since_epoch_secs().is_none(), + observed.direct_since_epoch_secs.is_none(), "middle cutover must clear direct-since timestamp" ); } @@ -90,7 +90,7 @@ fn adversarial_blackhat_probe_sequence_observes_consistent_mode_timestamp_pairs( let observed_direct = *rx.borrow(); assert_eq!(observed_direct, emitted_direct); assert!( - runtime.direct_since_epoch_secs().is_some(), + observed_direct.direct_since_epoch_secs.is_some(), "direct observation must never expose empty timestamp" ); @@ -100,7 +100,7 @@ fn adversarial_blackhat_probe_sequence_observes_consistent_mode_timestamp_pairs( let observed_middle = *rx.borrow(); assert_eq!(observed_middle, emitted_middle); assert!( - runtime.direct_since_epoch_secs().is_none(), + observed_middle.direct_since_epoch_secs.is_none(), "middle observation must never expose direct timestamp" ); } @@ -136,9 +136,9 @@ fn integration_subscriber_and_runtime_gates_stay_coherent_across_cutovers() { assert_eq!(snapshot, emitted); if matches!(mode, RelayRouteMode::Direct) { - assert!(runtime.direct_since_epoch_secs().is_some()); + assert!(snapshot.direct_since_epoch_secs.is_some()); } else { - assert!(runtime.direct_since_epoch_secs().is_none()); + assert!(snapshot.direct_since_epoch_secs.is_none()); } } } @@ -176,12 +176,12 @@ fn light_fuzz_random_mode_plan_preserves_timestamp_and_generation_invariants() { if matches!(snapshot.mode, RelayRouteMode::Direct) { assert!( - runtime.direct_since_epoch_secs().is_some(), + snapshot.direct_since_epoch_secs.is_some(), "direct fuzz state must expose timestamp" ); } else { assert!( - runtime.direct_since_epoch_secs().is_none(), + snapshot.direct_since_epoch_secs.is_none(), "middle fuzz state must clear timestamp" ); } @@ -199,12 +199,17 @@ fn stress_parallel_subscribers_never_observe_generation_regression() { let rx = runtime.subscribe(); let mut last = rx.borrow().generation; for _ in 0..10_000usize { - let current = rx.borrow().generation; + let current = *rx.borrow(); assert!( - current >= last, + current.generation >= last, "watch generation must be monotonic for every subscriber" ); - last = current; + assert_eq!( + matches!(current.mode, RelayRouteMode::Direct), + current.direct_since_epoch_secs.is_some(), + "one route snapshot must contain a coherent mode/timestamp pair" + ); + last = current.generation; std::thread::yield_now(); } })); @@ -227,8 +232,8 @@ fn stress_parallel_subscribers_never_observe_generation_regression() { let final_state = runtime.snapshot(); if matches!(final_state.mode, RelayRouteMode::Direct) { - assert!(runtime.direct_since_epoch_secs().is_some()); + assert!(final_state.direct_since_epoch_secs.is_some()); } else { - assert!(runtime.direct_since_epoch_secs().is_none()); + assert!(final_state.direct_since_epoch_secs.is_none()); } } diff --git a/src/proxy/tests/route_mode_security_tests.rs b/src/proxy/tests/route_mode_security_tests.rs index e5925fc..cc169ca 100644 --- a/src/proxy/tests/route_mode_security_tests.rs +++ b/src/proxy/tests/route_mode_security_tests.rs @@ -120,6 +120,7 @@ fn session_is_not_affected_when_mode_matches_even_if_generation_advanced() { let current = RouteCutoverState { mode: RelayRouteMode::Direct, generation: 2, + direct_since_epoch_secs: Some(1), }; let session_generation = 0; @@ -134,6 +135,7 @@ fn cutover_predicate_rejects_equal_generation_even_if_mode_differs() { let current = RouteCutoverState { mode: RelayRouteMode::Middle, generation: 77, + direct_since_epoch_secs: None, }; assert!( !is_session_affected_by_cutover(current, RelayRouteMode::Direct, 77), @@ -168,13 +170,15 @@ fn adversarial_route_oscillation_only_cuts_over_sessions_with_different_final_mo fn light_fuzz_cutover_predicate_matches_reference_oracle() { let mut rng = StdRng::seed_from_u64(0xC0DEC0DE5EED); for _ in 0..20_000 { + let mode = if rng.random::() { + RelayRouteMode::Direct + } else { + RelayRouteMode::Middle + }; let current = RouteCutoverState { - mode: if rng.random::() { - RelayRouteMode::Direct - } else { - RelayRouteMode::Middle - }, + mode, generation: rng.random_range(0u64..1_000_000), + direct_since_epoch_secs: matches!(mode, RelayRouteMode::Direct).then_some(1), }; let session_mode = if rng.random::() { RelayRouteMode::Direct diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index ea6bf6b..1c80f28 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -17,8 +17,8 @@ use crate::config::{ use crate::maestro::generation::test_runtime_generation; use crate::web::frame::{self, FrameType}; use crate::web::manager::{ - CloseOperationSelector, ControlError, SessionDetail, SessionFilter, SessionListRequest, - SessionRefError, WebProcessRuntime, + CloseOperationSelector, ControlError, ManagerError, SessionDetail, SessionFilter, + SessionListRequest, SessionRefError, WebProcessRuntime, }; #[path = "legacy_tests.rs"] @@ -433,6 +433,35 @@ async fn bridge_bootstrap_uses_the_generation_that_selected_its_profile() { replacement.stop_background_tasks().await; } +#[tokio::test] +async fn stale_generation_cannot_publish_bootstrap_after_disabled_cutover() { + let capability = [23u8; 32]; + let initial = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); + let runtime = WebProcessRuntime::start(active_runtime); + let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); + let mut disabled_config = runtime_config(capability, WebCarrier::Https); + disabled_config.web.enabled = false; + let disabled = test_runtime_generation(2, disabled_config); + + runtime.activate_generation(Arc::clone(&disabled)); + let result = runtime.issue_bootstrap_for_generation( + &initial, + profile, + "192.0.2.10".parse().unwrap(), + ); + + assert!(matches!(result, Err(ManagerError::Closed))); + let status = serde_json::to_value(runtime.try_status()).unwrap(); + assert_eq!(status["manager"]["bootstraps"], 0); + + runtime.shutdown().await; + initial.stop_sessions().await; + initial.stop_background_tasks().await; + disabled.stop_sessions().await; + disabled.stop_background_tasks().await; +} + #[tokio::test] async fn unused_bootstrap_is_rejected_after_profile_identity_change() { let capability = [11u8; 32]; diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs index e00df38..8d9fd86 100644 --- a/src/web/manager/credentials.rs +++ b/src/web/manager/credentials.rs @@ -102,7 +102,7 @@ impl WebProcessRuntime { self.record_limit_hit(); self.telemetry .record_rejection(WebRejectionReason::RuntimeClosed); - return Err(ManagerError::Limit); + return Err(ManagerError::Closed); } if state .bootstraps_per_ip