ensure each S3 API has an associated S3 Action

Author: Fabian Morgan

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

@@ -328,6 +328,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());
     }
@@ -1703,10 +1706,8 @@ 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

@@ -1572,7 +1572,6 @@ message CommitKeyRequest {
 }
 
 message CommitKeyResponse {
-
 }
 
 message AllocateBlockRequest {
@@ -2319,6 +2318,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/interface-client/src/main/resources/proto.lock

@@ -7683,6 +7683,12 @@
                 "name": "accessId",
                 "type": "string",
                 "optional": true
+              },
+              {
+                "id": 10,
+                "name": "s3Action",
+                "type": "string",
+                "optional": true
               }
             ]
           },

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,
@@ -634,7 +628,8 @@ public boolean checkAcls(ResourceType resType, StoreType storeType,
   public boolean checkAcls(OzoneObj obj, RequestContext context,
       boolean throwIfPermissionDenied) throws OMException {
 
-    final RequestContext normalizedRequestContext = maybeAttachSessionPolicyFromThreadLocal(context);
+    final RequestContext normalizedRequestContext = maybeAttachS3ActionFromThreadLocal(
+        maybeAttachSessionPolicyFromThreadLocal(context));
 
     if (!captureLatencyNs(perfMetrics::setCheckAccessLatencyNs,
         () -> accessAuthorizer.checkAccess(obj, normalizedRequestContext))) {
@@ -692,6 +687,22 @@ private RequestContext maybeAttachSessionPolicyFromThreadLocal(RequestContext co
         .build();
   }
 
+  /**
+   * Attaches s3 action to RequestContext if an S3Authentication is found in the Ozone Manager thread local,
+   * and it has an s3 action.  Otherwise, returns the RequestContext as it was before.
+   * @param context the original RequestContext
+   * @return RequestContext as before or with s3 action embedded
+   */
+  private RequestContext maybeAttachS3ActionFromThreadLocal(RequestContext context) {
+    final S3Authentication s3Authentication = OzoneManager.getS3Auth();
+    if (s3Authentication == null || !s3Authentication.hasS3Action()) {
+      return context;
+    }
+    return context.toBuilder()
+        .setS3Action(s3Authentication.getS3Action())
+        .build();
+  }
+
   static String getClientAddress() {
     String clientMachine = Server.getRemoteAddress();
     if (clientMachine == null) { //not a RPC client

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

@@ -619,6 +619,7 @@ public void close() {
    */
   @VisibleForTesting
   OMResponse runCommand(OMRequest request, TermIndex termIndex) {
+    boolean isS3AuthThreadLocalSet = false;
     boolean isStsThreadLocalSet = false;
     try {
       if (ozoneManager.isSecurityEnabled() && request.hasS3Authentication()) {
@@ -627,6 +628,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(
@@ -661,6 +666,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);
       }