feat(connect): add registration and credential rotation (#6353)

rustfs/tests/connect_registration.rs

@@ -0,0 +1,911 @@
+// 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::collections::VecDeque;
+use std::fs;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use base64::Engine as _;
+use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD};
+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 p256::ecdsa::signature::Verifier as _;
+use p256::ecdsa::{Signature, VerifyingKey};
+use p256::pkcs8::DecodePublicKey as _;
+use rcgen::{
+    BasicConstraints, CertificateParams, DistinguishedName, DnType, ExtendedKeyUsagePurpose, IsCa, Issuer, KeyPair,
+    KeyUsagePurpose, SanType, SerialNumber,
+};
+use rustfs::connect::{ClientError, ConnectClient, ConnectConfig, CredentialStore, IdentityStore, RegistrationToken, TokenError};
+use rustls::RootCertStore;
+use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject as _};
+use rustls::server::WebPkiClientVerifier;
+use serde_json::{Value, json};
+use sha2::{Digest as _, Sha256};
+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 FRESH_TOKEN_UID: &str = "0198f4b0-7f00-7c70-a381-8e9fa0b1c2d6";
+const SECOND_TOKEN_UID: &str = "0198f4b0-8f00-7d80-b491-9fa0b1c2d3e7";
+
+struct TestPki {
+    root_params: CertificateParams,
+    root_key: KeyPair,
+    root_der: CertificateDer<'static>,
+    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 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_der: root.der().clone(),
+            root_pem: root.pem(),
+            server_der: server.der().clone(),
+            server_key: PrivatePkcs8KeyDer::from(server_key.serialize_der()),
+        }
+    }
+
+    fn credential(&self, identity: &rustfs::connect::DeviceIdentity, uri: &str, serial_byte: u8) -> Value {
+        let now = OffsetDateTime::now_utc().replace_nanosecond(0).expect("whole second");
+        self.credential_window(identity, uri, serial_byte, now, now + time::Duration::days(1))
+    }
+
+    fn credential_window(
+        &self,
+        identity: &rustfs::connect::DeviceIdentity,
+        uri: &str,
+        serial_byte: u8,
+        not_before: OffsetDateTime,
+        not_after: OffsetDateTime,
+    ) -> Value {
+        let mut params = CertificateParams::default();
+        params.not_before = not_before;
+        params.not_after = not_after;
+        params.serial_number = Some(SerialNumber::from(vec![serial_byte; 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
+            .push(SanType::URI(uri.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 = format!("{serial_byte:02x}").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": not_before.format(&Rfc3339).expect("format notBefore"),
+            "notAfter": not_after.format(&Rfc3339).expect("format notAfter"),
+        })
+    }
+
+    fn server_config(&self, require_client: bool) -> rustls::ServerConfig {
+        let mut roots = RootCertStore::empty();
+        roots.add(self.root_der.clone()).expect("add client root");
+        let verifier = WebPkiClientVerifier::builder(Arc::new(roots));
+        let verifier = if require_client {
+            verifier.build()
+        } else {
+            verifier.allow_unauthenticated().build()
+        }
+        .expect("build client verifier");
+        rustls::ServerConfig::builder()
+            .with_client_cert_verifier(verifier)
+            .with_single_cert(vec![self.server_der.clone()], PrivateKeyDer::Pkcs8(self.server_key.clone_key()))
+            .expect("build server TLS")
+    }
+}
+
+#[derive(Clone)]
+enum Reply {
+    Json(StatusCode, Value),
+    DelayedClose(Duration),
+    VerifiedRotation {
+        response: Value,
+        current_public_key: Vec<u8>,
+        current_certificate_fingerprint: String,
+        device_name: String,
+    },
+}
+
+struct TestServer {
+    endpoint: String,
+    seen: Arc<Mutex<Vec<Value>>>,
+    task: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+async fn server(pki: &TestPki, replies: Vec<Reply>) -> TestServer {
+    server_with_client_auth(pki, replies, false).await
+}
+
+async fn server_with_client_auth(pki: &TestPki, replies: Vec<Reply>, require_client: bool) -> TestServer {
+    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(require_client)));
+    let replies = Arc::new(Mutex::new(VecDeque::from(replies)));
+    let seen = Arc::new(Mutex::new(Vec::new()));
+    let captured = seen.clone();
+    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();
+            tokio::spawn(async move {
+                let Ok(stream) = acceptor.accept(stream).await else {
+                    return;
+                };
+                let service = service_fn(move |request: Request<hyper::body::Incoming>| {
+                    let replies = replies.clone();
+                    let seen = seen.clone();
+                    async move {
+                        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");
+                        match reply {
+                            Reply::Json(status, value) => Ok::<_, hyper::Error>(
+                                Response::builder()
+                                    .status(status)
+                                    .header("content-type", "application/json")
+                                    .body(Full::new(Bytes::from(serde_json::to_vec(&value).expect("reply JSON"))))
+                                    .expect("response"),
+                            ),
+                            Reply::DelayedClose(delay) => {
+                                tokio::time::sleep(delay).await;
+                                Ok(Response::builder()
+                                    .status(StatusCode::SERVICE_UNAVAILABLE)
+                                    .body(Full::new(Bytes::new()))
+                                    .expect("response"))
+                            }
+                            Reply::VerifiedRotation {
+                                response,
+                                current_public_key,
+                                current_certificate_fingerprint,
+                                device_name,
+                            } => {
+                                verify_rotation_request(
+                                    &value,
+                                    &current_public_key,
+                                    &current_certificate_fingerprint,
+                                    &device_name,
+                                );
+                                Ok(Response::builder()
+                                    .status(StatusCode::OK)
+                                    .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()),
+        seen,
+        task,
+    }
+}
+
+fn verify_rotation_request(request: &Value, current_public_key: &[u8], fingerprint: &str, device_name: &str) {
+    assert_eq!(request["protocolVersion"], "v1");
+    assert_eq!(request["proof"]["algorithm"], "ES256");
+    let csr = BASE64_STANDARD
+        .decode(request["certificateRequest"].as_str().expect("certificateRequest"))
+        .expect("CSR base64");
+    let csr_digest = BASE64_URL_NO_PAD.encode(Sha256::digest(&csr));
+    let request_id = request["requestId"].as_str().expect("requestId");
+    let transcript = rebuilt_rotation_transcript(
+        b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1",
+        [fingerprint, device_name, request_id, &csr_digest],
+    );
+    let encoded = request["proof"]["value"].as_str().expect("proof value");
+    assert_eq!(encoded.len(), 86);
+    let raw = BASE64_URL_NO_PAD.decode(encoded).expect("proof base64url");
+    let signature = Signature::from_slice(&raw).expect("fixed-width signature");
+    assert!(signature.normalize_s().is_none(), "rotation proof must be low-S");
+    let verifying = VerifyingKey::from_public_key_der(current_public_key).expect("current public key");
+    verifying.verify(&transcript, &signature).expect("rotation proof verifies");
+
+    let wrong_domain = rebuilt_rotation_transcript(
+        b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V2",
+        [fingerprint, device_name, request_id, &csr_digest],
+    );
+    assert!(verifying.verify(&wrong_domain, &signature).is_err());
+    let wrong_order = rebuilt_rotation_transcript(
+        b"RUSTFS-CONNECT-CREDENTIAL-ROTATION-V1",
+        [device_name, fingerprint, request_id, &csr_digest],
+    );
+    assert!(verifying.verify(&wrong_order, &signature).is_err());
+}
+
+fn rebuilt_rotation_transcript(domain: &[u8], fields: [&str; 4]) -> Vec<u8> {
+    let mut transcript = Vec::new();
+    transcript.extend_from_slice(domain);
+    transcript.push(b'\n');
+    for field in fields {
+        transcript.extend_from_slice(field.len().to_string().as_bytes());
+        transcript.push(b':');
+        transcript.extend_from_slice(field.as_bytes());
+        transcript.push(b'\n');
+    }
+    transcript
+}
+
+fn certificate_fingerprint(pem: &str) -> String {
+    let certificate = CertificateDer::pem_slice_iter(pem.as_bytes())
+        .next()
+        .expect("leaf certificate")
+        .expect("certificate PEM");
+    Sha256::digest(certificate.as_ref())
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect()
+}
+
+fn token_document() -> Value {
+    json!({
+        "registrationTokenUid": TOKEN_UID,
+        "registrationTokenSecret": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
+        "organizationUid": ORGANIZATION_UID,
+        "clusterUid": CLUSTER_UID,
+        "challengeNonce": "a3f1c07d9b2e4856af0c1d3b5e7f9012c4a6b8d0e2f4061738495a6b7c8d9e0f",
+        "expiresUnix": OffsetDateTime::now_utc().unix_timestamp() + 3600,
+    })
+}
+
+fn token() -> RegistrationToken {
+    token_with_uid(TOKEN_UID)
+}
+
+fn token_with_uid(uid: &str) -> RegistrationToken {
+    let document = token_document();
+    let mut document = document;
+    document["registrationTokenUid"] = json!(uid);
+    RegistrationToken::from_reader(serde_json::to_vec(&document).expect("token JSON").as_slice()).expect("token parses")
+}
+
+fn stores(temp: &tempfile::TempDir) -> (IdentityStore, CredentialStore) {
+    (
+        IdentityStore::new(temp.path().join("identity")),
+        CredentialStore::new(temp.path().join("credential")),
+    )
+}
+
+fn client(server: &TestServer, pki: &TestPki, timeout: Duration) -> ConnectClient {
+    ConnectClient::new(ConnectConfig {
+        endpoint: &server.endpoint,
+        root_ca_pem: pki.root_pem.as_bytes(),
+        timeout,
+    })
+    .expect("build Connect client")
+}
+
+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();
+    wire.as_object_mut().expect("response object").remove("uid");
+    wire.as_object_mut().expect("response object").remove("cluster");
+    (wire, stored)
+}
+
+fn write_stored_credential(path: &std::path::Path, response: &Value) {
+    let not_before = OffsetDateTime::parse(response["notBefore"].as_str().expect("notBefore"), &Rfc3339)
+        .expect("parse notBefore")
+        .unix_timestamp();
+    let not_after = OffsetDateTime::parse(response["notAfter"].as_str().expect("notAfter"), &Rfc3339)
+        .expect("parse notAfter")
+        .unix_timestamp();
+    let stored = json!({
+        "name": response["name"],
+        "uid": DEVICE_UID,
+        "protocolVersion": response["protocolVersion"],
+        "keyId": response["keyId"],
+        "certificateSerial": response["certificateSerial"],
+        "certificate": response["certificate"],
+        "certificateChain": response["certificateChain"],
+        "notBeforeUnix": not_before,
+        "notAfterUnix": not_after,
+    });
+    fs::write(path, serde_json::to_vec(&stored).expect("stored credential JSON")).expect("write credential");
+    set_owner_only(path);
+}
+
+#[cfg(unix)]
+fn set_owner_only(path: &std::path::Path) {
+    use std::os::unix::fs::PermissionsExt as _;
+    fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("set owner-only mode");
+}
+
+#[cfg(not(unix))]
+fn set_owner_only(_path: &std::path::Path) {}
+
+#[tokio::test]
+async fn registration_reuses_request_and_csr_after_timeout_and_restart() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let (identity_store, credential_store) = stores(&temp);
+    let identity = identity_store.load_or_create().expect("create identity");
+    let pki = TestPki::new();
+    let response = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 0x80);
+    let server = server(
+        &pki,
+        vec![
+            Reply::DelayedClose(Duration::from_millis(200)),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::CREATED, response),
+        ],
+    )
+    .await;
+    let first = client(&server, &pki, Duration::from_millis(80));
+    assert!(matches!(
+        first.register(&identity_store, &credential_store, &token()).await,
+        Err(ClientError::Unavailable { .. })
+    ));
+
+    let restarted = client(&server, &pki, Duration::from_secs(2));
+    let credential = restarted
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("restart replays completed exchange");
+    assert_eq!(credential.uid, DEVICE_UID);
+    assert_eq!(credential.certificate_serial, "80".repeat(16));
+
+    let seen = server.seen.lock().expect("seen lock");
+    assert_eq!(seen.len(), 4);
+    for request in &seen[1..] {
+        assert_eq!(request["requestId"], seen[0]["requestId"]);
+        assert_eq!(request["certificateRequest"], seen[0]["certificateRequest"]);
+    }
+}
+
+#[tokio::test]
+async fn registration_rejects_untrusted_or_misbound_credentials() {
+    for case in ["san", "chain", "key", "key_id", "cluster", "name"] {
+        let temp = tempfile::tempdir().expect("temp dir");
+        let (identity_store, credential_store) = stores(&temp);
+        let identity = identity_store.load_or_create().expect("create identity");
+        let pki = TestPki::new();
+        let mut response = match case {
+            "san" => pki.credential(&identity, "urn:rustfs:connect:device:0198f4b0-3c00-7e30-8f41-4a5b6c7d8e93", 2),
+            "chain" => TestPki::new().credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 2),
+            "key" => pki.credential(
+                &rustfs::connect::DeviceIdentity::generate(),
+                &format!("urn:rustfs:connect:device:{DEVICE_UID}"),
+                2,
+            ),
+            _ => pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 2),
+        };
+        match case {
+            "key_id" => response["keyId"] = json!("x509-deadbeef"),
+            "cluster" => response["cluster"] = json!(format!("organizations/{ORGANIZATION_UID}/clusters/other")),
+            "name" => {
+                response["name"] = json!(format!("organizations/{ORGANIZATION_UID}/clusters/other/clusterDevices/{DEVICE_UID}"))
+            }
+            _ => {}
+        }
+        let server = server(&pki, vec![Reply::Json(StatusCode::CREATED, response)]).await;
+        let error = client(&server, &pki, Duration::from_secs(2))
+            .register(&identity_store, &credential_store, &token())
+            .await
+            .expect_err("invalid returned identity must fail closed");
+        assert!(matches!(error, ClientError::Credential(_)));
+        assert!(!temp.path().join("credential/device.crt.json").exists());
+    }
+}
+
+#[tokio::test]
+async fn stored_credential_is_revalidated_before_reuse() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let (identity_store, credential_store) = stores(&temp);
+    let identity = identity_store.load_or_create().expect("create identity");
+    let pki = TestPki::new();
+    let issued = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 4);
+    let server = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
+    let client = client(&server, &pki, Duration::from_secs(2));
+    client
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+
+    let path = temp.path().join("credential/device.crt.json");
+    let mut stored: Value = serde_json::from_slice(&fs::read(&path).expect("read credential")).expect("credential JSON");
+    stored["certificateSerial"] = json!("00".repeat(16));
+    fs::write(&path, serde_json::to_vec(&stored).expect("credential JSON")).expect("tamper credential");
+    let error = client
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect_err("tampered stored credential must fail closed");
+    assert!(matches!(error, ClientError::Credential(_)));
+}
+
+#[tokio::test]
+async fn register_rejects_expired_and_not_yet_valid_stored_credentials() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let (identity_store, credential_store) = stores(&temp);
+    let identity = identity_store.load_or_create().expect("create identity");
+    let pki = TestPki::new();
+    let issued = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 9);
+    let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
+    let client = client(&registration, &pki, Duration::from_secs(2));
+    client
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+
+    let now = OffsetDateTime::now_utc().replace_nanosecond(0).expect("whole second");
+    let path = temp.path().join("credential/device.crt.json");
+    let expired = pki.credential_window(
+        &identity,
+        &format!("urn:rustfs:connect:device:{DEVICE_UID}"),
+        10,
+        now - time::Duration::days(2),
+        now - time::Duration::days(1),
+    );
+    write_stored_credential(&path, &expired);
+    assert!(matches!(
+        client.register(&identity_store, &credential_store, &token()).await,
+        Err(ClientError::CredentialExpired)
+    ));
+
+    let future = pki.credential_window(
+        &identity,
+        &format!("urn:rustfs:connect:device:{DEVICE_UID}"),
+        11,
+        now + time::Duration::hours(1),
+        now + time::Duration::hours(25),
+    );
+    write_stored_credential(&path, &future);
+    assert!(matches!(
+        client.register(&identity_store, &credential_store, &token()).await,
+        Err(ClientError::CredentialNotYetValid)
+    ));
+}
+
+#[tokio::test]
+async fn concurrent_rotation_retries_converge_and_promote_the_next_key() {
+    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 pki = TestPki::new();
+    let mut issued = pki.credential(&current, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 5);
+    issued["certificateChain"] = json!(
+        issued["certificateChain"]
+            .as_str()
+            .expect("certificate chain")
+            .trim_end_matches('\n')
+    );
+    let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
+    let registered = client(&registration, &pki, Duration::from_secs(2))
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+
+    let retries = server(
+        &pki,
+        vec![
+            Reply::DelayedClose(Duration::from_millis(200)),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+            Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})),
+        ],
+    )
+    .await;
+    let retry_client = client(&retries, &pki, Duration::from_millis(80));
+    let due = registered.not_after_unix - 8 * 60 * 60;
+    let (first, second) = tokio::join!(
+        retry_client.rotate_if_due(&identity_store, &credential_store, due),
+        retry_client.rotate_if_due(&identity_store, &credential_store, due)
+    );
+    assert!(matches!(first, Err(ClientError::Unavailable { .. })));
+    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");
+        for request in &seen[1..] {
+            assert_eq!(request["requestId"], seen[0]["requestId"]);
+            assert_eq!(request["certificateRequest"], seen[0]["certificateRequest"]);
+        }
+        (seen[0]["requestId"].clone(), seen[0]["certificateRequest"].clone())
+    };
+
+    let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read staged next key");
+    let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse next key");
+    assert_ne!(current.public_key_der(), next.public_key_der());
+    assert_eq!(
+        identity_store
+            .load()
+            .expect("load current key")
+            .expect("current key")
+            .public_key_der(),
+        current.public_key_der()
+    );
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt as _;
+        let mode = fs::metadata(temp.path().join("identity/device.key.next"))
+            .expect("next key metadata")
+            .permissions()
+            .mode()
+            & 0o7777;
+        assert_eq!(mode, 0o600);
+    }
+    let (rotated, _) = rotation_response(&pki, &next, 6);
+    let success = server_with_client_auth(
+        &pki,
+        vec![Reply::VerifiedRotation {
+            response: rotated,
+            current_public_key: current.public_key_der(),
+            current_certificate_fingerprint: certificate_fingerprint(&registered.certificate),
+            device_name: registered.name.clone(),
+        }],
+        true,
+    )
+    .await;
+    let success_client = client(&success, &pki, Duration::from_secs(2));
+    let (due_result, current_result) = tokio::join!(
+        success_client.rotate_if_due(&identity_store, &credential_store, due),
+        success_client.rotate_if_due(&identity_store, &credential_store, OffsetDateTime::now_utc().unix_timestamp())
+    );
+    let credential = due_result
+        .expect("retry rotation")
+        .or(current_result.expect("concurrent current-state check"))
+        .expect("exactly one rotation is due");
+    assert_eq!(credential.certificate_serial, "06".repeat(16));
+    let success_seen = success.seen.lock().expect("seen lock");
+    assert_eq!(success_seen.len(), 1, "the post-commit actor must not publish stale state");
+    assert_eq!(success_seen[0]["requestId"], request_id);
+    assert_eq!(success_seen[0]["certificateRequest"], certificate_request);
+    drop(success_seen);
+    assert_eq!(
+        identity_store
+            .load()
+            .expect("load key")
+            .expect("current key")
+            .public_key_der(),
+        next.public_key_der()
+    );
+    assert!(!temp.path().join("identity/device.key.next").exists());
+    assert!(!temp.path().join("credential/rotation.pending.json").exists());
+}
+
+#[tokio::test]
+async fn rotation_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 pki = TestPki::new();
+    let issued = pki.credential(&current, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 7);
+    let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
+    let registered = client(&registration, &pki, Duration::from_secs(2))
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+    let failed = server(&pki, vec![Reply::Json(StatusCode::SERVICE_UNAVAILABLE, json!({})); 3]).await;
+    let due = registered.not_after_unix - 8 * 60 * 60;
+    assert!(matches!(
+        client(&failed, &pki, Duration::from_secs(2))
+            .rotate_if_due(&identity_store, &credential_store, due)
+            .await,
+        Err(ClientError::Unavailable { .. })
+    ));
+
+    let pending_path = temp.path().join("credential/rotation.pending.json");
+    let pending = fs::read(&pending_path).expect("read pending state");
+    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 (_, stored) = rotation_response(&pki, &next, 8);
+    write_stored_credential(&temp.path().join("credential/device.crt.json"), &stored);
+
+    let idle = server(&pki, vec![]).await;
+    assert!(
+        client(&idle, &pki, Duration::from_secs(2))
+            .rotate_if_due(&identity_store, &credential_store, OffsetDateTime::now_utc().unix_timestamp())
+            .await
+            .expect("recover after credential save")
+            .is_none()
+    );
+    assert_eq!(
+        identity_store
+            .load()
+            .expect("load key")
+            .expect("current key")
+            .public_key_der(),
+        next.public_key_der()
+    );
+
+    fs::write(&pending_path, pending).expect("restore pending after key commit");
+    set_owner_only(&pending_path);
+    assert!(
+        client(&idle, &pki, Duration::from_secs(2))
+            .rotate_if_due(&identity_store, &credential_store, OffsetDateTime::now_utc().unix_timestamp())
+            .await
+            .expect("recover after key commit")
+            .is_none()
+    );
+    assert!(!pending_path.exists());
+}
+
+#[tokio::test]
+async fn pending_reenrollment_blocks_rotation_and_resumes_original_exchange() {
+    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 pki = TestPki::new();
+    let issued = pki.credential(&current, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13);
+    let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
+    let registered = client(&registration, &pki, Duration::from_secs(2))
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+
+    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))
+            .await,
+        Err(ClientError::Unavailable { .. })
+    ));
+
+    let pending_path = temp.path().join("credential/registration.pending.json");
+    let pending = fs::read(&pending_path).expect("read pending reenrollment");
+    let pending_document: Value = serde_json::from_slice(&pending).expect("pending JSON");
+    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 rotation = server(&pki, vec![]).await;
+    let error = client(&rotation, &pki, Duration::from_secs(2))
+        .rotate_if_due(&identity_store, &credential_store, registered.not_after_unix - 8 * 60 * 60)
+        .await
+        .expect_err("pending reenrollment blocks rotation");
+    assert!(matches!(error, ClientError::PendingRegistration));
+    assert!(rotation.seen.lock().expect("seen lock").is_empty());
+
+    let resumed = server(&pki, vec![Reply::Json(StatusCode::CREATED, enrolled)]).await;
+    let credential = client(&resumed, &pki, Duration::from_secs(2))
+        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("resume reenrollment");
+    assert_eq!(credential.certificate_serial, "0e".repeat(16));
+    let resumed_seen = resumed.seen.lock().expect("seen lock");
+    assert_eq!(resumed_seen.len(), 1);
+    assert_eq!(resumed_seen[0]["requestId"], pending_document["requestId"]);
+    assert_eq!(resumed_seen[0]["certificateRequest"], pending_document["certificateRequest"]);
+}
+
+#[tokio::test]
+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 pki = TestPki::new();
+    let issued = pki.credential(&current, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 13);
+    let registration = server(&pki, vec![Reply::Json(StatusCode::CREATED, issued)]).await;
+    client(&registration, &pki, Duration::from_secs(2))
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+
+    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))
+            .await,
+        Err(ClientError::Unavailable { .. })
+    ));
+
+    let pending_path = temp.path().join("credential/registration.pending.json");
+    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);
+    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));
+    let recovered = idle_client
+        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("recover after reenrollment credential save");
+    assert_eq!(recovered.certificate_serial, "0e".repeat(16));
+    assert_eq!(
+        identity_store
+            .load()
+            .expect("load key")
+            .expect("current key")
+            .public_key_der(),
+        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");
+    set_owner_only(&pending_path);
+    let recovered = idle_client
+        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("recover after reenrollment key commit");
+    assert_eq!(recovered.certificate_serial, "0e".repeat(16));
+    assert!(!pending_path.exists());
+    let recovered = idle_client
+        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("completed reenrollment is idempotent after pending cleanup");
+    assert_eq!(recovered.certificate_serial, "0e".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))
+            .await,
+        Err(ClientError::Unavailable { .. })
+    ));
+    assert_eq!(different.seen.lock().expect("seen lock").len(), 3);
+}
+
+#[test]
+fn registration_token_schema_is_strict_and_bounded() {
+    let mut document = serde_json::to_value(token_document()).expect("token document");
+    document["unexpected"] = json!(true);
+    assert!(matches!(
+        RegistrationToken::from_reader(serde_json::to_vec(&document).expect("token JSON").as_slice()),
+        Err(TokenError::Invalid(_))
+    ));
+    assert!(matches!(
+        RegistrationToken::from_reader(vec![b' '; 16 * 1024 + 1].as_slice()),
+        Err(TokenError::TooLarge)
+    ));
+    let mut malformed = token_document();
+    malformed["challengeNonce"] = json!("A".repeat(64));
+    assert!(matches!(
+        RegistrationToken::from_reader(serde_json::to_vec(&malformed).expect("token JSON").as_slice()),
+        Err(TokenError::Shape)
+    ));
+}
+
+#[tokio::test]
+async fn rotation_waits_for_threshold_and_stops_on_revocation() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let (identity_store, credential_store) = stores(&temp);
+    let identity = identity_store.load_or_create().expect("create identity");
+    let pki = TestPki::new();
+    let issued = pki.credential(&identity, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 3);
+    let rotation_server = server(
+        &pki,
+        vec![
+            Reply::Json(StatusCode::CREATED, issued),
+            Reply::Json(StatusCode::UNAUTHORIZED, json!({"details": [{"reason": "DEVICE_REVOKED"}]})),
+        ],
+    )
+    .await;
+    let connect = client(&rotation_server, &pki, Duration::from_secs(2));
+    let current = connect
+        .register(&identity_store, &credential_store, &token())
+        .await
+        .expect("register");
+    assert!(
+        connect
+            .rotate_if_due(&identity_store, &credential_store, current.not_before_unix)
+            .await
+            .expect("not due")
+            .is_none()
+    );
+    let error = connect
+        .rotate_if_due(&identity_store, &credential_store, current.not_after_unix - 8 * 60 * 60)
+        .await
+        .expect_err("revocation must stop rotation");
+    assert!(matches!(error, ClientError::AccessRevoked { .. }));
+    assert!(error.to_string().contains("ConnectClient::reenroll"));
+    assert_eq!(rotation_server.seen.lock().expect("seen lock").len(), 2);
+    let stored: Value =
+        serde_json::from_slice(&fs::read(temp.path().join("credential/device.crt.json")).expect("read stored credential"))
+            .expect("stored credential JSON");
+    assert_eq!(stored["certificateSerial"], current.certificate_serial);
+
+    let next_der = fs::read(temp.path().join("identity/device.key.next")).expect("read staged next key");
+    let next = rustfs::connect::DeviceIdentity::from_pkcs8_der(&next_der).expect("parse staged next key");
+    let enrolled = pki.credential(&next, &format!("urn:rustfs:connect:device:{DEVICE_UID}"), 12);
+    let reenrollment = server(&pki, vec![Reply::Json(StatusCode::CREATED, enrolled)]).await;
+    let fresh = client(&reenrollment, &pki, Duration::from_secs(2))
+        .reenroll(&identity_store, &credential_store, &token_with_uid(FRESH_TOKEN_UID))
+        .await
+        .expect("fresh token reenrolls revoked credential");
+    assert_eq!(fresh.certificate_serial, "0c".repeat(16));
+    assert_eq!(
+        identity_store
+            .load()
+            .expect("load identity")
+            .expect("identity")
+            .public_key_der(),
+        next.public_key_der()
+    );
+}
+
+#[test]
+fn unconfigured_connect_has_no_side_effects() {
+    let temp = tempfile::tempdir().expect("temp dir");
+    let directory = temp.path().join("connect");
+    assert!(
+        ConnectClient::from_optional_config(None)
+            .expect("unconfigured is valid")
+            .is_none()
+    );
+    assert!(!directory.exists());
+}