diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index cf869c9629..2691132178 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -69,7 +69,7 @@ env: GIT_CLONE_PROTECTION_ACTIVE: 'false' jobs: linting-coverage: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 steps: - name: Checkout uses: actions/checkout@v4 @@ -125,7 +125,7 @@ jobs: if: always() build: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 steps: - name: Checkout uses: actions/checkout@v4 @@ -164,7 +164,7 @@ jobs: cache-to: type=gha,mode=max,scope=pykmip build-federation-image: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 steps: - name: Checkout uses: actions/checkout@v4 @@ -191,7 +191,7 @@ jobs: cache-to: type=gha,mode=max,scope=federation multiple-backend: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 needs: build env: CLOUDSERVER_IMAGE: ghcr.io/${{ github.repository }}:${{ github.sha }} @@ -228,7 +228,7 @@ jobs: if: always() file-ft-tests: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 needs: build env: S3BACKEND: file @@ -267,7 +267,7 @@ jobs: if: always() utapi-v2-tests: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 needs: build env: ENABLE_UTAPI_V2: t @@ -302,7 +302,7 @@ jobs: if: always() kmip-ft-tests: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 needs: build env: S3BACKEND: file diff --git a/lib/api/api.js b/lib/api/api.js index af4770085a..ffe49451c8 100644 --- a/lib/api/api.js +++ b/lib/api/api.js @@ -57,6 +57,8 @@ const objectPutCopyPart = require('./objectPutCopyPart'); const objectPutRetention = require('./objectPutRetention'); const prepareRequestContexts = require('./apiUtils/authorization/prepareRequestContexts'); +const destroyDecipherStreamsOnError + = require('./apiUtils/object/destroyDecipherStreamsOnError'); const serviceGet = require('./serviceGet'); const vault = require('../auth/vault'); const website = require('./website'); @@ -274,7 +276,10 @@ const api = { sourceObject, sourceVersionId, log, callback); } if (apiMethod === 'objectGet') { - return this[apiMethod](userInfo, request, returnTagCount, log, callback); + return this[apiMethod](userInfo, request, returnTagCount, log, (err, ...results) => { + destroyDecipherStreamsOnError(err, request, response); + return callback(err, ...results); + }); } return this[apiMethod](userInfo, request, log, callback); }); diff --git a/lib/api/apiUtils/object/destroyDecipherStreamsOnError.js b/lib/api/apiUtils/object/destroyDecipherStreamsOnError.js new file mode 100644 index 0000000000..5be8fd655a --- /dev/null +++ b/lib/api/apiUtils/object/destroyDecipherStreamsOnError.js @@ -0,0 +1,34 @@ +/** + * Destroy the decipher streams prepared upfront by objectGet when they are not + * all consumed: objectGet error, error response or interrupted streaming. + * They hold native OpenSSL memory that is better released than left to GC. + * @param {Error|null} err - error returned by objectGet + * @param {object} request - request, with the encrypted locations set by objectGet + * @param {http.ServerResponse} response - response sent to the client + * @return {undefined} + */ +function destroyDecipherStreamsOnError(err, request, response) { + // only set by objectGet when upfront KMS calls are not skipped + const { kmsDecipherLocations } = request; + if (!kmsDecipherLocations) { + return; + } + const destroyDecipherStreams = () => kmsDecipherLocations.forEach(loc => { + if (loc.decipherStream) { + loc.decipherStream.destroy(); + } + }); + // isclosed is set by the server when the client closed before objectGet returned + if (err || response.isclosed) { + destroyDecipherStreams(); + return; + } + response.once('close', () => { + // a successful response fully sent has consumed every decipher stream + if (!response.writableFinished || response.statusCode >= 400) { + destroyDecipherStreams(); + } + }); +} + +module.exports = destroyDecipherStreamsOnError; diff --git a/lib/api/objectGet.js b/lib/api/objectGet.js index 3f3b84c6da..90709626a6 100644 --- a/lib/api/objectGet.js +++ b/lib/api/objectGet.js @@ -17,6 +17,13 @@ const kms = require('../kms/wrapper'); const validateHeaders = s3middleware.validateConditionalHeaders; +// Max parallel KMS calls to decrypt the data keys of an object's parts before streaming. Defaults to 100. +// 0 to skip upfront calls: parts are decrypted while streaming (historical behavior before CLDSRV-629) +// but KMS errors are not reported, success response is sent before. +const envMaxParallelKmsDecrypt = Number.parseInt(process.env.GET_OBJECT_MAX_PARALLEL_KMS_DECRYPT, 10); +const MAX_PARALLEL_KMS_DECRYPT = envMaxParallelKmsDecrypt > 0 ? envMaxParallelKmsDecrypt : 100; +const SKIP_UPFRONT_KMS_DECRYPT = envMaxParallelKmsDecrypt === 0; + /** * GET Object - Get an object * @param {AuthInfo} authInfo - Instance of AuthInfo class with requester's info @@ -213,13 +220,21 @@ function objectGet(authInfo, request, returnTagCount, log, callback) { dataLocator = setPartRanges(dataLocator, byteRange); } } - // Check KMS Key access and usability before checking data + // KMS is called for all parts upfront to fail early on error before streaming, instead of + // closing the socket mid-stream. For many parts (10k), this can delay time to first byte + // and flood the KMS with simultaneous queries. Also the decipher streams hold native C++ + // memory for OpenSSL: should be destroyed early on error, not left to GC. // diff with AWS: for empty object (no dataLocator) KMS not checked - return async.each(dataLocator || [], + const upfrontKmsLocations = SKIP_UPFRONT_KMS_DECRYPT ? [] + : (dataLocator || []).filter(objectGetInfo => objectGetInfo.cipheredDataKey); + if (upfrontKmsLocations.length > 0) { + // Reference for destroyDecipherStreamsOnError if not consumed + // eslint-disable-next-line no-param-reassign + request.kmsDecipherLocations = upfrontKmsLocations; + } + let kmsFailed = false; + return async.eachLimit(upfrontKmsLocations, MAX_PARALLEL_KMS_DECRYPT, (objectGetInfo, next) => { - if (!objectGetInfo.cipheredDataKey) { - return next(); - } const serverSideEncryption = { cryptoScheme: objectGetInfo.cryptoScheme, masterKeyId: objectGetInfo.masterKeyId, @@ -232,8 +247,14 @@ function objectGet(authInfo, request, returnTagCount, log, callback) { if (err) { log.error('cannot get decipher bundle from kms', { method: 'objectGet' }); + kmsFailed = true; return next(err); } + if (kmsFailed) { + // Inflight success when another KMS failed earlier + decipherBundle.decipher.destroy(); // destroy early + return next(); + } // eslint-disable-next-line no-param-reassign objectGetInfo.decipherStream = decipherBundle.decipher; return next(); diff --git a/package.json b/package.json index 9e6808f4bf..b7074747fe 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "s3", - "version": "7.10.61", + "version": "7.10.62", "description": "S3 connector", "main": "index.js", "engines": { diff --git a/tests/unit/api/apiUtils/destroyDecipherStreamsOnError.js b/tests/unit/api/apiUtils/destroyDecipherStreamsOnError.js new file mode 100644 index 0000000000..b78f754225 --- /dev/null +++ b/tests/unit/api/apiUtils/destroyDecipherStreamsOnError.js @@ -0,0 +1,74 @@ +const assert = require('assert'); +const { EventEmitter } = require('events'); +const sinon = require('sinon'); + +const destroyDecipherStreamsOnError = + require('../../../../lib/api/apiUtils/object/destroyDecipherStreamsOnError'); + +describe('destroyDecipherStreamsOnError', () => { + let locations; + let request; + let response; + + beforeEach(() => { + locations = [ + { decipherStream: { destroy: sinon.spy() } }, + { decipherStream: { destroy: sinon.spy() } }, + // KMS call failed or skipped for this part + {}, + ]; + request = { kmsDecipherLocations: locations }; + response = new EventEmitter(); + response.statusCode = 200; + response.writableFinished = false; + }); + + const assertDestroyed = destroyed => locations + .filter(location => location.decipherStream) + .forEach(location => assert.strictEqual( + location.decipherStream.destroy.calledOnce, destroyed)); + + it('should do nothing without decipher locations', () => { + destroyDecipherStreamsOnError(new Error(), {}, response); + assert.strictEqual(response.listenerCount('close'), 0); + }); + + it('should destroy decipher streams on objectGet error', () => { + destroyDecipherStreamsOnError(new Error(), request, response); + assertDestroyed(true); + assert.strictEqual(response.listenerCount('close'), 0); + }); + + it('should destroy decipher streams if the response is already closed', + () => { + response.isclosed = true; + destroyDecipherStreamsOnError(null, request, response); + assertDestroyed(true); + assert.strictEqual(response.listenerCount('close'), 0); + }); + + it('should not destroy decipher streams once a successful response ' + + 'is fully sent', () => { + // Because they are fully consumed by the response + destroyDecipherStreamsOnError(null, request, response); + assertDestroyed(false); + response.writableFinished = true; + response.emit('close'); + assertDestroyed(false); + }); + + it('should destroy decipher streams if the response is closed before ' + + 'being fully sent', () => { + destroyDecipherStreamsOnError(null, request, response); + response.emit('close'); + assertDestroyed(true); + }); + + it('should destroy decipher streams if an error response is sent', () => { + destroyDecipherStreamsOnError(null, request, response); + response.statusCode = 500; + response.writableFinished = true; + response.emit('close'); + assertDestroyed(true); + }); +}); diff --git a/tests/unit/api/objectGet.js b/tests/unit/api/objectGet.js index 19f47bf62e..5a7c781833 100644 --- a/tests/unit/api/objectGet.js +++ b/tests/unit/api/objectGet.js @@ -14,10 +14,12 @@ const DummyRequest = require('../DummyRequest'); const initiateMultipartUpload = require('../../../lib/api/initiateMultipartUpload'); const kms = require('../../../lib/kms/wrapper'); +const metadata = require('../../../lib/metadata/wrapper'); const objectPut = require('../../../lib/api/objectPut'); const objectGet = require('../../../lib/api/objectGet'); const objectPutPart = require('../../../lib/api/objectPutPart'); const changeObjectLock = require('../../utilities/objectLock-util'); +const requireWithEnv = require('../../utilities/requireWithEnv-util'); const log = new DummyRequestLogger(); const canonicalID = 'accessKey1'; @@ -395,6 +397,36 @@ describe('objectGet API', () => { }); }); + it('should get the data retrieval info of an unencrypted object ' + + 'with 10000 parts', done => { + // This test is for upfront KMS decryption, with async 2.5, eachLimit + // recurses on synchronous callbacks and overflows the stack. + // Skipping unencrypted parts before calling eachLimit fixes this issue + const locationCount = 10000; + async.waterfall([ + next => bucketPut(authInfo, testPutBucketRequest, log, + err => next(err)), + next => objectPut(authInfo, testPutObjectRequest, undefined, log, + err => next(err)), + next => metadata.getObjectMD(bucketName, objectName, {}, log, + next), + (objMD, next) => { + const location = objMD.location[0]; + // eslint-disable-next-line no-param-reassign + objMD.location = Array.from({ length: locationCount }, + (_, i) => Object.assign({}, location, + { start: i * location.size })); + metadata.putObjectMD(bucketName, objectName, objMD, {}, log, + err => next(err)); + }, + next => objectGet(authInfo, testGetRequest, false, log, next), + ], (err, dataGetInfo) => { + assert.ifError(err); + assert.strictEqual(dataGetInfo.length, locationCount); + done(); + }); + }); + describe('with an encrypted object', () => { const testPutEncryptedObjectRequest = () => new DummyRequest({ bucketName, @@ -407,6 +439,18 @@ describe('objectGet API', () => { parsedContentLength: 12, url: `/${bucketName}/${objectName}`, }, postBody); + // fresh request: objectGet sets kmsDecipherLocations on it + let getRequest; + + beforeEach(done => { + getRequest = Object.assign({}, testGetRequest); + async.series([ + next => bucketPut(authInfo, testPutBucketRequest, log, + err => next(err)), + next => objectPut(authInfo, testPutEncryptedObjectRequest(), + undefined, log, err => next(err)), + ], done); + }); afterEach(() => { sinon.restore(); @@ -415,22 +459,126 @@ describe('objectGet API', () => { it('should return KMS error before checking data', done => { const kmsError = errors.InternalError .customizeDescription('KMS key is not usable'); - async.waterfall([ - next => bucketPut(authInfo, testPutBucketRequest, log, - err => next(err)), - next => objectPut(authInfo, testPutEncryptedObjectRequest(), - undefined, log, err => next(err)), - ], err => { - assert.ifError(err); - const createDecipherBundleStub = sinon + const createDecipherStub = sinon + .stub(kms, 'createDecipherBundle') + .callsFake((sse, offset, log, cb) => cb(kmsError)); + const dataHeadSpy = sinon.spy(data, 'head'); + objectGet(authInfo, getRequest, false, log, err => { + assert.strictEqual(err, kmsError); + assert(createDecipherStub.calledOnce); + assert(dataHeadSpy.notCalled); + done(); + }); + }); + + describe('with many parts', () => { + const locationCount = 150; + const envName = 'GET_OBJECT_MAX_PARALLEL_KMS_DECRYPT'; + const objectGetPath = require.resolve('../../../lib/api/objectGet'); + let maxInFlight; + let createDecipherStub; + + beforeEach(done => { + let inFlight = 0; + maxInFlight = 0; + createDecipherStub = sinon .stub(kms, 'createDecipherBundle') - .callsFake((sse, offset, log, cb) => cb(kmsError)); - const dataHeadSpy = sinon.spy(data, 'head'); - objectGet(authInfo, testGetRequest, false, log, err => { + .callsFake((sse, offset, log, cb) => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + setImmediate(() => { + inFlight -= 1; + cb(null, { decipher: {} }); + }); + }); + metadata.getObjectMD(bucketName, objectName, {}, log, + (err, objMD) => { + assert.ifError(err); + const location = objMD.location[0]; + // eslint-disable-next-line no-param-reassign + objMD.location = Array.from({ length: locationCount }, + (_, i) => Object.assign({}, location, + { start: i * location.size })); + metadata.putObjectMD(bucketName, objectName, objMD, {}, + log, err => done(err)); + }); + }); + + [ + [undefined, 100], + ['-1', 100], + ['abc', 100], + ['7', 7], + ['1', 1], + ['150', 150], + ].forEach(([envValue, expectedLimit]) => { + it(`should limit concurrent KMS calls to ${expectedLimit} ` + + `when ${envName} is ${envValue}`, done => { + const objectGetWithEnv = requireWithEnv(objectGetPath, + { [envName]: envValue }); + objectGetWithEnv(authInfo, getRequest, false, + log, (err, locations) => { + assert.ifError(err); + assert.strictEqual( + createDecipherStub.callCount, locationCount); + assert.strictEqual(maxInFlight, expectedLimit); + assert(locations.every(l => l.decipherStream)); + assert.strictEqual( + getRequest.kmsDecipherLocations.length, + locations.length + ); + done(); + }); + }); + }); + + it(`should not call KMS before streaming when ${envName} is 0`, + done => { + const objectGetWithEnv = requireWithEnv(objectGetPath, + { [envName]: '0' }); + objectGetWithEnv(authInfo, getRequest, false, log, + (err, locations) => { + assert.ifError(err); + assert(createDecipherStub.notCalled); + assert(locations.every(l => !l.decipherStream)); + assert.strictEqual(getRequest.kmsDecipherLocations, + undefined); + done(); + }); + }); + + it('should return the first KMS error and destroy decipher ' + + 'streams of KMS calls still in flight', done => { + const kmsError = errors.InternalError + .customizeDescription('KMS key is not usable'); + const deciphers = []; + createDecipherStub.callsFake((sse, offset, log, cb) => { + if (createDecipherStub.callCount === 3) { + return setImmediate(cb, kmsError); + } + const decipher = { destroy: sinon.spy() }; + deciphers.push(decipher); + return setImmediate(cb, null, { decipher }); + }); + objectGet(authInfo, getRequest, false, log, err => { assert.strictEqual(err, kmsError); - assert(createDecipherBundleStub.calledOnce); - assert(dataHeadSpy.notCalled); - done(); + // once the KMS calls still in flight returned + setImmediate(() => { + const attached = getRequest.kmsDecipherLocations + .filter(l => l.decipherStream); + const destroyed = deciphers + .filter(d => d.destroy.calledOnce); + // first batch of 100 and the 2 calls started when 1 and 2 + // returned, none started after the error + assert.strictEqual(createDecipherStub.callCount, 102); + // calls 1 and 2 returned before the async error: attached, + // destroyed later by destroyDecipherStreamsOnError + assert.strictEqual(attached.length, 2); + // in flight at the error: calls 4 to 100 and the 2 calls + // started when 1 and 2 returned, destroyed and not attached + assert.strictEqual(destroyed.length, 99); + done(); + }); }); }); }); diff --git a/tests/utilities/requireWithEnv-util.js b/tests/utilities/requireWithEnv-util.js new file mode 100644 index 0000000000..4c77a904dd --- /dev/null +++ b/tests/utilities/requireWithEnv-util.js @@ -0,0 +1,38 @@ +function setEnv(name, value) { + if (value === undefined) { + delete process.env[name]; + } else { + process.env[name] = value; + } +} + +/** + * Require a module uncached, applying temporary env vars for load-time. + * Resets both env and require cache after loading. + * @param {string} modulePath - absolute module path, use require.resolve + * @param {object} env - env variables to set, an undefined value unsets it + * @return {*} - exports of the freshly loaded module + */ +function requireWithEnv(modulePath, env) { + const resolvedPath = require.resolve(modulePath); // ensure exact cache key + const cachedModule = require.cache[resolvedPath]; + const previousEnv = {}; + Object.keys(env).forEach(name => { + previousEnv[name] = process.env[name]; + setEnv(name, env[name]); + }); + delete require.cache[resolvedPath]; + try { + // eslint-disable-next-line global-require, import/no-dynamic-require + return require(resolvedPath); + } finally { + Object.keys(previousEnv).forEach(name => setEnv(name, previousEnv[name])); + if (cachedModule) { + require.cache[resolvedPath] = cachedModule; + } else { + delete require.cache[resolvedPath]; + } + } +} + +module.exports = requireWithEnv;