HDDS-10775. Support bucket ownership verification

hadoop-ozone/integration-test-s3/dev-support/findbugsExcludeFile.xml

@@ -13,4 +13,8 @@
   limitations under the License. See accompanying LICENSE file.
 -->
 <FindBugsFilter>
+  <Match>
+    <Class name="org.apache.hadoop.ozone.s3.awssdk.v2.AbstractS3SDKV2Tests$S3BucketVerificationConditionTests"/>
+    <Bug pattern="SIC_INNER_SHOULD_BE_STATIC"/>
+  </Match>
 </FindBugsFilter>

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/BucketEndpoint.java

@@ -134,7 +134,7 @@ public Response get(
     try {
       if (aclMarker != null) {
         s3GAction = S3GAction.GET_ACL;
-        S3BucketAcl result = getAcl(bucketName);
+        S3BucketAcl result = getAcl(hh, bucketName);
         getMetrics().updateGetAclSuccessStats(startNanos);
         AUDIT.logReadSuccess(
             buildAuditMessageForSuccess(s3GAction, getAuditParameters()));
@@ -143,7 +143,7 @@ public Response get(
 
       if (uploads != null) {
         s3GAction = S3GAction.LIST_MULTIPART_UPLOAD;
-        return listMultipartUploads(bucketName, prefix, keyMarker, uploadIdMarker, maxUploads);
+        return listMultipartUploads(bucketName, prefix, keyMarker, uploadIdMarker, maxUploads, hh);
       }
 
       maxKeys = validateMaxKeys(maxKeys);
@@ -168,8 +168,9 @@ public Response get(
           && OZONE_URI_DELIMITER.equals(delimiter);
 
       bucket = getBucket(bucketName);
-      ozoneKeyIterator = bucket.listKeys(prefix, prevKey, shallow);
+      BucketOwnerCondition.verify(hh, bucket.getOwner());
 
+      ozoneKeyIterator = bucket.listKeys(prefix, prevKey, shallow);
     } catch (OMException ex) {
       AUDIT.logReadFailure(
           buildAuditMessageForFailure(s3GAction, getAuditParameters(), ex));
@@ -347,18 +348,21 @@ public Response listMultipartUploads(
       String prefix,
       String keyMarker,
       String uploadIdMarker,
-      int maxUploads)
+      int maxUploads,
+      HttpHeaders httpHeaders)
       throws OS3Exception, IOException {
 
     if (maxUploads < 1 || maxUploads > 1000) {
       throw newError(S3ErrorTable.INVALID_ARGUMENT, "max-uploads",
           new Exception("max-uploads must be between 1 and 1000"));
     }
 
+
     long startNanos = Time.monotonicNowNanos();
     S3GAction s3GAction = S3GAction.LIST_MULTIPART_UPLOAD;
 
     OzoneBucket bucket = getBucket(bucketName);
+    BucketOwnerCondition.verify(httpHeaders, bucket.getOwner());
 
     try {
       OzoneMultipartUploadList ozoneMultipartUploadList =
@@ -408,16 +412,25 @@ public Response listMultipartUploads(
    * for more details.
    */
   @HEAD
-  public Response head(@PathParam("bucket") String bucketName)
+  public Response head(@PathParam("bucket") String bucketName, @Context HttpHeaders httpHeaders)
       throws OS3Exception, IOException {
     long startNanos = Time.monotonicNowNanos();
     S3GAction s3GAction = S3GAction.HEAD_BUCKET;
     try {
-      getBucket(bucketName);
+      OzoneBucket bucket = getBucket(bucketName);
+      BucketOwnerCondition.verify(httpHeaders, bucket.getOwner());
       AUDIT.logReadSuccess(
           buildAuditMessageForSuccess(s3GAction, getAuditParameters()));
       getMetrics().updateHeadBucketSuccessStats(startNanos);
       return Response.ok().build();
+    } catch (OMException ex) {
+      AUDIT.logReadFailure(
+          buildAuditMessageForFailure(s3GAction, getAuditParameters(), ex));
+      if (isAccessDenied(ex)) {
+        throw newError(S3ErrorTable.ACCESS_DENIED, bucketName, ex);
+      } else {
+        throw ex;
+      }
     } catch (Exception e) {
       AUDIT.logReadFailure(
           buildAuditMessageForFailure(s3GAction, getAuditParameters(), e));
@@ -432,13 +445,13 @@ public Response head(@PathParam("bucket") String bucketName)
    * for more details.
    */
   @DELETE
-  public Response delete(@PathParam("bucket") String bucketName)
+  public Response delete(@PathParam("bucket") String bucketName, @Context HttpHeaders httpHeaders)
       throws IOException, OS3Exception {
     long startNanos = Time.monotonicNowNanos();
     S3GAction s3GAction = S3GAction.DELETE_BUCKET;
 
     try {
-      deleteS3Bucket(bucketName);
+      deleteS3Bucket(bucketName, httpHeaders);
     } catch (OMException ex) {
       AUDIT.logWriteFailure(
           buildAuditMessageForFailure(s3GAction, getAuditParameters(), ex));
@@ -477,7 +490,8 @@ public Response delete(@PathParam("bucket") String bucketName)
   @Produces(MediaType.APPLICATION_XML)
   public MultiDeleteResponse multiDelete(@PathParam("bucket") String bucketName,
                                          @QueryParam("delete") String delete,
-                                         MultiDeleteRequest request)
+                                         MultiDeleteRequest request,
+                                         @Context HttpHeaders httpHeaders)
       throws OS3Exception, IOException {
     S3GAction s3GAction = S3GAction.MULTI_DELETE;
 
@@ -492,6 +506,7 @@ public MultiDeleteResponse multiDelete(@PathParam("bucket") String bucketName,
       }
       long startNanos = Time.monotonicNowNanos();
       try {
+        BucketOwnerCondition.verify(httpHeaders, bucket.getOwner());
         undeletedKeyResultMap = bucket.deleteKeys(deleteKeys, true);
         for (DeleteObject d : request.getObjects()) {
           ErrorInfo error = undeletedKeyResultMap.get(d.getKey());
@@ -508,6 +523,14 @@ public MultiDeleteResponse multiDelete(@PathParam("bucket") String bucketName,
           }
         }
         getMetrics().updateDeleteKeySuccessStats(startNanos);
+      } catch (OMException ex) {
+        AUDIT.logReadFailure(
+            buildAuditMessageForFailure(s3GAction, getAuditParameters(), ex));
+        if (isAccessDenied(ex)) {
+          throw newError(S3ErrorTable.ACCESS_DENIED, bucketName, ex);
+        } else {
+          throw ex;
+        }
       } catch (IOException ex) {
         LOG.error("Delete key failed: {}", ex.getMessage());
         getMetrics().updateDeleteKeyFailureStats(startNanos);
@@ -534,16 +557,14 @@ public MultiDeleteResponse multiDelete(@PathParam("bucket") String bucketName,
    * <p>
    * see: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetBucketAcl.html
    */
-  public S3BucketAcl getAcl(String bucketName)
+  public S3BucketAcl getAcl(HttpHeaders headers, String bucketName)
       throws OS3Exception, IOException {
     long startNanos = Time.monotonicNowNanos();
     S3BucketAcl result = new S3BucketAcl();
     try {
       OzoneBucket bucket = getBucket(bucketName);
-      OzoneVolume volume = getVolume();
-      // TODO: use bucket owner instead of volume owner here once bucket owner
-      // TODO: is supported.
-      S3Owner owner = S3Owner.of(volume.getOwner());
+      BucketOwnerCondition.verify(headers, bucket.getOwner());
+      S3Owner owner = S3Owner.of(bucket.getOwner());
       result.setOwner(owner);
 
       // TODO: remove this duplication avoid logic when ACCESS and DEFAULT scope
@@ -592,6 +613,7 @@ public Response putAcl(String bucketName, HttpHeaders httpHeaders,
 
     try {
       OzoneBucket bucket = getBucket(bucketName);
+      BucketOwnerCondition.verify(httpHeaders, bucket.getOwner());
       OzoneVolume volume = getVolume();
 
       List<OzoneAcl> ozoneAclListOnBucket = new ArrayList<>();

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/BucketOwnerCondition.java

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.s3.endpoint;
+
+import javax.ws.rs.core.HttpHeaders;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.s3.util.S3Consts;
+
+/**
+ * Bucket owner condition to verify bucket ownership.
+ * <p>
+ * See: https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucket-owner-condition.html
+ * for more details
+ */
+public final class BucketOwnerCondition {
+
+  public static final String ERROR_MESSAGE = "Expected bucket owner does not match";
+
+  private BucketOwnerCondition() {
+
+  }
+
+  /**
+   * Verify the bucket owner condition.
+   *
+   * @param headers       HTTP headers
+   * @param bucketOwner   bucket owner
+   * @throws OMException if the expected bucket owner does not match the actual
+   *                     bucket owner
+   */
+  public static void verify(HttpHeaders headers, String bucketOwner) throws OMException {
+    if (headers == null) {
+      return;
+    }
+
+    final String expectedBucketOwner = headers.getHeaderString(S3Consts.EXPECTED_BUCKET_OWNER_HEADER);
+    // client does not use this feature
+    if (expectedBucketOwner == null) {
+      return;
+    }
+    if (expectedBucketOwner.equals(bucketOwner)) {
+      return;
+    }
+    throw new OMException(ERROR_MESSAGE, OMException.ResultCodes.PERMISSION_DENIED);
+  }
+
+  /**
+   * Verify the copy operation's source and destination bucket owners.
+   *
+   * @param headers       HTTP headers
+   * @param sourceOwner   source bucket owner
+   * @param destOwner     destination bucket owner
+   * @throws OMException if the expected source or destination bucket owner does
+   *                     not match the actual owners
+   */
+  public static void verifyCopyOperation(HttpHeaders headers, String sourceOwner, String destOwner) throws OMException {
+    if (headers == null) {
+      return;
+    }
+
+    final String expectedSourceOwner = headers.getHeaderString(S3Consts.EXPECTED_SOURCE_BUCKET_OWNER_HEADER);
+    final String expectedDestOwner = headers.getHeaderString(S3Consts.EXPECTED_BUCKET_OWNER_HEADER);
+
+    // expectedSourceOwner is null, means the client does not want to check source owner
+    if (expectedSourceOwner != null && !sourceOwner.equals(expectedSourceOwner)) {
+      throw new OMException(ERROR_MESSAGE, OMException.ResultCodes.PERMISSION_DENIED);
+    }
+
+    // expectedDestOwner is null, means the client does not want to check destination owner
+    if (expectedDestOwner != null && !destOwner.equals(expectedDestOwner)) {
+      throw new OMException(ERROR_MESSAGE, OMException.ResultCodes.PERMISSION_DENIED);
+    }
+  }
+}

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/EndpointBase.java

@@ -210,11 +210,14 @@ protected String createS3Bucket(String bucketName) throws
   /**
    * Deletes an s3 bucket and removes mapping of Ozone volume/bucket.
    * @param s3BucketName - S3 Bucket Name.
+   * @param headers - HttpHeaders
    * @throws  IOException in case the bucket cannot be deleted.
    */
-  protected void deleteS3Bucket(String s3BucketName)
+  protected void deleteS3Bucket(String s3BucketName, HttpHeaders headers)
       throws IOException, OS3Exception {
     try {
+      OzoneBucket bucket = getBucket(s3BucketName);
+      BucketOwnerCondition.verify(headers, bucket.getOwner());
       client.getObjectStore().deleteS3Bucket(s3BucketName);
     } catch (OMException ex) {
       if (ex.getResult() == ResultCodes.PERMISSION_DENIED) {