initiateMultipartUpload.js

const { v4: uuidv4 } = require('uuid');
const { errors, errorInstances, s3middleware, s3routes } = require('arsenal');
const { validateObjectKeyLength } = s3routes.routesUtils;
const async = require('async');
const getMetaHeaders = s3middleware.userMetadata.getMetaHeaders;
const convertToXml = s3middleware.convertToXml;
const { pushMetric } = require('../utapi/utilities');
const collectCorsHeaders = require('../utilities/collectCorsHeaders');
const { hasNonPrintables } = require('../utilities/stringChecks');
const { cleanUpBucket } = require('./apiUtils/bucket/bucketCreation');
const constants = require('../../constants');
const services = require('../services');
const { standardMetadataValidateBucketAndObj } = require('../metadata/metadataUtils');
const locationConstraintCheck
    = require('./apiUtils/object/locationConstraintCheck');
const validateWebsiteHeader = require('./apiUtils/object/websiteServing')
    .validateWebsiteHeader;
const monitoring = require('../utilities/monitoringHandler');
const { data } = require('../data/wrapper');
const applyZenkoUserMD = require('./apiUtils/object/applyZenkoUserMD');
const { validateHeaders, compareObjectLockInformation } =
    require('./apiUtils/object/objectLockHelpers');
const { getObjectSSEConfiguration } = require('./apiUtils/bucket/bucketEncryption');
const { setExpirationHeaders } = require('./apiUtils/object/expirationHeaders');
const { setSSEHeaders } = require('./apiUtils/object/sseHeaders');
const { updateEncryption } = require('./apiUtils/bucket/updateEncryption');
const { getChecksumDataFromMPUHeaders, arsenalErrorFromChecksumError } =
    require('./apiUtils/integrity/validateChecksums');
const { config } = require('../Config');
const kms = require('../kms/wrapper');

/*
Sample xml response:
<?xml version='1.0' encoding='UTF-8'?>
<InitiateMultipartUploadResult xmlns='http://s3.amazonaws.com/doc/2006-03-01/'>
  <Bucket>example-bucket</Bucket>
  <Key>example-object</Key>
  <UploadId>VXBsb2FkIElEIGZvciA2aWWpbmcncyBteS
  1tb3ZpZS5tMnRzIHVwbG9hZA</UploadId>
</InitiateMultipartUploadResult>
*/

/**
 * Initiate multipart upload returning xml that includes the UploadId
 * for the multipart upload
 * @param  {AuthInfo} authInfo - Instance of AuthInfo class with requester's
 *                               info
 * @param  {request} request - request object given by router,
 * includes normalized headers
 * @param  {object} log - the log request
 * @param  {function} callback - final callback to call with the result
 * @return {undefined} calls callback from router
 * with err and result as arguments
 */
function initiateMultipartUpload(authInfo, request, log, callback) {
    log.debug('processing request', { method: 'initiateMultipartUpload' });
    const bucketName = request.bucketName;
    const objectKey = request.objectKey;

    if (hasNonPrintables(objectKey)) {
        return callback(errorInstances.InvalidInput.customizeDescription(
            'object keys cannot contain non-printable characters',
        ));
    }

    const keyLengthError = validateObjectKeyLength(objectKey, config.objectKeyByteLimit);
    if (keyLengthError) {
        return callback(keyLengthError);
    }

    // Note that we are using the string set forth in constants.js
    // to split components in the storage
    // of each MPU.  AWS does not restrict characters in object keys so
    // there is the possiblity that the chosen splitter will occur in the object
    // name itself. To prevent this, we are restricting the creation of a
    // multipart upload object with a key containing the splitter.
    const websiteRedirectHeader =
        request.headers['x-amz-website-redirect-location'];
    if (request.headers['x-amz-storage-class'] &&
        !constants.validStorageClasses.includes(request.headers['x-amz-storage-class'])) {
        log.trace('invalid storage-class header');
        monitoring.promMetrics('PUT', bucketName,
            errorInstances.InvalidStorageClass.code, 'initiateMultipartUpload');
        return callback(errors.InvalidStorageClass);
    }
    if (!validateWebsiteHeader(websiteRedirectHeader)) {
        const err = errors.InvalidRedirectLocation;
        log.debug('invalid x-amz-website-redirect-location' +
            `value ${websiteRedirectHeader}`, { error: err });
        return callback(err);
    }
    const checksumConfig = getChecksumDataFromMPUHeaders(request.headers);
    if (checksumConfig.error) {
        const checksumErr = arsenalErrorFromChecksumError(checksumConfig);
        log.debug('checksum header validation failed', { error: checksumErr, method: 'initiateMultipartUpload' });
        monitoring.promMetrics('PUT', bucketName, checksumErr.code, 'initiateMultipartUpload');
        return callback(checksumErr);
    }
    const metaHeaders = getMetaHeaders(request.headers);
    if (metaHeaders instanceof Error) {
        log.debug('user metadata validation failed', {
            error: metaHeaders,
            method: 'initiateMultipartUpload',
        });
        return process.nextTick(() => callback(metaHeaders));
    }
    // if the request occurs within a Zenko deployment, we place a user-metadata
    // field on the object
    applyZenkoUserMD(metaHeaders);

    // TODO: Add this as a utility function for all object put requests
    // but after authentication so that string to sign is not impacted
    // This is GH Issue#89
    // TODO: remove in CLDSRV-639
    const storageClassOptions =
        ['standard', 'standard_ia', 'reduced_redundancy'];
    let storageClass = 'STANDARD';
    if (storageClassOptions.indexOf(request.headers['x-amz-storage-class']) > -1) {
        storageClass = request.headers['x-amz-storage-class']
            .toUpperCase();
    }
    const metadataValParams = {
        objectKey,
        authInfo,
        bucketName,
        // Required permissions for this action are same as objectPut
        requestType: request.apiMethods || 'initiateMultipartUpload',
        request,
    };
    const accountCanonicalID = authInfo.getCanonicalID();
    let initiatorID = accountCanonicalID;
    let initiatorDisplayName = authInfo.getAccountDisplayName();
    if (authInfo.isRequesterAnIAMUser()) {
        initiatorID = authInfo.getArn();
        initiatorDisplayName = authInfo.getIAMdisplayName();
    }

    const metadataStoreParams = {
        objectKey,
        storageClass,
        metaHeaders,
        eventualStorageBucket: bucketName,
        headers: request.headers,
        // The ownerID should be the account canonicalID.
        ownerID: accountCanonicalID,
        ownerDisplayName: authInfo.getAccountDisplayName(),
        // If initiator is an IAM user, the initiatorID is the ARN.
        // Otherwise, it is the same as the ownerID
        // (the account canonicalID)
        initiatorID,
        // If initiator is an IAM user, the initiatorDisplayName is the
        // IAM user's displayname.
        // Otherwise, it is the same as the ownerDisplayName.
        initiatorDisplayName,
        splitter: constants.splitter,
    };
    metadataStoreParams.checksumAlgorithm = checksumConfig.algorithm;
    metadataStoreParams.checksumType = checksumConfig.type;
    metadataStoreParams.checksumIsDefault = checksumConfig.isDefault;
    const tagging = request.headers['x-amz-tagging'];
    if (tagging) {
        metadataStoreParams.tagging = tagging;
    }

    function _getMPUBucket(destinationBucket, log, corsHeaders, uploadId, cipherBundle, locConstraint, callback) {
        const xmlParams = {
            bucketName,
            objectKey,
            uploadId,
        };
        const xml = convertToXml('initiateMultipartUpload', xmlParams);
        metadataStoreParams.uploadId = uploadId;

        services.getMPUBucket(destinationBucket, bucketName, log,
            (err, MPUbucket) => {
                if (err) {
                    log.trace('error getting MPUbucket', {
                        error: err,
                    });
                    return callback(err);
                }
                // BACKWARD: Remove to remove the old splitter
                if (MPUbucket.getMdBucketModelVersion() < 2) {
                    metadataStoreParams.splitter = constants.oldSplitter;
                }
                return services.metadataStoreMPObject(MPUbucket.getName(),
                    cipherBundle, metadataStoreParams,
                    log, (err, mpuMD) => {
                        if (err) {
                            log.trace('error storing multipart object', {
                                error: err,
                            });
                            monitoring.promMetrics('PUT', bucketName, err.code,
                                'initiateMultipartUpload');
                            return callback(err, null, corsHeaders);
                        }
                        log.addDefaultFields({ uploadId });
                        log.trace('successfully initiated mpu');
                        pushMetric('initiateMultipartUpload', log, {
                            authInfo,
                            bucket: bucketName,
                            keys: [objectKey],
                            location: locConstraint,
                        });

                        // TODO: rename corsHeaders to headers
                        setExpirationHeaders(corsHeaders, {
                            lifecycleConfig: destinationBucket.getLifecycleConfiguration(),
                            mpuParams: {
                                key: mpuMD.key,
                                date: mpuMD.initiated,
                            },
                        });

                        setSSEHeaders(corsHeaders,
                            mpuMD['x-amz-server-side-encryption'],
                            mpuMD['x-amz-server-side-encryption-aws-kms-key-id']);

                        // Only respond the headers if the user sent them
                        if (!checksumConfig.isDefault) {
                            // eslint-disable-next-line no-param-reassign
                            corsHeaders['x-amz-checksum-algorithm'] = checksumConfig.algorithm.toUpperCase();
                            // eslint-disable-next-line no-param-reassign
                            corsHeaders['x-amz-checksum-type'] = checksumConfig.type;
                        }

                        monitoring.promMetrics('PUT', bucketName, '200',
                            'initiateMultipartUpload');
                        return callback(null, xml, corsHeaders);
                    });
            });
    }

    function _storetheMPObject(destinationBucket, corsHeaders, serverSideEncryption) {
        let cipherBundle = null;
        if (serverSideEncryption) {
            const { algorithm, configuredMasterKeyId, masterKeyId } = serverSideEncryption;
            if (configuredMasterKeyId) {
                log.debug('using user configured kms master key id');
            }
            cipherBundle = {
                algorithm,
                masterKeyId: configuredMasterKeyId || masterKeyId,
            };
        }
        const backendInfoObj = locationConstraintCheck(request, null,
            destinationBucket, log);
        if (backendInfoObj.err) {
            return process.nextTick(() => {
                callback(backendInfoObj.err);
            });
        }
        const locConstraint = backendInfoObj.controllingLC;
        metadataStoreParams.controllingLocationConstraint = locConstraint;
        metadataStoreParams.dataStoreName = locConstraint;

        if (request.headers) {
            const objectLockValError =
                validateHeaders(destinationBucket, request.headers, log);
            if (objectLockValError) {
                return callback(objectLockValError);
            }
        }
        const defaultRetention = destinationBucket.getObjectLockConfiguration();
        const finalObjectLockInfo =
            compareObjectLockInformation(request.headers, defaultRetention);

        if (finalObjectLockInfo.retentionInfo) {
            metadataStoreParams.retentionMode =
                finalObjectLockInfo.retentionInfo.mode;
            metadataStoreParams.retentionDate =
                finalObjectLockInfo.retentionInfo.date;
        }
        if (finalObjectLockInfo.legalHold) {
            metadataStoreParams.legalHold = finalObjectLockInfo.legalHold;
        }

        let uploadId;
        const mpuInfo = {
            objectKey,
            metaHeaders,
            bucketName,
            locConstraint,
            destinationBucket,
            tagging,
        };

        const putVersionId = request.headers['x-scal-s3-version-id'];
        const isPutVersion = putVersionId || putVersionId === '';
        if (isPutVersion &&
            locConstraint === destinationBucket.getLocationConstraint() &&
            destinationBucket.isIngestionBucket()) {
            // When restoring to OOB bucket, we cannot force the versionId of the object written to the
            // backend, and it is thus not match the versionId of the ingested object. Thus we add extra
            // user metadata to allow OOB to allow ingestion processor to "match" the (new) restored
            // object with the existing ingested object.
            mpuInfo.metaHeaders['x-amz-meta-scal-version-id'] = putVersionId;
        }

        return data.initiateMPU(mpuInfo, websiteRedirectHeader, log,
            (err, dataBackendResObj, isVersionedObj) => {
                // will return as true and a custom error if external backend does
                // not support versioned objects
                if (isVersionedObj) {
                    monitoring.promMetrics('PUT', bucketName, 501,
                        'initiateMultipartUpload');
                    return callback(err);
                }
                if (err) {
                    monitoring.promMetrics('PUT', bucketName, err.code,
                        'initiateMultipartUpload');
                    return callback(err);
                }
                // if mpu not handled externally, dataBackendResObj will be null
                if (dataBackendResObj) {
                    uploadId = dataBackendResObj.UploadId;
                } else {
                    // Generate uniqueID without dashes so routing not messed up
                    uploadId = uuidv4().replace(/-/g, '');
                }
                return _getMPUBucket(destinationBucket, log, corsHeaders,
                    uploadId, cipherBundle, locConstraint, callback);
            });
    }

    async.waterfall([
        next => standardMetadataValidateBucketAndObj(metadataValParams, request.actionImplicitDenies, log,
            (error, destinationBucket, destObjMD) =>
                updateEncryption(error, destinationBucket, destObjMD, objectKey, log, { skipObject: true },
                    (error, destinationBucket) => {
                        const corsHeaders = collectCorsHeaders(
                            request.headers.origin, request.method, destinationBucket);
                        if (error) {
                            log.debug('error processing request', {
                                error,
                                method: 'metadataValidateBucketAndObj',
                            });
                            monitoring.promMetrics('PUT', bucketName, error.code, 'initiateMultipartUpload');
                            return next(error, corsHeaders);
                        }
                        return next(null, corsHeaders, destinationBucket);
                    })),
        (corsHeaders, destinationBucket, next) => {
            if (destinationBucket.hasDeletedFlag() && accountCanonicalID !== destinationBucket.getOwner()) {
                log.trace('deleted flag on bucket and request from non-owner account');
                monitoring.promMetrics('PUT', bucketName, 404, 'initiateMultipartUpload');
                return next(errors.NoSuchBucket, corsHeaders);
            }
            if (destinationBucket.hasTransientFlag() || destinationBucket.hasDeletedFlag()) {
                log.trace('transient or deleted flag so cleaning up bucket');
                return cleanUpBucket(
                    destinationBucket,
                    accountCanonicalID,
                    log,
                    error => {
                        if (error) {
                            log.debug('error cleaning up bucket with flag',
                                {
                                    error,
                                    transientFlag: destinationBucket.hasTransientFlag(),
                                    deletedFlag: destinationBucket.hasDeletedFlag(),
                                });
                            // To avoid confusing user with error
                            // from cleaning up
                            // bucket return InternalError
                            monitoring.promMetrics('PUT', bucketName, 500, 'initiateMultipartUpload');
                            return next(errors.InternalError, corsHeaders);
                        }
                        return next(null, corsHeaders, destinationBucket);
                    });
            }
            return next(null, corsHeaders, destinationBucket);
        },
        (corsHeaders, destinationBucket, next) =>
            getObjectSSEConfiguration(
                request.headers,
                destinationBucket,
                log,
                (error, objectSSEConfig) => {
                    if (error) {
                        log.error('error fetching server-side encryption config', {
                            error,
                            method: 'getObjectSSEConfiguration',
                        });
                        return next(error, corsHeaders);
                    }
                    return next(null, corsHeaders, destinationBucket, objectSSEConfig);
                }
            ),
        // If SSE configured, test kms key encryption access, but ignore cipher bundle
        (corsHeaders, destinationBucket, objectSSEConfig, next) => {
            if (objectSSEConfig) {
                return kms.createCipherBundle(objectSSEConfig, log,
                    err => next(err, corsHeaders, destinationBucket, objectSSEConfig));
            }
            return next(null, corsHeaders, destinationBucket, objectSSEConfig);
        },
    ],
        (error, corsHeaders, destinationBucket, objectSSEConfig) => {
            if (error) {
                return callback(error, null, corsHeaders);
            }
            return _storetheMPObject(destinationBucket, corsHeaders, objectSSEConfig);
        }
    );
    return undefined;
}

module.exports = initiateMultipartUpload;