HDDS-15137. [STS] Ensure each S3 API has an associated S3 Action

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/S3Auth.java

@@ -29,6 +29,8 @@ public class S3Auth {
   private String userPrincipal;
   // Optional STS session token when using temporary credentials
   private String sessionToken;
+  // S3 action without s3: prefix (e.g. PutObject), set by S3 Gateway for use in finer-grained STS permissions.
+  private String s3Action;
 
   public S3Auth(final String stringToSign,
                 final String signature,
@@ -67,4 +69,12 @@ public String getSessionToken() {
   public void setSessionToken(String sessionToken) {
     this.sessionToken = sessionToken;
   }
+
+  public String getS3Action() {
+    return s3Action;
+  }
+
+  public void setS3Action(String s3Action) {
+    this.s3Action = s3Action;
+  }
 }

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

@@ -332,6 +332,9 @@ private OMResponse submitRequest(OMRequest omRequest)
       if (threadLocalS3Auth.get().getSessionToken() != null) {
         s3AuthBuilder.setSessionToken(threadLocalS3Auth.get().getSessionToken());
       }
+      if (threadLocalS3Auth.get().getS3Action() != null) {
+        s3AuthBuilder.setS3Action(threadLocalS3Auth.get().getS3Action());
+      }
 
       builder.setS3Authentication(s3AuthBuilder.build());
     }
@@ -1788,10 +1791,7 @@ public OmMultipartCommitUploadPartInfo commitMultipartUploadPart(
         handleError(submitRequest(omRequest))
         .getCommitMultiPartUploadResponse();
 
-    OmMultipartCommitUploadPartInfo info = new
-        OmMultipartCommitUploadPartInfo(response.getPartName(),
-          response.getETag());
-    return info;
+    return new OmMultipartCommitUploadPartInfo(response.getPartName(), response.getETag());
   }
 
   @Override

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -2346,6 +2346,9 @@ message S3Authentication {
     optional string resolvedStsOriginalAccessKeyId = 7;
     optional string resolvedStsTempAccessKeyId = 8;
     optional string resolvedStsSecretKeyId = 9;
+    // S3 action without the s3: prefix for this request (e.g. GetObject), set by S3 Gateway for use
+    // in finer-grained STS permissions.
+    optional string s3Action = 10;
 }
 
 message RecoverLeaseRequest {

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

@@ -55,6 +55,7 @@
 import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight;
 import org.apache.hadoop.ozone.om.helpers.S3VolumeContext;
 import org.apache.hadoop.ozone.om.protocolPB.grpc.GrpcClientConstants;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.S3Authentication;
 import org.apache.hadoop.ozone.security.STSTokenIdentifier;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
@@ -236,9 +237,7 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
     try {
       if (isAclEnabled) {
         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
+          // When listPrefix is set (original S3 ListObjects prefix), authorize READ 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();
@@ -258,7 +257,7 @@ public List<OzoneFileStatus> listStatus(OmKeyArgs args, boolean recursive,
           } else {
             aclKey = "*";
           }
-          checkAcls(ResourceType.KEY, StoreType.OZONE, ACLType.LIST, bucket.realVolume(), bucket.realBucket(), aclKey);
+          checkAcls(ResourceType.KEY, StoreType.OZONE, ACLType.READ, bucket.realVolume(), bucket.realBucket(), aclKey);
         } else {
           checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ,
               bucket, args.getKeyName());
@@ -304,12 +303,7 @@ public OzoneFileStatus getFileStatus(OmKeyArgs args) throws IOException {
 
     try {
       if (isAclEnabled) {
-        if (isStsS3Request()) {
-          checkAcls(getResourceType(args), StoreType.OZONE, ACLType.LIST, bucket, args.getKeyName());
-        } else {
-          checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ,
-              bucket, args.getKeyName());
-        }
+        checkAcls(getResourceType(args), StoreType.OZONE, ACLType.READ, bucket, args.getKeyName());
       }
       metrics.incNumGetFileStatus();
       return keyManager.getFileStatus(args, getClientAddress());
@@ -384,7 +378,7 @@ public ListKeysResult listKeys(String volumeName, String bucketName,
           final String aclKey = (keyPrefix == null || keyPrefix.isEmpty()) ? "*" : keyPrefix;
           captureLatencyNs(
               perfMetrics.getListKeysAclCheckLatencyNs(), () -> checkAcls(
-                  ResourceType.KEY, StoreType.OZONE, ACLType.LIST, bucket.realVolume(), bucket.realBucket(), aclKey));
+                  ResourceType.KEY, StoreType.OZONE, ACLType.READ, bucket.realVolume(), bucket.realBucket(), aclKey));
         } else {
           captureLatencyNs(perfMetrics.getListKeysAclCheckLatencyNs(), () ->
               checkAcls(ResourceType.BUCKET, StoreType.OZONE, ACLType.LIST,
@@ -612,16 +606,15 @@ public boolean checkAcls(ResourceType resType, StoreType storeType,
         .setVolumeName(vol)
         .setBucketName(bucket)
         .setKeyName(key).build();
-    RequestContext context = RequestContext.newBuilder()
+    RequestContext.Builder contextBuilder = RequestContext.newBuilder()
         .setClientUgi(ugi)
         .setIp(remoteAddress)
         .setHost(hostName)
         .setAclType(ACLIdentityType.USER)
         .setAclRights(aclType)
-        .setOwnerName(owner)
-        .build();
+        .setOwnerName(owner);
 
-    return checkAcls(obj, context, throwIfPermissionDenied);
+    return checkAcls(obj, contextBuilder, throwIfPermissionDenied);
   }
 
   /**
@@ -631,13 +624,14 @@ public boolean checkAcls(ResourceType resType, StoreType storeType,
    * @throws OMException ResultCodes.PERMISSION_DENIED if permission denied
    *                     and throwOnPermissionDenied set to true.
    */
-  public boolean checkAcls(OzoneObj obj, RequestContext context,
+  public boolean checkAcls(OzoneObj obj, RequestContext.Builder contextBuilder,
       boolean throwIfPermissionDenied) throws OMException {
 
-    final RequestContext normalizedRequestContext = maybeAttachSessionPolicyFromThreadLocal(context);
+    maybeAddToContextFromThreadLocal(contextBuilder);
+    final RequestContext context = contextBuilder.build();
 
     if (!captureLatencyNs(perfMetrics::setCheckAccessLatencyNs,
-        () -> accessAuthorizer.checkAccess(obj, normalizedRequestContext))) {
+        () -> accessAuthorizer.checkAccess(obj, context))) {
       if (throwIfPermissionDenied) {
         String volumeName = obj.getVolumeName() != null ?
                 "Volume:" + obj.getVolumeName() + " " : "";
@@ -647,7 +641,7 @@ public boolean checkAcls(OzoneObj obj, RequestContext context,
                 "Key:" + obj.getKeyName() : "";
         // For STS tokens, make clear that the user is using an assumed role, otherwise the access denied
         // message could be confusing
-        String user = normalizedRequestContext.getClientUgi().getShortUserName();
+        String user = context.getClientUgi().getShortUserName();
         final STSTokenIdentifier stsTokenIdentifier = OzoneManager.getStsTokenIdentifier();
         if (stsTokenIdentifier != null) {
           final StringBuilder builder = new StringBuilder(user)
@@ -660,11 +654,11 @@ public boolean checkAcls(OzoneObj obj, RequestContext context,
         }
         log.warn("User {} doesn't have {} permission to access {} {}{}{}",
             user,
-            normalizedRequestContext.getAclRights(),
+            context.getAclRights(),
             obj.getResourceType(), volumeName, bucketName, keyName);
         throw new OMException(
             "User " + user +
-            " doesn't have " + normalizedRequestContext.getAclRights() +
+            " doesn't have " + context.getAclRights() +
             " permission to access " + obj.getResourceType() + " " +
             volumeName  + bucketName + keyName, ResultCodes.PERMISSION_DENIED);
       }
@@ -675,21 +669,22 @@ public boolean checkAcls(OzoneObj obj, RequestContext context,
   }
 
   /**
-   * Attaches session policy to RequestContext if an STSTokenIdentifier is found in the Ozone Manager thread local
-   * (meaning this is an STS request), and the STSTokenIdentifier has a session policy.  Otherwise, returns the
-   * RequestContext as it was before.
-   * @param context the original RequestContext
-   * @return RequestContext as before or with sessionPolicy embedded
+   * Enriches the given {@link RequestContext.Builder} with per-request fields from the Ozone Manager
+   * thread locals: the session policy from {@link STSTokenIdentifier} (set on STS requests) and the
+   * S3 action from {@link S3Authentication} (set on S3 requests). Either or both may be absent, in
+   * which case the corresponding field is left untouched on the builder.
+   * @param contextBuilder the builder to enrich in-place
    */
-  private RequestContext maybeAttachSessionPolicyFromThreadLocal(RequestContext context) {
+  public static void maybeAddToContextFromThreadLocal(RequestContext.Builder contextBuilder) {
     final STSTokenIdentifier stsTokenIdentifier = OzoneManager.getStsTokenIdentifier();
-    if (stsTokenIdentifier == null) {
-      return context;
+    if (stsTokenIdentifier != null) {
+      contextBuilder.setSessionPolicy(stsTokenIdentifier.getSessionPolicy());
     }
 
-    return context.toBuilder()
-        .setSessionPolicy(stsTokenIdentifier.getSessionPolicy())
-        .build();
+    final S3Authentication s3Authentication = OzoneManager.getS3Auth();
+    if (s3Authentication != null && s3Authentication.hasS3Action() && !s3Authentication.getS3Action().isEmpty()) {
+      contextBuilder.setS3Action(s3Authentication.getS3Action());
+    }
   }
 
   static String getClientAddress() {

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

@@ -2891,16 +2891,15 @@ public boolean checkAcls(ResourceType resType, StoreType storeType,
         .setVolumeName(vol)
         .setBucketName(bucket)
         .setKeyName(key).build();
-    RequestContext context = RequestContext.newBuilder()
+    RequestContext.Builder contextBuilder = RequestContext.newBuilder()
         .setClientUgi(ugi)
         .setIp(remoteAddress)
         .setHost(hostName)
         .setAclType(ACLIdentityType.USER)
         .setAclRights(aclType)
-        .setOwnerName(owner)
-        .build();
+        .setOwnerName(owner);
 
-    return omMetadataReader.checkAcls(obj, context, throwIfPermissionDenied);
+    return omMetadataReader.checkAcls(obj, contextBuilder, throwIfPermissionDenied);
   }
 
   /**

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

@@ -664,6 +664,7 @@ public void close() {
    */
   @VisibleForTesting
   OMResponse runCommand(OMRequest request, TermIndex termIndex) {
+    boolean isS3AuthThreadLocalSet = false;
     boolean isStsThreadLocalSet = false;
     try {
       if (ozoneManager.isSecurityEnabled() && request.hasS3Authentication()) {
@@ -672,6 +673,10 @@ OMResponse runCommand(OMRequest request, TermIndex termIndex) {
         STSSecurityUtil.ensureResolvedStsFieldsInvariants(request);
 
         final OzoneManagerProtocolProtos.S3Authentication s3Auth = request.getS3Authentication();
+        // ThreadLocal carries S3 action for OmMetadataReader.
+        OzoneManager.setS3Auth(s3Auth);
+        isS3AuthThreadLocalSet = true;
+
         if (s3Auth.hasSessionToken() && !s3Auth.getSessionToken().isEmpty()) {
           // ThreadLocal carries session policy for OmMetadataReader
           final STSTokenIdentifier rehydratedTokenIdentifier = new STSTokenIdentifier(
@@ -706,6 +711,9 @@ OMResponse runCommand(OMRequest request, TermIndex termIndex) {
       String errorMessage = "Request " + request + " failed with exception";
       ExitUtils.terminate(1, errorMessage, e, LOG);
     } finally {
+      if (isS3AuthThreadLocalSet) {
+        OzoneManager.setS3Auth(null);
+      }
       if (isStsThreadLocalSet) {
         OzoneManager.setStsTokenIdentifier(null);
       }

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

@@ -385,7 +385,7 @@ protected void checkACLsWithFSO(OzoneManager ozoneManager, String volumeName,
         OmMetadataReader omMetadataReader =
             (OmMetadataReader) rcMetadataReader.get();
 
-        omMetadataReader.checkAcls(obj, contextBuilder.build(), true);
+        omMetadataReader.checkAcls(obj, contextBuilder, true);
       }
     }
   }