@@ -87,7 +87,10 @@ public class CreateHSMProfileCmd extends BaseCmd {
     private String vendorName;
 
     @Parameter(name = ApiConstants.DETAILS, type = CommandType.MAP,
-               description = "HSM configuration details (protocol specific)")
+               description = "HSM configuration details (protocol specific). For pkcs11, the wrap_algorithm detail "
+                       + "selects how volume encryption keys get wrapped. Allowed values are AES/GCM/NoPadding "
+                       + "(default), AES/KWP/NoPadding (needs Java 18 or later) and AES/CBC/PKCS5Padding. "
+                       + "It can't be changed later.")
     private Map<String, String> details;
 
     public String getName() {
 
@@ -72,6 +72,12 @@ public VolumeInfo getVolume() {
      */
     AsyncCallFuture<VolumeApiResult> expungeVolumeAsync(VolumeInfo volume);
 
+    /**
+     * Releases the encryption key (KMS wrapped key or legacy passphrase) of removed or expunged volumes once nothing
+     * needs it anymore, e.g. after the last snapshot of such a volume was deleted.
+     */
+    void releaseKeysOfRemovedVolumes();
+
     void ensureVolumeIsExpungeReady(long volumeId);
 
     boolean cloneVolume(long volumeId, long baseVolId);
 
@@ -191,4 +191,27 @@ public interface VolumeDao extends GenericDao<VolumeVO, Long>, StateDao<Volume.S
     VolumeVO findByExternalUuid(String externalUuid);
 
     List<VolumeVO> findByKmsWrappedKeyId(Long kmsWrappedKeyId);
+
+    /**
+     * All volumes that reference the given KMS wrapped key, including removed ones. Snapshots of a removed volume
+     * still need the key of that volume.
+     */
+    List<VolumeVO> listIncludingRemovedByKmsWrappedKeyId(long kmsWrappedKeyId);
+
+    /**
+     * All volumes that reference the given passphrase, including removed ones.
+     */
+    List<VolumeVO> listIncludingRemovedByPassphraseId(long passphraseId);
+
+    /**
+     * Removed or expunged volumes whose KMS wrapped key has not been removed yet. Their key may no longer be needed
+     * once their snapshots are gone; used by the storage garbage collector to release it.
+     */
+    List<VolumeVO> listRemovedOrExpungedWithKmsWrappedKey();
+
+    /**
+     * Removed or expunged volumes that still reference a passphrase. Used by the storage garbage collector to release
+     * the passphrase once it is no longer needed.
+     */
+    List<VolumeVO> listRemovedOrExpungedWithPassphrase();
 }
@@ -23,13 +23,16 @@
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import javax.inject.Inject;
+import javax.naming.ConfigurationException;
 
+import org.apache.cloudstack.kms.KMSWrappedKeyVO;
+import org.apache.cloudstack.kms.dao.KMSWrappedKeyDao;
 import org.apache.cloudstack.reservation.ReservationVO;
 import org.apache.cloudstack.reservation.dao.ReservationDao;
-import org.apache.cloudstack.kms.dao.KMSWrappedKeyDao;
 import org.apache.commons.collections.CollectionUtils;
 import org.springframework.stereotype.Component;
 
@@ -50,6 +53,7 @@
 import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.GenericSearchBuilder;
+import com.cloud.utils.db.JoinBuilder;
 import com.cloud.utils.db.QueryBuilder;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
@@ -82,6 +86,8 @@ public class VolumeDaoImpl extends GenericDaoBase<VolumeVO, Long> implements Vol
     private final SearchBuilder<VolumeVO> poolAndPathSearch;
     final GenericSearchBuilder<VolumeVO, Integer> CountByOfferingId;
     private final SearchBuilder<VolumeVO> kmsMigrationSearch;
+    private SearchBuilder<VolumeVO> removedWithPassphraseSearch;
+    private SearchBuilder<VolumeVO> removedWithLiveKmsWrappedKeySearch;
 
     @Inject
     ReservationDao reservationDao;
@@ -543,6 +549,32 @@ public VolumeDaoImpl() {
         kmsMigrationSearch.done();
     }
 
+    @Override
+    public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
+        boolean result = super.configure(name, params);
+
+        removedWithPassphraseSearch = createSearchBuilder();
+        removedWithPassphraseSearch.and("passphraseId", removedWithPassphraseSearch.entity().getPassphraseId(), Op.NNULL);
+        removedWithPassphraseSearch.and().op("removed", removedWithPassphraseSearch.entity().getRemoved(), Op.NNULL);
+        removedWithPassphraseSearch.or("state", removedWithPassphraseSearch.entity().getState(), Op.EQ);
+        removedWithPassphraseSearch.cp();
+        removedWithPassphraseSearch.done();
+
+        // Joins with the table of another DAO, so it is built here rather than in the constructor.
+        SearchBuilder<KMSWrappedKeyVO> liveWrappedKeySearch = kmsWrappedKeyDao.createSearchBuilder();
+        liveWrappedKeySearch.and("wrappedKeyRemoved", liveWrappedKeySearch.entity().getRemoved(), Op.NULL);
+
+        removedWithLiveKmsWrappedKeySearch = createSearchBuilder();
+        removedWithLiveKmsWrappedKeySearch.and().op("removed", removedWithLiveKmsWrappedKeySearch.entity().getRemoved(), Op.NNULL);
+        removedWithLiveKmsWrappedKeySearch.or("state", removedWithLiveKmsWrappedKeySearch.entity().getState(), Op.EQ);
+        removedWithLiveKmsWrappedKeySearch.cp();
+        removedWithLiveKmsWrappedKeySearch.join("wrappedKey", liveWrappedKeySearch, liveWrappedKeySearch.entity().getId(),
+                removedWithLiveKmsWrappedKeySearch.entity().getKmsWrappedKeyId(), JoinBuilder.JoinType.INNER);
+        removedWithLiveKmsWrappedKeySearch.done();
+
+        return result;
+    }
+
     @Override
     @DB()
     public Pair<Long, Long> getCountAndTotalByPool(long poolId) {
@@ -784,17 +816,6 @@ public boolean remove(Long id) {
         logger.debug(String.format("Removing volume %s from DB", id));
         VolumeVO entry = findById(id);
         if (entry != null) {
-            // Clean up KMS wrapped key if volume was encrypted with KMS
-            if (entry.getKmsWrappedKeyId() != null) {
-                try {
-                    kmsWrappedKeyDao.remove(entry.getKmsWrappedKeyId());
-                    logger.debug("Removed KMS wrapped key [id={}] for volume [id={}, uuid={}]",
-                            entry.getKmsWrappedKeyId(), id, entry.getUuid());
-                } catch (Exception e) {
-                    logger.warn("Failed to remove KMS wrapped key [id={}] for volume [id={}, uuid={}]: {}",
-                            entry.getKmsWrappedKeyId(), id, entry.getUuid(), e.getMessage(), e);
-                }
-            }
             tagsDao.removeByIdAndType(id, ResourceObjectType.Volume);
         }
         boolean result = super.remove(id);
@@ -1011,4 +1032,32 @@ public List<VolumeVO> findByKmsWrappedKeyId(Long kmsWrappedKeyId) {
         sc.setParameters("notDestroyed", Volume.State.Expunged);
         return listBy(sc);
     }
+
+    @Override
+    public List<VolumeVO> listIncludingRemovedByKmsWrappedKeyId(long kmsWrappedKeyId) {
+        SearchCriteria<VolumeVO> sc = AllFieldsSearch.create();
+        sc.setParameters("kmsWrappedKeyId", kmsWrappedKeyId);
+        return listIncludingRemovedBy(sc);
+    }
+
+    @Override
+    public List<VolumeVO> listIncludingRemovedByPassphraseId(long passphraseId) {
+        SearchCriteria<VolumeVO> sc = AllFieldsSearch.create();
+        sc.setParameters("passphraseId", passphraseId);
+        return listIncludingRemovedBy(sc);
+    }
+
+    @Override
+    public List<VolumeVO> listRemovedOrExpungedWithKmsWrappedKey() {
+        SearchCriteria<VolumeVO> sc = removedWithLiveKmsWrappedKeySearch.create();
+        sc.setParameters("state", Volume.State.Expunged);
+        return listIncludingRemovedBy(sc);
+    }
+
+    @Override
+    public List<VolumeVO> listRemovedOrExpungedWithPassphrase() {
+        SearchCriteria<VolumeVO> sc = removedWithPassphraseSearch.create();
+        sc.setParameters("state", Volume.State.Expunged);
+        return listIncludingRemovedBy(sc);
+    }
 }
@@ -16,8 +16,27 @@
 // under the License.
 package com.cloud.upgrade.dao;
 
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+import com.cloud.utils.exception.CloudRuntimeException;
+
 public class Upgrade42300to2400 extends DbUpgradeAbstractImpl implements DbUpgrade, DbUpgradeSystemVmTemplate {
 
+    // A removed KMS wrapped key that a volume (not expunged) or a snapshot (not destroyed) still needs
+    private static final String REMOVED_KMS_WRAPPED_KEY_IN_USE = "w.removed IS NOT NULL AND ("
+            + "EXISTS (SELECT 1 FROM `cloud`.`volumes` v WHERE v.kms_wrapped_key_id = w.id"
+            + " AND v.removed IS NULL AND v.state <> 'Expunged')"
+            + " OR EXISTS (SELECT 1 FROM `cloud`.`snapshots` s JOIN `cloud`.`volumes` v ON v.id = s.volume_id"
+            + " WHERE v.kms_wrapped_key_id = w.id AND s.removed IS NULL AND s.status <> 'Destroyed'))";
+    // The KMS key and the KEK the key was wrapped with still exist. Deleting a KMS key deletes its KEKs but leaves
+    // its version rows as they are.
+    private static final String KEK_VERSION_USABLE = "EXISTS (SELECT 1 FROM `cloud`.`kms_kek_versions` k"
+            + " JOIN `cloud`.`kms_keys` kk ON kk.id = k.kms_key_id"
+            + " WHERE k.id = w.kek_version_id AND k.removed IS NULL AND k.status <> 'Archived' AND kk.removed IS NULL)";
+
     @Override
     public String[] getUpgradableVersionRange() {
         return new String[]{"4.23.0.0", "24.0.0"};
@@ -27,4 +46,40 @@ public String[] getUpgradableVersionRange() {
     public String getUpgradedVersion() {
         return "24.0.0";
     }
+
+    @Override
+    public void performDataMigration(Connection conn) {
+        restoreKmsWrappedKeysStillInUse(conn);
+    }
+
+    /**
+     * In 4.23 removing a volume also removed its KMS wrapped key, even when the key was still needed by the volume
+     * it was migrated to or by snapshots of the volume. Restores such keys while the KEK they were wrapped with still
+     * exists, and logs the ones that can't be recovered anymore.
+     */
+    private void restoreKmsWrappedKeysStillInUse(Connection conn) {
+        String lostKeysQuery = "SELECT w.uuid, v.uuid FROM `cloud`.`kms_wrapped_key` w"
+                + " JOIN `cloud`.`volumes` v ON v.kms_wrapped_key_id = w.id"
+                + " WHERE " + REMOVED_KMS_WRAPPED_KEY_IN_USE + " AND NOT " + KEK_VERSION_USABLE;
+        try (PreparedStatement pstmt = conn.prepareStatement(lostKeysQuery); ResultSet rs = pstmt.executeQuery()) {
+            while (rs.next()) {
+                logger.warn("KMS wrapped key [{}] of volume [{}] was removed while still in use and the KEK it was "
+                        + "wrapped with is gone; the volume and its snapshots can't be decrypted anymore",
+                        rs.getString(1), rs.getString(2));
+            }
+        } catch (SQLException e) {
+            throw new CloudRuntimeException("Unable to look up KMS wrapped keys that were removed while still in use", e);
+        }
+
+        String restoreQuery = "UPDATE `cloud`.`kms_wrapped_key` w SET w.removed = NULL"
+                + " WHERE " + REMOVED_KMS_WRAPPED_KEY_IN_USE + " AND " + KEK_VERSION_USABLE;
+        try (PreparedStatement pstmt = conn.prepareStatement(restoreQuery)) {
+            int restored = pstmt.executeUpdate();
+            if (restored > 0) {
+                logger.info("Restored {} KMS wrapped key(s) that were removed while still in use", restored);
+            }
+        } catch (SQLException e) {
+            throw new CloudRuntimeException("Unable to restore KMS wrapped keys that were removed while still in use", e);
+        }
+    }
 }
@@ -61,6 +61,16 @@ public class KMSWrappedKeyVO {
     @Column(name = "wrapped_blob", nullable = false)
     private byte[] wrappedBlob;
 
+    @Column(name = "wrap_algorithm", nullable = false)
+    private String wrapAlgorithm;
+
+    /**
+     * Set on every row that existed when upgrading from 4.23. Such a blob may hold a volume key that was migrated
+     * from a legacy passphrase in the wrong format; it is checked once and the flag is cleared.
+     */
+    @Column(name = "format_check_pending", nullable = false)
+    private boolean formatCheckPending;
+
     @Column(name = GenericDao.CREATED_COLUMN, nullable = false)
     @Temporal(TemporalType.TIMESTAMP)
     private Date created;
@@ -152,6 +162,22 @@ public void setWrappedBlob(byte[] wrappedBlob) {
         this.wrappedBlob = wrappedBlob != null ? Arrays.copyOf(wrappedBlob, wrappedBlob.length) : null;
     }
 
+    public String getWrapAlgorithm() {
+        return wrapAlgorithm;
+    }
+
+    public void setWrapAlgorithm(String wrapAlgorithm) {
+        this.wrapAlgorithm = wrapAlgorithm;
+    }
+
+    public boolean isFormatCheckPending() {
+        return formatCheckPending;
+    }
+
+    public void setFormatCheckPending(boolean formatCheckPending) {
+        this.formatCheckPending = formatCheckPending;
+    }
+
     public Date getCreated() {
         return created;
     }
@@ -171,6 +197,6 @@ public void setRemoved(Date removed) {
     @Override
     public String toString() {
         return String.format("KMSWrappedKey %s", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(
-                this, "id", "uuid", "kmsKeyId", "kekVersionId", "zoneId", "created", "removed"));
+                this, "id", "uuid", "kmsKeyId", "kekVersionId", "zoneId", "wrapAlgorithm", "created", "removed"));
     }
 }
@@ -32,4 +32,10 @@ public interface KMSWrappedKeyDao extends GenericDao<KMSWrappedKeyVO, Long> {
     List<KMSWrappedKeyVO> listByKekVersionId(Long kekVersionId, int limit);
 
     long countByKekVersionId(Long kekVersionId);
+
+    /**
+     * Rows still flagged for the one-time format check after the upgrade from 4.23
+     * (see {@link KMSWrappedKeyVO#isFormatCheckPending()}), with an id greater than {@code afterId}, by id.
+     */
+    List<KMSWrappedKeyVO> listFormatCheckPending(long afterId, int limit);
 }
@@ -38,6 +38,8 @@ public KMSWrappedKeyDaoImpl() {
         allFieldSearch.and("kmsKeyId", allFieldSearch.entity().getKmsKeyId(), SearchCriteria.Op.EQ);
         allFieldSearch.and("kekVersionId", allFieldSearch.entity().getKekVersionId(), SearchCriteria.Op.EQ);
         allFieldSearch.and("zoneId", allFieldSearch.entity().getZoneId(), SearchCriteria.Op.EQ);
+        allFieldSearch.and("formatCheckPending", allFieldSearch.entity().isFormatCheckPending(), SearchCriteria.Op.EQ);
+        allFieldSearch.and("idAfter", allFieldSearch.entity().getId(), SearchCriteria.Op.GT);
         allFieldSearch.done();
     }
 
@@ -67,4 +69,12 @@ public long countByKekVersionId(Long kekVersionId) {
         Integer count = getCount(sc);
         return count != null ? count.longValue() : 0L;
     }
+
+    @Override
+    public List<KMSWrappedKeyVO> listFormatCheckPending(long afterId, int limit) {
+        SearchCriteria<KMSWrappedKeyVO> sc = allFieldSearch.create();
+        sc.setParameters("formatCheckPending", true);
+        sc.setParameters("idAfter", afterId);
+        return listBy(sc, new Filter(KMSWrappedKeyVO.class, "id", true, 0L, (long) limit));
+    }
 }
@@ -226,3 +226,36 @@ CREATE TABLE IF NOT EXISTS `cloud`.`instance_boot_group_details` (
     PRIMARY KEY (`id`),
     CONSTRAINT `fk_instance_boot_group_details__group_id` FOREIGN KEY (`boot_group_id`) REFERENCES `instance_boot_group` (`id`) ON DELETE CASCADE
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- KMS: record the algorithm each key was wrapped with. Keys of the PKCS#11 provider were wrapped with
+-- AES/CBC/PKCS5Padding (the column default), keys of the database provider with AES/GCM/NoPadding.
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.kms_wrapped_key', 'wrap_algorithm', 'VARCHAR(64) NOT NULL DEFAULT ''AES/CBC/PKCS5Padding'' COMMENT ''Algorithm the key was wrapped with''');
+UPDATE `cloud`.`kms_wrapped_key` w
+    JOIN `cloud`.`kms_kek_versions` v ON v.id = w.kek_version_id
+    JOIN `cloud`.`kms_hsm_profiles` p ON p.id = v.hsm_profile_id
+SET w.wrap_algorithm = 'AES/GCM/NoPadding'
+WHERE LOWER(p.protocol) = 'database';
+
+-- KMS: 4.23 migrated legacy volume passphrases to KMS in a wrong format. Flag every existing key for a one-time
+-- check by the management server, which repairs the affected ones.
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.kms_wrapped_key', 'format_check_pending', 'TINYINT(1) NOT NULL DEFAULT 0 COMMENT ''1 if the key still has to be checked for the format 4.23 migrated legacy passphrases in''');
+UPDATE `cloud`.`kms_wrapped_key` SET format_check_pending = 1;
+
+-- KMS: existing PKCS#11 HSM profiles keep wrapping keys with AES/CBC/PKCS5Padding. New profiles default to
+-- AES/GCM/NoPadding.
+INSERT INTO `cloud`.`kms_hsm_profile_details` (profile_id, name, value)
+    SELECT p.id, 'wrap_algorithm', 'AES/CBC/PKCS5Padding' FROM `cloud`.`kms_hsm_profiles` p
+    WHERE LOWER(p.protocol) = 'pkcs11'
+      AND NOT EXISTS (SELECT 1 FROM `cloud`.`kms_hsm_profile_details` d WHERE d.profile_id = p.id AND d.name = 'wrap_algorithm');
+UPDATE `cloud`.`kms_keys` k
+    JOIN `cloud`.`kms_hsm_profiles` p ON p.id = k.hsm_profile_id
+SET k.algorithm = 'AES/CBC/PKCS5Padding'
+WHERE LOWER(p.protocol) = 'pkcs11';
+
+-- KMS: the database provider doesn't use an HSM
+UPDATE `cloud`.`kms_hsm_profiles` SET name = 'Database Provider' WHERE protocol = 'database' AND name = 'HSM Database Provider';
+
+-- KMS: remove legacy passphrases no volume refers to anymore, e.g. those left behind when 4.23 migrated a volume to KMS
+DELETE p FROM `cloud`.`passphrase` p
+    LEFT JOIN `cloud`.`volumes` v ON v.passphrase_id = p.id
+WHERE v.id IS NULL;