CLDSRV-863: pipe requests to ChecksumTransform and validate checksum after upload

Author: leif-scality

lib/api/apiUtils/object/prepareStream.js

@@ -1,50 +1,111 @@
 const V4Transform = require('../../../auth/streamingV4/V4Transform');
 const TrailingChecksumTransform = require('../../../auth/streamingV4/trailingChecksumTransform');
+const ChecksumTransform = require('../../../auth/streamingV4/ChecksumTransform');
+const {
+    getChecksumDataFromHeaders,
+    arsenalErrorFromChecksumError,
+} = require('../../apiUtils/integrity/validateChecksums');
+const { errors, jsutil } = require('arsenal');
+const { unsupportedSignatureChecksums } = require('../../../../constants');
 
 /**
- * Prepares the stream if the chunks are sent in a v4 Auth request
- * @param {object} stream - stream containing the data
- * @param {object | null } streamingV4Params - if v4 auth, object containing
- * accessKey, signatureFromRequest, region, scopeDate, timestamp, and
- * credentialScope (to be used for streaming v4 auth if applicable)
- * @param {RequestLogger} log - the current request logger
- * @param {function} errCb - callback called if an error occurs
- * @return {object|null} - V4Transform object if v4 Auth request, or
- * the original stream, or null if the request has no V4 params but
- * the type of request requires them
+ * Prepares the request stream for data storage by wrapping it in the
+ * appropriate transform pipeline based on the x-amz-content-sha256 header.
+ * Always returns a ChecksumTransform as the final stream.
+ * If no checksum was sent by the client a CRC64NVME ChecksumTransform is returned.
+ *
+ * @param {object} request - incoming HTTP request with headers and body stream
+ * @param {object|null} streamingV4Params - v4 streaming auth params (accessKey,
+ * signatureFromRequest, region, scopeDate, timestamp, credentialScope), or
+ * null/undefined for non-v4 requests
+ * @param {RequestLogger} log - request logger
+ * @param {function} errCb - error callback invoked if a stream error occurs
+ * @return {{ error: Arsenal.Error|null, stream: ChecksumTransform|null }}
+ * error is set and stream is null if the request headers are invalid;
+ * otherwise error is null and stream is the ChecksumTransform to read from
  */
-function prepareStream(stream, streamingV4Params, log, errCb) {
-    if (stream.headers['x-amz-content-sha256'] ===
-        'STREAMING-AWS4-HMAC-SHA256-PAYLOAD') {
-        if (typeof streamingV4Params !== 'object') {
-            // this might happen if the user provided a valid V2
-            // Authentication header, while the chunked upload method
-            // requires V4: in such case we don't get any V4 params
-            // and we should return an error to the client.
-            return null;
-        }
-        const v4Transform = new V4Transform(streamingV4Params, log, errCb);
-        stream.pipe(v4Transform);
-        v4Transform.headers = stream.headers;
-        return v4Transform;
-    }
-    return stream;
-}
+function prepareStream(request, streamingV4Params, log, errCb) {
+    const xAmzContentSHA256 = request.headers['x-amz-content-sha256'];
 
-function stripTrailingChecksumStream(stream, log, errCb) {
-    // don't do anything if we are not in the correct integrity check mode
-    if (stream.headers['x-amz-content-sha256'] !== 'STREAMING-UNSIGNED-PAYLOAD-TRAILER') {
-        return stream;
+    const checksumAlgo = getChecksumDataFromHeaders(request.headers);
+    if (checksumAlgo.error) {
+        log.debug('prepareStream invalid checksum headers', checksumAlgo);
+        return { error: arsenalErrorFromChecksumError(checksumAlgo), stream: null };
     }
 
-    const trailingChecksumTransform = new TrailingChecksumTransform(log);
-    trailingChecksumTransform.on('error', errCb);
-    stream.pipe(trailingChecksumTransform);
-    trailingChecksumTransform.headers = stream.headers;
-    return trailingChecksumTransform;
+    let stream = request;
+    switch (xAmzContentSHA256) {
+        case 'STREAMING-AWS4-HMAC-SHA256-PAYLOAD': {
+            if (streamingV4Params === null || typeof streamingV4Params !== 'object') {
+                // this might happen if the user provided a valid V2
+                // Authentication header, while the chunked upload method
+                // requires V4: in such case we don't get any V4 params
+                // and we should return an error to the client.
+                log.error('missing v4 streaming params for chunked upload', {
+                    method: 'prepareStream',
+                    streamingV4ParamsType: typeof streamingV4Params,
+                    streamingV4Params,
+                });
+                return { error: errors.InvalidArgument, stream: null };
+            }
+            const v4Transform = new V4Transform(streamingV4Params, log, errCb);
+            request.pipe(v4Transform);
+            v4Transform.headers = request.headers;
+            stream = v4Transform;
+
+            const checksumedStream = new ChecksumTransform(
+                checksumAlgo.algorithm,
+                checksumAlgo.expected,
+                checksumAlgo.isTrailer,
+                log,
+            );
+            checksumedStream.on('error', errCb);
+            stream.pipe(checksumedStream);
+            return { error: null, stream: checksumedStream };
+        }
+        case 'STREAMING-UNSIGNED-PAYLOAD-TRAILER': {
+            // Use a once-guard so that auto-destroying both piped streams
+            // when one errors does not result in errCb being called twice.
+            const onStreamError = jsutil.once(errCb);
+            const trailingChecksumTransform = new TrailingChecksumTransform(log);
+            trailingChecksumTransform.on('error', onStreamError);
+            request.pipe(trailingChecksumTransform);
+            trailingChecksumTransform.headers = request.headers;
+            stream = trailingChecksumTransform;
+
+            const checksumedStream = new ChecksumTransform(
+                checksumAlgo.algorithm,
+                checksumAlgo.expected,
+                checksumAlgo.isTrailer,
+                log,
+            );
+            checksumedStream.on('error', onStreamError);
+            trailingChecksumTransform.on('trailer', (name, value) => {
+                checksumedStream.setExpectedChecksum(name, value);
+            });
+            stream.pipe(checksumedStream);
+            return { error: null, stream: checksumedStream };
+        }
+        case 'UNSIGNED-PAYLOAD': // Fallthrough
+        default: {
+            if (unsupportedSignatureChecksums.has(xAmzContentSHA256)) {
+                return {
+                    error: errors.BadRequest.customizeDescription(`${xAmzContentSHA256} is not supported`),
+                    stream: null,
+                };
+            }
+
+            const checksumedStream = new ChecksumTransform(
+                checksumAlgo.algorithm,
+                checksumAlgo.expected,
+                checksumAlgo.isTrailer,
+                log,
+            );
+            checksumedStream.on('error', errCb);
+            stream.pipe(checksumedStream);
+            return { error: null, stream: checksumedStream };
+        }
+    }
 }
 
-module.exports = {
-    prepareStream,
-    stripTrailingChecksumStream,
-};
+module.exports = { prepareStream };

lib/api/apiUtils/object/storeObject.js

@@ -1,7 +1,8 @@
 const { errors, jsutil } = require('arsenal');
 
 const { data } = require('../../../data/wrapper');
-const { prepareStream, stripTrailingChecksumStream } = require('./prepareStream');
+const { prepareStream } = require('./prepareStream');
+const { arsenalErrorFromChecksumError } = require('../../apiUtils/integrity/validateChecksums');
 
 /**
  * Check that `hashedStream.completedHash` matches header `stream.contentMD5`
@@ -58,31 +59,67 @@ function checkHashMatchMD5(stream, hashedStream, dataRetrievalInfo, log, cb) {
 function dataStore(objectContext, cipherBundle, stream, size,
     streamingV4Params, backendInfo, log, cb) {
     const cbOnce = jsutil.once(cb);
-    const dataStreamTmp = prepareStream(stream, streamingV4Params, log, cbOnce);
-    if (!dataStreamTmp) {
-        return process.nextTick(() => cb(errors.InvalidArgument));
+
+    // errCb is delegated through a mutable reference so it can be upgraded to
+    // include batchDelete once data.put has actually stored data.
+    let onStreamError = cbOnce;
+    const errCb = err => onStreamError(err);
+
+    const checksumedStream = prepareStream(stream, streamingV4Params, log, errCb);
+    if (checksumedStream.error) {
+        log.debug('dataStore failed to prepare stream', checksumedStream);
+        return process.nextTick(() => cbOnce(checksumedStream.error));
     }
-    const dataStream = stripTrailingChecksumStream(dataStreamTmp, log, cbOnce);
     return data.put(
-        cipherBundle, dataStream, size, objectContext, backendInfo, log,
+        cipherBundle, checksumedStream.stream, size, objectContext, backendInfo, log,
         (err, dataRetrievalInfo, hashedStream) => {
             if (err) {
-                log.error('error in datastore', {
-                    error: err,
-                });
+                log.error('error in datastore', { error: err });
                 return cbOnce(err);
             }
             if (!dataRetrievalInfo) {
-                log.fatal('data put returned neither an error nor a key', {
-                    method: 'storeObject::dataStore',
-                });
+                log.fatal('data put returned neither an error nor a key', { method: 'storeObject::dataStore' });
                 return cbOnce(errors.InternalError);
             }
-            log.trace('dataStore: backend stored key', {
-                dataRetrievalInfo,
-            });
-            return checkHashMatchMD5(stream, hashedStream,
-                                     dataRetrievalInfo, log, cbOnce);
+            log.trace('dataStore: backend stored key', { dataRetrievalInfo });
+
+            // Data is now stored. Upgrade the error handler so that any stream
+            // error from this point on cleans up the stored data first.
+            onStreamError = streamErr => {
+                log.error('checksum stream error after data.put', { error: streamErr });
+                return data.batchDelete([dataRetrievalInfo], null, null, log, deleteErr => {
+                    if (deleteErr) {
+                        log.error('dataStore failed to delete old data', { error: deleteErr });
+                    }
+                    cbOnce(streamErr);
+                });
+            };
+
+            const doValidate = () => {
+                const checksumErr = checksumedStream.stream.validateChecksum();
+                if (checksumErr) {
+                    log.debug('failed checksum validation stream', checksumErr);
+                    return data.batchDelete([dataRetrievalInfo], null, null, log, deleteErr => {
+                        if (deleteErr) {
+                            // Failure of batch delete is only logged.
+                            log.error('dataStore failed to delete old data', { error: deleteErr });
+                        }
+                        return cbOnce(arsenalErrorFromChecksumError(checksumErr));
+                    });
+                }
+                return checkHashMatchMD5(stream, hashedStream, dataRetrievalInfo, log, cbOnce);
+            };
+
+            // ChecksumTransform._flush computes the digest asynchronously for
+            // some algorithms (e.g. crc64nvme). writableFinished is true once
+            // _flush has called its callback, guaranteeing this.digest is set.
+            // Stream piping ordering means this is virtually always true here,
+            // but we wait for 'finish' explicitly to be safe.
+            if (checksumedStream.stream.writableFinished) {
+                return doValidate();
+            }
+            checksumedStream.stream.once('finish', doValidate);
+            return undefined;
         });
 }