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$S3BucketOwnershipVerificationConditionsTests"/>
+    <Bug pattern="SIC_INNER_SHOULD_BE_STATIC"/>
+  </Match>
 </FindBugsFilter>

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

@@ -93,6 +93,9 @@ public class BucketEndpoint extends EndpointBase {
   private static final Logger LOG =
       LoggerFactory.getLogger(BucketEndpoint.class);
 
+  @Context
+  private HttpHeaders headers;
+
   private boolean listKeysShallowEnabled;
   private int maxKeysLimit = 1000;
 
@@ -120,8 +123,7 @@ public Response get(
       @QueryParam("acl") String aclMarker,
       @QueryParam("key-marker") String keyMarker,
       @QueryParam("upload-id-marker") String uploadIdMarker,
-      @DefaultValue("1000") @QueryParam("max-uploads") int maxUploads,
-      @Context HttpHeaders hh) throws OS3Exception, IOException {
+      @DefaultValue("1000") @QueryParam("max-uploads") int maxUploads) throws OS3Exception, IOException {
     long startNanos = Time.monotonicNowNanos();
     S3GAction s3GAction = S3GAction.GET_BUCKET;
     PerformanceStringBuilder perf = new PerformanceStringBuilder();
@@ -132,6 +134,8 @@ public Response get(
     OzoneBucket bucket = null;
 
     try {
+      bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       if (aclMarker != null) {
         s3GAction = S3GAction.GET_ACL;
         S3BucketAcl result = getAcl(bucketName);
@@ -167,7 +171,6 @@ public Response get(
       boolean shallow = listKeysShallowEnabled
           && OZONE_URI_DELIMITER.equals(delimiter);
 
-      bucket = getBucket(bucketName);
       ozoneKeyIterator = bucket.listKeys(prefix, prevKey, shallow);
 
     } catch (OMException ex) {
@@ -309,15 +312,14 @@ private int validateMaxKeys(int maxKeys) throws OS3Exception {
   @PUT
   public Response put(@PathParam("bucket") String bucketName,
                       @QueryParam("acl") String aclMarker,
-                      @Context HttpHeaders httpHeaders,
                       InputStream body) throws IOException, OS3Exception {
     long startNanos = Time.monotonicNowNanos();
     S3GAction s3GAction = S3GAction.CREATE_BUCKET;
 
     try {
       if (aclMarker != null) {
         s3GAction = S3GAction.PUT_ACL;
-        Response response =  putAcl(bucketName, httpHeaders, body);
+        Response response =  putAcl(bucketName, body);
         AUDIT.logWriteSuccess(
             buildAuditMessageForSuccess(s3GAction, getAuditParameters()));
         return response;
@@ -413,7 +415,8 @@ public Response head(@PathParam("bucket") String bucketName)
     long startNanos = Time.monotonicNowNanos();
     S3GAction s3GAction = S3GAction.HEAD_BUCKET;
     try {
-      getBucket(bucketName);
+      OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       AUDIT.logReadSuccess(
           buildAuditMessageForSuccess(s3GAction, getAuditParameters()));
       getMetrics().updateHeadBucketSuccessStats(startNanos);
@@ -438,6 +441,8 @@ public Response delete(@PathParam("bucket") String bucketName)
     S3GAction s3GAction = S3GAction.DELETE_BUCKET;
 
     try {
+      OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       deleteS3Bucket(bucketName);
     } catch (OMException ex) {
       AUDIT.logWriteFailure(
@@ -492,6 +497,7 @@ public MultiDeleteResponse multiDelete(@PathParam("bucket") String bucketName,
       }
       long startNanos = Time.monotonicNowNanos();
       try {
+        S3Owner.verifyBucketOwnerCondition(headers, 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
@@ -581,17 +584,18 @@ public S3BucketAcl getAcl(String bucketName)
    * <p>
    * see: https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketAcl.html
    */
-  public Response putAcl(String bucketName, HttpHeaders httpHeaders,
+  public Response putAcl(String bucketName,
                          InputStream body) throws IOException, OS3Exception {
     long startNanos = Time.monotonicNowNanos();
-    String grantReads = httpHeaders.getHeaderString(S3Acl.GRANT_READ);
-    String grantWrites = httpHeaders.getHeaderString(S3Acl.GRANT_WRITE);
-    String grantReadACP = httpHeaders.getHeaderString(S3Acl.GRANT_READ_CAP);
-    String grantWriteACP = httpHeaders.getHeaderString(S3Acl.GRANT_WRITE_CAP);
-    String grantFull = httpHeaders.getHeaderString(S3Acl.GRANT_FULL_CONTROL);
+    String grantReads = headers.getHeaderString(S3Acl.GRANT_READ);
+    String grantWrites = headers.getHeaderString(S3Acl.GRANT_WRITE);
+    String grantReadACP = headers.getHeaderString(S3Acl.GRANT_READ_CAP);
+    String grantWriteACP = headers.getHeaderString(S3Acl.GRANT_WRITE_CAP);
+    String grantFull = headers.getHeaderString(S3Acl.GRANT_FULL_CONTROL);
 
     try {
       OzoneBucket bucket = getBucket(bucketName);
+      S3Owner.verifyBucketOwnerCondition(headers, bucketName, bucket.getOwner());
       OzoneVolume volume = getVolume();
 
       List<OzoneAcl> ozoneAclListOnBucket = new ArrayList<>();
@@ -770,6 +774,11 @@ public OzoneConfiguration getOzoneConfiguration() {
     return this.ozoneConfiguration;
   }
 
+  @VisibleForTesting
+  public void setHeaders(HttpHeaders headers) {
+    this.headers = headers;
+  }
+
   @Override
   @PostConstruct
   public void init() {

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,16 +898,17 @@ 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());
       }
       if (LOG.isDebugEnabled()) {
         LOG.debug("Parts map {}", partsMap);
       }
 
-      omMultipartUploadCompleteInfo = getClientProtocol()
-          .completeMultipartUpload(volume.getName(), bucket, key, uploadID,
-              partsMap);
+      omMultipartUploadCompleteInfo = ozoneBucket.completeMultipartUpload(key, uploadID, partsMap);
       CompleteMultipartUploadResponse completeMultipartUploadResponse =
           new CompleteMultipartUploadResponse();
       completeMultipartUploadResponse.setBucket(bucket);
@@ -990,6 +1000,9 @@ 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, sourceBucket, sourceBucketOwner, bucket,
+            ozoneBucket.getOwner());
 
         OzoneKeyDetails sourceKeyDetails = getClientProtocol().getKeyDetails(
             volume.getName(), sourceBucket, sourceKey);
@@ -1236,6 +1249,10 @@ private CopyObjectResponse copyObject(OzoneVolume volume,
     String sourceBucket = result.getLeft();
     String sourceKey = result.getRight();
     DigestInputStream sourceDigestInputStream = null;
+
+    String sourceBucketOwner = volume.getBucket(sourceBucket).getOwner();
+    // The destBucket owner has already been checked in the caller method
+    S3Owner.verifyBucketOwnerConditionOnCopyOperation(headers, sourceBucket, sourceBucketOwner, null, null);
     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,59 @@ 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_MISMATCH, bucketName);
+  }
+
+  /**
+   * Verify the bucket owner condition on copy operation.
+   *
+   * @param headers          HTTP headers
+   * @param sourceBucketName source bucket name
+   * @param sourceOwner      source bucket owner
+   * @param destBucketName   dest bucket name
+   * @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 sourceBucketName,
+                                                               String sourceOwner, String destBucketName,
+                                                               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_MISMATCH, sourceBucketName);
+    }
+
+    if (expectedDestOwner != null && destOwner != null && !destOwner.equals(expectedDestOwner)) {
+      throw S3ErrorTable.newError(S3ErrorTable.BUCKET_OWNER_MISMATCH, destBucketName);
+    }
+  }
 }

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_MISMATCH = 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/endpoint/BucketEndpointBuilder.java

@@ -31,6 +31,7 @@ public BucketEndpointBuilder() {
   public BucketEndpoint build() {
     BucketEndpoint endpoint = super.build();
     endpoint.setOzoneConfiguration(getConfig());
+    endpoint.setHeaders(getHeaders());
 
     return endpoint;
   }