feat(connect): add registration bootstrap command (#6452)

rustfs/src/config/cli.rs

@@ -56,7 +56,7 @@ pub(super) const LONG_VERSION: &str = concat!(
 );
 
 /// Known subcommands. When the first arg matches one of these, it is treated as a subcommand.
-pub const KNOWN_SUBCOMMANDS: &[&str] = &["server", "info", "tls", "diagnose", "inspect"];
+pub const KNOWN_SUBCOMMANDS: &[&str] = &["server", "info", "tls", "diagnose", "inspect", "connect"];
 
 /// Preprocess argv for legacy compatibility: `rustfs <volume>` and `rustfs --address ...` are
 /// treated as `rustfs server <volume>` and `rustfs server --address ...` respectively.
@@ -118,6 +118,42 @@ pub enum Commands {
     Diagnose(DiagnoseOpts),
     /// Offline, read-only inspection of on-disk data (no server required)
     Inspect(InspectOpts),
+    /// Configure outbound RustFS Connect integration
+    Connect(ConnectOpts),
+}
+
+/// RustFS Connect subcommand options
+#[derive(Args, Clone)]
+pub struct ConnectOpts {
+    #[command(subcommand)]
+    pub command: ConnectCommands,
+}
+
+/// Allow-listed RustFS Connect operations
+#[derive(Subcommand, Clone)]
+pub enum ConnectCommands {
+    /// Exchange a protected one-time token for a durable device credential (Unix only)
+    Register(ConnectRegisterOpts),
+}
+
+/// `connect register` options
+#[derive(Args, Clone)]
+pub struct ConnectRegisterOpts {
+    /// Connect agent API HTTPS base URL
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub endpoint: String,
+
+    /// PEM root CA file used only for this Connect endpoint
+    #[arg(long = "ca-file")]
+    pub ca_file: PathBuf,
+
+    /// Explicit directory shared with the Connect heartbeat runtime
+    #[arg(long = "state-dir")]
+    pub state_dir: PathBuf,
+
+    /// Owner-only regular token file; omit to read the token from stdin
+    #[arg(long = "token-file")]
+    pub token_file: Option<PathBuf>,
 }
 
 /// Offline inspection subcommand options
@@ -412,6 +448,8 @@ pub enum CommandResult {
     Diagnose(DiagnoseOpts),
     /// Inspect command with options
     Inspect(InspectOpts),
+    /// One-time Connect registration command
+    ConnectRegister(ConnectRegisterOpts),
 }
 
 /// Create default ServerOpts from environment variables
@@ -451,7 +489,7 @@ pub fn default_server_opts() -> ServerOpts {
 
 #[cfg(test)]
 mod tests {
-    use super::{Cli, Commands, InspectCommands, preprocess_args_for_legacy};
+    use super::{Cli, Commands, ConnectCommands, InspectCommands, preprocess_args_for_legacy};
     use clap::Parser;
     use clap::error::ErrorKind;
 
@@ -515,4 +553,63 @@ mod tests {
 
         assert_eq!(err.kind(), ErrorKind::MissingRequiredArgument);
     }
+
+    #[test]
+    fn connect_register_accepts_only_paths_and_endpoint_configuration() {
+        let cli = Cli::try_parse_from([
+            "rustfs",
+            "connect",
+            "register",
+            "--endpoint",
+            "https://connect.example/agent/",
+            "--ca-file",
+            "/etc/rustfs/connect-ca.pem",
+            "--state-dir",
+            "/var/lib/rustfs/connect",
+        ])
+        .expect("connect register arguments should parse");
+
+        let Some(Commands::Connect(connect)) = cli.command else {
+            panic!("connect command expected");
+        };
+        let ConnectCommands::Register(register) = connect.command;
+        assert_eq!(register.endpoint, "https://connect.example/agent/");
+        assert_eq!(register.ca_file, std::path::Path::new("/etc/rustfs/connect-ca.pem"));
+        assert_eq!(register.state_dir, std::path::Path::new("/var/lib/rustfs/connect"));
+        assert!(register.token_file.is_none());
+    }
+
+    #[test]
+    fn connect_register_has_no_token_value_or_environment_option() {
+        for forbidden in ["--token", "--registration-token", "--token-env"] {
+            let result = Cli::try_parse_from([
+                "rustfs",
+                "connect",
+                "register",
+                "--endpoint",
+                "https://connect.example/agent/",
+                "--ca-file",
+                "/etc/rustfs/connect-ca.pem",
+                "--state-dir",
+                "/var/lib/rustfs/connect",
+                forbidden,
+                "secret",
+            ]);
+            let Err(error) = result else {
+                panic!("secret-bearing command-line options must be rejected");
+            };
+            assert_eq!(error.kind(), ErrorKind::UnknownArgument);
+        }
+    }
+
+    #[test]
+    fn connect_register_help_states_the_unix_only_security_scope() {
+        let result = Cli::try_parse_from(["rustfs", "connect", "register", "--help"]);
+        let Err(help) = result else {
+            panic!("help exits without running registration");
+        };
+
+        assert_eq!(help.kind(), ErrorKind::DisplayHelp);
+        assert!(help.to_string().contains("Unix only"));
+    }
 }

rustfs/src/config/opt.rs

@@ -18,7 +18,7 @@
 //! and methods for parsing command line arguments.
 
 use super::Config;
-use super::cli::{Cli, CommandResult, Commands, ServerOpts, default_server_opts, preprocess_args_for_legacy};
+use super::cli::{Cli, CommandResult, Commands, ConnectCommands, ServerOpts, default_server_opts, preprocess_args_for_legacy};
 use crate::apply_external_env_compat;
 use CommandResult::Server;
 use clap::Parser;
@@ -98,9 +98,11 @@ impl Opt {
         let cli = Cli::parse_from(args);
         match cli.command {
             Some(Commands::Server(opts)) => Self::from_server_opts(*opts),
-            Some(Commands::Info(_)) | Some(Commands::Tls(_)) | Some(Commands::Diagnose(_)) | Some(Commands::Inspect(_)) => {
+            Some(Commands::Info(_))
-                Self::from_server_opts(default_server_opts())
+            | Some(Commands::Tls(_))
-            }
+            | Some(Commands::Diagnose(_))
+            | Some(Commands::Inspect(_))
+            | Some(Commands::Connect(_)) => Self::from_server_opts(default_server_opts()),
             None => {
                 // Default to server with empty volumes (will be filled from env)
                 Self::from_server_opts(default_server_opts())
@@ -135,6 +137,9 @@ impl Opt {
             Some(Commands::Tls(opts)) => Ok(CommandResult::Tls(opts)),
             Some(Commands::Diagnose(opts)) => Ok(CommandResult::Diagnose(opts)),
             Some(Commands::Inspect(opts)) => Ok(CommandResult::Inspect(opts)),
+            Some(Commands::Connect(opts)) => match opts.command {
+                ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)),
+            },
             Some(Commands::Server(opts)) => Self::server_command_result(Self::from_server_opts(*opts)),
             None => {
                 // Default to server with empty volumes (will be filled from env)
@@ -163,9 +168,11 @@ impl Opt {
         let cli = Cli::try_parse_from(args)?;
         match cli.command {
             Some(Commands::Server(opts)) => Ok(Self::from_server_opts(*opts)),
-            Some(Commands::Info(_)) | Some(Commands::Tls(_)) | Some(Commands::Diagnose(_)) | Some(Commands::Inspect(_)) => {
+            Some(Commands::Info(_))
-                Err(clap::Error::new(clap::error::ErrorKind::DisplayHelp))
+            | Some(Commands::Tls(_))
-            }
+            | Some(Commands::Diagnose(_))
+            | Some(Commands::Inspect(_))
+            | Some(Commands::Connect(_)) => Err(clap::Error::new(clap::error::ErrorKind::DisplayHelp)),
             None => {
                 // Default to server with empty volumes
                 Ok(Self::from_server_opts(default_server_opts()))

rustfs/src/connect/mod.rs

@@ -33,6 +33,7 @@ pub mod identity;
 pub mod identity_store;
 pub mod offline;
 pub mod registration;
+pub mod registration_bootstrap;
 pub mod runtime;
 
 pub use client::{ClientError, ConnectClient, ConnectConfig};
@@ -43,4 +44,5 @@ pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, Registratio
 pub use identity_store::{IdentityStore, StoreError};
 pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge};
 pub use registration::{RegistrationToken, TokenError};
+pub use registration_bootstrap::{RegistrationBootstrapError, RegistrationBootstrapResult, register_from_protected_input};
 pub use runtime::{HeartbeatRuntime, spawn_heartbeat_runtime};

rustfs/src/connect/registration_bootstrap.rs

@@ -0,0 +1,372 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io;
+use std::path::Path;
+#[cfg(unix)]
+use std::{
+    fs::{self, File, OpenOptions},
+    path::PathBuf,
+    time::Duration,
+};
+
+use super::TokenError;
+#[cfg(unix)]
+use super::{ConnectClient, ConnectConfig, CredentialStore, IdentityStore, RegistrationToken};
+
+#[cfg(unix)]
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
+
+#[derive(Debug, PartialEq, Eq)]
+pub struct RegistrationBootstrapResult {
+    pub device_uid: String,
+    pub cluster_name: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum RegistrationBootstrapError {
+    #[error("the Connect registration token file must be an owner-readable, owner-only regular file")]
+    TokenFileSecurity,
+    #[error("the Connect root CA file must be a trusted, non-shared-writable regular file")]
+    RootCaFileSecurity,
+    #[error("the Connect state path must be an explicit directory, not a symlink")]
+    StateDirectorySecurity,
+    #[error("failed to read protected Connect registration input")]
+    Input(#[source] io::Error),
+    #[error("Connect registration configuration is invalid")]
+    Configuration,
+    #[error("Connect registration exchange failed")]
+    Exchange,
+    #[error("Connect registration bootstrap requires Unix owner and permission guarantees")]
+    PlatformSecurity,
+    #[error(transparent)]
+    Token(#[from] TokenError),
+}
+
+#[cfg(not(unix))]
+pub async fn register_from_protected_input(
+    endpoint: &str,
+    root_ca_file: &Path,
+    state_directory: &Path,
+    token_file: Option<&Path>,
+) -> Result<RegistrationBootstrapResult, RegistrationBootstrapError> {
+    let _ = (endpoint, root_ca_file, state_directory, token_file);
+    Err(RegistrationBootstrapError::PlatformSecurity)
+}
+
+#[cfg(unix)]
+pub async fn register_from_protected_input(
+    endpoint: &str,
+    root_ca_file: &Path,
+    state_directory: &Path,
+    token_file: Option<&Path>,
+) -> Result<RegistrationBootstrapResult, RegistrationBootstrapError> {
+    let root_ca_pem = read_regular_file(root_ca_file, false)?;
+    let client = ConnectClient::new(ConnectConfig {
+        endpoint,
+        root_ca_pem: &root_ca_pem,
+        timeout: REQUEST_TIMEOUT,
+    })
+    .map_err(|_| RegistrationBootstrapError::Configuration)?;
+
+    let state_directory = prepare_state_directory(state_directory)?;
+    let token = match token_file {
+        Some(path) => RegistrationToken::from_reader(open_regular_file(path, true)?),
+        None => RegistrationToken::from_reader(io::stdin().lock()),
+    }?;
+    let cluster_name = format!("organizations/{}/clusters/{}", token.organization_uid, token.cluster_uid);
+    let credential = client
+        .register(
+            &IdentityStore::new(state_directory.join("identity")),
+            &CredentialStore::new(state_directory.join("credential")),
+            &token,
+        )
+        .await
+        .map_err(|_| RegistrationBootstrapError::Exchange)?;
+    if credential.name != format!("{cluster_name}/clusterDevices/{}", credential.uid) {
+        return Err(RegistrationBootstrapError::Exchange);
+    }
+
+    Ok(RegistrationBootstrapResult {
+        device_uid: credential.uid,
+        cluster_name,
+    })
+}
+
+#[cfg(unix)]
+fn prepare_state_directory(path: &Path) -> Result<PathBuf, RegistrationBootstrapError> {
+    prepare_state_directory_with_sync(path, sync_directory)
+}
+
+#[cfg(unix)]
+fn prepare_state_directory_with_sync(
+    path: &Path,
+    mut sync: impl FnMut(&Path) -> io::Result<()>,
+) -> Result<PathBuf, RegistrationBootstrapError> {
+    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<_>>();
+    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)?;
+        }
+    }
+    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)?;
+        }
+    }
+    for directory in &directories {
+        validate_directory(directory, directory == &path)?;
+    }
+    for directory in &store_directories {
+        validate_directory(directory, true)?;
+    }
+    Ok(path)
+}
+
+#[cfg(unix)]
+fn ensure_directory(path: &Path, require_process_owner: bool) -> Result<bool, RegistrationBootstrapError> {
+    match fs::symlink_metadata(path) {
+        Ok(_) => {
+            validate_directory(path, require_process_owner)?;
+            Ok(false)
+        }
+        Err(error) if error.kind() == io::ErrorKind::NotFound => {
+            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)
+                }
+                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
+                    Err(RegistrationBootstrapError::StateDirectorySecurity)
+                }
+                Err(error) => Err(RegistrationBootstrapError::Input(error)),
+            }
+        }
+        Err(error) => Err(RegistrationBootstrapError::Input(error)),
+    }
+}
+
+#[cfg(unix)]
+fn sync_directory(path: &Path) -> io::Result<()> {
+    File::open(path)?.sync_all()
+}
+
+#[cfg(unix)]
+fn validate_directory(path: &Path, require_process_owner: bool) -> Result<(), RegistrationBootstrapError> {
+    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
+
+    let metadata = fs::symlink_metadata(path).map_err(RegistrationBootstrapError::Input)?;
+    if metadata.file_type().is_symlink() || !metadata.is_dir() {
+        return Err(RegistrationBootstrapError::StateDirectorySecurity);
+    }
+    let mode = metadata.permissions().mode() & 0o777;
+    if !unix_directory_is_trusted(metadata.uid(), mode, process_uid(), require_process_owner) {
+        return Err(RegistrationBootstrapError::StateDirectorySecurity);
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn unix_directory_is_trusted(owner_uid: u32, mode: u32, process_uid: u32, require_process_owner: bool) -> bool {
+    (owner_uid == process_uid || (!require_process_owner && owner_uid == 0)) && mode & 0o022 == 0
+}
+
+#[cfg(unix)]
+// SAFETY: geteuid has no pointer arguments or caller preconditions.
+#[allow(unsafe_code)]
+fn process_uid() -> u32 {
+    unsafe { libc::geteuid() }
+}
+
+#[cfg(unix)]
+fn read_regular_file(path: &Path, owner_only: bool) -> Result<Vec<u8>, RegistrationBootstrapError> {
+    let mut file = open_regular_file(path, owner_only)?;
+    let mut contents = Vec::new();
+    io::Read::read_to_end(&mut file, &mut contents).map_err(RegistrationBootstrapError::Input)?;
+    Ok(contents)
+}
+
+#[cfg(unix)]
+fn open_regular_file(path: &Path, owner_only: bool) -> Result<File, RegistrationBootstrapError> {
+    let insecure = || {
+        if owner_only {
+            RegistrationBootstrapError::TokenFileSecurity
+        } else {
+            RegistrationBootstrapError::RootCaFileSecurity
+        }
+    };
+    let initial = fs::symlink_metadata(path).map_err(RegistrationBootstrapError::Input)?;
+    if initial.file_type().is_symlink() || !initial.is_file() {
+        return Err(insecure());
+    }
+
+    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 file = options.open(path).map_err(RegistrationBootstrapError::Input)?;
+    let metadata = file.metadata().map_err(RegistrationBootstrapError::Input)?;
+    if !metadata.is_file() {
+        return Err(insecure());
+    }
+    use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
+    let mode = metadata.permissions().mode() & 0o777;
+    if owner_only {
+        if metadata.uid() != process_uid() || mode & 0o400 == 0 || mode & 0o177 != 0 {
+            return Err(RegistrationBootstrapError::TokenFileSecurity);
+        }
+    } else if !unix_ca_file_is_trusted(metadata.uid(), mode, process_uid()) {
+        return Err(RegistrationBootstrapError::RootCaFileSecurity);
+    }
+    Ok(file)
+}
+
+#[cfg(unix)]
+fn unix_ca_file_is_trusted(owner_uid: u32, mode: u32, process_uid: u32) -> bool {
+    (owner_uid == process_uid || owner_uid == 0) && mode & 0o022 == 0
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use std::cell::{Cell, RefCell};
+    use std::fs;
+    use std::io;
+    use std::os::unix::fs::PermissionsExt as _;
+
+    use super::{
+        RegistrationBootstrapError, prepare_state_directory_with_sync, unix_ca_file_is_trusted, unix_directory_is_trusted,
+    };
+
+    #[test]
+    fn unix_directory_policy_rejects_wrong_owners_and_writable_modes_only() {
+        let process_uid = 501;
+
+        assert!(unix_directory_is_trusted(process_uid, 0o700, process_uid, true));
+        assert!(unix_directory_is_trusted(process_uid, 0o755, process_uid, true));
+        assert!(unix_directory_is_trusted(0, 0o755, process_uid, false));
+        assert!(!unix_directory_is_trusted(0, 0o755, process_uid, true));
+        assert!(!unix_directory_is_trusted(process_uid + 1, 0o700, process_uid, false));
+        assert!(!unix_directory_is_trusted(process_uid, 0o720, process_uid, true));
+        assert!(!unix_directory_is_trusted(process_uid, 0o702, process_uid, true));
+    }
+
+    #[test]
+    fn unix_ca_policy_accepts_only_process_or_root_owned_non_writable_files() {
+        let process_uid = 501;
+
+        assert!(unix_ca_file_is_trusted(process_uid, 0o600, process_uid));
+        assert!(unix_ca_file_is_trusted(0, 0o644, process_uid));
+        assert!(!unix_ca_file_is_trusted(process_uid + 1, 0o600, process_uid));
+        assert!(!unix_ca_file_is_trusted(process_uid, 0o620, process_uid));
+        assert!(!unix_ca_file_is_trusted(0, 0o646, process_uid));
+    }
+
+    #[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);
+
+        let error = 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");
+
+        assert!(matches!(error, RegistrationBootstrapError::Input(_)));
+        assert_eq!(
+            observed.into_inner(),
+            vec![
+                ancestor.clone(),
+                temp.path().to_path_buf(),
+                state.clone(),
+                ancestor,
+                state.join("identity"),
+                state.clone(),
+                state.join("credential"),
+                state,
+            ]
+        );
+    }
+
+    #[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");
+        }
+        let calls = Cell::new(0);
+
+        let prepared = prepare_state_directory_with_sync(&state, |_| {
+            calls.set(calls.get() + 1);
+            Err(io::Error::other("existing directories must not be synced"))
+        })
+        .expect("existing secure state tree should be ready");
+
+        assert_eq!(prepared, state);
+        assert_eq!(calls.get(), 0);
+    }
+}
+
+#[cfg(all(test, not(unix)))]
+mod non_unix_tests {
+    use std::path::Path;
+
+    use super::{RegistrationBootstrapError, register_from_protected_input};
+
+    #[tokio::test]
+    async fn bootstrap_fails_closed_for_stdin_and_token_files_without_unix_guarantees() {
+        let root = Path::new("unreadable-root-ca");
+        let state = Path::new("registration-bootstrap-must-not-create-state");
+        let token = Path::new("unreadable-token");
+        assert!(!state.exists());
+
+        for token_file in [None, Some(token)] {
+            let error = register_from_protected_input("https://connect.invalid/agent/", root, state, token_file)
+                .await
+                .expect_err("non-Unix bootstrap must fail closed");
+            assert!(matches!(error, RegistrationBootstrapError::PlatformSecurity));
+        }
+        assert!(!state.exists());
+    }
+}

rustfs/src/startup_entrypoint.rs

@@ -90,6 +90,18 @@ async fn async_main() -> Result<()> {
         // Inspect is offline like diagnose: read-only against drive paths, output
         // to stdout/--out, and must run before any observability/storage init.
         CommandResult::Inspect(opts) => return crate::inspect::execute_inspect(&opts).await,
+        CommandResult::ConnectRegister(opts) => {
+            let registered = crate::connect::register_from_protected_input(
+                &opts.endpoint,
+                &opts.ca_file,
+                &opts.state_dir,
+                opts.token_file.as_deref(),
+            )
+            .await
+            .map_err(Error::other)?;
+            println!("device={} cluster={}", registered.device_uid, registered.cluster_name);
+            return Ok(());
+        }
         CommandResult::Server(config) => config,
     };
 

rustfs/tests/connect_registration_bootstrap.rs

@@ -0,0 +1,626 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![cfg(unix)]
+
+use std::collections::VecDeque;
+use std::fs;
+use std::io::{self, Write as _};
+use std::process::{Command, Stdio};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use bytes::Bytes;
+use http_body_util::{BodyExt as _, Full};
+use hyper::service::service_fn;
+use hyper::{Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
+use rcgen::{
+    BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair,
+    KeyUsagePurpose, SanType, SerialNumber,
+};
+use rustfs::connect::{IdentityStore, RegistrationBootstrapError, register_from_protected_input};
+use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
+use serde_json::{Value, json};
+use time::OffsetDateTime;
+use time::format_description::well_known::Rfc3339;
+use tokio::net::TcpListener;
+use tokio_rustls::TlsAcceptor;
+
+const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70";
+const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81";
+const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92";
+const TOKEN_UID: &str = "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5";
+const TOKEN_SECRET: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+const REMOTE_REASON: &str = "remote-reason-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+
+struct TestPki {
+    root_params: CertificateParams,
+    root_key: KeyPair,
+    root_pem: String,
+    server_der: CertificateDer<'static>,
+    server_key: PrivatePkcs8KeyDer<'static>,
+}
+
+impl TestPki {
+    fn new() -> Self {
+        let now = OffsetDateTime::now_utc();
+        let root_key = KeyPair::generate().expect("generate root key");
+        let mut root_params = CertificateParams::default();
+        root_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
+        root_params.not_before = now - time::Duration::days(1);
+        root_params.not_after = now + time::Duration::days(30);
+        root_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::DigitalSignature];
+        root_params
+            .distinguished_name
+            .push(DnType::CommonName, "Connect bootstrap test root");
+        let root = root_params.self_signed(&root_key).expect("sign root");
+
+        let server_key = KeyPair::generate().expect("generate server key");
+        let mut server_params = CertificateParams::default();
+        server_params.not_before = now - time::Duration::hours(1);
+        server_params.not_after = now + time::Duration::days(2);
+        server_params
+            .subject_alt_names
+            .push(SanType::DnsName("localhost".try_into().expect("valid DNS name")));
+        server_params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
+        let issuer = Issuer::from_params(&root_params, &root_key);
+        let server = server_params
+            .signed_by(&server_key, &issuer)
+            .expect("sign server certificate");
+
+        Self {
+            root_params,
+            root_key,
+            root_pem: root.pem(),
+            server_der: server.der().clone(),
+            server_key: PrivatePkcs8KeyDer::from(server_key.serialize_der()),
+        }
+    }
+
+    fn server_config(&self) -> rustls::ServerConfig {
+        rustls::ServerConfig::builder()
+            .with_no_client_auth()
+            .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key()))
+            .expect("build server TLS")
+    }
+
+    fn credential(&self, identity: &rustfs::connect::DeviceIdentity) -> Value {
+        let now = OffsetDateTime::now_utc()
+            .replace_nanosecond(0)
+            .expect("whole-second test time");
+        let mut params = CertificateParams::default();
+        params.not_before = now;
+        params.not_after = now + time::Duration::days(1);
+        params.serial_number = Some(SerialNumber::from(vec![7; 16]));
+        params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
+        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
+        params.distinguished_name = DistinguishedName::new();
+        params.distinguished_name.push(DnType::CommonName, DEVICE_UID);
+        params.subject_alt_names = vec![SanType::URI(
+            format!("urn:rustfs:connect:device:{DEVICE_UID}")
+                .try_into()
+                .expect("valid URI SAN"),
+        )];
+        let private_key = identity.to_pkcs8_der().expect("serialize device key");
+        let private_key = PrivatePkcs8KeyDer::from(private_key.to_vec());
+        let device_key =
+            KeyPair::from_pkcs8_der_and_sign_algo(&private_key, &rcgen::PKCS_ECDSA_P256_SHA256).expect("parse device key");
+        let issuer = Issuer::from_params(&self.root_params, &self.root_key);
+        let certificate = params.signed_by(&device_key, &issuer).expect("sign device certificate");
+        let serial = "07".repeat(16);
+        let cluster = format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}");
+
+        json!({
+            "name": format!("{cluster}/clusterDevices/{DEVICE_UID}"),
+            "uid": DEVICE_UID,
+            "cluster": cluster,
+            "protocolVersion": "v1",
+            "keyId": format!("x509-{serial}"),
+            "certificateSerial": serial,
+            "certificate": certificate.pem(),
+            "certificateChain": certificate.pem(),
+            "notBefore": now.format(&Rfc3339).expect("format notBefore"),
+            "notAfter": (now + time::Duration::days(1)).format(&Rfc3339).expect("format notAfter"),
+        })
+    }
+}
+
+#[derive(Clone)]
+enum Reply {
+    Register,
+    RegisterAfter(Duration),
+    Reject(StatusCode, &'static str),
+    DropConnection,
+}
+
+struct TestServer {
+    endpoint: String,
+    root_pem: String,
+    seen: Arc<Mutex<Vec<Value>>>,
+    task: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+async fn server(state_directory: &std::path::Path, replies: Vec<Reply>) -> TestServer {
+    let pki = Arc::new(TestPki::new());
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind test server");
+    let address = listener.local_addr().expect("server address");
+    let acceptor = TlsAcceptor::from(Arc::new(pki.server_config()));
+    let replies = Arc::new(Mutex::new(VecDeque::from(replies)));
+    let seen = Arc::new(Mutex::new(Vec::new()));
+    let captured = seen.clone();
+    let root_pem = pki.root_pem.clone();
+    let identity_directory = state_directory.join("identity");
+    let task = tokio::spawn(async move {
+        loop {
+            let Ok((stream, _)) = listener.accept().await else {
+                return;
+            };
+            let acceptor = acceptor.clone();
+            let replies = replies.clone();
+            let seen = captured.clone();
+            let pki = pki.clone();
+            let identity_directory = identity_directory.clone();
+            tokio::spawn(async move {
+                let Ok(stream) = acceptor.accept(stream).await else {
+                    return;
+                };
+                let service = service_fn(move |request: Request<hyper::body::Incoming>| {
+                    let replies = replies.clone();
+                    let seen = seen.clone();
+                    let pki = pki.clone();
+                    let identity_directory = identity_directory.clone();
+                    async move {
+                        let body = request.into_body().collect().await.expect("read request body").to_bytes();
+                        let request: Value = serde_json::from_slice(&body).expect("request JSON");
+                        seen.lock().expect("seen lock").push(request.clone());
+                        let reply = replies.lock().expect("reply lock").pop_front().expect("planned reply");
+                        let (status, response) = match reply {
+                            Reply::Register => {
+                                let identity = IdentityStore::new(&identity_directory)
+                                    .load()
+                                    .expect("load bootstrap identity")
+                                    .expect("bootstrap identity exists before exchange");
+                                (StatusCode::CREATED, pki.credential(&identity))
+                            }
+                            Reply::RegisterAfter(delay) => {
+                                tokio::time::sleep(delay).await;
+                                let identity = IdentityStore::new(&identity_directory)
+                                    .load()
+                                    .expect("load bootstrap identity")
+                                    .expect("bootstrap identity exists before exchange");
+                                (StatusCode::CREATED, pki.credential(&identity))
+                            }
+                            Reply::Reject(status, reason) => (status, json!({"details": [{"reason": reason}]})),
+                            Reply::DropConnection => {
+                                return Err::<Response<Full<Bytes>>, io::Error>(io::Error::new(
+                                    io::ErrorKind::ConnectionAborted,
+                                    "planned response loss",
+                                ));
+                            }
+                        };
+                        Ok(Response::builder()
+                            .status(status)
+                            .header("content-type", "application/json")
+                            .body(Full::new(Bytes::from(serde_json::to_vec(&response).expect("reply JSON"))))
+                            .expect("response"))
+                    }
+                });
+                let _ = hyper::server::conn::http1::Builder::new()
+                    .serve_connection(TokioIo::new(stream), service)
+                    .await;
+            });
+        }
+    });
+
+    TestServer {
+        endpoint: format!("https://localhost:{}/agent/", address.port()),
+        root_pem,
+        seen,
+        task,
+    }
+}
+
+fn token_document(expires_unix: i64) -> Vec<u8> {
+    serde_json::to_vec(&json!({
+        "registrationTokenUid": TOKEN_UID,
+        "registrationTokenSecret": TOKEN_SECRET,
+        "organizationUid": ORGANIZATION_UID,
+        "clusterUid": CLUSTER_UID,
+        "challengeNonce": "a3f1c07d9b2e4856af0c1d3b5e7f9012c4a6b8d0e2f4061738495a6b7c8d9e0f",
+        "expiresUnix": expires_unix,
+    }))
+    .expect("token JSON")
+}
+
+fn write_file(path: &std::path::Path, bytes: &[u8], mode: u32) {
+    fs::write(path, bytes).expect("write test file");
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt as _;
+        fs::set_permissions(path, fs::Permissions::from_mode(mode)).expect("set test mode");
+    }
+    let _ = mode;
+}
+
+fn prepare_inputs(temp: &tempfile::TempDir, root_pem: &str) -> (std::path::PathBuf, std::path::PathBuf) {
+    let root = temp.path().join("root.pem");
+    let token = temp.path().join("token.json");
+    write_file(&root, root_pem.as_bytes(), 0o644);
+    write_file(&token, &token_document(OffsetDateTime::now_utc().unix_timestamp() + 3600), 0o600);
+    (root, token)
+}
+
+fn secure_tempdir() -> tempfile::TempDir {
+    tempfile::Builder::new()
+        .prefix(".connect-registration-")
+        .tempdir_in(env!("CARGO_MANIFEST_DIR"))
+        .expect("temporary directory inside the protected checkout")
+}
+
+fn run_binary(endpoint: String, root: std::path::PathBuf, state: std::path::PathBuf, token: Vec<u8>) -> std::process::Output {
+    let mut child = Command::new(env!("CARGO_BIN_EXE_rustfs"))
+        .args(["connect", "register", "--endpoint"])
+        .arg(endpoint)
+        .arg("--ca-file")
+        .arg(root)
+        .arg("--state-dir")
+        .arg(state)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("start production rustfs binary");
+    child
+        .stdin
+        .take()
+        .expect("command stdin")
+        .write_all(&token)
+        .expect("write protected token to stdin");
+    child.wait_with_output().expect("wait for registration command")
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn production_command_registers_once_and_emits_only_stable_identifiers() {
+    let temp = secure_tempdir();
+    let state = temp.path().join("state");
+    let server = server(&state, vec![Reply::Register]).await;
+    let (root, _) = prepare_inputs(&temp, &server.root_pem);
+    let token = token_document(OffsetDateTime::now_utc().unix_timestamp() + 3600);
+    let output = tokio::task::spawn_blocking({
+        let endpoint = server.endpoint.clone();
+        let state = state.clone();
+        move || run_binary(endpoint, root, state, token)
+    })
+    .await
+    .expect("registration process task");
+
+    assert!(output.status.success(), "stderr: {}", String::from_utf8_lossy(&output.stderr));
+    assert_eq!(
+        String::from_utf8(output.stdout).expect("UTF-8 stdout"),
+        format!("device={DEVICE_UID} cluster=organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}\n")
+    );
+    let stderr = String::from_utf8(output.stderr).expect("UTF-8 stderr");
+    assert!(stderr.is_empty());
+    assert!(!stderr.contains(TOKEN_SECRET));
+    assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
+    assert!(
+        IdentityStore::new(state.join("identity"))
+            .load()
+            .expect("load identity")
+            .is_some()
+    );
+    assert!(state.join("credential/device.crt.json").is_file());
+    assert!(!state.join("credential/registration.pending.json").exists());
+    assert_no_staging_files(&state);
+    #[cfg(unix)]
+    assert_owner_only_files(&state);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn production_command_never_echoes_a_remote_reason() {
+    let temp = secure_tempdir();
+    let state = temp.path().join("state");
+    let server = server(
+        &state,
+        vec![
+            Reply::Reject(StatusCode::BAD_REQUEST, REMOTE_REASON),
+            Reply::Reject(StatusCode::BAD_REQUEST, REMOTE_REASON),
+        ],
+    )
+    .await;
+    let (root, token_file) = prepare_inputs(&temp, &server.root_pem);
+    let error = register_from_protected_input(&server.endpoint, &root, &temp.path().join("direct-state"), Some(&token_file))
+        .await
+        .expect_err("remote rejection must fail");
+    assert_sanitized_error(&error, &[TOKEN_SECRET, REMOTE_REASON]);
+
+    let token = token_document(OffsetDateTime::now_utc().unix_timestamp() + 3600);
+    let output = tokio::task::spawn_blocking({
+        let endpoint = server.endpoint.clone();
+        let state = state.clone();
+        move || run_binary(endpoint, root, state, token)
+    })
+    .await
+    .expect("registration process task");
+
+    assert!(!output.status.success());
+    assert!(output.stdout.is_empty());
+    let stderr = String::from_utf8(output.stderr).expect("UTF-8 stderr");
+    assert_eq!(stderr, "[FATAL] Server runtime failed: Connect registration exchange failed\n");
+    for forbidden in [TOKEN_SECRET, REMOTE_REASON] {
+        assert!(!stderr.contains(forbidden));
+    }
+    assert_eq!(server.seen.lock().expect("seen lock").len(), 2);
+}
+
+#[tokio::test]
+async fn response_loss_reuses_the_pending_request_and_existing_credential_is_idempotent() {
+    let temp = secure_tempdir();
+    let state = temp.path().join("state");
+    let server = server(
+        &state,
+        vec![
+            Reply::DropConnection,
+            Reply::DropConnection,
+            Reply::DropConnection,
+            Reply::Register,
+        ],
+    )
+    .await;
+    let (root, token) = prepare_inputs(&temp, &server.root_pem);
+
+    while server.seen.lock().expect("seen lock").len() < 3 {
+        let error = register_from_protected_input(&server.endpoint, &root, &state, Some(&token))
+            .await
+            .expect_err("lost response must leave a retryable failure");
+        assert!(!error.to_string().contains(TOKEN_SECRET));
+        assert!(state.join("credential/registration.pending.json").is_file());
+    }
+
+    let registered = register_from_protected_input(&server.endpoint, &root, &state, Some(&token))
+        .await
+        .expect("retry registration");
+    assert_eq!(registered.device_uid, DEVICE_UID);
+    let requests = server.seen.lock().expect("seen lock");
+    assert_eq!(requests.len(), 4);
+    for request in &requests[1..] {
+        assert_eq!(request["requestId"], requests[0]["requestId"]);
+        assert_eq!(request["certificateRequest"], requests[0]["certificateRequest"]);
+    }
+    drop(requests);
+
+    let idle = server(&state, vec![]).await;
+    let idempotent = register_from_protected_input(&idle.endpoint, &root, &state, Some(&token))
+        .await
+        .expect("valid stored credential is idempotent");
+    assert_eq!(idempotent, registered);
+    assert!(idle.seen.lock().expect("seen lock").is_empty());
+    assert!(!state.join("credential/registration.pending.json").exists());
+    assert_no_staging_files(&state);
+}
+
+#[tokio::test]
+async fn concurrent_bootstraps_share_one_identity_and_one_exchange() {
+    let temp = secure_tempdir();
+    let state = temp.path().join("state");
+    let server = server(&state, vec![Reply::RegisterAfter(Duration::from_millis(150))]).await;
+    let (root, token) = prepare_inputs(&temp, &server.root_pem);
+
+    let (first, second) = tokio::join!(
+        register_from_protected_input(&server.endpoint, &root, &state, Some(&token)),
+        register_from_protected_input(&server.endpoint, &root, &state, Some(&token)),
+    );
+    assert_eq!(first.expect("first registration"), second.expect("second registration"));
+    assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
+    assert_no_staging_files(&state);
+}
+
+#[tokio::test]
+async fn endpoint_ca_token_state_and_service_failures_are_closed_and_sanitized() {
+    let temp = secure_tempdir();
+    let server = server(temp.path(), vec![Reply::Reject(StatusCode::BAD_REQUEST, "REGISTRATION_TOKEN_EXPIRED")]).await;
+    let (root, token) = prepare_inputs(&temp, &server.root_pem);
+
+    let http_state = temp.path().join("http-state");
+    let error = register_from_protected_input("http://localhost/agent/", &root, &http_state, Some(&token))
+        .await
+        .expect_err("HTTP endpoint must fail");
+    assert!(matches!(error, RegistrationBootstrapError::Configuration));
+    assert!(!http_state.exists());
+
+    let malformed = temp.path().join("malformed-token.json");
+    write_file(&malformed, b"not a token", 0o600);
+    let malformed_state = temp.path().join("malformed-state");
+    let error = register_from_protected_input(&server.endpoint, &root, &malformed_state, Some(&malformed))
+        .await
+        .expect_err("malformed token must fail");
+    assert!(matches!(error, RegistrationBootstrapError::Token(_)));
+    assert!(!malformed_state.exists());
+
+    let expired = temp.path().join("expired-token.json");
+    write_file(&expired, &token_document(OffsetDateTime::now_utc().unix_timestamp() - 1), 0o600);
+    let rejected_state = temp.path().join("rejected-state");
+    let error = register_from_protected_input(&server.endpoint, &root, &rejected_state, Some(&expired))
+        .await
+        .expect_err("expired token must be refused by Connect");
+    assert!(matches!(&error, RegistrationBootstrapError::Exchange));
+    assert_eq!(error.to_string(), "Connect registration exchange failed");
+    assert_sanitized_error(&error, &[TOKEN_SECRET, "REGISTRATION_TOKEN_EXPIRED"]);
+    assert!(!rejected_state.join("credential/device.crt.json").exists());
+    assert!(!rejected_state.join("credential/registration.pending.json").exists());
+
+    let other_pki = TestPki::new();
+    let wrong_root = temp.path().join("wrong-root.pem");
+    write_file(&wrong_root, other_pki.root_pem.as_bytes(), 0o644);
+    let wrong_ca_state = temp.path().join("wrong-ca-state");
+    let error = register_from_protected_input(&server.endpoint, &wrong_root, &wrong_ca_state, Some(&token))
+        .await
+        .expect_err("wrong CA must fail TLS");
+    assert!(matches!(error, RegistrationBootstrapError::Exchange));
+    assert!(!wrong_ca_state.join("credential/device.crt.json").exists());
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn token_ca_and_state_paths_reject_sharing_symlinks_and_non_files() {
+    use std::os::unix::fs::{PermissionsExt as _, symlink};
+
+    let pki = TestPki::new();
+    let temp = secure_tempdir();
+    let (root, token) = prepare_inputs(&temp, &pki.root_pem);
+    let endpoint = "https://localhost:1/agent/";
+
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o640)).expect("share token mode");
+    let error = register_from_protected_input(endpoint, &root, &temp.path().join("shared"), Some(&token))
+        .await
+        .expect_err("group-readable token must fail");
+    assert!(matches!(error, RegistrationBootstrapError::TokenFileSecurity));
+
+    fs::set_permissions(&token, fs::Permissions::from_mode(0o600)).expect("restore token mode");
+    let token_link = temp.path().join("token-link");
+    symlink(&token, &token_link).expect("token symlink");
+    let error = register_from_protected_input(endpoint, &root, &temp.path().join("token-link-state"), Some(&token_link))
+        .await
+        .expect_err("token symlink must fail");
+    assert!(matches!(error, RegistrationBootstrapError::TokenFileSecurity));
+
+    let root_link = temp.path().join("root-link");
+    symlink(&root, &root_link).expect("root symlink");
+    let error = register_from_protected_input(endpoint, &root_link, &temp.path().join("root-link-state"), Some(&token))
+        .await
+        .expect_err("CA symlink must fail");
+    assert!(matches!(error, RegistrationBootstrapError::RootCaFileSecurity));
+
+    for mode in [0o664, 0o646] {
+        fs::set_permissions(&root, fs::Permissions::from_mode(mode)).expect("make CA writable by another user");
+        let error =
+            register_from_protected_input(endpoint, &root, &temp.path().join(format!("writable-ca-{mode:o}")), Some(&token))
+                .await
+                .expect_err("group/world-writable CA must fail");
+        assert!(matches!(error, RegistrationBootstrapError::RootCaFileSecurity));
+    }
+    fs::set_permissions(&root, fs::Permissions::from_mode(0o644)).expect("restore CA mode");
+
+    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");
+    let error = register_from_protected_input(endpoint, &root, &shared_state, Some(&token))
+        .await
+        .expect_err("shared-writable state must fail");
+    assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+
+    let shared_ancestor = temp.path().join("shared-ancestor");
+    fs::create_dir(&shared_ancestor).expect("shared ancestor directory");
+    fs::set_permissions(&shared_ancestor, fs::Permissions::from_mode(0o770)).expect("make ancestor group-writable");
+    let error = register_from_protected_input(endpoint, &root, &shared_ancestor.join("state"), Some(&token))
+        .await
+        .expect_err("shared-writable state ancestor must fail");
+    assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+
+    let ancestor_target = temp.path().join("ancestor-target");
+    fs::create_dir(&ancestor_target).expect("ancestor target");
+    let ancestor_link = temp.path().join("ancestor-link");
+    symlink(&ancestor_target, &ancestor_link).expect("ancestor symlink");
+    let error = register_from_protected_input(endpoint, &root, &ancestor_link.join("state"), Some(&token))
+        .await
+        .expect_err("state ancestor symlink must fail");
+    assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+
+    for nested in ["identity", "credential"] {
+        let state = temp.path().join(format!("nested-{nested}"));
+        fs::create_dir(&state).expect("state directory");
+        let target = temp.path().join(format!("{nested}-target"));
+        fs::create_dir(&target).expect("nested target");
+        symlink(&target, state.join(nested)).expect("nested store symlink");
+        let error = register_from_protected_input(endpoint, &root, &state, Some(&token))
+            .await
+            .expect_err("nested store symlink must fail");
+        assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+    }
+
+    for (nested, mode) in [("identity", 0o720), ("credential", 0o702)] {
+        let state = temp.path().join(format!("writable-{nested}"));
+        fs::create_dir(&state).expect("state directory");
+        let nested = state.join(nested);
+        fs::create_dir(&nested).expect("nested store directory");
+        fs::set_permissions(&nested, fs::Permissions::from_mode(mode)).expect("make nested store writable");
+        let error = register_from_protected_input(endpoint, &root, &state, Some(&token))
+            .await
+            .expect_err("writable nested store must fail");
+        assert!(matches!(error, RegistrationBootstrapError::StateDirectorySecurity));
+    }
+}
+
+fn assert_sanitized_error(error: &(dyn std::error::Error + 'static), forbidden: &[&str]) {
+    let mut current = Some(error);
+    while let Some(candidate) = current {
+        let display = candidate.to_string();
+        let debug = format!("{candidate:?}");
+        for forbidden in forbidden {
+            assert!(!display.contains(forbidden), "error Display leaked forbidden text");
+            assert!(!debug.contains(forbidden), "error Debug leaked forbidden text");
+        }
+        current = candidate.source();
+    }
+}
+
+fn assert_no_staging_files(root: &std::path::Path) {
+    fn visit(path: &std::path::Path, found: &mut Vec<String>) {
+        let Ok(entries) = fs::read_dir(path) else {
+            return;
+        };
+        for entry in entries {
+            let entry = entry.expect("directory entry");
+            let path = entry.path();
+            if path.is_dir() {
+                visit(&path, found);
+            } else if path
+                .file_name()
+                .and_then(|name| name.to_str())
+                .is_some_and(|name| name.starts_with('.') && name.ends_with(".tmp"))
+            {
+                found.push(path.display().to_string());
+            }
+        }
+    }
+
+    let mut found = Vec::new();
+    visit(root, &mut found);
+    assert!(found.is_empty(), "staging files remained: {found:?}");
+}
+
+#[cfg(unix)]
+fn assert_owner_only_files(state: &std::path::Path) {
+    use std::os::unix::fs::PermissionsExt as _;
+
+    for path in [
+        state.to_path_buf(),
+        state.join("identity"),
+        state.join("credential"),
+        state.join("identity/device.key"),
+        state.join("credential/device.crt.json"),
+        state.join("credential/.state.lock"),
+    ] {
+        let mode = fs::metadata(&path).expect("stored file metadata").permissions().mode() & 0o777;
+        let expected = if path.is_dir() { 0o700 } else { 0o600 };
+        assert_eq!(mode, expected, "unexpected mode for {}", path.display());
+    }
+}