CLDSRV-863: emit trailer in TrailingChecksumTransform

Author: leif-scality

lib/auth/streamingV4/trailingChecksumTransform.js

@@ -3,10 +3,11 @@ const { errors } = require('arsenal');
 const { maximumAllowedPartSize } = require('../../../constants');
 
 /**
- * This class is designed to handle the chunks sent in a streaming
- * unsigned playload trailer request. In this iteration, we are not checking
- * the checksums, but we are removing them from the stream.
- * S3C-9732 will deal with checksum verification.
+ * This class handles the chunked-upload body format used by
+ * STREAMING-UNSIGNED-PAYLOAD-TRAILER requests. It strips the chunk-size
+ * headers and trailing checksum trailer from the stream, forwarding only
+ * the raw object data. The trailer name and value are emitted via a
+ * 'trailer' event so that ChecksumTransform can validate the checksum.
  */
 class TrailingChecksumTransform extends Transform {
     /**
@@ -20,6 +21,10 @@ class TrailingChecksumTransform extends Transform {
         this.bytesToDiscard = 0; // when trailing \r\n are present, we discard them but they can be in different chunks
         this.bytesToRead = 0; // when a chunk is advertised, the size is put here and we forward all bytes
         this.streamClosed = false;
+        this.readingTrailer = false;
+        this.trailerBuffer = Buffer.alloc(0);
+        this.trailerName = null;
+        this.trailerValue = null;
     }
 
     /**
@@ -30,9 +35,16 @@ class TrailingChecksumTransform extends Transform {
      * @return {function} executes callback with err if applicable
      */
     _flush(callback) {
-        if (!this.streamClosed) {
+        if (!this.streamClosed && this.readingTrailer && this.trailerBuffer.length === 0) {
+            // Nothing came after "0\r\n", don't fail.
+            // If the x-amz-trailer header was present then the trailer is required and ChecksumTransform will fail.
+            return callback();
+        } else if (!this.streamClosed && this.readingTrailer && this.trailerBuffer.length !== 0) {
+            this.log.error('stream ended without trailer "\r\n"');
+            return callback(errors.IncompleteBody);
+        } else if (!this.streamClosed && !this.readingTrailer) {
             this.log.error('stream ended without closing chunked encoding');
-            return callback(errors.InvalidArgument);
+            return callback(errors.IncompleteBody);
         }
         return callback();
     }
@@ -66,6 +78,49 @@ class TrailingChecksumTransform extends Transform {
                 continue;
             }
 
+            // after the 0-size chunk, read the trailer line (e.g. "x-amz-checksum-crc32:YABb/g==")
+            if (this.readingTrailer) {
+                const combined = Buffer.concat([this.trailerBuffer, chunk]);
+                const lineBreakIndex = combined.indexOf('\r\n');
+                if (lineBreakIndex === -1) {
+                    if (combined.byteLength > 1024) {
+                        this.log.error('trailer line too long');
+                        return callback(errors.MalformedTrailerError);
+                    }
+                    // The trailer is not complete yet, continue.
+                    this.trailerBuffer = combined;
+                    return callback();
+                }
+                this.trailerBuffer = Buffer.alloc(0);
+                const fullTrailer = combined.subarray(0, lineBreakIndex);
+                if (fullTrailer.length === 0) {
+                    // The trailer is empty, stop reading.
+                    this.readingTrailer = false;
+                    this.streamClosed = true;
+                    return callback();
+                }
+                let trailerLine = fullTrailer.toString();
+                // Some clients terminate the trailer with \n\r\n instead of
+                // just \r\n, producing a trailing \n in the parsed line.
+                if (trailerLine.endsWith('\n')) {
+                    trailerLine = trailerLine.slice(0, -1);
+                }
+                const colonIndex = trailerLine.indexOf(':');
+                if (colonIndex > 0) {
+                    this.trailerName = trailerLine.slice(0, colonIndex);
+                    this.trailerValue = trailerLine.slice(colonIndex + 1);
+                    this.emit('trailer', this.trailerName, this.trailerValue);
+                } else {
+                    this.log.error('incomplete trailer missing ":"', { trailerLine });
+                    return callback(errors.IncompleteBody);
+                }
+                this.readingTrailer = false;
+                this.streamClosed = true;
+                // The trailer \r\n is the last bytes of the stream per the AWS
+                // chunked upload format, so any remaining bytes are discarded.
+                return callback();
+            }
+
             // we are now looking for the chunk size field
             // no need to look further than 10 bytes since the field cannot be bigger: the max
             // chunk size is 5GB (see constants.maximumAllowedPartSize)
@@ -100,9 +155,9 @@ class TrailingChecksumTransform extends Transform {
             }
             this.chunkSizeBuffer = Buffer.alloc(0);
             if (dataSize === 0) {
-                // TODO: check if the checksum is correct (S3C-9732)
-                // last chunk, no more data to read, the stream is closed
-                this.streamClosed = true;
+                // last chunk, no more data to read; enter trailer-reading mode
+                // bytesToDiscard = 2 below will consume the \r\n after "0"
+                this.readingTrailer = true;
             }
             if (dataSize > maximumAllowedPartSize) {
                 this.log.error('chunk size too big', { dataSize });

tests/functional/raw-node/test/trailingChecksums.js

@@ -6,10 +6,9 @@ const HttpRequestAuthV4 = require('../utils/HttpRequestAuthV4');
 const bucket = 'testunsupportedchecksumsbucket';
 const objectKey = 'key';
 const objData = Buffer.alloc(1024, 'a');
-// note this is not the correct checksum in objDataWithTrailingChecksum
 const objDataWithTrailingChecksum = '10\r\n0123456789abcdef\r\n' +
                                     '10\r\n0123456789abcdef\r\n' +
-                                    '0\r\nx-amz-checksum-crc64nvme:YeIDuLa7tU0=\r\n';
+                                    '0\r\nx-amz-checksum-crc64nvme:skQv82y5rgE=\r\n';
 const objDataWithoutTrailingChecksum = '0123456789abcdef0123456789abcdef';
 
 const config = require('../../config.json');

tests/unit/auth/TrailingChecksumTransform.js

@@ -4,7 +4,6 @@ const async = require('async');
 const { Readable } = require('stream');
 
 const TrailingChecksumTransform = require('../../../lib/auth/streamingV4/trailingChecksumTransform');
-const { stripTrailingChecksumStream } = require('../../../lib/api/apiUtils/object/prepareStream');
 const { DummyRequestLogger } = require('../helpers');
 
 const log = new DummyRequestLogger();
@@ -174,22 +173,24 @@ describe('TrailingChecksumTransform class', () => {
     it('should propagate _flush error via errCb when stream closes without chunked encoding', done => {
         const incompleteData = '10\r\n01234\r6789abcd\r\n\r\n';
         const source = new ChunkedReader([Buffer.from(incompleteData)]);
-        source.headers = { 'x-amz-content-sha256': 'STREAMING-UNSIGNED-PAYLOAD-TRAILER' };
-        const stream = stripTrailingChecksumStream(source, log, err => {
-            assert.deepStrictEqual(err, errors.InvalidArgument);
+        const stream = new TrailingChecksumTransform(log);
+        stream.on('error', err => {
+            assert.deepStrictEqual(err, errors.IncompleteBody);
             done();
         });
+        source.pipe(stream);
         stream.resume();
     });
 
     it('should propagate _transform error via errCb for invalid chunk size', done => {
         const badData = '500000000000\r\n';
         const source = new ChunkedReader([Buffer.from(badData)]);
-        source.headers = { 'x-amz-content-sha256': 'STREAMING-UNSIGNED-PAYLOAD-TRAILER' };
-        const stream = stripTrailingChecksumStream(source, log, err => {
+        const stream = new TrailingChecksumTransform(log);
+        stream.on('error', err => {
             assert.deepStrictEqual(err, errors.InvalidArgument);
             done();
         });
+        source.pipe(stream);
         stream.resume();
     });