Direct-since-epoch

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,
+    AccessSection, compute_snapshot_revision, compute_source_revision, load_candidate_snapshot,
-    resolve_single_source_owner, toml_path_exists,
+    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<String, ApiFailure> {
+    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<String, ApiFailure> {
     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<Option<ExistingTarget>> {
+    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, &current) {
+            (Some(expected), Some(current)) => {
+                same_target(&expected.metadata, &current.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();

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
     };

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<File>,
+    pid_identity: Option<FileIdentity>,
     lock_file: Option<Flock<File>>,
 }
 
+#[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<P: AsRef<Path>>(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) {
+        let removal = match fs::symlink_metadata(&self.path) {
-            Ok(()) => Ok(()),
+            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<Option<i32>, 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<Option<i32>, 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<fs::Metadata, DaemonError> {
+    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<P: AsRef<Path>>(path: P) -> Result<i32, DaemonError> {
@@ -202,13 +282,6 @@ pub fn signal_pid_file<P: AsRef<Path>>(
     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();

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(&current) != 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<ProxyConfig>,
@@ -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(&regular, b"preserve").unwrap();
+        symlink(&regular, &link).unwrap();
+
+        assert!(remove_stale_unix_socket(&regular).is_err());
+        assert!(remove_stale_unix_socket(&link).is_err());
+        assert_eq!(std::fs::read(&regular).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());
+    }
+}

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<u64>,
 }
 
 #[derive(Clone)]
 pub(crate) struct RouteRuntimeController {
-    direct_since_epoch_secs: Arc<AtomicU64>,
     tx: watch::Sender<RouteCutoverState>,
 }
 
@@ -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) {
+        Self { tx }
-            now_epoch_secs()
-        } else {
-            0
-        };
-        Self {
-            direct_since_epoch_secs: Arc::new(AtomicU64::new(direct_since_epoch_secs)),
-            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<u64> {
-        let value = self.direct_since_epoch_secs.load(Ordering::Relaxed);
-        (value > 0).then_some(value)
-    }
-
     pub(crate) fn set_mode(&self, mode: RelayRouteMode) -> Option<RouteCutoverState> {
         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
         });

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());
     }
 }

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 current = RouteCutoverState {
+        let mode = if rng.random::<bool>() {
-            mode: if rng.random::<bool>() {
             RelayRouteMode::Direct
         } else {
             RelayRouteMode::Middle
-            },
+        };
+        let current = RouteCutoverState {
+            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::<bool>() {
             RelayRouteMode::Direct

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,
+    CloseOperationSelector, ControlError, ManagerError, SessionDetail, SessionFilter,
-    SessionRefError, WebProcessRuntime,
+    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];

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