CLDSRV-910: Rewrite the data of a self-copy when its encryption changes

lib/api/objectCopy.js

@@ -11,6 +11,7 @@ const locationConstraintCheck = require('./apiUtils/object/locationConstraintChe
 const { checkQueryVersionId, versioningPreprocessing, decodeVID } = require('./apiUtils/object/versioning');
 const getReplicationInfo = require('./apiUtils/object/getReplicationInfo');
 const { data } = require('../data/wrapper');
+const kms = require('../kms/wrapper');
 const services = require('../services');
 const { pushMetric } = require('../utapi/utilities');
 const removeAWSChunked = require('./apiUtils/object/removeAWSChunked');
@@ -32,6 +33,52 @@ const versioningNotImplBackends = constants.versioningNotImplBackends;
 const externalVersioningErrorMessage =
     'We do not currently support putting ' + 'a versioned object to a location-constraint of type AWS or Azure or GCP.';
 
+/**
+ * Compute the prior data locations that are orphaned.
+ *
+ * @param {Array} dataToDelete - prior data locations from versioningPreprocessing
+ * @param {Array} newDataGetInfo - new data locations stored in the new metadata
+ * @returns {Array|null} orphaned entries to batchDelete, or null if none
+ */
+function _orphanedDataLocations(dataToDelete, newDataGetInfo) {
+    // Identity is (dataStoreName, key): the same key under a different
+    // dataStoreName points to a different backend slot, so the old slot
+    // is an orphan even when the key string collides. dataStoreName is a
+    // location-constraint name (never contains '|'), so the delimiter is
+    // unambiguous.
+
+    // We need to normalize to support old legacy location names.
+    const normalize = loc => (typeof loc === 'string' ? { key: loc } : loc);
+    const locId = loc => {
+        const n = normalize(loc);
+        return `${n.dataStoreName}|${n.key}`;
+    };
+    const newIds = new Set((newDataGetInfo || []).map(locId));
+    const orphans = (dataToDelete || []).filter(loc => !newIds.has(locId(loc)));
+    return orphans.length > 0 ? orphans : null;
+}
+
+/**
+ * Tell whether the data must be encrypted again under the destination key:
+ * the destination is encrypted, and the source data is not encrypted or is
+ * encrypted under another master key. Encrypted data without a master key ID
+ * counts as encrypted under another key.
+ *
+ * @param {Array} dataLocator - source data locations, with the source master key
+ * @param {object|null} serverSideEncryption - SSE config for the destination
+ * @param {object} log - request logger
+ * @returns {boolean} true if the data must be rewritten
+ */
+function _mustReencryptData(dataLocator, serverSideEncryption, log) {
+    if (!serverSideEncryption || !serverSideEncryption.algorithm) {
+        return false;
+    }
+    const masterKeyId = serverSideEncryption.configuredMasterKeyId || serverSideEncryption.masterKeyId;
+    return dataLocator.some(
+        location => !location.cipheredDataKey || !kms.isSameMasterKey(location.masterKeyId, masterKeyId, log),
+    );
+}
+
 /**
  * Preps metadata to be saved (based on copy or replace request header)
  * @param {object} request - request
@@ -495,12 +542,16 @@ function objectCopy(authInfo, request, sourceBucket, sourceObject, sourceVersion
                 const vcfg = destBucketMD.getVersioningConfiguration();
                 const isVersionedObj = vcfg && vcfg.Status === 'Enabled';
                 const destLocationConstraintName = storeMetadataParams.dataStoreName;
-                const needsEncryption = serverSideEncryption && !!serverSideEncryption.algo;
-                // skip if source and dest and location constraint the same and
-                // versioning is not enabled
+                // skip if source and dest and location constraint the same,
+                // versioning is not enabled and the data keeps its encryption
                 // still send along serverSideEncryption info so algo
                 // and masterKeyId stored properly in metadata
-                if (sourceIsDestination && storeMetadataParams.locationMatch && !isVersionedObj && !needsEncryption) {
+                if (
+                    sourceIsDestination &&
+                    storeMetadataParams.locationMatch &&
+                    !isVersionedObj &&
+                    !_mustReencryptData(dataLocator, serverSideEncryption, log)
+                ) {
                     return next(null, storeMetadataParams, dataLocator, destObjMD, serverSideEncryption, destBucketMD);
                 }
 
@@ -633,7 +684,7 @@ function objectCopy(authInfo, request, sourceBucket, sourceObject, sourceVersion
                         if (options.extraMD) {
                             Object.assign(storeMetadataParams, options.extraMD);
                         }
-                        const dataToDelete = options.dataToDelete;
+                        const dataToDelete = _orphanedDataLocations(options.dataToDelete, destDataGetInfoArr);
                         return next(
                             null,
                             storeMetadataParams,
@@ -706,11 +757,7 @@ function objectCopy(authInfo, request, sourceBucket, sourceObject, sourceVersion
                 destObjPrevSize,
                 next,
             ) {
-                // Clean up any potential orphans in data if object
-                // put is an overwrite of already existing
-                // object with same name, so long as the source is not
-                // the same as the destination
-                if (!sourceIsDestination && dataToDelete) {
+                if (dataToDelete) {
                     const newDataStoreName = storeMetadataParams.dataStoreName;
                     return data.batchDelete(dataToDelete, request.method, newDataStoreName, log, err => {
                         if (err) {
@@ -831,3 +878,6 @@ function objectCopy(authInfo, request, sourceBucket, sourceObject, sourceVersion
 }
 
 module.exports = objectCopy;
+// Exposed for unit testing only.
+module.exports._orphanedDataLocations = _orphanedDataLocations;
+module.exports._mustReencryptData = _mustReencryptData;