From ba629bdae041d5df928d09d083c6f996edf82775 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 26 Aug 2026 09:34:57 +0800 Subject: [PATCH] feat(connect): rotate device credentials at runtime (#6586) * feat(connect): rotate credentials from heartbeat runtime * fix(connect): make rotation safe with in-flight telemetry * fix(connect): keep rotation retry state private * fix(connect): preserve public rotation retries * fix(connect): preserve heartbeat error API * fix(connect): keep heartbeat alive during reenrollment * fix(connect): validate pending reenrollment before skipping * fix(connect): validate pending reenrollment token * fix(connect): bind pending reenrollment state * fix(connect): recover credentials before telemetry --- rustfs/src/connect/client.rs | 216 +++++-- rustfs/src/connect/credential_store.rs | 1 + rustfs/src/connect/heartbeat.rs | 1 + rustfs/src/connect/identity.rs | 20 +- rustfs/src/connect/registration.rs | 2 +- rustfs/src/connect/runtime.rs | 77 +++ rustfs/src/connect/telemetry.rs | 131 ++-- rustfs/tests/connect_registration.rs | 796 ++++++++++++++++++++++++- 8 files changed, 1145 insertions(+), 99 deletions(-) diff --git a/rustfs/src/connect/client.rs b/rustfs/src/connect/client.rs index a13ec953d..39191c7ca 100644 --- a/rustfs/src/connect/client.rs +++ b/rustfs/src/connect/client.rs @@ -15,7 +15,8 @@ use std::time::Duration; use base64::Engine as _; -use reqwest::{Client, StatusCode, Url}; +use chrono::{DateTime, Utc}; +use reqwest::{Client, StatusCode, Url, header}; use rustls::RootCertStore; use rustls::pki_types::{CertificateDer, pem::PemObject as _}; use serde::Deserialize; @@ -23,19 +24,21 @@ use uuid::Uuid; use zeroize::Zeroizing; use super::credential_store::{ - CompletedRegistration, CredentialStore, CredentialStoreError, DeviceCredential, PendingRegistration, PendingRotation, + CompletedRegistration, CredentialLock, CredentialStore, CredentialStoreError, DeviceCredential, PendingRegistration, + PendingRotation, }; use super::identity::{IdentityError, RegistrationTranscript}; use super::identity_store::{IdentityStore, StoreError}; use super::registration::{ CredentialResponse, CredentialValidationError, ExpectedDevice, RegistrationRequest, RegistrationToken, RotationRequest, - certificate_fingerprint, certificate_request_matches, public_key_fingerprint, validate_credential, + certificate_fingerprint, certificate_request_matches, is_uuid_v7, public_key_fingerprint, validate_credential, validate_stored_credential, }; const MAX_ATTEMPTS: usize = 3; const MAX_RESPONSE_BYTES: usize = 1024 * 1024; const ROTATION_THRESHOLD_SECONDS: i64 = 8 * 60 * 60; +const PENDING_REGISTRATION_STATE_DOMAIN: &[u8] = b"RUSTFS-CONNECT-PENDING-REGISTRATION-V1"; pub struct ConnectConfig<'a> { pub endpoint: &'a str, @@ -51,6 +54,23 @@ pub struct ConnectClient { timeout: Duration, } +pub(crate) enum RotationAttempt { + Completed(Option), + ReenrollmentPending, + Unavailable { + status: Option, + retry_after: Option, + }, +} + +enum SingleRequest { + Response(CredentialResponse), + Unavailable { + status: Option, + retry_after: Option, + }, +} + impl ConnectClient { pub fn from_optional_config(config: Option>) -> Result, ClientError> { config.map(Self::new).transpose() @@ -100,26 +120,31 @@ impl ConnectClient { credential_store: &CredentialStore, token: &RegistrationToken, ) -> Result { - let _lock = credential_store.lock().await?; - if let Some((credential, _)) = self.load_valid_credential(identity_store, credential_store)? { + let lock = credential_store.lock().await?; + if let Some((credential, _, _)) = + Self::recover_valid_credential_locked(&lock, identity_store, credential_store, &self.roots, &self.root_certificates)? + { ensure_credential_time(&credential, unix_now())?; return Ok(credential); } let identity = identity_store.load_or_create()?; - let candidate = PendingRegistration { + let mut candidate = PendingRegistration { token_uid: token.registration_token_uid.clone(), request_id: Uuid::new_v4().to_string(), certificate_request: identity.certificate_request_base64()?, previous_credential_fingerprint: None, next_public_key_sha256: None, + state_proof: String::new(), }; + candidate.state_proof = identity.sign_pending_registration_state(&pending_registration_state(&candidate)); let pending = credential_store.claim_pending_registration(&candidate)?; if pending.token_uid != token.registration_token_uid || pending.previous_credential_fingerprint.is_some() || pending.next_public_key_sha256.is_some() || !is_request_id(&pending.request_id) || !certificate_request_matches(&pending.certificate_request, &identity)? + || !pending_registration_is_bound(&pending, &identity) { return Err(ClientError::PendingRegistration); } @@ -143,10 +168,10 @@ impl ConnectClient { credential_store: &CredentialStore, token: &RegistrationToken, ) -> Result { - let _lock = credential_store.lock().await?; - let (credential, _) = self - .load_valid_credential(identity_store, credential_store)? - .ok_or(ClientError::NotRegistered)?; + let lock = credential_store.lock().await?; + let (credential, _, _) = + Self::recover_valid_credential_locked(&lock, identity_store, credential_store, &self.roots, &self.root_certificates)? + .ok_or(ClientError::NotRegistered)?; let fingerprint = certificate_fingerprint(&credential.certificate)?; if credential_store.load_completed_registration()?.is_some_and(|completed| { completed.token_uid == token.registration_token_uid && completed.credential_fingerprint == fingerprint @@ -156,19 +181,22 @@ impl ConnectClient { credential_store.clear_pending_rotation()?; let next = identity_store.load_or_create_next()?; let next_fingerprint = public_key_fingerprint(&next); - let candidate = PendingRegistration { + let mut candidate = PendingRegistration { token_uid: token.registration_token_uid.clone(), request_id: Uuid::new_v4().to_string(), certificate_request: next.certificate_request_base64()?, previous_credential_fingerprint: Some(fingerprint.clone()), next_public_key_sha256: Some(next_fingerprint.clone()), + state_proof: String::new(), }; + candidate.state_proof = next.sign_pending_registration_state(&pending_registration_state(&candidate)); let pending = credential_store.claim_pending_registration(&candidate)?; if pending.token_uid != token.registration_token_uid || pending.previous_credential_fingerprint.as_deref() != Some(&fingerprint) || pending.next_public_key_sha256.as_deref() != Some(&next_fingerprint) || !is_request_id(&pending.request_id) || !certificate_request_matches(&pending.certificate_request, &next)? + || !pending_registration_is_bound(&pending, &next) { return Err(ClientError::PendingRegistration); } @@ -232,16 +260,39 @@ impl ConnectClient { credential_store: &CredentialStore, now_unix: i64, ) -> Result, ClientError> { - let _lock = credential_store.lock().await?; - let (credential, identity) = self - .load_valid_credential(identity_store, credential_store)? - .ok_or(ClientError::NotRegistered)?; - if credential_store.load_pending_registration()?.is_some() { - return Err(ClientError::PendingRegistration); + let mut last_status = None; + for attempt in 0..MAX_ATTEMPTS { + match self.rotate_if_due_once(identity_store, credential_store, now_unix).await? { + RotationAttempt::Completed(credential) => return Ok(credential), + RotationAttempt::ReenrollmentPending => return Err(ClientError::PendingRegistration), + RotationAttempt::Unavailable { + status: Some(status), .. + } => last_status = Some(status), + RotationAttempt::Unavailable { status: None, .. } => {} + } + if attempt + 1 < MAX_ATTEMPTS { + tokio::time::sleep(Duration::from_millis(50 * (attempt as u64 + 1))).await; + } + } + Err(ClientError::Unavailable { status: last_status }) + } + + pub(crate) async fn rotate_if_due_once( + &self, + identity_store: &IdentityStore, + credential_store: &CredentialStore, + now_unix: i64, + ) -> Result { + let lock = credential_store.lock().await?; + let (credential, identity, reenrollment_pending) = + Self::recover_valid_credential_locked(&lock, identity_store, credential_store, &self.roots, &self.root_certificates)? + .ok_or(ClientError::NotRegistered)?; + if reenrollment_pending { + return Ok(RotationAttempt::ReenrollmentPending); } ensure_credential_time(&credential, now_unix)?; if credential.not_after_unix - now_unix > ROTATION_THRESHOLD_SECONDS { - return Ok(None); + return Ok(RotationAttempt::Completed(None)); } let fingerprint = certificate_fingerprint(&credential.certificate)?; let next = identity_store.load_or_create_next()?; @@ -277,7 +328,12 @@ impl ConnectClient { let client = build_client(&self.root_certificates, self.timeout, Some(tls_identity))?; let path = format!("clusterDevices/{}:rotateCredential", credential.uid); let url = self.url(&path)?; - let response = self.send(StatusCode::OK, || client.post(url.clone()).json(&body)).await?; + let response = match self.send_once(StatusCode::OK, client.post(url).json(&body)).await? { + SingleRequest::Response(response) => response, + SingleRequest::Unavailable { status, retry_after } => { + return Ok(RotationAttempt::Unavailable { status, retry_after }); + } + }; let rotated = validate_credential( response, &next, @@ -291,29 +347,35 @@ impl ConnectClient { credential_store.save(&rotated)?; identity_store.commit_next(&next)?; credential_store.clear_pending_rotation()?; - Ok(Some(rotated)) + Ok(RotationAttempt::Completed(Some(rotated))) } - fn load_valid_credential( - &self, + pub(crate) fn recover_valid_credential_locked( + _lock: &CredentialLock, identity_store: &IdentityStore, credential_store: &CredentialStore, - ) -> Result, ClientError> { + roots: &RootCertStore, + root_certificates: &[CertificateDer<'static>], + ) -> Result, ClientError> { let Some(credential) = credential_store.load()? else { return Ok(None); }; let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?; if let Some(pending) = credential_store.load_pending_registration()? { + if !is_uuid_v7(&pending.token_uid) { + return Err(ClientError::PendingRegistration); + } let Some(previous) = pending.previous_credential_fingerprint.as_deref() else { if pending.next_public_key_sha256.is_some() || !is_request_id(&pending.request_id) || !certificate_request_matches(&pending.certificate_request, ¤t)? + || !pending_registration_is_bound(&pending, ¤t) { return Err(ClientError::PendingRegistration); } - validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + validate_stored_credential(&credential, ¤t, roots, root_certificates)?; credential_store.clear_pending_registration()?; - return Ok(Some((credential, current))); + return Ok(Some((credential, current, false))); }; let next_fingerprint = pending .next_public_key_sha256 @@ -324,28 +386,38 @@ impl ConnectClient { } let fingerprint = certificate_fingerprint(&credential.certificate)?; if fingerprint == previous { - validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + validate_stored_credential(&credential, ¤t, roots, root_certificates)?; let next = identity_store.load_next()?.ok_or(ClientError::PendingRegistration)?; if public_key_fingerprint(&next) != next_fingerprint || !certificate_request_matches(&pending.certificate_request, &next)? + || !pending_registration_is_bound(&pending, &next) { return Err(ClientError::PendingRegistration); } - return Ok(Some((credential, current))); - } - if public_key_fingerprint(¤t) == next_fingerprint { - if !certificate_request_matches(&pending.certificate_request, ¤t)? { + if credential_store.load_completed_registration()?.is_some_and(|completed| { + completed.credential_fingerprint == fingerprint + && (!is_uuid_v7(&completed.token_uid) || completed.token_uid == pending.token_uid) + }) { return Err(ClientError::PendingRegistration); } - validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + return Ok(Some((credential, current, true))); + } + if public_key_fingerprint(¤t) == next_fingerprint { + if !certificate_request_matches(&pending.certificate_request, ¤t)? + || !pending_registration_is_bound(&pending, ¤t) + { + return Err(ClientError::PendingRegistration); + } + validate_stored_credential(&credential, ¤t, roots, root_certificates)?; } else { let next = identity_store.load_next()?.ok_or(ClientError::PendingRegistration)?; if public_key_fingerprint(&next) != next_fingerprint || !certificate_request_matches(&pending.certificate_request, &next)? + || !pending_registration_is_bound(&pending, &next) { return Err(ClientError::PendingRegistration); } - validate_stored_credential(&credential, &next, &self.roots, &self.root_certificates)?; + validate_stored_credential(&credential, &next, roots, root_certificates)?; identity_store.commit_next(&next)?; } credential_store.save_completed_registration(&CompletedRegistration { @@ -354,32 +426,32 @@ impl ConnectClient { })?; credential_store.clear_pending_registration()?; let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?; - return Ok(Some((credential, current))); + return Ok(Some((credential, current, false))); } let Some(pending) = credential_store.load_pending_rotation()? else { - validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; - return Ok(Some((credential, current))); + validate_stored_credential(&credential, ¤t, roots, root_certificates)?; + return Ok(Some((credential, current, false))); }; let fingerprint = certificate_fingerprint(&credential.certificate)?; if pending.device_name != credential.name || !is_request_id(&pending.request_id) { return Err(ClientError::PendingRotation); } if fingerprint == pending.credential_fingerprint { - validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + validate_stored_credential(&credential, ¤t, roots, root_certificates)?; let next = identity_store.load_next()?.ok_or(ClientError::PendingRotation)?; if pending.next_public_key_sha256 != public_key_fingerprint(&next) || !certificate_request_matches(&pending.certificate_request, &next)? { return Err(ClientError::PendingRotation); } - return Ok(Some((credential, current))); + return Ok(Some((credential, current, false))); } if public_key_fingerprint(¤t) == pending.next_public_key_sha256 { if !certificate_request_matches(&pending.certificate_request, ¤t)? { return Err(ClientError::PendingRotation); } - validate_stored_credential(&credential, ¤t, &self.roots, &self.root_certificates)?; + validate_stored_credential(&credential, ¤t, roots, root_certificates)?; } else { let next = identity_store.load_next()?.ok_or(ClientError::PendingRotation)?; if public_key_fingerprint(&next) != pending.next_public_key_sha256 @@ -387,12 +459,12 @@ impl ConnectClient { { return Err(ClientError::PendingRotation); } - validate_stored_credential(&credential, &next, &self.roots, &self.root_certificates)?; + validate_stored_credential(&credential, &next, roots, root_certificates)?; identity_store.commit_next(&next)?; } credential_store.clear_pending_rotation()?; let current = identity_store.load()?.ok_or(ClientError::IdentityMissing)?; - Ok(Some((credential, current))) + Ok(Some((credential, current, false))) } async fn send(&self, success: StatusCode, mut request: F) -> Result @@ -435,6 +507,35 @@ impl ConnectClient { Err(ClientError::Unavailable { status: last_status }) } + async fn send_once(&self, success: StatusCode, request: reqwest::RequestBuilder) -> Result { + let response = match request.send().await { + Ok(response) => response, + Err(error) if error.is_timeout() || error.is_connect() => { + return Ok(SingleRequest::Unavailable { + status: None, + retry_after: None, + }); + } + Err(error) => return Err(ClientError::Transport(error)), + }; + let status = response.status(); + if status == success { + return decode_response(response).await.map(SingleRequest::Response); + } + if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { + let reason = decode_reason(response).await; + return Err(ClientError::AccessRevoked { status, reason }); + } + if matches!(status, StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS) || status.is_server_error() { + return Ok(SingleRequest::Unavailable { + status: Some(status), + retry_after: retry_after(response.headers(), Utc::now()), + }); + } + let reason = decode_reason(response).await; + Err(ClientError::Rejected { status, reason }) + } + fn url(&self, path: &str) -> Result { self.endpoint.join(path).map_err(|_| ClientError::Endpoint) } @@ -444,6 +545,32 @@ fn is_request_id(value: &str) -> bool { Uuid::parse_str(value).is_ok_and(|uuid| uuid.get_version() == Some(uuid::Version::Random) && uuid.to_string() == value) } +fn pending_registration_is_bound(pending: &PendingRegistration, identity: &super::identity::DeviceIdentity) -> bool { + identity.verifies_pending_registration_state(&pending_registration_state(pending), &pending.state_proof) +} + +fn pending_registration_state(pending: &PendingRegistration) -> Vec { + let mut state = Vec::with_capacity(PENDING_REGISTRATION_STATE_DOMAIN.len() + 512); + state.extend_from_slice(PENDING_REGISTRATION_STATE_DOMAIN); + for field in [ + Some(pending.token_uid.as_str()), + Some(pending.request_id.as_str()), + Some(pending.certificate_request.as_str()), + pending.previous_credential_fingerprint.as_deref(), + pending.next_public_key_sha256.as_deref(), + ] { + match field { + Some(value) => { + state.push(1); + state.extend_from_slice(&(value.len() as u64).to_be_bytes()); + state.extend_from_slice(value.as_bytes()); + } + None => state.push(0), + } + } + state +} + fn unix_now() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -460,6 +587,15 @@ fn ensure_credential_time(credential: &DeviceCredential, now_unix: i64) -> Resul Ok(()) } +fn retry_after(headers: &header::HeaderMap, now: DateTime) -> Option { + let value = headers.get(header::RETRY_AFTER)?.to_str().ok()?; + value.parse::().ok().map(Duration::from_secs).or_else(|| { + DateTime::parse_from_rfc2822(value) + .ok() + .and_then(|at| (at.with_timezone(&Utc) - now).to_std().ok()) + }) +} + fn build_client( roots: &[CertificateDer<'static>], timeout: Duration, diff --git a/rustfs/src/connect/credential_store.rs b/rustfs/src/connect/credential_store.rs index 863ae1096..be9ff137d 100644 --- a/rustfs/src/connect/credential_store.rs +++ b/rustfs/src/connect/credential_store.rs @@ -66,6 +66,7 @@ pub(crate) struct PendingRegistration { pub previous_credential_fingerprint: Option, #[serde(default)] pub next_public_key_sha256: Option, + pub state_proof: String, } #[derive(Serialize, Deserialize)] diff --git a/rustfs/src/connect/heartbeat.rs b/rustfs/src/connect/heartbeat.rs index 1a95d0d6f..f71de4c2f 100644 --- a/rustfs/src/connect/heartbeat.rs +++ b/rustfs/src/connect/heartbeat.rs @@ -430,6 +430,7 @@ impl From for HeartbeatError { TelemetryError::IdentityCertificate => Self::IdentityCertificate, TelemetryError::CredentialName => Self::CredentialName, TelemetryError::CredentialExpired => Self::CredentialExpired, + TelemetryError::StateConflict => Self::StateConflict, TelemetryError::ResponseTooLarge => Self::ResponseTooLarge, TelemetryError::Url(error) => Self::Url(error), TelemetryError::Transport(error) => Self::Transport(error), diff --git a/rustfs/src/connect/identity.rs b/rustfs/src/connect/identity.rs index e9cdbba58..e3028c770 100644 --- a/rustfs/src/connect/identity.rs +++ b/rustfs/src/connect/identity.rs @@ -22,7 +22,7 @@ use base64::Engine as _; use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD}; -use p256::ecdsa::signature::Signer as _; +use p256::ecdsa::signature::{Signer as _, Verifier as _}; use p256::ecdsa::{Signature, SigningKey}; use p256::elliptic_curve::Generate as _; use p256::pkcs8::{DecodePrivateKey as _, EncodePrivateKey as _, LineEnding}; @@ -256,6 +256,24 @@ impl DeviceIdentity { } } + pub(crate) fn sign_pending_registration_state(&self, state: &[u8]) -> String { + let signature: Signature = self.signing_key.sign(state); + BASE64_URL_NO_PAD.encode(signature.normalize_s().to_bytes()) + } + + pub(crate) fn verifies_pending_registration_state(&self, state: &[u8], proof: &str) -> bool { + let Ok(octets) = BASE64_URL_NO_PAD.decode(proof) else { + return false; + }; + if BASE64_URL_NO_PAD.encode(&octets) != proof { + return false; + } + let Ok(signature) = Signature::from_slice(&octets) else { + return false; + }; + signature.normalize_s() == signature && self.signing_key.verifying_key().verify(state, &signature).is_ok() + } + /// The device public key, DER SubjectPublicKeyInfo. pub fn public_key_der(&self) -> Vec { use p256::pkcs8::EncodePublicKey as _; diff --git a/rustfs/src/connect/registration.rs b/rustfs/src/connect/registration.rs index 87802e897..3be43c4ac 100644 --- a/rustfs/src/connect/registration.rs +++ b/rustfs/src/connect/registration.rs @@ -496,7 +496,7 @@ fn canonical_serial(raw: &[u8]) -> Result { Ok(hex_lower(&padded)) } -fn is_uuid_v7(value: &str) -> bool { +pub(crate) fn is_uuid_v7(value: &str) -> bool { Uuid::parse_str(value).is_ok_and(|uuid| uuid.get_version() == Some(Version::SortRand) && uuid.to_string() == value) } diff --git a/rustfs/src/connect/runtime.rs b/rustfs/src/connect/runtime.rs index cb8bf0467..14c54c9f5 100644 --- a/rustfs/src/connect/runtime.rs +++ b/rustfs/src/connect/runtime.rs @@ -20,8 +20,10 @@ use chrono::Utc; use rand::RngExt as _; use tokio::sync::watch; use tokio::task::JoinHandle; +use tokio::time::Instant; use tokio_util::sync::CancellationToken; +use super::client::{ClientError, ConnectClient, ConnectConfig, RotationAttempt}; use super::config::HeartbeatConfig; use super::heartbeat::{CoarseNodeSummary, Delivery, HeartbeatError, HeartbeatSender, HeartbeatStateStore, HeartbeatStatus}; use super::inventory::{ @@ -109,6 +111,14 @@ where return Ok(None); } let sender = HeartbeatSender::new(config.clone())?; + let rotation = ConnectClient::new(ConnectConfig { + endpoint: &config.endpoint, + root_ca_pem: &config.root_ca_pem, + timeout: config.schedule.timeout, + }) + .map_err(rotation_failure)?; + let identity_store = config.identity_store.clone(); + let credential_store = config.credential_store.clone(); let store = HeartbeatStateStore::new(config.state_path.clone()); let lock = store.try_runtime_lock()?; let schedule = config.schedule; @@ -118,10 +128,46 @@ where let task = tokio::spawn(async move { let _lock = lock; let mut backoff = schedule.initial_backoff; + let mut rotation_backoff = schedule.initial_backoff; + let mut rotation_retry_at = None; loop { if task_shutdown.is_cancelled() { break; } + if rotation_retry_at.is_none_or(|retry_at| Instant::now() >= retry_at) { + match cancellable( + &task_shutdown, + rotation.rotate_if_due_once(&identity_store, &credential_store, Utc::now().timestamp()), + ) + .await + { + Some(Ok(RotationAttempt::Completed(_))) => { + rotation_backoff = schedule.initial_backoff; + rotation_retry_at = None; + } + Some(Ok(RotationAttempt::ReenrollmentPending)) => {} + Some(Ok(RotationAttempt::Unavailable { retry_after, .. })) => { + let delay = retry_after + .unwrap_or(rotation_backoff) + .clamp(schedule.initial_backoff, schedule.max_backoff); + rotation_backoff = rotation_backoff.saturating_mul(2).min(schedule.max_backoff); + rotation_retry_at = Some(Instant::now() + delay); + } + Some(Err(ClientError::Transport(_))) => { + rotation_retry_at = Some(Instant::now() + rotation_backoff); + rotation_backoff = rotation_backoff.saturating_mul(2).min(schedule.max_backoff); + } + Some(Err(ClientError::AccessRevoked { status, reason })) => { + let _ = status_tx.send(HeartbeatStatus::AuthenticationStopped { + status: status.as_u16(), + reason, + }); + return; + } + Some(Err(error)) => return failed(&status_tx, rotation_failure(error)), + None => break, + } + } let pending = match store.prepare(sample(), Utc::now()).await { Ok(pending) => pending, Err(error) => return failed(&status_tx, error), @@ -344,6 +390,28 @@ fn failed(status: &watch::Sender, error: HeartbeatError) { }); } +fn rotation_failure(error: ClientError) -> HeartbeatError { + match error { + ClientError::Endpoint => HeartbeatError::Endpoint, + ClientError::RootCertificate => HeartbeatError::RootCertificate, + ClientError::NotRegistered => HeartbeatError::NotRegistered, + ClientError::IdentityMissing => HeartbeatError::IdentityMissing, + ClientError::CredentialExpired | ClientError::CredentialNotYetValid => HeartbeatError::CredentialExpired, + ClientError::IdentityCertificate => HeartbeatError::IdentityCertificate, + ClientError::Identity(error) => HeartbeatError::Identity(error), + ClientError::IdentityStore(error) => HeartbeatError::IdentityStore(error), + ClientError::CredentialStore(error) => HeartbeatError::CredentialStore(error), + ClientError::Credential(error) => HeartbeatError::CredentialValidation(error), + ClientError::Transport(error) => HeartbeatError::Transport(error), + ClientError::ResponseTooLarge => HeartbeatError::ResponseTooLarge, + ClientError::PendingRegistration | ClientError::PendingRotation => HeartbeatError::StateConflict, + ClientError::AccessRevoked { .. } + | ClientError::Rejected { .. } + | ClientError::Unavailable { .. } + | ClientError::Response => HeartbeatError::Response, + } +} + pub(crate) fn heartbeat_failure_reason(error: &HeartbeatError) -> &'static str { use super::registration::CredentialValidationError; @@ -448,6 +516,15 @@ mod tests { heartbeat_failure_reason(&HeartbeatError::CredentialExpired), "connect_heartbeat_credential_expired" ); + let pending_rotation = rotation_failure(ClientError::PendingRotation); + assert!(matches!(pending_rotation, HeartbeatError::StateConflict)); + assert_eq!(heartbeat_failure_reason(&pending_rotation), "connect_heartbeat_state_conflict"); + let oversized_rotation_response = rotation_failure(ClientError::ResponseTooLarge); + assert!(matches!(oversized_rotation_response, HeartbeatError::ResponseTooLarge)); + assert_eq!( + heartbeat_failure_reason(&oversized_rotation_response), + "connect_heartbeat_response_too_large" + ); assert_eq!( heartbeat_failure_reason(&HeartbeatError::CredentialValidation( super::super::registration::CredentialValidationError::Identity diff --git a/rustfs/src/connect/telemetry.rs b/rustfs/src/connect/telemetry.rs index bcee54320..3baef8cc2 100644 --- a/rustfs/src/connect/telemetry.rs +++ b/rustfs/src/connect/telemetry.rs @@ -21,11 +21,12 @@ use rustls::pki_types::{CertificateDer, pem::PemObject as _}; use serde::{Deserialize, Serialize}; use zeroize::Zeroizing; +use super::client::{ClientError, ConnectClient}; use super::config::HeartbeatConfig; use super::credential_store::{CredentialStoreError, DeviceCredential}; use super::identity::IdentityError; use super::identity_store::StoreError; -use super::registration::{CredentialValidationError, validate_stored_credential}; +use super::registration::{CredentialValidationError, certificate_fingerprint}; const MAX_RESPONSE_BYTES: usize = 64 * 1024; @@ -43,6 +44,13 @@ pub(crate) struct TelemetryTransport { config: HeartbeatConfig, } +struct AuthenticatedClient { + cluster_name: String, + cluster_uid: String, + credential_fingerprint: String, + client: Client, +} + impl TelemetryTransport { pub(crate) fn new(config: HeartbeatConfig) -> Result { let mut endpoint = Url::parse(&config.endpoint).map_err(|_| TelemetryError::Endpoint)?; @@ -87,55 +95,80 @@ impl TelemetryTransport { } pub(crate) async fn post(&self, collection: &str, value: &T) -> Result { - let (cluster_name, cluster_uid, client) = self.authenticated_client().await?; - let url = self.endpoint.join(&format!("clusters/{cluster_uid}/{collection}"))?; - let response = match client.post(url).json(value).send().await { - Ok(response) => response, - Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => { + let mut authenticated = self.authenticated_client().await?; + let mut refreshed = false; + loop { + let url = self + .endpoint + .join(&format!("clusters/{}/{collection}", authenticated.cluster_uid))?; + let response = match authenticated.client.post(url).json(value).send().await { + Ok(response) => response, + Err(error) if error.is_timeout() || error.is_connect() || error.is_request() => { + return Ok(TelemetryDelivery::Retry { retry_after: None }); + } + Err(error) => return Err(error.into()), + }; + let status = response.status(); + if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) && !refreshed { + let current = self.authenticated_client().await?; + if current.credential_fingerprint != authenticated.credential_fingerprint { + authenticated = current; + refreshed = true; + continue; + } + } + if status == StatusCode::TOO_MANY_REQUESTS { + return Ok(TelemetryDelivery::Retry { + retry_after: retry_after(response.headers(), Utc::now(), self.config.schedule.max_backoff), + }); + } + if status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() { return Ok(TelemetryDelivery::Retry { retry_after: None }); } - Err(error) => return Err(error.into()), - }; - let status = response.status(); - if status == StatusCode::TOO_MANY_REQUESTS { - return Ok(TelemetryDelivery::Retry { - retry_after: retry_after(response.headers(), Utc::now(), self.config.schedule.max_backoff), + if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { + return Ok(TelemetryDelivery::AuthenticationStopped { + status: status.as_u16(), + reason: response_reason(response).await, + }); + } + if status != StatusCode::OK { + return Ok(TelemetryDelivery::Rejected { + status: status.as_u16(), + reason: response_reason(response).await, + }); + } + return Ok(TelemetryDelivery::Accepted { + cluster_name: authenticated.cluster_name, + body: bounded_body(response).await?, }); } - if status == StatusCode::REQUEST_TIMEOUT || status.is_server_error() { - return Ok(TelemetryDelivery::Retry { retry_after: None }); - } - if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { - return Ok(TelemetryDelivery::AuthenticationStopped { - status: status.as_u16(), - reason: response_reason(response).await, - }); - } - if status != StatusCode::OK { - return Ok(TelemetryDelivery::Rejected { - status: status.as_u16(), - reason: response_reason(response).await, - }); - } - Ok(TelemetryDelivery::Accepted { - cluster_name, - body: bounded_body(response).await?, - }) } - async fn authenticated_client(&self) -> Result<(String, String, Client), TelemetryError> { - let _lock = self.config.credential_store.lock().await?; - let credential = self.config.credential_store.load()?.ok_or(TelemetryError::NotRegistered)?; - let identity = self.config.identity_store.load()?.ok_or(TelemetryError::IdentityMissing)?; - validate_stored_credential(&credential, &identity, &self.root_store, &self.roots)?; + async fn authenticated_client(&self) -> Result { + let lock = self.config.credential_store.lock().await?; + let (credential, identity, _) = ConnectClient::recover_valid_credential_locked( + &lock, + &self.config.identity_store, + &self.config.credential_store, + &self.root_store, + &self.roots, + ) + .map_err(credential_recovery_error)? + .ok_or(TelemetryError::NotRegistered)?; let now = Utc::now().timestamp(); if now < credential.not_before_unix || now >= credential.not_after_unix { return Err(TelemetryError::CredentialExpired); } let (organization_uid, cluster_uid) = credential_parent(&credential)?; let cluster_name = format!("organizations/{organization_uid}/clusters/{cluster_uid}"); + let credential_fingerprint = certificate_fingerprint(&credential.certificate)?; let client = self.client(&credential, &identity.to_pkcs8_pem()?)?; - Ok((cluster_name, cluster_uid.to_owned(), client)) + Ok(AuthenticatedClient { + cluster_name, + cluster_uid: cluster_uid.to_owned(), + credential_fingerprint, + client, + }) } fn client(&self, credential: &DeviceCredential, key: &Zeroizing) -> Result { @@ -160,6 +193,28 @@ impl TelemetryTransport { } } +fn credential_recovery_error(error: ClientError) -> TelemetryError { + match error { + ClientError::Endpoint => TelemetryError::Endpoint, + ClientError::RootCertificate => TelemetryError::RootCertificate, + ClientError::PendingRegistration | ClientError::PendingRotation => TelemetryError::StateConflict, + ClientError::NotRegistered => TelemetryError::NotRegistered, + ClientError::IdentityMissing => TelemetryError::IdentityMissing, + ClientError::CredentialExpired | ClientError::CredentialNotYetValid => TelemetryError::CredentialExpired, + ClientError::IdentityCertificate => TelemetryError::IdentityCertificate, + ClientError::Identity(error) => TelemetryError::Identity(error), + ClientError::IdentityStore(error) => TelemetryError::IdentityStore(error), + ClientError::CredentialStore(error) => TelemetryError::CredentialStore(error), + ClientError::Credential(error) => TelemetryError::CredentialValidation(error), + ClientError::AccessRevoked { .. } + | ClientError::Rejected { .. } + | ClientError::Unavailable { .. } + | ClientError::ResponseTooLarge + | ClientError::Response + | ClientError::Transport(_) => TelemetryError::StateConflict, + } +} + fn credential_parent(credential: &DeviceCredential) -> Result<(&str, &str), TelemetryError> { let mut parts = credential.name.split('/'); let valid = parts.next() == Some("organizations"); @@ -248,6 +303,8 @@ pub(crate) enum TelemetryError { CredentialName, #[error("the stored Connect device certificate is not currently valid")] CredentialExpired, + #[error("the persisted Connect credential transition is invalid")] + StateConflict, #[error("Connect telemetry response exceeded 64 KiB")] ResponseTooLarge, #[error(transparent)] diff --git a/rustfs/tests/connect_registration.rs b/rustfs/tests/connect_registration.rs index f33cf23e6..9915d198e 100644 --- a/rustfs/tests/connect_registration.rs +++ b/rustfs/tests/connect_registration.rs @@ -31,7 +31,12 @@ use rcgen::{ BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair, KeyUsagePurpose, SanType, SerialNumber, }; -use rustfs::connect::{ClientError, ConnectClient, ConnectConfig, CredentialStore, IdentityStore, RegistrationToken, TokenError}; +use rustfs::connect::{ + ClientError, CoarseNodeSummary, ConnectClient, ConnectConfig, CredentialStore, HeartbeatConfig, HeartbeatError, + HeartbeatSchedule, HeartbeatStatus, IdentityStore, RegistrationToken, TokenError, spawn_heartbeat_runtime, +}; +#[cfg(target_os = "linux")] +use rustfs::connect::{InventorySchedule, InventorySnapshot, InventoryStatus, spawn_inventory_runtime}; use rustls::RootCertStore; use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject as _}; use rustls::server::WebPkiClientVerifier; @@ -40,7 +45,9 @@ use sha2::{Digest as _, Sha256}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::net::TcpListener; +use tokio::sync::watch; use tokio_rustls::TlsAcceptor; +use tokio_util::sync::CancellationToken; const ORGANIZATION_UID: &str = "0198f4b0-1a00-7c10-8d21-2e3f4a5b6c70"; const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81"; @@ -48,6 +55,8 @@ const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92"; const TOKEN_UID: &str = "0198f4b0-6f00-7b60-9271-7d8e9fa0b1c5"; const FRESH_TOKEN_UID: &str = "0198f4b0-7f00-7c70-a381-8e9fa0b1c2d6"; const SECOND_TOKEN_UID: &str = "0198f4b0-8f00-7d80-b491-9fa0b1c2d3e7"; +const UNRELATED_TOKEN_UID: &str = "0198f4b0-9f00-7e90-85a1-afb1c2d3e4f8"; +const INVENTORY_UID: &str = "0198f4b0-af00-7fa0-96b1-bfc1d2e3f509"; struct TestPki { root_params: CertificateParams, @@ -161,6 +170,11 @@ impl TestPki { enum Reply { Json(StatusCode, Value), DelayedClose(Duration), + RateLimited(&'static str), + UnauthorizedAfterCredential { + path: std::path::PathBuf, + certificate_serial: String, + }, VerifiedRotation { response: Value, current_public_key: Vec, @@ -172,6 +186,8 @@ enum Reply { struct TestServer { endpoint: String, seen: Arc>>, + paths: Arc>>, + client_certificates: Arc>>>, task: tokio::task::JoinHandle<()>, } @@ -191,7 +207,11 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec, require_cli let acceptor = TlsAcceptor::from(Arc::new(pki.server_config(require_client))); let replies = Arc::new(Mutex::new(VecDeque::from(replies))); let seen = Arc::new(Mutex::new(Vec::new())); + let paths = Arc::new(Mutex::new(Vec::new())); + let client_certificates = Arc::new(Mutex::new(Vec::new())); let captured = seen.clone(); + let captured_paths = paths.clone(); + let captured_certificates = client_certificates.clone(); let task = tokio::spawn(async move { loop { let Ok((stream, _)) = listener.accept().await else { @@ -200,18 +220,35 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec, require_cli let acceptor = acceptor.clone(); let replies = replies.clone(); let seen = captured.clone(); + let paths = captured_paths.clone(); + let client_certificates = captured_certificates.clone(); tokio::spawn(async move { let Ok(stream) = acceptor.accept(stream).await else { return; }; + let client_certificate = stream + .get_ref() + .1 + .peer_certificates() + .and_then(|certificates| certificates.first()) + .map(|certificate| certificate_der_fingerprint(certificate.as_ref())); let service = service_fn(move |request: Request| { let replies = replies.clone(); let seen = seen.clone(); + let paths = paths.clone(); + let client_certificates = client_certificates.clone(); + let client_certificate = client_certificate.clone(); async move { + let path = request.uri().path().to_owned(); let body = request.into_body().collect().await.expect("read request body").to_bytes(); let value: Value = serde_json::from_slice(&body).expect("request JSON"); - seen.lock().expect("seen lock").push(value.clone()); let reply = replies.lock().expect("reply lock").pop_front().expect("planned reply"); + client_certificates + .lock() + .expect("client certificates lock") + .push(client_certificate); + paths.lock().expect("paths lock").push(path); + seen.lock().expect("seen lock").push(value.clone()); match reply { Reply::Json(status, value) => Ok::<_, hyper::Error>( Response::builder() @@ -227,6 +264,38 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec, require_cli .body(Full::new(Bytes::new())) .expect("response")) } + Reply::RateLimited(retry_after) => Ok(Response::builder() + .status(StatusCode::TOO_MANY_REQUESTS) + .header("content-type", "application/json") + .header("retry-after", retry_after) + .body(Full::new(Bytes::from_static(br#"{"details":[]}"#))) + .expect("response")), + Reply::UnauthorizedAfterCredential { + path, + certificate_serial, + } => { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let changed = fs::read(&path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .is_some_and(|credential| { + credential["certificateSerial"].as_str() == Some(certificate_serial.as_str()) + }); + if changed { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("rotated credential commit"); + Ok(Response::builder() + .status(StatusCode::UNAUTHORIZED) + .header("content-type", "application/json") + .body(Full::new(Bytes::from_static(br#"{"details":[{"reason":"CREDENTIAL_REVOKED"}]}"#))) + .expect("response")) + } Reply::VerifiedRotation { response, current_public_key, @@ -257,6 +326,8 @@ async fn server_with_client_auth(pki: &TestPki, replies: Vec, require_cli TestServer { endpoint: format!("https://localhost:{}/agent/", address.port()), seen, + paths, + client_certificates, task, } } @@ -311,10 +382,11 @@ fn certificate_fingerprint(pem: &str) -> String { .next() .expect("leaf certificate") .expect("certificate PEM"); - Sha256::digest(certificate.as_ref()) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect() + certificate_der_fingerprint(certificate.as_ref()) +} + +fn certificate_der_fingerprint(certificate: &[u8]) -> String { + Sha256::digest(certificate).iter().map(|byte| format!("{byte:02x}")).collect() } fn token_document() -> Value { @@ -355,6 +427,89 @@ fn client(server: &TestServer, pki: &TestPki, timeout: Duration) -> ConnectClien .expect("build Connect client") } +fn due_runtime_config( + temp: &tempfile::TempDir, + pki: &TestPki, + endpoint: &str, +) -> (HeartbeatConfig, Value, rustfs::connect::DeviceIdentity) { + runtime_config(temp, pki, endpoint, 17) +} + +fn runtime_config( + temp: &tempfile::TempDir, + pki: &TestPki, + endpoint: &str, + elapsed_hours: i64, +) -> (HeartbeatConfig, Value, rustfs::connect::DeviceIdentity) { + let (identity_store, credential_store) = stores(temp); + let identity = identity_store.load_or_create().expect("create current identity"); + let now = OffsetDateTime::now_utc().replace_nanosecond(0).expect("whole second"); + let credential = pki.credential_window( + &identity, + &format!("urn:rustfs:connect:device:{DEVICE_UID}"), + 0x20, + now - time::Duration::hours(elapsed_hours), + now + time::Duration::hours(24 - elapsed_hours), + ); + let not_before = + OffsetDateTime::parse(credential["notBefore"].as_str().expect("notBefore"), &Rfc3339).expect("parse notBefore"); + let not_after = OffsetDateTime::parse(credential["notAfter"].as_str().expect("notAfter"), &Rfc3339).expect("parse notAfter"); + assert_eq!(not_after - not_before, time::Duration::hours(24)); + fs::create_dir_all(temp.path().join("credential")).expect("credential directory"); + write_stored_credential(&temp.path().join("credential/device.crt.json"), &credential); + let next = identity_store.load_or_create_next().expect("create next identity"); + let mut config = HeartbeatConfig::new( + endpoint, + pki.root_pem.as_bytes(), + identity_store, + credential_store, + temp.path().join("heartbeat/state.json"), + ); + config.schedule = HeartbeatSchedule { + cadence: Duration::from_millis(100), + jitter: Duration::ZERO, + timeout: Duration::from_millis(250), + initial_backoff: Duration::from_millis(20), + max_backoff: Duration::from_secs(2), + }; + (config, credential, next) +} + +fn heartbeat_response(server_time: &str) -> Value { + json!({ + "serverTime": server_time, + "acceptedVersion": "v1", + "capabilityHints": [], + }) +} + +async fn wait_for_requests(server: &TestServer, count: usize) { + tokio::time::timeout(Duration::from_secs(3), async { + while server.paths.lock().expect("paths lock").len() < count { + tokio::task::yield_now().await; + } + }) + .await + .expect("planned requests"); +} + +async fn wait_for_heartbeat_status( + status: &mut watch::Receiver, + predicate: impl Fn(&HeartbeatStatus) -> bool, +) -> HeartbeatStatus { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let current = status.borrow_and_update().clone(); + if predicate(¤t) { + return current; + } + status.changed().await.expect("heartbeat status channel"); + } + }) + .await + .expect("heartbeat status") +} + fn rotation_response(pki: &TestPki, identity: &rustfs::connect::DeviceIdentity, serial: u8) -> (Value, Value) { let stored = pki.credential(identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), serial); let mut wire = stored.clone(); @@ -577,7 +732,7 @@ async fn concurrent_rotation_retries_converge_and_promote_the_next_key() { assert!(matches!(second, Err(ClientError::Unavailable { .. }))); let (request_id, certificate_request) = { let seen = retries.seen.lock().expect("seen lock"); - assert!(seen.len() >= 3, "bounded retries must reach the server"); + assert_eq!(seen.len(), 6, "each public rotation call makes three attempts"); for request in &seen[1..] { assert_eq!(request["requestId"], seen[0]["requestId"]); assert_eq!(request["certificateRequest"], seen[0]["certificateRequest"]); @@ -645,6 +800,539 @@ async fn concurrent_rotation_retries_converge_and_promote_the_next_key() { assert!(!temp.path().join("credential/rotation.pending.json").exists()); } +#[tokio::test] +async fn heartbeat_runtime_retries_rotation_and_reloads_the_rotated_credential() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let (mut config, current, next) = due_runtime_config(&temp, &pki, "https://localhost/agent/"); + let (rotated, rotated_stored) = rotation_response(&pki, &next, 0x21); + let server = server_with_client_auth( + &pki, + vec![ + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")), + Reply::VerifiedRotation { + response: rotated, + current_public_key: config + .identity_store + .load() + .expect("load current identity") + .expect("current identity") + .public_key_der(), + current_certificate_fingerprint: certificate_fingerprint( + current["certificate"].as_str().expect("current certificate"), + ), + device_name: current["name"].as_str().expect("device name").to_owned(), + }, + Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:04Z")), + ], + true, + ) + .await; + config.endpoint = server.endpoint.clone(); + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("start heartbeat runtime") + .expect("configured runtime"); + + wait_for_requests(&server, 4).await; + runtime.shutdown().await; + + let rotation_path = format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential"); + let heartbeat_path = format!("/agent/clusters/{CLUSTER_UID}/heartbeats"); + assert_eq!( + &server.paths.lock().expect("paths lock")[..4], + [rotation_path.clone(), heartbeat_path.clone(), rotation_path, heartbeat_path,] + ); + let current_fingerprint = certificate_fingerprint(current["certificate"].as_str().expect("current certificate")); + let rotated_fingerprint = certificate_fingerprint(rotated_stored["certificate"].as_str().expect("rotated certificate")); + let certificates = server.client_certificates.lock().expect("client certificates lock"); + assert_eq!(certificates[1].as_deref(), Some(current_fingerprint.as_str())); + assert_eq!(certificates[3].as_deref(), Some(rotated_fingerprint.as_str())); + drop(certificates); + let stored: Value = + serde_json::from_slice(&fs::read(temp.path().join("credential/device.crt.json")).expect("stored rotated credential")) + .expect("stored credential JSON"); + assert_eq!(stored["name"], current["name"]); + assert_eq!(stored["uid"], current["uid"]); + assert_eq!(stored["certificateSerial"], rotated_stored["certificateSerial"]); +} + +#[tokio::test] +async fn heartbeat_runtime_respects_rotation_retry_after_without_blocking_heartbeats() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let server = server_with_client_auth( + &pki, + std::iter::once(Reply::RateLimited("1")) + .chain((0..8).map(|_| Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")))) + .collect(), + true, + ) + .await; + let (config, _, _) = due_runtime_config(&temp, &pki, &server.endpoint); + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("start heartbeat runtime") + .expect("configured runtime"); + let mut status = runtime.status(); + + wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await; + tokio::time::sleep(Duration::from_millis(350)).await; + runtime.shutdown().await; + + let paths = server.paths.lock().expect("paths lock"); + assert_eq!(paths.iter().filter(|path| path.ends_with(":rotateCredential")).count(), 1); + assert!(paths.iter().filter(|path| path.ends_with("/heartbeats")).count() >= 3); +} + +#[tokio::test] +async fn heartbeat_runtime_skips_only_valid_pending_reenrollment() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let (mut config, _, next) = due_runtime_config(&temp, &pki, "https://localhost/agent/"); + let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x24); + let server = server_with_client_auth( + &pki, + vec![ + Reply::Json(StatusCode::CREATED, enrolled), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")), + ], + true, + ) + .await; + config.endpoint = server.endpoint.clone(); + let validation_config = config.clone(); + client(&server, &pki, Duration::from_millis(250)) + .reenroll(&config.identity_store, &config.credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("complete prior reenrollment"); + assert!(matches!( + client(&server, &pki, Duration::from_millis(250)) + .reenroll(&config.identity_store, &config.credential_store, &token_with_uid(SECOND_TOKEN_UID)) + .await, + Err(ClientError::Unavailable { .. }) + )); + let pending_path = temp.path().join("credential/registration.pending.json"); + let next_path = temp.path().join("identity/device.key.next"); + let credential_path = temp.path().join("credential/device.crt.json"); + let pending = fs::read(&pending_path).expect("pending reenrollment"); + let next = fs::read(&next_path).expect("pending reenrollment key"); + let credential = fs::read(&credential_path).expect("current credential"); + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("start heartbeat runtime") + .expect("configured runtime"); + let mut status = runtime.status(); + + assert!(matches!( + wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await, + HeartbeatStatus::Online { .. } + )); + runtime.shutdown().await; + + assert_eq!(fs::read(&pending_path).expect("preserved pending reenrollment"), pending); + assert_eq!(fs::read(&next_path).expect("preserved pending reenrollment key"), next); + assert_eq!(fs::read(&credential_path).expect("preserved current credential"), credential); + let paths = server.paths.lock().expect("paths lock"); + assert_eq!( + paths + .iter() + .filter(|path| path.ends_with("registrationTokens:exchange")) + .count(), + 4 + ); + assert_eq!(paths.iter().filter(|path| path.ends_with(":rotateCredential")).count(), 0); + assert_eq!(paths.iter().filter(|path| path.ends_with("/heartbeats")).count(), 1); + drop(paths); + + for (field, value) in [ + ("requestId", "not-a-request-id"), + ("tokenUid", ""), + ("tokenUid", "not-a-token-uid"), + ("tokenUid", FRESH_TOKEN_UID), + ("tokenUid", UNRELATED_TOKEN_UID), + ] { + let mut corrupted_pending: Value = serde_json::from_slice(&pending).expect("pending reenrollment JSON"); + corrupted_pending[field] = json!(value); + let corrupted_pending = serde_json::to_vec(&corrupted_pending).expect("corrupted pending reenrollment JSON"); + fs::write(&pending_path, &corrupted_pending).expect("corrupt pending reenrollment"); + set_owner_only(&pending_path); + let corrupted_runtime = spawn_heartbeat_runtime(Some(validation_config.clone()), &shutdown, || { + CoarseNodeSummary::new(1, 1, 0).expect("node summary") + }) + .expect("start heartbeat runtime with corrupt pending state") + .expect("configured runtime"); + let mut corrupted_status = corrupted_runtime.status(); + assert_eq!( + wait_for_heartbeat_status(&mut corrupted_status, |status| matches!(status, HeartbeatStatus::Failed { .. })).await, + HeartbeatStatus::Failed { + reason: HeartbeatError::StateConflict.to_string(), + } + ); + corrupted_runtime.shutdown().await; + assert_eq!(fs::read(&pending_path).expect("preserved corrupt pending state"), corrupted_pending); + } + assert_eq!(fs::read(&next_path).expect("preserved pending reenrollment key"), next); + assert_eq!(fs::read(&credential_path).expect("preserved current credential"), credential); + let paths = server.paths.lock().expect("paths lock"); + assert_eq!(paths.iter().filter(|path| path.ends_with(":rotateCredential")).count(), 0); + assert_eq!(paths.iter().filter(|path| path.ends_with("/heartbeats")).count(), 1); +} + +#[tokio::test] +async fn heartbeat_retries_with_the_new_credential_after_concurrent_rotation() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let (mut config, current, next) = runtime_config(&temp, &pki, "https://localhost/agent/", 1); + let (rotated, rotated_stored) = rotation_response(&pki, &next, 0x22); + let server = server_with_client_auth( + &pki, + vec![ + Reply::UnauthorizedAfterCredential { + path: temp.path().join("credential/device.crt.json"), + certificate_serial: rotated_stored["certificateSerial"] + .as_str() + .expect("rotated serial") + .to_owned(), + }, + Reply::VerifiedRotation { + response: rotated, + current_public_key: config + .identity_store + .load() + .expect("load current identity") + .expect("current identity") + .public_key_der(), + current_certificate_fingerprint: certificate_fingerprint( + current["certificate"].as_str().expect("current certificate"), + ), + device_name: current["name"].as_str().expect("device name").to_owned(), + }, + Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:03Z")), + ], + true, + ) + .await; + config.endpoint = server.endpoint.clone(); + let identity_store = config.identity_store.clone(); + let credential_store = config.credential_store.clone(); + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("start heartbeat runtime") + .expect("configured runtime"); + let mut status = runtime.status(); + + wait_for_requests(&server, 1).await; + let due = OffsetDateTime::parse(current["notAfter"].as_str().expect("notAfter"), &Rfc3339) + .expect("parse notAfter") + .unix_timestamp() + - 8 * 60 * 60; + client(&server, &pki, Duration::from_millis(250)) + .rotate_if_due(&identity_store, &credential_store, due) + .await + .expect("concurrent rotation") + .expect("rotation due"); + assert!(matches!( + wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::Online { .. })).await, + HeartbeatStatus::Online { .. } + )); + runtime.shutdown().await; + + let heartbeat_path = format!("/agent/clusters/{CLUSTER_UID}/heartbeats"); + assert_eq!( + server.paths.lock().expect("paths lock").as_slice(), + [ + heartbeat_path.clone(), + format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential"), + heartbeat_path, + ] + ); + let current_fingerprint = certificate_fingerprint(current["certificate"].as_str().expect("current certificate")); + let rotated_fingerprint = certificate_fingerprint(rotated_stored["certificate"].as_str().expect("rotated certificate")); + let certificates = server.client_certificates.lock().expect("client certificates lock"); + assert_eq!(certificates[0].as_deref(), Some(current_fingerprint.as_str())); + assert_eq!(certificates[2].as_deref(), Some(rotated_fingerprint.as_str())); +} + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn inventory_retries_with_the_new_credential_after_concurrent_rotation() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let (mut config, current, next) = runtime_config(&temp, &pki, "https://localhost/agent/", 1); + let (rotated, rotated_stored) = rotation_response(&pki, &next, 0x23); + let server = server_with_client_auth( + &pki, + vec![ + Reply::UnauthorizedAfterCredential { + path: temp.path().join("credential/device.crt.json"), + certificate_serial: rotated_stored["certificateSerial"] + .as_str() + .expect("rotated serial") + .to_owned(), + }, + Reply::VerifiedRotation { + response: rotated, + current_public_key: config + .identity_store + .load() + .expect("load current identity") + .expect("current identity") + .public_key_der(), + current_certificate_fingerprint: certificate_fingerprint( + current["certificate"].as_str().expect("current certificate"), + ), + device_name: current["name"].as_str().expect("device name").to_owned(), + }, + Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})), + ], + true, + ) + .await; + config.endpoint = server.endpoint.clone(); + let identity_store = config.identity_store.clone(); + let credential_store = config.credential_store.clone(); + let snapshot = InventorySnapshot::current(1, 1, 100, 50, []).expect("inventory snapshot"); + let shutdown = CancellationToken::new(); + let runtime = spawn_inventory_runtime( + Some(config), + InventorySchedule { + cadence: Duration::from_millis(100), + jitter: Duration::ZERO, + }, + &shutdown, + move || std::future::ready(Ok(snapshot.clone())), + ) + .expect("start inventory runtime") + .expect("configured runtime"); + let mut status = runtime.status(); + + wait_for_requests(&server, 1).await; + let due = OffsetDateTime::parse(current["notAfter"].as_str().expect("notAfter"), &Rfc3339) + .expect("parse notAfter") + .unix_timestamp() + - 8 * 60 * 60; + client(&server, &pki, Duration::from_millis(250)) + .rotate_if_due(&identity_store, &credential_store, due) + .await + .expect("concurrent rotation") + .expect("rotation due"); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let current = status.borrow_and_update().clone(); + if matches!(current, InventoryStatus::BackingOff { .. }) { + break current; + } + status.changed().await.expect("inventory status channel"); + } + }) + .await + .expect("inventory retry status"), + InventoryStatus::BackingOff { .. } + )); + runtime.shutdown().await; + + let inventory_path = format!("/agent/clusters/{CLUSTER_UID}/inventorySnapshots"); + assert_eq!( + server.paths.lock().expect("paths lock").as_slice(), + [ + inventory_path.clone(), + format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential"), + inventory_path, + ] + ); + let current_fingerprint = certificate_fingerprint(current["certificate"].as_str().expect("current certificate")); + let rotated_fingerprint = certificate_fingerprint(rotated_stored["certificate"].as_str().expect("rotated certificate")); + let certificates = server.client_certificates.lock().expect("client certificates lock"); + assert_eq!(certificates[0].as_deref(), Some(current_fingerprint.as_str())); + assert_eq!(certificates[2].as_deref(), Some(rotated_fingerprint.as_str())); +} + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn inventory_first_recovers_a_saved_reenrollment_before_telemetry() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; + let (mut config, _, next) = runtime_config(&temp, &pki, &failed.endpoint, 1); + assert!(matches!( + client(&failed, &pki, Duration::from_millis(250)) + .reenroll(&config.identity_store, &config.credential_store, &token_with_uid(SECOND_TOKEN_UID)) + .await, + Err(ClientError::Unavailable { .. }) + )); + let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x25); + write_stored_credential(&temp.path().join("credential/device.crt.json"), &enrolled); + assert!(temp.path().join("credential/registration.pending.json").exists()); + assert!(temp.path().join("identity/device.key.next").exists()); + assert_ne!( + config + .identity_store + .load() + .expect("load pre-recovery identity") + .expect("pre-recovery identity") + .public_key_der(), + next.public_key_der() + ); + + let snapshot = InventorySnapshot::current(1, 1, 100, 50, []).expect("inventory snapshot"); + let content_hash = snapshot.content_hash().expect("inventory content hash"); + let telemetry = server_with_client_auth( + &pki, + vec![ + Reply::Json( + StatusCode::OK, + json!({ + "name": format!("organizations/{ORGANIZATION_UID}/clusters/{CLUSTER_UID}/inventorySnapshots/{INVENTORY_UID}"), + "uid": INVENTORY_UID, + "contentHash": content_hash, + "receivedAt": "2026-08-25T01:02:03Z", + }), + ), + Reply::Json(StatusCode::OK, heartbeat_response("2026-08-25T01:02:04Z")), + ], + true, + ) + .await; + config.endpoint = telemetry.endpoint.clone(); + config.schedule.cadence = Duration::from_secs(60); + let heartbeat_config = config.clone(); + let shutdown = CancellationToken::new(); + let inventory_snapshot = snapshot.clone(); + let inventory = spawn_inventory_runtime( + Some(config), + InventorySchedule { + cadence: Duration::from_secs(60), + jitter: Duration::ZERO, + }, + &shutdown, + move || std::future::ready(Ok(inventory_snapshot.clone())), + ) + .expect("start inventory runtime") + .expect("configured inventory runtime"); + let mut inventory_status = inventory.status(); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let current = inventory_status.borrow_and_update().clone(); + if matches!(current, InventoryStatus::Online { .. }) { + break current; + } + inventory_status.changed().await.expect("inventory status channel"); + } + }) + .await + .expect("inventory online status"), + InventoryStatus::Online { .. } + )); + + let pending_path = temp.path().join("credential/registration.pending.json"); + assert!(!pending_path.exists()); + assert!(!temp.path().join("identity/device.key.next").exists()); + assert_eq!( + heartbeat_config + .identity_store + .load() + .expect("load recovered identity") + .expect("recovered identity") + .public_key_der(), + next.public_key_der() + ); + let completed: Value = serde_json::from_slice( + &fs::read(temp.path().join("credential/registration.completed.json")).expect("completed reenrollment receipt"), + ) + .expect("completed reenrollment JSON"); + assert_eq!(completed["tokenUid"], SECOND_TOKEN_UID); + + let heartbeat = spawn_heartbeat_runtime(Some(heartbeat_config), &shutdown, || { + CoarseNodeSummary::new(1, 1, 0).expect("node summary") + }) + .expect("start heartbeat runtime") + .expect("configured heartbeat runtime"); + let mut heartbeat_status = heartbeat.status(); + assert!(matches!( + wait_for_heartbeat_status(&mut heartbeat_status, |status| matches!(status, HeartbeatStatus::Online { .. })).await, + HeartbeatStatus::Online { .. } + )); + heartbeat.shutdown().await; + inventory.shutdown().await; + + assert_eq!(failed.paths.lock().expect("registration paths").len(), 3); + assert_eq!( + telemetry.paths.lock().expect("telemetry paths").as_slice(), + [ + format!("/agent/clusters/{CLUSTER_UID}/inventorySnapshots"), + format!("/agent/clusters/{CLUSTER_UID}/heartbeats"), + ] + ); + let enrolled_fingerprint = certificate_fingerprint(enrolled["certificate"].as_str().expect("enrolled certificate")); + assert_eq!( + telemetry.client_certificates.lock().expect("client certificates").as_slice(), + [Some(enrolled_fingerprint.clone()), Some(enrolled_fingerprint)] + ); +} + +#[tokio::test] +async fn heartbeat_runtime_cancels_an_in_flight_rotation() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let server = server_with_client_auth(&pki, vec![Reply::DelayedClose(Duration::from_secs(5))], true).await; + let (config, _, _) = due_runtime_config(&temp, &pki, &server.endpoint); + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("start heartbeat runtime") + .expect("configured runtime"); + + wait_for_requests(&server, 1).await; + tokio::time::timeout(Duration::from_millis(250), runtime.shutdown()) + .await + .expect("rotation cancellation"); + assert_eq!( + server.paths.lock().expect("paths lock").as_slice(), + [format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential")] + ); +} + +#[tokio::test] +async fn heartbeat_runtime_stops_when_rotation_reports_revocation() { + let temp = tempfile::tempdir().expect("temp dir"); + let pki = TestPki::new(); + let server = server_with_client_auth( + &pki, + vec![Reply::Json( + StatusCode::UNAUTHORIZED, + json!({"details": [{"reason": "DEVICE_REVOKED"}]}), + )], + true, + ) + .await; + let (config, _, _) = due_runtime_config(&temp, &pki, &server.endpoint); + let shutdown = CancellationToken::new(); + let runtime = spawn_heartbeat_runtime(Some(config), &shutdown, || CoarseNodeSummary::new(1, 1, 0).expect("node summary")) + .expect("start heartbeat runtime") + .expect("configured runtime"); + let mut status = runtime.status(); + + assert_eq!( + wait_for_heartbeat_status(&mut status, |status| matches!(status, HeartbeatStatus::AuthenticationStopped { .. })).await, + HeartbeatStatus::AuthenticationStopped { + status: 401, + reason: Some("DEVICE_REVOKED".to_owned()), + } + ); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + server.paths.lock().expect("paths lock").as_slice(), + [format!("/agent/clusterDevices/{DEVICE_UID}:rotateCredential")] + ); + runtime.shutdown().await; +} + #[tokio::test] async fn rotation_commit_recovers_after_each_durable_step() { let temp = tempfile::tempdir().expect("temp dir"); @@ -665,6 +1353,13 @@ async fn rotation_commit_recovers_after_each_durable_step() { .await, Err(ClientError::Unavailable { .. }) )); + let failed_seen = failed.seen.lock().expect("seen lock"); + assert_eq!(failed_seen.len(), 3, "public rotation keeps its bounded retry contract"); + for request in &failed_seen[1..] { + assert_eq!(request["requestId"], failed_seen[0]["requestId"]); + assert_eq!(request["certificateRequest"], failed_seen[0]["certificateRequest"]); + } + drop(failed_seen); let pending_path = temp.path().join("credential/rotation.pending.json"); let pending = fs::read(&pending_path).expect("read pending state"); @@ -754,19 +1449,31 @@ async fn pending_reenrollment_blocks_rotation_and_resumes_original_exchange() { async fn reenrollment_commit_recovers_after_each_durable_step() { let temp = tempfile::tempdir().expect("temp dir"); let (identity_store, credential_store) = stores(&temp); - let current = identity_store.load_or_create().expect("create identity"); + let initial = identity_store.load_or_create().expect("create identity"); let pki = TestPki::new(); - let issued = pki.credential(¤t, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13); + let issued = pki.credential(&initial, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13); let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await; client(®istration, &pki, Duration::from_secs(2)) .register(&identity_store, &credential_store, &token()) .await .expect("register"); + let first_next = identity_store + .load_or_create_next() + .expect("create first reenrollment identity"); + let first_enrolled = pki.credential(&first_next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 14); + let first_reenrollment = server(&pki, vec![Reply::Json(StatusCode::CREATED, first_enrolled)]).await; + client(&first_reenrollment, &pki, Duration::from_secs(2)) + .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .await + .expect("complete prior reenrollment"); + let completed_path = temp.path().join("credential/registration.completed.json"); + let previous_completed = fs::read(&completed_path).expect("read prior completed receipt"); + let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; assert!(matches!( client(&failed, &pki, Duration::from_secs(2)) - .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) .await, Err(ClientError::Unavailable { .. }) )); @@ -775,16 +1482,41 @@ async fn reenrollment_commit_recovers_after_each_durable_step() { let pending = fs::read(&pending_path).expect("read pending reenrollment"); let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read next key"); let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key"); - let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 14); + let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 15); write_stored_credential(&temp.path().join("credential/device.crt.json"), &enrolled); let idle = server(&pki, vec![]).await; let idle_client = client(&idle, &pki, Duration::from_secs(2)); + for token_uid in [UNRELATED_TOKEN_UID, FRESH_TOKEN_UID] { + let mut document: Value = serde_json::from_slice(&pending).expect("pending reenrollment JSON"); + document["tokenUid"] = json!(token_uid); + let tampered_pending = serde_json::to_vec(&document).expect("tampered pending reenrollment JSON"); + fs::write(&pending_path, &tampered_pending).expect("tamper pending after credential save"); + set_owner_only(&pending_path); + assert!(matches!( + idle_client + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) + .await, + Err(ClientError::PendingRegistration) + )); + assert_eq!(fs::read(&pending_path).expect("preserved tampered pending"), tampered_pending); + assert_eq!(fs::read(&completed_path).expect("preserved prior completed receipt"), previous_completed); + assert_eq!( + identity_store + .load() + .expect("load current key") + .expect("current key") + .public_key_der(), + first_next.public_key_der() + ); + } + fs::write(&pending_path, &pending).expect("restore bound pending after credential save"); + set_owner_only(&pending_path); let recovered = idle_client - .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) .await .expect("recover after reenrollment credential save"); - assert_eq!(recovered.certificate_serial, "0e".repeat(16)); + assert_eq!(recovered.certificate_serial, "0f".repeat(16)); assert_eq!( identity_store .load() @@ -794,26 +1526,50 @@ async fn reenrollment_commit_recovers_after_each_durable_step() { next.public_key_der() ); - fs::remove_file(temp.path().join("credential/registration.completed.json")).expect("remove completed receipt"); - fs::write(&pending_path, pending).expect("restore pending after key commit"); + fs::write(&completed_path, &previous_completed).expect("restore prior completed receipt after key commit"); + set_owner_only(&completed_path); + for token_uid in [UNRELATED_TOKEN_UID, FRESH_TOKEN_UID] { + let mut document: Value = serde_json::from_slice(&pending).expect("pending reenrollment JSON"); + document["tokenUid"] = json!(token_uid); + let tampered_pending = serde_json::to_vec(&document).expect("tampered pending reenrollment JSON"); + fs::write(&pending_path, &tampered_pending).expect("tamper pending after key commit"); + set_owner_only(&pending_path); + assert!(matches!( + idle_client + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) + .await, + Err(ClientError::PendingRegistration) + )); + assert_eq!(fs::read(&pending_path).expect("preserved tampered pending"), tampered_pending); + assert_eq!(fs::read(&completed_path).expect("preserved prior completed receipt"), previous_completed); + assert_eq!( + identity_store + .load() + .expect("load current key") + .expect("current key") + .public_key_der(), + next.public_key_der() + ); + } + fs::write(&pending_path, pending).expect("restore bound pending after key commit"); set_owner_only(&pending_path); let recovered = idle_client - .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) .await .expect("recover after reenrollment key commit"); - assert_eq!(recovered.certificate_serial, "0e".repeat(16)); + assert_eq!(recovered.certificate_serial, "0f".repeat(16)); assert!(!pending_path.exists()); let recovered = idle_client - .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID)) + .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) .await .expect("completed reenrollment is idempotent after pending cleanup"); - assert_eq!(recovered.certificate_serial, "0e".repeat(16)); + assert_eq!(recovered.certificate_serial, "0f".repeat(16)); assert!(idle.seen.lock().expect("seen lock").is_empty()); let different = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await; assert!(matches!( client(&different, &pki, Duration::from_secs(2)) - .reenroll(&identity_store, &credential_store, &token_with_uid(SECOND_TOKEN_UID)) + .reenroll(&identity_store, &credential_store, &token_with_uid(TOKEN_UID)) .await, Err(ClientError::Unavailable { .. }) ));