HDDS-14801. [STS] Part 2 - IAM Session Policy and ListBucket improvements

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -2253,14 +2253,17 @@ public OzoneOutputStream createFile(String volumeName, String bucketName,
   }
 
   private OmKeyArgs prepareOmKeyArgs(String volumeName, String bucketName,
-      String keyName) {
-    return new OmKeyArgs.Builder()
+      String keyName, String listPrefix) {
+    final OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
         .setVolumeName(volumeName)
         .setBucketName(bucketName)
         .setKeyName(keyName)
         .setSortDatanodesInPipeline(topologyAwareReadEnabled)
-        .setLatestVersionLocation(getLatestVersionLocation)
-        .build();
+        .setLatestVersionLocation(getLatestVersionLocation);
+    if (listPrefix != null && !listPrefix.isEmpty()) {
+      builder.setListPrefix(listPrefix);
+    }
+    return builder.build();
   }
 
   @Override
@@ -2288,7 +2291,8 @@ public OzoneDataStreamOutput createStreamFile(String volumeName,
   public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
       String keyName, boolean recursive, String startKey, long numEntries)
       throws IOException {
-    OmKeyArgs keyArgs = prepareOmKeyArgs(volumeName, bucketName, keyName);
+    final OmKeyArgs keyArgs = prepareOmKeyArgs(volumeName, bucketName, keyName,
+        null);
     return ozoneManagerClient
         .listStatus(keyArgs, recursive, startKey, numEntries);
   }
@@ -2297,7 +2301,7 @@ public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
   public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
       String keyName, boolean recursive, String startKey,
       long numEntries, boolean allowPartialPrefixes) throws IOException {
-    OmKeyArgs keyArgs = prepareOmKeyArgs(volumeName, bucketName, keyName);
+    final OmKeyArgs keyArgs = prepareOmKeyArgs(volumeName, bucketName, keyName, null);
     return ozoneManagerClient
         .listStatus(keyArgs, recursive, startKey, numEntries,
             allowPartialPrefixes);
@@ -2306,11 +2310,8 @@ public List<OzoneFileStatus> listStatus(String volumeName, String bucketName,
   @Override
   public List<OzoneFileStatusLight> listStatusLight(ListStatusLightOptions options)
       throws IOException {
-    OmKeyArgs keyArgs = prepareOmKeyArgs(options.getVolumeName(),
-        options.getBucketName(), options.getKeyName());
-    if (options.getListPrefix() != null && !options.getListPrefix().isEmpty()) {
-      keyArgs = keyArgs.toBuilder().setListPrefix(options.getListPrefix()).build();
-    }
+    final OmKeyArgs keyArgs = prepareOmKeyArgs(
+        options.getVolumeName(), options.getBucketName(), options.getKeyName(), options.getListPrefix());
     if (omVersion.compareTo(OzoneManagerVersion.LIGHTWEIGHT_LIST_STATUS) >= 0) {
       return ozoneManagerClient.listStatusLight(
           keyArgs, options.isRecursive(), options.getStartKey(), options.getNumEntries(),

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -2369,8 +2369,8 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
         .setLatestVersionLocation(args.getLatestVersionLocation())
         .build();
 
-    ListStatusRequest.Builder listStatusRequestBuilder = createListStatusRequestBuilder(keyArgs, recursive, startKey,
-        numEntries, allowPartialPrefixes, null);
+    final ListStatusRequest.Builder listStatusRequestBuilder = createListStatusRequestBuilder(
+        keyArgs, recursive, startKey, numEntries, allowPartialPrefixes);
 
     OMRequest omRequest = createOMRequest(Type.ListStatus)
         .setListStatusRequest(listStatusRequestBuilder.build())
@@ -2398,8 +2398,12 @@ public List<OzoneFileStatusLight> listStatusLight(OmKeyArgs args,
         .setLatestVersionLocation(true)
         .build();
 
-    ListStatusRequest.Builder listStatusRequestBuilder = createListStatusRequestBuilder(keyArgs, recursive, startKey,
-        numEntries, allowPartialPrefixes, args.getListPrefix());
+    final ListStatusRequest.Builder listStatusRequestBuilder = createListStatusRequestBuilder(
+        keyArgs, recursive, startKey, numEntries, allowPartialPrefixes);
+    final String listPrefix = args.getListPrefix();
+    if (listPrefix != null && !listPrefix.isEmpty()) {
+      listStatusRequestBuilder.setListPrefix(listPrefix);
+    }
 
     OMRequest omRequest = createOMRequest(Type.ListStatusLight)
         .setListStatusRequest(listStatusRequestBuilder.build())
@@ -2417,7 +2421,7 @@ public List<OzoneFileStatusLight> listStatusLight(OmKeyArgs args,
   }
 
   private ListStatusRequest.Builder createListStatusRequestBuilder(KeyArgs keyArgs, boolean recursive, String startKey,
-      long numEntries, boolean allowPartialPrefixes, String listPrefix) {
+      long numEntries, boolean allowPartialPrefixes) {
     ListStatusRequest.Builder listStatusRequestBuilder =
         ListStatusRequest.newBuilder()
             .setKeyArgs(keyArgs)
@@ -2433,9 +2437,6 @@ private ListStatusRequest.Builder createListStatusRequestBuilder(KeyArgs keyArgs
     if (allowPartialPrefixes) {
       listStatusRequestBuilder.setAllowPartialPrefix(allowPartialPrefixes);
     }
-    if (listPrefix != null && !listPrefix.isEmpty()) {
-      listStatusRequestBuilder.setListPrefix(listPrefix);
-    }
     return listStatusRequestBuilder;
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataReader.java

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.ipc_.ProtobufRpcEngine;
 import org.apache.hadoop.ipc_.Server;
@@ -234,8 +235,34 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
 
     try {
       if (isAclEnabled) {
-        checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ,
-            bucket, args.getKeyName());
+        if (isStsS3Request()) {
+          // We need to be able to tell the difference between being able to download a file and merely seeing the file
+          // name in a list.  Use READ for download ability and LIST (here) for listing.
+          // When listPrefix is set (original S3 ListObjects prefix), authorize LIST on that prefix for the whole
+          // listing, including FSO traversal where keyName is an internal directory (e.g. userA) under prefix user.
+          final String listPrefix = args.getListPrefix();
+          final String keyName = args.getKeyName();
+          final String aclKey;
+          if (StringUtils.isNotBlank(listPrefix)) {
+            if (StringUtils.isBlank(keyName)) {
+              aclKey = listPrefix;
+            } else if (isStsListPathUnderRequestPrefix(keyName, listPrefix)) {
+              aclKey = listPrefix;
+            } else {
+              throw new OMException(
+                  "STS listStatus: key path: " + keyName + " does not match authorized list prefix: " + listPrefix,
+                  ResultCodes.PERMISSION_DENIED);
+            }
+          } else if (keyName != null && !keyName.isEmpty()) {
+            aclKey = keyName;
+          } else {
+            aclKey = "*";
+          }
+          checkAcls(ResourceType.KEY, StoreType.OZONE, ACLType.LIST, bucket.realVolume(), bucket.realBucket(), aclKey);
+        } else {
+          checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ,
+              bucket, args.getKeyName());
+        }
       }
       metrics.incNumListStatus();
       return keyManager.listStatus(args, recursive, startKey,
@@ -277,8 +304,12 @@ public OzoneFileStatus getFileStatus(OmKeyArgs args) throws IOException {
 
     try {
       if (isAclEnabled) {
-        checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ,
-            bucket, args.getKeyName());
+        if (isStsS3Request()) {
+          checkAcls(getResourceType(args), StoreType.OZONE, ACLType.LIST, bucket, args.getKeyName());
+        } else {
+          checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ,
+              bucket, args.getKeyName());
+        }
       }
       metrics.incNumGetFileStatus();
       return keyManager.getFileStatus(args, getClientAddress());
@@ -343,10 +374,23 @@ public ListKeysResult listKeys(String volumeName, String bucketName,
 
     try {
       if (isAclEnabled) {
-        captureLatencyNs(perfMetrics.getListKeysAclCheckLatencyNs(), () ->
-            checkAcls(ResourceType.BUCKET, StoreType.OZONE, ACLType.LIST,
-            bucket.realVolume(), bucket.realBucket(), keyPrefix)
-        );
+        if (isStsS3Request()) {
+          // Check with key null to ensure there is LIST permission on the bucket
+          captureLatencyNs(
+              perfMetrics.getListKeysAclCheckLatencyNs(), () -> checkAcls(
+                  ResourceType.BUCKET, StoreType.OZONE, ACLType.LIST, bucket.realVolume(), bucket.realBucket(),
+                  null));
+          // With STS we must check acl on the prefix to be compliant with AWS
+          final String aclKey = (keyPrefix == null || keyPrefix.isEmpty()) ? "*" : keyPrefix;
+          captureLatencyNs(
+              perfMetrics.getListKeysAclCheckLatencyNs(), () -> checkAcls(
+                  ResourceType.KEY, StoreType.OZONE, ACLType.LIST, bucket.realVolume(), bucket.realBucket(), aclKey));
+        } else {
+          captureLatencyNs(perfMetrics.getListKeysAclCheckLatencyNs(), () ->
+              checkAcls(ResourceType.BUCKET, StoreType.OZONE, ACLType.LIST,
+                  bucket.realVolume(), bucket.realBucket(), keyPrefix)
+          );
+        }
       }
       metrics.incNumKeyLists();
       return keyManager.listKeys(bucket.realVolume(), bucket.realBucket(),
@@ -698,6 +742,29 @@ public boolean isNativeAuthorizerEnabled() {
     return accessAuthorizer.isNative();
   }
 
+  private boolean isStsS3Request() {
+    return getS3Auth() != null && OzoneManager.getStsTokenIdentifier() != null;
+  }
+
+  /**
+   * For STS, {@code listPrefix} is the original S3 ListObjects prefix. Internal FSO listing may call
+   * {@code listStatus} with {@code keyName} set to a subdirectory (e.g. userA) while the session policy
+   * still authorizes only the request prefix (e.g. user). This returns true when {@code keyName} is the
+   * prefix itself, a key path under that prefix, or an ancestor directory on the way to a deeper prefix.
+   */
+  private static boolean isStsListPathUnderRequestPrefix(String keyName, String listPrefix) {
+    if (StringUtils.isBlank(listPrefix) || StringUtils.isBlank(keyName)) {
+      return false;
+    }
+    if (keyName.equals(listPrefix)) {
+      return true;
+    }
+    if (keyName.startsWith(listPrefix)) {
+      return true;
+    }
+    return listPrefix.startsWith(keyName + "/");
+  }
+
   private ResourceType getResourceType(OmKeyArgs args) {
     if (args.getKeyName() == null || args.getKeyName().isEmpty()) {
       return ResourceType.BUCKET;