feat(kms): allow key description and tag updates (#5546)

crates/kms/src/backends/local.rs

@@ -14,7 +14,10 @@
 
 //! Local file-based KMS backend implementation
 
-use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits};
+use crate::backends::{
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits,
+    ensure_tag_keys_are_mutable,
+};
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -1294,6 +1297,32 @@ impl LocalKmsClient {
         Ok(())
     }
 
+    /// Read-modify-write of a key's mutable metadata under the per-key write
+    /// lock, carrying the existing key material over untouched.
+    ///
+    /// `mutate` reports whether it changed anything; an unchanged record is
+    /// never rewritten, so a repeated no-op update neither rewrites the key
+    /// file nor risks a failed commit on an unrelated call.
+    async fn update_key_metadata<F>(&self, key_id: &str, mutate: F) -> Result<()>
+    where
+        F: FnOnce(&mut MasterKeyInfo) -> Result<bool>,
+    {
+        let _write_guard = self.lock_key_for_write(key_id).await;
+        let mut master_key = self.load_master_key(key_id).await?;
+        if !mutate(&mut master_key)? {
+            return Ok(());
+        }
+
+        // Preserve the existing key material (see enable_key): a metadata edit
+        // must never regenerate the master key, or every DEK wrapped by it
+        // becomes undecryptable.
+        let key_material = self.get_key_material(key_id).await?;
+        self.save_master_key(&master_key, &key_material).await?;
+
+        debug!(key_id, "Local KMS key metadata updated");
+        Ok(())
+    }
+
     /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
     #[cfg(test)]
     pub(crate) async fn schedule_key_deletion(
@@ -1412,12 +1441,15 @@ impl KmsBackend for LocalKmsBackend {
             // Generate key material
             let key_material = generate_key_material(algorithm)?;
 
-            let master_key = MasterKeyInfo::new_with_description(
+            let mut master_key = MasterKeyInfo::new_with_description(
                 key_id.clone(),
                 algorithm.to_string(),
                 Some("local-kms".to_string()),
                 request.description.clone(),
             );
+            // Persist the caller's tags: the response below reports them, and
+            // describe_key reads them back out of this record.
+            master_key.metadata = request.tags.clone();
 
             // Save to disk
             self.client.save_new_master_key(&master_key, &key_material).await?;
@@ -1685,6 +1717,44 @@ impl KmsBackend for LocalKmsBackend {
         self.client.disable_key(key_id, None).await
     }
 
+    async fn update_key_description(&self, key_id: &str, description: Option<&str>) -> Result<()> {
+        self.client
+            .update_key_metadata(key_id, |master_key| {
+                if master_key.description.as_deref() == description {
+                    return Ok(false);
+                }
+                master_key.description = description.map(str::to_string);
+                Ok(true)
+            })
+            .await
+    }
+
+    async fn tag_key(&self, key_id: &str, tags: &HashMap<String, String>) -> Result<()> {
+        ensure_tag_keys_are_mutable(tags.keys().map(String::as_str))?;
+        self.client
+            .update_key_metadata(key_id, |master_key| {
+                let mut changed = false;
+                for (tag_key, value) in tags {
+                    changed |= master_key.metadata.insert(tag_key.clone(), value.clone()).as_ref() != Some(value);
+                }
+                Ok(changed)
+            })
+            .await
+    }
+
+    async fn untag_key(&self, key_id: &str, tag_keys: &[String]) -> Result<()> {
+        ensure_tag_keys_are_mutable(tag_keys.iter().map(String::as_str))?;
+        self.client
+            .update_key_metadata(key_id, |master_key| {
+                let mut changed = false;
+                for tag_key in tag_keys {
+                    changed |= master_key.metadata.remove(tag_key).is_some();
+                }
+                Ok(changed)
+            })
+            .await
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
@@ -1697,6 +1767,7 @@ impl KmsBackend for LocalKmsBackend {
             .with_enable_disable(true)
             .with_schedule_deletion(true)
             .with_physical_delete(true)
+            .with_update_key_metadata(true)
     }
 
     async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {

crates/kms/src/backends/mod.rs

@@ -19,6 +19,7 @@ use crate::types::*;
 use async_trait::async_trait;
 use jiff::Zoned;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 
 pub mod aws;
 #[cfg(test)]
@@ -99,6 +100,32 @@ pub(crate) fn ensure_key_status_permits(key_id: &str, status: &KeyStatus, operat
     ensure_key_state_permits(key_id, &state, operation)
 }
 
+/// Tag key that carries the key's identity rather than user metadata.
+///
+/// The key-creation path lifts `name` out of the caller's tag map and uses it
+/// as the key id, so a key's `name` tag and its id are the same string.
+/// Rewriting or dropping it after creation would leave the key addressable
+/// under an id its own metadata no longer states. Metadata updates therefore
+/// reject it; only creation may set it.
+pub const RESERVED_KEY_NAME_TAG: &str = "name";
+
+/// Reject a metadata update that would rewrite or remove
+/// [`RESERVED_KEY_NAME_TAG`].
+///
+/// Enforced by every backend that implements tag updates, so no call path —
+/// including a direct [`KmsBackend`] user that bypasses the manager — can
+/// detach a key from its identity.
+pub(crate) fn ensure_tag_keys_are_mutable<'a>(tag_keys: impl IntoIterator<Item = &'a str>) -> Result<()> {
+    for tag_key in tag_keys {
+        if tag_key == RESERVED_KEY_NAME_TAG {
+            return Err(KmsError::invalid_parameter(format!(
+                "Tag '{RESERVED_KEY_NAME_TAG}' identifies the key and cannot be updated or removed"
+            )));
+        }
+    }
+    Ok(())
+}
+
 /// Simplified KMS backend interface for manager
 #[async_trait]
 pub trait KmsBackend: Send + Sync {
@@ -154,6 +181,38 @@ pub trait KmsBackend: Send + Sync {
         Err(KmsError::unsupported_capability("backend without rotation support", "rotate_key"))
     }
 
+    /// Replace a key's free-form description; `None` clears it.
+    ///
+    /// Backends that advertise [`BackendCapabilities::update_key_metadata`]
+    /// must override this method; the default rejects the operation.
+    async fn update_key_description(&self, _key_id: &str, _description: Option<&str>) -> Result<()> {
+        Err(KmsError::unsupported_capability(
+            "backend without key metadata updates",
+            "update_key_description",
+        ))
+    }
+
+    /// Add or overwrite the given tags, leaving every other tag untouched.
+    ///
+    /// Implementations must run [`ensure_tag_keys_are_mutable`] before
+    /// persisting anything. Backends that advertise
+    /// [`BackendCapabilities::update_key_metadata`] must override this method;
+    /// the default rejects the operation.
+    async fn tag_key(&self, _key_id: &str, _tags: &HashMap<String, String>) -> Result<()> {
+        Err(KmsError::unsupported_capability("backend without key metadata updates", "tag_key"))
+    }
+
+    /// Remove the given tags.
+    ///
+    /// Tags that are not set are ignored, so repeating the call is a no-op
+    /// rather than an error. Implementations must run
+    /// [`ensure_tag_keys_are_mutable`] before persisting anything. Backends
+    /// that advertise [`BackendCapabilities::update_key_metadata`] must
+    /// override this method; the default rejects the operation.
+    async fn untag_key(&self, _key_id: &str, _tag_keys: &[String]) -> Result<()> {
+        Err(KmsError::unsupported_capability("backend without key metadata updates", "untag_key"))
+    }
+
     /// Health check
     async fn health_check(&self) -> Result<bool>;
 
@@ -223,6 +282,8 @@ pub struct BackendCapabilities {
     pub versioning: bool,
     /// Irreversible physical deletion of key material
     pub physical_delete: bool,
+    /// Updating a key's description and tags after creation
+    pub update_key_metadata: bool,
 }
 
 impl BackendCapabilities {
@@ -239,6 +300,7 @@ impl BackendCapabilities {
             schedule_deletion: false,
             versioning: false,
             physical_delete: false,
+            update_key_metadata: false,
         }
     }
 
@@ -289,6 +351,12 @@ impl BackendCapabilities {
         self.physical_delete = physical_delete;
         self
     }
+
+    /// Set whether description and tag updates are supported
+    pub const fn with_update_key_metadata(mut self, update_key_metadata: bool) -> Self {
+        self.update_key_metadata = update_key_metadata;
+        self
+    }
 }
 
 impl Default for BackendCapabilities {
@@ -367,6 +435,7 @@ mod tests {
         assert!(!capabilities.schedule_deletion);
         assert!(!capabilities.versioning);
         assert!(!capabilities.physical_delete);
+        assert!(!capabilities.update_key_metadata);
     }
 
     #[tokio::test]
@@ -375,6 +444,12 @@ mod tests {
             ("enable_key", MinimalBackend.enable_key("any-key").await),
             ("disable_key", MinimalBackend.disable_key("any-key").await),
             ("rotate_key", MinimalBackend.rotate_key("any-key").await),
+            (
+                "update_key_description",
+                MinimalBackend.update_key_description("any-key", Some("new")).await,
+            ),
+            ("tag_key", MinimalBackend.tag_key("any-key", &HashMap::new()).await),
+            ("untag_key", MinimalBackend.untag_key("any-key", &[]).await),
         ] {
             let error = result.expect_err("backends must opt in to lifecycle operations by overriding them");
             assert!(
@@ -396,6 +471,20 @@ mod tests {
         );
     }
 
+    #[test]
+    fn identity_tag_is_rejected_by_metadata_updates() {
+        let error = ensure_tag_keys_are_mutable([RESERVED_KEY_NAME_TAG])
+            .expect_err("the identity tag must not be writable through a metadata update");
+        assert!(
+            matches!(&error, KmsError::InvalidOperation { message } if message.contains(RESERVED_KEY_NAME_TAG)),
+            "expected a typed rejection naming the tag, got {error:?}"
+        );
+
+        // Ordinary tags — including ones that merely contain the reserved name
+        // — stay writable.
+        ensure_tag_keys_are_mutable(["team", "nickname", "Name"]).expect("ordinary tags must remain writable");
+    }
+
     #[tokio::test]
     async fn local_backend_capabilities_golden() {
         let temp_dir = tempfile::tempdir().expect("temp dir should be created");

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__local_backend_capabilities.snap

@@ -10,5 +10,6 @@ expression: capabilities_snapshot(backend.capabilities())
   "physical_delete": true,
   "rotate": false,
   "schedule_deletion": true,
+  "update_key_metadata": true,
   "versioning": false
 }

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__static_backend_capabilities.snap

@@ -10,5 +10,6 @@ expression: capabilities_snapshot(backend.capabilities())
   "physical_delete": false,
   "rotate": false,
   "schedule_deletion": false,
+  "update_key_metadata": false,
   "versioning": false
 }

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_kv2_backend_capabilities.snap

@@ -10,5 +10,6 @@ expression: capabilities_snapshot(backend.capabilities())
   "physical_delete": true,
   "rotate": true,
   "schedule_deletion": true,
+  "update_key_metadata": true,
   "versioning": true
 }

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_transit_backend_capabilities.snap

@@ -10,5 +10,6 @@ expression: capabilities_snapshot(backend.capabilities())
   "physical_delete": true,
   "rotate": true,
   "schedule_deletion": true,
+  "update_key_metadata": true,
   "versioning": true
 }

crates/kms/src/backends/static_kms.rs

@@ -431,6 +431,32 @@ mod tests {
         (backend, key_id, key)
     }
 
+    #[tokio::test]
+    async fn metadata_updates_report_the_capability_gap() {
+        let (backend, key_id, _key) = create_test_backend().await;
+        assert!(
+            !backend.capabilities().update_key_metadata,
+            "Static KMS owns no mutable key record, so it must not advertise metadata updates"
+        );
+
+        for (operation, result) in [
+            ("update_key_description", backend.update_key_description(&key_id, Some("new")).await),
+            (
+                "tag_key",
+                backend
+                    .tag_key(&key_id, &HashMap::from([("team".to_string(), "storage".to_string())]))
+                    .await,
+            ),
+            ("untag_key", backend.untag_key(&key_id, &["team".to_string()]).await),
+        ] {
+            let error = result.expect_err("Static KMS is read-only: metadata updates must be rejected");
+            assert!(
+                matches!(error, KmsError::UnsupportedCapability { .. }),
+                "expected UnsupportedCapability for {operation}, got {error:?}"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_generate_and_decrypt_data_key() {
         let (backend, key_id, _key) = create_test_backend().await;

crates/kms/src/backends/vault.rs

@@ -20,6 +20,7 @@ use crate::backends::vault_credentials::{
 };
 use crate::backends::{
     BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits, ensure_key_status_permits,
+    ensure_tag_keys_are_mutable,
 };
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -1020,6 +1021,67 @@ impl VaultKmsClient {
         Ok(())
     }
 
+    /// Replace the key's description; `None` clears it.
+    ///
+    /// The write goes through the check-and-set read-modify-write loop, so a
+    /// rotation or state transition landing in between is carried over instead
+    /// of clobbered. A description that already matches is not rewritten.
+    pub(crate) async fn update_key_description(&self, key_id: &str, description: Option<&str>) -> Result<()> {
+        self.update_key_data_with_cas(key_id, |key_data| {
+            if key_data.description.as_deref() == description {
+                return Ok(CasMutation::Skip(()));
+            }
+            key_data.description = description.map(str::to_string);
+            Ok(CasMutation::Write(()))
+        })
+        .await?;
+
+        debug!(key_id, "Vault KMS key description updated");
+        Ok(())
+    }
+
+    /// Add or overwrite tags, leaving every other tag untouched.
+    pub(crate) async fn tag_key(&self, key_id: &str, tags: &HashMap<String, String>) -> Result<()> {
+        ensure_tag_keys_are_mutable(tags.keys().map(String::as_str))?;
+
+        self.update_key_data_with_cas(key_id, |key_data| {
+            let mut changed = false;
+            for (tag_key, value) in tags {
+                changed |= key_data.tags.insert(tag_key.clone(), value.clone()).as_ref() != Some(value);
+            }
+            Ok(if changed {
+                CasMutation::Write(())
+            } else {
+                CasMutation::Skip(())
+            })
+        })
+        .await?;
+
+        debug!(key_id, "Vault KMS key tags updated");
+        Ok(())
+    }
+
+    /// Remove tags; tags that are not set are ignored.
+    pub(crate) async fn untag_key(&self, key_id: &str, tag_keys: &[String]) -> Result<()> {
+        ensure_tag_keys_are_mutable(tag_keys.iter().map(String::as_str))?;
+
+        self.update_key_data_with_cas(key_id, |key_data| {
+            let mut changed = false;
+            for tag_key in tag_keys {
+                changed |= key_data.tags.remove(tag_key).is_some();
+            }
+            Ok(if changed {
+                CasMutation::Write(())
+            } else {
+                CasMutation::Skip(())
+            })
+        })
+        .await?;
+
+        debug!(key_id, "Vault KMS key tags removed");
+        Ok(())
+    }
+
     /// Rotate the master key while keeping every historical version decryptable.
     ///
     /// Commit protocol (all writes check-and-set, in this order):
@@ -1447,6 +1509,18 @@ impl KmsBackend for VaultKmsBackend {
         self.client.rotate_key(key_id, None).await.map(|_| ())
     }
 
+    async fn update_key_description(&self, key_id: &str, description: Option<&str>) -> Result<()> {
+        self.client.update_key_description(key_id, description).await
+    }
+
+    async fn tag_key(&self, key_id: &str, tags: &HashMap<String, String>) -> Result<()> {
+        self.client.tag_key(key_id, tags).await
+    }
+
+    async fn untag_key(&self, key_id: &str, tag_keys: &[String]) -> Result<()> {
+        self.client.untag_key(key_id, tag_keys).await
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
@@ -1462,6 +1536,7 @@ impl KmsBackend for VaultKmsBackend {
             .with_schedule_deletion(true)
             .with_versioning(true)
             .with_physical_delete(true)
+            .with_update_key_metadata(true)
     }
 
     async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {
@@ -2845,6 +2920,63 @@ mod tests {
         );
     }
 
+    /// Tag updates are check-and-set read-modify-writes over the live record,
+    /// never blind overwrites: they preserve the material and the tags they did
+    /// not address.
+    #[tokio::test]
+    async fn wired_tag_key_writeback_is_check_and_set() {
+        let mut key_data = healthy_key_data();
+        key_data.tags = HashMap::from([("name".to_string(), "wired-key".to_string())]);
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(kv2_metadata_read_data(1)),
+            ScriptedResponse::ok(kv2_read_data(&key_data)),
+            ScriptedResponse::ok(kv2_write_ack()),
+        ])
+        .await;
+
+        client
+            .tag_key("wired-key", &HashMap::from([("team".to_string(), "storage".to_string())]))
+            .await
+            .expect("tagging must succeed");
+
+        let bodies = vault.request_bodies();
+        let writeback = parse_write_body(&bodies[2]);
+        assert_eq!(writeback["options"]["cas"], serde_json::json!(1), "{writeback}");
+        assert_eq!(writeback["data"]["tags"]["team"], serde_json::json!("storage"), "{writeback}");
+        assert_eq!(
+            writeback["data"]["tags"]["name"],
+            serde_json::json!("wired-key"),
+            "a tag update must not drop tags it did not address: {writeback}"
+        );
+        assert_eq!(
+            writeback["data"]["encrypted_key_material"],
+            serde_json::json!(healthy_key_data().encrypted_key_material),
+            "the write-back must preserve the material of the freshly read record: {writeback}"
+        );
+    }
+
+    /// Rejecting the identity tag happens before any Vault call, so a rejected
+    /// request cannot leave a partial write behind.
+    #[tokio::test]
+    async fn wired_identity_tag_update_is_rejected_before_any_vault_call() {
+        let (vault, client) = scripted_client(Vec::new()).await;
+
+        for result in [
+            client
+                .tag_key("wired-key", &HashMap::from([("name".to_string(), "other".to_string())]))
+                .await,
+            client.untag_key("wired-key", &["name".to_string()]).await,
+        ] {
+            let error = result.expect_err("the identity tag must not be writable");
+            assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+        }
+        assert!(
+            vault.request_bodies().is_empty(),
+            "a rejected metadata update must not reach Vault: {:?}",
+            vault.request_bodies()
+        );
+    }
+
     /// A version record above the current pointer means the top-level record
     /// regressed (a lost update rolled back a committed rotation). Resolving
     /// material through such a record must fail closed instead of quietly

crates/kms/src/backends/vault_transit.rs

@@ -18,7 +18,10 @@ use crate::backends::vault_credentials::{
     CredentialTaskHandle, VaultClientHandle, VaultConnectionSettings, VaultCredentialPolicy, VaultCredentialProvider,
     token_source_for,
 };
-use crate::backends::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits};
+use crate::backends::{
+    BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits,
+    ensure_tag_keys_are_mutable,
+};
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
 use crate::error::{KmsError, Result};
@@ -908,6 +911,46 @@ impl VaultTransitKmsClient {
         .map(|_| ())
     }
 
+    /// Replace the key's description; `None` clears it.
+    ///
+    /// Metadata edits carry no state gate: they neither use nor invalidate key
+    /// material, so they stay available for whatever lifecycle state the key
+    /// is in.
+    pub(crate) async fn update_key_description(&self, key_id: &str, description: Option<&str>) -> Result<()> {
+        self.mutate_key_metadata(key_id, |metadata| {
+            metadata.description = description.map(str::to_string);
+            Ok(())
+        })
+        .await
+        .map(|_| ())
+    }
+
+    /// Add or overwrite tags, leaving every other tag untouched.
+    pub(crate) async fn tag_key(&self, key_id: &str, tags: &HashMap<String, String>) -> Result<()> {
+        ensure_tag_keys_are_mutable(tags.keys().map(String::as_str))?;
+        self.mutate_key_metadata(key_id, |metadata| {
+            metadata
+                .tags
+                .extend(tags.iter().map(|(key, value)| (key.clone(), value.clone())));
+            Ok(())
+        })
+        .await
+        .map(|_| ())
+    }
+
+    /// Remove tags; tags that are not set are ignored.
+    pub(crate) async fn untag_key(&self, key_id: &str, tag_keys: &[String]) -> Result<()> {
+        ensure_tag_keys_are_mutable(tag_keys.iter().map(String::as_str))?;
+        self.mutate_key_metadata(key_id, |metadata| {
+            for tag_key in tag_keys {
+                metadata.tags.remove(tag_key);
+            }
+            Ok(())
+        })
+        .await
+        .map(|_| ())
+    }
+
     /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
     #[cfg(test)]
     pub(crate) async fn schedule_key_deletion(
@@ -1237,6 +1280,18 @@ impl KmsBackend for VaultTransitKmsBackend {
         self.client.rotate_key(key_id, None).await.map(|_| ())
     }
 
+    async fn update_key_description(&self, key_id: &str, description: Option<&str>) -> Result<()> {
+        self.client.update_key_description(key_id, description).await
+    }
+
+    async fn tag_key(&self, key_id: &str, tags: &HashMap<String, String>) -> Result<()> {
+        self.client.tag_key(key_id, tags).await
+    }
+
+    async fn untag_key(&self, key_id: &str, tag_keys: &[String]) -> Result<()> {
+        self.client.untag_key(key_id, tag_keys).await
+    }
+
     async fn health_check(&self) -> Result<bool> {
         self.client.health_check().await.map(|_| true)
     }
@@ -1251,6 +1306,7 @@ impl KmsBackend for VaultTransitKmsBackend {
             .with_schedule_deletion(true)
             .with_versioning(true)
             .with_physical_delete(true)
+            .with_update_key_metadata(true)
     }
 
     async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {

crates/kms/src/manager.rs

@@ -24,6 +24,7 @@ use crate::types::{
     DeleteKeyRequest, DeleteKeyResponse, DescribeKeyRequest, DescribeKeyResponse, EncryptRequest, EncryptResponse,
     GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse, OperationContext,
 };
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::sync::RwLock;
@@ -348,6 +349,27 @@ impl KmsManager {
         result
     }
 
+    /// Replace a key's description; `None` clears it
+    pub async fn update_key_description(&self, key_id: &str, description: Option<&str>) -> Result<()> {
+        self.backend.update_key_description(key_id, description).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
+    }
+
+    /// Add or overwrite key tags, leaving every other tag untouched
+    pub async fn tag_key(&self, key_id: &str, tags: &HashMap<String, String>) -> Result<()> {
+        self.backend.tag_key(key_id, tags).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
+    }
+
+    /// Remove key tags; tags that are not set are ignored
+    pub async fn untag_key(&self, key_id: &str, tag_keys: &[String]) -> Result<()> {
+        self.backend.untag_key(key_id, tag_keys).await?;
+        self.invalidate_cached_metadata(key_id).await;
+        Ok(())
+    }
+
     /// Drop cached metadata after a state mutation so the next describe
     /// observes backend truth instead of the pre-mutation snapshot.
     async fn invalidate_cached_metadata(&self, key_id: &str) {

crates/kms/src/service.rs

@@ -14,6 +14,9 @@
 
 //! Object encryption service for S3-compatible encryption
 
+use crate::api_types::{
+    TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse, UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
+};
 use crate::cache::KmsCacheStats;
 use crate::encryption::ciphers::{create_cipher, generate_iv};
 use crate::error::{KmsError, Result};
@@ -184,6 +187,64 @@ impl ObjectEncryptionService {
         self.kms_manager.list_keys_with_context(request, context).await
     }
 
+    /// Replace a master key's description (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - UpdateKeyDescriptionRequest with key ID and the new
+    ///   description; an empty description clears the stored value
+    ///
+    /// # Returns
+    /// UpdateKeyDescriptionResponse acknowledging the update
+    ///
+    pub async fn update_key_description(&self, request: UpdateKeyDescriptionRequest) -> Result<UpdateKeyDescriptionResponse> {
+        let description = (!request.description.is_empty()).then_some(request.description.as_str());
+        self.kms_manager.update_key_description(&request.key_id, description).await?;
+
+        Ok(UpdateKeyDescriptionResponse {
+            success: true,
+            message: "key description updated".to_string(),
+            key_id: request.key_id,
+        })
+    }
+
+    /// Add or overwrite master key tags (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - TagKeyRequest with key ID and the tags to set; tags
+    ///   outside the request are left untouched
+    ///
+    /// # Returns
+    /// TagKeyResponse acknowledging the update
+    ///
+    pub async fn tag_key(&self, request: TagKeyRequest) -> Result<TagKeyResponse> {
+        self.kms_manager.tag_key(&request.key_id, &request.tags).await?;
+
+        Ok(TagKeyResponse {
+            success: true,
+            message: "key tags updated".to_string(),
+            key_id: request.key_id,
+        })
+    }
+
+    /// Remove master key tags (delegates to KMS manager)
+    ///
+    /// # Arguments
+    /// * `request` - UntagKeyRequest with key ID and the tag keys to remove;
+    ///   tag keys that are not set are ignored, so the call is idempotent
+    ///
+    /// # Returns
+    /// UntagKeyResponse acknowledging the removal
+    ///
+    pub async fn untag_key(&self, request: UntagKeyRequest) -> Result<UntagKeyResponse> {
+        self.kms_manager.untag_key(&request.key_id, &request.tag_keys).await?;
+
+        Ok(UntagKeyResponse {
+            success: true,
+            message: "key tags removed".to_string(),
+            key_id: request.key_id,
+        })
+    }
+
     /// Generate a data encryption key (delegates to KMS manager)
     ///
     /// # Arguments
@@ -966,6 +1027,128 @@ mod tests {
         );
     }
 
+    async fn describe(service: &ObjectEncryptionService, key_id: &str) -> KeyMetadata {
+        service
+            .describe_key(DescribeKeyRequest {
+                key_id: key_id.to_string(),
+            })
+            .await
+            .expect("describe should succeed")
+            .key_metadata
+    }
+
+    async fn create_metadata_test_key(service: &ObjectEncryptionService, key_id: &str) {
+        service
+            .create_key(CreateKeyRequest {
+                key_name: Some(key_id.to_string()),
+                key_usage: KeyUsage::EncryptDecrypt,
+                description: Some("original".to_string()),
+                policy: None,
+                tags: HashMap::from([
+                    ("name".to_string(), key_id.to_string()),
+                    ("team".to_string(), "storage".to_string()),
+                ]),
+                origin: None,
+            })
+            .await
+            .expect("test key should be created");
+    }
+
+    #[tokio::test]
+    async fn key_metadata_updates_are_visible_to_describe() {
+        let (service, _temp_dir) = create_test_service().await;
+        create_metadata_test_key(&service, "metadata-key").await;
+
+        service
+            .update_key_description(UpdateKeyDescriptionRequest {
+                key_id: "metadata-key".to_string(),
+                description: "updated".to_string(),
+            })
+            .await
+            .expect("description update should succeed");
+        service
+            .tag_key(TagKeyRequest {
+                key_id: "metadata-key".to_string(),
+                tags: HashMap::from([
+                    ("team".to_string(), "platform".to_string()),
+                    ("env".to_string(), "prod".to_string()),
+                ]),
+            })
+            .await
+            .expect("tagging should succeed");
+
+        // Reading through the manager's metadata cache must observe the
+        // updates, not the snapshot cached when the key was created.
+        let metadata = describe(&service, "metadata-key").await;
+        assert_eq!(metadata.description.as_deref(), Some("updated"));
+        assert_eq!(metadata.tags.get("team").map(String::as_str), Some("platform"));
+        assert_eq!(metadata.tags.get("env").map(String::as_str), Some("prod"));
+        assert_eq!(
+            metadata.tags.get("name").map(String::as_str),
+            Some("metadata-key"),
+            "tags set at creation must survive a later tag update"
+        );
+
+        // Untagging is idempotent: removing an absent tag is a no-op, not an
+        // error, so a retried request stays safe.
+        for attempt in 1..=2 {
+            service
+                .untag_key(UntagKeyRequest {
+                    key_id: "metadata-key".to_string(),
+                    tag_keys: vec!["env".to_string()],
+                })
+                .await
+                .unwrap_or_else(|error| panic!("untag attempt {attempt} should succeed: {error:?}"));
+        }
+        let metadata = describe(&service, "metadata-key").await;
+        assert!(!metadata.tags.contains_key("env"), "untagged tag must be gone: {:?}", metadata.tags);
+        assert_eq!(
+            metadata.tags.get("team").map(String::as_str),
+            Some("platform"),
+            "untagging must not touch other tags"
+        );
+
+        // An empty description clears the stored value.
+        service
+            .update_key_description(UpdateKeyDescriptionRequest {
+                key_id: "metadata-key".to_string(),
+                description: String::new(),
+            })
+            .await
+            .expect("clearing the description should succeed");
+        assert_eq!(describe(&service, "metadata-key").await.description, None);
+    }
+
+    #[tokio::test]
+    async fn identity_tag_is_not_writable_through_metadata_updates() {
+        let (service, _temp_dir) = create_test_service().await;
+        create_metadata_test_key(&service, "identity-key").await;
+
+        let rewrite = service
+            .tag_key(TagKeyRequest {
+                key_id: "identity-key".to_string(),
+                tags: HashMap::from([("name".to_string(), "other-key".to_string())]),
+            })
+            .await
+            .expect_err("rewriting the identity tag must be rejected");
+        assert!(matches!(rewrite, KmsError::InvalidOperation { .. }), "got {rewrite:?}");
+
+        let removal = service
+            .untag_key(UntagKeyRequest {
+                key_id: "identity-key".to_string(),
+                tag_keys: vec!["team".to_string(), "name".to_string()],
+            })
+            .await
+            .expect_err("removing the identity tag must be rejected");
+        assert!(matches!(removal, KmsError::InvalidOperation { .. }), "got {removal:?}");
+
+        // Both rejections are pre-write: the record is untouched, including
+        // the ordinary tag that shared the rejected untag request.
+        let metadata = describe(&service, "identity-key").await;
+        assert_eq!(metadata.tags.get("name").map(String::as_str), Some("identity-key"));
+        assert_eq!(metadata.tags.get("team").map(String::as_str), Some("storage"));
+    }
+
     #[tokio::test]
     async fn test_decrypt_data_key_uses_object_encryption_context() {
         let (service, _temp_dir) = create_test_service().await;