From c9e4bb6fa216a5b50cb1f0e608a7af17b5cefffa Mon Sep 17 00:00:00 2001 From: sylvain senechal Date: Mon, 17 Aug 2026 18:13:10 +0200 Subject: [PATCH] copy location task reads source location from s3 with STS Issue: BB-812 --- .../queueProcessor/QueueProcessor.js | 11 ++ .../replication/tasks/CopyLocationTask.js | 158 +++++++++++++-- lib/Config.js | 14 ++ lib/management/operatorBackend.js | 1 + .../unit/replication/CopyLocationTask.spec.js | 184 ++++++++++++++++++ 5 files changed, 347 insertions(+), 21 deletions(-) diff --git a/extensions/replication/queueProcessor/QueueProcessor.js b/extensions/replication/queueProcessor/QueueProcessor.js index 24edf7d054..305ec224dd 100644 --- a/extensions/replication/queueProcessor/QueueProcessor.js +++ b/extensions/replication/queueProcessor/QueueProcessor.js @@ -15,6 +15,7 @@ const RoundRobin = require('arsenal').network.RoundRobin; const BackbeatProducer = require('../../../lib/BackbeatProducer'); const BackbeatConsumer = require('../../../lib/BackbeatConsumer'); const VaultClientCache = require('../../../lib/clients/VaultClientCache'); +const CredentialsManager = require('../../../lib/credentials/CredentialsManager'); const QueueEntry = require('../../../lib/models/QueueEntry'); const TaskScheduler = require('../../../lib/tasks/TaskScheduler'); const { getTaskSchedulerQueueKey, @@ -229,6 +230,12 @@ class QueueProcessor extends EventEmitter { this.logger = new Logger( `Backbeat:Replication:QueueProcessor:${this.site}`); + this.assumedRoleCredentialsManager = new CredentialsManager( + 'replication-copy-location', this.logger); + this.assumedRoleS3Clients = {}; + this.assumedRoleHTTPAgent = new HttpAgent.Agent({ keepAlive: true }); + this.assumedRoleHTTPSAgent = new HttpsAgent.Agent({ keepAlive: true }); + // global variables if (sourceConfig.transport === 'https') { this.sourceHTTPAgent = new HttpsAgent.Agent({ @@ -690,6 +697,10 @@ class QueueProcessor extends EventEmitter { destHTTPAgent: this.destHTTPAgent, vaultclientCache: this.vaultclientCache, accountCredsCache: this.accountCredsCache, + assumedRoleCredentialsManager: this.assumedRoleCredentialsManager, + assumedRoleS3Clients: this.assumedRoleS3Clients, + assumedRoleHTTPAgent: this.assumedRoleHTTPAgent, + assumedRoleHTTPSAgent: this.assumedRoleHTTPSAgent, replicationStatusProducer: this.replicationStatusProducer, mProducer: this._mProducer, logger: this.logger, diff --git a/extensions/replication/tasks/CopyLocationTask.js b/extensions/replication/tasks/CopyLocationTask.js index b7d455e909..0fcb68890a 100644 --- a/extensions/replication/tasks/CopyLocationTask.js +++ b/extensions/replication/tasks/CopyLocationTask.js @@ -3,12 +3,14 @@ const { v4: uuid } = require('uuid'); const { errors, jsutil, models } = require('arsenal'); const { ObjectMD } = models; +const { S3Client: AwsS3Client, GetObjectCommand: AwsGetObjectCommand } = + require('@aws-sdk/client-s3'); const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy'); const BackbeatTask = require('../../../lib/tasks/BackbeatTask'); -const { +const { BackbeatRoutesClient, - GetObjectCommand, + GetObjectCommand: BackbeatRoutesGetObjectCommand, MultipleBackendPutObjectCommand, MultipleBackendInitiateMPUCommand, MultipleBackendPutMPUPartCommand, @@ -24,6 +26,8 @@ const { getAccountCredentials } = require('../../../lib/credentials/AccountCredentials'); const RoleCredentials = require('../../../lib/credentials/RoleCredentials'); +const config = require('../../../lib/Config'); +const { authTypeAssumeRole } = require('../../../lib/constants'); const { metricsExtension, metricsTypeQueued, metricsTypeCompleted } = require('../constants'); @@ -101,6 +105,117 @@ class CopyLocationTask extends BackbeatTask { .setSourceClient(log); } + /** + * Get a cached S3 client, authenticated with the assumed-role + * credentials for the role carried on a location part. + * @param {Object} locationConfig - the location's config + * @param {String} roleArn - the role ARN carried by the location part + * @param {Werelogs} log - the logger instance + * @return {AwsS3Client} the client + * @throws {ArsenalError} AccessDenied (retryable) if credentials + * could not be obtained for the role + */ + _getAssumedRoleS3Client(locationConfig, roleArn, log) { + const { details } = locationConfig; + const s3Endpoint = `${details.transport}://${details.servers[0]}`; + const cacheKey = `${s3Endpoint}::${roleArn}`; + if (this.assumedRoleS3Clients[cacheKey]) { + return this.assumedRoleS3Clients[cacheKey]; + } + const accountId = roleArn.split(':')[4]; + const roleName = roleArn.split(':role/')[1]; + const credentials = this.assumedRoleCredentialsManager.getCredentials({ + id: roleArn, + accountId, + authConfig: { + type: authTypeAssumeRole, + roleName, + }, + stsConfig: { + endpoint: `${details.transport}://${details.sts.host}:${details.sts.port}`, + credentials: { + accessKeyId: details.sts.accessKey, + secretAccessKey: details.sts.secretKey, + }, + }, + }); + if (!credentials) { + log.error('unable to obtain assumed-role credentials for source location', { + method: 'CopyLocationTask._getAssumedRoleS3Client', + roleArn, + endpoint: s3Endpoint, + }); + const err = errors.AccessDenied.customizeDescription( + `unable to assume role ${roleArn} for isCRR source location`); + err.retryable = true; + throw err; + } + const isHttps = details.transport === 'https'; + const client = new AwsS3Client({ + endpoint: s3Endpoint, + credentials: credentials.getCredentialsProvider(), + region: 'us-east-1', + forcePathStyle: true, + requestHandler: { + [isHttps ? 'httpsAgent' : 'httpAgent']: + isHttps ? this.assumedRoleHTTPSAgent : this.assumedRoleHTTPAgent, + requestTimeout: TIMEOUT_MS, + }, + maxAttempts: 1, + }); + client.middlewareStack.add(isRetryableMiddleware(), { + step: 'deserialize', + priority: 'high', + }); + this.assumedRoleS3Clients[cacheKey] = client; + return this.assumedRoleS3Clients[cacheKey]; + } + + /** + * Send a GetObject request for the object's data, + * reading either through Cloudserver's multiple-backend routes, + * or directly from a CRR source location's own S3 + * endpoint via an assumed role. + * @param {ActionQueueEntry} actionEntry - the action entry + * @param {ObjectMD} objMD - metadata object + * @param {Object} [range] - byte range to request, or undefined for the whole object + * @param {Werelogs} log - the logger instance + * @param {AbortController} abortController - abort controller for the GET request + * @return {Promise} resolves to the GetObject response + */ + async _sendGetObject(actionEntry, objMD, range, log, abortController) { + const locationConfig = config.getLocationConstraint(objMD.getDataStoreName()); + if (locationConfig?.isCRR === true) { + const locations = objMD.getLocation(); + const part = locations && locations[0]; + if (!part || !part.role) { + const err = errors.AccessDenied.customizeDescription( + 'missing role on location part for isCRR source location'); + err.retryable = true; + throw err; + } + const s3Client = this._getAssumedRoleS3Client(locationConfig, part.role, log); + const command = new AwsGetObjectCommand({ + Bucket: part.bucket, + Key: objMD.getKey(), + VersionId: part.dataStoreVersionId, + Range: range && `bytes=${range.start}-${range.end}`, + }); + return await s3Client.send(command, { abortSignal: abortController.signal }); + } + + const { bucket, key, version } = actionEntry.getAttribute('target'); + const command = new BackbeatRoutesGetObjectCommand({ + Bucket: bucket, + Key: key, + VersionId: version, + Range: range && `bytes=${range.start}-${range.end}`, + LocationConstraint: objMD.getDataStoreName(), + RequestUids: log.getSerializedUids(), + }); + return await this.backbeatClient.send(command, { abortSignal: abortController.signal }); + } + processQueueEntry(actionEntry, kafkaEntry, done) { const startTime = Date.now(); const log = this.logger.newRequestLogger(); @@ -237,16 +352,8 @@ class CopyLocationTask extends BackbeatTask { let sourceStreamAborted = false; let abortedByPut = false; const abortController = new AbortController(); - const { bucket, key, version } = actionEntry.getAttribute('target'); - const getObjectCommand = new GetObjectCommand({ - Bucket: bucket, - Key: key, - VersionId: version, - LocationConstraint: objMD.getDataStoreName(), - RequestUids: log.getSerializedUids(), - }); - return this.backbeatClient.send(getObjectCommand, { abortSignal: abortController.signal }) + return this._sendGetObject(actionEntry, objMD, undefined, log, abortController) .then(response => { const incomingMsg = response.Body; incomingMsg.on('error', err => { @@ -293,6 +400,14 @@ class CopyLocationTask extends BackbeatTask { actionEntry, objMD, size, incomingMsg, log, putDone); }) .catch(err => { + if (err.name === 'NoSuchVersion') { + log.info('source version no longer exists', Object.assign({ + method: 'CopyLocationTask._getAndPutObjectOnce', + error: err.message, + }, actionEntry.getLogInfo())); + return doneOnce(errors.InvalidObjectState.customizeDescription( + 'source version no longer exists')); + } if (err.$metadata?.httpStatusCode === 404) { log.error('the source object was not found', Object.assign({ method: 'CopyLocationTask._getAndPutObjectOnce', @@ -307,6 +422,7 @@ class CopyLocationTask extends BackbeatTask { method: 'CopyLocationTask._getAndPutObjectOnce', peer: this.sourceConfig.s3, error: err.message, + errorName: err.name, httpStatus: err.$metadata?.httpStatusCode, }, actionEntry.getLogInfo())); return doneOnce(err); @@ -418,20 +534,19 @@ class CopyLocationTask extends BackbeatTask { } const abortController = new AbortController(); - const { bucket, key, version } = actionEntry.getAttribute('target'); - const getObjectCommand = new GetObjectCommand({ - Bucket: bucket, - Key: key, - VersionId: version, - Range: range && `bytes=${range.start}-${range.end}`, - LocationConstraint: objMD.getDataStoreName(), - RequestUids: log.getSerializedUids(), - }); - return this.backbeatClient.send(getObjectCommand, { abortSignal: abortController.signal }) + return this._sendGetObject(actionEntry, objMD, range, log, abortController) .then(response => this._putMPUPart(actionEntry, objMD, response.Body, size, uploadId, partNumber, log, abortController, done)) .catch(err => { + if (err.name === 'NoSuchVersion') { + log.info('source version no longer exists', Object.assign({ + method: 'CopyLocationTask._getRangeAndPutMPUPartOnce', + error: err.message, + }, actionEntry.getLogInfo())); + return done(errors.InvalidObjectState.customizeDescription( + 'source version no longer exists')); + } if (err.$metadata?.httpStatusCode === 404) { return done(err); } @@ -439,6 +554,7 @@ class CopyLocationTask extends BackbeatTask { Object.assign({ method: 'CopyLocationTask._getRangeAndPutMPUPartOnce', error: err.message, + errorName: err.name, httpStatus: err.$metadata?.httpStatusCode, }, actionEntry.getLogInfo())); return done(err); diff --git a/lib/Config.js b/lib/Config.js index 57a9aaa880..58a2786a03 100644 --- a/lib/Config.js +++ b/lib/Config.js @@ -178,6 +178,7 @@ class Config extends EventEmitter { Object.assign(this, parsedConfig); this.transientLocations = {}; + this.locationConstraints = {}; this._setTimeOptions(); this._setLifecycleConductorOptions(); @@ -311,6 +312,19 @@ class Config extends EventEmitter { return this.transientLocations[locationName] || false; } + setLocationConstraints(locationConstraints) { + this.locationConstraints = locationConstraints; + } + + /** + * Get the raw location constraint config for a given location name + * @param {String} locationName - the location constraint name + * @return {Object|undefined} the location config, or undefined if unknown + */ + getLocationConstraint(locationName) { + return this.locationConstraints[locationName]; + } + getPublicInstanceId() { return this.publicInstanceId; } diff --git a/lib/management/operatorBackend.js b/lib/management/operatorBackend.js index f886899081..3273407220 100644 --- a/lib/management/operatorBackend.js +++ b/lib/management/operatorBackend.js @@ -78,6 +78,7 @@ function initManagement(params, done) { })); const locations = require('../../conf/locationConfig.json') || {}; + config.setLocationConstraints(locations); Object.keys(locations).forEach(locName => { config.setIsTransientLocation( locName, locations[locName].isTransient); diff --git a/tests/unit/replication/CopyLocationTask.spec.js b/tests/unit/replication/CopyLocationTask.spec.js index dcf9d84818..2c282406a8 100644 --- a/tests/unit/replication/CopyLocationTask.spec.js +++ b/tests/unit/replication/CopyLocationTask.spec.js @@ -299,4 +299,188 @@ describe('CopyLocationTask', () => { assert.strictEqual(task.retryParams.maxRetries, 13); }); }); + + describe('_sendGetObject', () => { + let task; + let config; + + beforeEach(() => { + config = require('../../../lib/Config'); + task = new CopyLocationTask({ + getStateVars: () => ({ + mProducer: { getProducer: () => {} }, + sourceConfig: { transport: 'http' }, + }), + }); + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should read through Cloudserver when the location is not isCRR', () => { + sinon.stub(config, 'getLocationConstraint').returns({ locationType: 'location-aws-s3-v1', isCRR: false }); + task.backbeatClient = { send: sinon.stub().resolves({ Body: 'stream' }) }; + + const entry = new ActionQueueEntry({ + target: { bucket: 'bucket', key: 'key', version: 'v1' }, + }); + const objMd = new ObjectMD(); + objMd.setDataStoreName('some-location'); + + return task._sendGetObject(entry, objMd, undefined, fakeLogger, new AbortController()) + .then(response => { + assert.deepStrictEqual(response, { Body: 'stream' }); + assert(task.backbeatClient.send.calledOnce); + const command = task.backbeatClient.send.firstCall.args[0]; + assert.strictEqual(command.input.Bucket, 'bucket'); + assert.strictEqual(command.input.Key, 'key'); + assert.strictEqual(command.input.VersionId, 'v1'); + assert.strictEqual(command.input.LocationConstraint, 'some-location'); + }); + }); + + it('should read directly from the CRR source location when isCRR', () => { + sinon.stub(config, 'getLocationConstraint').returns({ + locationType: 'location-scality-crr-v1', + isCRR: true, + details: { + servers: ['production.example.com:443'], + transport: 'https', + sts: { + host: 'sts.production.example.com', + port: '443', + accessKey: 'AK', + secretKey: 'SK', + }, + }, + }); + + const fakeS3Client = { send: sinon.stub().resolves({ Body: 'remote-stream' }) }; + sinon.stub(task, '_getAssumedRoleS3Client').returns(fakeS3Client); + + const entry = new ActionQueueEntry({ + target: { bucket: 'local-bucket', key: 'key', version: 'v1' }, + }); + const objMd = new ObjectMD(); + objMd.setDataStoreName('source-site'); + objMd.setKey('backups/vm001.vbk'); + objMd.setLocation([{ + key: 'backups/vm001.vbk', + size: 1048576, + start: 0, + dataStoreName: 'source-site', + dataStoreType: 'aws_s3', + dataStoreETag: '1:9b2cf535f27731c974343645a3985328', + dataStoreVersionId: 'aJdO95zrzY5BKLXf9GHFItC0d1CkQ0Ei', + bucket: 'backup-repo-01', + role: 'arn:aws:iam::123456789012:role/clean-room-read', + }]); + + return task._sendGetObject(entry, objMd, undefined, fakeLogger, new AbortController()) + .then(response => { + assert.deepStrictEqual(response, { Body: 'remote-stream' }); + assert(task._getAssumedRoleS3Client.calledOnce); + const [locationConfig, roleArn] = task._getAssumedRoleS3Client.firstCall.args; + assert.strictEqual(locationConfig.isCRR, true); + assert.strictEqual(roleArn, 'arn:aws:iam::123456789012:role/clean-room-read'); + assert(fakeS3Client.send.calledOnce); + const command = fakeS3Client.send.firstCall.args[0]; + assert.strictEqual(command.input.Bucket, 'backup-repo-01'); + assert.strictEqual(command.input.Key, 'backups/vm001.vbk'); + assert.strictEqual(command.input.VersionId, 'aJdO95zrzY5BKLXf9GHFItC0d1CkQ0Ei'); + }); + }); + + it('should reject without calling Cloudserver or the remote site when the role is missing', () => { + sinon.stub(config, 'getLocationConstraint').returns({ + locationType: 'location-scality-crr-v1', + isCRR: true, + details: {}, + }); + task.backbeatClient = { send: sinon.stub() }; + sinon.stub(task, '_getAssumedRoleS3Client'); + + const entry = new ActionQueueEntry({ target: {} }); + const objMd = new ObjectMD(); + objMd.setDataStoreName('source-site'); + objMd.setLocation([{ + key: 'k', + bucket: 'b', + dataStoreName: 'source-site', + // no role: owner absent from the ownerId->role map + }]); + + return task._sendGetObject(entry, objMd, undefined, fakeLogger, new AbortController()) + .then(() => assert.fail('expected rejection')) + .catch(err => { + assert(err.AccessDenied); + assert.strictEqual(err.retryable, true); + assert(task.backbeatClient.send.notCalled); + assert(task._getAssumedRoleS3Client.notCalled); + }); + }); + }); + + describe('_getAssumedRoleS3Client', () => { + let task; + const locationConfig = { + details: { + servers: ['production.example.com:443'], + transport: 'https', + sts: { + host: 'sts.production.example.com', + port: '443', + accessKey: 'AK', + secretKey: 'SK', + }, + }, + }; + const roleArn = 'arn:aws:iam::123456789012:role/clean-room-read'; + + beforeEach(() => { + task = new CopyLocationTask({ + getStateVars: () => ({ + mProducer: { getProducer: () => {} }, + sourceConfig: { transport: 'http' }, + assumedRoleCredentialsManager: { + getCredentials: sinon.stub().returns({ + getCredentialsProvider: () => async () => ({}), + }), + }, + assumedRoleS3Clients: {}, + }), + }); + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should cache and reuse the S3 client for the same endpoint and role', () => { + const client1 = task._getAssumedRoleS3Client(locationConfig, roleArn, fakeLogger); + const client2 = task._getAssumedRoleS3Client(locationConfig, roleArn, fakeLogger); + assert.strictEqual(client1, client2); + assert(task.assumedRoleCredentialsManager.getCredentials.calledOnce); + }); + + it('should log and throw a retryable AccessDenied when credentials cannot be obtained', () => { + task.assumedRoleCredentialsManager.getCredentials.returns(null); + const logSpy = sinon.spy(fakeLogger, 'error'); + + assert.throws( + () => task._getAssumedRoleS3Client(locationConfig, roleArn, fakeLogger), + err => err.AccessDenied && err.retryable === true); + assert(logSpy.calledOnce); + }); + + it('should keep the full role name, including any path, when the role ARN has one', () => { + const pathedRoleArn = 'arn:aws:iam::123456789012:role/service-role/clean-room-read'; + + task._getAssumedRoleS3Client(locationConfig, pathedRoleArn, fakeLogger); + + const params = task.assumedRoleCredentialsManager.getCredentials.firstCall.args[0]; + assert.strictEqual(params.authConfig.roleName, 'service-role/clean-room-read'); + }); + }); });