From e11f63f463039eb77f9d627799eb61768b79b0a9 Mon Sep 17 00:00:00 2001
From: andyhuangdev <ky052632@gmail.com>
Date: Wed, 28 Oct 2026 13:37:04 +0800
Subject: [PATCH] HDDS-16963. Require SCM admin access for secret key rotation

Test plan: TestSecureOzoneCluster and TestBlockTokensCLI (12 tests passed); checkstyle passed.
---
 .../scm/server/SCMSecurityProtocolServer.java |  3 +-
 .../hadoop/ozone/TestSecureOzoneCluster.java  | 32 +++++++++++++++++++
 2 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMSecurityProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMSecurityProtocolServer.java
index bed276bc9129..ebfd1e59133b 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMSecurityProtocolServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMSecurityProtocolServer.java
@@ -244,7 +244,8 @@ private void validateSecretKeyStatus() throws SCMSecretKeyException {
   }
 
   @Override
-  public boolean checkAndRotate(boolean force) throws SCMSecretKeyException {
+  public boolean checkAndRotate(boolean force) throws IOException {
+    storageContainerManager.checkAdminAccess(getRpcRemoteUser(), false);
     validateSecretKeyStatus();
     try {
       return secretKeyManager.checkAndRotate(force);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestSecureOzoneCluster.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestSecureOzoneCluster.java
index 777101d26acc..e8a4bba4b14c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestSecureOzoneCluster.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestSecureOzoneCluster.java
@@ -94,11 +94,14 @@
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMGetCertResponseProto;
 import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolClientSideTranslatorPB;
+import org.apache.hadoop.hdds.protocolPB.SecretKeyProtocolClientSideTranslatorPB;
+import org.apache.hadoop.hdds.protocolPB.SecretKeyProtocolScmPB;
 import org.apache.hadoop.hdds.scm.HddsTestUtils;
 import org.apache.hadoop.hdds.scm.ScmInfo;
 import org.apache.hadoop.hdds.scm.client.ScmTopologyClient;
 import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
 import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationInternalInterface;
+import org.apache.hadoop.hdds.scm.proxy.SecretKeyProtocolFailoverProxyProvider;
 import org.apache.hadoop.hdds.scm.server.SCMStorageConfig;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.security.SecurityConfig;
@@ -292,6 +295,7 @@ void testSecureScmAndOmStartupAndAccessControl() throws Exception {
     assertScmIssuesCertificateOnlyToServiceUserOfRole();
     assertScmAdminProtocolDeniesNonAdminUser();
     assertScmAdminProtocolRejectsNonKerberosUser();
+    assertScmSecretKeyRotationRequiresAdmin();
     assertScmSecurityProtocolRequiresAdminForScmRoleCertificate();
 
     // Secure OM initialization against the shared SCM.
@@ -439,6 +443,34 @@ private void assertScmAdminProtocolRejectsNonKerberosUser() throws IOException {
         .hasMessageContaining(CANNOT_AUTHENTICATE_MESSAGE);
   }
 
+  /** SCM secret key protocol - rotation requires SCM admin access. */
+  private void assertScmSecretKeyRotationRequiresAdmin() throws Exception {
+    GenericTestUtils.waitFor(() -> scm.getSecretKeyManager().isInitialized(), 100, 10_000);
+    UserGroupInformation testUser = UserGroupInformation.loginUserFromKeytabAndReturnUGI(
+        getTestUserPrincipal(), getTestUserKeytab().getCanonicalPath());
+    UUID currentKeyId = scm.getSecretKeyManager().getCurrentSecretKey().getId();
+    try (SecretKeyProtocolClientSideTranslatorPB secretKeyClient = getSecretKeyClientForScm(testUser)) {
+      for (boolean force : new boolean[]{false, true}) {
+        IOException denied = assertThrows(IOException.class, () -> secretKeyClient.checkAndRotate(force));
+        assertThat(denied).hasMessageContaining("Access denied");
+        assertEquals(currentKeyId, scm.getSecretKeyManager().getCurrentSecretKey().getId());
+      }
+    }
+
+    UserGroupInformation scmUser = UserGroupInformation.loginUserFromKeytabAndReturnUGI(
+        conf.get(HDDS_SCM_KERBEROS_PRINCIPAL_KEY), conf.get(HDDS_SCM_KERBEROS_KEYTAB_FILE_KEY));
+    try (SecretKeyProtocolClientSideTranslatorPB secretKeyClient = getSecretKeyClientForScm(scmUser)) {
+      assertThat(secretKeyClient.checkAndRotate(true)).isTrue();
+      assertNotEquals(currentKeyId, scm.getSecretKeyManager().getCurrentSecretKey().getId());
+    }
+  }
+
+  private SecretKeyProtocolClientSideTranslatorPB getSecretKeyClientForScm(UserGroupInformation ugi) {
+    return new SecretKeyProtocolClientSideTranslatorPB(
+        new SecretKeyProtocolFailoverProxyProvider<>(conf, ugi, SecretKeyProtocolScmPB.class),
+        SecretKeyProtocolScmPB.class);
+  }
+
   /**
    * Secure OM initialization fails when the delegation token and secret key
    * configuration don't meet requirement.