Skip to content

CLDSRV-960: backbeat checksum cleanup #6241

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bert-e merged 4 commits into from
Aug 4, 2026
Merged

CLDSRV-960: backbeat checksum cleanup #6241

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dc9d5cc
CLDSRV-960: accept null checksums in prepareStream
leif-scality Jul 30, 2026
4c531c8
CLDSRV-960: remove x-amz-checksum- handling in backbeat routes
leif-scality Jul 30, 2026
153e2fb
CLDSRV-960: validate the primary checksum stream in the Veeam route
leif-scality Aug 4, 2026
a840367
CLDSRV-960: prettier lint
leif-scality Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 80 additions & 62 deletions lib/api/apiUtils/object/prepareStream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,30 +6,81 @@ const { parseContentSHA256, ContentSHA256Type } = require('../integrity/validate
const { errors, errorInstances, jsutil } = require('arsenal');
const { unsupportedSignatureChecksums } = require('../../../../constants');

/**
* Instantiates a ChecksumTransform for the given checksum configuration, or
* returns null if no checksum is requested.
*
* @param {object|null} checksum - { algorithm, isTrailer, expected }, or null
* @param {function} onStreamError - error listener for the transform
* @param {RequestLogger} log - request logger
* @return {ChecksumTransform|null} the transform, or null if no checksum
*/
function createChecksumStream(checksum, onStreamError, log) {
if (!checksum) {
return null;
}
const checksumStream = new ChecksumTransform(checksum.algorithm, checksum.expected, checksum.isTrailer, log);
checksumStream.on('error', onStreamError);
return checksumStream;
}

/**
* Appends the requested checksum transforms to the pipeline: the secondary
* (only validated) first, then the primary, so that the primary always ends the
* pipeline and its digest covers the whole body. Either may be absent.
*
* @param {stream.Readable} inputStream - stream to append the transforms to
* @param {object|null} primary - primary checksum ({ algorithm, isTrailer,
* expected }), or null to compute no stored checksum
* @param {object|null} secondary - secondary checksum, or null
* @param {function} onStreamError - error listener for the transforms
* @param {RequestLogger} log - request logger
* @return {{ stream: stream.Readable, primaryChecksumStream:
* ChecksumTransform|null, secondaryChecksumStream: ChecksumTransform|null }}
*/
function pipeChecksumStreams(inputStream, primary, secondary, onStreamError, log) {
let stream = inputStream;
const secondaryChecksumStream = createChecksumStream(secondary, onStreamError, log);
if (secondaryChecksumStream) {
stream = stream.pipe(secondaryChecksumStream);
}
const primaryChecksumStream = createChecksumStream(primary, onStreamError, log);
if (primaryChecksumStream) {
stream = stream.pipe(primaryChecksumStream);
}
return { stream, primaryChecksumStream, secondaryChecksumStream };
}

/**
* Prepares the request stream for data storage by wrapping it in the
* appropriate transform pipeline based on the x-amz-content-sha256 header.
* The returned stream is always the primary ChecksumTransform (the stored
* checksum). When a secondary checksum is requested it is inserted upstream
* of the primary and exposed via secondaryChecksumStream.
* The primary ChecksumTransform (the stored checksum) is the last transform of
* the pipeline and is returned as primaryChecksumStream; when a secondary
* checksum is requested it is inserted upstream of the primary and exposed via
* secondaryChecksumStream. Callers that need no checksum computed (the
* Backbeat routes, which rely on content-md5) pass no checksums at all, and
* get a pipeline without any ChecksumTransform.
*
* @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 {object} checksums - checksum configuration
* @param {object} checksums.primary - primary checksum
* @param {object|null} checksums - checksum configuration, or null to compute
* no checksum at all
* @param {object|null} checksums.primary - primary checksum
* ({ algorithm, isTrailer, expected }) — validated and its digest returned
* @param {object|null} checksums.secondary - optional secondary checksum
* ({ algorithm, isTrailer, expected }) — only validated; used for MPU parts
* @param {RequestLogger} log - request logger
* @param {function} errCb - error callback invoked if a stream error occurs
* @return {{ error: Arsenal.Error|null, stream: ChecksumTransform|null,
* secondaryChecksumStream: ChecksumTransform|null }}
* @return {{ error: Arsenal.Error|null, stream: stream.Readable|null,
Comment thread
leif-scality marked this conversation as resolved.
* primaryChecksumStream: ChecksumTransform|null,
* secondaryChecksumStream: ChecksumTransform|null,
* contentSHA256Stream: ContentSHA256Transform|null }}

@dvasilas dvasilas Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In some cases we return undefined instead of null for contentSHA256Stream.

@leif-scality leif-scality Aug 3, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed

*/
function prepareStream(request, streamingV4Params, checksums, log, errCb) {
const xAmzContentSHA256 = request.headers['x-amz-content-sha256'];
const { primary, secondary } = checksums;
const { primary = null, secondary = null } = checksums || {};

switch (xAmzContentSHA256) {
case 'STREAMING-AWS4-HMAC-SHA256-PAYLOAD': {
Expand All @@ -52,22 +103,11 @@ function prepareStream(request, streamingV4Params, checksums, log, errCb) {
request.pipe(v4Transform);
v4Transform.headers = request.headers;

let secondaryChecksumStream = null;
let stream = v4Transform;
if (secondary) {
secondaryChecksumStream = new ChecksumTransform(
secondary.algorithm,
secondary.expected,
secondary.isTrailer,
log,
);
secondaryChecksumStream.on('error', onStreamError);
stream = v4Transform.pipe(secondaryChecksumStream);
}

const primaryStream = new ChecksumTransform(primary.algorithm, primary.expected, primary.isTrailer, log);
primaryStream.on('error', onStreamError);
return { error: null, stream: stream.pipe(primaryStream), secondaryChecksumStream };
return {
error: null,
...pipeChecksumStreams(v4Transform, primary, secondary, onStreamError, log),
contentSHA256Stream: null,
};
}
case 'STREAMING-UNSIGNED-PAYLOAD-TRAILER': {
const onStreamError = jsutil.once(errCb);
Expand All @@ -76,30 +116,23 @@ function prepareStream(request, streamingV4Params, checksums, log, errCb) {
request.pipe(trailingChecksumTransform);
trailingChecksumTransform.headers = request.headers;

let secondaryChecksumStream = null;
let stream = trailingChecksumTransform;
if (secondary) {
secondaryChecksumStream = new ChecksumTransform(
secondary.algorithm,
secondary.expected,
secondary.isTrailer,
log,
);
secondaryChecksumStream.on('error', onStreamError);
stream = trailingChecksumTransform.pipe(secondaryChecksumStream);
const checksumStreams = pipeChecksumStreams(
trailingChecksumTransform,
primary,
secondary,
onStreamError,
log,
);
// The trailer is validated against the secondary checksum when
// there is one, otherwise against the primary.
const trailerChecksumStream =
checksumStreams.secondaryChecksumStream || checksumStreams.primaryChecksumStream;
if (trailerChecksumStream) {
trailingChecksumTransform.on('trailer', (name, value) => {
secondaryChecksumStream.setExpectedChecksum(name, value);
trailerChecksumStream.setExpectedChecksum(name, value);
});
}

const primaryStream = new ChecksumTransform(primary.algorithm, primary.expected, primary.isTrailer, log);
primaryStream.on('error', onStreamError);
if (!secondary) {
trailingChecksumTransform.on('trailer', (name, value) => {
primaryStream.setExpectedChecksum(name, value);
});
}
return { error: null, stream: stream.pipe(primaryStream), secondaryChecksumStream };
return { error: null, ...checksumStreams, contentSHA256Stream: null };
}
case 'UNSIGNED-PAYLOAD': // Fallthrough
default: {
Expand All @@ -112,32 +145,17 @@ function prepareStream(request, streamingV4Params, checksums, log, errCb) {

const parsedContentSHA256 = parseContentSHA256(request.headers);
const shouldValidateContentSHA256 = parsedContentSHA256.type === ContentSHA256Type.HexSHA256;
const onStreamError = secondary || shouldValidateContentSHA256 ? jsutil.once(errCb) : errCb;
const onStreamError = jsutil.once(errCb);
let contentSHA256Stream = null;
let secondaryChecksumStream = null;
let stream = request;
if (shouldValidateContentSHA256) {
contentSHA256Stream = new ContentSHA256Transform(parsedContentSHA256.value, log);
contentSHA256Stream.on('error', onStreamError);
stream = stream.pipe(contentSHA256Stream);
}
if (secondary) {
secondaryChecksumStream = new ChecksumTransform(
secondary.algorithm,
secondary.expected,
secondary.isTrailer,
log,
);
secondaryChecksumStream.on('error', onStreamError);
stream = stream.pipe(secondaryChecksumStream);
}

const primaryStream = new ChecksumTransform(primary.algorithm, primary.expected, primary.isTrailer, log);
primaryStream.on('error', onStreamError);
return {
error: null,
stream: stream.pipe(primaryStream),
secondaryChecksumStream,
...pipeChecksumStreams(stream, primary, secondary, onStreamError, log),
contentSHA256Stream,
};
}
Expand Down
66 changes: 37 additions & 29 deletions lib/api/apiUtils/object/storeObject.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,14 @@ const { arsenalErrorFromChecksumError } = require('../../apiUtils/integrity/vali
* @param {object} dataRetrievalInfo - object containing the keys of stored data
* @param {number} dataRetrievalInfo.key - key of the stored data
* @param {string} dataRetrievalInfo.dataStoreName - the implName of the data
* @param {object} checksumStream - checksum transform stream with digest/algoName properties
* @param {object|null} checksumStream - checksum transform stream with
* digest/algoName properties, or null if no checksum was computed
* @param {object} log - request logger instance
* @param {function} cb - callback to send error or move to next task
* @return {function} - calls callback with arguments:

@jonathan-gramain jonathan-gramain Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems we are missing the checksum argument description in the return value

@leif-scality leif-scality Aug 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added

* error, dataRetrievalInfo, and completedHash (if any)
* error, dataRetrievalInfo, completedHash (if any), and checksum —
* `{ algorithm, value }` from checksumStream, or undefined when no checksum
* was computed
*/
function checkHashMatchMD5(stream, hashedStream, dataRetrievalInfo, checksumStream, log, cb) {
const contentMD5 = stream.contentMD5;
Expand All @@ -38,7 +41,7 @@ function checkHashMatchMD5(stream, hashedStream, dataRetrievalInfo, checksumStre
return cb(errors.BadDigest);
});
}
const checksum = { algorithm: checksumStream.algoName, value: checksumStream.digest };
const checksum = checksumStream ? { algorithm: checksumStream.algoName, value: checksumStream.digest } : undefined;
return cb(null, dataRetrievalInfo, completedHash, checksum);
}

Expand All @@ -54,8 +57,9 @@ function checkHashMatchMD5(stream, hashedStream, dataRetrievalInfo, checksumStre
* credentialScope (to be used for streaming v4 auth if applicable)
* @param {BackendInfo} backendInfo - info to determine which data
* backend to use
* @param {object} checksums - checksum configuration
* @param {object} checksums.primary - primary checksum data
* @param {object|null} checksums - checksum configuration, or null to compute
* no checksum at all (callers relying on content-md5 only, e.g. Backbeat)
* @param {object|null} checksums.primary - primary checksum data
* @param {object|null} checksums.secondary - secondary checksum data
* @param {RequestLogger} log - the current stream logger
* @param {function} cb - callback containing result for the next task
Expand All @@ -69,14 +73,14 @@ function dataStore(objectContext, cipherBundle, stream, size, streamingV4Params,
let onStreamError = cbOnce;
const errCb = err => onStreamError(err);

const checksumedStream = prepareStream(stream, streamingV4Params, checksums, log, errCb);
if (checksumedStream.error) {
log.debug('dataStore failed to prepare stream', checksumedStream);
return process.nextTick(() => cbOnce(checksumedStream.error));
const preparedStream = prepareStream(stream, streamingV4Params, checksums, log, errCb);
if (preparedStream.error) {
log.debug('dataStore failed to prepare stream', preparedStream);
return process.nextTick(() => cbOnce(preparedStream.error));
}
return data.put(
cipherBundle,
checksumedStream.stream,
preparedStream.stream,
size,
objectContext,
backendInfo,
Expand Down Expand Up @@ -104,10 +108,10 @@ function dataStore(objectContext, cipherBundle, stream, size, streamingV4Params,
});
};

// stream is always the primary (end of pipe, stored checksum).
// secondaryChecksumStream and contentSHA256Stream are upstream and
// only validated.
const { secondaryChecksumStream, contentSHA256Stream } = checksumedStream;
// The primary checksum stream (when there is one) is the end of the
// pipe and holds the stored checksum. secondaryChecksumStream and
// contentSHA256Stream are upstream and only validated.
const { primaryChecksumStream, secondaryChecksumStream, contentSHA256Stream } = preparedStream;

const doValidate = () => {
// Validate the SigV4 payload hash (x-amz-content-sha256) first.
Expand Down Expand Up @@ -137,22 +141,24 @@ function dataStore(objectContext, cipherBundle, stream, size, streamingV4Params,
}
}
// Validate the primary (stored) checksum.
const primaryErr = checksumedStream.stream.validateChecksum();
if (primaryErr) {
log.debug('failed primary checksum validation', { error: primaryErr });
return data.batchDelete([dataRetrievalInfo], null, null, log, deleteErr => {
if (deleteErr) {
log.error('dataStore failed to delete old data', { error: deleteErr });
}
return cbOnce(arsenalErrorFromChecksumError(primaryErr));
});
if (primaryChecksumStream) {
const primaryErr = primaryChecksumStream.validateChecksum();
if (primaryErr) {
log.debug('failed primary checksum validation', { error: primaryErr });
return data.batchDelete([dataRetrievalInfo], null, null, log, deleteErr => {
if (deleteErr) {
log.error('dataStore failed to delete old data', { error: deleteErr });
}
return cbOnce(arsenalErrorFromChecksumError(primaryErr));
});
}
}
if (!secondaryChecksumStream) {
return checkHashMatchMD5(
stream,
hashedStream,
dataRetrievalInfo,
checksumedStream.stream,
primaryChecksumStream,
log,
cbOnce,
);
Expand All @@ -165,7 +171,7 @@ function dataStore(objectContext, cipherBundle, stream, size, streamingV4Params,
stream,
hashedStream,
dataRetrievalInfo,
checksumedStream.stream,
primaryChecksumStream,
log,
(err, dataInfo, hash, primaryChecksum) => {
if (err) {
Expand All @@ -184,12 +190,14 @@ function dataStore(objectContext, cipherBundle, stream, size, streamingV4Params,
// 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 is the primary (end of pipe) — when it finishes all
// upstream transforms (including the secondary) have flushed.
if (checksumedStream.stream.writableFinished) {
// The last checksum stream of the pipe is awaited: when it finishes
// all upstream transforms have flushed. If there is none, there is
// no digest to wait for.
const lastChecksumStream = primaryChecksumStream || secondaryChecksumStream || contentSHA256Stream;
if (!lastChecksumStream || lastChecksumStream.writableFinished) {
return doValidate();
}
checksumedStream.stream.once('finish', doValidate);
lastChecksumStream.once('finish', doValidate);
return undefined;
},
);
Expand Down
Loading
Loading