feat(kms): report which keys have outlived their rotation period (#5769)

rustfs/backlog#1636 rejected a built-in rotation scheduler: rotation is a policy decision with a per-backend cost and a hard upgrade-ordering constraint, and a server that rotated on its own would make that decision on an operator's behalf at a moment they did not choose. This is what that issue resolved to deliver instead — the signal, without the actuator.

RUSTFS_KMS_ROTATION_MAX_AGE_SECS names the period. Unset leaves the verdict unreported rather than assuming a policy, because how often keys must be rotated is a compliance decision and a built-in default would report keys as overdue against a rule nobody wrote; an unparsable value is refused the same way, loudly. Values below an hour are raised to it, since a threshold of seconds reports every key as overdue moments after it was rotated and teaches operators to ignore the signal.

KeyInfo gains rotation_due and rotation_due_reason, both additive on the wire and both filled in by the manager rather than by each backend, so no two backends can disagree about what overdue means. A backend that does not advertise rotation reports unsupported and is never reported as due — it must not be told to do something it cannot. A key with no recorded rotation is measured from creation, which is how long its material has actually been in use, and is distinguished from a stale rotation so an operator can tell "overdue again" from "never once". Ages are computed saturating, so a timestamp from a node running ahead cannot manufacture an overdue key.

The verdict is advisory in the strongest sense: nothing consults it before encrypting or decrypting, a key reported as due keeps serving traffic, and readiness is unaffected.

The single-key describe response deliberately does not carry the verdict. Its type records a creation date but no rotation timestamp, so a verdict computed there could not tell a key rotated last week from one never rotated, and reporting never_rotated for a key that was in fact rotated is worse than reporting nothing.

The wraps-based branch the issue also specifies is not implemented: it depends on the per-key wrap accounting that does not exist yet.

crates/kms/src/backends/aws.rs

@@ -655,6 +655,8 @@ impl KmsBackend for AwsKmsBackend {
                 created_at: metadata.creation_date,
                 rotated_at: None,
                 created_by: None,
+                rotation_due: false,
+                rotation_due_reason: None,
             });
         }
 

crates/kms/src/backends/static_kms.rs

@@ -278,6 +278,8 @@ impl StaticKmsBackend {
             created_at: metadata.creation_date,
             rotated_at: None,
             created_by: None,
+            rotation_due: false,
+            rotation_due_reason: None,
         })
     }
 

crates/kms/src/backends/vault.rs

@@ -1225,6 +1225,8 @@ impl VaultKmsClient {
             created_at: key_data.created_at,
             rotated_at: key_data.rotated_at,
             created_by: None,
+            rotation_due: false,
+            rotation_due_reason: None,
         })
     }
 

crates/kms/src/backends/vault_transit.rs

@@ -704,6 +704,8 @@ impl VaultTransitKmsClient {
             created_at: metadata.created_at,
             rotated_at: None,
             created_by: metadata.created_by,
+            rotation_due: false,
+            rotation_due_reason: None,
         })
     }
 

crates/kms/src/config.rs

@@ -37,6 +37,10 @@ pub const ENV_KMS_VAULT_APPROLE_MOUNT: &str = "RUSTFS_KMS_VAULT_APPROLE_MOUNT";
 pub const ENV_KMS_VAULT_TOKEN_FILE: &str = "RUSTFS_KMS_VAULT_TOKEN_FILE";
 pub const ENV_KMS_AWS_REGION: &str = "RUSTFS_KMS_AWS_REGION";
 pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
+/// Age in whole seconds beyond which a key is reported as due for rotation;
+/// unset leaves rotation readiness unreported. Read once when the manager is
+/// built, by [`crate::manager::KmsManager`].
+pub const ENV_KMS_ROTATION_MAX_AGE_SECS: &str = "RUSTFS_KMS_ROTATION_MAX_AGE_SECS";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata";
 pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle";

crates/kms/src/deletion_worker.rs

@@ -770,6 +770,8 @@ mod tests {
             created_at,
             rotated_at,
             created_by: None,
+            rotation_due: false,
+            rotation_due_reason: None,
         }
     }
 

crates/kms/src/manager.rs

@@ -17,24 +17,57 @@
 use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::KmsBackend;
 use crate::cache::{KmsCache, KmsCacheStats};
-use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, KmsConfig};
+use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, KmsConfig};
 use crate::deletion_worker::DeletionReferenceChecker;
 use crate::error::{KmsError, Result};
 use crate::types::{
     CancelKeyDeletionRequest, CancelKeyDeletionResponse, CreateKeyRequest, CreateKeyResponse,
     DEFAULT_PENDING_DELETION_WINDOW_DAYS, DecryptRequest, DecryptResponse, DeleteKeyRequest, DeleteKeyResponse,
     DescribeDataKeyWrappingRequest, DescribeDataKeyWrappingResponse, DescribeKeyRequest, DescribeKeyResponse, EncryptRequest,
-    EncryptResponse, GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse,
+    EncryptResponse, GenerateDataKeyRequest, GenerateDataKeyResponse, KeyInfo, ListKeysRequest, ListKeysResponse,
     MAX_PENDING_DELETION_WINDOW_DAYS, MIN_PENDING_DELETION_WINDOW_DAYS, OperationContext, RewrapDataKeyRequest,
-    RewrapDataKeyResponse,
+    RewrapDataKeyResponse, RotationDueReason,
 };
+use jiff::Zoned;
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use tokio::sync::RwLock;
 use tracing::warn;
 
 /// KMS Manager coordinates operations between backends and caching
+/// Smallest rotation age that can be configured.
+///
+/// A threshold shorter than this would report every key as overdue moments
+/// after it was rotated, which trains operators to ignore the signal.
+const MIN_ROTATION_MAX_AGE: Duration = Duration::from_secs(3600);
+
+/// Rotation age from the environment, or `None` when the signal is off.
+///
+/// Unset leaves it off rather than guessing a policy: how often a deployment
+/// must rotate is a compliance decision, and inventing a default would report
+/// keys as overdue against a rule nobody chose. An unparsable value is refused
+/// the same way, loudly, instead of falling back to a number the operator did
+/// not write.
+fn configured_rotation_max_age() -> Option<Duration> {
+    parse_rotation_max_age(std::env::var(ENV_KMS_ROTATION_MAX_AGE_SECS).ok().as_deref())
+}
+
+fn parse_rotation_max_age(value: Option<&str>) -> Option<Duration> {
+    let value = value?;
+    let Ok(seconds) = value.trim().parse::<u64>() else {
+        warn!(
+            variable = ENV_KMS_ROTATION_MAX_AGE_SECS,
+            "ignoring unparsable KMS rotation age; rotation readiness stays unreported"
+        );
+        return None;
+    };
+    if seconds == 0 {
+        return None;
+    }
+    Some(Duration::from_secs(seconds).max(MIN_ROTATION_MAX_AGE))
+}
+
 #[derive(Clone)]
 pub struct KmsManager {
     backend: Arc<dyn KmsBackend>,
@@ -45,6 +78,10 @@ pub struct KmsManager {
     audit_sink: Option<Arc<dyn KmsAuditSink>>,
     allow_immediate_deletion: bool,
     reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
+    /// Age beyond which a key is reported as due for rotation; `None` leaves
+    /// the verdict unreported. Read once at construction so a listing cannot
+    /// change its answer halfway through.
+    rotation_max_age: Option<Duration>,
 }
 
 impl KmsManager {
@@ -65,6 +102,7 @@ impl KmsManager {
             audit_sink: None,
             allow_immediate_deletion: config.allow_immediate_deletion,
             reference_checker: None,
+            rotation_max_age: configured_rotation_max_age(),
         }
     }
 
@@ -248,12 +286,59 @@ impl KmsManager {
     /// List keys on behalf of `context`'s principal
     pub async fn list_keys_with_context(&self, request: ListKeysRequest, context: &OperationContext) -> Result<ListKeysResponse> {
         let started = Instant::now();
-        let result = self.backend.list_keys(request).await;
+        let mut result = self.backend.list_keys(request).await;
+        if let Ok(response) = result.as_mut() {
+            let rotates = self.backend.capabilities().rotate;
+            let now = Zoned::now();
+            for key in &mut response.keys {
+                self.apply_rotation_readiness(key, rotates, &now);
+            }
+        }
         // Listing spans keys, so the record carries no key id.
         self.audit(KmsAuditOperation::ListKeys, context, None, started, &result);
         result
     }
 
+    /// Fill in the advisory rotation verdict for one listed key.
+    ///
+    /// Decided here rather than in each backend so no two backends can disagree
+    /// about what "overdue" means, and so a backend that cannot rotate is never
+    /// the one deciding whether it should. Nothing consults the verdict before
+    /// encrypting or decrypting: a key reported as due keeps serving traffic.
+    fn apply_rotation_readiness(&self, key: &mut KeyInfo, backend_rotates: bool, now: &Zoned) {
+        if !backend_rotates {
+            // Reported, not silently omitted: an operator chasing an overdue key
+            // needs to know the answer is "this backend cannot rotate at all",
+            // which is a backend choice to revisit rather than a key to fix.
+            key.rotation_due = false;
+            key.rotation_due_reason = Some(RotationDueReason::Unsupported);
+            return;
+        }
+        let Some(max_age) = self.rotation_max_age else {
+            key.rotation_due = false;
+            key.rotation_due_reason = None;
+            return;
+        };
+
+        // A key that was never rotated is measured from when it was created:
+        // that is how long its material has been in use, which is the quantity
+        // the threshold is about.
+        let (since, reason) = match key.rotated_at.as_ref() {
+            Some(rotated_at) => (rotated_at, RotationDueReason::Age),
+            None => (&key.created_at, RotationDueReason::NeverRotated),
+        };
+        // Saturating: a timestamp from a node running ahead must not read as an
+        // enormous age and report a fresh key as overdue.
+        let age = Duration::from_secs((now.timestamp().as_second() - since.timestamp().as_second()).max(0) as u64);
+        if age >= max_age {
+            key.rotation_due = true;
+            key.rotation_due_reason = Some(reason);
+        } else {
+            key.rotation_due = false;
+            key.rotation_due_reason = None;
+        }
+    }
+
     /// Get cache statistics, or `None` when caching is disabled
     pub async fn cache_stats(&self) -> Option<KmsCacheStats> {
         if self.enable_cache {
@@ -542,7 +627,7 @@ mod tests {
     use crate::audit::KmsAuditOutcome;
     use crate::backends::local::LocalKmsBackend;
     use crate::error::KmsError;
-    use crate::types::{KeyMetadata, KeySpec, KeyState, KeyUsage};
+    use crate::types::{KeyMetadata, KeySpec, KeyState, KeyStatus, KeyUsage};
     use async_trait::async_trait;
     use base64::Engine as _;
     use jiff::Zoned;
@@ -1578,4 +1663,140 @@ mod tests {
             assert_eq!(state, KeyState::Enabled, "cancelling must restore the key");
         }
     }
+    /// The threshold is a compliance decision, so nothing is inferred: unset
+    /// and unparsable both leave the signal off rather than reporting keys
+    /// overdue against a rule nobody wrote. A value below the floor is raised,
+    /// because a threshold of a few seconds reports every key as overdue moments
+    /// after it was rotated and trains operators to ignore the signal.
+    #[test]
+    fn rotation_age_is_configured_or_absent_never_guessed() {
+        assert_eq!(parse_rotation_max_age(None), None);
+        assert_eq!(parse_rotation_max_age(Some("not-a-number")), None);
+        assert_eq!(parse_rotation_max_age(Some("")), None);
+        assert_eq!(parse_rotation_max_age(Some("-1")), None);
+        assert_eq!(parse_rotation_max_age(Some("0")), None);
+
+        assert_eq!(parse_rotation_max_age(Some("1")), Some(MIN_ROTATION_MAX_AGE));
+        assert_eq!(parse_rotation_max_age(Some(" 86400 ")), Some(Duration::from_secs(86_400)));
+        assert_eq!(
+            parse_rotation_max_age(Some(&MIN_ROTATION_MAX_AGE.as_secs().to_string())),
+            Some(MIN_ROTATION_MAX_AGE)
+        );
+    }
+
+    fn readiness_manager(rotation_max_age: Option<Duration>) -> KmsManager {
+        let temp_dir = tempfile::tempdir().expect("temp dir");
+        let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        let mut manager = KmsManager::new(Arc::new(ScriptedBackend::succeeding()), config);
+        manager.rotation_max_age = rotation_max_age;
+        manager
+    }
+
+    fn aged_key(rotated_at: Option<Zoned>, created_at: Zoned) -> KeyInfo {
+        KeyInfo {
+            key_id: "key-a".to_string(),
+            description: None,
+            algorithm: "AES_256".to_string(),
+            usage: KeyUsage::EncryptDecrypt,
+            status: KeyStatus::Active,
+            version: 1,
+            metadata: HashMap::new(),
+            tags: HashMap::new(),
+            created_at,
+            rotated_at,
+            created_by: None,
+            rotation_due: false,
+            rotation_due_reason: None,
+        }
+    }
+
+    /// The whole decision matrix, including the one case that must never fire:
+    /// a backend that cannot rotate is never told to rotate.
+    #[test]
+    fn rotation_readiness_reports_but_never_asks_the_impossible() {
+        let now = Zoned::now();
+        let long_ago = &now - jiff::Span::new().days(400);
+        let recently = &now - jiff::Span::new().hours(1);
+        let day = Duration::from_secs(86_400);
+
+        // A backend without rotation is reported as such and never as overdue,
+        // however ancient the key and however short the threshold.
+        let manager = readiness_manager(Some(Duration::from_secs(1)));
+        let mut key = aged_key(None, long_ago.clone());
+        manager.apply_rotation_readiness(&mut key, false, &now);
+        assert!(!key.rotation_due, "a backend that cannot rotate must never be told to");
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Unsupported));
+
+        // No threshold configured: no verdict, on any key.
+        let manager = readiness_manager(None);
+        let mut key = aged_key(None, long_ago.clone());
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert!(!key.rotation_due);
+        assert_eq!(key.rotation_due_reason, None);
+
+        let manager = readiness_manager(Some(day));
+
+        // Rotated, but longer ago than the threshold.
+        let mut key = aged_key(Some(long_ago.clone()), long_ago.clone());
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert!(key.rotation_due);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Age));
+
+        // Never rotated, and in use longer than the threshold. Measured from
+        // creation, and distinguished from a stale rotation so an operator can
+        // tell "overdue again" from "never once".
+        let mut key = aged_key(None, long_ago.clone());
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert!(key.rotation_due);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::NeverRotated));
+
+        // Rotated within the threshold, and created long ago: recency wins.
+        let mut key = aged_key(Some(recently), long_ago);
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert!(!key.rotation_due);
+        assert_eq!(key.rotation_due_reason, None);
+
+        // A timestamp from a node running ahead must not read as an enormous
+        // age and report a fresh key as overdue.
+        let ahead = &now + jiff::Span::new().days(2);
+        let mut key = aged_key(Some(ahead.clone()), ahead);
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert!(!key.rotation_due, "clock skew must not manufacture an overdue key");
+    }
+
+    /// The two fields are additive on the wire: a payload written before they
+    /// existed still deserializes, and a key with no verdict serializes exactly
+    /// as it did before.
+    #[test]
+    fn rotation_readiness_fields_are_additive_on_the_wire() {
+        let legacy = serde_json::json!({
+            "key_id": "key-a",
+            "description": null,
+            "algorithm": "AES_256",
+            "usage": "EncryptDecrypt",
+            "status": "Active",
+            "version": 1,
+            "metadata": {},
+            "tags": {},
+            "created_at": "2026-01-01T00:00:00Z[UTC]",
+            "rotated_at": null,
+            "created_by": null,
+        });
+        let decoded: KeyInfo = serde_json::from_value(legacy).expect("a payload without the fields must still decode");
+        assert!(!decoded.rotation_due);
+        assert_eq!(decoded.rotation_due_reason, None);
+
+        let encoded = serde_json::to_value(&decoded).expect("encode");
+        assert_eq!(encoded.get("rotation_due"), Some(&serde_json::Value::Bool(false)));
+        assert!(
+            encoded.get("rotation_due_reason").is_none(),
+            "an absent verdict must not add a field for old consumers to trip over"
+        );
+
+        let mut due = decoded;
+        due.rotation_due = true;
+        due.rotation_due_reason = Some(RotationDueReason::NeverRotated);
+        let encoded = serde_json::to_value(&due).expect("encode");
+        assert_eq!(encoded.get("rotation_due_reason").and_then(|value| value.as_str()), Some("never_rotated"));
+    }
 }

crates/kms/src/types.rs

@@ -204,6 +204,23 @@ pub enum KeyStatus {
     Deleted,
 }
 
+/// Why a key carries the rotation-readiness verdict it does.
+///
+/// Reported alongside [`KeyInfo::rotation_due`] so an operator can tell an
+/// overdue key from one the server cannot judge at all; new variants may be
+/// added, so consumers must treat an unknown value as "no verdict".
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum RotationDueReason {
+    /// The key was last rotated longer ago than the configured maximum age.
+    Age,
+    /// The key has never been rotated and has existed longer than the
+    /// configured maximum age.
+    NeverRotated,
+    /// The backend cannot rotate keys at all, so no age makes one due.
+    Unsupported,
+}
+
 /// Information about a key
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct KeyInfo {
@@ -229,6 +246,19 @@ pub struct KeyInfo {
     pub rotated_at: Option<Zoned>,
     /// Key creator
     pub created_by: Option<String>,
+    /// Whether the key has outlived the configured rotation age.
+    ///
+    /// Advisory: nothing consults it before encrypting or decrypting, and a key
+    /// reported as due keeps serving traffic unchanged. Backends leave it at
+    /// its default — the verdict is filled in by the manager, which is the only
+    /// place that knows both the configured age and whether the backend can
+    /// rotate at all.
+    #[serde(default)]
+    pub rotation_due: bool,
+    /// Why [`KeyInfo::rotation_due`] holds its value, absent when there is no
+    /// verdict to explain.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub rotation_due_reason: Option<RotationDueReason>,
 }
 
 impl From<MasterKeyInfo> for KeyInfo {
@@ -245,6 +275,8 @@ impl From<MasterKeyInfo> for KeyInfo {
             created_at: master_key.created_at,
             rotated_at: master_key.rotated_at,
             created_by: master_key.created_by,
+            rotation_due: false,
+            rotation_due_reason: None,
         }
     }
 }