Support streaming uploads on the Veeam SOSAPI route

lib/routes/veeam/utils.js

@@ -1,11 +1,16 @@
 const xml2js = require('xml2js');
-const { errors, errorInstances, jsutil } = require('arsenal');
+const { errors, errorInstances } = require('arsenal');
 const { Readable, Writable, pipeline: streamPipeline } = require('stream');
 const { promisify } = require('util');
 const collectResponseHeaders = require('../../utilities/collectResponseHeaders');
 const collectCorsHeaders = require('../../utilities/collectCorsHeaders');
 const crypto = require('crypto');
-const { prepareStream } = require('arsenal/build/lib/s3middleware/prepareStream');
+const { prepareStream } = require('../../api/apiUtils/object/prepareStream');
+const {
+    getChecksumDataFromHeaders,
+    arsenalErrorFromChecksumError,
+    defaultChecksumData,
+} = require('../../api/apiUtils/integrity/validateChecksums');
 const UtilizationService = require('../../utilization/instance');
 const metadata = require('../../metadata/wrapper');
 
@@ -25,6 +30,10 @@
 /**
  * Generic function to get data from a client request.
  *
+ * The request stream is decoded and validated according to its
+ * x-amz-content-sha256 value (plain, signed streaming or unsigned streaming
+ * with trailing checksum), with the same semantics as the object data path.
+ *
  * @param {object} request - incoming request
  * @param {object} log - logger object
  * @returns {Promise<string>}
@@ -40,33 +49,54 @@
             `maximum allowed content-length is ${ContentLengthThreshold} bytes`,
         );
     }
-    return await new Promise((resolve, reject) => {
-        const settle = jsutil.once((err, result) => {
-            if (err) {
-                return reject(err);
+    const headerChecksum = getChecksumDataFromHeaders(request.headers);
+    if (headerChecksum && headerChecksum.error) {
+        throw arsenalErrorFromChecksumError(headerChecksum);
+    }
+    const checksums = { primary: headerChecksum || defaultChecksumData, secondary: null };
+    let totalLength = 0;
+    const chunks = [];
+    const collector = new Writable({
+        write(chunk, _enc, cb) {
+            totalLength += chunk.length;
+            if (totalLength > parsedContentLength) {
+                log.error('data stream exceed announced size', { parsedContentLength, overflow: totalLength });
+                return cb(
+                    errorInstances.InvalidRequest.customizeDescription(
+                        'request body exceeds the announced content-length',
+                    ),
+                );
             }
-            return resolve(result);
-        });
-        let totalLength = 0;
-        const chunks = [];
-        const collector = new Writable({
-            write(chunk, _enc, cb) {
-                totalLength += chunk.length;
-                if (totalLength > parsedContentLength) {
-                    log.error('data stream exceed announced size', { parsedContentLength, overflow: totalLength });
-                    return cb(errors.InternalError);
-                }
-                chunks.push(chunk);
-                return cb();
-            },
-            final(cb) {
-                settle(null, Buffer.concat(chunks).toString());
-                cb();
-            },
-        });
-        const dataStream = prepareStream(request, request.streamingV4Params, log, settle);
-        pipeline(dataStream, collector).catch(err => settle(err));
+            chunks.push(chunk);
+            return cb();
+        },
+    });
+    // Transform errors can be delivered through the errCb side-channel
+    // without necessarily erroring the pipeline, so bridge them into a
+    // promise raced against the pipeline completion. A repeated rejection
+    // after settlement is a no-op; the no-op handler below keeps the
+    // rejection handled even on paths that throw before the race subscribes.
+    let onStreamError;
+    const streamError = new Promise((resolve, reject) => {
+        onStreamError = reject;
     });
+    streamError.catch(() => {});
+    const prepared = prepareStream(request, request.streamingV4Params, checksums, log, onStreamError);
+    if (prepared.error) {
+        throw prepared.error;
+    }
+    await Promise.race([pipeline(prepared.stream, collector), streamError]);
+    // Checksum transforms only compute digests while streaming: validation
+    // against the expected values (header or trailer) must be done once the
+    // stream is fully consumed.
+    const checksumErr =
+        (prepared.contentSHA256Stream && prepared.contentSHA256Stream.validateChecksum()) ||
+        prepared.stream.validateChecksum();
+    if (checksumErr) {
+        log.debug('failed checksum validation', { error: checksumErr });
+        throw arsenalErrorFromChecksumError(checksumErr);
+    }
+    return Buffer.concat(chunks).toString();
 }
 
 /**