HDDS-14207. Inconsistent Ozone admin check

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/SecurityConfig.java

@@ -71,6 +71,8 @@
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_X509_SIGNATURE_ALGO;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_X509_SIGNATURE_ALGO_DEFAULT;
 import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_METADATA_DIRS;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_AUTHORIZATION_ENABLED;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_AUTHORIZATION_ENABLED_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SECURITY_ENABLED_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SECURITY_ENABLED_KEY;
 
@@ -104,6 +106,15 @@ public class SecurityConfig {
   private static final Logger LOG =
       LoggerFactory.getLogger(SecurityConfig.class);
   private static volatile Provider provider;
+
+  /**
+   * Test-only configuration property to enable authorization checks without
+   * requiring full security (Kerberos) setup. This is for testing purposes
+   * only.
+   */
+  public static final String OZONE_TEST_AUTHORIZATION_ENABLED = "ozone.test.authorization.enabled";
+  public static final boolean OZONE_TEST_AUTHORIZATION_ENABLED_DEFAULT = false;
+
   private final int size;
   private final String keyAlgo;
   private final String providerString;
@@ -136,6 +147,7 @@ public class SecurityConfig {
   private final Duration rootCaCertificatePollingInterval;
   private final boolean autoCARotationEnabled;
   private final Duration expiredCertificateCheckInterval;
+  private final boolean authorizationEnabled;
 
   /**
    * Constructs a SecurityConfig.
@@ -200,6 +212,14 @@ public SecurityConfig(ConfigurationSource configuration) {
         OZONE_SECURITY_ENABLED_KEY,
         OZONE_SECURITY_ENABLED_DEFAULT);
 
+    // Authorization is only effective when security is enabled, unless test mode is enabled
+    boolean testAuthorizationEnabled = configuration.getBoolean(
+        OZONE_TEST_AUTHORIZATION_ENABLED,
+        OZONE_TEST_AUTHORIZATION_ENABLED_DEFAULT);
+    this.authorizationEnabled = (isSecurityEnabled || testAuthorizationEnabled) &&
+        configuration.getBoolean(OZONE_AUTHORIZATION_ENABLED,
+            OZONE_AUTHORIZATION_ENABLED_DEFAULT);
+
     String certDurationString =
         configuration.get(HDDS_X509_DEFAULT_DURATION,
             HDDS_X509_DEFAULT_DURATION_DEFAULT);
@@ -608,4 +628,15 @@ public boolean useTestCert() {
   public boolean isTokenEnabled() {
     return blockTokenEnabled || containerTokenEnabled;
   }
+
+  /**
+   * Check if authorization checks should be performed in Ozone.
+   * Authorization is only effective when security is enabled, unless test mode is enabled.
+   * This controls both admin privilege checks and ACL checks.
+   *
+   * @return true if authorization checks should be performed
+   */
+  public boolean isAuthorizationEnabled() {
+    return authorizationEnabled;
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -469,6 +469,8 @@ public final class OzoneConfigKeys {
       "ozone.acl.enabled";
   public static final boolean OZONE_ACL_ENABLED_DEFAULT =
       false;
+  public static final String OZONE_AUTHORIZATION_ENABLED = "ozone.authorization.enabled";
+  public static final boolean OZONE_AUTHORIZATION_ENABLED_DEFAULT = true;
   public static final String OZONE_S3_VOLUME_NAME =
           "ozone.s3g.volume.name";
   public static final String OZONE_S3_VOLUME_NAME_DEFAULT =

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneSecurityUtil.java

@@ -17,6 +17,10 @@
 
 package org.apache.hadoop.ozone;
 
+import static org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED;
+import static org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_AUTHORIZATION_ENABLED;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_AUTHORIZATION_ENABLED_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_HTTP_SECURITY_ENABLED_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_HTTP_SECURITY_ENABLED_KEY;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SECURITY_ENABLED_DEFAULT;
@@ -60,6 +64,23 @@ public static boolean isHttpSecurityEnabled(ConfigurationSource conf) {
         OZONE_HTTP_SECURITY_ENABLED_DEFAULT);
   }
 
+  /**
+   * Check if authorization checks should be performed in Ozone.
+   * Authorization is only effective when security is enabled, unless test mode is enabled.
+   * This controls both admin privilege checks and ACL checks.
+   *
+   * @param conf Configuration source
+   * @return true if authorization checks should be performed
+   */
+  public static boolean isAuthorizationEnabled(ConfigurationSource conf) {
+    // Check if test mode is enabled (allows authorization without full security)
+    boolean testAuthorizationEnabled = conf.getBoolean(OZONE_TEST_AUTHORIZATION_ENABLED,
+        OZONE_TEST_AUTHORIZATION_ENABLED_DEFAULT);
+    return (isSecurityEnabled(conf) || testAuthorizationEnabled) &&
+        conf.getBoolean(OZONE_AUTHORIZATION_ENABLED,
+            OZONE_AUTHORIZATION_ENABLED_DEFAULT);
+  }
+
   /**
    * Returns Keys status.
    *

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -2376,6 +2376,20 @@
     <tag>OZONE, SECURITY, ACL</tag>
     <description>Key to enable/disable ozone acls.</description>
   </property>
+  <property>
+    <name>ozone.authorization.enabled</name>
+    <value>true</value>
+    <tag>OZONE, SECURITY, AUTHORIZATION</tag>
+    <description>
+      Master switch to enable/disable authorization checks in Ozone
+      (admin privilege checks and ACL checks).
+      This property only takes effect when ozone.security.enabled is true.
+      When true: admin privilege checks are always performed, and object
+      ACL checks are controlled by ozone.acl.enabled.
+      When false: no authorization checks are performed.
+      Default is true.
+    </description>
+  </property>
   <property>
     <name>ozone.om.kerberos.keytab.file</name>
     <value>/etc/security/keytabs/OM.keytab</value>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java

@@ -669,6 +669,11 @@ public boolean isStopped() {
    */
   private void checkAdminPrivilege(String operation)
       throws IOException {
+    // Skip check if authorization is disabled
+    if (secConf == null || !secConf.isAuthorizationEnabled()) {
+      return;
+    }
+
     final UserGroupInformation ugi = getRemoteUser();
     admins.checkAdminUserPrivilege(ugi);
   }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/SCMTestUtils.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.common;
 
+import static org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 
@@ -137,6 +138,7 @@ public static OzoneConfiguration getConf(File testDir) {
     conf.setClass(SpaceUsageCheckFactory.Conf.configKeyForClassName(),
         MockSpaceUsageCheckFactory.None.class,
         SpaceUsageCheckFactory.class);
+    conf.setBoolean(OZONE_TEST_AUTHORIZATION_ENABLED, true);
     return conf;
   }
 

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/OzoneAdmins.java

@@ -103,6 +103,8 @@ public static OzoneAdmins getReadonlyAdmins(
 
   /**
    * Check ozone admin privilege, throws exception if not admin.
+   * Note: This method does NOT check if authorization is enabled.
+   * Callers should check authorization before calling this method.
    */
   public void checkAdminUserPrivilege(UserGroupInformation ugi)
       throws AccessControlException {

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/DBCheckpointServlet.java

@@ -73,14 +73,14 @@ public class DBCheckpointServlet extends HttpServlet
   private transient DBStore dbStore;
   private transient DBCheckpointMetrics dbMetrics;
 
-  private boolean aclEnabled;
+  private boolean authorizationEnabled;
   private boolean isSpnegoEnabled;
   private transient OzoneAdmins admins;
   private transient BootstrapStateHandler.Lock lock;
   private transient File bootstrapTempData;
 
   public void initialize(DBStore store, DBCheckpointMetrics metrics,
-                         boolean omAclEnabled,
+                         boolean isAuthorizationEnabled,
                          Collection<String> allowedAdminUsers,
                          Collection<String> allowedAdminGroups,
                          boolean isSpnegoAuthEnabled)
@@ -94,7 +94,7 @@ public void initialize(DBStore store, DBCheckpointMetrics metrics,
       throw new ServletException("DB Store is null");
     }
 
-    this.aclEnabled = omAclEnabled;
+    this.authorizationEnabled = isAuthorizationEnabled;
     this.admins = new OzoneAdmins(allowedAdminUsers, allowedAdminGroups);
     this.isSpnegoEnabled = isSpnegoAuthEnabled;
     lock = new NoOpLock();
@@ -129,9 +129,9 @@ public File getBootstrapTempData() {
   }
 
   private boolean hasPermission(UserGroupInformation user) {
-    // Check ACL for dbCheckpoint only when global Ozone ACL and SPNEGO is
+    // Check admin access for dbCheckpoint only when authorization and SPNEGO is
     // enabled
-    if (aclEnabled && isSpnegoEnabled) {
+    if (authorizationEnabled && isSpnegoEnabled) {
       return admins.isAdmin(user);
     } else {
       return true;
@@ -165,8 +165,8 @@ private void generateSnapshotCheckpoint(HttpServletRequest request,
       return;
     }
 
-    // Check ACL for dbCheckpoint only when global Ozone ACL is enabled
-    if (aclEnabled) {
+    // Check authorization for dbCheckpoint only when authorization is enabled
+    if (authorizationEnabled) {
       final java.security.Principal userPrincipal = request.getUserPrincipal();
       if (userPrincipal == null) {
         final String remoteUser = request.getRemoteUser();

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDBCheckpointServlet.java

@@ -27,10 +27,10 @@
 /**
  * Provides the current checkpoint Snapshot of the SCM DB. (tar.gz)
  *
- * When Ozone ACL is enabled (`ozone.acl.enabled`=`true`), only users/principals
- * configured in `ozone.administrator` (along with the user that starts OM,
- * which automatically becomes an Ozone administrator but not necessarily in
- * the config) are allowed to access this endpoint.
+ * When Ozone authorization is enabled (`ozone.authorization.enabled`=`true`),
+ * only users/principals configured in `ozone.administrator` (along with the
+ * user that starts SCM, which automatically becomes an Ozone administrator
+ * but not necessarily in the config) are allowed to access this endpoint.
  *
  * If Kerberos is enabled, the principal should be appended to
  * `ozone.administrator`, e.g. `scm/scm@EXAMPLE.COM`
@@ -56,7 +56,7 @@ public void init() throws ServletException {
 
     initialize(scm.getScmMetadataStore().getStore(),
         scm.getMetrics().getDBCheckpointMetrics(),
-        false,
+        scm.isAdminAuthorizationEnabled(),
         Collections.emptyList(),
         Collections.emptyList(),
         false);

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java

@@ -1952,8 +1952,23 @@ private void checkAdminAccess(String op) throws IOException {
     checkAdminAccess(getRemoteUser(), false);
   }
 
+  /**
+   * Check if admin privilege authorization should be enforced.
+   * This controls system-level admin operations (upgrades, decommission, etc.)
+   *
+   * @return true if admin authorization checks should be performed
+   */
+  public boolean isAdminAuthorizationEnabled() {
+    return securityConfig != null && securityConfig.isAuthorizationEnabled();
+  }
+
   public void checkAdminAccess(UserGroupInformation remoteUser, boolean isRead)
       throws IOException {
+    // Skip check if authorization is disabled
+    if (!isAdminAuthorizationEnabled()) {
+      return;
+    }
+
     if (remoteUser != null && !scmAdmins.isAdmin(remoteUser)) {
       if (!isRead || !scmReadOnlyAdmins.isAdmin(remoteUser)) {
         throw new AccessControlException(

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestStorageContainerManager.java

@@ -24,6 +24,7 @@
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_SCM_SAFEMODE_PIPELINE_CREATION;
 import static org.apache.hadoop.hdds.scm.HddsTestUtils.mockRemoteUser;
 import static org.apache.hadoop.hdds.scm.HddsWhiteboxTestUtils.setInternalState;
+import static org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL;
 import static org.apache.hadoop.ozone.common.BlockGroup.SIZE_NOT_AVAILABLE;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -167,6 +168,7 @@ public class TestStorageContainerManager {
   @Test
   void test(@TempDir Path tempDir) throws Exception {
     OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(OZONE_TEST_AUTHORIZATION_ENABLED, true);
     configureTopology(conf);
     configureBlockDeletion(conf);
     Path scmPath = tempDir.resolve("scm-meta");

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java

@@ -17,6 +17,8 @@
 
 package org.apache.hadoop.ozone.client.rpc;
 
+import static org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED;
+
 import java.io.IOException;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
@@ -32,6 +34,7 @@ class TestOzoneRpcClient extends OzoneRpcClientTests {
   @BeforeAll
   public static void init() throws Exception {
     OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(OZONE_TEST_AUTHORIZATION_ENABLED, true);
     conf.setInt(ScmConfigKeys.OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT, 1);
     conf.setBoolean(OzoneConfigKeys.OZONE_ACL_ENABLED, true);
     conf.set(OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS,

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestSecureOzoneRpcClient.java

@@ -19,6 +19,7 @@
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_METADATA_DIRS;
+import static org.apache.hadoop.hdds.security.SecurityConfig.OZONE_TEST_AUTHORIZATION_ENABLED;
 import static org.apache.hadoop.ozone.OzoneConsts.FORCE_LEASE_RECOVERY_ENV;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_OFS_URI_SCHEME;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_ROOT;
@@ -116,6 +117,7 @@ public static void init() throws Exception {
     conf.setBoolean(HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED, true);
     conf.set(OZONE_METADATA_DIRS, testDir.getAbsolutePath());
     conf.setBoolean(OzoneConfigKeys.OZONE_ACL_ENABLED, true);
+    conf.setBoolean(OZONE_TEST_AUTHORIZATION_ENABLED, true);
     conf.set(OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS,
         OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS_NATIVE);
     CertificateClientTestImpl certificateClientTest =