CLDSRV-1020: Improve KMS upfront decrypt with many parts

.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

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);
         });

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;

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();

package.json

@@ -1,6 +1,6 @@
 {
   "name": "s3",
-  "version": "7.10.61",
+  "version": "7.10.62",
   "description": "S3 connector",
   "main": "index.js",
   "engines": {

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);
+    });
+});