HDDS-15853. Fix Ranger ACL validation for lifecycle requests

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

@@ -115,7 +115,7 @@ public OMRequest preExecute(OzoneManager ozoneManager)
         .setVersion(ozoneManager.getVersionManager().getMetadataLayoutVersion())
         .build();
     omRequest = getOmRequest().toBuilder()
-        .setUserInfo(getUserInfo())
+        .setUserInfo(getUserIfNotExists(ozoneManager))
         .setLayoutVersion(layoutVersion).build();
     return omRequest;
   }
@@ -193,18 +193,41 @@ public OzoneManagerProtocolProtos.UserInfo getUserInfo() throws IOException {
         && grpcContextClientIpAddress != null) {
       userInfo.setHostName(grpcContextClientHostname);
       userInfo.setRemoteAddress(grpcContextClientIpAddress);
-    } else if (omRequest.hasUserInfo()
-        && omRequest.getUserInfo().hasRemoteAddress()) {
-      // For non-RPC internal service requests (e.g. the Trash emptier) that
-      // populate their own UserInfo, preserve the supplied host/address since
-      // there is no RPC or gRPC client context to derive it from.
-      userInfo.setHostName(omRequest.getUserInfo().getHostName());
-      userInfo.setRemoteAddress(omRequest.getUserInfo().getRemoteAddress());
     }
 
     return userInfo.build();
   }
 
+  /**
+   * For non-rpc internal calls Server.getRemoteUser()
+   * and Server.getRemoteIp() will be null.
+   * Passing getCurrentUser() and Ip of the Om node that started it.
+   * @return User Info.
+   */
+  public OzoneManagerProtocolProtos.UserInfo getUserIfNotExists(
+      OzoneManager ozoneManager) throws IOException {
+    OzoneManagerProtocolProtos.UserInfo userInfo = getUserInfo();
+    if (!userInfo.hasRemoteAddress() || !userInfo.hasUserName()) {
+      OzoneManagerProtocolProtos.UserInfo.Builder newuserInfo =
+          OzoneManagerProtocolProtos.UserInfo.newBuilder();
+      UserGroupInformation user;
+      InetAddress remoteAddress;
+      try {
+        user = UserGroupInformation.getCurrentUser();
+        remoteAddress = ozoneManager.getOmRpcServerAddr()
+            .getAddress();
+      } catch (Exception e) {
+        LOG.debug("Couldn't get om Rpc server address", e);
+        return getUserInfo();
+      }
+      newuserInfo.setUserName(user.getUserName());
+      newuserInfo.setHostName(remoteAddress.getHostName());
+      newuserInfo.setRemoteAddress(remoteAddress.getHostAddress());
+      return newuserInfo.build();
+    }
+    return getUserInfo();
+  }
+
   /**
    * Check Acls of ozone object.
    * @param ozoneManager

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

@@ -102,7 +102,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     //  BlockOutputStreamEntryPool, so we are fine for now. But if one some
     //  one uses direct omclient we might be in trouble.
 
-    UserInfo userInfo = getOmRequest().getUserInfo();
+    UserInfo userInfo = getUserIfNotExists(ozoneManager);
     ReplicationConfig repConfig = ReplicationConfig.fromProto(keyArgs.getType(),
         keyArgs.getFactor(), keyArgs.getEcReplicationConfig());
     // To allocate atleast one block passing requested size and scmBlockSize
@@ -133,7 +133,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     newAllocatedBlockRequest.setKeyLocation(
         omKeyLocationInfoList.get(0).getProtobuf(getOmRequest().getVersion()));
 
-    return getOmRequest().toBuilder()
+    return getOmRequest().toBuilder().setUserInfo(userInfo)
         .setAllocateBlockRequest(newAllocatedBlockRequest).build();
 
   }

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

@@ -92,7 +92,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     return getOmRequest().toBuilder()
         .setDeleteKeyRequest(deleteKeyRequest.toBuilder()
             .setKeyArgs(resolvedArgs))
-        .build();
+        .setUserInfo(getUserIfNotExists(ozoneManager)).build();
   }
 
   protected KeyArgs resolveBucketAndCheckAcls(OzoneManager ozoneManager,

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

@@ -95,7 +95,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     return getOmRequest().toBuilder()
         .setRenameKeyRequest(renameKeyRequest.toBuilder().setToKeyName(dstKey)
             .setKeyArgs(resolvedArgs))
-        .build();
+        .setUserInfo(getUserIfNotExists(ozoneManager)).build();
 
   }
 

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

@@ -49,6 +49,7 @@
 import org.apache.hadoop.ozone.request.validation.RequestProcessingPhase;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -77,9 +78,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
         Pair.of(volumeName, bucketName), this);
 
     if (ozoneManager.getAclsEnabled()) {
-      checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET, OzoneObj.StoreType.OZONE,
-          IAccessAuthorizer.ACLType.ALL, resolvedBucket.realVolume(), 
-          resolvedBucket.realBucket(), null);
+      checkAclPermission(ozoneManager, resolvedBucket.realVolume(), resolvedBucket.realBucket());
     }
 
     // Update the request with resolved volume and bucket names
@@ -175,6 +174,22 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
     }
   }
 
+  private void checkAclPermission(OzoneManager ozoneManager, String volumeName, String bucketName)
+      throws IOException {
+    if (ozoneManager.getAccessAuthorizer().isNative()) {
+      UserGroupInformation ugi = createUGIForApi();
+      String bucketOwner = ozoneManager.getBucketOwner(volumeName, bucketName,
+          IAccessAuthorizer.ACLType.READ, OzoneObj.ResourceType.BUCKET);
+      if (!ozoneManager.isAdmin(ugi) && !ozoneManager.isOwner(ugi, bucketOwner)) {
+        throw new OMException("Lifecycle configuration can only be deleted by cluster Admin or bucket Owner",
+            OMException.ResultCodes.PERMISSION_DENIED);
+      }
+    } else {
+      checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET, OzoneObj.StoreType.OZONE,
+          IAccessAuthorizer.ACLType.WRITE, volumeName, bucketName, null);
+    }
+  }
+
   @RequestFeatureValidator(
       conditions = ValidationCondition.CLUSTER_NEEDS_FINALIZATION,
       processingPhase = RequestProcessingPhase.PRE_PROCESS,

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

@@ -54,6 +54,7 @@
 import org.apache.hadoop.ozone.request.validation.RequestProcessingPhase;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -86,9 +87,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
         Pair.of(volumeName, bucketName), this);
 
     if (ozoneManager.getAclsEnabled()) {
-      checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET, OzoneObj.StoreType.OZONE,
-          IAccessAuthorizer.ACLType.ALL, resolvedBucket.realVolume(), 
-          resolvedBucket.realBucket(), null);
+      checkAclPermission(ozoneManager, resolvedBucket.realVolume(), resolvedBucket.realBucket());
     }
 
     if (resolvedBucket.bucketLayout().toProto() != request.getLifecycleConfiguration().getBucketLayout()) {
@@ -204,6 +203,22 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
     }
   }
 
+  private void checkAclPermission(OzoneManager ozoneManager, String volumeName, String bucketName)
+      throws IOException {
+    if (ozoneManager.getAccessAuthorizer().isNative()) {
+      UserGroupInformation ugi = createUGIForApi();
+      String bucketOwner = ozoneManager.getBucketOwner(volumeName, bucketName,
+          IAccessAuthorizer.ACLType.READ, OzoneObj.ResourceType.BUCKET);
+      if (!ozoneManager.isAdmin(ugi) && !ozoneManager.isOwner(ugi, bucketOwner)) {
+        throw new OMException("Lifecycle configuration can only be set by cluster Admin or bucket Owner",
+            OMException.ResultCodes.PERMISSION_DENIED);
+      }
+    } else {
+      checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET, OzoneObj.StoreType.OZONE,
+          IAccessAuthorizer.ACLType.WRITE, volumeName, bucketName, null);
+    }
+  }
+
   @RequestFeatureValidator(
       conditions = ValidationCondition.CLUSTER_NEEDS_FINALIZATION,
       processingPhase = RequestProcessingPhase.PRE_PROCESS,

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

@@ -55,54 +55,53 @@ public OMLifecycleSetServiceStatusRequest(OMRequest omRequest) {
     super(omRequest);
   }
 
+  @Override
+  public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
+    OMRequest request = super.preExecute(ozoneManager);
+
+    if (ozoneManager.getAclsEnabled()) {
+      boolean suspend = request.getSetLifecycleServiceStatusRequest().getSuspend();
+      UserGroupInformation ugi = createUGIForApi();
+      if (!ozoneManager.isAdmin(ugi)) {
+        throw new OMException("Access denied for user " + ugi + ". "
+            + "Superuser privilege is required to " + (suspend ? "suspend" : "resume") + " Lifecycle Service.",
+            OMException.ResultCodes.ACCESS_DENIED);
+      }
+    }
+
+    return request;
+  }
+
   @Override
   public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, ExecutionContext context) {
     OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder(getOmRequest());
     AuditLogger auditLogger = ozoneManager.getAuditLogger();
     UserInfo userInfo = getOmRequest().getUserInfo();
     HashMap<String, String> auditMap = new HashMap<>();
-    IOException exception = null;
-    OMClientResponse omClientResponse;
     boolean suspend = getOmRequest().getSetLifecycleServiceStatusRequest().getSuspend();
     auditMap.put("suspend", String.valueOf(suspend));
 
-    try {
-      if (ozoneManager.getAclsEnabled()) {
-        UserGroupInformation ugi = createUGIForApi();
-        if (!ozoneManager.isAdmin(ugi)) {
-          throw new OMException("Access denied for user " + ugi + ". "
-              + "Superuser privilege is required to " + (suspend ? "suspend" : "resume") + " Lifecycle Service.",
-              OMException.ResultCodes.ACCESS_DENIED);
-        }
-      }
-
-      KeyLifecycleService keyLifecycleService = ozoneManager.getKeyManager().getKeyLifecycleService();
-      if (keyLifecycleService != null) {
-        if (suspend) {
-          keyLifecycleService.suspend();
-          LOG.info("KeyLifecycleService has been suspended by user: {}",
-              userInfo != null ? userInfo.getUserName() : "unknown");
-        } else {
-          keyLifecycleService.resume();
-          LOG.info("KeyLifecycleService resume called by user: {}",
-              userInfo != null ? userInfo.getUserName() : "unknown");
-        }
+    KeyLifecycleService keyLifecycleService = ozoneManager.getKeyManager().getKeyLifecycleService();
+    if (keyLifecycleService != null) {
+      if (suspend) {
+        keyLifecycleService.suspend();
+        LOG.info("KeyLifecycleService has been suspended by user: {}",
+            userInfo != null ? userInfo.getUserName() : "unknown");
       } else {
-        LOG.warn("KeyLifecycleService is not available");
+        keyLifecycleService.resume();
+        LOG.info("KeyLifecycleService resume called by user: {}",
+            userInfo != null ? userInfo.getUserName() : "unknown");
       }
-
-      omResponse.setSetLifecycleServiceStatusResponse(
-          SetLifecycleServiceStatusResponse.newBuilder().build());
-      omClientResponse = new OMLifecycleSetServiceStatusResponse(omResponse.build());
-    } catch (IOException ex) {
-      exception = ex;
-      LOG.error("Failed to " + (suspend ? "suspend" : "resume") + " KeyLifecycleService", ex);
-      omClientResponse = new OMLifecycleSetServiceStatusResponse(
-          createErrorOMResponse(omResponse, ex));
+    } else {
+      LOG.warn("KeyLifecycleService is not available");
     }
 
+    omResponse.setSetLifecycleServiceStatusResponse(
+        SetLifecycleServiceStatusResponse.newBuilder().build());
+    OMClientResponse omClientResponse = new OMLifecycleSetServiceStatusResponse(omResponse.build());
+
     markForAudit(auditLogger, buildAuditMessage(OMAction.SET_LIFECYCLE_SERVICE_STATUS,
-        auditMap, exception, userInfo));
+        auditMap, null, userInfo));
     return omClientResponse;
   }