/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.proxy.s3;

import alluxio.AlluxioURI;
import alluxio.client.WriteType;
import alluxio.client.file.FileSystem;
import alluxio.client.file.URIStatus;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AccessControlException;
import alluxio.exception.AlluxioException;
import alluxio.exception.DirectoryNotEmptyException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.FileAlreadyExistsException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.grpc.Bits;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.PMode;
import alluxio.grpc.SetAttributePOptions;
import alluxio.grpc.WritePType;
import alluxio.grpc.XAttrPropagationStrategy;
import alluxio.proto.journal.File;
import alluxio.s3.S3AuditContext;
import alluxio.s3.S3Constants;
import alluxio.s3.S3ErrorCode;
import alluxio.s3.S3ErrorResponse;
import alluxio.s3.S3Exception;
import alluxio.s3.TaggingData;
import alluxio.s3.auth.Authenticator;
import alluxio.s3.auth.AwsAuthInfo;
import alluxio.s3.signature.AwsSignatureProcessor;
import alluxio.security.User;
import alluxio.security.authentication.AuthType;
import alluxio.security.authentication.AuthenticatedClientUser;
import alluxio.security.user.ServerUserState;
import alluxio.util.SecurityUtils;
import alluxio.util.ThreadUtils;

import com.fasterxml.jackson.dataformat.xml.XmlMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.Cache;
import com.google.common.util.concurrent.RateLimiter;
import com.google.protobuf.ByteString;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
import java.util.regex.Pattern;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.security.auth.Subject;
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.container.ContainerRequestContext;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;

/**
 * Utilities for handling S3 REST calls.
 */
public final class S3RestUtils {
  private static final Logger LOG = LoggerFactory.getLogger(S3RestUtils.class);

  public static final String ALLUXIO_USER_HEADER = "Alluxio-Fs-User";

  public static final String MULTIPART_UPLOADS_METADATA_DIR = AlluxioURI.SEPARATOR
      + S3Constants.S3_METADATA_ROOT_DIR + AlluxioURI.SEPARATOR
      + S3Constants.S3_METADATA_UPLOADS_DIR;

  /**
   * Calls the given {@link S3RestUtils.RestCallable} and handles any exceptions thrown.
   *
   * @param <T> the return type of the callable
   * @param resource the resource (bucket or object) to be operated on
   * @param callable the callable to call
   * @return the response object
   */
  public static <T> Response call(String resource, S3RestUtils.RestCallable<T> callable) {
    try {
      // TODO(cc): reconsider how to enable authentication
      if (SecurityUtils.isSecurityEnabled(Configuration.global())
              && AuthenticatedClientUser.get(Configuration.global()) == null) {
        AuthenticatedClientUser.set(ServerUserState.global().getUser().getName());
      }
    } catch (IOException e) {
      LOG.warn("Failed to set AuthenticatedClientUser in REST service handler: {}", e.toString());
      return S3ErrorResponse.createErrorResponse(new S3Exception(
          e, resource, S3ErrorCode.INTERNAL_ERROR), resource);
    }

    try {
      T result = callable.call();
      if (result == null) {
        return Response.ok().build();
      }
      if (result instanceof Response) {
        return (Response) result;
      }
      if (result instanceof Response.Status) {
        switch ((Response.Status) result) {
          case OK:
            return Response.ok().build();
          case ACCEPTED:
            return Response.accepted().build();
          case NO_CONTENT:
            return Response.noContent().build();
          default:
            return S3ErrorResponse.createErrorResponse(new S3Exception(
                "Response status is invalid", resource, S3ErrorCode.INTERNAL_ERROR), resource);
        }
      }
      // Need to explicitly encode the string as XML because Jackson will not do it automatically.
      XmlMapper mapper = new XmlMapper();
      return Response.ok(mapper.writeValueAsString(result)).build();
    } catch (Exception e) {
      String errOutputMsg = e.getMessage();
      if (StringUtils.isEmpty(errOutputMsg)) {
        errOutputMsg = ThreadUtils.formatStackTrace(e);
      }
      LOG.warn("Error invoking REST endpoint for {}:\n{}", resource, errOutputMsg);
      return S3ErrorResponse.createErrorResponse(e, resource);
    }
  }

  /**
   * An interface representing a callable.
   *
   * @param <T> the return type of the callable
   */
  public interface RestCallable<T> {
    /**
     * The REST endpoint implementation.
     *
     * @return the return value from the callable
     */
    T call() throws S3Exception;
  }

  /**
   * @param bucketPath the bucket path like "/bucket", "/mount/point/bucket"
   * @param objectKey  the object key like "img/2017/9/1/s3.jpg"
   * @param uploadId   the upload ID
   * @return the temporary directory used to hold parts of the object during multipart uploads
   */
  public static String getMultipartTemporaryDirForObject(
      String bucketPath, String objectKey, String uploadId) {
    return bucketPath + AlluxioURI.SEPARATOR + objectKey
        + "_" + uploadId;
  }

  /**
   * @param uploadId the upload ID
   * @return the Alluxio UFS filepath containing the metadata for this upload
   */
  public static String getMultipartMetaFilepathForUploadId(String uploadId) {
    return MULTIPART_UPLOADS_METADATA_DIR + AlluxioURI.SEPARATOR + uploadId;
  }

  /**
   * Initiate the S3 API metadata directories.
   * @param fileSystem
   * @throws IOException
   * @throws AlluxioException
   */
  public static void initMultipartUploadsMetadataDir(FileSystem fileSystem)
      throws IOException, AlluxioException {
    // Initiate the S3 API metadata directories
    if (!fileSystem.exists(new AlluxioURI(MULTIPART_UPLOADS_METADATA_DIR))) {
      try {
        fileSystem.createDirectory(new AlluxioURI(MULTIPART_UPLOADS_METADATA_DIR),
            CreateDirectoryPOptions.newBuilder()
                .setRecursive(true)
                .setMode(PMode.newBuilder()
                    .setOwnerBits(Bits.ALL).setGroupBits(Bits.ALL)
                    .setOtherBits(Bits.NONE)
                    .build())
                .setWriteType(S3RestUtils.getS3WriteType())
                .setXattrPropStrat(XAttrPropagationStrategy.LEAF_NODE)
                .build());
      } catch (FileAlreadyExistsException e) {
        LOG.warn("Multipart uploads metadata dir {} already exists.",
            MULTIPART_UPLOADS_METADATA_DIR);
      }
    }
  }

  /**
   * @param epoch the milliseconds from the epoch
   * @return the string representation of the epoch in the S3 date format
   */
  public static String toS3Date(long epoch) {
    final DateFormat s3DateFormat = new SimpleDateFormat(S3Constants.S3_DATE_FORMAT_REGEXP);
    return s3DateFormat.format(new Date(epoch));
  }

  /**
   * Format bucket path.
   *
   * @param bucketPath bucket path
   * @return bucket path after format
   */
  public static String parsePath(String bucketPath) {
    // Normalize the bucket by replacing ":" with "/"
    return bucketPath.replace(S3Constants.BUCKET_SEPARATOR, AlluxioURI.SEPARATOR);
  }

  /**
   * Convert an exception to instance of {@link S3Exception}.
   *
   * @param exception Exception thrown when process s3 object rest request
   * @param resource complete bucket path
   * @return instance of {@link S3Exception}
   */
  public static S3Exception toBucketS3Exception(Exception exception, String resource) {
    try {
      throw exception;
    } catch (S3Exception e) {
      e.setResource(resource);
      return e;
    } catch (DirectoryNotEmptyException e) {
      return new S3Exception(e, resource, S3ErrorCode.BUCKET_NOT_EMPTY);
    } catch (FileAlreadyExistsException e) {
      return new S3Exception(e, resource, S3ErrorCode.BUCKET_ALREADY_EXISTS);
    } catch (FileDoesNotExistException e) {
      return new S3Exception(e, resource, S3ErrorCode.NO_SUCH_BUCKET);
    } catch (InvalidPathException e) {
      return new S3Exception(e, resource, S3ErrorCode.INVALID_BUCKET_NAME);
    } catch (AccessControlException e) {
      return new S3Exception(e, resource, S3ErrorCode.ACCESS_DENIED_ERROR);
    } catch (Exception e) {
      return new S3Exception(e, resource, S3ErrorCode.INTERNAL_ERROR);
    }
  }

  /**
   * Convert an exception to instance of {@link S3Exception}.
   *
   * @param exception Exception thrown when process s3 object rest request
   * @param resource complete bucket path
   * @param auditContext the audit context for exception
   * @return instance of {@link S3Exception}
   */
  public static S3Exception toBucketS3Exception(Exception exception, String resource,
                                                @Nonnull S3AuditContext auditContext) {
    if (exception instanceof AccessControlException) {
      auditContext.setAllowed(false);
    }
    auditContext.setSucceeded(false);
    return toBucketS3Exception(exception, resource);
  }

  /**
   * Convert an exception to instance of {@link S3Exception}.
   *
   * @param exception Exception thrown when process s3 object rest request
   * @param resource object complete path
   * @return instance of {@link S3Exception}
   */
  public static S3Exception toObjectS3Exception(Exception exception, String resource) {
    try {
      throw exception;
    } catch (S3Exception e) {
      e.setResource(resource);
      return e;
    } catch (DirectoryNotEmptyException e) {
      return new S3Exception(e, resource, S3ErrorCode.PRECONDITION_FAILED);
    } catch (FileDoesNotExistException e) {
      if (Pattern.matches(ExceptionMessage.BUCKET_DOES_NOT_EXIST.getMessage(".*"),
          e.getMessage())) {
        return new S3Exception(e, resource, S3ErrorCode.NO_SUCH_BUCKET);
      }
      return new S3Exception(e, resource, S3ErrorCode.NO_SUCH_KEY);
    } catch (AccessControlException e) {
      return new S3Exception(e, resource, S3ErrorCode.ACCESS_DENIED_ERROR);
    } catch (Exception e) {
      return new S3Exception(e, resource, S3ErrorCode.INTERNAL_ERROR);
    }
  }

  /**
   * Convert an exception to instance of {@link S3Exception}.
   *
   * @param exception Exception thrown when process s3 object rest request
   * @param resource object complete path
   * @param auditContext the audit context for exception
   * @return instance of {@link S3Exception}
   */
  public static S3Exception toObjectS3Exception(Exception exception, String resource,
                                                @Nonnull S3AuditContext auditContext) {
    if (exception instanceof AccessControlException) {
      auditContext.setAllowed(false);
    }
    auditContext.setSucceeded(false);
    return toObjectS3Exception(exception, resource);
  }

  /**
   * Check if a path in alluxio is a directory.
   *
   * @param fs instance of {@link FileSystem}
   * @param bucketPath bucket complete path
   * @param auditContext the audit context for exception
   */
  public static void checkPathIsAlluxioDirectory(FileSystem fs, String bucketPath,
                                                 @Nullable S3AuditContext auditContext)
      throws S3Exception {
    try {
      URIStatus status = fs.getStatus(new AlluxioURI(bucketPath));
      if (!status.isFolder()) {
        throw new FileDoesNotExistException(
            ExceptionMessage.BUCKET_DOES_NOT_EXIST.getMessage(bucketPath));
      }
    } catch (Exception e) {
      if (auditContext != null) {
        throw toBucketS3Exception(e, bucketPath, auditContext);
      }
      throw toBucketS3Exception(e, bucketPath);
    }
  }

  /**
   * Check if a path in alluxio is a directory.
   *
   * @param fs instance of {@link FileSystem}
   * @param bucketPath bucket complete path
   * @param auditContext the audit context for exception
   * @param bucketPathCache cache the bucket path for a certain time period
   */
  public static void checkPathIsAlluxioDirectory(FileSystem fs, String bucketPath,
                                                 @Nullable S3AuditContext auditContext,
                                                 Cache<String, Boolean> bucketPathCache)
      throws S3Exception {
    if (Boolean.TRUE.equals(bucketPathCache.getIfPresent(bucketPath))) {
      return;
    }
    checkPathIsAlluxioDirectory(fs, bucketPath, auditContext);
    bucketPathCache.put(bucketPath, true);
  }

  /**
   * Fetches and returns the corresponding {@link URIStatus} for both
   * the multipart upload temp directory and the Alluxio S3 metadata file.
   * @param metaFs instance of {@link FileSystem} - used for metadata operations
   * @param userFs instance of {@link FileSystem} - under the scope of a user agent
   * @param multipartTempDirUri multipart upload tmp directory URI
   * @param uploadId multipart upload Id
   * @return a list of file statuses:
   *         - first, the multipart upload temp directory
   *         - second, the metadata file
   */
  public static List<URIStatus> checkStatusesForUploadId(
      FileSystem metaFs, FileSystem userFs, AlluxioURI multipartTempDirUri, String uploadId)
      throws AlluxioException, IOException {
    // Verify the multipart upload dir exists and is a folder
    URIStatus multipartTempDirStatus = userFs.getStatus(multipartTempDirUri);
    if (!multipartTempDirStatus.isFolder()) {
      //TODO(czhu): determine intended behavior in this edge-case
      throw new RuntimeException(
          "Alluxio UFS contains a file at intended multipart-upload directory path: "
              + multipartTempDirUri);
    }

    // Verify the multipart upload meta file exists and matches the file ID
    final AlluxioURI metaUri = new AlluxioURI(
        S3RestUtils.getMultipartMetaFilepathForUploadId(uploadId));
    URIStatus metaStatus = metaFs.getStatus(metaUri);
    if (metaStatus.getXAttr() == null
        || !metaStatus.getXAttr().containsKey(S3Constants.UPLOADS_FILE_ID_XATTR_KEY)) {
      //TODO(czhu): determine intended behavior in this edge-case
      throw new RuntimeException(
          "Alluxio is missing multipart-upload metadata for upload ID: " + uploadId);
    }
    if (Long.parseLong(
        ByteString.copyFrom(metaStatus.getXAttr().get(S3Constants.UPLOADS_FILE_ID_XATTR_KEY))
            .toStringUtf8()) != multipartTempDirStatus.getFileId()) {
      throw new RuntimeException(
          "Alluxio mismatched file ID for multipart-upload with upload ID: " + uploadId);
    }
    return new ArrayList<>(Arrays.asList(multipartTempDirStatus, metaStatus));
  }

  /**
   * Delete an existing key.
   *
   * @param fs instance of {@link FileSystem}
   * @param objectURI the key uri
   */
  public static void deleteExistObject(final FileSystem fs, AlluxioURI objectURI)
      throws IOException, AlluxioException {
    deleteExistObject(fs, objectURI, false);
  }

  /**
   * Delete an existing key.
   *
   * @param fs instance of {@link FileSystem}
   * @param objectURI the key uri
   * @param recursive if delete option is recursive
   */
  public static void deleteExistObject(final FileSystem fs, AlluxioURI objectURI, Boolean recursive)
      throws IOException, AlluxioException {
    if (fs.exists(objectURI)) {
      if (recursive) {
        fs.delete(objectURI, DeletePOptions.newBuilder().setRecursive(true).build());
      } else {
        fs.delete(objectURI);
      }
      LOG.info("Remove exist object: {} for overwrite.", objectURI.getPath());
    }
  }

  /**
   * @return s3 WritePType
   */
  public static WritePType getS3WriteType() {
    return Configuration.getEnum(PropertyKey.PROXY_S3_WRITE_TYPE, WriteType.class).toProto();
  }

  /**
   * Checks if authentication is enabled.
   *
   * @param conf Alluxio configuration
   * @return true if authentication is enabled, false otherwise
   */
  public static boolean isAuthenticationEnabled(AlluxioConfiguration conf) {
    return conf.getBoolean(PropertyKey.S3_REST_AUTHENTICATION_ENABLED);
  }

  /**
   * @param user the {@link Subject} name of the filesystem user
   * @param fs the source {@link FileSystem} to base off of
   * @return A {@link FileSystem} with the subject set to the provided user
   */
  public static FileSystem createFileSystemForUser(
      String user, FileSystem fs) {
    if (user == null) {
      // Used to return the top-level FileSystem view when not using Authentication
      return fs;
    }

    final Subject subject = new Subject();
    subject.getPrincipals().add(new User(user));
    // Use local conf to create filesystem rather than fs.getConf()
    // due to fs conf will be changed by merged cluster conf.
    return FileSystem.Factory.get(subject, Configuration.global());
  }

  /**
   * Convert MultivaluedMap to a single value map.
   *
   * @param queryParameters MultivaluedMap
   * @param lowerCase whether to use lower case
   * @return a single value map
   */
  public static Map<String, String> fromMultiValueToSingleValueMap(
      MultivaluedMap<String, String> queryParameters, boolean lowerCase) {
    Map<String, String> result = lowerCase
        ? new TreeMap<>(new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
              return o1.compareToIgnoreCase(o2);
          }
        }) : new HashMap<>();
    for (String key : queryParameters.keySet()) {
      result.put(key, queryParameters.getFirst(key));
    }
    return result;
  }

  /**
   * Given xAttr, parses and returns the Content-Type header metadata
   * as its corresponding {@link MediaType}, or otherwise defaults
   * to {@link MediaType#APPLICATION_OCTET_STREAM_TYPE}.
   * @param xAttr the Inode's xAttrs
   * @return the {@link MediaType} corresponding to the Content-Type header
   */
  public static MediaType deserializeContentType(Map<String, byte[]> xAttr) {
    MediaType type = MediaType.APPLICATION_OCTET_STREAM_TYPE;
    // Fetch the Content-Type from the Inode xAttr
    if (xAttr == null) {
      return type;
    }
    if (xAttr.containsKey(S3Constants.CONTENT_TYPE_XATTR_KEY)) {
      String contentType = new String(xAttr.get(
          S3Constants.CONTENT_TYPE_XATTR_KEY), S3Constants.HEADER_CHARSET);
      if (!contentType.isEmpty()) {
        type = MediaType.valueOf(contentType);
      }
    }
    return type;
  }

  /**
   * Given xAttr, parses and deserializes the Tagging metadata
   * into a {@link TaggingData} object. Returns null if no data exists.
   * @param xAttr the Inode's xAttrs
   * @return the deserialized {@link TaggingData} object
   */
  public static TaggingData deserializeTags(Map<String, byte[]> xAttr)
      throws IOException {
    Map<String, String> tagMap = new HashMap<>(xAttr.size());
    for (Map.Entry<String, byte[]> tags : xAttr.entrySet()) {
      tagMap.put(tags.getKey(), new String(tags.getValue()));
    }
    return new TaggingData().addTags(tagMap);
  }

  /**
   * Given a URL-encoded Tagging header, parses and deserializes the Tagging metadata
   * into a {@link TaggingData} object. Returns null on empty strings.
   * @param taggingHeader the URL-encoded Tagging header
   * @param maxHeaderMetadataSize Header user-metadata size limit validation (default max: 2 KB)
   * - https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingMetadata.html
   * @return the deserialized {@link TaggingData} object
   */
  public static TaggingData deserializeTaggingHeader(
      String taggingHeader, int maxHeaderMetadataSize) throws S3Exception {
    if (taggingHeader == null || taggingHeader.isEmpty()) { return null; }
    if (maxHeaderMetadataSize > 0
        && taggingHeader.getBytes(S3Constants.TAGGING_CHARSET).length
        > maxHeaderMetadataSize) {
      throw new S3Exception(S3ErrorCode.METADATA_TOO_LARGE);
    }
    Map<String, String> tagMap = new HashMap<>();
    for (String tag : taggingHeader.split("&")) {
      String[] entries = tag.split("=");
      if (entries.length > 1) {
        tagMap.put(entries[0], entries[1]);
      } else { // Key was provided without a value
        tagMap.put(entries[0], "");
      }
    }
    return new TaggingData().addTags(tagMap);
  }

  /**
   * This helper method is used to set the ETag xAttr on an object.
   * @param fs The {@link FileSystem} used to make the gRPC request
   * @param objectUri The {@link AlluxioURI} for the object to update
   * @param entityTag The entity tag of the object (MD5 checksum of the object contents)
   * @throws IOException
   * @throws AlluxioException
   */
  public static void setEntityTag(FileSystem fs, AlluxioURI objectUri, String entityTag)
      throws IOException, AlluxioException {
    fs.setAttribute(objectUri, SetAttributePOptions.newBuilder()
        .putXattr(S3Constants.ETAG_XATTR_KEY,
            ByteString.copyFrom(entityTag, S3Constants.XATTR_STR_CHARSET))
        .setXattrUpdateStrategy(File.XAttrUpdateStrategy.UNION_REPLACE)
        .build());
  }

  /**
   * This helper method is used to get the ETag xAttr on an object.
   * @param status The {@link URIStatus} of the object
   * @return the entityTag String, or null if it does not exist
   */
  public static String getEntityTag(URIStatus status) {
    String contenthash = status.getFileInfo().getContentHash();
    if (StringUtils.isNotEmpty(contenthash)) {
      return contenthash;
    } else {
      return null;
    }
  }

  /**
   * Get username from header info.
   *
   * @param authorization authorization info
   * @param requestContext request context
   * @return username
   * @throws S3Exception
   */
  public static String getUser(String authorization, ContainerRequestContext requestContext)
      throws S3Exception {
    // TODO(czhu): refactor PropertyKey.S3_REST_AUTHENTICATION_ENABLED to an ENUM
    //             to specify between using custom Authenticator class vs. Alluxio Master schemes
    if (S3RestUtils.isAuthenticationEnabled(Configuration.global())) {
      return getUserFromSignature(requestContext);
    }
    try {
      return getUserFromAuthorization(authorization, Configuration.global());
    } catch (RuntimeException e) {
      throw new S3Exception(new S3ErrorCode(S3ErrorCode.INTERNAL_ERROR.getCode(),
          e.getMessage(), S3ErrorCode.INTERNAL_ERROR.getStatus()));
    }
  }

  /**
   * Get username from header info from HttpServletRequest.
   *
   * @param authorization
   * @param request
   * @return user name
   * @throws S3Exception
   */
  public static String getUser(String authorization, HttpServletRequest request)
          throws S3Exception {
    if (S3RestUtils.isAuthenticationEnabled(Configuration.global())) {
      return getUserFromSignature(request);
    }
    try {
      return getUserFromAuthorization(authorization, Configuration.global());
    } catch (RuntimeException e) {
      throw new S3Exception(new S3ErrorCode(S3ErrorCode.INTERNAL_ERROR.getCode(),
              e.getMessage(), S3ErrorCode.INTERNAL_ERROR.getStatus()));
    }
  }

  private static String getUserFromSignature(HttpServletRequest request)
          throws S3Exception {
    AwsSignatureProcessor signatureProcessor = new AwsSignatureProcessor(request);
    Authenticator authenticator = Authenticator.Factory.create(Configuration.global());
    AwsAuthInfo authInfo = signatureProcessor.getAuthInfo();
    if (authenticator.isAuthenticated(authInfo)) {
      return authInfo.getAccessID();
    }
    throw new S3Exception(authInfo.toString(), S3ErrorCode.INVALID_IDENTIFIER);
  }

  /**
   * Get username from parsed header info.
   *
   * @return username
   * @throws S3Exception
   */
  private static String getUserFromSignature(ContainerRequestContext requestContext)
      throws S3Exception {
    AwsSignatureProcessor signatureProcessor = new AwsSignatureProcessor(requestContext);
    Authenticator authenticator = Authenticator.Factory.create(Configuration.global());
    AwsAuthInfo authInfo = signatureProcessor.getAuthInfo();
    if (authenticator.isAuthenticated(authInfo)) {
      return authInfo.getAccessID();
    }
    throw new S3Exception(authInfo.toString(), S3ErrorCode.INVALID_IDENTIFIER);
  }

  /**
   * Gets the user from the authorization header string for AWS Signature Version 4.
   * @param authorization the authorization header string
   * @param conf the {@link AlluxioConfiguration} Alluxio conf
   * @return the user
   */
  @VisibleForTesting
  public static String getUserFromAuthorization(String authorization, AlluxioConfiguration conf)
      throws S3Exception {
    if (conf.get(PropertyKey.SECURITY_AUTHENTICATION_TYPE) == AuthType.NOSASL) {
      return null;
    }
    if (authorization == null) {
      throw new S3Exception("The authorization header that you provided is not valid.",
          S3ErrorCode.AUTHORIZATION_HEADER_MALFORMED);
    }

    // Parse the authorization header defined at
    // https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-auth-using-authorization-header.html
    // All other authorization types are deprecated or EOL (as of writing)
    // Example Header value (spaces turned to line breaks):
    // AWS4-HMAC-SHA256
    // Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request,
    // SignedHeaders=host;range;x-amz-date,
    // Signature=fe5f80f77d5fa3beca038a248ff027d0445342fe2855ddc963176630326f1024

    // We only care about the credential key, so split the header by " " and then take everything
    // after the "=" and before the first "/"
    String[] fields = authorization.split(" ");
    if (fields.length < 2) {
      throw new S3Exception("The authorization header that you provided is not valid.",
          S3ErrorCode.AUTHORIZATION_HEADER_MALFORMED);
    }
    String credentials = fields[1];
    String[] creds = credentials.split("=");
    // only support version 4 signature
    if (creds.length < 2 || !StringUtils.equals("Credential", creds[0])
        || !creds[1].contains("/")) {
      throw new S3Exception("The authorization header that you provided is not valid.",
          S3ErrorCode.AUTHORIZATION_HEADER_MALFORMED);
    }

    final String user = creds[1].substring(0, creds[1].indexOf("/")).trim();
    if (user.isEmpty()) {
      throw new S3Exception("The authorization header that you provided is not valid.",
          S3ErrorCode.AUTHORIZATION_HEADER_MALFORMED);
    }

    return user;
  }

  /**
   * Populate xattr with content type info from header.
   * @param xattrMap
   * @param contentTypeHeader
   */
  public static void populateContentTypeInXAttr(Map<String, ByteString> xattrMap,
                                                String contentTypeHeader) {
    if (contentTypeHeader != null) {
      xattrMap.put(S3Constants.CONTENT_TYPE_XATTR_KEY,
              ByteString.copyFrom(contentTypeHeader, S3Constants.HEADER_CHARSET));
    }
  }

  /**
   * Populate xattr map with tagging info from tagging header.
   * @param xattrMap
   * @param taggingHeader
   * @param auditContext
   * @param objectPath
   * @throws S3Exception
   */
  public static void populateTaggingInXAttr(Map<String, ByteString> xattrMap, String taggingHeader,
                                            S3AuditContext auditContext, String objectPath)
      throws S3Exception {
    TaggingData tagData = null;
    if (taggingHeader != null) { // Parse the tagging header if it exists for PutObject
      try {
        tagData = S3RestUtils.deserializeTaggingHeader(
            taggingHeader, S3Handler.MAX_HEADER_METADATA_SIZE);
      } catch (IllegalArgumentException e) {
        Throwable cause = e.getCause();
        if (cause instanceof S3Exception) {
          throw S3RestUtils.toObjectS3Exception((S3Exception) cause, objectPath,
                  auditContext);
        }
        throw S3RestUtils.toObjectS3Exception(e, objectPath, auditContext);
      }
    }
    LOG.debug("tagData={}", tagData);
    // Populate the xattr Map with the metadata tags if provided
    if (tagData != null) {
      try {
        xattrMap.put(S3Constants.TAGGING_XATTR_KEY, TaggingData.serialize(tagData));
      } catch (Exception e) {
        throw S3RestUtils.toObjectS3Exception(e, objectPath, auditContext);
      }
    }
  }

  /**
   * Create a rate limiter for given rate.
   * @param rate bytes per second
   * @return empty if rate <= 0
   */
  public static Optional<RateLimiter> createRateLimiter(long rate) {
    if (rate <= 0) {
      return Optional.empty();
    }
    return Optional.of(RateLimiter.create(rate));
  }

    /**
     * Comparator based on uri name， treat uri name as a Long number.
     */
  public static class URIStatusNameComparator implements Comparator<URIStatus>, Serializable {

    private static final long serialVersionUID = 733270188584155565L;

    @Override
    public int compare(URIStatus o1, URIStatus o2) {
      long part1 = Long.parseLong(o1.getName());
      long part2 = Long.parseLong(o2.getName());
      return Long.compare(part1, part2);
    }
  }

  private S3RestUtils() {} // prevent instantiation
}
