TQ: Support for ZFS Key Rotation (#9737)

When Trust Quorum commits a new epoch, all U.2 crypt datasets must have
their encryption keys rotated to use the new epoch's derived key. This
change implements the key rotation flow triggered by epoch commits.

## Trust Quorum Integration

- Add watch channel to `NodeTaskHandle` for epoch change notifications
- Initialize channel with current committed epoch on startup
- Notify subscribers via `send_if_modified()` when epoch changes

## Config Reconciler Integration

- Accept `committed_epoch_rx` watch channel from trust quorum
- Trigger reconciliation when epoch changes
- Track per-disk encryption epoch in `ExternalDisks`
- Add `rekey_for_epoch()` to coordinate key rotation:
  - Filter disks needing rekey (cached epoch < target OR unknown)
  - Derive keys for each disk via `StorageKeyRequester`
  - Send batch request to dataset task
  - Update cached epochs on success
  - Retry on failure via normal reconciliation retry logic

## Dataset Task Changes

- Add `RekeyRequest`/`RekeyResult` types for batch rekey operations
- Add `datasets_rekey()` with idempotency check (skip if already at
target)
- Use `Zfs::change_key()` for atomic key + epoch property update

## ZFS Utilities

- Add `Zfs::change_key()` using native `-o user:property=value` support
recently added to Illumos ZFS
- Add `Zfs::load_key()`, `unload_key()`, `dataset_exists()`
- Add `epoch` field to `DatasetProperties`
- Add structured error types for key operations

## Crash Recovery

- Add trial decryption recovery in `sled-storage` for datasets with
missing epoch property (e.g., crash during initial creation)
- Unload key before each trial attempt to handle crash-after-load-key
- Set epoch property after successful recovery

---------

Co-authored-by: Andrew J. Stone <andrew@oxidecomputer.com>

Author: plaidfinch

Cargo.lock

@@ -75,6 +75,7 @@ members = [
     "internal-dns/types/versions",
     "ipcc",
     "key-manager",
+    "key-manager/types",
     "live-tests",
     "live-tests/macros",
     "nexus",
@@ -250,6 +251,7 @@ default-members = [
     "internal-dns/types/versions",
     "ipcc",
     "key-manager",
+    "key-manager/types",
     "live-tests",
     "live-tests/macros",
     "nexus",
@@ -560,6 +562,7 @@ ipnetwork = { version = "0.21", features = ["schemars", "serde"] }
 ispf = { git = "https://github.com/oxidecomputer/ispf" }
 jiff = "0.2.15"
 key-manager = { path = "key-manager" }
+key-manager-types = { path = "key-manager/types" }
 kstat-rs = "0.2.4"
 libc = "0.2.174"
 libipcc = { git = "https://github.com/oxidecomputer/ipcc-rs", rev = "524eb8f125003dff50b9703900c6b323f00f9e1b" }

Cargo.toml

@@ -25,6 +25,7 @@ http.workspace = true
 iddqd.workspace = true
 ipnetwork.workspace = true
 itertools.workspace = true
+key-manager-types.workspace = true
 libc.workspace = true
 macaddr.workspace = true
 nix.workspace = true

illumos-utils/Cargo.toml

@@ -206,6 +206,42 @@ pub struct GetValueError {
 #[error("Failed to list snapshots: {0}")]
 pub struct ListSnapshotsError(#[from] crate::ExecutionError);
 
+/// Error returned by [`Zfs::change_key`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to change encryption key for dataset '{name}'")]
+pub struct ChangeKeyError {
+    pub name: String,
+    #[source]
+    pub err: anyhow::Error,
+}
+
+/// Error returned by [`Zfs::load_key`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to load encryption key for dataset '{name}'")]
+pub struct LoadKeyError {
+    pub name: String,
+    #[source]
+    pub err: crate::ExecutionError,
+}
+
+/// Error returned by [`Zfs::dataset_exists`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to check if dataset '{name}' exists")]
+pub struct DatasetExistsError {
+    pub name: String,
+    #[source]
+    pub err: crate::ExecutionError,
+}
+
+/// Error returned by [`Zfs::unload_key`].
+#[derive(thiserror::Error, Debug)]
+#[error("Failed to unload encryption key for dataset '{name}'")]
+pub struct UnloadKeyError {
+    pub name: String,
+    #[source]
+    pub err: crate::ExecutionError,
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error(
     "Failed to create snapshot '{snap_name}' from filesystem '{filesystem}': {err}"
@@ -523,11 +559,19 @@ pub struct DatasetProperties {
     /// string so that unexpected compression formats don't prevent inventory
     /// from being collected.
     pub compression: String,
+    /// The encryption key epoch for this dataset.
+    ///
+    /// Only present on encrypted datasets that directly hold a key (e.g.,
+    /// crypt datasets on U.2s). Not present on datasets that inherit
+    /// encryption from a parent.
+    pub epoch: Option<u64>,
 }
 
 impl DatasetProperties {
-    const ZFS_GET_PROPS: &'static str =
-        "oxide:uuid,name,mounted,avail,used,quota,reservation,compression";
+    const ZFS_GET_PROPS: &'static str = concat!(
+        "oxide:uuid,oxide:epoch,",
+        "name,mounted,avail,used,quota,reservation,compression",
+    );
 }
 
 impl TryFrom<&DatasetProperties> for SharedDatasetConfig {
@@ -648,6 +692,18 @@ impl DatasetProperties {
                     .get("compression")
                     .map(|(prop, _source)| prop.to_string())
                     .ok_or_else(|| anyhow!("Missing 'compression'"))?;
+                // The epoch property is only present on encrypted datasets.
+                // Like oxide:uuid, we ignore inherited values.
+                let epoch = props
+                    .get("oxide:epoch")
+                    .filter(|(prop, source)| {
+                        !source.starts_with("inherited") && *prop != "-"
+                    })
+                    .map(|(prop, _source)| {
+                        prop.parse::<u64>()
+                            .context("Failed to parse 'oxide:epoch'")
+                    })
+                    .transpose()?;
 
                 Ok(DatasetProperties {
                     id,
@@ -658,6 +714,7 @@ impl DatasetProperties {
                     quota,
                     reservation,
                     compression,
+                    epoch,
                 })
             })
             .collect::<Result<Vec<_>, _>>()
@@ -1197,7 +1254,7 @@ impl Zfs {
         name: &str,
         mountpoint: &Mountpoint,
     ) -> Result<(), EnsureDatasetErrorRaw> {
-        let mount_info = Self::dataset_exists(name, mountpoint).await?;
+        let mount_info = Self::dataset_mount_info(name, mountpoint).await?;
         if !mount_info.exists {
             return Err(EnsureDatasetErrorRaw::DoesNotExist);
         }
@@ -1246,7 +1303,7 @@ impl Zfs {
             additional_options,
         }: DatasetEnsureArgs<'_>,
     ) -> Result<(), EnsureDatasetErrorRaw> {
-        let dataset_info = Self::dataset_exists(name, &mountpoint).await?;
+        let dataset_info = Self::dataset_mount_info(name, &mountpoint).await?;
 
         // Non-zoned datasets with an explicit mountpoint and the
         // "canmount=on" property should be mounted within the global zone.
@@ -1365,9 +1422,29 @@ impl Zfs {
         Ok(())
     }
 
-    // Return (true, mounted) if the dataset exists, (false, false) otherwise,
-    // where mounted is if the dataset is mounted.
-    async fn dataset_exists(
+    /// Check if a ZFS dataset exists.
+    pub async fn dataset_exists(
+        name: &str,
+    ) -> Result<bool, DatasetExistsError> {
+        let mut cmd = Command::new(ZFS);
+        cmd.args(&["list", "-H", name]);
+        match execute_async(&mut cmd).await {
+            Ok(_) => Ok(true),
+            Err(crate::ExecutionError::CommandFailure(ref info))
+                if info.stderr.contains("does not exist") =>
+            {
+                Ok(false)
+            }
+            Err(err) => Err(DatasetExistsError { name: name.to_string(), err }),
+        }
+    }
+
+    /// Get mount info for a dataset, validating its mountpoint.
+    ///
+    /// Returns (exists=true, mounted) if the dataset exists with the expected
+    /// mountpoint, (exists=false, mounted=false) if it doesn't exist.
+    /// Returns an error if the dataset exists but has an unexpected mountpoint.
+    async fn dataset_mount_info(
         name: &str,
         mountpoint: &Mountpoint,
     ) -> Result<DatasetMountInfo, EnsureDatasetErrorRaw> {
@@ -1523,6 +1600,52 @@ impl Zfs {
         })
     }
 
+    /// Change the encryption key and set the oxide:epoch property.
+    ///
+    /// This operation is used for ZFS key rotation when a new Trust Quorum
+    /// epoch is committed. The caller is responsible for writing the new key
+    /// to the dataset's keylocation before calling this, and zeroing the
+    /// keyfile afterward.
+    pub async fn change_key(
+        dataset: &str,
+        epoch: u64,
+    ) -> Result<(), ChangeKeyError> {
+        let epoch_prop = format!("oxide:epoch={epoch}");
+        let mut cmd = Command::new(PFEXEC);
+        cmd.args(&[ZFS, "change-key", "-o", &epoch_prop, dataset]);
+        execute_async(&mut cmd).await.map_err(|e| ChangeKeyError {
+            name: dataset.to_string(),
+            err: e.into(),
+        })?;
+        Ok(())
+    }
+
+    /// Load the encryption key for an encrypted ZFS dataset.
+    ///
+    /// This makes the dataset accessible for mounting. The key must have
+    /// previously been written to the dataset's keylocation.
+    pub async fn load_key(name: &str) -> Result<(), LoadKeyError> {
+        let mut cmd = Command::new(PFEXEC);
+        cmd.args(&[ZFS, "load-key", name]);
+        execute_async(&mut cmd)
+            .await
+            .map(|_| ())
+            .map_err(|err| LoadKeyError { name: name.to_string(), err })
+    }
+
+    /// Unload the encryption key for an encrypted ZFS dataset.
+    ///
+    /// This is used for cleanup after failed key operations or during
+    /// trial decryption recovery. The dataset must not be mounted.
+    pub async fn unload_key(name: &str) -> Result<(), UnloadKeyError> {
+        let mut cmd = Command::new(PFEXEC);
+        cmd.args(&[ZFS, "unload-key", name]);
+        execute_async(&mut cmd)
+            .await
+            .map(|_| ())
+            .map_err(|err| UnloadKeyError { name: name.to_string(), err })
+    }
+
     /// Calls "zfs get" to acquire multiple values
     ///
     /// - `names`: The properties being acquired

illumos-utils/src/zfs.rs

@@ -10,6 +10,7 @@ workspace = true
 [dependencies]
 async-trait.workspace = true
 hkdf.workspace = true
+key-manager-types.workspace = true
 omicron-common.workspace = true
 secrecy.workspace = true
 sha3.workspace = true

key-manager/Cargo.toml

@@ -9,7 +9,9 @@ use std::fmt::Debug;
 
 use async_trait::async_trait;
 use hkdf::Hkdf;
-use secrecy::{ExposeSecret, ExposeSecretMut, SecretBox};
+use key_manager_types::Aes256GcmDiskEncryptionKey;
+pub use key_manager_types::VersionedAes256GcmDiskEncryptionKey;
+use secrecy::{ExposeSecret, SecretBox};
 use sha3::Sha3_256;
 use slog::{Logger, o, warn};
 use tokio::sync::{mpsc, oneshot};
@@ -52,27 +54,6 @@ pub enum Error {
     SecretRetrieval(#[from] SecretRetrieverError),
 }
 
-/// Derived Disk Encryption key
-#[derive(Default)]
-struct Aes256GcmDiskEncryptionKey(SecretBox<[u8; 32]>);
-
-/// A Disk encryption key for a given epoch to be used with ZFS datasets for
-/// U.2 devices
-pub struct VersionedAes256GcmDiskEncryptionKey {
-    epoch: u64,
-    key: Aes256GcmDiskEncryptionKey,
-}
-
-impl VersionedAes256GcmDiskEncryptionKey {
-    pub fn epoch(&self) -> u64 {
-        self.epoch
-    }
-
-    pub fn expose_secret(&self) -> &[u8; 32] {
-        &self.key.0.expose_secret()
-    }
-}
-
 /// A request sent from a [`StorageKeyRequester`] to the [`KeyManager`].
 enum StorageKeyRequest {
     GetKey {
@@ -255,11 +236,11 @@ impl<S: SecretRetriever> KeyManager<S> {
                 disk_id.model.as_bytes(),
                 disk_id.serial.as_bytes(),
             ],
-            key.0.expose_secret_mut(),
+            key.expose_secret_mut(),
         )
         .unwrap();
 
-        Ok(VersionedAes256GcmDiskEncryptionKey { epoch, key })
+        Ok(VersionedAes256GcmDiskEncryptionKey::new(epoch, key))
     }
 
     /// Return the epochs for all secrets which are loaded
@@ -309,6 +290,9 @@ pub enum SecretRetrieverError {
 
     #[error("Trust quorum error: {0}")]
     TrustQuorum(String),
+
+    #[error("Timeout retrieving secret")]
+    Timeout,
 }
 
 /// A mechanism for retrieving a secrets to use as input key material to HKDF-
@@ -405,7 +389,7 @@ mod tests {
         };
         let epoch = 0;
         let key = km.disk_encryption_key(epoch, &disk_id).await.unwrap();
-        assert_eq!(key.epoch, epoch);
+        assert_eq!(key.epoch(), epoch);
 
         // Key derivation is deterministic based on disk_id and loaded secrets
         let key2 = km.disk_encryption_key(epoch, &disk_id).await.unwrap();
@@ -436,8 +420,8 @@ mod tests {
         let epoch = 0;
         let key1 = km.disk_encryption_key(epoch, &id_1).await.unwrap();
         let key2 = km.disk_encryption_key(epoch, &id_2).await.unwrap();
-        assert_eq!(key1.epoch, epoch);
-        assert_eq!(key2.epoch, epoch);
+        assert_eq!(key1.epoch(), epoch);
+        assert_eq!(key2.epoch(), epoch);
         assert_ne!(key1.expose_secret(), key2.expose_secret());
     }
 

key-manager/src/lib.rs

@@ -0,0 +1,12 @@
+[package]
+name = "key-manager-types"
+version = "0.1.0"
+edition.workspace = true
+license = "MPL-2.0"
+
+[lints]
+workspace = true
+
+[dependencies]
+secrecy.workspace = true
+omicron-workspace-hack.workspace = true