From ba8f2e90bed571e5cf43cdf0383b50ca27d09419 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 23 Aug 2026 20:24:23 +0800 Subject: [PATCH] feat(connect): add registration bootstrap command (#6452) --- rustfs/src/config/cli.rs | 101 ++- rustfs/src/config/opt.rs | 21 +- rustfs/src/connect/mod.rs | 2 + rustfs/src/connect/registration_bootstrap.rs | 372 +++++++++++ rustfs/src/startup_entrypoint.rs | 12 + .../tests/connect_registration_bootstrap.rs | 626 ++++++++++++++++++ 6 files changed, 1125 insertions(+), 9 deletions(-) create mode 100644 rustfs/src/connect/registration_bootstrap.rs create mode 100644 rustfs/tests/connect_registration_bootstrap.rs diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 1e3436848..172101911 100644 --- a/rustfs/src/config/cli.rs +++ b/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 ` and `rustfs --address ...` are /// treated as `rustfs server ` 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, } /// 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")); + } } diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index ee861ebe6..e9dbbe42c 100644 --- a/rustfs/src/config/opt.rs +++ b/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(_)) => { - Self::from_server_opts(default_server_opts()) - } + Some(Commands::Info(_)) + | 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(_)) => { - Err(clap::Error::new(clap::error::ErrorKind::DisplayHelp)) - } + Some(Commands::Info(_)) + | 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())) diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 3ce2ad975..515dd13e8 100644 --- a/rustfs/src/connect/mod.rs +++ b/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}; diff --git a/rustfs/src/connect/registration_bootstrap.rs b/rustfs/src/connect/registration_bootstrap.rs new file mode 100644 index 000000000..30c071083 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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::>(); + 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 { + 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, 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 { + 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()); + } +} diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 8ffd53d71..cad258689 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/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, }; diff --git a/rustfs/tests/connect_registration_bootstrap.rs b/rustfs/tests/connect_registration_bootstrap.rs new file mode 100644 index 000000000..95a68b2ec --- /dev/null +++ b/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>>, + 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) -> 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| { + 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::>, 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 { + 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) -> 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) { + 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()); + } +}