HBASE-29368 [Precursor] Add API surface and refactoring for key management feature

This commit prepares the codebase for the upcoming key management feature
(HBASE-29368) by introducing the necessary API definitions, protocol buffer
changes, and infrastructure refactoring. No functional changes are included;
all implementation will follow in the feature PR.

This precursor PR essentially extracts the API surface definitions and
infrastructure refactoring from the main feature PR (apache#7421) to facilitate
easier review.  By separating the ~15k line feature PR into a smaller precursor
containing interface definitions, protocol changes, and method signature
updates, the subsequent feature PR will focus purely on implementation logic.

API Surface Additions:
* New interfaces:
  - KeymetaAdmin: Admin API for key management operations
  - Server methods for cache management (getManagedKeyDataCache, getSystemKeyCache)

* Protocol buffer definitions:
  - ManagedKeys.proto: Definitions for managed key data and operations
  - Admin.proto: RPC methods for key management admin operations
  - Procedure.proto: Key rotation procedure support

Infrastructure Refactoring:
* Encryption context creation:
  - Moved createEncryptionContext from EncryptionUtil (client) to SecurityUtil (server)
    where it properly belongs, as it requires server-side resources
  - Added overloads to support future key encryption key (KEK) parameters

* Method signature updates:
  - Added ManagedKeyDataCache and SystemKeyCache parameters to encryption-related
    methods throughout HRegion, HStore, HStoreFile, and HFile classes
  - Updated constructors and factory methods to thread cache references
  - All cache parameters are currently null/unused, enabling gradual feature rollout

* New utility methods:
  - Encryption.encryptWithGivenKey() / decryptWithGivenKey(): Extract method
    refactoring to support both subject-based and KEK-based encryption
  - EncryptionUtil.wrapKey() / unwrapKey() overloads with KEK parameter
  - Bytes.add() 4-argument overload for concatenation

Stub Infrastructure:
* Blank place holder shells for some public data classes such as
  ManagedKeyData and KeymetaAdminClient
* Stub implementations for key management services and caches that return null
  or throw UnsupportedOperationException, clearly documented as placeholders
* New package org.apache.hadoop.hbase.keymeta for key management classes
* Mock services updated to support new cache getter methods for testing

Code Organization:
* Procedure framework: Added support for region-level server name tracking
  to support future key rotation procedures
* Testing infrastructure updated to support new constructor signatures

All stub implementations clearly document they are placeholders for the
upcoming feature PR. Existing encryption functionality remains unchanged
and continues to work as before.

Testing:
* All existing tests pass (precursor introduces no functional changes)
* Build completes successfully with new API surface
* Backward compatibility maintained for non-key-management code paths

Author: haridsv

.gitignore

@@ -25,5 +25,11 @@ linklint/
 **/*.log
 tmp
 **/.flattened-pom.xml
+.sw*
+.*.sw*
+ID
+filenametags
+tags
+.codegenie/
 .vscode/
 **/__pycache__

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

@@ -114,6 +114,9 @@ public interface ColumnFamilyDescriptor {
   /** Returns Return the raw crypto key attribute for the family, or null if not set */
   byte[] getEncryptionKey();
 
+  /** Returns the encryption key namespace for this family */
+  String getEncryptionKeyNamespace();
+
   /** Returns Return the encryption algorithm in use by this family */
   String getEncryptionType();
 

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

@@ -167,6 +167,10 @@ public class ColumnFamilyDescriptorBuilder {
   @InterfaceAudience.Private
   public static final String ENCRYPTION_KEY = "ENCRYPTION_KEY";
   private static final Bytes ENCRYPTION_KEY_BYTES = new Bytes(Bytes.toBytes(ENCRYPTION_KEY));
+  @InterfaceAudience.Private
+  public static final String ENCRYPTION_KEY_NAMESPACE = "ENCRYPTION_KEY_NAMESPACE";
+  private static final Bytes ENCRYPTION_KEY_NAMESPACE_BYTES =
+    new Bytes(Bytes.toBytes(ENCRYPTION_KEY_NAMESPACE));
 
   private static final boolean DEFAULT_MOB = false;
   @InterfaceAudience.Private
@@ -320,6 +324,7 @@ public static Map<String, String> getDefaultValues() {
     DEFAULT_VALUES.keySet().forEach(s -> RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(s))));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(ENCRYPTION)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(ENCRYPTION_KEY)));
+    RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(ENCRYPTION_KEY_NAMESPACE)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(IS_MOB)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(MOB_THRESHOLD)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(MOB_COMPACT_PARTITION_POLICY)));
@@ -522,6 +527,11 @@ public ColumnFamilyDescriptorBuilder setEncryptionKey(final byte[] value) {
     return this;
   }
 
+  public ColumnFamilyDescriptorBuilder setEncryptionKeyNamespace(final String value) {
+    desc.setEncryptionKeyNamespace(value);
+    return this;
+  }
+
   public ColumnFamilyDescriptorBuilder setEncryptionType(String value) {
     desc.setEncryptionType(value);
     return this;
@@ -1337,6 +1347,20 @@ public ModifyableColumnFamilyDescriptor setEncryptionKey(byte[] keyBytes) {
       return setValue(ENCRYPTION_KEY_BYTES, new Bytes(keyBytes));
     }
 
+    @Override
+    public String getEncryptionKeyNamespace() {
+      return getStringOrDefault(ENCRYPTION_KEY_NAMESPACE_BYTES, Function.identity(), null);
+    }
+
+    /**
+     * Set the encryption key namespace attribute for the family
+     * @param keyNamespace the key namespace, or null to remove existing setting
+     * @return this (for chained invocation)
+     */
+    public ModifyableColumnFamilyDescriptor setEncryptionKeyNamespace(String keyNamespace) {
+      return setValue(ENCRYPTION_KEY_NAMESPACE_BYTES, keyNamespace);
+    }
+
     @Override
     public long getMobThreshold() {
       return getStringOrDefault(MOB_THRESHOLD_BYTES, Long::valueOf, DEFAULT_MOB_THRESHOLD);

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

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.keymeta;
+
+import java.io.IOException;
+import java.security.KeyException;
+import java.util.List;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.io.crypto.ManagedKeyData;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * STUB IMPLEMENTATION - Feature not yet complete. This class will be fully implemented in
+ * HBASE-29368 feature PR.
+ */
+@InterfaceAudience.Public
+public class KeymetaAdminClient implements KeymetaAdmin {
+
+  public KeymetaAdminClient(Connection conn) throws IOException {
+    // Stub constructor
+  }
+
+  @Override
+  public ManagedKeyData enableKeyManagement(byte[] keyCust, String keyNamespace)
+    throws IOException, KeyException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public List<ManagedKeyData> getManagedKeys(byte[] keyCust, String keyNamespace)
+    throws IOException, KeyException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public boolean rotateSTK() throws IOException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public void ejectManagedKeyDataCacheEntry(byte[] keyCustodian, String keyNamespace,
+    String keyMetadata) throws IOException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public void clearManagedKeyDataCache() throws IOException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public ManagedKeyData disableKeyManagement(byte[] keyCust, String keyNamespace)
+    throws IOException, KeyException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public ManagedKeyData disableManagedKey(byte[] keyCust, String keyNamespace,
+    byte[] keyMetadataHash) throws IOException, KeyException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public ManagedKeyData rotateManagedKey(byte[] keyCust, String keyNamespace)
+    throws IOException, KeyException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+
+  @Override
+  public void refreshManagedKeys(byte[] keyCust, String keyNamespace)
+    throws IOException, KeyException {
+    throw new UnsupportedOperationException("KeymetaAdmin feature not yet implemented");
+  }
+}

hbase-client/src/main/java/org/apache/hadoop/hbase/security/EncryptionUtil.java

@@ -27,7 +27,6 @@
 import org.apache.commons.crypto.cipher.CryptoCipherFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.io.crypto.Cipher;
 import org.apache.hadoop.hbase.io.crypto.Encryption;
 import org.apache.hadoop.hbase.io.crypto.aes.CryptoAES;
@@ -80,6 +79,21 @@ public static byte[] wrapKey(Configuration conf, byte[] key, String algorithm)
    * @return the encrypted key bytes
    */
   public static byte[] wrapKey(Configuration conf, String subject, Key key) throws IOException {
+    return wrapKey(conf, subject, key, null);
+  }
+
+  /**
+   * Protect a key by encrypting it with the secret key of the given subject or kek. The
+   * configuration must be set up correctly for key alias resolution. Only one of the
+   * {@code subject} or {@code kek} needs to be specified and the other one can be {@code null}.
+   * @param conf    configuration
+   * @param subject subject key alias
+   * @param key     the key
+   * @param kek     the key encryption key
+   * @return the encrypted key bytes
+   */
+  public static byte[] wrapKey(Configuration conf, String subject, Key key, Key kek)
+    throws IOException {
     // Wrap the key with the configured encryption algorithm.
     String algorithm = conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
     Cipher cipher = Encryption.getCipher(conf, algorithm);
@@ -100,8 +114,12 @@ public static byte[] wrapKey(Configuration conf, String subject, Key key) throws
     builder
       .setHash(UnsafeByteOperations.unsafeWrap(Encryption.computeCryptoKeyHash(conf, keyBytes)));
     ByteArrayOutputStream out = new ByteArrayOutputStream();
-    Encryption.encryptWithSubjectKey(out, new ByteArrayInputStream(keyBytes), subject, conf, cipher,
-      iv);
+    if (kek != null) {
+      Encryption.encryptWithGivenKey(kek, out, new ByteArrayInputStream(keyBytes), cipher, iv);
+    } else {
+      Encryption.encryptWithSubjectKey(out, new ByteArrayInputStream(keyBytes), subject, conf,
+        cipher, iv);
+    }
     builder.setData(UnsafeByteOperations.unsafeWrap(out.toByteArray()));
     // Build and return the protobuf message
     out.reset();
@@ -118,6 +136,21 @@ public static byte[] wrapKey(Configuration conf, String subject, Key key) throws
    * @return the raw key bytes
    */
   public static Key unwrapKey(Configuration conf, String subject, byte[] value)
+    throws IOException, KeyException {
+    return unwrapKey(conf, subject, value, null);
+  }
+
+  /**
+   * Unwrap a key by decrypting it with the secret key of the given subject. The configuration must
+   * be set up correctly for key alias resolution. Only one of the {@code subject} or {@code kek}
+   * needs to be specified and the other one can be {@code null}.
+   * @param conf    configuration
+   * @param subject subject key alias
+   * @param value   the encrypted key bytes
+   * @param kek     the key encryption key
+   * @return the raw key bytes
+   */
+  public static Key unwrapKey(Configuration conf, String subject, byte[] value, Key kek)
     throws IOException, KeyException {
     EncryptionProtos.WrappedKey wrappedKey =
       EncryptionProtos.WrappedKey.parser().parseDelimitedFrom(new ByteArrayInputStream(value));
@@ -126,11 +159,12 @@ public static Key unwrapKey(Configuration conf, String subject, byte[] value)
     if (cipher == null) {
       throw new RuntimeException("Cipher '" + algorithm + "' not available");
     }
-    return getUnwrapKey(conf, subject, wrappedKey, cipher);
+    return getUnwrapKey(conf, subject, wrappedKey, cipher, kek);
   }
 
   private static Key getUnwrapKey(Configuration conf, String subject,
-    EncryptionProtos.WrappedKey wrappedKey, Cipher cipher) throws IOException, KeyException {
+    EncryptionProtos.WrappedKey wrappedKey, Cipher cipher, Key kek)
+    throws IOException, KeyException {
     String configuredHashAlgorithm = Encryption.getConfiguredHashAlgorithm(conf);
     String wrappedHashAlgorithm = wrappedKey.getHashAlgorithm().trim();
     if (!configuredHashAlgorithm.equalsIgnoreCase(wrappedHashAlgorithm)) {
@@ -143,8 +177,13 @@ private static Key getUnwrapKey(Configuration conf, String subject,
     }
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     byte[] iv = wrappedKey.hasIv() ? wrappedKey.getIv().toByteArray() : null;
-    Encryption.decryptWithSubjectKey(out, wrappedKey.getData().newInput(), wrappedKey.getLength(),
-      subject, conf, cipher, iv);
+    if (kek != null) {
+      Encryption.decryptWithGivenKey(kek, out, wrappedKey.getData().newInput(),
+        wrappedKey.getLength(), cipher, iv);
+    } else {
+      Encryption.decryptWithSubjectKey(out, wrappedKey.getData().newInput(), wrappedKey.getLength(),
+        subject, conf, cipher, iv);
+    }
     byte[] keyBytes = out.toByteArray();
     if (wrappedKey.hasHash()) {
       if (
@@ -176,58 +215,7 @@ public static Key unwrapWALKey(Configuration conf, String subject, byte[] value)
     if (cipher == null) {
       throw new RuntimeException("Cipher '" + algorithm + "' not available");
     }
-    return getUnwrapKey(conf, subject, wrappedKey, cipher);
-  }
-
-  /**
-   * Helper to create an encyption context.
-   * @param conf   The current configuration.
-   * @param family The current column descriptor.
-   * @return The created encryption context.
-   * @throws IOException           if an encryption key for the column cannot be unwrapped
-   * @throws IllegalStateException in case of encryption related configuration errors
-   */
-  public static Encryption.Context createEncryptionContext(Configuration conf,
-    ColumnFamilyDescriptor family) throws IOException {
-    Encryption.Context cryptoContext = Encryption.Context.NONE;
-    String cipherName = family.getEncryptionType();
-    if (cipherName != null) {
-      if (!Encryption.isEncryptionEnabled(conf)) {
-        throw new IllegalStateException("Encryption for family '" + family.getNameAsString()
-          + "' configured with type '" + cipherName + "' but the encryption feature is disabled");
-      }
-      Cipher cipher;
-      Key key;
-      byte[] keyBytes = family.getEncryptionKey();
-      if (keyBytes != null) {
-        // Family provides specific key material
-        key = unwrapKey(conf, keyBytes);
-        // Use the algorithm the key wants
-        cipher = Encryption.getCipher(conf, key.getAlgorithm());
-        if (cipher == null) {
-          throw new IllegalStateException("Cipher '" + key.getAlgorithm() + "' is not available");
-        }
-        // Fail if misconfigured
-        // We use the encryption type specified in the column schema as a sanity check on
-        // what the wrapped key is telling us
-        if (!cipher.getName().equalsIgnoreCase(cipherName)) {
-          throw new IllegalStateException(
-            "Encryption for family '" + family.getNameAsString() + "' configured with type '"
-              + cipherName + "' but key specifies algorithm '" + cipher.getName() + "'");
-        }
-      } else {
-        // Family does not provide key material, create a random key
-        cipher = Encryption.getCipher(conf, cipherName);
-        if (cipher == null) {
-          throw new IllegalStateException("Cipher '" + cipherName + "' is not available");
-        }
-        key = cipher.getRandomKey();
-      }
-      cryptoContext = Encryption.newContext(conf);
-      cryptoContext.setCipher(cipher);
-      cryptoContext.setKey(key);
-    }
-    return cryptoContext;
+    return getUnwrapKey(conf, subject, wrappedKey, cipher, null);
   }
 
   /**

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

@@ -34,6 +34,8 @@ public class Context implements Configurable {
   private Configuration conf;
   private Cipher cipher;
   private Key key;
+  private ManagedKeyData kekData;
+  private String keyNamespace;
   private String keyHash;
 
   Context(Configuration conf) {
@@ -97,4 +99,22 @@ public Context setKey(Key key) {
     this.keyHash = new String(Hex.encodeHex(Encryption.computeCryptoKeyHash(conf, encoded)));
     return this;
   }
+
+  public Context setKeyNamespace(String keyNamespace) {
+    this.keyNamespace = keyNamespace;
+    return this;
+  }
+
+  public String getKeyNamespace() {
+    return keyNamespace;
+  }
+
+  public Context setKEKData(ManagedKeyData kekData) {
+    this.kekData = kekData;
+    return this;
+  }
+
+  public ManagedKeyData getKEKData() {
+    return kekData;
+  }
 }