content: /*
   * 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.Constants;
  import alluxio.client.file.FileSystem;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.grpc.Bits;
  import alluxio.grpc.CreateDirectoryPOptions;
  import alluxio.grpc.PMode;
  import alluxio.grpc.XAttrPropagationStrategy;
  import alluxio.master.audit.AsyncUserAccessAuditLogWriter;
  import alluxio.util.CommonUtils;
  import alluxio.util.ThreadUtils;
  import alluxio.web.ProxyWebServer;
  
  import com.google.common.base.Stopwatch;
  import com.google.common.cache.Cache;
  import com.google.common.cache.CacheBuilder;
  import org.eclipse.jetty.server.Request;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.net.URLDecoder;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  import javax.annotation.Nullable;
  import javax.servlet.ServletContext;
  import javax.servlet.ServletInputStream;
  import javax.servlet.ServletOutputStream;
  import javax.servlet.http.HttpServletRequest;
  import javax.servlet.http.HttpServletResponse;
  import javax.ws.rs.core.MultivaluedMap;
  import javax.ws.rs.core.Response;
  
  /**
   *
   */
  public class S3Handler {
    public static final boolean BUCKET_NAMING_RESTRICTION_ENABLED =
            Configuration.getBoolean(PropertyKey.PROXY_S3_BUCKET_NAMING_RESTRICTIONS_ENABLED);
    public static final int MAX_HEADER_METADATA_SIZE =
            (int) Configuration.getBytes(PropertyKey.PROXY_S3_METADATA_HEADER_MAX_SIZE);
    public static final boolean MULTIPART_CLEANER_ENABLED =
            Configuration.getBoolean(PropertyKey.PROXY_S3_MULTIPART_UPLOAD_CLEANER_ENABLED);
    // https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
    // - Undocumented edge-case, no adjacent periods with hyphens, i.e: '.-' or '-.'
    public static final Pattern BUCKET_ADJACENT_DOTS_DASHES_PATTERN = Pattern.compile("([-\\.]{2})");
    public static final Pattern BUCKET_INVALIDATION_PREFIX_PATTERN = Pattern.compile("^xn--.*");
    public static final Pattern BUCKET_INVALID_SUFFIX_PATTERN = Pattern.compile(".*-s3alias$");
    public static final Pattern BUCKET_VALID_NAME_PATTERN =
            Pattern.compile("[a-z0-9][a-z0-9\\.-]{1,61}[a-z0-9]");
    public static final Pattern BUCKET_PATH_PATTERN =
            Pattern.compile("^" + S3RequestServlet.S3_V2_SERVICE_PATH_PREFIX + "/[^/]*$");
    public static final Pattern OBJECT_PATH_PATTERN =
            Pattern.compile("^" + S3RequestServlet.S3_V2_SERVICE_PATH_PREFIX + "/[^/]*/.*$");
    public static final int BUCKET_PATH_CACHE_SIZE = 65536;
    /* BUCKET_PATH_CACHE caches bucket path during specific period.
     BUCKET_PATH_CACHE.put(bucketPath,true) means bucket path exists.
     BUCKET_PATH_CACHE.put(bucketPath,false) plays the same effect
     as BUCKET_PATH_CACHE.remove(bucketPath). */
    public static final Cache<String, Boolean> BUCKET_PATH_CACHE = CacheBuilder.newBuilder()
        .maximumSize(BUCKET_PATH_CACHE_SIZE)
        .expireAfterWrite(
            Configuration.global().getMs(PropertyKey.PROXY_S3_BUCKETPATHCACHE_TIMEOUT_MS),
            TimeUnit.MILLISECONDS)
        .build();
    private static final Logger LOG = LoggerFactory.getLogger(S3Handler.class);
    private static final ThreadLocal<byte[]> TLS_BYTES =
            ThreadLocal.withInitial(() -> new byte[8 * 1024]);
    // Position read will consume additional memory, so here we limit the maximum memory usage.
    private static final int MAX_POSITION_READ_LENGTH = 4 * Constants.MB;
    public static final int USE_POSITION_READ_SIZE = (int) Math.min(MAX_POSITION_READ_LENGTH,
        Configuration.getBytes(PropertyKey.PROXY_S3_USE_POSITION_READ_RANGE_SIZE));
    private final String mBucket;
    private final String mObject;
    private final HttpServletRequest mServletRequest;
    private final HttpServletResponse mServletResponse;
    private final ServletContext mServletContext;
    public AsyncUserAccessAuditLogWriter mAsyncAuditLogWriter;
    String[] mUnsupportedSubResources = {"acl", "policy", "versioning", "cors",
      "encryption", "intelligent-tiering", "inventory", "lifecycle",
      "metrics", "ownershipControls", "replication", "website", "accelerate",
      "location", "logging", "metrics", "notification", "ownershipControls",
      "policyStatus", "requestPayment", "attributes", "legal-hold", "object-lock",
      "retention", "torrent", "publicAccessBlock", "restore", "select"};
    Set<String> mUnsupportedSubResourcesSet = new HashSet<>(Arrays.asList(mUnsupportedSubResources));
    Map<String, String> mAmzHeaderMap = new HashMap<>();
    Request mBaseRequest;
    private Stopwatch mStopwatch;
    private String mUser;
    private S3BaseTask mS3Task;
    private FileSystem mMetaFS;
  
    /**
     * S3Handler Constructor.
     * @param bucket
     * @param object
     * @param request
     * @param response
     */
    public S3Handler(String bucket, String object,
                     HttpServletRequest request, HttpServletResponse response) {
      mBucket = bucket;
      mObject = object;
      mServletRequest = request;
      mServletResponse = response;
      mServletContext = request.getServletContext();
    }
  
    /**
     * Create a S3Handler based on the incoming Request.
     * @param path
     * @param request
     * @param response
     * @return A S3Handler
     * @throws Exception
     *
     */
    public static S3Handler createHandler(String path,
                                          HttpServletRequest request,
                                          HttpServletResponse response) throws Exception {
      Stopwatch stopwatch = Stopwatch.createStarted();
      Matcher bucketMatcher = BUCKET_PATH_PATTERN.matcher(path);
      Matcher objectMatcher = OBJECT_PATH_PATTERN.matcher(path);
  
      String pathStr = path;
      String bucket = null;
      String object = null;
      S3Handler handler = null;
      try {
        if (bucketMatcher.matches()) {
          pathStr = path.substring(S3RequestServlet.S3_V2_SERVICE_PATH_PREFIX.length() + 1);
          bucket = URLDecoder.decode(pathStr, "UTF-8");
        } else if (objectMatcher.matches()) {
          pathStr = path.substring(S3RequestServlet.S3_V2_SERVICE_PATH_PREFIX.length() + 1);
          bucket = URLDecoder.decode(
            pathStr.substring(0, pathStr.indexOf(AlluxioURI.SEPARATOR)), "UTF-8");
          object = URLDecoder.decode(
            pathStr.substring(pathStr.indexOf(AlluxioURI.SEPARATOR) + 1), "UTF-8");
        }
        handler = new S3Handler(bucket, object, request, response);
        handler.setStopwatch(stopwatch);
        handler.init();
        S3BaseTask task = null;
        if (object != null && !object.isEmpty()) {
          task = S3ObjectTask.Factory.create(handler);
        } else {
          task = S3BucketTask.Factory.create(handler);
        }
        handler.setS3Task(task);
        return handler;
      } catch (Exception ex) {
        LOG.error("Exception during create s3handler:{}", ThreadUtils.formatStackTrace(ex));
        throw ex;
      }
    }
  
    /**
     * Process the response returned from S3Task core logic to write to downstream.
     * @param servletResponse
     * @param response
     * @throws IOException
     */
    public static void processResponse(HttpServletResponse servletResponse,
                                       Response response) throws IOException {
      try {
        // Status
        servletResponse.setStatus(response.getStatus());
        // Headers
        final MultivaluedMap<String, String> headers = response.getStringHeaders();
        for (final Map.Entry<String, List<String>> e : headers.entrySet()) {
          final Iterator<String> it = e.getValue().iterator();
          if (!it.hasNext()) {
            continue;
          }
          final String header = e.getKey();
          if (servletResponse.containsHeader(header)) {
            // replace any headers previously set with values from Jersey container response.
            servletResponse.setHeader(header, it.next());
          }
          while (it.hasNext()) {
            servletResponse.addHeader(header, it.next());
          }
        }
        // Entity
        if (response.hasEntity()) {
          ServletOutputStream servletOut = servletResponse.getOutputStream();
          Object entity = response.getEntity();
          if (entity instanceof InputStream) {
            InputStream is = (InputStream) entity;
            byte[] bytesArray = TLS_BYTES.get();
            int read;
            do {
              try {
                read = is.read(bytesArray);
              } catch (IOException ex) {
                /* Alluxio thrown IOException, remapping the exception
                and send new response to downstream again */
                Response errorResponse = S3ErrorResponse.createErrorResponse(ex, "");
                S3Handler.processResponse(servletResponse, errorResponse);
                return;
              }
              if (read == -1) {
                break;
              }
              servletOut.write(bytesArray, 0, read);
            } while (true);
          } else {
            String contentStr = entity.toString();
            int contentLen = contentStr.length();
            servletResponse.setContentLength(contentLen);
            servletOut.write(contentStr.getBytes());
          }
        }
      } finally {
        response.close();
      }
    }
  
    /**
     * Initialize the S3Handler object in preparation for handling the request.
     * @throws Exception
     */
    public void init() throws Exception {
      // Do Authentication of the request.
      doAuthentication();
      // Extract x-amz- headers.
      extractAMZHeaders();
      // Reject unsupported subresources.
      rejectUnsupportedResources();
      // Init utils
      ServletContext context = getServletContext();
      mMetaFS = (FileSystem) context.getAttribute(ProxyWebServer.FILE_SYSTEM_SERVLET_RESOURCE_KEY);
      mAsyncAuditLogWriter = (AsyncUserAccessAuditLogWriter) context.getAttribute(
          ProxyWebServer.ALLUXIO_PROXY_AUDIT_LOG_WRITER_KEY);
      // Initiate the S3 API metadata directories
      if (!mMetaFS.exists(new AlluxioURI(S3RestUtils.MULTIPART_UPLOADS_METADATA_DIR))) {
        mMetaFS.createDirectory(new AlluxioURI(S3RestUtils.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());
      }
    }
  
    /**
     * get S3Task of this S3Handler.
     * @return S3BaseTask
     */
    public S3BaseTask getS3Task() {
      return mS3Task;
    }
  
    /**
     * set S3Task for this S3Handler.
     * @param task
     */
    public void setS3Task(S3BaseTask task) {
      mS3Task = task;
    }
  
    /**
     * get HTTP verb of this request.
     * @return HTTP Verb
     */
    public String getHTTPVerb() {
      return mServletRequest.getMethod();
    }
  
    /**
     * get specified HTTP header value of this request.
     * @param headerName
     * @return header value
     */
    public String getHeader(String headerName) {
      return mServletRequest.getHeader(headerName);
    }
  
    /**
     * get specified HTTP header with a default if not exist.
     * @param headerName
     * @param defaultHeaderValue
     * @return header value
     */
    public String getHeaderOrDefault(String headerName, String defaultHeaderValue) {
      String headerVal = mServletRequest.getHeader(headerName);
      if (headerVal == null) {
        headerVal = defaultHeaderValue;
      }
      return headerVal;
    }
  
    /**
     * get HttpServletResponse of this request.
     * @return HttpServletResponse
     */
    public HttpServletResponse getServletResponse() {
      return mServletResponse;
    }
  
    /**
     * get HttpServletRequest of this request.
     * @return HttpServletRequest
     */
    public HttpServletRequest getServletRequest() {
      return mServletRequest;
    }
  
    /**
     * get ServletContext from current http conversation.
     * @return ServletContext
     */
    public ServletContext getServletContext() {
      return mServletContext;
    }
  
    /**
     * retrieve given query parameter value.
     * @param queryParam
     * @return query parameter value
     */
    public String getQueryParameter(String queryParam) {
      return mServletRequest.getParameter(queryParam);
    }
  
    /**
     * retrieve inputstream from incoming request.
     * @return ServletInputStream
     */
    public ServletInputStream getInputStream() throws IOException {
      return mServletRequest.getInputStream();
    }
  
    /**
     * Creates a {@link S3AuditContext} instance.
     *
     * @param command the command to be logged by this {@link S3AuditContext}
     * @param user  user name
     * @param bucket  bucket name
     * @param object  object name
     * @return newly-created {@link S3AuditContext} instance
     */
    public S3AuditContext createAuditContext(String command,
                                             String user,
                                             @Nullable String bucket,
                                             @Nullable String object) {
      // Audit log may be enabled during runtime
      AsyncUserAccessAuditLogWriter auditLogWriter = null;
      if (Configuration.getBoolean(PropertyKey.PROXY_AUDIT_LOGGING_ENABLED)) {
        auditLogWriter = mAsyncAuditLogWriter;
      }
      S3AuditContext auditContext = new S3AuditContext(auditLogWriter);
      if (auditLogWriter != null) {
        String ugi = "";
        if (user != null) {
          try {
            String primaryGroup = CommonUtils.getPrimaryGroupName(user, Configuration.global());
            ugi = user + "," + primaryGroup;
          } catch (IOException e) {
            LOG.debug("Failed to get primary group for user {}.", user);
            ugi = user + ",N/A";
          }
        } else {
          ugi = "N/A";
        }
        auditContext.setUgi(ugi)
            .setCommand(command)
            .setIp(String.format("%s:%s",
                mServletRequest.getRemoteAddr(), mServletRequest.getRemotePort()))
            .setBucket(bucket)
            .setObject(object)
            .setAllowed(true)
            .setSucceeded(true)
            .setCreationTimeNs(System.nanoTime());
      }
      return auditContext;
    }
  
    /**
     * Utility function to dump a collection into a string.
     * @param prefix
     * @param collection
     * @return result string
     */
    public String printCollection(String prefix, Collection<? extends Object> collection) {
      StringBuilder sb = new StringBuilder(prefix + ":[");
      Iterator<? extends Object> it = collection.iterator();
      while (it.hasNext()) {
        sb.append(it.next().toString());
        if (it.hasNext()) {
          sb.append(",");
        }
      }
      sb.append("]");
      return sb.toString();
    }
  
    /**
     * Utility function to dump a map into a string.
     * @param prefix
     * @param map
     * @return result string
     */
    public String printMap(String prefix, Map<? extends Object, ? extends Object> map) {
      StringBuilder sb = new StringBuilder(prefix + ":[");
      Iterator<? extends Map.Entry<?, ?>> it = map.entrySet().iterator();
      while (it.hasNext()) {
        Map.Entry<?, ?> entry = it.next();
        sb.append(entry.getKey().toString() + ":" + entry.getValue().toString());
        if (it.hasNext()) {
          sb.append(",");
        }
      }
      sb.append("]");
      return sb.toString();
    }
  
    /**
     * Utility function to help extract x-amz- headers from request.
     */
    public void extractAMZHeaders() {
      java.util.Enumeration<String> headerNamesIt = mServletRequest.getHeaderNames();
      while (headerNamesIt.hasMoreElements()) {
        String header = headerNamesIt.nextElement();
        mAmzHeaderMap.putIfAbsent(header, mServletRequest.getHeader(header));
      }
    }
  
    /**
     * Reject unsupported request from the given subresources from request.
     * @throws S3Exception
     */
    public void rejectUnsupportedResources() throws S3Exception {
      java.util.Enumeration<String> parameterNamesIt = mServletRequest.getParameterNames();
      while (parameterNamesIt.hasMoreElements()) {
        if (mUnsupportedSubResourcesSet.contains(parameterNamesIt.nextElement())) {
          throw new S3Exception(S3Constants.EMPTY, S3ErrorCode.NOT_IMPLEMENTED);
        }
      }
    }
  
    /**
     * Do S3 request authentication.
     * @throws Exception
     */
    public void doAuthentication() throws Exception {
      try {
        String authorization = mServletRequest.getHeader("Authorization");
        String user = S3RestUtils.getUser(authorization, mServletRequest);
        // replace the authorization header value to user
        LOG.debug("request origin Authorization Header is: {}, new user header is: {}",
            authorization, user);
        mUser = user;
      } catch (Exception e) {
        LOG.warn("exception happened in Authentication.");
        throw e;
      }
    }
  
    /**
     * Get the user name of this request.
     * @return user name
     */
    public String getUser() {
      return mUser;
    }
  
    /**
     * Get the bucket name of this request.
     * @return bucket name
     */
    public String getBucket() {
      return mBucket;
    }
  
    /**
     * Get the object name of this request.
     * @return object name
     */
    public String getObject() {
      return mObject;
    }
  
    /**
     * Get system user FileSystem object.
     * @return FileSystem object
     */
    public FileSystem getMetaFS() {
      return mMetaFS;
    }
  
    /**
     * Get Stopwatch object used for recording this request's latency.
     * @return Stopwatch object
     */
    public Stopwatch getStopwatch() {
      return mStopwatch;
    }
  
    /**
     * Set the Stopwatch object used for recording this request's latency.
     * @param stopwatch
     */
    public void setStopwatch(Stopwatch stopwatch) {
      mStopwatch = stopwatch;
    }
  }
  