CLDSRV-863: add ChecksumTransform to calculate and verify stream checksums

Author: leif-scality

lib/auth/streamingV4/ChecksumTransform.js

@@ -0,0 +1,92 @@
+const { errors } = require('arsenal');
+const { algorithms, ChecksumError } = require('../../api/apiUtils/integrity/validateChecksums');
+const { Transform } = require('stream');
+
+class ChecksumTransform extends Transform {
+    constructor(algoName, expectedDigest, isTrailer, log) {
+        super({});
+        this.log = log;
+        this.algoName = algoName;
+        this.algo = algorithms[algoName];
+        this.hash = this.algo.createHash();
+        this.digest = undefined;
+        this.expectedDigest = expectedDigest;
+        this.isTrailer = isTrailer;
+        this.trailerChecksumName = undefined;
+        this.trailerChecksumValue = undefined;
+    }
+
+    setExpectedChecksum(name, value) {
+        this.trailerChecksumName = name;
+        this.trailerChecksumValue = value;
+    }
+
+    validateChecksum() {
+        if (this.isTrailer) {
+            // x-amz-trailer in headers but no trailer in body.
+            if (this.trailerChecksumValue === undefined) {
+                return {
+                    error: ChecksumError.TrailerMissing,
+                    details: { expectedTrailer: `x-amz-checksum-${this.algoName}` },
+                };
+            }
+
+            if (this.trailerChecksumName !== `x-amz-checksum-${this.algoName}`) {
+                return { error: ChecksumError.TrailerAlgoMismatch, details: { algorithm: this.algoName } };
+            }
+
+            const expected = this.trailerChecksumValue;
+            if (!this.algo.isValidDigest(expected)) {
+                return {
+                    error: ChecksumError.TrailerChecksumMalformed,
+                    details: { algorithm: this.algoName, expected },
+                };
+            }
+
+            if (this.digest !== this.trailerChecksumValue) {
+                return {
+                    error: ChecksumError.XAmzMismatch,
+                    details: { algorithm: this.algoName, calculated: this.digest, expected },
+                };
+            }
+
+            return null;
+        }
+
+        if (this.trailerChecksumValue) {
+            // Trailer found in the body but no x-amz-trailer in the headers.
+            return {
+                error: ChecksumError.TrailerUnexpected,
+                details: { name: this.trailerChecksumName, val: this.trailerChecksumValue },
+            };
+        }
+
+        if (this.expectedDigest) {
+            if (this.digest !== this.expectedDigest) {
+                return {
+                    error: ChecksumError.XAmzMismatch,
+                    details: { algorithm: this.algoName, calculated: this.digest, expected: this.expectedDigest },
+                };
+            }
+        }
+
+        return null;
+    }
+
+    _flush(callback) {
+        Promise.resolve(this.algo.digestFromHash(this.hash))
+            .then(digest => { this.digest = digest; })
+            .then(() => callback(), err => {
+                this.log.error('failed to compute checksum digest', { error: err, algorithm: this.algoName });
+                callback(errors.InternalError);
+            });
+    }
+
+    _transform(chunk, encoding, callback) {
+        const input = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+        this.hash.update(input, encoding);
+        callback(null, input);
+    }
+}
+
+module.exports = ChecksumTransform;

tests/unit/auth/ChecksumTransform.js

@@ -0,0 +1,182 @@
+const assert = require('assert');
+const { errors } = require('arsenal');
+
+const { algorithms, ChecksumError } = require('../../../lib/api/apiUtils/integrity/validateChecksums');
+const ChecksumTransform = require('../../../lib/auth/streamingV4/ChecksumTransform');
+const { DummyRequestLogger } = require('../helpers');
+
+const log = new DummyRequestLogger();
+const testData = Buffer.from('hello world');
+const algos = ['crc32', 'crc32c', 'sha1', 'sha256', 'crc64nvme'];
+
+// Helper: pipe chunks into a ChecksumTransform, collect output, resolve on finish
+function runTransform(stream, chunks) {
+    return new Promise((resolve, reject) => {
+        const output = [];
+        stream.on('data', chunk => output.push(chunk));
+        stream.on('end', () => resolve(Buffer.concat(output)));
+        stream.on('error', reject);
+        for (const chunk of chunks) {
+            stream.write(chunk);
+        }
+        stream.end();
+    });
+}
+
+// Helper: pipe data through and wait for finish without collecting output
+function drainTransform(stream, chunks) {
+    return new Promise((resolve, reject) => {
+        stream.resume();
+        stream.on('finish', resolve);
+        stream.on('error', reject);
+        for (const chunk of chunks) {
+            stream.write(chunk);
+        }
+        stream.end();
+    });
+}
+
+describe('ChecksumTransform basic behaviour', () => {
+    let expectedDigests;
+
+    before(async () => {
+        expectedDigests = {};
+        for (const algo of algos) {
+            expectedDigests[algo] = await Promise.resolve(algorithms[algo].digest(testData));
+        }
+    });
+
+    for (const algo of algos) {
+        it(`[${algo}] passes data through unchanged`, async () => {
+            const stream = new ChecksumTransform(algo, undefined, false, log);
+            const output = await runTransform(stream, [testData]);
+            assert.deepStrictEqual(output, testData);
+        });
+
+        it(`[${algo}] computes digest correctly after stream ends`, async () => {
+            const stream = new ChecksumTransform(algo, undefined, false, log);
+            await drainTransform(stream, [testData]);
+            assert.strictEqual(stream.digest, expectedDigests[algo]);
+        });
+
+        it(`[${algo}] handles multi-chunk input: digest matches single-chunk equivalent`, async () => {
+            const half = Math.floor(testData.length / 2);
+            const stream = new ChecksumTransform(algo, undefined, false, log);
+            await drainTransform(stream, [testData.subarray(0, half), testData.subarray(half)]);
+            assert.strictEqual(stream.digest, expectedDigests[algo]);
+        });
+
+        it(`[${algo}] handles Buffer and string chunks equally`, async () => {
+            const streamBuf = new ChecksumTransform(algo, undefined, false, log);
+            const streamStr = new ChecksumTransform(algo, undefined, false, log);
+            await drainTransform(streamBuf, [testData]);
+            await drainTransform(streamStr, [testData.toString()]);
+            assert.strictEqual(streamBuf.digest, streamStr.digest);
+        });
+    }
+
+    it('emits error via stream error event if digestFromHash fails', done => {
+        const stream = new ChecksumTransform('crc32', undefined, false, log);
+        // Replace digestFromHash to return a rejected Promise
+        stream.algo = Object.assign({}, stream.algo, {
+            digestFromHash: () => Promise.reject(new Error('simulated digest failure')),
+        });
+        stream.on('error', err => {
+            assert.deepStrictEqual(err, errors.InternalError);
+            done();
+        });
+        stream.write(testData);
+        stream.end();
+        stream.resume();
+    });
+});
+
+describe('ChecksumTransform validateChecksum — non-trailer mode (isTrailer=false)', () => {
+    let crc32Digest;
+
+    before(async () => {
+        crc32Digest = await Promise.resolve(algorithms.crc32.digest(testData));
+    });
+
+    it('returns null when no expectedDigest and no trailer received', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, false, log);
+        await drainTransform(stream, [testData]);
+        assert.strictEqual(stream.validateChecksum(), null);
+    });
+
+    it('returns null when expectedDigest matches computed digest', async () => {
+        const stream = new ChecksumTransform('crc32', crc32Digest, false, log);
+        await drainTransform(stream, [testData]);
+        assert.strictEqual(stream.validateChecksum(), null);
+    });
+
+    it('returns XAmzMismatch when expectedDigest does not match computed digest', async () => {
+        const stream = new ChecksumTransform('crc32', 'AAAAAA==', false, log);
+        await drainTransform(stream, [testData]);
+        const result = stream.validateChecksum();
+        assert.strictEqual(result.error, ChecksumError.XAmzMismatch);
+        assert.strictEqual(result.details.algorithm, 'crc32');
+        assert.strictEqual(result.details.calculated, crc32Digest);
+        assert.strictEqual(result.details.expected, 'AAAAAA==');
+    });
+
+    it('returns TrailerUnexpected when setExpectedChecksum was called but isTrailer=false', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, false, log);
+        stream.setExpectedChecksum('x-amz-checksum-crc32', crc32Digest);
+        await drainTransform(stream, [testData]);
+        const result = stream.validateChecksum();
+        assert.strictEqual(result.error, ChecksumError.TrailerUnexpected);
+    });
+});
+
+describe('ChecksumTransform validateChecksum — trailer mode (isTrailer=true)', () => {
+    let crc32Digest;
+
+    before(async () => {
+        crc32Digest = await Promise.resolve(algorithms.crc32.digest(testData));
+    });
+
+    it('returns TrailerMissing when setExpectedChecksum was never called', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, true, log);
+        await drainTransform(stream, [testData]);
+        const result = stream.validateChecksum();
+        assert.strictEqual(result.error, ChecksumError.TrailerMissing);
+        assert.strictEqual(result.details.expectedTrailer, 'x-amz-checksum-crc32');
+    });
+
+    it('returns TrailerAlgoMismatch when trailer name does not match algo', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, true, log);
+        stream.setExpectedChecksum('x-amz-checksum-sha256', 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=');
+        await drainTransform(stream, [testData]);
+        const result = stream.validateChecksum();
+        assert.strictEqual(result.error, ChecksumError.TrailerAlgoMismatch);
+        assert.strictEqual(result.details.algorithm, 'crc32');
+    });
+
+    it('returns TrailerChecksumMalformed when trailer value is not a valid digest for the algo', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, true, log);
+        stream.setExpectedChecksum('x-amz-checksum-crc32', 'not-valid!');
+        await drainTransform(stream, [testData]);
+        const result = stream.validateChecksum();
+        assert.strictEqual(result.error, ChecksumError.TrailerChecksumMalformed);
+        assert.strictEqual(result.details.algorithm, 'crc32');
+    });
+
+    it('returns XAmzMismatch when trailer value is valid but does not match computed digest', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, true, log);
+        stream.setExpectedChecksum('x-amz-checksum-crc32', 'AAAAAA==');
+        await drainTransform(stream, [testData]);
+        const result = stream.validateChecksum();
+        assert.strictEqual(result.error, ChecksumError.XAmzMismatch);
+        assert.strictEqual(result.details.algorithm, 'crc32');
+        assert.strictEqual(result.details.calculated, crc32Digest);
+        assert.strictEqual(result.details.expected, 'AAAAAA==');
+    });
+
+    it('returns null when trailer name and value match computed digest', async () => {
+        const stream = new ChecksumTransform('crc32', undefined, true, log);
+        stream.setExpectedChecksum('x-amz-checksum-crc32', crc32Digest);
+        await drainTransform(stream, [testData]);
+        assert.strictEqual(stream.validateChecksum(), null);
+    });
+});