HBASE-29643: Admin API to trigger for System Key rotation

.rubocop.yml

@@ -9,3 +9,12 @@ Layout/LineLength:
 
 Metrics/MethodLength:
   Max: 75
+
+GlobalVars:
+  AllowedVariables:
+    - $CUST1_ENCODED
+    - $CUST1_ALIAS
+    - $CUST1_ENCODED
+    - $GLOB_CUST_ENCODED
+    - $TEST
+    - $TEST_CLUSTER

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java

@@ -2664,4 +2664,10 @@ List<LogEntry> getLogEntries(Set<ServerName> serverNames, String logType, Server
 
   @InterfaceAudience.Private
   void restoreBackupSystemTable(String snapshotName) throws IOException;
+
+  /**
+   * Refresh the system key cache on all specified region servers.
+   * @param regionServers the list of region servers to refresh the system key cache on
+   */
+  void refreshSystemKeyCacheOnAllServers(Set<ServerName> regionServers) throws IOException;
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java

@@ -1146,4 +1146,9 @@ public List<String> getCachedFilesList(ServerName serverName) throws IOException
   public void restoreBackupSystemTable(String snapshotName) throws IOException {
     get(admin.restoreBackupSystemTable(snapshotName));
   }
+
+  @Override
+  public void refreshSystemKeyCacheOnAllServers(Set<ServerName> regionServers) throws IOException {
+    get(admin.refreshSystemKeyCacheOnAllServers(regionServers));
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -1874,4 +1874,10 @@ CompletableFuture<List<LogEntry>> getLogEntries(Set<ServerName> serverNames, Str
 
   @InterfaceAudience.Private
   CompletableFuture<Void> restoreBackupSystemTable(String snapshotName);
+
+  /**
+   * Refresh the system key cache on all specified region servers.
+   * @param regionServers the list of region servers to refresh the system key cache on
+   */
+  CompletableFuture<Void> refreshSystemKeyCacheOnAllServers(Set<ServerName> regionServers);
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java

@@ -686,6 +686,11 @@ public CompletableFuture<Void> updateConfiguration(String groupName) {
     return wrap(rawAdmin.updateConfiguration(groupName));
   }
 
+  @Override
+  public CompletableFuture<Void> refreshSystemKeyCacheOnAllServers(Set<ServerName> regionServers) {
+    return wrap(rawAdmin.refreshSystemKeyCacheOnAllServers(regionServers));
+  }
+
   @Override
   public CompletableFuture<Void> rollWALWriter(ServerName serverName) {
     return wrap(rawAdmin.rollWALWriter(serverName));

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -150,6 +150,7 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.EmptyMsg;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.LastHighestWalFilenum;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.NameStringPair;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ProcedureDescription;
@@ -4662,4 +4663,28 @@ MasterProtos.RestoreBackupSystemTableResponse> procedureCall(request,
         MasterProtos.RestoreBackupSystemTableResponse::getProcId,
         new RestoreBackupSystemTableProcedureBiConsumer());
   }
+
+  @Override
+  public CompletableFuture<Void> refreshSystemKeyCacheOnAllServers(Set<ServerName> regionServers) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(this::refreshSystemKeyCache).collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> refreshSystemKeyCache(ServerName serverName) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<EmptyMsg, EmptyMsg, Void> adminCall(controller, stub,
+        EmptyMsg.getDefaultInstance(),
+        (s, c, req, done) -> s.refreshSystemKeyCache(controller, req, done), resp -> null))
+      .serverName(serverName).call();
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/keymeta/KeymetaAdminClient.java

@@ -25,19 +25,18 @@
 import org.apache.hadoop.hbase.io.crypto.ManagedKeyData;
 import org.apache.hadoop.hbase.io.crypto.ManagedKeyState;
 import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos;
-import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos.ManagedKeysRequest;
-import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos.ManagedKeysResponse;
+import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos.ManagedKeyRequest;
+import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos.ManagedKeyResponse;
 import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
 
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.EmptyMsg;
 
 @InterfaceAudience.Public
 public class KeymetaAdminClient implements KeymetaAdmin {
-  private static final Logger LOG = LoggerFactory.getLogger(KeymetaAdminClient.class);
   private ManagedKeysProtos.ManagedKeysService.BlockingInterface stub;
 
   public KeymetaAdminClient(Connection conn) throws IOException {
@@ -46,38 +45,54 @@ public KeymetaAdminClient(Connection conn) throws IOException {
   }
 
   @Override
-  public List<ManagedKeyData> enableKeyManagement(String keyCust, String keyNamespace)
+  public ManagedKeyData enableKeyManagement(byte[] keyCust, String keyNamespace)
     throws IOException {
     try {
-      ManagedKeysProtos.GetManagedKeysResponse response = stub.enableKeyManagement(null,
-        ManagedKeysRequest.newBuilder().setKeyCust(keyCust).setKeyNamespace(keyNamespace).build());
-      return generateKeyDataList(response);
+      ManagedKeysProtos.ManagedKeyResponse response =
+        stub.enableKeyManagement(null, ManagedKeyRequest.newBuilder()
+          .setKeyCust(ByteString.copyFrom(keyCust)).setKeyNamespace(keyNamespace).build());
+      return generateKeyData(response);
     } catch (ServiceException e) {
       throw ProtobufUtil.handleRemoteException(e);
     }
   }
 
   @Override
-  public List<ManagedKeyData> getManagedKeys(String keyCust, String keyNamespace)
+  public List<ManagedKeyData> getManagedKeys(byte[] keyCust, String keyNamespace)
     throws IOException, KeyException {
     try {
-      ManagedKeysProtos.GetManagedKeysResponse statusResponse = stub.getManagedKeys(null,
-        ManagedKeysRequest.newBuilder().setKeyCust(keyCust).setKeyNamespace(keyNamespace).build());
+      ManagedKeysProtos.GetManagedKeysResponse statusResponse =
+        stub.getManagedKeys(null, ManagedKeyRequest.newBuilder()
+          .setKeyCust(ByteString.copyFrom(keyCust)).setKeyNamespace(keyNamespace).build());
       return generateKeyDataList(statusResponse);
     } catch (ServiceException e) {
       throw ProtobufUtil.handleRemoteException(e);
     }
   }
 
+  @Override
+  public boolean rotateSTK() throws IOException {
+    try {
+      ManagedKeysProtos.RotateSTKResponse response =
+        stub.rotateSTK(null, EmptyMsg.getDefaultInstance());
+      return response.getRotated();
+    } catch (ServiceException e) {
+      throw ProtobufUtil.handleRemoteException(e);
+    }
+  }
+
   private static List<ManagedKeyData>
     generateKeyDataList(ManagedKeysProtos.GetManagedKeysResponse stateResponse) {
     List<ManagedKeyData> keyStates = new ArrayList<>();
-    for (ManagedKeysResponse state : stateResponse.getStateList()) {
-      keyStates
-        .add(new ManagedKeyData(state.getKeyCustBytes().toByteArray(), state.getKeyNamespace(),
-          null, ManagedKeyState.forValue((byte) state.getKeyState().getNumber()),
-          state.getKeyMetadata(), state.getRefreshTimestamp()));
+    for (ManagedKeyResponse state : stateResponse.getStateList()) {
+      keyStates.add(generateKeyData(state));
     }
     return keyStates;
   }
+
+  private static ManagedKeyData generateKeyData(ManagedKeysProtos.ManagedKeyResponse response) {
+    return new ManagedKeyData(response.getKeyCust().toByteArray(), response.getKeyNamespace(), null,
+      ManagedKeyState.forValue((byte) response.getKeyState().getNumber()),
+      response.getKeyMetadata(), response.getRefreshTimestamp());
+  }
 }

hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/Encryption.java

@@ -506,12 +506,11 @@ public static void decryptWithSubjectKey(OutputStream out, InputStream in, int o
       // is configured
       String alternateAlgorithm = conf.get(HConstants.CRYPTO_ALTERNATE_KEY_ALGORITHM_CONF_KEY);
       if (alternateAlgorithm != null) {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("Unable to decrypt data with current cipher algorithm '"
-            + conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES)
-            + "'. Trying with the alternate cipher algorithm '" + alternateAlgorithm
-            + "' configured.");
-        }
+        LOG.debug(
+          "Unable to decrypt data with current cipher algorithm '{}'. "
+            + "Trying with the alternate cipher algorithm '{}' configured.",
+          conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES),
+          alternateAlgorithm);
         Cipher alterCipher = Encryption.getCipher(conf, alternateAlgorithm);
         if (alterCipher == null) {
           throw new RuntimeException("Cipher '" + alternateAlgorithm + "' not available");
@@ -575,7 +574,7 @@ private static Object createProvider(final Configuration conf, String classNameK
         throw new RuntimeException(e);
       }
       keyProviderCache.put(providerCacheKey, provider);
-      LOG.debug("Installed " + providerClassName + " into key provider cache");
+      LOG.debug("Installed {} into key provider cache", providerClassName);
     }
     return provider;
   }

hbase-common/src/main/java/org/apache/hadoop/hbase/keymeta/KeymetaAdmin.java

@@ -31,23 +31,31 @@
 public interface KeymetaAdmin {
   /**
    * Enables key management for the specified custodian and namespace.
-   * @param keyCust      The key custodian in base64 encoded format.
+   * @param keyCust      The key custodian identifier.
    * @param keyNamespace The namespace for the key management.
    * @return The list of {@link ManagedKeyData} objects each identifying the key and its current
    *         status.
    * @throws IOException if an error occurs while enabling key management.
    */
-  List<ManagedKeyData> enableKeyManagement(String keyCust, String keyNamespace)
+  ManagedKeyData enableKeyManagement(byte[] keyCust, String keyNamespace)
     throws IOException, KeyException;
 
   /**
    * Get the status of all the keys for the specified custodian.
-   * @param keyCust      The key custodian in base64 encoded format.
+   * @param keyCust      The key custodian identifier.
    * @param keyNamespace The namespace for the key management.
    * @return The list of {@link ManagedKeyData} objects each identifying the key and its current
    *         status.
    * @throws IOException if an error occurs while enabling key management.
    */
-  List<ManagedKeyData> getManagedKeys(String keyCust, String keyNamespace)
+  List<ManagedKeyData> getManagedKeys(byte[] keyCust, String keyNamespace)
     throws IOException, KeyException;
+
+  /**
+   * Triggers rotation of the System Key (STK) by checking for a new key and propagating it to all
+   * region servers.
+   * @return true if a new STK was found and rotated, false if no change was detected
+   * @throws IOException if an error occurs while rotating the STK
+   */
+  boolean rotateSTK() throws IOException;
 }

hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/MockManagedKeyProvider.java

@@ -44,18 +44,27 @@ public class MockManagedKeyProvider extends MockAesKeyProvider implements Manage
   private Map<String, ManagedKeyState> keyState = new HashMap<>();
   private String systemKeyAlias = "default_system_key_alias";
 
+  private boolean shouldThrowExceptionOnGetSystemKey = false;
+  private boolean shouldThrowExceptionOnGetManagedKey = false;
+
   @Override
   public void initConfig(Configuration conf, String providerParameters) {
     super.init(providerParameters);
   }
 
   @Override
   public ManagedKeyData getSystemKey(byte[] systemId) throws IOException {
+    if (shouldThrowExceptionOnGetSystemKey) {
+      throw new IOException("Test exception on getSystemKey");
+    }
     return getKey(systemId, systemKeyAlias, ManagedKeyData.KEY_SPACE_GLOBAL);
   }
 
   @Override
   public ManagedKeyData getManagedKey(byte[] key_cust, String key_namespace) throws IOException {
+    if (shouldThrowExceptionOnGetManagedKey) {
+      throw new IOException("Test exception on getManagedKey");
+    }
     String alias = Bytes.toString(key_cust);
     return getKey(key_cust, alias, key_namespace);
   }
@@ -118,6 +127,14 @@ public String getSystemKeyAlias() {
     return this.systemKeyAlias;
   }
 
+  public void setShouldThrowExceptionOnGetSystemKey(boolean shouldThrowExceptionOnGetSystemKey) {
+    this.shouldThrowExceptionOnGetSystemKey = shouldThrowExceptionOnGetSystemKey;
+  }
+
+  public void setShouldThrowExceptionOnGetManagedKey(boolean shouldThrowExceptionOnGetManagedKey) {
+    this.shouldThrowExceptionOnGetManagedKey = shouldThrowExceptionOnGetManagedKey;
+  }
+
   /**
    * Generate a new secret key.
    * @return the key

hbase-protocol-shaded/src/main/protobuf/server/ManagedKeys.proto

@@ -24,8 +24,10 @@ option java_generic_services = true;
 option java_generate_equals_and_hash = true;
 option optimize_for = SPEED;
 
-message ManagedKeysRequest {
-  required string key_cust = 1;
+import "HBase.proto";
+
+message ManagedKeyRequest {
+  required bytes key_cust = 1;
   required string key_namespace = 2;
 }
 
@@ -36,21 +38,27 @@ enum ManagedKeyState {
   KEY_DISABLED = 4;
 }
 
-message ManagedKeysResponse {
-  required string key_cust = 1;
+message ManagedKeyResponse {
+  required bytes key_cust = 1;
   required string key_namespace = 2;
   required ManagedKeyState key_state = 3;
   optional string key_metadata = 4;
   optional int64 refresh_timestamp = 5;
 }
 
 message GetManagedKeysResponse {
-  repeated ManagedKeysResponse state = 1;
+  repeated ManagedKeyResponse state = 1;
+}
+
+message RotateSTKResponse {
+  required bool rotated = 1;
 }
 
 service ManagedKeysService {
-  rpc EnableKeyManagement(ManagedKeysRequest)
-    returns (GetManagedKeysResponse);
-  rpc GetManagedKeys(ManagedKeysRequest)
+  rpc EnableKeyManagement(ManagedKeyRequest)
+    returns (ManagedKeyResponse);
+  rpc GetManagedKeys(ManagedKeyRequest)
     returns (GetManagedKeysResponse);
+  rpc RotateSTK(EmptyMsg)
+    returns (RotateSTKResponse);
 }

hbase-protocol-shaded/src/main/protobuf/server/region/Admin.proto

@@ -420,4 +420,7 @@ service AdminService {
 
   rpc GetCachedFilesList(GetCachedFilesListRequest)
     returns(GetCachedFilesListResponse);
+
+  rpc RefreshSystemKeyCache(EmptyMsg)
+    returns(EmptyMsg);
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java

@@ -194,7 +194,7 @@ public abstract class HBaseServerBase<R extends HBaseRpcServicesBase<?>> extends
 
   protected final NettyEventLoopGroupConfig eventLoopGroupConfig;
 
-  private SystemKeyCache systemKeyCache;
+  protected SystemKeyCache systemKeyCache;
   protected KeymetaAdminImpl keymetaAdmin;
   protected ManagedKeyDataCache managedKeyDataCache;
 
@@ -437,6 +437,17 @@ protected void buildSystemKeyCache() throws IOException {
     }
   }
 
+  /**
+   * Rebuilds the system key cache. This method can be called to refresh the system key cache when
+   * the system key has been rotated.
+   * @throws IOException if there is an error rebuilding the cache
+   */
+  public void rebuildSystemKeyCache() throws IOException {
+    if (SecurityUtil.isKeyManagementEnabled(conf)) {
+      systemKeyCache = SystemKeyCache.createCache(new SystemKeyAccessor(this));
+    }
+  }
+
   protected final void shutdownChore(ScheduledChore chore) {
     if (chore != null) {
       chore.shutdown();

hbase-server/src/main/java/org/apache/hadoop/hbase/client/AsyncRegionServerAdmin.java

@@ -66,6 +66,7 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WarmupRegionRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WarmupRegionResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.EmptyMsg;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaSnapshotsRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaSnapshotsResponse;
 
@@ -217,4 +218,8 @@ public CompletableFuture<GetRegionLoadResponse> getRegionLoad(GetRegionLoadReque
     executeProcedures(ExecuteProceduresRequest request) {
     return call((stub, controller, done) -> stub.executeProcedures(controller, request, done));
   }
+
+  public CompletableFuture<EmptyMsg> refreshSystemKeyCache(EmptyMsg request) {
+    return call((stub, controller, done) -> stub.refreshSystemKeyCache(controller, request, done));
+  }
 }