fix(heal): fence format repair during pool transitions

feat(connect): answer offline enrolment challenges without a network (#6335 )
2026-08-21 20:06:37 +00:00 · 2026-08-22 00:25:32 +08:00 · 2026-08-21 13:16:18 +00:00
24 changed files with 2364 additions and 31 deletions
@@ -866,7 +866,7 @@ impl BucketTargetSys {
            return Some(cli);
        }

-        // TODO(backlog): spawn an async task to proactively reload the replication target
+        // TODO: spawn a task to reload the target
        if self.is_reloading_target(bucket, arn).await {
            return None;
        }
@@ -454,7 +454,7 @@ impl S3PeerSys {
                }
            }
            topology_complete &= bucket_map.values().all(|count| *count >= quorum);
-            // TODO(backlog): integrate MRF backlog stats into scanner bucket listing
+            // TODO: MRF
        }

        let mut buckets: Vec<BucketInfo> = result_map.into_values().collect();
@@ -2406,7 +2406,7 @@ impl DiskAPI for RemoteDisk {
            return errors;
        }

-        // TODO(backlog): replace string errors with typed `StorageError` variants
+        // TODO: use Error not string

        let result = self
            .execute_with_timeout(
@@ -1675,7 +1675,7 @@ impl PoolMeta {
        self.load_no_lock(pool).await
    }

-    async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
+    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
    where
        S: EcstoreObjectIO,
    {
@@ -249,7 +249,7 @@ impl Sets {

        self.connect_disks().await;

-        // TODO(backlog): make monitor_and_connect interval configurable instead of hardcoded 15s
+        // TODO: config interval
        let mut interval = tokio::time::interval(Duration::from_secs(15));
        loop {
            tokio::select! {
@@ -5215,8 +5215,8 @@ impl LocalDisk {

        let cache = Cache::new(update_fn, Duration::from_secs(1), Opts::default());

-        // TODO(backlog): add O_DIRECT I/O support for performance-critical paths
-        // TODO(backlog): populate DiskInfo in constructor
+        // TODO: DIRECT support
+        // TODD: DiskInfo
        let mut disk = Self {
            root: root.clone(),
            publication_root,
@@ -5751,7 +5751,7 @@ impl LocalDisk {

        // return Ok(());

-        // TODO(backlog): make disk space checks and trash cleanup event-driven instead of poll-based
+        // TODO: async notifications for disk space checks and trash cleanup

        let trash_path = self.io_get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
        // if let Some(parent) = trash_path.parent() {
@@ -5997,7 +5997,7 @@ impl LocalDisk {

    #[hotpath::measure(impl_type = "LocalDisk")]
    async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
-        // TODO(backlog): add configurable timeout for read_all_data operations
+        // TODO: timeout support
        let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
        Ok(data)
    }
@@ -6674,7 +6674,7 @@ impl LocalDisk {
            return Ok(());
        }

-        // TODO(backlog): add directory listing lock to prevent concurrent enumeration
+        // TODO: add lock

        let stall = opts.stall_timeout_duration();

@@ -8796,7 +8796,7 @@ impl DiskAPI for LocalDisk {
        Ok(entries)
    }

-    // TODO(backlog): support io.writer cancellation and early termination in walk_dir
+    // FIXME: TODO: io.writer TODO cancel
    #[tracing::instrument(level = "trace", skip_all)]
    async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
        self.wait_for_startup_cleanup().await;
@@ -9880,7 +9880,7 @@ impl DiskAPI for LocalDisk {
                );
                return Err(e);
            }
-            // TODO(backlog): add post-setup disk health verification
+            // TODO: health check
        }
        Ok(())
    }
@@ -249,7 +249,7 @@ impl PoolEndpointList {
            endpoint.set_set_index(0);
            endpoint.set_disk_index(0);

-            // TODO(backlog): check for cross-device mounts in single-drive setup
+            // TODO Check for cross device mounts if any.

            return Ok(Self {
                inner: vec![Endpoints::from(vec![endpoint])],
@@ -264,7 +264,7 @@ impl PoolEndpointList {
                // Convert args to endpoints
                let mut eps = Endpoints::try_from(set_layout.as_slice())?;

-                // TODO(backlog): check for cross-device mounts in multi-pool setup
+                // TODO Check for cross device mounts if any.

                for (disk_idx, ep) in eps.as_mut().iter_mut().enumerate() {
                    ep.set_pool_index(pool_idx);
@@ -1091,7 +1091,7 @@ impl ObjectInfo {
                    }
                };

-                // TODO(backlog): handle VersionPurgeStatus in object listing
+                // TODO:VersionPurgeStatus
                let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
                objects.push(ObjectInfo::from_file_info(&fi, bucket, &entry.name, versioned));

@@ -1575,7 +1575,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
        let parts_metadata = vec![fi.clone(); disks.len()];

        if !user_defined.contains_key("content-type") {
-            // TODO(backlog): detect content-type from part data when header is missing
+            // TODO: get content-type
        }

        if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS)
@@ -1971,7 +1971,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
            }

-            // TODO(backlog): integrate encryption verification during complete multipart
+            // TODO: crypto

            if (i < uploaded_parts.len() - 1)
                && !(opts.data_movement && ext_part.actual_size < 0)
@@ -6161,7 +6161,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {

        join_all(rollback_futures).await;

-        // TODO(backlog): support partial object deletion for multi-part objects
+        // TODO: add_partial

        if let Some(api) = opts.tier_delete_journal_api.as_ref() {
            for (idx, je) in persisted_journal_entries {
@@ -6371,7 +6371,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
            }
        }

-        // TODO(backlog): integrate lifecycle evaluation before object deletion
+        // TODO: Lifecycle

        let mut version_found = true;
        // delete_object_version below derives its own majority quorum from the
@@ -6465,7 +6465,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                mark_deleted: mark_delete,
                mod_time: Some(mod_time),
                replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta),
-                ..Default::default() // TODO(backlog): populate transition state on delete markers
+                ..Default::default() // TODO: Transition
            };

            fi.set_tier_free_version_id(&find_vid.to_string());
@@ -601,7 +601,7 @@ impl ECStore {

    #[instrument(skip(self))]
    pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
-        // TODO(backlog): support cached bucket listing via opts.cached
+        // TODO: opts.cached

        let mut buckets = self.peer_sys.list_bucket(opts).await?;

@@ -13,7 +13,12 @@
 // limitations under the License.

 use super::*;
+use crate::core::pools::POOL_META_NAME;
+use crate::services::rebalance::{REBAL_META_NAME, RebalStatus};
+use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::heal::HealOperations as _;
+use crate::storage_api_contracts::namespace::NamespaceLocking as _;
+use rustfs_lock::NamespaceLockGuard;
 use tracing::trace;

 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -30,7 +35,119 @@ fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error {
    )
 }

+#[derive(Debug, Clone, Copy)]
+enum HealFormatPoolSkip {
+    Completed,
+    Retryable,
+}
+
+fn classify_heal_format_pool(
+    pool_idx: usize,
+    pool_cmd_line: &str,
+    pool_meta: &PoolMeta,
+    rebalance_meta: Option<&RebalanceMeta>,
+) -> Option<HealFormatPoolSkip> {
+    let Some(pool) = pool_meta.pools.get(pool_idx) else {
+        return Some(HealFormatPoolSkip::Retryable);
+    };
+
+    if pool.id != pool_idx || pool_cmd_line.is_empty() || pool.cmd_line.is_empty() || pool.cmd_line != pool_cmd_line {
+        return Some(HealFormatPoolSkip::Retryable);
+    }
+
+    if let Some(decommission) = pool.decommission.as_ref() {
+        if decommission.complete {
+            return Some(HealFormatPoolSkip::Completed);
+        }
+        if decommission.failed || decommission.canceled || decommission.queued || pool_meta.is_suspended(pool_idx) {
+            return Some(HealFormatPoolSkip::Retryable);
+        }
+    }
+
+    if let Some(meta) = rebalance_meta {
+        let Some(pool_stats) = meta.pool_stats.get(pool_idx) else {
+            return Some(HealFormatPoolSkip::Retryable);
+        };
+        if pool_stats.info.stopping || (pool_stats.participating && pool_stats.info.status == RebalStatus::Started) {
+            return Some(HealFormatPoolSkip::Retryable);
+        }
+    }
+
+    None
+}
+
+fn heal_format_pool_skip_error(skip: HealFormatPoolSkip) -> Error {
+    match skip {
+        HealFormatPoolSkip::Completed => StorageError::NoHealRequired,
+        HealFormatPoolSkip::Retryable => StorageError::SlowDown,
+    }
+}
+
+fn heal_format_fence_lost_error() -> Error {
+    StorageError::SlowDown
+}
+
 impl ECStore {
+    async fn acquire_heal_format_fence(
+        &self,
+    ) -> Result<(NamespaceLockGuard, NamespaceLockGuard, PoolMeta, Option<RebalanceMeta>)> {
+        let metadata_pool = self
+            .pools
+            .first()
+            .cloned()
+            .ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
+
+        // Metadata fence order is part of the decommission/rebalance protocol:
+        // pool.bin must always be acquired before rebalance.bin.
+        let pool_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
+        let pool_guard = pool_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+        let rebalance_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
+        let rebalance_guard = rebalance_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+
+        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        let mut pool_meta = PoolMeta::default();
+        pool_meta.load_no_lock(metadata_pool.clone()).await?;
+        if pool_meta.pools.len() != self.pools.len()
+            || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
+                pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
+            })
+        {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        let mut rebalance_meta = RebalanceMeta::new();
+        let rebalance_meta = match rebalance_meta
+            .load_with_opts(
+                metadata_pool,
+                ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+        {
+            Ok(()) => Some(rebalance_meta),
+            Err(Error::ConfigNotFound) => None,
+            Err(err) => return Err(err),
+        };
+
+        if rebalance_meta
+            .as_ref()
+            .is_some_and(|meta| meta.pool_stats.len() != self.pools.len())
+        {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+            return Err(heal_format_fence_lost_error());
+        }
+
+        Ok((pool_guard, rebalance_guard, pool_meta, rebalance_meta))
+    }
+
    fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result<Vec<Arc<Sets>>> {
        match opts.pool {
            Some(pool_idx) => Ok(vec![
@@ -52,8 +169,24 @@ impl ECStore {
        };

        let mut count_no_heal = 0;
+        let mut count_completed = 0;
        let mut first_error = None;
-        for pool in self.pools.iter() {
+        for (pool_idx, pool) in self.pools.iter().enumerate() {
+            let (pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?;
+            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+                first_error.get_or_insert(heal_format_fence_lost_error());
+                break;
+            }
+            if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref())
+            {
+                if matches!(skip, HealFormatPoolSkip::Completed) {
+                    count_completed += 1;
+                } else {
+                    first_error.get_or_insert(heal_format_pool_skip_error(skip));
+                }
+                continue;
+            }
+
            let (mut result, err) = pool.heal_format(dry_run).await?;
            if let Some(err) = err {
                match err {
@@ -69,11 +202,18 @@ impl ECStore {
            r.set_count += result.set_count;
            r.before.drives.append(&mut result.before.drives);
            r.after.drives.append(&mut result.after.drives);
+
+            // Sets::heal_format cannot observe this guard before each disk write;
+            // fail closed after the call if the lease was lost during format IO.
+            if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
+                first_error.get_or_insert(heal_format_fence_lost_error());
+                break;
+            }
        }
        if let Some(err) = first_error {
            return Ok((r, Some(err)));
        }
-        if count_no_heal == self.pools.len() {
+        if count_no_heal + count_completed == self.pools.len() {
            info!(
                event = EVENT_HEAL_FORMAT_COMPLETED,
                component = LOG_COMPONENT_ECSTORE,
@@ -300,6 +440,7 @@ mod tests {
    use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
    use crate::disk::{DiskOption, format::FormatV3, new_disk};
    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
    use crate::store::init_format::{load_format_erasure, save_format_file};

    async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
@@ -347,6 +488,164 @@ mod tests {
        }
    }

+    fn pool_meta_with_decommission(info: PoolDecommissionInfo) -> PoolMeta {
+        PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: OffsetDateTime::UNIX_EPOCH,
+                decommission: Some(info),
+            }],
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn heal_format_pool_state_barriers_are_classified() {
+        let active = pool_meta_with_decommission(PoolDecommissionInfo {
+            start_time: Some(OffsetDateTime::UNIX_EPOCH),
+            ..Default::default()
+        });
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &active, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        for info in [
+            PoolDecommissionInfo {
+                failed: true,
+                ..Default::default()
+            },
+            PoolDecommissionInfo {
+                canceled: true,
+                ..Default::default()
+            },
+        ] {
+            assert!(matches!(
+                classify_heal_format_pool(0, "pool-0", &pool_meta_with_decommission(info), None),
+                Some(HealFormatPoolSkip::Retryable)
+            ));
+        }
+
+        let completed = pool_meta_with_decommission(PoolDecommissionInfo {
+            complete: true,
+            ..Default::default()
+        });
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &completed, None),
+            Some(HealFormatPoolSkip::Completed)
+        ));
+    }
+
+    #[test]
+    fn heal_format_pool_rebalance_barriers_and_identity_are_fail_closed() {
+        let identity_meta = pool_meta_with_decommission(PoolDecommissionInfo::default());
+        let rebalance = RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&rebalance)),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let stopping = RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                info: RebalanceInfo {
+                    stopping: true,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping)),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let identity = pool_meta_with_decommission(PoolDecommissionInfo::default());
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-new", &identity, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let identity_without_decommission = PoolMeta {
+            pools: vec![PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: OffsetDateTime::UNIX_EPOCH,
+                decommission: None,
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-new", &identity_without_decommission, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        assert!(matches!(
+            classify_heal_format_pool(0, "", &identity_meta, None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &PoolMeta::default(), None),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+
+        let stopped = RebalanceMeta {
+            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Stopped,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopped)).is_none());
+
+        let stopping_after_stop = RebalanceMeta {
+            stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    stopping: true,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping_after_stop)),
+            Some(HealFormatPoolSkip::Retryable)
+        ));
+    }
+
+    #[test]
+    fn skipped_heal_format_pool_is_never_reported_as_success() {
+        assert!(matches!(
+            heal_format_pool_skip_error(HealFormatPoolSkip::Retryable),
+            StorageError::SlowDown
+        ));
+        assert!(matches!(
+            heal_format_pool_skip_error(HealFormatPoolSkip::Completed),
+            StorageError::NoHealRequired
+        ));
+    }
+
    #[tokio::test]
    async fn heal_object_pool_scope_selects_only_requested_pool() {
        let store = minimal_heal_store().await;
@@ -615,6 +914,18 @@ mod tests {
            bucket_fence_registry: std::sync::Arc::default(),
        };

+        let err = store
+            .handle_heal_format(false)
+            .await
+            .expect_err("missing pool metadata must fail closed before format writes");
+        assert!(matches!(err, StorageError::SlowDown));
+
+        let pool_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
+        pool_meta
+            .save(store.pools.clone())
+            .await
+            .expect("pool metadata should be persisted before format heal");
+
        let (result, err) = store
            .handle_heal_format(false)
            .await
@@ -628,5 +939,22 @@ mod tests {
            .await
            .expect("the later pool should be healed despite the first pool error");
        assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
+
+        let mut completed_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
+        for status in &mut completed_meta.pools {
+            status.decommission = Some(PoolDecommissionInfo {
+                complete: true,
+                ..Default::default()
+            });
+        }
+        completed_meta
+            .save(store.pools.clone())
+            .await
+            .expect("completed pool metadata should be persisted");
+        let (_, err) = store
+            .handle_heal_format(false)
+            .await
+            .expect("completed pools should be reported as a no-op");
+        assert!(matches!(err, Some(StorageError::NoHealRequired)));
    }
 }
@@ -4673,7 +4673,7 @@ async fn gather_results(
            entry.name = entry.name.replace("\\", "/");
        }

-        // TODO(backlog): integrate rx.recv() for incremental listing results
+        // TODO: rx.recv()

        if let Some(marker) = &opts.marker
            && ((!opts.include_marker && &entry.name <= marker) || (opts.include_marker && &entry.name < marker))
@@ -4703,7 +4703,7 @@ async fn gather_results(
            continue;
        }

-        // TODO(backlog): integrate lifecycle evaluation during object listing
+        // TODO: Lifecycle

        entries.push(Some(entry));
        candidate_entries += 1;
@@ -332,7 +332,7 @@ impl ECStore {
        let expected_incarnation_id = opts.expected_bucket_incarnation_id;

        if request.prefix.is_empty() {
-            // TODO(backlog): return cached multipart listing when prefix is empty
+            // TODO: return from cache
        }

        if self.single_pool() {
@@ -610,7 +610,7 @@ impl ECStore {
        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
        let opts = &opts;

-        // TODO(backlog): defer DeleteUploadID to background for faster abort response
+        // TODO: defer DeleteUploadID

        if self.single_pool() {
            return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
@@ -385,7 +385,7 @@ impl ECStore {
    }

    pub(super) async fn is_suspended(&self, idx: usize) -> bool {
-        // TODO(backlog): acquire pool metadata lock for consistent suspension check
+        // TODO: LOCK

        let pool_meta = self.pool_meta.read().await;

@@ -231,6 +231,10 @@ impl HealTask {
                            "Heal erasure set format repair skipped because no format heal was required"
                        );
                    } else {
+                        let error = e;
+                        if error.is_recoverable_heal() {
+                            return Err(error);
+                        }
                        error!(
                            target: "rustfs::heal::task",
                            event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -239,7 +243,7 @@ impl HealTask {
                            task_id = %self.id,
                            set_disk_id,
                            result = "format_failed",
-                            error = %e,
+                            error = %error,
                            "Heal erasure set failed"
                        );
                        {
@@ -247,7 +251,7 @@ impl HealTask {
                            progress.update_progress(4, 4, 0, 0);
                        }
                        return Err(Error::TaskExecutionFailed {
-                            message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
+                            message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
                        });
                    }
                } else {
@@ -284,6 +288,9 @@ impl HealTask {
            Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
            Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
            Err(e) => {
+                if e.is_recoverable_heal() {
+                    return Err(e);
+                }
                error!(
                    target: "rustfs::heal::task",
                    event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -547,6 +547,7 @@ struct MockStorage {
    heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
    heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
    format_no_heal_required: Mutex<bool>,
+    format_error: Mutex<Option<Error>>,
    global_format_calls: Mutex<u32>,
    replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
    replacement_targets_ready: Mutex<bool>,
@@ -867,6 +868,9 @@ impl HealStorageAPI for MockStorage {

    async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
        *self.global_format_calls.lock().unwrap() += 1;
+        if let Some(error) = self.format_error.lock().unwrap().take() {
+            return Err(error);
+        }
        let no_heal_required = *self.format_no_heal_required.lock().unwrap();
        if no_heal_required {
            Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2052,6 +2056,30 @@ async fn test_erasure_set_heal_continues_after_format_no_heal_required() {
    );
 }

+#[tokio::test]
+async fn erasure_set_format_slowdown_is_propagated() {
+    let storage = Arc::new(MockStorage {
+        format_error: Mutex::new(Some(Error::Storage(EcstoreError::SlowDown))),
+        ..Default::default()
+    });
+    let request = HealRequest::new(
+        HealType::ErasureSet {
+            buckets: Vec::new(),
+            set_disk_id: "pool_0_set_0".to_string(),
+        },
+        HealOptions::default(),
+        HealPriority::Normal,
+    );
+    let task = HealTask::from_request(request, storage);
+
+    let error = task
+        .execute()
+        .await
+        .expect_err("format SlowDown must remain recoverable for the task manager");
+
+    assert!(matches!(error, Error::Storage(EcstoreError::SlowDown)));
+}
+
 #[tokio::test]
 async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
    let temp = TempDir::new().expect("temporary directory should be created");
@@ -245,6 +245,18 @@ impl TestECStoreEnvBuilder {
            .await
            .expect("build test ECStore");

+        // The production bootstrap only persists pool.bin from the elected
+        // first cluster node.  Test stores intentionally have no cluster
+        // election, but heal-format still requires that durable fence before
+        // it can write any disk format.  Materialize the validated topology
+        // here so the shared fixture models a ready single-node store.
+        let mut pool_meta = ecstore.pool_meta.read().await.clone();
+        pool_meta.dont_save = false;
+        pool_meta
+            .save(ecstore.pools.clone())
+            .await
+            .expect("persist test pool metadata");
+
        if self.init_bucket_metadata {
            let buckets_list = ecstore
                .list_bucket(&BucketOptions {
@@ -17,6 +17,220 @@
 //! This binary shares RustFS's existing subcommand dispatcher and provides the
 //! documented entry point for offline tooling such as `inspect bucket-meta`.

-fn main() {
+use std::fs;
+use std::io::{Read as _, Write as _};
+use std::path::{Path, PathBuf};
+use std::process::ExitCode;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use rustfs::connect::offline::{OfflineEnrollment, OfflineKeyStore};
+
+/// Owner read/write only. The response names the key being enrolled and the
+/// challenge it answers; neither belongs to anyone else on the machine.
+#[cfg(unix)]
+const RESPONSE_MODE: u32 = 0o600;
+
+const USAGE: &str = "\
+Usage: rustfs-cli connect offline enroll --challenge <path|-> --output <path> [--key-dir <path>]
+
+Answers a Connect offline enrolment challenge without a network. Reads the
+challenge from a file or from stdin when the path is `-`, verifies it against the
+enrolment root compiled into this binary, mints the key being enrolled on first
+use, and writes the signed response.
+
+No secret is ever accepted on the command line.
+";
+
+fn main() -> ExitCode {
+    let arguments: Vec<String> = std::env::args().skip(1).collect();
+
+    // Offline enrolment is handled before the server dispatcher is reached, and
+    // the reason is the surface's whole point: `run_process` builds a Tokio
+    // runtime and enters the server's async main. An air-gapped enrolment must
+    // not start a runtime, a task, or anything that could open a socket, so the
+    // two paths cannot share an entry.
+    if matches!(
+        arguments.first().map(String::as_str),
+        Some("connect") if matches!(arguments.get(1).map(String::as_str), Some("offline"))
+    ) {
+        return match run_offline(&arguments[2..]) {
+            Ok(()) => ExitCode::SUCCESS,
+            Err(message) => {
+                eprintln!("rustfs-cli: {message}");
+                ExitCode::FAILURE
+            }
+        };
+    }
+
    rustfs::startup_entrypoint::run_process();
+
+    ExitCode::SUCCESS
+}
+
+fn run_offline(arguments: &[String]) -> Result<(), String> {
+    match arguments.first().map(String::as_str) {
+        Some("enroll") => enroll(&arguments[1..]),
+        Some(other) => Err(format!("unknown offline subcommand `{other}`\n\n{USAGE}")),
+        None => Err(format!("missing offline subcommand\n\n{USAGE}")),
+    }
+}
+
+fn enroll(arguments: &[String]) -> Result<(), String> {
+    let mut challenge_path: Option<String> = None;
+    let mut output_path: Option<String> = None;
+    let mut key_directory: Option<String> = None;
+
+    let mut index = 0;
+    while index < arguments.len() {
+        let flag = arguments[index].as_str();
+        let take_value = |name: &str| -> Result<String, String> {
+            arguments
+                .get(index + 1)
+                .cloned()
+                .ok_or_else(|| format!("`{name}` needs a value\n\n{USAGE}"))
+        };
+
+        match flag {
+            "--challenge" => challenge_path = Some(take_value("--challenge")?),
+            "--output" => output_path = Some(take_value("--output")?),
+            "--key-dir" => key_directory = Some(take_value("--key-dir")?),
+            "-h" | "--help" => {
+                println!("{USAGE}");
+                return Ok(());
+            }
+            other => return Err(format!("unknown option `{other}`\n\n{USAGE}")),
+        }
+
+        index += 2;
+    }
+
+    let challenge_path = challenge_path.ok_or_else(|| format!("`--challenge` is required\n\n{USAGE}"))?;
+    let output_path = output_path.ok_or_else(|| format!("`--output` is required\n\n{USAGE}"))?;
+    let key_directory = key_directory.unwrap_or_else(|| ".".to_string());
+
+    let challenge = read_challenge(&challenge_path)?;
+
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(|_| "the system clock is before the Unix epoch".to_string())?
+        .as_secs() as i64;
+
+    let verified = OfflineEnrollment::verify_challenge(&challenge, now).map_err(|error| error.to_string())?;
+
+    // First use mints the key; a retry answers with the one already enrolled,
+    // because the operator may already be carrying a response naming it.
+    let key = OfflineKeyStore::new(&key_directory)
+        .load_or_create()
+        .map_err(|error| error.to_string())?;
+
+    let mut device_nonce = [0u8; 32];
+    getrandom(&mut device_nonce)?;
+
+    let response = OfflineEnrollment::build_response(&verified, &key, &device_nonce, now).map_err(|error| error.to_string())?;
+
+    write_response(Path::new(&output_path), &response)?;
+
+    Ok(())
+}
+
+/// Reads the challenge from a file, or from stdin when the path is `-`.
+///
+/// A challenge is not a secret — it is signed, public, and carried in by hand —
+/// so accepting a path is safe. The response's key never arrives this way.
+fn read_challenge(path: &str) -> Result<Vec<u8>, String> {
+    if path == "-" {
+        let mut buffer = Vec::new();
+        std::io::stdin()
+            .read_to_end(&mut buffer)
+            .map_err(|error| format!("cannot read the challenge from stdin: {error}"))?;
+        return Ok(buffer);
+    }
+
+    fs::read(path).map_err(|error| format!("cannot read the challenge at {path}: {error}"))
+}
+
+/// Writes the response durably and atomically at mode 0600.
+///
+/// Not the no-clobber publish `IdentityStore` performs for a key: an operator
+/// who reruns an enrolment expects the response file to be replaced, whereas a
+/// second key would strand the first. Same durability, deliberately different
+/// publication rule.
+fn write_response(path: &Path, response: &[u8]) -> Result<(), String> {
+    let parent = path.parent().filter(|parent| !parent.as_os_str().is_empty());
+    let temporary: PathBuf = match parent {
+        Some(parent) => parent.join(format!(".{}.tmp", file_name(path))),
+        None => PathBuf::from(format!(".{}.tmp", file_name(path))),
+    };
+
+    let mut options = fs::OpenOptions::new();
+    options.write(true).create(true).truncate(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt as _;
+        options.mode(RESPONSE_MODE);
+    }
+
+    let write = (|| -> std::io::Result<()> {
+        let mut file = options.open(&temporary)?;
+        file.write_all(response)?;
+
+        // The umask can only narrow the creation mode, so set the exact mode
+        // before the bytes become durable.
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt as _;
+            file.set_permissions(fs::Permissions::from_mode(RESPONSE_MODE))?;
+        }
+
+        file.sync_all()
+    })();
+
+    if let Err(error) = write {
+        let _ = fs::remove_file(&temporary);
+        return Err(format!("cannot write the response to {}: {error}", path.display()));
+    }
+
+    fs::rename(&temporary, path).map_err(|error| {
+        let _ = fs::remove_file(&temporary);
+        format!("cannot publish the response at {}: {error}", path.display())
+    })?;
+
+    if let Some(parent) = parent {
+        sync_directory(parent);
+    }
+
+    Ok(())
+}
+
+fn file_name(path: &Path) -> String {
+    path.file_name()
+        .map(|name| name.to_string_lossy().into_owned())
+        .unwrap_or_else(|| "response".to_string())
+}
+
+/// Fsync the directory so the renamed entry survives power loss. Directories
+/// cannot be opened for syncing on Windows, where this is a no-op.
+fn sync_directory(directory: &Path) {
+    #[cfg(unix)]
+    {
+        if let Ok(handle) = fs::File::open(directory) {
+            let _ = handle.sync_all();
+        }
+    }
+    #[cfg(not(unix))]
+    let _ = directory;
+}
+
+/// Fills `buffer` with operating-system randomness.
+///
+/// The device nonce must be unpredictable: it is what stops a captured response
+/// being replayed as a fresh one. Sourced through p256's pinned rand_core 0.6
+/// rather than the workspace `rand` 0.10, matching `identity.rs`; the two are
+/// different crate versions and only the pinned one is on p256's own path.
+fn getrandom(buffer: &mut [u8]) -> Result<(), String> {
+    use p256::elliptic_curve::rand_core::{OsRng, RngCore as _};
+
+    OsRng
+        .try_fill_bytes(buffer)
+        .map_err(|error| format!("the operating system random source failed: {error}"))
 }
@@ -27,6 +27,8 @@

 pub mod identity;
 pub mod identity_store;
+pub mod offline;

 pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript};
 pub use identity_store::{IdentityStore, StoreError};
+pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge};
@@ -0,0 +1,684 @@
+// 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.
+
+//! Challenge verification and response production for offline enrolment.
+//!
+//! Two invariants carry the security of this surface and both are easy to break
+//! by accident:
+//!
+//! - Every signature is checked over the octets that arrived, never over a
+//!   re-serialised document. Parsing happens only to route the verification, and
+//!   nothing a parse yields is believed until the signature over those same
+//!   octets has verified.
+//! - The enrolment root is the constant in this file. It is never taken from a
+//!   challenge, a configuration file, or an operator prompt, so there is no
+//!   trust-on-first-use path an operator could be talked into.
+//!
+//! The order of the checks in [`OfflineEnrollment::verify_challenge`] is frozen
+//! by `verificationOrder.enrollmentChallenge` in
+//! `protocol/agent/v1/fixtures/offline-enrollment/trust-model.json`, and the
+//! signature encoding, the domain separation tags, and every rejection reason
+//! are frozen beside it. Reordering the checks changes which reason a given
+//! artifact produces, which is itself part of the contract.
+
+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 _, Verifier as _};
+use p256::ecdsa::{Signature, SigningKey, VerifyingKey};
+use p256::pkcs8::DecodePrivateKey as _;
+use serde::{Deserialize, Serialize};
+use sha2::{Digest as _, Sha256};
+use time::{Date, Month, OffsetDateTime, PrimitiveDateTime, Time};
+
+use crate::connect::identity::DeviceIdentity;
+
+/// The hosted enrolment root, compiled in. Both halves are pinned: the
+/// fingerprint identifies the root, and the point is what actually verifies the
+/// first link, so a build cannot be pointed at a different key by supplying one.
+const PINNED_ROOT_KEY_ID: &str = "df22e2806112debbe953672aafa186d699af0e97dd3fd2b09fa8359005fe348f";
+const PINNED_ROOT_PUBLIC_KEY: &str = "BFfx-K-FfEA5nK_Rz3IHacvRCkJyQ7JOd1geLyU6HKRZDgNezmVuKhvJ22VhemyjV__Gshk8JGGqOBzYPMD0p6s";
+
+/// Domain separation tags. A document that verifies under one of these must not
+/// be accepted for another artifact type, so the tag is part of the signature
+/// input rather than a property of the caller.
+const TAG_TRUST_LINK: &[u8] = b"rustfs-offline-trust-link-v1";
+const TAG_CHALLENGE: &[u8] = b"rustfs-offline-enrollment-challenge-v1";
+const TAG_RESPONSE: &[u8] = b"rustfs-offline-enrollment-response-v1";
+
+/// The single octet between the tag and the signed document.
+const DOMAIN_SEPARATOR: u8 = 0x00;
+
+const SIGNATURE_ALGORITHM: &str = "ES256";
+const PROTOCOL_VERSION: &str = "v1";
+const FORMAT_TRUST_LINK: &str = "rustfs.connect.offline.trustLink/1";
+const FORMAT_CHALLENGE: &str = "rustfs.connect.offline.enrollmentChallenge/1";
+const FORMAT_RESPONSE: &str = "rustfs.connect.offline.enrollmentResponse/1";
+
+/// DER SubjectPublicKeyInfo header for an uncompressed P-256 point. A keyId is
+/// the SHA-256 of this prefix followed by the 65 octet point, so the prefix is
+/// also how a device public key is recovered from its own DER encoding.
+const SPKI_PREFIX: [u8; 26] = [
+    0x30, 0x59, 0x30, 0x13, 0x06, 0x07, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x02, 0x01, 0x06, 0x08, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x03,
+    0x01, 0x07, 0x03, 0x42, 0x00,
+];
+
+/// Order of the P-256 group, and half of it. `r` and `s` must lie in `[1, n)`,
+/// and `s` additionally in `[1, n/2]`: ECDSA admits both `s` and `n - s`, and a
+/// signature with two spellings cannot serve as an artifact identity. Every
+/// ECDSA library accepts the malleated form, so the encoding layer rejects it.
+const GROUP_ORDER: [u8; 32] = [
+    0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xbc, 0xe6, 0xfa, 0xad, 0xa7,
+    0x17, 0x9e, 0x84, 0xf3, 0xb9, 0xca, 0xc2, 0xfc, 0x63, 0x25, 0x51,
+];
+const MAX_S: [u8; 32] = [
+    0x7f, 0xff, 0xff, 0xff, 0x80, 0x00, 0x00, 0x00, 0x7f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xde, 0x73, 0x7d, 0x56, 0xd3,
+    0x8b, 0xcf, 0x42, 0x79, 0xdc, 0xe5, 0x61, 0x7e, 0x31, 0x92, 0xa8,
+];
+
+const SCALAR_OCTETS: usize = 32;
+const SIGNATURE_OCTETS: usize = 64;
+/// 64 octets as unpadded base64url. The length is checked before decoding so
+/// that `=` padding, the standard alphabet, DER, and a truncated value are all
+/// refused rather than repaired.
+const SIGNATURE_VALUE_CHARS: usize = 86;
+
+const PUBLIC_KEY_OCTETS: usize = 65;
+const PUBLIC_KEY_CHARS: usize = 87;
+/// SEC1 tag of an uncompressed point. Compressed and hybrid forms are refused.
+const UNCOMPRESSED_POINT: u8 = 0x04;
+
+const TIMESTAMP_CHARS: usize = 20;
+
+/// The chain is exactly two links: a pinned root issues the intermediate, and
+/// the intermediate issues the signing key. Roles are positional and the
+/// enumeration is closed.
+const CHAIN_LINK_COUNT: usize = 2;
+const CHAIN_ROLES: [&str; CHAIN_LINK_COUNT] = ["intermediate", "signing"];
+
+/// Skew allowed on the challenge window. A device may have no synchronised
+/// clock at all, so its own reading of "now" is advisory.
+const CLOCK_SKEW_TOLERANCE: i64 = 300;
+
+/// Longest life a challenge may claim. The issuer sets both ends of its own
+/// window, so the protocol bound is applied on top of the declared expiry
+/// rather than trusted from it.
+const MAX_CHALLENGE_LIFETIME: i64 = 604_800;
+
+/// A challenge that verified, with the fields the response has to echo.
+///
+/// Construction is the proof: a value of this type only exists after the chain
+/// closed on the pinned root and the challenge signature verified over the
+/// received octets.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct VerifiedChallenge {
+    pub challenge_id: String,
+    pub organization_name: String,
+    pub cluster_name: String,
+    pub nonce: String,
+    pub issued_at: String,
+    pub expires_at: String,
+    pub connect_key_id: String,
+    /// The signature value of the challenge, verbatim. It binds a response to
+    /// the one challenge it answers, so it is carried rather than recomputed.
+    pub challenge_proof: String,
+}
+
+/// Why an offline enrolment artifact was refused.
+///
+/// The variants are the frozen `reason` vocabulary of
+/// `fixtures/offline-enrollment/error-codes.json`, which spans both halves of
+/// the exchange. The device half implemented here produces the encoding, chain,
+/// version, and freshness reasons; the reasons that describe a response being
+/// evaluated against stored state — [`Self::ChallengeUnknown`],
+/// [`Self::ChallengeProofInvalid`], [`Self::DeviceProofInvalid`],
+/// [`Self::EnrollmentReplayed`], [`Self::OrganizationMismatch`], and
+/// [`Self::ClusterMismatch`] — are Connect's to raise and are named here so the
+/// two sides share one vocabulary.
+///
+/// No variant carries a payload: a rejection must never disclose key material,
+/// signature octets, nonces, or document bytes.
+#[derive(Debug, PartialEq, Eq, thiserror::Error)]
+pub enum EnrollmentError {
+    #[error("protocolVersion is missing, malformed, or names an unsupported major version")]
+    UnsupportedProtocol,
+
+    #[error("formatVersion is not a supported offline enrollment format")]
+    UnsupportedFormat,
+
+    #[error("the signature is not 64 octets of fixed-width r||s in unpadded base64url")]
+    SignatureMalformed,
+
+    #[error("the signature is not in its canonical low-S form")]
+    SignatureNotCanonical,
+
+    #[error("the signature does not verify over the received octets")]
+    SignatureInvalid,
+
+    #[error("the trust chain is not issued by a root pinned in this build")]
+    EnrollmentRootUnknown,
+
+    #[error("a trust link is invalid, misordered, or outside its validity at the challenge issuedAt")]
+    TrustChainInvalid,
+
+    #[error("connectKeyId is not the subject of the last trust link")]
+    ConnectKeyUnchained,
+
+    #[error("no issued challenge matches this challengeId")]
+    ChallengeUnknown,
+
+    #[error("the challenge is not yet valid at the evaluation time")]
+    ChallengeNotYetValid,
+
+    #[error("the challenge has expired at the evaluation time")]
+    ChallengeExpired,
+
+    #[error("the response nonce or challengeProof is not the one issued for this challenge")]
+    ChallengeProofInvalid,
+
+    #[error("the response does not prove possession of the device key it presents")]
+    DeviceProofInvalid,
+
+    #[error("the challenge was already consumed")]
+    EnrollmentReplayed,
+
+    #[error("the response names a different organization than the challenge it answers")]
+    OrganizationMismatch,
+
+    #[error("the response names a different cluster than the challenge it answers")]
+    ClusterMismatch,
+
+    /// The artifact could not be read as a signed enrolment document at all: the
+    /// envelope, the base64 of the signed octets, or a field the frozen order
+    /// reads before the signature verifies did not parse. The frozen reason set
+    /// has no code for a structurally unreadable document, so this variant maps
+    /// to none of them.
+    #[error("the offline enrollment document is not well formed")]
+    MalformedDocument,
+
+    /// A fault on this side of the exchange rather than in the artifact: the
+    /// device key did not round-trip through its own PKCS#8 encoding, or the
+    /// caller named an instant outside the representable calendar. Fails closed
+    /// because a half-produced response must never reach removable media.
+    #[error("the enrollment response could not be produced on this device")]
+    ResponseNotProduced,
+}
+
+impl EnrollmentError {
+    /// The frozen `reason` an operator and Connect both branch on.
+    ///
+    /// The `Display` message is prose and may be reworded; this is the stable
+    /// identifier, so nothing should parse the message instead. The two
+    /// variants with no frozen counterpart deliberately return codes outside
+    /// the frozen set rather than borrowing the nearest one, so a document that
+    /// simply failed to parse can never be reported as a signature or freshness
+    /// failure.
+    pub fn reason(&self) -> &'static str {
+        match self {
+            Self::UnsupportedProtocol => "UNSUPPORTED_PROTOCOL",
+            Self::UnsupportedFormat => "UNSUPPORTED_FORMAT",
+            Self::SignatureMalformed => "SIGNATURE_MALFORMED",
+            Self::SignatureNotCanonical => "SIGNATURE_NOT_CANONICAL",
+            Self::SignatureInvalid => "SIGNATURE_INVALID",
+            Self::EnrollmentRootUnknown => "ENROLLMENT_ROOT_UNKNOWN",
+            Self::TrustChainInvalid => "TRUST_CHAIN_INVALID",
+            Self::ConnectKeyUnchained => "CONNECT_KEY_UNCHAINED",
+            Self::ChallengeUnknown => "CHALLENGE_UNKNOWN",
+            Self::ChallengeNotYetValid => "CHALLENGE_NOT_YET_VALID",
+            Self::ChallengeExpired => "CHALLENGE_EXPIRED",
+            Self::ChallengeProofInvalid => "CHALLENGE_PROOF_INVALID",
+            Self::DeviceProofInvalid => "DEVICE_PROOF_INVALID",
+            Self::EnrollmentReplayed => "ENROLLMENT_REPLAYED",
+            Self::OrganizationMismatch => "ORGANIZATION_MISMATCH",
+            Self::ClusterMismatch => "CLUSTER_MISMATCH",
+            Self::MalformedDocument => "MALFORMED_DOCUMENT",
+            Self::ResponseNotProduced => "RESPONSE_NOT_PRODUCED",
+        }
+    }
+}
+
+/// A signed document, in the shape both directions carry it. `bytes` is
+/// standard padded base64 of the exact octets that were signed; nothing else is
+/// ever used as the signature input.
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SignedDocument {
+    bytes: String,
+    signature: DocumentSignature,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct DocumentSignature {
+    algorithm: String,
+    key_id: String,
+    value: String,
+}
+
+/// The three fields the frozen order permits reading before anything verifies.
+/// They route the verification and are not facts until it has.
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ChallengeRouting {
+    connect_key_id: String,
+    issued_at: String,
+    trust_chain: Vec<SignedDocument>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ChallengeDocument {
+    format_version: String,
+    protocol_version: String,
+    challenge_id: String,
+    organization_name: String,
+    cluster_name: String,
+    nonce: String,
+    issued_at: String,
+    expires_at: String,
+    connect_key_id: String,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct TrustLink {
+    format_version: String,
+    protocol_version: String,
+    role: String,
+    issuer_key_id: String,
+    subject_key_id: String,
+    subject_public_key: String,
+    not_before: String,
+    not_after: String,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ResponseDocument<'a> {
+    format_version: &'a str,
+    protocol_version: &'a str,
+    challenge_id: &'a str,
+    organization_name: &'a str,
+    cluster_name: &'a str,
+    challenge_nonce: &'a str,
+    challenge_proof: &'a str,
+    device_key_id: String,
+    device_public_key: String,
+    device_nonce: String,
+    produced_at: String,
+}
+
+/// The device half of the offline enrolment exchange: bytes in, bytes out.
+pub struct OfflineEnrollment;
+
+impl OfflineEnrollment {
+    /// Verify an enrolment challenge and return what a response must echo.
+    ///
+    /// `now_unix` is the device's reading of the current time, which the clock
+    /// skew tolerance treats as advisory.
+    pub fn verify_challenge(document: &[u8], now_unix: i64) -> Result<VerifiedChallenge, EnrollmentError> {
+        let envelope: SignedDocument = serde_json::from_slice(document).map_err(|_| EnrollmentError::MalformedDocument)?;
+
+        // Step 1: the encoding is checked before anything is decoded from it, so
+        // a DER, padded, truncated, out-of-range, or high-S signature is refused
+        // on its spelling rather than handed to a library that would accept it.
+        let signature = decode_signature(&envelope.signature)?;
+
+        // The octets that were transmitted. They are never re-serialised: every
+        // later step signs and parses this same buffer.
+        let bytes = BASE64_STANDARD
+            .decode(envelope.bytes.as_bytes())
+            .map_err(|_| EnrollmentError::MalformedDocument)?;
+
+        // Step 2: routing only.
+        let routing: ChallengeRouting = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::MalformedDocument)?;
+        let issued_at = parse_timestamp(&routing.issued_at)?;
+
+        // Steps 3 to 5.
+        let connect_key = verify_trust_chain(&routing.trust_chain, &routing.connect_key_id, issued_at)?;
+
+        // Step 6. The verification key comes from the chain, so `signature.keyId`
+        // is a label rather than an input: a value naming some other key simply
+        // fails to verify here.
+        if !verifies(&connect_key, TAG_CHALLENGE, &bytes, &signature) {
+            return Err(EnrollmentError::SignatureInvalid);
+        }
+
+        // Step 7: only now is the document read as a fact.
+        let challenge: ChallengeDocument = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::MalformedDocument)?;
+        if challenge.protocol_version != PROTOCOL_VERSION {
+            return Err(EnrollmentError::UnsupportedProtocol);
+        }
+        if challenge.format_version != FORMAT_CHALLENGE {
+            return Err(EnrollmentError::UnsupportedFormat);
+        }
+
+        // Step 8.
+        let expires_at = parse_timestamp(&challenge.expires_at)?;
+        check_challenge_window(issued_at, expires_at, now_unix)?;
+
+        Ok(VerifiedChallenge {
+            challenge_id: challenge.challenge_id,
+            organization_name: challenge.organization_name,
+            cluster_name: challenge.cluster_name,
+            nonce: challenge.nonce,
+            issued_at: challenge.issued_at,
+            expires_at: challenge.expires_at,
+            connect_key_id: challenge.connect_key_id,
+            challenge_proof: envelope.signature.value,
+        })
+    }
+
+    /// Build the signed response an operator carries back to Connect.
+    ///
+    /// `device_nonce` is the response's own replay value and must come from a
+    /// cryptographic source. The private key never appears in the result: only
+    /// the public point, its fingerprint, and a signature over the document
+    /// that presents them, which is what makes presenting the key safe.
+    pub fn build_response(
+        challenge: &VerifiedChallenge,
+        key: &DeviceIdentity,
+        device_nonce: &[u8; 32],
+        produced_at_unix: i64,
+    ) -> Result<Vec<u8>, EnrollmentError> {
+        let issued_at = parse_timestamp(&challenge.issued_at)?;
+        let expires_at = parse_timestamp(&challenge.expires_at)?;
+        // Connect re-checks producedAt against the same window, so a response
+        // outside it is refused here rather than written to media and rejected
+        // after the operator has carried it out.
+        check_challenge_window(issued_at, expires_at, produced_at_unix)?;
+
+        let point = device_public_point(key)?;
+        let produced_at = format_timestamp(produced_at_unix)?;
+
+        let document = ResponseDocument {
+            format_version: FORMAT_RESPONSE,
+            protocol_version: PROTOCOL_VERSION,
+            challenge_id: &challenge.challenge_id,
+            organization_name: &challenge.organization_name,
+            cluster_name: &challenge.cluster_name,
+            challenge_nonce: &challenge.nonce,
+            challenge_proof: &challenge.challenge_proof,
+            device_key_id: key_id(&point),
+            device_public_key: BASE64_URL_NO_PAD.encode(point),
+            device_nonce: BASE64_URL_NO_PAD.encode(device_nonce),
+            produced_at,
+        };
+
+        // Serialised once. These octets are what is signed and what is carried,
+        // so no second serialisation can disagree with the signature.
+        let bytes = serde_json::to_vec(&document).map_err(|_| EnrollmentError::ResponseNotProduced)?;
+        let signature = sign(key, TAG_RESPONSE, &bytes)?;
+
+        let envelope = SignedDocument {
+            bytes: BASE64_STANDARD.encode(&bytes),
+            signature: DocumentSignature {
+                algorithm: SIGNATURE_ALGORITHM.to_owned(),
+                key_id: document.device_key_id,
+                value: signature,
+            },
+        };
+
+        serde_json::to_vec(&envelope).map_err(|_| EnrollmentError::ResponseNotProduced)
+    }
+}
+
+/// Walk the chain from the pinned root to the signing key, returning the key
+/// `connect_key_id` names once the chain vouches for it.
+fn verify_trust_chain(
+    chain: &[SignedDocument],
+    connect_key_id: &str,
+    challenge_issued_at: i64,
+) -> Result<VerifyingKey, EnrollmentError> {
+    // The pinned root gate runs before the chain's shape is examined, so a
+    // chain that is internally consistent under a foreign root — exactly what
+    // trust on first use would have accepted — is refused for its root rather
+    // than for its length.
+    let first = chain.first().ok_or(EnrollmentError::EnrollmentRootUnknown)?;
+    let root = decode_trust_link(first)?;
+    if root.0.issuer_key_id != PINNED_ROOT_KEY_ID {
+        return Err(EnrollmentError::EnrollmentRootUnknown);
+    }
+
+    let [_, second] = chain else {
+        return Err(EnrollmentError::TrustChainInvalid);
+    };
+    let links = [root, decode_trust_link(second)?];
+
+    let mut issuer_key_id = PINNED_ROOT_KEY_ID.to_owned();
+    let (mut issuer_key, _) = decode_public_key(PINNED_ROOT_PUBLIC_KEY).ok_or(EnrollmentError::EnrollmentRootUnknown)?;
+
+    for (index, ((link, link_bytes), entry)) in links.iter().zip(chain).enumerate() {
+        if link.format_version != FORMAT_TRUST_LINK
+            || link.protocol_version != PROTOCOL_VERSION
+            || link.role != CHAIN_ROLES[index]
+            || link.issuer_key_id != issuer_key_id
+            // A link that names itself as its own issuer would let a stolen
+            // intermediate mint its own root.
+            || link.subject_key_id == link.issuer_key_id
+        {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        let (subject_key, subject_point) =
+            decode_public_key(&link.subject_public_key).ok_or(EnrollmentError::TrustChainInvalid)?;
+        if key_id(&subject_point) != link.subject_key_id {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        let signature = decode_signature(&entry.signature)?;
+        if !verifies(&issuer_key, TAG_TRUST_LINK, link_bytes, &signature) {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        // The issuer controls both ends of a link's window, so it is evaluated
+        // with no skew tolerance, and against the challenge's issuedAt rather
+        // than against the device clock: a challenge carries the chain that was
+        // valid when it was issued.
+        let not_before = parse_timestamp(&link.not_before)?;
+        let not_after = parse_timestamp(&link.not_after)?;
+        if challenge_issued_at < not_before || challenge_issued_at > not_after {
+            return Err(EnrollmentError::TrustChainInvalid);
+        }
+
+        issuer_key_id = link.subject_key_id.clone();
+        issuer_key = subject_key;
+    }
+
+    if issuer_key_id != connect_key_id {
+        return Err(EnrollmentError::ConnectKeyUnchained);
+    }
+
+    Ok(issuer_key)
+}
+
+/// Decode a link and keep the octets it was signed over: the signature is
+/// checked against these, never against a re-encoding of the parsed link.
+fn decode_trust_link(entry: &SignedDocument) -> Result<(TrustLink, Vec<u8>), EnrollmentError> {
+    let bytes = BASE64_STANDARD
+        .decode(entry.bytes.as_bytes())
+        .map_err(|_| EnrollmentError::MalformedDocument)?;
+    let link = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::TrustChainInvalid)?;
+    Ok((link, bytes))
+}
+
+/// Check a signature's spelling and range, then admit it.
+///
+/// `r` and `s` are compared against the group order here rather than left to
+/// the ECDSA library, because a library that accepts high-S — every library
+/// does — would let a malleated copy of an artifact pass as a second artifact.
+fn decode_signature(signature: &DocumentSignature) -> Result<Signature, EnrollmentError> {
+    if signature.algorithm != SIGNATURE_ALGORITHM {
+        return Err(EnrollmentError::SignatureMalformed);
+    }
+
+    let value = signature.value.as_bytes();
+    if value.len() != SIGNATURE_VALUE_CHARS || !value.iter().all(|byte| is_base64url(*byte)) {
+        return Err(EnrollmentError::SignatureMalformed);
+    }
+
+    let decoded = BASE64_URL_NO_PAD
+        .decode(value)
+        .map_err(|_| EnrollmentError::SignatureMalformed)?;
+    let octets: [u8; SIGNATURE_OCTETS] = decoded
+        .as_slice()
+        .try_into()
+        .map_err(|_| EnrollmentError::SignatureMalformed)?;
+
+    // Big-endian octets of equal length order lexicographically exactly as the
+    // integers they spell, so a slice comparison is the range check.
+    let (r, s) = octets.split_at(SCALAR_OCTETS);
+    let out_of_range = |scalar: &[u8]| scalar.iter().all(|byte| *byte == 0) || scalar >= &GROUP_ORDER[..];
+    if out_of_range(r) || out_of_range(s) {
+        return Err(EnrollmentError::SignatureMalformed);
+    }
+    if s > &MAX_S[..] {
+        return Err(EnrollmentError::SignatureNotCanonical);
+    }
+
+    Signature::from_slice(&octets).map_err(|_| EnrollmentError::SignatureMalformed)
+}
+
+fn verifies(key: &VerifyingKey, tag: &[u8], bytes: &[u8], signature: &Signature) -> bool {
+    key.verify(&signature_input(tag, bytes), signature).is_ok()
+}
+
+fn signature_input(tag: &[u8], bytes: &[u8]) -> Vec<u8> {
+    let mut input = Vec::with_capacity(tag.len() + 1 + bytes.len());
+    input.extend_from_slice(tag);
+    input.push(DOMAIN_SEPARATOR);
+    input.extend_from_slice(bytes);
+    input
+}
+
+fn sign(key: &DeviceIdentity, tag: &[u8], bytes: &[u8]) -> Result<String, EnrollmentError> {
+    // `DeviceIdentity` publishes no general signing operation, so the key is
+    // rebuilt from its own PKCS#8 encoding; the encoding is wiped when the
+    // wrapper drops.
+    let pkcs8 = key.to_pkcs8_der().map_err(|_| EnrollmentError::ResponseNotProduced)?;
+    let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| EnrollmentError::ResponseNotProduced)?;
+
+    let signature: Signature = signing_key.sign(&signature_input(tag, bytes));
+    let canonical = signature.normalize_s().unwrap_or(signature);
+
+    Ok(BASE64_URL_NO_PAD.encode(canonical.to_bytes()))
+}
+
+/// The device's public point, recovered from the DER encoding the identity
+/// publishes so that one prefix constant governs both the fingerprint and the
+/// wire form.
+fn device_public_point(key: &DeviceIdentity) -> Result<[u8; PUBLIC_KEY_OCTETS], EnrollmentError> {
+    key.public_key_der()
+        .strip_prefix(&SPKI_PREFIX)
+        .and_then(|point| <[u8; PUBLIC_KEY_OCTETS]>::try_from(point).ok())
+        .ok_or(EnrollmentError::ResponseNotProduced)
+}
+
+/// Decode an uncompressed SEC1 point and check that it is on the curve.
+///
+/// The length and alphabet are checked before decoding so that a padded or
+/// standard-alphabet spelling is refused, and the point tag is checked so that
+/// the compressed and hybrid forms — which no keyId would match — cannot be
+/// spelled at all.
+fn decode_public_key(value: &str) -> Option<(VerifyingKey, [u8; PUBLIC_KEY_OCTETS])> {
+    let value = value.as_bytes();
+    if value.len() != PUBLIC_KEY_CHARS || !value.iter().all(|byte| is_base64url(*byte)) {
+        return None;
+    }
+
+    let point: [u8; PUBLIC_KEY_OCTETS] = BASE64_URL_NO_PAD.decode(value).ok()?.try_into().ok()?;
+    if point[0] != UNCOMPRESSED_POINT {
+        return None;
+    }
+
+    VerifyingKey::from_sec1_bytes(&point).ok().map(|key| (key, point))
+}
+
+/// Lowercase SHA-256 hex of the DER SubjectPublicKeyInfo built from a 65 octet
+/// uncompressed point.
+fn key_id(point: &[u8]) -> String {
+    let mut digest = Sha256::new();
+    digest.update(SPKI_PREFIX);
+    digest.update(point);
+    hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower)
+}
+
+fn is_base64url(byte: u8) -> bool {
+    byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_'
+}
+
+/// Parse `YYYY-MM-DDTHH:MM:SSZ` into a Unix instant.
+///
+/// The shape is checked before the fields are read: offsets other than `Z` and
+/// fractional seconds are refused rather than normalised, so two producers
+/// cannot spell the same instant two ways.
+fn parse_timestamp(value: &str) -> Result<i64, EnrollmentError> {
+    let octets = value.as_bytes();
+    if octets.len() != TIMESTAMP_CHARS
+        || octets[4] != b'-'
+        || octets[7] != b'-'
+        || octets[10] != b'T'
+        || octets[13] != b':'
+        || octets[16] != b':'
+        || octets[19] != b'Z'
+    {
+        return Err(EnrollmentError::MalformedDocument);
+    }
+
+    let field = |range: std::ops::Range<usize>| -> Result<u32, EnrollmentError> {
+        let text = &value[range];
+        if !text.bytes().all(|byte| byte.is_ascii_digit()) {
+            return Err(EnrollmentError::MalformedDocument);
+        }
+        text.parse().map_err(|_| EnrollmentError::MalformedDocument)
+    };
+
+    let month = Month::try_from(field(5..7)? as u8).map_err(|_| EnrollmentError::MalformedDocument)?;
+    let date = Date::from_calendar_date(field(0..4)? as i32, month, field(8..10)? as u8)
+        .map_err(|_| EnrollmentError::MalformedDocument)?;
+    let clock = Time::from_hms(field(11..13)? as u8, field(14..16)? as u8, field(17..19)? as u8)
+        .map_err(|_| EnrollmentError::MalformedDocument)?;
+
+    Ok(PrimitiveDateTime::new(date, clock).assume_utc().unix_timestamp())
+}
+
+fn format_timestamp(unix: i64) -> Result<String, EnrollmentError> {
+    let moment = OffsetDateTime::from_unix_timestamp(unix).map_err(|_| EnrollmentError::ResponseNotProduced)?;
+    Ok(format!(
+        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
+        moment.year(),
+        u8::from(moment.month()),
+        moment.day(),
+        moment.hour(),
+        moment.minute(),
+        moment.second()
+    ))
+}
+
+/// `at` must fall within `[issuedAt - 300, expiresAt + 300]`.
+///
+/// The declared expiry is capped at the protocol's maximum challenge lifetime
+/// because the issuer sets both ends of its own window; a challenge claiming a
+/// longer life expires at the bound.
+fn check_challenge_window(issued_at: i64, expires_at: i64, at: i64) -> Result<(), EnrollmentError> {
+    if at < issued_at.saturating_sub(CLOCK_SKEW_TOLERANCE) {
+        return Err(EnrollmentError::ChallengeNotYetValid);
+    }
+
+    let effective_expiry = expires_at.min(issued_at.saturating_add(MAX_CHALLENGE_LIFETIME));
+    if at > effective_expiry.saturating_add(CLOCK_SKEW_TOLERANCE) {
+        return Err(EnrollmentError::ChallengeExpired);
+    }
+
+    Ok(())
+}
@@ -0,0 +1,72 @@
+// 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.
+
+//! On-disk home of the offline enrollment key.
+//!
+//! An air-gapped device enrols with a key that is not its online device
+//! identity: the online key is minted during a registration exchange this
+//! device cannot perform, and an operator who carries an enrolment response out
+//! on removable media is enrolling exactly one key that Connect will pin. Losing
+//! it means asking for a fresh challenge, so it is written durably and published
+//! exactly once.
+//!
+//! The durability protocol is not reimplemented here. [`IdentityStore`] already
+//! seals a P-256 key at mode 0600, fsyncs it, and publishes it through a
+//! no-clobber link so a retry or a concurrent start converges on one key; it is
+//! pointed at a directory of this key's own rather than generalised into a
+//! key-store abstraction that would have to describe both lifecycles.
+
+use std::path::{Path, PathBuf};
+
+use super::super::identity::DeviceIdentity;
+use super::super::identity_store::{IdentityStore, StoreError};
+
+/// Subdirectory holding the offline enrolment key, kept apart from the online
+/// device identity so neither can be read in place of the other.
+const OFFLINE_DIRECTORY: &str = "offline";
+
+/// The offline enrolment key of one deployment.
+#[derive(Clone, Debug)]
+pub struct OfflineKeyStore {
+    inner: IdentityStore,
+}
+
+impl OfflineKeyStore {
+    pub fn new(directory: impl AsRef<Path>) -> Self {
+        Self {
+            inner: IdentityStore::new(directory.as_ref().join(OFFLINE_DIRECTORY)),
+        }
+    }
+
+    pub fn key_path(&self) -> PathBuf {
+        self.inner.key_path()
+    }
+
+    /// Return the stored key, or `None` when this deployment has never enrolled
+    /// offline. Reading never creates one, so a deployment that only ever
+    /// registers online holds no offline key.
+    pub fn load(&self) -> Result<Option<DeviceIdentity>, StoreError> {
+        self.inner.load()
+    }
+
+    /// Return the stored key, generating and publishing one the first time.
+    ///
+    /// A second enrolment attempt returns the original key rather than minting a
+    /// replacement: the operator may already be carrying a response for it, and
+    /// two keys would mean the response and the device disagree about which one
+    /// Connect pinned.
+    pub fn load_or_create(&self) -> Result<DeviceIdentity, StoreError> {
+        self.inner.load_or_create()
+    }
+}
@@ -0,0 +1,35 @@
+// 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.
+
+//! Offline enrolment: joining a Connect tenant without a network.
+//!
+//! An air-gapped cluster cannot perform the registration exchange, so an
+//! operator carries a signed challenge in and a signed response out. The device
+//! half of that exchange lives here: verifying the challenge against a root
+//! whose fingerprint is compiled into this binary, minting the key being
+//! enrolled, and signing the response.
+//!
+//! Nothing here opens a socket. That is the point of the surface, and it is
+//! asserted rather than assumed: the enrolment path takes bytes and returns
+//! bytes.
+//!
+//! The trust model, the signing convention, and every rejection reason are
+//! frozen by `protocol/agent/v1/fixtures/offline-enrollment/` and by
+//! `docs/adr/0009-offline-signing.md` on the Connect side.
+
+pub mod enrollment;
+pub mod key_store;
+
+pub use enrollment::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
+pub use key_store::OfflineKeyStore;
@@ -0,0 +1,951 @@
+// 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.
+
+//! Offline enrollment conformance against the frozen Connect fixtures.
+//!
+//! The device half of the air-gapped exchange verifies a challenge Connect
+//! signed and produces a response Connect will verify. Neither side can talk to
+//! the other while it does so, which means every disagreement about encoding,
+//! trust, or clock windows surfaces as a failed enrollment in the field rather
+//! than as an error at development time. The fixtures under
+//! `protocol/agent/v1/fixtures/offline-enrollment/` are the shared statement of
+//! what both sides must do, so this suite replays them rather than restating
+//! them: accept vectors must be accepted with the fields the document carries,
+//! reject vectors must fail with the single reason `error-codes.json` freezes,
+//! and the signature encoding rules in `trust-model.json` must hold even where
+//! the underlying ECDSA library is happy.
+
+use std::fs;
+use std::path::PathBuf;
+
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64_URL_NO_PAD;
+use rustfs::connect::identity::DeviceIdentity;
+use rustfs::connect::offline::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
+use serde_json::Value;
+use sha2::{Digest as _, Sha256};
+
+/// DER prefix of a P-256 `SubjectPublicKeyInfo`, frozen by
+/// `trust-model.json` as `signature.subjectPublicKeyInfoDerPrefix`. The 65
+/// octet uncompressed point follows it, so a SEC1 point published in a fixture
+/// becomes a decodable public key by concatenation.
+const SPKI_PREFIX_HEX: &str = "3059301306072a8648ce3d020106082a8648ce3d030107034200";
+
+/// `clockSkew.toleranceSeconds` in `trust-model.json`.
+const SKEW_TOLERANCE_SECONDS: i64 = 300;
+
+// ---------------------------------------------------------------------------
+// Fixture access
+// ---------------------------------------------------------------------------
+
+fn fixture_dir() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures/offline-enrollment")
+}
+
+fn sha256_hex(bytes: &[u8]) -> String {
+    Sha256::digest(bytes).iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+/// Read one fixture file and refuse it unless its bytes match the digest
+/// `MANIFEST.sha256` freezes.
+///
+/// Every vector in this suite arrives through here. A fixture edited on this
+/// side therefore fails the tests that depend on it instead of quietly
+/// redefining what conformance means, which is the failure mode a
+/// fixture-driven suite is otherwise blind to.
+fn read_fixture(name: &str) -> Vec<u8> {
+    let dir = fixture_dir();
+    let manifest = fs::read_to_string(dir.join("MANIFEST.sha256")).expect("read MANIFEST.sha256");
+
+    let expected = manifest
+        .lines()
+        .filter(|line| !line.trim().is_empty())
+        .find_map(|line| {
+            let (digest, file) = line
+                .split_once("  ")
+                .unwrap_or_else(|| panic!("malformed manifest line: {line}"));
+            (file == name).then(|| digest.to_string())
+        })
+        .unwrap_or_else(|| panic!("{name} is not listed in MANIFEST.sha256"));
+
+    let bytes = fs::read(dir.join(name)).unwrap_or_else(|error| panic!("read {name}: {error}"));
+    assert_eq!(sha256_hex(&bytes), expected, "{name} does not match the digest MANIFEST.sha256 freezes");
+    bytes
+}
+
+fn fixture_json(name: &str) -> Value {
+    serde_json::from_slice(&read_fixture(name)).unwrap_or_else(|error| panic!("{name} parses: {error}"))
+}
+
+fn accept_vectors() -> Value {
+    fixture_json("accept-vectors.json")
+}
+
+fn reject_vectors() -> Value {
+    fixture_json("reject-vectors.json")
+}
+
+fn trust_model() -> Value {
+    fixture_json("trust-model.json")
+}
+
+fn vector_list(fixture: &Value) -> Vec<Value> {
+    fixture["vectors"].as_array().expect("fixture carries a vector list").clone()
+}
+
+fn field<'a>(value: &'a Value, key: &str) -> &'a str {
+    value[key]
+        .as_str()
+        .unwrap_or_else(|| panic!("expected a string at '{key}' in {value}"))
+}
+
+/// The octets an operator carries in on removable media.
+///
+/// The fixture's `document` object *is* the transmitted artifact: a padded
+/// base64 `bytes` field holding the raw signed octets, plus the detached
+/// signature over them. Only `bytes` is covered by the signature, so
+/// re-serialising the surrounding envelope here cannot change what a verifier
+/// checks.
+fn envelope(document: &Value) -> Vec<u8> {
+    serde_json::to_vec(document).expect("envelope serialises")
+}
+
+/// The raw octets the signature covers, exactly as transmitted.
+fn signed_octets(document: &Value) -> Vec<u8> {
+    BASE64_STANDARD
+        .decode(field(document, "bytes"))
+        .expect("document bytes are padded base64")
+}
+
+/// The parsed signed document. Parsing is a convenience for the assertions
+/// below; the implementation under test is required to verify before it parses.
+fn signed_document(document: &Value) -> Value {
+    serde_json::from_slice(&signed_octets(document)).expect("signed document parses")
+}
+
+fn unix(rfc3339: &str) -> i64 {
+    chrono::DateTime::parse_from_rfc3339(rfc3339)
+        .unwrap_or_else(|error| panic!("'{rfc3339}' is not RFC 3339: {error}"))
+        .timestamp()
+}
+
+fn hex_to_bytes(hex: &str) -> Vec<u8> {
+    (0..hex.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("valid hex"))
+        .collect()
+}
+
+/// Turn a fixture's unpadded-base64url SEC1 point into a usable verifying key.
+fn verifying_key(sec1_base64url: &str) -> p256::ecdsa::VerifyingKey {
+    let point = BASE64_URL_NO_PAD.decode(sec1_base64url).expect("public key is base64url");
+    assert_eq!(point.len(), 65, "the protocol freezes a 65 octet uncompressed SEC1 point");
+
+    let mut der = hex_to_bytes(SPKI_PREFIX_HEX);
+    der.extend_from_slice(&point);
+    <p256::ecdsa::VerifyingKey as p256::pkcs8::DecodePublicKey>::from_public_key_der(&der).expect("public key decodes")
+}
+
+fn published_key(role_or_name: &str) -> Value {
+    fixture_json("trust-chain.json")["keys"]
+        .as_array()
+        .expect("trust chain publishes keys")
+        .iter()
+        .find(|key| field(key, "name") == role_or_name)
+        .unwrap_or_else(|| panic!("trust-chain.json publishes no key named '{role_or_name}'"))
+        .clone()
+}
+
+/// `signatureInput = domainSeparationTag || 0x00 || the received octets`, the
+/// rule `trust-model.json` freezes under `domainSeparation`.
+fn signing_input(artifact_tag: &str, received: &[u8]) -> Vec<u8> {
+    let mut input = artifact_tag.as_bytes().to_vec();
+    input.push(0x00);
+    input.extend_from_slice(received);
+    input
+}
+
+fn domain_tag(artifact: &str) -> String {
+    let model = trust_model();
+    assert_eq!(
+        field(&model["domainSeparation"], "separatorByte"),
+        "0x00",
+        "the separator byte this suite encodes is the one the trust model freezes"
+    );
+    field(&model["domainSeparation"]["tags"], artifact).to_string()
+}
+
+/// Locate an accept vector by the name other vectors reference it by.
+fn accept_vector_named(name: &str) -> Value {
+    vector_list(&accept_vectors())
+        .into_iter()
+        .find(|vector| field(vector, "name") == name)
+        .unwrap_or_else(|| panic!("accept-vectors.json carries no vector named '{name}'"))
+}
+
+/// Verify the challenge a response vector answers, at that challenge's own
+/// evaluation time.
+fn answered_challenge(response_vector: &Value) -> (Value, VerifiedChallenge) {
+    let challenge_vector = accept_vector_named(field(response_vector, "answersChallenge"));
+    let now = unix(field(&challenge_vector, "evaluationTime"));
+    let verified = OfflineEnrollment::verify_challenge(&envelope(&challenge_vector["document"]), now)
+        .expect("the answered challenge is an accept vector and must verify");
+    (challenge_vector, verified)
+}
+
+fn device_nonce_of(document: &Value) -> [u8; 32] {
+    let raw = BASE64_URL_NO_PAD
+        .decode(field(&signed_document(document), "deviceNonce"))
+        .expect("deviceNonce is base64url");
+    raw.try_into().expect("replay.nonceLengthBytes freezes a 32 octet nonce")
+}
+
+// ---------------------------------------------------------------------------
+// Accept vectors
+// ---------------------------------------------------------------------------
+
+/// Every challenge accept vector must verify at its own evaluation time and
+/// expose exactly what the signed document says.
+///
+/// Two of these vectors sit on the skew boundary — 120 seconds before
+/// `issuedAt` and 300 seconds after `expiresAt` — so a verifier that compares
+/// against the raw window instead of the tolerated one fails here rather than
+/// in an air-gapped data centre. `challenge_proof` is pinned to the challenge's
+/// own detached signature value because that is what the response has to echo;
+/// deriving it from anything else would silently break the binding.
+#[test]
+fn every_challenge_accept_vector_verifies_and_exposes_the_signed_fields() {
+    let mut verified_count = 0usize;
+
+    for vector in vector_list(&accept_vectors()) {
+        if field(&vector, "artifact") != "challenge" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let document = &vector["document"];
+        let now = unix(field(&vector, "evaluationTime"));
+
+        let verified = OfflineEnrollment::verify_challenge(&envelope(document), now)
+            .unwrap_or_else(|error| panic!("accept vector '{name}' must verify: {}", error.reason()));
+
+        let signed = signed_document(document);
+        assert_eq!(verified.challenge_id, field(&signed, "challengeId"), "vector '{name}' challengeId");
+        assert_eq!(
+            verified.organization_name,
+            field(&signed, "organizationName"),
+            "vector '{name}' organizationName"
+        );
+        assert_eq!(verified.cluster_name, field(&signed, "clusterName"), "vector '{name}' clusterName");
+        assert_eq!(verified.nonce, field(&signed, "nonce"), "vector '{name}' nonce");
+        assert_eq!(verified.issued_at, field(&signed, "issuedAt"), "vector '{name}' issuedAt");
+        assert_eq!(verified.expires_at, field(&signed, "expiresAt"), "vector '{name}' expiresAt");
+        assert_eq!(verified.connect_key_id, field(&signed, "connectKeyId"), "vector '{name}' connectKeyId");
+        assert_eq!(
+            verified.challenge_proof,
+            field(&document["signature"], "value"),
+            "vector '{name}' must carry the challenge's own signature as the proof a response echoes"
+        );
+
+        verified_count += 1;
+    }
+
+    assert_eq!(
+        verified_count, 3,
+        "accept-vectors.json publishes three challenge vectors; a fourth is a protocol change"
+    );
+}
+
+/// Connect's own producer wrote the response accept vectors. Rebuilding them
+/// from the challenge they answer, with the device nonce and production time
+/// they used, must reproduce every field that does not depend on which device
+/// key signed — including the discarded-unknown-field vector, whose extra
+/// `telemetryHint` must not survive into anything this side produces.
+#[test]
+fn response_accept_vectors_are_reproduced_field_for_field_by_build_response() {
+    let key = DeviceIdentity::generate();
+    let mut reproduced = 0usize;
+
+    for vector in vector_list(&accept_vectors()) {
+        if field(&vector, "artifact") != "response" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let published = signed_document(&vector["document"]);
+        let (_, challenge) = answered_challenge(&vector);
+
+        let produced_at = unix(field(&published, "producedAt"));
+        let nonce = device_nonce_of(&vector["document"]);
+
+        let built_envelope: Value = serde_json::from_slice(
+            &OfflineEnrollment::build_response(&challenge, &key, &nonce, produced_at)
+                .unwrap_or_else(|error| panic!("vector '{name}' must be reproducible: {}", error.reason())),
+        )
+        .expect("the built response is JSON");
+        let built = signed_document(&built_envelope);
+
+        for shared in [
+            "formatVersion",
+            "protocolVersion",
+            "challengeId",
+            "organizationName",
+            "clusterName",
+            "challengeNonce",
+            "challengeProof",
+            "deviceNonce",
+        ] {
+            assert_eq!(
+                field(&built, shared),
+                field(&published, shared),
+                "vector '{name}' field {shared} must match the response Connect published"
+            );
+        }
+        assert_eq!(
+            unix(field(&built, "producedAt")),
+            produced_at,
+            "vector '{name}' producedAt must be the instant it was given"
+        );
+
+        // `versioning.additive` says an unknown optional field is discarded and
+        // never echoed back; a producer that copied the challenge or a previous
+        // response wholesale would carry it forward.
+        assert!(
+            built.get("telemetryHint").is_none(),
+            "vector '{name}' must not echo an unknown optional field"
+        );
+
+        reproduced += 1;
+    }
+
+    assert_eq!(
+        reproduced, 2,
+        "accept-vectors.json publishes two response vectors; a third is a protocol change"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// Reject vectors
+// ---------------------------------------------------------------------------
+
+/// Every challenge reject vector must fail, and fail for the one reason
+/// `error-codes.json` freezes.
+///
+/// Asserting only that verification failed would pass for an implementation
+/// that rejects everything, and would let a tampered document be reported as an
+/// expiry — a rejection reason is what an operator acts on, so it is part of the
+/// contract rather than a diagnostic detail.
+#[test]
+fn every_challenge_reject_vector_fails_with_its_frozen_reason() {
+    let known_reasons: Vec<String> = fixture_json("error-codes.json")["reasons"]
+        .as_array()
+        .expect("error-codes.json carries reasons")
+        .iter()
+        .map(|entry| field(entry, "reason").to_string())
+        .collect();
+
+    let mut rejected = 0usize;
+
+    for vector in vector_list(&reject_vectors()) {
+        if field(&vector, "artifact") != "challenge" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let expected = field(&vector["expected"], "reason");
+        assert!(
+            known_reasons.iter().any(|reason| reason == expected),
+            "vector '{name}' names reason {expected}, which error-codes.json does not freeze"
+        );
+
+        let now = unix(field(&vector, "evaluationTime"));
+        let error = OfflineEnrollment::verify_challenge(&envelope(&vector["document"]), now)
+            .expect_err(&format!("reject vector '{name}' must not verify"));
+
+        assert_eq!(error.reason(), expected, "vector '{name}' must fail as {expected}");
+        rejected += 1;
+    }
+
+    assert_eq!(
+        rejected, 8,
+        "reject-vectors.json publishes eight challenge vectors; losing one silently narrows the suite"
+    );
+}
+
+/// The response reject vectors are artifacts Connect refuses. This side never
+/// verifies a response, so the device-side statement is the stronger one: given
+/// the challenge each vector answers, `build_response` must not be capable of
+/// emitting that artifact in the first place.
+///
+/// Each arm pins the specific field a compromised or careless producer would
+/// have to get wrong, so an implementation that copied values out of the wrong
+/// place — the response's own document, an operator-supplied argument, a
+/// previous exchange — fails here.
+#[test]
+fn response_reject_vectors_are_artifacts_build_response_cannot_emit() {
+    let key = DeviceIdentity::generate();
+    let mut covered = 0usize;
+
+    for vector in vector_list(&reject_vectors()) {
+        if field(&vector, "artifact") != "response" {
+            continue;
+        }
+
+        let name = field(&vector, "name");
+        let refused = signed_document(&vector["document"]);
+        let (_, challenge) = answered_challenge(&vector);
+        let produced_at = unix(field(&refused, "producedAt"));
+        let nonce = device_nonce_of(&vector["document"]);
+
+        let outcome = OfflineEnrollment::build_response(&challenge, &key, &nonce, produced_at);
+
+        match field(&vector["expected"], "reason") {
+            // `responseWindow` in trust-model.json: a device that emits a
+            // response outside the tolerated challenge window has produced an
+            // artifact Connect will refuse, so the refusal belongs here rather
+            // than at the far end of a courier run.
+            "CHALLENGE_EXPIRED" => {
+                let error = outcome.expect_err(&format!("vector '{name}': producing this response must be refused"));
+                assert_eq!(error.reason(), "CHALLENGE_EXPIRED", "vector '{name}' must refuse as CHALLENGE_EXPIRED");
+                covered += 1;
+                continue;
+            }
+            reason => {
+                let built_envelope: Value = serde_json::from_slice(
+                    &outcome.unwrap_or_else(|error| panic!("vector '{name}' baseline must build: {}", error.reason())),
+                )
+                .expect("the built response is JSON");
+                let built = signed_document(&built_envelope);
+
+                match reason {
+                    "ORGANIZATION_MISMATCH" => {
+                        assert_ne!(
+                            field(&refused, "organizationName"),
+                            challenge.organization_name,
+                            "vector '{name}' is only a mismatch if it names another organization"
+                        );
+                        assert_eq!(
+                            field(&built, "organizationName"),
+                            challenge.organization_name,
+                            "vector '{name}': the organization must come from the challenge, never from elsewhere"
+                        );
+                    }
+                    "CLUSTER_MISMATCH" => {
+                        assert_ne!(
+                            field(&refused, "clusterName"),
+                            challenge.cluster_name,
+                            "vector '{name}' is only a mismatch if it names another cluster"
+                        );
+                        assert_eq!(
+                            field(&built, "clusterName"),
+                            challenge.cluster_name,
+                            "vector '{name}': the cluster must come from the challenge, never from elsewhere"
+                        );
+                    }
+                    "CHALLENGE_PROOF_INVALID" => {
+                        // Two distinct vectors land here: a nonce the challenge
+                        // never carried, and a proof lifted from another
+                        // challenge. Both must be impossible to produce.
+                        assert_eq!(
+                            field(&built, "challengeNonce"),
+                            challenge.nonce,
+                            "vector '{name}': the echoed nonce must be the challenge's own"
+                        );
+                        assert_eq!(
+                            field(&built, "challengeProof"),
+                            challenge.challenge_proof,
+                            "vector '{name}': the proof must be the answered challenge's signature"
+                        );
+                        assert!(
+                            field(&refused, "challengeNonce") != challenge.nonce
+                                || field(&refused, "challengeProof") != challenge.challenge_proof,
+                            "vector '{name}' must differ from the challenge in nonce or proof to be rejectable"
+                        );
+                    }
+                    "DEVICE_PROOF_INVALID" => {
+                        // The refused vector presents one key and is signed by
+                        // another; hold the fixture to that claim, then require
+                        // the built response to be the opposite. Proof of
+                        // possession is the only thing that makes presenting a
+                        // key in an unauthenticated document safe.
+                        use p256::ecdsa::signature::Verifier as _;
+
+                        let presented = verifying_key(field(&refused, "devicePublicKey"));
+                        let raw = BASE64_URL_NO_PAD
+                            .decode(field(&vector["document"]["signature"], "value"))
+                            .expect("signature is base64url");
+                        let signature = p256::ecdsa::Signature::from_slice(&raw).expect("signature parses");
+                        assert!(
+                            presented
+                                .verify(
+                                    &signing_input(&domain_tag("enrollmentResponse"), &signed_octets(&vector["document"])),
+                                    &signature
+                                )
+                                .is_err(),
+                            "vector '{name}' is only a possession failure if it does not verify under the key it presents"
+                        );
+
+                        assert_response_proves_possession(&built_envelope, name);
+                    }
+                    "UNSUPPORTED_FORMAT" => {
+                        assert_ne!(
+                            field(&refused, "formatVersion"),
+                            field(&built, "formatVersion"),
+                            "vector '{name}' is only unsupported if it names another format version"
+                        );
+                        assert_eq!(
+                            field(&built, "formatVersion"),
+                            "rustfs.connect.offline.enrollmentResponse/1",
+                            "vector '{name}': the format version is frozen"
+                        );
+                    }
+                    "UNSUPPORTED_PROTOCOL" => {
+                        assert_ne!(
+                            field(&refused, "protocolVersion"),
+                            field(&built, "protocolVersion"),
+                            "vector '{name}' is only unsupported if it names another protocol major"
+                        );
+                        assert_eq!(field(&built, "protocolVersion"), "v1", "vector '{name}': the protocol major is frozen");
+                    }
+                    "ENROLLMENT_REPLAYED" => {
+                        // The vector claims to be a byte-identical replay of an
+                        // accepted response; hold it to that, because a replay
+                        // vector that is not byte identical proves nothing about
+                        // single use.
+                        let accepted = accept_vector_named("response binding the device public key and the challenge proof");
+                        assert_eq!(
+                            signed_octets(&vector["document"]),
+                            signed_octets(&accepted["document"]),
+                            "vector '{name}' must be the accepted response octet for octet"
+                        );
+                        assert_eq!(
+                            field(&vector["document"]["signature"], "value"),
+                            field(&accepted["document"]["signature"], "value"),
+                            "vector '{name}' must carry the accepted response's signature"
+                        );
+
+                        // A fresh device nonce is a different artifact, so a
+                        // second enrollment is never mistaken for a replay of
+                        // the first.
+                        let other = OfflineEnrollment::build_response(&challenge, &key, &[0x5a; 32], produced_at)
+                            .expect("a second response builds");
+                        assert_ne!(
+                            signed_octets(&built_envelope),
+                            signed_octets(&serde_json::from_slice::<Value>(&other).expect("JSON")),
+                            "vector '{name}': a different device nonce must yield a different artifact"
+                        );
+                    }
+                    other => panic!("vector '{name}' names an unhandled reason {other}; extend this test"),
+                }
+            }
+        }
+
+        covered += 1;
+    }
+
+    assert_eq!(
+        covered, 9,
+        "reject-vectors.json publishes nine response vectors; losing one silently narrows the suite"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// Signature encoding
+// ---------------------------------------------------------------------------
+
+/// The high-S malleation is the rejection the whole encoding rule exists for.
+///
+/// `(r, n - s)` is a second valid signature over the same document under the
+/// same key. Every mainstream ECDSA library verifies it, so an implementation
+/// that hands the decoded octets straight to `p256` accepts a forged-looking
+/// duplicate of a genuine challenge — and because the 64 octets differ, that
+/// duplicate is a distinct artifact identity that slips past any deduplication
+/// keyed on the signature. This test proves the rejection came from the
+/// encoding rule and not from a failed verification: it first shows the
+/// malleated signature verifying mathematically, then requires
+/// `verify_challenge` to refuse it as SIGNATURE_NOT_CANONICAL.
+#[test]
+fn malleated_high_s_signature_is_refused_although_it_verifies_mathematically() {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let model = trust_model();
+    let malleated = model["rejectedSignatureEncodings"]
+        .as_array()
+        .expect("trust-model.json publishes rejected encodings")
+        .iter()
+        .find(|entry| field(entry, "reason") == "SIGNATURE_NOT_CANONICAL")
+        .expect("trust-model.json publishes the high-S malleation")
+        .clone();
+    assert!(
+        malleated["acceptedByALenientVerifier"].as_bool() == Some(true),
+        "this vector is only interesting because a lenient verifier accepts it"
+    );
+
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let genuine_value = field(&vector["document"]["signature"], "value").to_string();
+    let malleated_value = field(&malleated, "value").to_string();
+    assert_ne!(genuine_value, malleated_value, "the malleation must be a different encoding");
+
+    let genuine = BASE64_URL_NO_PAD.decode(&genuine_value).expect("signature is base64url");
+    let raw = BASE64_URL_NO_PAD.decode(&malleated_value).expect("signature is base64url");
+    assert_eq!(raw.len(), 64, "the malleation is well formed at 64 octets");
+    assert_eq!(raw[..32], genuine[..32], "the malleation shares r with the genuine signature");
+    assert_ne!(raw[32..], genuine[32..], "the malleation replaces s with n - s");
+
+    // Step one: the malleated pair really does verify under the signing key, so
+    // a verifier cannot be excused for accepting it on mathematical grounds.
+    let signature = p256::ecdsa::Signature::from_slice(&raw).expect("the malleated signature parses");
+    assert!(signature.normalize_s().is_some(), "the malleated signature must be the high-S form");
+    let key = verifying_key(field(&published_key("signing"), "publicKey"));
+    let input = signing_input(&domain_tag("enrollmentChallenge"), &signed_octets(&vector["document"]));
+    key.verify(&input, &signature)
+        .expect("the malleated signature must verify mathematically, or this test proves nothing");
+
+    // Step two: the implementation must refuse it anyway, and say why.
+    let mut tampered = vector["document"].clone();
+    tampered["signature"]["value"] = Value::String(malleated_value);
+
+    let now = unix(field(&vector, "evaluationTime"));
+    let error = OfflineEnrollment::verify_challenge(&envelope(&tampered), now)
+        .expect_err("a high-S signature must be refused even though it verifies");
+    assert_eq!(
+        error.reason(),
+        "SIGNATURE_NOT_CANONICAL",
+        "a malleated signature is a canonicality failure, not a verification failure"
+    );
+}
+
+/// Every encoding `trust-model.json` names as rejected must fail with the
+/// reason it names — DER, padded base64url, truncation, and out-of-range
+/// scalars alongside the malleation. Three of the five are accepted by a
+/// lenient verifier, so a single blanket "signature did not verify" answer would
+/// be both wrong and undiagnosable.
+#[test]
+fn every_rejected_signature_encoding_fails_with_its_frozen_reason() {
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let now = unix(field(&vector, "evaluationTime"));
+    let model = trust_model();
+    let encodings = model["rejectedSignatureEncodings"]
+        .as_array()
+        .expect("trust-model.json publishes rejected encodings");
+
+    for entry in encodings {
+        let name = field(entry, "name");
+        let mut tampered = vector["document"].clone();
+        tampered["signature"]["value"] = Value::String(field(entry, "value").to_string());
+
+        let error: EnrollmentError = OfflineEnrollment::verify_challenge(&envelope(&tampered), now)
+            .err()
+            .unwrap_or_else(|| panic!("rejected encoding '{name}' must not verify"));
+
+        assert_eq!(error.reason(), field(entry, "reason"), "rejected encoding '{name}'");
+    }
+
+    assert_eq!(encodings.len(), 5, "trust-model.json freezes five rejected encodings");
+}
+
+// ---------------------------------------------------------------------------
+// Clock window
+// ---------------------------------------------------------------------------
+
+/// The tolerated window is `[issuedAt - 300, expiresAt + 300]`, inclusive at
+/// both ends. An air-gapped device has no synchronised clock, so an
+/// off-by-one here either strands a legitimate enrollment or widens the window
+/// a stolen challenge stays usable in. Both ends are checked at the exact bound
+/// and one second past it, and the reason distinguishes the two directions.
+#[test]
+fn challenge_is_accepted_at_the_exact_skew_bound_and_refused_one_second_past_it() {
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let document = envelope(&vector["document"]);
+    let signed = signed_document(&vector["document"]);
+
+    let issued_at = unix(field(&signed, "issuedAt"));
+    let expires_at = unix(field(&signed, "expiresAt"));
+
+    let earliest = issued_at - SKEW_TOLERANCE_SECONDS;
+    OfflineEnrollment::verify_challenge(&document, earliest).expect("the earliest tolerated instant is inside the window");
+    let error =
+        OfflineEnrollment::verify_challenge(&document, earliest - 1).expect_err("one second earlier is outside the window");
+    assert_eq!(error.reason(), "CHALLENGE_NOT_YET_VALID");
+
+    let latest = expires_at + SKEW_TOLERANCE_SECONDS;
+    OfflineEnrollment::verify_challenge(&document, latest).expect("the latest tolerated instant is inside the window");
+    let error = OfflineEnrollment::verify_challenge(&document, latest + 1).expect_err("one second later is outside the window");
+    assert_eq!(error.reason(), "CHALLENGE_EXPIRED");
+}
+
+// ---------------------------------------------------------------------------
+// Response production
+// ---------------------------------------------------------------------------
+
+/// Assert a built response proves possession of the key it presents: the
+/// fingerprint matches the presented key, and the detached signature is a
+/// canonical low-S ES256 signature that verifies under that key over the exact
+/// octets transmitted.
+fn assert_response_proves_possession(built_envelope: &Value, label: &str) {
+    use p256::ecdsa::signature::Verifier as _;
+
+    let raw = signed_octets(built_envelope);
+    let built = signed_document(built_envelope);
+    let signature_block = &built_envelope["signature"];
+
+    assert_eq!(field(signature_block, "algorithm"), "ES256", "{label}: the algorithm is frozen");
+
+    let value = field(signature_block, "value");
+    assert_eq!(value.len(), 86, "{label}: the transfer encoding is 86 unpadded base64url characters");
+    assert!(
+        value.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'),
+        "{label}: the signature must use the base64url alphabet with no padding"
+    );
+
+    let bytes = BASE64_URL_NO_PAD.decode(value).expect("signature is base64url");
+    assert_eq!(bytes.len(), 64, "{label}: the signature is a fixed-width r || s");
+    let signature = p256::ecdsa::Signature::from_slice(&bytes).expect("signature parses");
+    assert!(
+        signature.normalize_s().is_none(),
+        "{label}: this side must never emit the malleated high-S form it refuses to accept"
+    );
+
+    let presented = field(&built, "devicePublicKey");
+    let key = verifying_key(presented);
+    key.verify(&signing_input(&domain_tag("enrollmentResponse"), &raw), &signature)
+        .unwrap_or_else(|error| panic!("{label}: the response must verify under the key it presents: {error}"));
+
+    // `signature.keyIdAlgorithm`: the lowercase SHA-256 of the DER
+    // SubjectPublicKeyInfo, not of the bare point and not of the transfer
+    // encoding.
+    let mut spki = hex_to_bytes(SPKI_PREFIX_HEX);
+    spki.extend_from_slice(&BASE64_URL_NO_PAD.decode(presented).expect("public key is base64url"));
+    let fingerprint = sha256_hex(&spki);
+    assert_eq!(
+        field(&built, "deviceKeyId"),
+        fingerprint,
+        "{label}: deviceKeyId must be the fingerprint of the key the document presents"
+    );
+    assert_eq!(
+        field(signature_block, "keyId"),
+        fingerprint,
+        "{label}: the detached signature must name the same key"
+    );
+}
+
+/// A response is the only thing Connect will ever see from this device, so it
+/// has to carry the whole binding on its own: the challenge it answers, the
+/// proof that challenge was genuine, the key being enrolled, and possession of
+/// that key.
+#[test]
+fn built_response_binds_the_challenge_proof_and_proves_possession_of_the_device_key() {
+    let vector = accept_vector_named("response binding the device public key and the challenge proof");
+    let (challenge_vector, challenge) = answered_challenge(&vector);
+    let key = DeviceIdentity::generate();
+    let produced_at = unix(field(&signed_document(&vector["document"]), "producedAt"));
+
+    let bytes = OfflineEnrollment::build_response(&challenge, &key, &[0x11; 32], produced_at).expect("the response builds");
+    let built_envelope: Value = serde_json::from_slice(&bytes).expect("the response is JSON");
+    let built = signed_document(&built_envelope);
+
+    assert_response_proves_possession(&built_envelope, "built response");
+
+    // The proof is the challenge's own detached signature. A producer that
+    // echoed the nonce alone, or hashed something, would let a response be
+    // built from an unverified challenge.
+    assert_eq!(
+        field(&built, "challengeProof"),
+        field(&challenge_vector["document"]["signature"], "value"),
+        "the proof must be the signature of the challenge being answered"
+    );
+    assert_eq!(field(&built, "challengeNonce"), challenge.nonce);
+    assert_eq!(field(&built, "challengeId"), challenge.challenge_id);
+
+    assert_eq!(
+        field(&built, "devicePublicKey"),
+        BASE64_URL_NO_PAD.encode(&key.public_key_der()[hex_to_bytes(SPKI_PREFIX_HEX).len()..]),
+        "the presented key must be the key that was passed in"
+    );
+    assert_eq!(
+        field(&built, "deviceNonce"),
+        BASE64_URL_NO_PAD.encode([0x11; 32]),
+        "the device nonce must be the one that was passed in"
+    );
+    assert!(field(&built, "producedAt").ends_with('Z'), "producedAt is a UTC RFC 3339 instant");
+}
+
+/// The response leaves the air gap on removable media and is read by anyone who
+/// handles it. A producer that serialised the key pair instead of the public
+/// key, or logged a debug rendering into the document, would put the enrolled
+/// private key on that medium — and the enrollment would still succeed, so
+/// nothing else in this suite would notice.
+#[test]
+fn built_response_carries_no_private_key_material() {
+    let vector = accept_vector_named("response binding the device public key and the challenge proof");
+    let (_, challenge) = answered_challenge(&vector);
+    let key = DeviceIdentity::generate();
+    let produced_at = unix(field(&signed_document(&vector["document"]), "producedAt"));
+
+    let response = OfflineEnrollment::build_response(&challenge, &key, &[0x22; 32], produced_at).expect("the response builds");
+
+    // The envelope carries the signed document base64-encoded, so a needle
+    // present in the document is not present in the envelope octets. Both
+    // layers are searched: an operator handling the medium can read either.
+    let envelope_value: Value = serde_json::from_slice(&response).expect("the response is JSON");
+    let mut haystack = response;
+    haystack.extend_from_slice(&signed_octets(&envelope_value));
+
+    let pkcs8 = key.to_pkcs8_der().expect("serialise the key");
+    let secret = <p256::SecretKey as p256::pkcs8::DecodePrivateKey>::from_pkcs8_der(&pkcs8).expect("the key parses");
+    let scalar = secret.to_bytes();
+
+    // Every spelling the scalar could plausibly reach a document in: raw, and
+    // the three encodings this protocol already uses elsewhere.
+    let scalar_hex: String = scalar.iter().map(|byte| format!("{byte:02x}")).collect();
+    for (description, needle) in [
+        ("the PKCS#8 encoding", pkcs8.to_vec()),
+        ("the raw private scalar", scalar.to_vec()),
+        ("the scalar in base64url", BASE64_URL_NO_PAD.encode(scalar).into_bytes()),
+        ("the scalar in standard base64", BASE64_STANDARD.encode(scalar).into_bytes()),
+        ("the scalar in hex", scalar_hex.into_bytes()),
+    ] {
+        assert!(
+            !haystack.windows(needle.len()).any(|window| window == needle.as_slice()),
+            "the response must not contain {description}"
+        );
+    }
+
+    // The public half must be there, so the absence above is a statement about
+    // what was excluded rather than about a haystack that would not have found
+    // the private half either.
+    let point = BASE64_URL_NO_PAD.encode(&key.public_key_der()[hex_to_bytes(SPKI_PREFIX_HEX).len()..]);
+    assert!(
+        haystack.windows(point.len()).any(|window| window == point.as_bytes()),
+        "the response must still present the public key"
+    );
+}
+
+// ---------------------------------------------------------------------------
+// The offline invariant
+// ---------------------------------------------------------------------------
+
+/// The whole surface exists because there is no network. This asserts that
+/// three different ways, because no single one of them is conclusive on its own.
+///
+/// 1. The process opens no descriptor across a full verify-and-respond cycle. A
+///    socket, a DNS resolver, a pooled HTTP client, or a revocation-list fetch
+///    all show up here — including one that is opened and cached rather than
+///    opened and closed, which is what a lazily built client does.
+/// 2. The cycle is a pure byte transform: the same inputs produce the same
+///    verified fields, and the evaluation instant is an argument rather than an
+///    ambient read, so nothing about the outcome can depend on reachability.
+/// 3. Repeating the cycle changes nothing observable, so a first call cannot be
+///    quietly initialising shared state that a later one reuses.
+#[cfg(unix)]
+#[test]
+fn enrollment_opens_no_descriptor_and_is_a_pure_byte_transform() {
+    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
+    let document = envelope(&vector["document"]);
+    let now = unix(field(&vector, "evaluationTime"));
+    let key = DeviceIdentity::generate();
+
+    // Warm anything the test harness itself lazily opens before the baseline.
+    let _ = open_descriptors();
+    let baseline = open_descriptors();
+    assert!(
+        !baseline.is_empty(),
+        "the descriptor table must be readable for this test to mean anything"
+    );
+
+    let mut fields = Vec::new();
+    for _ in 0..2 {
+        let challenge = OfflineEnrollment::verify_challenge(&document, now).expect("the challenge verifies");
+        let response = OfflineEnrollment::build_response(&challenge, &key, &[0x33; 32], now).expect("the response builds");
+        fields.push((
+            challenge.challenge_id.clone(),
+            challenge.nonce.clone(),
+            challenge.challenge_proof.clone(),
+            signed_octets(&serde_json::from_slice::<Value>(&response).expect("JSON")),
+        ));
+    }
+
+    assert_eq!(
+        open_descriptors(),
+        baseline,
+        "the enrollment path must not open a descriptor: no socket, no resolver, no cached client"
+    );
+
+    let (first, second) = (&fields[0], &fields[1]);
+    assert_eq!(first.0, second.0, "verification must be deterministic");
+    assert_eq!(first.1, second.1, "verification must be deterministic");
+    assert_eq!(first.2, second.2, "verification must be deterministic");
+    assert_eq!(
+        first.3, second.3,
+        "the signed response octets are a function of the challenge, the key, the nonce, and the instant"
+    );
+}
+
+#[cfg(unix)]
+fn open_descriptors() -> Vec<String> {
+    // Linux publishes the table at /proc/self/fd; the BSDs and macOS at /dev/fd.
+    let path = if PathBuf::from("/proc/self/fd").is_dir() {
+        "/proc/self/fd"
+    } else {
+        "/dev/fd"
+    };
+
+    let mut entries: Vec<String> = fs::read_dir(path)
+        .unwrap_or_else(|error| panic!("read {path}: {error}"))
+        .map(|entry| entry.expect("read dir entry").file_name().to_string_lossy().into_owned())
+        .collect();
+    entries.sort();
+    entries
+}
+
+/// A descriptor count taken around a call cannot see a socket that was opened
+/// and closed inside it, so the invariant is also asserted where it can be
+/// stated absolutely: the implementation names no network API at all.
+///
+/// This is the shape the regression actually takes — someone adds a
+/// revocation-list fetch, a time-server check, or a "just confirm the challenge
+/// with Connect" call — and it is caught at the source rather than by observing
+/// its effects.
+#[test]
+fn enrollment_implementation_names_no_network_api() {
+    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src/connect/offline/enrollment.rs");
+    let source = fs::read_to_string(&path).unwrap_or_else(|error| panic!("read {}: {error}", path.display()));
+
+    // Prose is allowed to discuss the invariant it is documenting, so only code
+    // is scanned.
+    let code: String = source
+        .lines()
+        .filter(|line| !line.trim_start().starts_with("//"))
+        .collect::<Vec<_>>()
+        .join("\n");
+
+    for forbidden in [
+        "std::net",
+        "tokio::net",
+        "TcpStream",
+        "TcpListener",
+        "UdpSocket",
+        "UnixStream",
+        "ToSocketAddrs",
+        "reqwest",
+        "hyper",
+        "tonic",
+    ] {
+        assert!(
+            !code.contains(forbidden),
+            "offline enrollment must not reach the network, but the implementation names {forbidden}"
+        );
+    }
+}
Author	SHA1	Message	Date
马登山	eeab9d201b	fix(heal): fence format repair during pool transitions	2026-08-22 00:25:32 +08:00
Zhengchao An	bce5922aef	feat(connect): answer offline enrolment challenges without a network (#6335 )	2026-08-21 13:16:18 +00:00