CLDSRV-908: CopyObject handle checksums

lib/api/apiUtils/integrity/validateChecksums.js

@@ -39,6 +39,12 @@ const errMPUTypeWithoutAlgo = errorInstances.InvalidRequest.customizeDescription
     'The x-amz-checksum-type header can only be used ' + 'with the x-amz-checksum-algorithm header.',
 );
 
+// TODO: Update with 'MD5', 'SHA512', 'XXHASH128', 'XXHASH3', 'XXHASH64' when they are introduced.
+const errCopyChecksumAlgoNotSupported = errorInstances.InvalidRequest.customizeDescription(
+    'Checksum algorithm provided is unsupported. Please try again with any of the valid types: ' +
+        '[CRC32, CRC32C, CRC64NVME, SHA1, SHA256]',
+);
+
 const checksumedMethods = Object.freeze({
     completeMultipartUpload: true,
     multiObjectDelete: true,
@@ -81,6 +87,7 @@ const ChecksumError = Object.freeze({
     MPUTypeInvalid: 'MPUTypeInvalid',
     MPUTypeWithoutAlgo: 'MPUTypeWithoutAlgo',
     MPUInvalidCombination: 'MPUInvalidCombination',
+    CopyChecksumAlgoNotSupported: 'CopyChecksumAlgoNotSupported',
 });
 
 const base64Regex = /^[A-Za-z0-9+/]*={0,2}$/;
@@ -401,6 +408,8 @@ function arsenalErrorFromChecksumError(err) {
                 `The ${err.details.type} checksum type cannot be used ` +
                     `with the ${err.details.algorithm.toUpperCase()} checksum algorithm.`,
             );
+        case ChecksumError.CopyChecksumAlgoNotSupported:
+            return errCopyChecksumAlgoNotSupported;
         default:
             return ArsenalErrors.BadDigest;
     }
@@ -585,6 +594,31 @@ function computeFullObjectMPUChecksum(algorithm, parts) {
     return { checksum: combinePartCrcs(parts, params.polyReversed, params.dim), error: null };
 }
 
+/**
+ * Read and validate the x-amz-checksum-algorithm header for CopyObject.
+ *
+ * @param {object} headers - request headers
+ * @returns {{ error: null, algorithm: string|null }
+ *   | { error: { error: string, details: { algorithm: string } }, algorithm: null }}
+ */
+function getCopyObjectChecksumAlgorithm(headers) {
+    const requested = headers['x-amz-checksum-algorithm'];
+    if (requested === undefined) {
+        return { error: null, algorithm: null };
+    }
+    const algorithm = requested.toLowerCase();
+    if (!algorithms[algorithm]) {
+        return {
+            error: {
+                error: ChecksumError.CopyChecksumAlgoNotSupported,
+                details: { algorithm: requested },
+            },
+            algorithm: null,
+        };
+    }
+    return { error: null, algorithm };
+}
+
 module.exports = {
     ChecksumError,
     defaultChecksumData,
@@ -597,4 +631,5 @@ module.exports = {
     getChecksumDataFromMPUHeaders,
     computeCompositeMPUChecksum,
     computeFullObjectMPUChecksum,
+    getCopyObjectChecksumAlgorithm,
 };

lib/api/objectCopy.js

@@ -1,6 +1,7 @@
 const async = require('async');
+const { PassThrough } = require('stream');
 
-const { errors, errorInstances, versioning, s3middleware, s3routes } = require('arsenal');
+const { errors, errorInstances, jsutil, versioning, s3middleware, s3routes } = require('arsenal');
 const { validateObjectKeyLength } = s3routes.routesUtils;
 const getMetaHeaders = s3middleware.userMetadata.getMetaHeaders;
 const validateHeaders = s3middleware.validateConditionalHeaders;
@@ -28,13 +29,82 @@
 const { setSSEHeaders } = require('./apiUtils/object/sseHeaders');
 const { updateEncryption } = require('./apiUtils/bucket/updateEncryption');
 const { initializeInternalLogRequestQueue, queueInternalLogRequest } = require('../utilities/serverAccessLogger');
+const { algorithms, arsenalErrorFromChecksumError, getCopyObjectChecksumAlgorithm } =
+    require('./apiUtils/integrity/validateChecksums');
+const ChecksumTransform = require('../auth/streamingV4/ChecksumTransform');
+const kms = require('../kms/wrapper');
 
 const versionIdUtils = versioning.VersionID;
 const locationHeader = constants.objectLocationConstraintHeader;
 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.';
 
+/**
+ * Concatenate the source object's parts into a single Readable stream by
+ * reading them sequentially through `data.get`. Returns the PassThrough
+ * immediately; consumers should pipe it to the next stage and observe
+ * `error` on this stream for any read failure along the way.
+ *
+ * @param {Array} dataLocator - ordered source parts
+ * @param {object} log - request logger
+ * @return {PassThrough}
+ */
+function _pipeSourcePartsThrough(dataLocator, log) {
+    const passthrough = new PassThrough();
+    async.eachSeries(dataLocator, (part, cb) => {
+        const done = jsutil.once(cb);
+        if (part.dataStoreType === 'azure') {
+            // Azure's data.get writes part bytes into the provided writable
+            // instead of returning a Readable. Pipe a per-part PassThrough
+            // into the master passthrough and use its 'end' as the completion
+            // signal — same pattern arsenal's data.copyObject uses.
+            const perPart = new PassThrough();
+            perPart.once('error', done);
+            perPart.once('end', () => done());
+            perPart.pipe(passthrough, { end: false });
+            return data.get(part, perPart, log, err => {
+                if (err) {
+                    done(err);
+                }
+            });
+        }
+        return data.get(part, null, log, (err, partStream) => {
+            if (err) {
+                return done(err);
+            }
+            partStream.once('error', done);
+            partStream.once('end', () => done());
+            partStream.pipe(passthrough, { end: false });
+            return undefined;
+        });
+    }, err => {
+        if (err) {
+            passthrough.destroy(err);
+        } else {
+            passthrough.end();
+        }
+    });
+    return passthrough;
+}
+
+/**
+ * Decide whether the destination's checksum needs to be recomputed by
+ * streaming the source bytes through a ChecksumTransform.
+ *
+ * @param {object} headers - request headers
+ * @param {object} sourceObjMD - source object metadata
+ * @returns {boolean}
+ */
+function _shouldRecomputeChecksum(headers, sourceObjMD) {
+    const requestedAlgo = headers['x-amz-checksum-algorithm'] && headers['x-amz-checksum-algorithm'].toLowerCase();
+    if (sourceObjMD.checksum && sourceObjMD.checksum.checksumType === 'FULL_OBJECT'
+        && (!requestedAlgo || requestedAlgo === sourceObjMD.checksum.checksumAlgorithm)) {
+        return false;
+    }
+    return Boolean(sourceObjMD.checksum || requestedAlgo);
+}
+
 /**
  * Preps metadata to be saved (based on copy or replace request header)
  * @param {object} request - request
@@ -188,6 +258,14 @@
         storeMetadataParams.defaultRetention = defaultRetentionConfig;
     }
 
+    if (sourceObjMD.checksum && !_shouldRecomputeChecksum(headers, sourceObjMD)) {
+        storeMetadataParams.checksum = {
+            algorithm: sourceObjMD.checksum.checksumAlgorithm,
+            value: sourceObjMD.checksum.checksumValue,
+            type: sourceObjMD.checksum.checksumType,
+        };
+    }
+
     // In case whichMetadata === 'REPLACE' but contentType is undefined in copy
     // request headers, make sure to keep the original header instead
     if (!storeMetadataParams.contentType) {
@@ -278,6 +356,13 @@
             'PUT', destBucketName, err.code, 'copyObject');
         return callback(err);
     }
+    const { error: checksumAlgoErr, algorithm: requestedAlgo } = getCopyObjectChecksumAlgorithm(request.headers);
+    if (checksumAlgoErr) {
+        const err = arsenalErrorFromChecksumError(checksumAlgoErr);
+        log.debug('invalid x-amz-checksum-algorithm', { error: checksumAlgoErr });
+        monitoring.promMetrics('PUT', destBucketName, err.code, 'copyObject');
+        return callback(err);
+    }
     const queryContainsVersionId = checkQueryVersionId(request.query);
     if (queryContainsVersionId instanceof Error) {
         return callback(queryContainsVersionId);
@@ -417,35 +502,37 @@
 
                     return next(null, storeMetadataParams, dataLocator,
                         sourceBucketMD, destBucketMD, destObjMD,
-                        sourceLocationConstraintName, backendInfoDest);
+                        sourceLocationConstraintName, backendInfoDest, sourceObjMD);
                 });
         },
         function getSSEConfiguration(storeMetadataParams, dataLocator, sourceBucketMD,
             destBucketMD, destObjMD, sourceLocationConstraintName,
-            backendInfoDest, next) {
+            backendInfoDest, sourceObjMD, next) {
             getObjectSSEConfiguration(
                     request.headers,
                     destBucketMD,
                     log,
                     (err, sseConfig) =>
                         next(err, storeMetadataParams, dataLocator, sourceBucketMD,
                             destBucketMD, destObjMD, sourceLocationConstraintName,
-                            backendInfoDest, sseConfig));
+                            backendInfoDest, sseConfig, sourceObjMD));
         },
         function goGetData(storeMetadataParams, dataLocator, sourceBucketMD,
             destBucketMD, destObjMD, sourceLocationConstraintName,
-            backendInfoDest, serverSideEncryption, next) {
+            backendInfoDest, serverSideEncryption, sourceObjMD, next) {
             const vcfg = destBucketMD.getVersioningConfiguration();
             const isVersionedObj = vcfg && vcfg.Status === 'Enabled';
             const destLocationConstraintName =
                 storeMetadataParams.dataStoreName;
             const needsEncryption = serverSideEncryption && !!serverSideEncryption.algo;
+            const shouldRecomputeChecksum = _shouldRecomputeChecksum(request.headers, sourceObjMD);
             // skip if source and dest and location constraint the same and
-            // versioning is not enabled
+            // versioning is not enabled, unless we need to recompute a
+            // checksum (which requires streaming the bytes through us)
             // still send along serverSideEncryption info so algo
             // and masterKeyId stored properly in metadata
             if (sourceIsDestination && storeMetadataParams.locationMatch
-                && !isVersionedObj && !needsEncryption) {
+                && !isVersionedObj && !needsEncryption && !shouldRecomputeChecksum) {
                 return next(null, storeMetadataParams, dataLocator, destObjMD,
                     serverSideEncryption, destBucketMD);
             }
@@ -468,30 +555,109 @@
                   externalVersioningErrorMessage), destBucketMD);
             }
             if (dataLocator.length === 0) {
-                if (!storeMetadataParams.locationMatch &&
-                destLocationConstraintType &&
-                constants.externalBackends[destLocationConstraintType]) {
-                    return data.put(null, null, storeMetadataParams.size,
-                        dataStoreContext, backendInfoDest,
-                        log, (error, objectRetrievalInfo) => {
-                            if (error) {
-                                return next(error, destBucketMD);
-                            }
-                            const putResult = {
-                                key: objectRetrievalInfo.key,
-                                dataStoreName: objectRetrievalInfo.
-                                    dataStoreName,
-                                dataStoreType: objectRetrievalInfo.
-                                    dataStoreType,
-                                size: storeMetadataParams.size,
+                const finishZeroByte = () => {
+                    if (!storeMetadataParams.locationMatch && destLocationConstraintType
+                        && constants.externalBackends[destLocationConstraintType]) {
+                        return data.put(null, null, storeMetadataParams.size,
+                            dataStoreContext, backendInfoDest,
+                            log, (error, objectRetrievalInfo) => {
+                                if (error) {
+                                    return next(error, destBucketMD);
+                                }
+                                const putResult = {
+                                    key: objectRetrievalInfo.key,
+                                    dataStoreName: objectRetrievalInfo.dataStoreName,
+                                    dataStoreType: objectRetrievalInfo.dataStoreType,
+                                    size: storeMetadataParams.size,
+                                };
+                                return next(null, storeMetadataParams, [putResult],
+                                    destObjMD, serverSideEncryption, destBucketMD);
+                            });
+                    }
+                    return next(null, storeMetadataParams, dataLocator, destObjMD,
+                        serverSideEncryption, destBucketMD);
+                };
+                if (shouldRecomputeChecksum) {
+                    // No bytes to stream, but AWS still writes the empty-bytes digest of the chosen algorithm.
+                    const algoName = requestedAlgo || sourceObjMD.checksum.checksumAlgorithm;
+                    return Promise.resolve(algorithms[algoName].digest(Buffer.alloc(0)))
+                        .then(digest => {
+                            // eslint-disable-next-line no-param-reassign
+                            storeMetadataParams.checksum = {
+                                algorithm: algoName,
+                                value: digest,
+                                type: 'FULL_OBJECT',
                             };
-                            const putResultArr = [putResult];
-                            return next(null, storeMetadataParams, putResultArr,
-                                destObjMD, serverSideEncryption, destBucketMD);
+                            return finishZeroByte();
+                        }, err => {
+                            log.error('failed to compute empty checksum digest',
+                                { algorithm: algoName, error: err });
+                            return next(errors.InternalError, destBucketMD);
                         });
                 }
-                return next(null, storeMetadataParams, dataLocator, destObjMD,
-                    serverSideEncryption, destBucketMD);
+                return finishZeroByte();
+            }
+            if (shouldRecomputeChecksum) {
+                // Choose algorithm: client-requested if any, otherwise use the source object algo in FULL_OBJECT mode.
+                const recomputeAlgo = requestedAlgo || sourceObjMD.checksum.checksumAlgorithm;
+                // Stream source bytes through a ChecksumTransform and write them out as a single put.
+                log.debug('recomputing checksum on CopyObject',
+                    { algorithm: recomputeAlgo, size: storeMetadataParams.size });
+                const done = jsutil.once((err, results) => {
+                    if (err) {
+                        if (request.sourceServerAccessLog) {
+                            // eslint-disable-next-line no-param-reassign
+                            request.sourceServerAccessLog.error = err;
+                        }
+                        return next(err, destBucketMD);
+                    }
+                    return next(null, storeMetadataParams, results, destObjMD, serverSideEncryption, destBucketMD);
+                });
+                const sourceStream = _pipeSourcePartsThrough(dataLocator, log);
+                const checksumStream = new ChecksumTransform(recomputeAlgo, undefined, false, log);
+                sourceStream.once('error', done);
+                checksumStream.once('error', done);
+                sourceStream.pipe(checksumStream);
+                const doPut = cipherBundle => data.put(
+                    cipherBundle, checksumStream, storeMetadataParams.size,
+                    dataStoreContext, backendInfoDest, log,
+                    (err, dataRetrievalInfo) => {
+                        if (err) {
+                            return done(err);
+                        }
+                        // eslint-disable-next-line no-param-reassign
+                        storeMetadataParams.checksum = {
+                            algorithm: recomputeAlgo,
+                            value: checksumStream.digest,
+                            type: 'FULL_OBJECT',
+                        };
+                        const putResult = {
+                            key: dataRetrievalInfo.key,
+                            size: storeMetadataParams.size,
+                            start: 0,
+                            dataStoreName: dataRetrievalInfo.dataStoreName,
+                            dataStoreType: dataRetrievalInfo.dataStoreType,
+                            dataStoreETag: dataRetrievalInfo.dataStoreETag,
+                            dataStoreVersionId:
+                                dataRetrievalInfo.dataStoreVersionId,
+                        };
+                        if (cipherBundle) {
+                            putResult.cryptoScheme = cipherBundle.cryptoScheme;
+                            putResult.cipheredDataKey =
+                                cipherBundle.cipheredDataKey;
+                        }
+                        return done(null, [putResult]);
+                    });
+                if (serverSideEncryption && serverSideEncryption.algorithm) {
+                    return kms.createCipherBundle(serverSideEncryption, log,
+                        (err, cipherBundle) => {
+                            if (err) {
+                                return done(err);
+                            }
+                            return doPut(cipherBundle);
+                        });
+                }
+                return doPut(null);
             }
             const originalIdentityImpDenies = request.actionImplicitDenies;
             // eslint-disable-next-line no-param-reassign
@@ -639,12 +805,23 @@
                 'PUT', destBucketName, err.code, 'copyObject');
             return callback(err, null, corsHeaders);
         }
+        let checksumXml = '';
+        const checksum = storeMetadataParams && storeMetadataParams.checksum;
+        const checksumAlgo = checksum && algorithms[checksum.algorithm];
+        if (checksum && checksumAlgo) {
+            checksumXml =
+                `<${checksumAlgo.xmlTag}>${checksum.value}</${checksumAlgo.xmlTag}>` +
+                `<ChecksumType>${checksum.type}</ChecksumType>`;
+        } else if (checksum) {
+            log.error('unknown checksum algorithm in source object metadata', { algorithm: checksum.algorithm });
+        }
         const xml = [
             '<?xml version="1.0" encoding="UTF-8"?>',
             '<CopyObjectResult>',
             '<LastModified>', new Date(storeMetadataParams.lastModifiedDate)
                 .toISOString(), '</LastModified>',
             '<ETag>&quot;', storeMetadataParams.contentMD5, '&quot;</ETag>',
+            checksumXml,
             '</CopyObjectResult>',
         ].join('');
         const additionalHeaders = corsHeaders || {};