fix(connect): make registration bootstrap retry durable (#6468)

* fix(connect): make registration state durability retry-safe

* fix(connect): reject parent state paths

* fix(connect): harden state directory creation

* fix(connect): bound bootstrap directory syncs

* fix(connect): require durable state parent

* fix(connect): close bootstrap marker race

* test(connect): cover marker sync failure

rustfs/src/connect/registration_bootstrap.rs

@@ -17,7 +17,9 @@ use std::path::Path;
 #[cfg(unix)]
 use std::{
     fs::{self, File, OpenOptions},
-    path::PathBuf,
+    io::Write as _,
+    path::{Component, PathBuf},
+    sync::atomic::{AtomicU64, Ordering},
     time::Duration,
 };
 
@@ -27,6 +29,12 @@ use super::{ConnectClient, ConnectConfig, CredentialStore, IdentityStore, Regist
 
 #[cfg(unix)]
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
+#[cfg(unix)]
+const BOOTSTRAP_READY_FILE: &str = ".bootstrap-ready";
+#[cfg(unix)]
+const BOOTSTRAP_READY_CONTENTS: &[u8] = b"v1\n";
+#[cfg(unix)]
+static BOOTSTRAP_STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0);
 
 #[derive(Debug, PartialEq, Eq)]
 pub struct RegistrationBootstrapResult {
@@ -42,6 +50,10 @@ pub enum RegistrationBootstrapError {
     RootCaFileSecurity,
     #[error("the Connect state path must be an explicit directory, not a symlink")]
     StateDirectorySecurity,
+    #[error("the Connect state directory parent must be pre-provisioned as a secure durable directory")]
+    StateParentRequired,
+    #[error("the Connect bootstrap readiness marker must be an owner-only regular file")]
+    StateMarkerSecurity,
     #[error("failed to read protected Connect registration input")]
     Input(#[source] io::Error),
     #[error("Connect registration configuration is invalid")]
@@ -112,62 +124,207 @@ fn prepare_state_directory(path: &Path) -> Result<PathBuf, RegistrationBootstrap
 #[cfg(unix)]
 fn prepare_state_directory_with_sync(
     path: &Path,
-    mut sync: impl FnMut(&Path) -> io::Result<()>,
+    sync: impl FnMut(&Path) -> io::Result<()>,
 ) -> Result<PathBuf, RegistrationBootstrapError> {
+    prepare_state_directory_with_sync_and_missing_observer(path, sync, |_| {})
+}
+
+#[cfg(unix)]
+fn prepare_state_directory_with_sync_and_missing_observer(
+    path: &Path,
+    mut sync: impl FnMut(&Path) -> io::Result<()>,
+    mut observed_missing: impl FnMut(&Path),
+) -> Result<PathBuf, RegistrationBootstrapError> {
+    if path.components().any(|component| matches!(component, Component::ParentDir)) {
+        return Err(RegistrationBootstrapError::StateDirectorySecurity);
+    }
     let path = if path.is_absolute() {
         path.to_path_buf()
     } else {
         std::env::current_dir().map_err(RegistrationBootstrapError::Input)?.join(path)
     };
 
-    let mut directories = path.ancestors().map(Path::to_path_buf).collect::<Vec<_>>();
+    let state_parent = path.parent().ok_or(RegistrationBootstrapError::StateParentRequired)?;
+    match fs::symlink_metadata(state_parent) {
+        Ok(_) => {}
+        Err(error) if error.kind() == io::ErrorKind::NotFound => {
+            return Err(RegistrationBootstrapError::StateParentRequired);
+        }
+        Err(error) => return Err(RegistrationBootstrapError::Input(error)),
+    }
+
+    let mut directories = state_parent.ancestors().map(Path::to_path_buf).collect::<Vec<_>>();
     directories.reverse();
     for directory in &directories {
-        if ensure_directory(directory, directory == &path)? {
-            sync(directory).map_err(RegistrationBootstrapError::Input)?;
-            let parent = directory.parent().ok_or_else(|| {
-                RegistrationBootstrapError::Input(io::Error::new(io::ErrorKind::InvalidInput, "directory has no parent"))
-            })?;
-            sync(parent).map_err(RegistrationBootstrapError::Input)?;
-        }
+        validate_directory(directory, false)?;
     }
-    let store_directories = [path.join("identity"), path.join("credential")];
-    for directory in &store_directories {
-        if ensure_directory(directory, true)? {
-            sync(directory).map_err(RegistrationBootstrapError::Input)?;
-            let parent = directory.parent().ok_or_else(|| {
-                RegistrationBootstrapError::Input(io::Error::new(io::ErrorKind::InvalidInput, "directory has no parent"))
-            })?;
-            sync(parent).map_err(RegistrationBootstrapError::Input)?;
+
+    let state_exists = match fs::symlink_metadata(&path) {
+        Ok(_) => {
+            validate_directory(&path, true)?;
+            true
         }
+        Err(error) if error.kind() == io::ErrorKind::NotFound => false,
+        Err(error) => return Err(RegistrationBootstrapError::Input(error)),
+    };
+    let store_directories = [path.join("identity"), path.join("credential")];
+    let ready = path.join(BOOTSTRAP_READY_FILE);
+    if state_exists && ready_marker_exists(&ready)? {
+        for directory in &store_directories {
+            validate_directory(directory, true)?;
+        }
+        return Ok(path);
+    }
+
+    ensure_directory(&path, true, &mut observed_missing)?;
+    for directory in &store_directories {
+        ensure_directory(directory, true, &mut observed_missing)?;
     }
     for directory in &directories {
-        validate_directory(directory, directory == &path)?;
+        validate_directory(directory, false)?;
     }
+    validate_directory(&path, true)?;
     for directory in &store_directories {
         validate_directory(directory, true)?;
     }
+
+    // The caller-provisioned parent is the durability anchor. Without a ready
+    // marker, repeat this complete commit sequence after every interruption.
+    for directory in [
+        store_directories[0].clone(),
+        store_directories[1].clone(),
+        path.clone(),
+        state_parent.to_path_buf(),
+    ] {
+        sync(&directory).map_err(RegistrationBootstrapError::Input)?;
+    }
+    publish_ready_marker(&path, &ready)?;
     Ok(path)
 }
 
 #[cfg(unix)]
-fn ensure_directory(path: &Path, require_process_owner: bool) -> Result<bool, RegistrationBootstrapError> {
+fn ready_marker_exists(path: &Path) -> Result<bool, RegistrationBootstrapError> {
+    let initial = match fs::symlink_metadata(path) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
+        Err(error) => return Err(RegistrationBootstrapError::Input(error)),
+    };
+    if initial.file_type().is_symlink() || !initial.is_file() {
+        return Err(RegistrationBootstrapError::StateMarkerSecurity);
+    }
+
+    let mut options = OpenOptions::new();
+    options.read(true);
+    use std::os::unix::fs::OpenOptionsExt as _;
+    options.custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW);
+    let mut file = options.open(path).map_err(RegistrationBootstrapError::Input)?;
+    let metadata = file.metadata().map_err(RegistrationBootstrapError::Input)?;
+    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
+    let mode = metadata.permissions().mode() & 0o777;
+    if !metadata.is_file() || !unix_ready_marker_is_trusted(metadata.uid(), mode, process_uid()) {
+        return Err(RegistrationBootstrapError::StateMarkerSecurity);
+    }
+
+    let mut contents = Vec::with_capacity(BOOTSTRAP_READY_CONTENTS.len() + 1);
+    io::Read::read_to_end(&mut io::Read::take(&mut file, (BOOTSTRAP_READY_CONTENTS.len() + 1) as u64), &mut contents)
+        .map_err(RegistrationBootstrapError::Input)?;
+    if contents != BOOTSTRAP_READY_CONTENTS {
+        return Err(RegistrationBootstrapError::StateMarkerSecurity);
+    }
+    Ok(true)
+}
+
+#[cfg(unix)]
+fn publish_ready_marker(state_directory: &Path, ready: &Path) -> Result<(), RegistrationBootstrapError> {
+    publish_ready_marker_with_existing_observer(state_directory, ready, File::sync_all, || {})
+}
+
+#[cfg(unix)]
+fn publish_ready_marker_with_existing_observer(
+    state_directory: &Path,
+    ready: &Path,
+    mut sync_staging: impl FnMut(&File) -> io::Result<()>,
+    mut existing_observer: impl FnMut(),
+) -> Result<(), RegistrationBootstrapError> {
+    let (staging_path, mut staging) = loop {
+        let staging_path = state_directory.join(format!(
+            "{BOOTSTRAP_READY_FILE}.{}.{}.tmp",
+            std::process::id(),
+            BOOTSTRAP_STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed)
+        ));
+        let mut options = OpenOptions::new();
+        options.write(true).create_new(true);
+        use std::os::unix::fs::OpenOptionsExt as _;
+        options.mode(0o600);
+        match options.open(&staging_path) {
+            Ok(file) => break (staging_path, file),
+            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
+            Err(error) => return Err(RegistrationBootstrapError::Input(error)),
+        }
+    };
+
+    let staged = (|| -> io::Result<()> {
+        staging.write_all(BOOTSTRAP_READY_CONTENTS)?;
+        use std::os::unix::fs::PermissionsExt as _;
+        staging.set_permissions(fs::Permissions::from_mode(0o600))?;
+        sync_staging(&staging)
+    })();
+    drop(staging);
+    if let Err(error) = staged {
+        let _ = fs::remove_file(&staging_path);
+        return Err(RegistrationBootstrapError::Input(error));
+    }
+
+    // Hard-link publication is atomic and, unlike rename, cannot replace a
+    // marker planted between validation and publication.
+    let published = fs::hard_link(&staging_path, ready);
+    if let Err(error) = fs::remove_file(&staging_path) {
+        if published.is_ok() {
+            let _ = fs::remove_file(ready);
+        }
+        return Err(RegistrationBootstrapError::Input(error));
+    }
+    match published {
+        Ok(()) => Ok(()),
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
+            existing_observer();
+            if ready_marker_exists(ready)? {
+                Ok(())
+            } else {
+                Err(RegistrationBootstrapError::Input(io::Error::new(
+                    io::ErrorKind::NotFound,
+                    "bootstrap readiness marker disappeared during publication",
+                )))
+            }
+        }
+        Err(error) => Err(RegistrationBootstrapError::Input(error)),
+    }
+}
+
+#[cfg(unix)]
+fn ensure_directory(
+    path: &Path,
+    require_process_owner: bool,
+    observed_missing: &mut impl FnMut(&Path),
+) -> Result<(), RegistrationBootstrapError> {
     match fs::symlink_metadata(path) {
         Ok(_) => {
             validate_directory(path, require_process_owner)?;
-            Ok(false)
+            Ok(())
         }
         Err(error) if error.kind() == io::ErrorKind::NotFound => {
+            observed_missing(path);
             let mut builder = fs::DirBuilder::new();
             use std::os::unix::fs::DirBuilderExt as _;
             builder.mode(0o700);
             match builder.create(path) {
                 Ok(()) => {
                     validate_directory(path, true)?;
-                    Ok(true)
+                    Ok(())
                 }
                 Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
-                    Err(RegistrationBootstrapError::StateDirectorySecurity)
+                    validate_directory(path, require_process_owner)?;
+                    Ok(())
                 }
                 Err(error) => Err(RegistrationBootstrapError::Input(error)),
             }
@@ -201,6 +358,11 @@ fn unix_directory_is_trusted(owner_uid: u32, mode: u32, process_uid: u32, requir
     (owner_uid == process_uid || (!require_process_owner && owner_uid == 0)) && mode & 0o022 == 0
 }
 
+#[cfg(unix)]
+fn unix_ready_marker_is_trusted(owner_uid: u32, mode: u32, process_uid: u32) -> bool {
+    owner_uid == process_uid && mode & 0o400 != 0 && mode & 0o177 == 0
+}
+
 #[cfg(unix)]
 // SAFETY: geteuid has no pointer arguments or caller preconditions.
 #[allow(unsafe_code)]
@@ -259,14 +421,32 @@ fn unix_ca_file_is_trusted(owner_uid: u32, mode: u32, process_uid: u32) -> bool
 #[cfg(all(test, unix))]
 mod tests {
     use std::cell::{Cell, RefCell};
-    use std::fs;
+    use std::fs::{self, File};
     use std::io;
-    use std::os::unix::fs::PermissionsExt as _;
+    use std::os::unix::fs::{PermissionsExt as _, symlink};
+    use std::path::Path;
+    use std::sync::{Arc, Barrier};
 
     use super::{
-        RegistrationBootstrapError, prepare_state_directory_with_sync, unix_ca_file_is_trusted, unix_directory_is_trusted,
+        BOOTSTRAP_READY_CONTENTS, BOOTSTRAP_READY_FILE, RegistrationBootstrapError, prepare_state_directory_with_sync,
+        prepare_state_directory_with_sync_and_missing_observer, publish_ready_marker_with_existing_observer, ready_marker_exists,
+        sync_directory, unix_ca_file_is_trusted, unix_directory_is_trusted, unix_ready_marker_is_trusted,
     };
 
+    fn secure_tempdir() -> tempfile::TempDir {
+        tempfile::Builder::new()
+            .prefix(".connect-registration-bootstrap-")
+            .tempdir_in(env!("CARGO_MANIFEST_DIR"))
+            .expect("temporary directory inside the protected checkout")
+    }
+
+    fn create_secure_state_tree(state: &Path) {
+        for directory in [state.to_path_buf(), state.join("identity"), state.join("credential")] {
+            fs::create_dir_all(&directory).expect("create state directory");
+            fs::set_permissions(&directory, fs::Permissions::from_mode(0o700)).expect("secure state directory");
+        }
+    }
+
     #[test]
     fn unix_directory_policy_rejects_wrong_owners_and_writable_modes_only() {
         let process_uid = 501;
@@ -292,60 +472,385 @@ mod tests {
     }
 
     #[test]
-    fn new_state_chain_syncs_each_created_directory_then_parent_and_propagates_failure() {
-        let temp = tempfile::tempdir().expect("temporary directory");
-        let ancestor = temp.path().join("connect");
-        let state = ancestor.join("state");
-        let observed = RefCell::new(Vec::new());
-        let calls = Cell::new(0);
+    fn unix_ready_marker_policy_requires_process_owner_and_owner_only_readability() {
+        let process_uid = 501;
 
-        let error = prepare_state_directory_with_sync(&state, |path| {
+        assert!(unix_ready_marker_is_trusted(process_uid, 0o400, process_uid));
+        assert!(unix_ready_marker_is_trusted(process_uid, 0o600, process_uid));
+        assert!(!unix_ready_marker_is_trusted(process_uid + 1, 0o600, process_uid));
+        assert!(!unix_ready_marker_is_trusted(process_uid, 0o200, process_uid));
+        assert!(!unix_ready_marker_is_trusted(process_uid, 0o640, process_uid));
+        assert!(!unix_ready_marker_is_trusted(process_uid, 0o602, process_uid));
+    }
+
+    #[test]
+    fn new_state_syncs_managed_directories_and_preprovisioned_parent() {
+        let temp = secure_tempdir();
+        let state = temp.path().join("state");
+        let observed = RefCell::new(Vec::new());
+
+        let prepared = prepare_state_directory_with_sync(&state, |path| {
             assert!(path.is_dir(), "directory must exist before it is synced");
             observed.borrow_mut().push(path.to_path_buf());
-            calls.set(calls.get() + 1);
-            if calls.get() == 8 {
-                return Err(io::Error::other("injected final parent sync failure"));
-            }
             Ok(())
         })
-        .expect_err("parent sync failure must stop bootstrap preparation");
+        .expect("new state tree must become ready");
 
-        assert!(matches!(error, RegistrationBootstrapError::Input(_)));
+        assert_eq!(prepared, state);
+        let observed = observed.into_inner();
         assert_eq!(
-            observed.into_inner(),
-            vec![
-                ancestor.clone(),
-                temp.path().to_path_buf(),
-                state.clone(),
-                ancestor,
+            observed,
+            [
                 state.join("identity"),
-                state.clone(),
                 state.join("credential"),
                 state,
-            ]
+                temp.path().to_path_buf(),
+            ],
+            "new state trees must sync only managed directories and their durable parent"
         );
     }
 
     #[test]
-    fn existing_state_tree_is_validated_without_syncing() {
-        let temp = tempfile::tempdir().expect("temporary directory");
-        let state = temp.path().join("state");
-        let directories = [state.clone(), state.join("identity"), state.join("credential")];
-        for directory in &directories {
-            fs::create_dir_all(directory).expect("create existing state directory");
-            fs::set_permissions(directory, fs::Permissions::from_mode(0o700)).expect("secure existing state directory");
-        }
+    fn missing_state_parent_fails_before_creation_or_sync() {
+        let temp = secure_tempdir();
+        let missing_parent = temp.path().join("missing");
+        let state = missing_parent.join("state");
         let calls = Cell::new(0);
 
-        let prepared = prepare_state_directory_with_sync(&state, |_| {
+        let error = prepare_state_directory_with_sync(&state, |_| {
             calls.set(calls.get() + 1);
-            Err(io::Error::other("existing directories must not be synced"))
+            Ok(())
         })
-        .expect("existing secure state tree should be ready");
+        .expect_err("state parent must be provisioned before bootstrap");
+
+        assert!(matches!(error, RegistrationBootstrapError::StateParentRequired));
+        assert_eq!(calls.get(), 0);
+        assert!(!missing_parent.exists());
+        assert_eq!(
+            error.to_string(),
+            "the Connect state directory parent must be pre-provisioned as a secure durable directory"
+        );
+    }
+
+    #[test]
+    fn parent_components_fail_before_creation_or_sync_while_current_directory_is_allowed() {
+        let temp = secure_tempdir();
+        let absolute_parent = temp.path().join("secure/connect/..");
+
+        for path in [absolute_parent.as_path(), Path::new("state/../other")] {
+            let calls = Cell::new(0);
+            let error = prepare_state_directory_with_sync(path, |_| {
+                calls.set(calls.get() + 1);
+                Ok(())
+            })
+            .expect_err("parent components must fail before filesystem preparation");
+
+            assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+            assert_eq!(calls.get(), 0);
+        }
+        assert!(!temp.path().join("secure").exists());
+
+        let dotted_state = temp.path().join("./state");
+        prepare_state_directory_with_sync(&dotted_state, |_| Ok(())).expect("current-directory component is safe");
+        assert_eq!(
+            fs::read(temp.path().join("state").join(BOOTSTRAP_READY_FILE)).expect("read ready marker"),
+            BOOTSTRAP_READY_CONTENTS
+        );
+    }
+
+    #[test]
+    fn missing_marker_retries_all_managed_syncs_after_final_parent_failure() {
+        let temp = secure_tempdir();
+        let state = temp.path().join("state");
+        create_secure_state_tree(&state);
+        let ready = state.join(BOOTSTRAP_READY_FILE);
+        let first_attempt = RefCell::new(Vec::new());
+
+        let error = prepare_state_directory_with_sync(&state, |path| {
+            first_attempt.borrow_mut().push(path.to_path_buf());
+            if path == temp.path() {
+                return Err(io::Error::other("injected final parent sync failure"));
+            }
+            Ok(())
+        })
+        .expect_err("final parent sync failure must stop preparation");
+
+        assert!(matches!(error, RegistrationBootstrapError::Input(_)));
+        assert_eq!(
+            first_attempt.into_inner(),
+            vec![
+                state.join("identity"),
+                state.join("credential"),
+                state.clone(),
+                temp.path().to_path_buf()
+            ]
+        );
+        assert!(!ready.exists(), "failed durability preparation must not publish ready");
+
+        let retry = RefCell::new(Vec::new());
+        let prepared = prepare_state_directory_with_sync(&state, |path| {
+            retry.borrow_mut().push(path.to_path_buf());
+            Ok(())
+        })
+        .expect("retry must repeat durability preparation");
 
+        assert_eq!(prepared, state);
+        assert_eq!(
+            retry.into_inner(),
+            [
+                state.join("identity"),
+                state.join("credential"),
+                state.clone(),
+                temp.path().to_path_buf(),
+            ],
+            "an interrupted complete tree must resync only its managed path and direct parent"
+        );
+        assert_eq!(fs::read(&ready).expect("read ready marker"), BOOTSTRAP_READY_CONTENTS);
+
+        let calls = Cell::new(0);
+        let prepared = prepare_state_directory_with_sync(&state, |_| {
+            calls.set(calls.get() + 1);
+            Err(io::Error::other("ready trees must not sync"))
+        })
+        .expect("safe ready tree should bypass sync");
         assert_eq!(prepared, state);
         assert_eq!(calls.get(), 0);
     }
+
+    #[test]
+    fn every_pre_marker_sync_failure_retries_the_complete_commit() {
+        let temp = secure_tempdir();
+
+        for fail_at in 1..=4 {
+            let state = temp.path().join(format!("state-{fail_at}"));
+            create_secure_state_tree(&state);
+            let expected = [
+                state.join("identity"),
+                state.join("credential"),
+                state.clone(),
+                temp.path().to_path_buf(),
+            ];
+            let calls = Cell::new(0);
+            let error = prepare_state_directory_with_sync(&state, |path| {
+                assert_eq!(path, &expected[calls.get()]);
+                calls.set(calls.get() + 1);
+                if calls.get() == fail_at {
+                    return Err(io::Error::other("injected pre-marker sync failure"));
+                }
+                Ok(())
+            })
+            .expect_err("every managed durability failure must stop preparation");
+
+            assert!(matches!(error, RegistrationBootstrapError::Input(_)));
+            assert_eq!(calls.get(), fail_at);
+            assert!(!state.join(BOOTSTRAP_READY_FILE).exists());
+
+            let retry = RefCell::new(Vec::new());
+            prepare_state_directory_with_sync(&state, |path| {
+                retry.borrow_mut().push(path.to_path_buf());
+                Ok(())
+            })
+            .expect("retry must repeat the complete durability commit");
+            assert_eq!(retry.into_inner(), expected);
+            assert!(ready_marker_exists(&state.join(BOOTSTRAP_READY_FILE)).expect("validate ready marker"));
+        }
+    }
+
+    #[test]
+    fn missing_directory_replacement_is_revalidated_after_create_race() {
+        let temp = secure_tempdir();
+
+        let target = temp.path().join("target");
+        fs::create_dir(&target).expect("create symlink target");
+        let linked_state = temp.path().join("linked-state");
+        let error = prepare_state_directory_with_sync_and_missing_observer(
+            &linked_state,
+            |_| panic!("a raced symlink must fail before syncing"),
+            |missing| {
+                if missing == linked_state {
+                    symlink(&target, missing).expect("replace missing state with symlink");
+                }
+            },
+        )
+        .expect_err("raced symlink must fail closed");
+        assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+        assert!(!target.join("identity").exists());
+        assert!(!target.join("credential").exists());
+        assert!(!target.join(BOOTSTRAP_READY_FILE).exists());
+
+        let shared_state = temp.path().join("shared-state-race");
+        let error = prepare_state_directory_with_sync_and_missing_observer(
+            &shared_state,
+            |_| panic!("a raced shared directory must fail before syncing"),
+            |missing| {
+                if missing == shared_state {
+                    fs::create_dir(missing).expect("replace missing state with directory");
+                    fs::set_permissions(missing, fs::Permissions::from_mode(0o770)).expect("share raced directory");
+                }
+            },
+        )
+        .expect_err("raced shared directory must fail closed");
+        assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+        assert!(!shared_state.join("identity").exists());
+        assert!(!shared_state.join("credential").exists());
+        assert!(!shared_state.join(BOOTSTRAP_READY_FILE).exists());
+    }
+
+    #[test]
+    fn ready_marker_rejects_symlinks_shared_modes_non_files_and_invalid_contents() {
+        let temp = secure_tempdir();
+
+        for case in ["symlink", "shared", "directory", "contents"] {
+            let state = temp.path().join(case);
+            create_secure_state_tree(&state);
+            let ready = state.join(BOOTSTRAP_READY_FILE);
+            match case {
+                "symlink" => {
+                    let target = temp.path().join("marker-target");
+                    fs::write(&target, BOOTSTRAP_READY_CONTENTS).expect("write marker target");
+                    fs::set_permissions(&target, fs::Permissions::from_mode(0o600)).expect("secure marker target");
+                    symlink(target, &ready).expect("create marker symlink");
+                }
+                "shared" => {
+                    fs::write(&ready, BOOTSTRAP_READY_CONTENTS).expect("write shared marker");
+                    fs::set_permissions(&ready, fs::Permissions::from_mode(0o640)).expect("share marker");
+                }
+                "directory" => fs::create_dir(&ready).expect("create marker directory"),
+                "contents" => {
+                    fs::write(&ready, b"not-ready\n").expect("write invalid marker");
+                    fs::set_permissions(&ready, fs::Permissions::from_mode(0o600)).expect("secure invalid marker");
+                }
+                _ => unreachable!(),
+            }
+
+            let error = prepare_state_directory_with_sync(&state, |_| panic!("an unsafe marker must fail before syncing"))
+                .expect_err("unsafe marker must fail closed");
+            assert!(matches!(error, RegistrationBootstrapError::StateMarkerSecurity));
+        }
+    }
+
+    #[test]
+    fn marker_removed_after_publication_conflict_fails_closed() {
+        let temp = secure_tempdir();
+        let state = temp.path().join("state");
+        create_secure_state_tree(&state);
+        let ready = state.join(BOOTSTRAP_READY_FILE);
+        fs::write(&ready, BOOTSTRAP_READY_CONTENTS).expect("write existing ready marker");
+        fs::set_permissions(&ready, fs::Permissions::from_mode(0o600)).expect("secure existing ready marker");
+
+        let error = publish_ready_marker_with_existing_observer(&state, &ready, File::sync_all, || {
+            fs::remove_file(&ready).expect("remove marker after no-replace conflict");
+        })
+        .expect_err("a marker removed before validation must fail closed");
+
+        assert!(matches!(
+            error,
+            RegistrationBootstrapError::Input(ref source) if source.kind() == io::ErrorKind::NotFound
+        ));
+        assert!(!ready.exists());
+        assert!(
+            fs::read_dir(&state)
+                .expect("read state directory")
+                .filter_map(Result::ok)
+                .all(|entry| !entry.file_name().to_string_lossy().ends_with(".tmp")),
+            "failed publication must not leave staging files"
+        );
+    }
+
+    #[test]
+    fn staging_sync_failure_does_not_publish_ready_marker_and_can_retry() {
+        let temp = secure_tempdir();
+        let state = temp.path().join("state");
+        create_secure_state_tree(&state);
+        let ready = state.join(BOOTSTRAP_READY_FILE);
+
+        let error = publish_ready_marker_with_existing_observer(
+            &state,
+            &ready,
+            |_| Err(io::Error::other("injected staging sync failure")),
+            || {},
+        )
+        .expect_err("staging sync failure must stop publication");
+
+        assert!(matches!(error, RegistrationBootstrapError::Input(ref source) if source.kind() == io::ErrorKind::Other));
+        assert!(!ready.exists(), "staging sync failure must not publish the marker");
+        assert!(
+            fs::read_dir(&state)
+                .expect("read state directory")
+                .filter_map(Result::ok)
+                .all(|entry| !entry.file_name().to_string_lossy().ends_with(".tmp")),
+            "staging sync failure must not leave staging files"
+        );
+
+        publish_ready_marker_with_existing_observer(&state, &ready, File::sync_all, || {})
+            .expect("retry should publish a durable marker");
+        assert!(ready_marker_exists(&ready).expect("validate retried ready marker"));
+    }
+
+    #[test]
+    fn ready_marker_does_not_bypass_ancestor_validation() {
+        let temp = secure_tempdir();
+        let ancestor = temp.path().join("ancestor");
+        let state = ancestor.join("state");
+        create_secure_state_tree(&state);
+        prepare_state_directory_with_sync(&state, |_| Ok(())).expect("publish ready marker");
+
+        fs::set_permissions(&ancestor, fs::Permissions::from_mode(0o770)).expect("share state ancestor");
+        let error = prepare_state_directory_with_sync(&state, |_| panic!("unsafe ancestor must fail before syncing"))
+            .expect_err("shared ancestor must fail even with ready marker");
+        assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+        fs::set_permissions(&ancestor, fs::Permissions::from_mode(0o700)).expect("restore state ancestor");
+
+        let linked_ancestor = temp.path().join("linked-ancestor");
+        symlink(&ancestor, &linked_ancestor).expect("create ancestor symlink");
+        let error = prepare_state_directory_with_sync(&linked_ancestor.join("state"), |_| {
+            panic!("symlink ancestor must fail before syncing")
+        })
+        .expect_err("symlink ancestor must fail even with ready marker");
+        assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+    }
+
+    #[test]
+    fn concurrent_preparation_publishes_one_safe_ready_marker() {
+        let temp = secure_tempdir();
+        let state_parent = temp.path().join("preprovisioned");
+        fs::create_dir(&state_parent).expect("preprovision state parent");
+        fs::set_permissions(&state_parent, fs::Permissions::from_mode(0o700)).expect("secure state parent");
+        let state = state_parent.join("state");
+        let barrier = Arc::new(Barrier::new(2));
+
+        std::thread::scope(|scope| {
+            let handles = (0..2)
+                .map(|_| {
+                    let barrier = barrier.clone();
+                    let state = state.clone();
+                    scope.spawn(move || {
+                        let first_sync = Cell::new(true);
+                        prepare_state_directory_with_sync(&state, |path| {
+                            if first_sync.replace(false) {
+                                barrier.wait();
+                            }
+                            sync_directory(path)
+                        })
+                    })
+                })
+                .collect::<Vec<_>>();
+            for handle in handles {
+                let prepared = handle
+                    .join()
+                    .expect("preparation thread")
+                    .expect("concurrent preparation succeeds");
+                assert_eq!(prepared, state);
+            }
+        });
+
+        assert!(ready_marker_exists(&state.join(BOOTSTRAP_READY_FILE)).expect("validate ready marker"));
+        let staging = fs::read_dir(&state)
+            .expect("read state directory")
+            .filter_map(Result::ok)
+            .any(|entry| entry.file_name().to_string_lossy().ends_with(".tmp"));
+        assert!(!staging, "concurrent publication must not leave staging files");
+    }
 }
 
 #[cfg(all(test, not(unix)))]

rustfs/tests/connect_registration_bootstrap.rs

@@ -456,7 +456,11 @@ async fn endpoint_ca_token_state_and_service_failures_are_closed_and_sanitized()
         .await
         .expect_err("malformed token must fail");
     assert!(matches!(error, RegistrationBootstrapError::Token(_)));
-    assert!(!malformed_state.exists());
+    assert_eq!(
+        fs::read(malformed_state.join(".bootstrap-ready")).expect("read durable state marker"),
+        b"v1\n"
+    );
+    assert_no_staging_files(&malformed_state);
 
     let expired = temp.path().join("expired-token.json");
     write_file(&expired, &token_document(OffsetDateTime::now_utc().unix_timestamp() - 1), 0o600);
@@ -522,6 +526,44 @@ async fn token_ca_and_state_paths_reject_sharing_symlinks_and_non_files() {
     }
     fs::set_permissions(&root, fs::Permissions::from_mode(0o644)).expect("restore CA mode");
 
+    let missing_parent = temp.path().join("missing-parent");
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o640)).expect("make token unsafe behind parent gate");
+    let error = register_from_protected_input(endpoint, &root, &missing_parent.join("state"), Some(&token))
+        .await
+        .expect_err("missing state parent must fail before token access or network");
+    assert!(matches!(error, RegistrationBootstrapError::StateParentRequired));
+    assert!(!missing_parent.exists(), "bootstrap must not create the durable parent");
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o600)).expect("restore token after parent gate");
+
+    let parent_state = temp.path().join("secure/connect/..");
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o640)).expect("make token unsafe behind state gate");
+    let error = register_from_protected_input(endpoint, &root, &parent_state, Some(&token))
+        .await
+        .expect_err("parent state component must fail before token access or network");
+    assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+    assert!(!temp.path().join("secure").exists());
+    assert!(!parent_state.join(".bootstrap-ready").exists());
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o600)).expect("restore token after state gate");
+
+    let marker_state = temp.path().join("unsafe-marker-state");
+    for directory in [
+        marker_state.clone(),
+        marker_state.join("identity"),
+        marker_state.join("credential"),
+    ] {
+        fs::create_dir_all(&directory).expect("create marker state directory");
+        fs::set_permissions(&directory, fs::Permissions::from_mode(0o700)).expect("secure marker state directory");
+    }
+    let marker_target = temp.path().join("unsafe-marker-target");
+    write_file(&marker_target, b"v1\n", 0o600);
+    symlink(marker_target, marker_state.join(".bootstrap-ready")).expect("ready marker symlink");
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o640)).expect("make token unsafe behind marker gate");
+    let error = register_from_protected_input(endpoint, &root, &marker_state, Some(&token))
+        .await
+        .expect_err("unsafe marker must fail before token access or network");
+    assert!(matches!(error, RegistrationBootstrapError::StateMarkerSecurity));
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o600)).expect("restore token after marker gate");
+
     let shared_state = temp.path().join("shared-state");
     fs::create_dir(&shared_state).expect("shared state directory");
     fs::set_permissions(&shared_state, fs::Permissions::from_mode(0o770)).expect("make state group-writable");
@@ -616,6 +658,7 @@ fn assert_owner_only_files(state: &std::path::Path) {
 
     for path in [
         state.to_path_buf(),
+        state.join(".bootstrap-ready"),
         state.join("identity"),
         state.join("credential"),
         state.join("identity/device.key"),