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

@@ -132,6 +132,8 @@ public Response get(
     OzoneBucket bucket = null;
 
     try {
+      bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(hh, bucketName, bucket.getOwner());
       if (aclMarker != null) {
         s3GAction = S3GAction.GET_ACL;
         S3BucketAcl result = getAcl(bucketName);
@@ -167,7 +169,6 @@ public Response get(
       boolean shallow = listKeysShallowEnabled
           && OZONE_URI_DELIMITER.equals(delimiter);
 
-      bucket = getBucket(bucketName);
       ozoneKeyIterator = bucket.listKeys(prefix, prevKey, shallow);
 
     } catch (OMException ex) {
@@ -408,12 +409,13 @@ 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);
+      S3Owner.verifyBucketOwnerCondition(httpHeaders, bucketName, bucket.getOwner());
       AUDIT.logReadSuccess(
           buildAuditMessageForSuccess(s3GAction, getAuditParameters()));
       getMetrics().updateHeadBucketSuccessStats(startNanos);
@@ -432,12 +434,14 @@ 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 {
+      OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(httpHeaders, bucketName, bucket.getOwner());
       deleteS3Bucket(bucketName);
     } catch (OMException ex) {
       AUDIT.logWriteFailure(
@@ -477,7 +481,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 +497,7 @@ public MultiDeleteResponse multiDelete(@PathParam("bucket") String bucketName,
       }
       long startNanos = Time.monotonicNowNanos();
       try {
+        S3Owner.verifyBucketOwnerCondition(httpHeaders, bucketName, bucket.getOwner());
         undeletedKeyResultMap = bucket.deleteKeys(deleteKeys, true);
         for (DeleteObject d : request.getObjects()) {
           ErrorInfo error = undeletedKeyResultMap.get(d.getKey());
@@ -540,10 +546,7 @@ public S3BucketAcl getAcl(String bucketName)
     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());
+      S3Owner owner = S3Owner.of(bucket.getOwner());
       result.setOwner(owner);
 
       // TODO: remove this duplication avoid logic when ACCESS and DEFAULT scope
@@ -592,6 +595,7 @@ public Response putAcl(String bucketName, HttpHeaders httpHeaders,
 
     try {
       OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(httpHeaders, bucketName, bucket.getOwner());
       OzoneVolume volume = getVolume();
 
       List<OzoneAcl> ozoneAclListOnBucket = new ArrayList<>();

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

@@ -243,6 +243,9 @@ public Response put(
         throw newError(NOT_IMPLEMENTED, keyPath);
       }
       OzoneVolume volume = getVolume();
+      OzoneBucket bucket = volume.getBucket(bucketName);
+      String bucketOwner = bucket.getOwner();
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucketOwner);
       if (taggingMarker != null) {
         s3GAction = S3GAction.PUT_OBJECT_TAGGING;
         return putObjectTagging(volume, bucketName, keyPath, body);
@@ -265,7 +268,6 @@ public Response put(
       boolean storageTypeDefault = StringUtils.isEmpty(storageType);
 
       // Normal put object
-      OzoneBucket bucket = volume.getBucket(bucketName);
       ReplicationConfig replicationConfig =
           getReplicationConfig(bucket, storageType, storageConfig);
 
@@ -428,6 +430,8 @@ public Response get(
     S3GAction s3GAction = S3GAction.GET_KEY;
     PerformanceStringBuilder perf = new PerformanceStringBuilder();
     try {
+      OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       if (taggingMarker != null) {
         s3GAction = S3GAction.GET_OBJECT_TAGGING;
         return getObjectTagging(bucketName, keyPath);
@@ -622,6 +626,8 @@ public Response head(
 
     OzoneKey key;
     try {
+      OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       key = getClientProtocol().headS3Object(bucketName, keyPath);
 
       isFile(keyPath, key);
@@ -743,6 +749,8 @@ public Response delete(
 
     try {
       OzoneVolume volume = getVolume();
+      OzoneBucket bucket = volume.getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       if (taggingMarker != null) {
         s3GAction = S3GAction.DELETE_OBJECT_TAGGING;
         return deleteObjectTagging(volume, bucketName, keyPath);
@@ -818,6 +826,7 @@ public Response initializeMultipartUpload(
 
     try {
       OzoneBucket ozoneBucket = getBucket(bucket);
+      S3Owner.verifyBucketOwnerCondition(headers, bucket, ozoneBucket.getOwner());
       String storageType = headers.getHeaderString(STORAGE_CLASS_HEADER);
       String storageConfig = headers.getHeaderString(CUSTOM_METADATA_HEADER_PREFIX + STORAGE_CONFIG_HEADER);
 
@@ -889,6 +898,9 @@ public Response completeMultipartUpload(@PathParam("bucket") String bucket,
 
     OmMultipartUploadCompleteInfo omMultipartUploadCompleteInfo;
     try {
+      OzoneBucket ozoneBucket = volume.getBucket(bucket);
+      S3Owner.verifyBucketOwnerCondition(headers, bucket, ozoneBucket.getOwner());
+
       for (CompleteMultipartUploadRequest.Part part : partList) {
         partsMap.put(part.getPartNumber(), part.getETag());
       }
@@ -990,6 +1002,8 @@ private Response createMultipartKey(OzoneVolume volume, String bucket,
         Pair<String, String> result = parseSourceHeader(copyHeader);
         String sourceBucket = result.getLeft();
         String sourceKey = result.getRight();
+        String sourceBucketOwner = volume.getBucket(sourceBucket).getOwner();
+        S3Owner.verifyBucketOwnerConditionOnCopyOperation(headers, bucket, sourceBucketOwner, ozoneBucket.getOwner());
 
         OzoneKeyDetails sourceKeyDetails = getClientProtocol().getKeyDetails(
             volume.getName(), sourceBucket, sourceKey);
@@ -1236,6 +1250,10 @@ private CopyObjectResponse copyObject(OzoneVolume volume,
     String sourceBucket = result.getLeft();
     String sourceKey = result.getRight();
     DigestInputStream sourceDigestInputStream = null;
+
+    String sourceBucketOwner = volume.getBucket(sourceBucket).getOwner();
+    String destBucketOwner = volume.getBucket(destBucket).getOwner();
+    S3Owner.verifyBucketOwnerConditionOnCopyOperation(headers, destBucket, sourceBucketOwner, destBucketOwner);
     try {
       OzoneKeyDetails sourceKeyDetails = getClientProtocol().getKeyDetails(
           volume.getName(), sourceBucket, sourceKey);

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

@@ -17,10 +17,15 @@
 
 package org.apache.hadoop.ozone.s3.endpoint;
 
+import javax.ws.rs.core.HttpHeaders;
 import javax.xml.bind.annotation.XmlAccessType;
 import javax.xml.bind.annotation.XmlAccessorType;
 import javax.xml.bind.annotation.XmlElement;
 import javax.xml.bind.annotation.XmlRootElement;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.ozone.s3.exception.OS3Exception;
+import org.apache.hadoop.ozone.s3.exception.S3ErrorTable;
+import org.apache.hadoop.ozone.s3.util.S3Consts;
 
 /**
  * Represents an owner of S3 resources in the Ozone S3 compatibility layer.
@@ -80,4 +85,57 @@ public String toString() {
         ", id='" + id + '\'' +
         '}';
   }
+
+  /**
+   * Verify the bucket owner condition.
+   *
+   * @param headers       HTTP headers
+   * @param bucketName    bucket name
+   * @param bucketOwner   bucket owner
+   * @throws OS3Exception if the expected bucket owner does not match
+   */
+  public static void verifyBucketOwnerCondition(HttpHeaders headers, String bucketName, String bucketOwner)
+      throws OS3Exception {
+    if (headers == null || bucketOwner == null) {
+      return;
+    }
+
+    final String expectedBucketOwner = headers.getHeaderString(S3Consts.EXPECTED_BUCKET_OWNER_HEADER);
+
+    if (StringUtils.isEmpty(expectedBucketOwner)) {
+      return;
+    }
+    if (expectedBucketOwner.equals(bucketOwner)) {
+      return;
+    }
+    throw S3ErrorTable.newError(S3ErrorTable.BUCKET_OWNER_MISSMATCH, bucketName);
+  }
+
+  /**
+   * Verify the bucket owner condition on copy operation.
+   *
+   * @param headers       HTTP headers
+   * @param bucketName    bucket name
+   * @param sourceOwner   source bucket owner
+   * @param destOwner     destination bucket owner
+   * @throws OS3Exception if the expected source or destination bucket owner does not match
+   */
+  public static void verifyBucketOwnerConditionOnCopyOperation(HttpHeaders headers, String bucketName,
+                                                               String sourceOwner, String destOwner)
+      throws OS3Exception {
+    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);
+
+    if (expectedSourceOwner != null && sourceOwner != null && !sourceOwner.equals(expectedSourceOwner)) {
+      throw S3ErrorTable.newError(S3ErrorTable.BUCKET_OWNER_MISSMATCH, bucketName);
+    }
+
+    if (expectedDestOwner != null && destOwner != null && !destOwner.equals(expectedDestOwner)) {
+      throw S3ErrorTable.newError(S3ErrorTable.BUCKET_OWNER_MISSMATCH, bucketName);
+    }
+  }
 }

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/exception/S3ErrorTable.java

@@ -156,6 +156,10 @@ public final class S3ErrorTable {
       "a valid custom EC storage config for if using STANDARD_IA.",
       HTTP_BAD_REQUEST);
 
+  public static final OS3Exception BUCKET_OWNER_MISSMATCH = new OS3Exception(
+      "Access Denied", "User doesn't have permission to access this resource due to a " +
+      "bucket ownership mismatch.", HTTP_FORBIDDEN);
+
   private static Function<Exception, OS3Exception> generateInternalError =
       e -> new OS3Exception("InternalError", e.getMessage(), HTTP_INTERNAL_ERROR);
 

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/util/S3Consts.java

@@ -91,6 +91,11 @@ public final class S3Consts {
   public static final Pattern TAG_REGEX_PATTERN = Pattern.compile("^([\\p{L}\\p{Z}\\p{N}_.:/=+\\-]*)$");
   public static final String MP_PARTS_COUNT = "x-amz-mp-parts-count";
 
+  // Bucket owner condition headers
+  // See https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucket-owner-condition.html
+  public static final String EXPECTED_BUCKET_OWNER_HEADER = "x-amz-expected-bucket-owner";
+  public static final String EXPECTED_SOURCE_BUCKET_OWNER_HEADER = "x-amz-source-expected-bucket-owner";
+
   //Never Constructed
   private S3Consts() {
 

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestS3GatewayAuditLog.java

@@ -121,7 +121,7 @@ public static void tearDown() {
   public void testHeadBucket() throws Exception {
     parametersMap.put("bucket", "[bucket]");
 
-    bucketEndpoint.head(bucketName);
+    bucketEndpoint.head(bucketName, null);
 
     String expected = "INFO  | S3GAudit | ? | user=null | ip=null | " +
         "op=HEAD_BUCKET {bucket=[bucket], x-amz-request-id=" + 

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestBucketDelete.java

@@ -19,7 +19,9 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.Mockito.mock;
 
+import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.Response;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.client.ObjectStore;
@@ -41,6 +43,7 @@ public class TestBucketDelete {
   private OzoneClient clientStub;
   private ObjectStore objectStoreStub;
   private BucketEndpoint bucketEndpoint;
+  private HttpHeaders httpHeaders;
 
   @BeforeEach
   public void setup() throws Exception {
@@ -55,21 +58,20 @@ public void setup() throws Exception {
     bucketEndpoint = EndpointBuilder.newBucketEndpointBuilder()
         .setClient(clientStub)
         .build();
-
-
+    httpHeaders = mock(HttpHeaders.class);
   }
 
   @Test
   public void testBucketEndpoint() throws Exception {
-    Response response = bucketEndpoint.delete(bucketName);
+    Response response = bucketEndpoint.delete(bucketName, httpHeaders);
     assertEquals(HttpStatus.SC_NO_CONTENT, response.getStatus());
 
   }
 
   @Test
   public void testDeleteWithNoSuchBucket() throws Exception {
     try {
-      bucketEndpoint.delete("unknownbucket");
+      bucketEndpoint.delete("unknownbucket", httpHeaders);
     } catch (OS3Exception ex) {
       assertEquals(S3ErrorTable.NO_SUCH_BUCKET.getCode(), ex.getCode());
       assertEquals(S3ErrorTable.NO_SUCH_BUCKET.getErrorMessage(),
@@ -84,7 +86,7 @@ public void testDeleteWithBucketNotEmpty() throws Exception {
     try {
       ObjectStoreStub stub = (ObjectStoreStub) objectStoreStub;
       stub.setBucketEmptyStatus(bucketName, false);
-      bucketEndpoint.delete(bucketName);
+      bucketEndpoint.delete(bucketName, httpHeaders);
     } catch (OS3Exception ex) {
       assertEquals(S3ErrorTable.BUCKET_NOT_EMPTY.getCode(), ex.getCode());
       assertEquals(S3ErrorTable.BUCKET_NOT_EMPTY.getErrorMessage(),

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestBucketHead.java

@@ -20,7 +20,9 @@
 import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
 
+import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.Response;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.client.OzoneClient;
@@ -37,11 +39,13 @@ public class TestBucketHead {
   private String bucketName = OzoneConsts.BUCKET;
   private OzoneClient clientStub;
   private BucketEndpoint bucketEndpoint;
+  private HttpHeaders httpHeaders;
 
   @BeforeEach
   public void setup() throws Exception {
     clientStub = new OzoneClientStub();
     clientStub.getObjectStore().createS3Bucket(bucketName);
+    httpHeaders = mock(HttpHeaders.class);
 
     // Create HeadBucket and setClient to OzoneClientStub
     bucketEndpoint = EndpointBuilder.newBucketEndpointBuilder()
@@ -52,15 +56,15 @@ public void setup() throws Exception {
   @Test
   public void testHeadBucket() throws Exception {
 
-    Response response = bucketEndpoint.head(bucketName);
+    Response response = bucketEndpoint.head(bucketName, httpHeaders);
     assertEquals(200, response.getStatus());
 
   }
 
   @Test
   public void testHeadFail() throws Exception {
     OS3Exception e = assertThrows(OS3Exception.class, () ->
-        bucketEndpoint.head("unknownbucket"));
+        bucketEndpoint.head("unknownbucket", httpHeaders));
     assertEquals(HTTP_NOT_FOUND, e.getHttpCode());
     assertEquals("NoSuchBucket", e.getCode());
   }

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestBucketList.java

@@ -525,6 +525,7 @@ public void testEncodingTypeException() throws IOException {
   @Test
   public void testListObjectsWithInvalidMaxKeys() throws Exception {
     OzoneClient client = createClientWithKeys("file1");
+    client.getObjectStore().createS3Bucket("bucket");
     BucketEndpoint bucketEndpoint = EndpointBuilder.newBucketEndpointBuilder()
         .setClient(client)
         .build();