diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcClient.java
index ed0c4fffc724..44689d695d97 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcClient.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcClient.java
@@ -116,7 +116,7 @@ SslContext getSslContext() throws X509Exception, IOException {
         keyStoreWatcher.get() == null && trustStoreWatcher.get() == null
           && conf.getBoolean(X509Util.TLS_CERT_RELOAD, false)
       ) {
-        X509Util.enableCertFileReloading(conf, keyStoreWatcher, trustStoreWatcher,
+        X509Util.enableCertFileReloadingForClient(conf, keyStoreWatcher, trustStoreWatcher,
           () -> sslContextForClient.set(null));
       }
     }
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/tls/X509Util.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/tls/X509Util.java
index d6be0eed844e..6d7387895218 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/tls/X509Util.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/tls/X509Util.java
@@ -28,6 +28,8 @@
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ssl.CertPathTrustManagerParameters;
 import javax.net.ssl.KeyManager;
@@ -89,6 +91,44 @@ public final class X509Util {
   public static final String TLS_CERT_RELOAD = CONFIG_PREFIX + "certReload";
   public static final String TLS_USE_OPENSSL = CONFIG_PREFIX + "useOpenSsl";
 
+  //
+  // Role-scoped keystore/truststore configs for single-EKU certificate support.
+  //
+
+  /**
+   * When set, these take precedence over the unscoped keys above; when unset, the
+   * unscoped keys are used as a fallback so existing deployments keep working
+   * unchanged.
+   */
+  static final String CLIENT_CONFIG_PREFIX = CONFIG_PREFIX + "client.";
+  static final String SERVER_CONFIG_PREFIX = CONFIG_PREFIX + "server.";
+
+  public static final String TLS_CONFIG_CLIENT_KEYSTORE_LOCATION =
+    CLIENT_CONFIG_PREFIX + "keystore.location";
+  public static final String TLS_CONFIG_CLIENT_KEYSTORE_TYPE =
+    CLIENT_CONFIG_PREFIX + "keystore.type";
+  public static final String TLS_CONFIG_CLIENT_KEYSTORE_PASSWORD =
+    CLIENT_CONFIG_PREFIX + "keystore.password";
+  public static final String TLS_CONFIG_CLIENT_TRUSTSTORE_LOCATION =
+    CLIENT_CONFIG_PREFIX + "truststore.location";
+  public static final String TLS_CONFIG_CLIENT_TRUSTSTORE_TYPE =
+    CLIENT_CONFIG_PREFIX + "truststore.type";
+  public static final String TLS_CONFIG_CLIENT_TRUSTSTORE_PASSWORD =
+    CLIENT_CONFIG_PREFIX + "truststore.password";
+
+  public static final String TLS_CONFIG_SERVER_KEYSTORE_LOCATION =
+    SERVER_CONFIG_PREFIX + "keystore.location";
+  public static final String TLS_CONFIG_SERVER_KEYSTORE_TYPE =
+    SERVER_CONFIG_PREFIX + "keystore.type";
+  public static final String TLS_CONFIG_SERVER_KEYSTORE_PASSWORD =
+    SERVER_CONFIG_PREFIX + "keystore.password";
+  public static final String TLS_CONFIG_SERVER_TRUSTSTORE_LOCATION =
+    SERVER_CONFIG_PREFIX + "truststore.location";
+  public static final String TLS_CONFIG_SERVER_TRUSTSTORE_TYPE =
+    SERVER_CONFIG_PREFIX + "truststore.type";
+  public static final String TLS_CONFIG_SERVER_TRUSTSTORE_PASSWORD =
+    SERVER_CONFIG_PREFIX + "truststore.password";
+
   //
   // Server-side specific configs
   //
@@ -165,6 +205,78 @@ public org.apache.hbase.thirdparty.io.netty.handler.ssl.ClientAuth toNettyClient
     }
   }
 
+  /**
+   * Identifies which side of a TLS connection is being configured. Used by role-aware helpers
+   * to pick the correct role-scoped configuration key for keystore/truststore material.
+   */
+  enum Role {
+    CLIENT,
+    SERVER
+  }
+
+  /**
+   * Tracks which configuration keys have already been logged as the effective source of a piece of
+   * TLS material, so {@link #resolveConfig} / {@link #resolvePassword} emit at most one INFO line
+   * per key per JVM. Keys are unique full property names (role-scoped or legacy).
+   */
+  private static final Set<String> LOGGED_RESOLVED_KEYS = ConcurrentHashMap.newKeySet();
+
+  private static void logResolvedKeyOnce(String key) {
+    if (LOGGED_RESOLVED_KEYS.add(key)) {
+      LOG.info("Using configuration key '{}' for TLS material", key);
+    }
+  }
+
+  /**
+   * Returns the value of a role-scoped TLS configuration key, falling back to the unscoped legacy
+   * key if the role-scoped key is unset, and finally to {@code defaultValue} if both are unset.
+   * Logs (once per JVM at INFO) which key supplied the effective value, to aid diagnosing which
+   * keystore/truststore is actually in use on each side of a TLS handshake.
+   * @param config       the configuration to read from
+   * @param roleKey      the role-scoped key name (e.g. {@code hbase.rpc.tls.client.keystore.location})
+   * @param legacyKey    the unscoped fallback key name (e.g. {@code hbase.rpc.tls.keystore.location})
+   * @param defaultValue value to return when neither key is set; may be {@code null}
+   * @return the resolved value, or {@code defaultValue} if neither key is set
+   */
+  public static String resolveConfig(Configuration config, String roleKey, String legacyKey,
+    String defaultValue) {
+    String value = config.get(roleKey);
+    if (value != null) {
+      logResolvedKeyOnce(roleKey);
+      return value;
+    }
+    value = config.get(legacyKey);
+    if (value != null) {
+      logResolvedKeyOnce(legacyKey);
+      return value;
+    }
+    return defaultValue;
+  }
+
+  /**
+   * Password-flavored counterpart to {@link #resolveConfig}. Uses
+   * {@link Configuration#getPassword(String)} so that credential providers configured via
+   * {@code hadoop.security.credential.provider.path} are honored. Returns {@code null} if neither
+   * the role-scoped nor the legacy key resolves to a value.
+   * @param config    the configuration to read from
+   * @param roleKey   the role-scoped password key name
+   * @param legacyKey the unscoped fallback password key name
+   * @return the resolved password as a char array, or {@code null} if neither key is set
+   */
+  public static char[] resolvePassword(Configuration config, String roleKey, String legacyKey)
+    throws IOException {
+    char[] value = config.getPassword(roleKey);
+    if (value != null) {
+      logResolvedKeyOnce(roleKey);
+      return value;
+    }
+    value = config.getPassword(legacyKey);
+    if (value != null) {
+      logResolvedKeyOnce(legacyKey);
+    }
+    return value;
+  }
+
   private X509Util() {
     // disabled
   }
@@ -175,20 +287,27 @@ public static SslContext createSslContextForClient(Configuration config)
     SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();
 
     configureOpenSslIfAvailable(sslContextBuilder, config);
-    String keyStoreLocation = config.get(TLS_CONFIG_KEYSTORE_LOCATION, "");
-    char[] keyStorePassword = config.getPassword(TLS_CONFIG_KEYSTORE_PASSWORD);
-    String keyStoreType = config.get(TLS_CONFIG_KEYSTORE_TYPE, "");
+    String keyStoreLocation =
+      resolveConfig(config, TLS_CONFIG_CLIENT_KEYSTORE_LOCATION, TLS_CONFIG_KEYSTORE_LOCATION, "");
+    char[] keyStorePassword =
+      resolvePassword(config, TLS_CONFIG_CLIENT_KEYSTORE_PASSWORD, TLS_CONFIG_KEYSTORE_PASSWORD);
+    String keyStoreType =
+      resolveConfig(config, TLS_CONFIG_CLIENT_KEYSTORE_TYPE, TLS_CONFIG_KEYSTORE_TYPE, "");
 
     if (keyStoreLocation.isEmpty()) {
-      LOG.warn(TLS_CONFIG_KEYSTORE_LOCATION + " not specified");
+      LOG.warn("Neither {} nor {} specified", TLS_CONFIG_CLIENT_KEYSTORE_LOCATION,
+        TLS_CONFIG_KEYSTORE_LOCATION);
     } else {
       sslContextBuilder
         .keyManager(createKeyManager(keyStoreLocation, keyStorePassword, keyStoreType));
     }
 
-    String trustStoreLocation = config.get(TLS_CONFIG_TRUSTSTORE_LOCATION, "");
-    char[] trustStorePassword = config.getPassword(TLS_CONFIG_TRUSTSTORE_PASSWORD);
-    String trustStoreType = config.get(TLS_CONFIG_TRUSTSTORE_TYPE, "");
+    String trustStoreLocation = resolveConfig(config, TLS_CONFIG_CLIENT_TRUSTSTORE_LOCATION,
+      TLS_CONFIG_TRUSTSTORE_LOCATION, "");
+    char[] trustStorePassword = resolvePassword(config, TLS_CONFIG_CLIENT_TRUSTSTORE_PASSWORD,
+      TLS_CONFIG_TRUSTSTORE_PASSWORD);
+    String trustStoreType =
+      resolveConfig(config, TLS_CONFIG_CLIENT_TRUSTSTORE_TYPE, TLS_CONFIG_TRUSTSTORE_TYPE, "");
 
     boolean sslCrlEnabled = config.getBoolean(TLS_CONFIG_CLR, false);
     boolean sslOcspEnabled = config.getBoolean(TLS_CONFIG_OCSP, false);
@@ -198,7 +317,8 @@ public static SslContext createSslContextForClient(Configuration config)
     boolean allowReverseDnsLookup = config.getBoolean(TLS_CONFIG_REVERSE_DNS_LOOKUP_ENABLED, true);
 
     if (trustStoreLocation.isEmpty()) {
-      LOG.warn(TLS_CONFIG_TRUSTSTORE_LOCATION + " not specified");
+      LOG.warn("Neither {} nor {} specified", TLS_CONFIG_CLIENT_TRUSTSTORE_LOCATION,
+        TLS_CONFIG_TRUSTSTORE_LOCATION);
     } else {
       sslContextBuilder
         .trustManager(createTrustManager(trustStoreLocation, trustStorePassword, trustStoreType,
@@ -249,13 +369,16 @@ private static boolean configureOpenSslIfAvailable(SslContextBuilder sslContextB
 
   public static SslContext createSslContextForServer(Configuration config)
     throws X509Exception, IOException {
-    String keyStoreLocation = config.get(TLS_CONFIG_KEYSTORE_LOCATION, "");
-    char[] keyStorePassword = config.getPassword(TLS_CONFIG_KEYSTORE_PASSWORD);
-    String keyStoreType = config.get(TLS_CONFIG_KEYSTORE_TYPE, "");
+    String keyStoreLocation =
+      resolveConfig(config, TLS_CONFIG_SERVER_KEYSTORE_LOCATION, TLS_CONFIG_KEYSTORE_LOCATION, "");
+    char[] keyStorePassword =
+      resolvePassword(config, TLS_CONFIG_SERVER_KEYSTORE_PASSWORD, TLS_CONFIG_KEYSTORE_PASSWORD);
+    String keyStoreType =
+      resolveConfig(config, TLS_CONFIG_SERVER_KEYSTORE_TYPE, TLS_CONFIG_KEYSTORE_TYPE, "");
 
     if (keyStoreLocation.isEmpty()) {
-      throw new SSLContextException(
-        "Keystore is required for SSL server: " + TLS_CONFIG_KEYSTORE_LOCATION);
+      throw new SSLContextException("Keystore is required for SSL server: set either "
+        + TLS_CONFIG_SERVER_KEYSTORE_LOCATION + " or " + TLS_CONFIG_KEYSTORE_LOCATION);
     }
 
     SslContextBuilder sslContextBuilder;
@@ -263,9 +386,12 @@ public static SslContext createSslContextForServer(Configuration config)
       .forServer(createKeyManager(keyStoreLocation, keyStorePassword, keyStoreType));
 
     configureOpenSslIfAvailable(sslContextBuilder, config);
-    String trustStoreLocation = config.get(TLS_CONFIG_TRUSTSTORE_LOCATION, "");
-    char[] trustStorePassword = config.getPassword(TLS_CONFIG_TRUSTSTORE_PASSWORD);
-    String trustStoreType = config.get(TLS_CONFIG_TRUSTSTORE_TYPE, "");
+    String trustStoreLocation = resolveConfig(config, TLS_CONFIG_SERVER_TRUSTSTORE_LOCATION,
+      TLS_CONFIG_TRUSTSTORE_LOCATION, "");
+    char[] trustStorePassword = resolvePassword(config, TLS_CONFIG_SERVER_TRUSTSTORE_PASSWORD,
+      TLS_CONFIG_TRUSTSTORE_PASSWORD);
+    String trustStoreType =
+      resolveConfig(config, TLS_CONFIG_SERVER_TRUSTSTORE_TYPE, TLS_CONFIG_TRUSTSTORE_TYPE, "");
 
     boolean sslCrlEnabled = config.getBoolean(TLS_CONFIG_CLR, false);
     boolean sslOcspEnabled = config.getBoolean(TLS_CONFIG_OCSP, false);
@@ -277,7 +403,8 @@ public static SslContext createSslContextForServer(Configuration config)
     boolean allowReverseDnsLookup = config.getBoolean(TLS_CONFIG_REVERSE_DNS_LOOKUP_ENABLED, true);
 
     if (trustStoreLocation.isEmpty()) {
-      LOG.warn(TLS_CONFIG_TRUSTSTORE_LOCATION + " not specified");
+      LOG.warn("Neither {} nor {} specified", TLS_CONFIG_SERVER_TRUSTSTORE_LOCATION,
+        TLS_CONFIG_TRUSTSTORE_LOCATION);
     } else {
       sslContextBuilder
         .trustManager(createTrustManager(trustStoreLocation, trustStorePassword, trustStoreType,
@@ -422,20 +549,53 @@ private static String[] getCipherSuites(Configuration config) {
   }
 
   /**
-   * Enable certificate file reloading by creating FileWatchers for keystore and truststore.
-   * AtomicReferences will be set with the new instances. resetContext - if not null - will be
-   * called when the file has been modified.
+   * Enable certificate file reloading for the RPC <em>client</em> side by creating FileWatchers for
+   * the keystore and truststore whose paths are resolved by
+   * {@link Role#CLIENT} (role-scoped keys first, legacy keys as fallback). AtomicReferences will be
+   * set with the new instances. {@code resetContext} - if not null - will be called when the file
+   * has been modified.
    * @param keystoreWatcher   Reference to keystoreFileWatcher.
    * @param trustStoreWatcher Reference to truststoreFileWatcher.
    * @param resetContext      Callback for file changes.
    */
-  public static void enableCertFileReloading(Configuration config,
+  public static void enableCertFileReloadingForClient(Configuration config,
     AtomicReference<FileChangeWatcher> keystoreWatcher,
     AtomicReference<FileChangeWatcher> trustStoreWatcher, Runnable resetContext)
     throws IOException {
-    String keyStoreLocation = config.get(TLS_CONFIG_KEYSTORE_LOCATION, "");
+    enableCertFileReloading(config, Role.CLIENT, keystoreWatcher, trustStoreWatcher, resetContext);
+  }
+
+  /**
+   * Enable certificate file reloading for the RPC <em>server</em> side. See
+   * {@link #enableCertFileReloadingForClient} for parameter semantics; the only difference is that
+   * paths are resolved via the {@link Role#SERVER} role-scoped keys, falling back to the legacy
+   * unscoped keys when unset.
+   */
+  public static void enableCertFileReloadingForServer(Configuration config,
+    AtomicReference<FileChangeWatcher> keystoreWatcher,
+    AtomicReference<FileChangeWatcher> trustStoreWatcher, Runnable resetContext)
+    throws IOException {
+    enableCertFileReloading(config, Role.SERVER, keystoreWatcher, trustStoreWatcher, resetContext);
+  }
+
+  private static void enableCertFileReloading(Configuration config, Role role,
+    AtomicReference<FileChangeWatcher> keystoreWatcher,
+    AtomicReference<FileChangeWatcher> trustStoreWatcher, Runnable resetContext)
+    throws IOException {
+    String keyStoreLocation;
+    String trustStoreLocation;
+    if (role == Role.CLIENT) {
+      keyStoreLocation = resolveConfig(config, TLS_CONFIG_CLIENT_KEYSTORE_LOCATION,
+        TLS_CONFIG_KEYSTORE_LOCATION, "");
+      trustStoreLocation = resolveConfig(config, TLS_CONFIG_CLIENT_TRUSTSTORE_LOCATION,
+        TLS_CONFIG_TRUSTSTORE_LOCATION, "");
+    } else {
+      keyStoreLocation = resolveConfig(config, TLS_CONFIG_SERVER_KEYSTORE_LOCATION,
+        TLS_CONFIG_KEYSTORE_LOCATION, "");
+      trustStoreLocation = resolveConfig(config, TLS_CONFIG_SERVER_TRUSTSTORE_LOCATION,
+        TLS_CONFIG_TRUSTSTORE_LOCATION, "");
+    }
     keystoreWatcher.set(newFileChangeWatcher(config, keyStoreLocation, resetContext));
-    String trustStoreLocation = config.get(TLS_CONFIG_TRUSTSTORE_LOCATION, "");
     // we are using the same callback for both. there's no reason to kick off two
     // threads if keystore/truststore are both at the same location
     if (!keyStoreLocation.equals(trustStoreLocation)) {
diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestFileChangeWatcher.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestFileChangeWatcher.java
index 4b822c9d402d..48f74cf57801 100644
--- a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestFileChangeWatcher.java
+++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestFileChangeWatcher.java
@@ -111,7 +111,7 @@ public void testEnableCertFileReloading() throws IOException {
     myConf.set(X509Util.TLS_CONFIG_TRUSTSTORE_LOCATION, sharedPath);
     AtomicReference<FileChangeWatcher> keystoreWatcher = new AtomicReference<>();
     AtomicReference<FileChangeWatcher> truststoreWatcher = new AtomicReference<>();
-    X509Util.enableCertFileReloading(myConf, keystoreWatcher, truststoreWatcher, () -> {
+    X509Util.enableCertFileReloadingForServer(myConf, keystoreWatcher, truststoreWatcher, () -> {
     });
     assertNotNull(keystoreWatcher.get());
     assertThat(keystoreWatcher.get().getWatcherThread().getName(), endsWith("foo.jks"));
@@ -122,7 +122,7 @@ public void testEnableCertFileReloading() throws IOException {
 
     String truststorePath = File.createTempFile("bar", "bar.jks").getAbsolutePath();
     myConf.set(X509Util.TLS_CONFIG_TRUSTSTORE_LOCATION, truststorePath);
-    X509Util.enableCertFileReloading(myConf, keystoreWatcher, truststoreWatcher, () -> {
+    X509Util.enableCertFileReloadingForServer(myConf, keystoreWatcher, truststoreWatcher, () -> {
     });
 
     assertNotNull(keystoreWatcher.get());
diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/tls/TestX509Util.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/tls/TestX509Util.java
index f2499fc24d7d..0c4ed1a2085c 100644
--- a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/tls/TestX509Util.java
+++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/tls/TestX509Util.java
@@ -20,6 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
@@ -375,4 +376,142 @@ public void testLoadPKCS12TrustStoreWithWrongPassword() {
     });
   }
 
+  // ---------------------------------------------------------------------------
+  // Role-scoped configuration resolution (single-EKU certificate support)
+  // ---------------------------------------------------------------------------
+
+  @TestTemplate
+  public void testResolveConfigPrefersRoleScopedOverLegacy() {
+    conf.set("test.role", "role-value");
+    conf.set("test.legacy", "legacy-value");
+    assertEquals("role-value",
+      X509Util.resolveConfig(conf, "test.role", "test.legacy", "default-value"));
+  }
+
+  @TestTemplate
+  public void testResolveConfigFallsBackToLegacyWhenRoleUnset() {
+    conf.unset("test.role");
+    conf.set("test.legacy", "legacy-value");
+    assertEquals("legacy-value",
+      X509Util.resolveConfig(conf, "test.role", "test.legacy", "default-value"));
+  }
+
+  @TestTemplate
+  public void testResolveConfigReturnsDefaultWhenBothUnset() {
+    conf.unset("test.role");
+    conf.unset("test.legacy");
+    assertEquals("default-value",
+      X509Util.resolveConfig(conf, "test.role", "test.legacy", "default-value"));
+  }
+
+  @TestTemplate
+  public void testResolvePasswordPrefersRoleScopedOverLegacy() throws Exception {
+    conf.set("test.role.password", "role-pw");
+    conf.set("test.legacy.password", "legacy-pw");
+    assertArrayEquals("role-pw".toCharArray(),
+      X509Util.resolvePassword(conf, "test.role.password", "test.legacy.password"));
+  }
+
+  @TestTemplate
+  public void testResolvePasswordFallsBackToLegacyWhenRoleUnset() throws Exception {
+    conf.unset("test.role.password");
+    conf.set("test.legacy.password", "legacy-pw");
+    assertArrayEquals("legacy-pw".toCharArray(),
+      X509Util.resolvePassword(conf, "test.role.password", "test.legacy.password"));
+  }
+
+  @TestTemplate
+  public void testResolvePasswordReturnsNullWhenBothUnset() throws Exception {
+    conf.unset("test.role.password");
+    conf.unset("test.legacy.password");
+    assertNull(X509Util.resolvePassword(conf, "test.role.password", "test.legacy.password"));
+  }
+
+  @TestTemplate
+  public void testCreateSSLContextForClientUsesRoleScopedKeystoreWhenSet() throws Exception {
+    // Move the legacy keystore values to the client-scoped keys and clear the legacy keys, so
+    // that a successful context build proves the client-scoped keys were consulted.
+    String location = conf.get(X509Util.TLS_CONFIG_KEYSTORE_LOCATION);
+    String password = conf.get(X509Util.TLS_CONFIG_KEYSTORE_PASSWORD);
+    String type = conf.get(X509Util.TLS_CONFIG_KEYSTORE_TYPE);
+    conf.set(X509Util.TLS_CONFIG_CLIENT_KEYSTORE_LOCATION, location);
+    conf.set(X509Util.TLS_CONFIG_CLIENT_KEYSTORE_PASSWORD, password);
+    conf.set(X509Util.TLS_CONFIG_CLIENT_KEYSTORE_TYPE, type);
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_PASSWORD);
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_TYPE);
+
+    SslContext sslContext = X509Util.createSslContextForClient(conf);
+    ByteBufAllocator byteBufAllocatorMock = mock(ByteBufAllocator.class);
+    // Handshake would fail if the key manager weren't wired; smoke-test that engine creation works.
+    assertTrue(sslContext.newEngine(byteBufAllocatorMock).getSSLParameters().getProtocols().length
+        > 0);
+  }
+
+  @TestTemplate
+  public void testCreateSSLContextForClientFallsBackToLegacyKeystore() throws Exception {
+    // The base setUp() only sets the legacy TLS_CONFIG_KEYSTORE_* / TLS_CONFIG_TRUSTSTORE_* keys.
+    // The role-scoped keys are intentionally unset; the context must still build using the legacy
+    // values (backward-compat regression guard).
+    conf.unset(X509Util.TLS_CONFIG_CLIENT_KEYSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_CLIENT_KEYSTORE_PASSWORD);
+    conf.unset(X509Util.TLS_CONFIG_CLIENT_KEYSTORE_TYPE);
+    conf.unset(X509Util.TLS_CONFIG_CLIENT_TRUSTSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_CLIENT_TRUSTSTORE_PASSWORD);
+    conf.unset(X509Util.TLS_CONFIG_CLIENT_TRUSTSTORE_TYPE);
+
+    SslContext sslContext = X509Util.createSslContextForClient(conf);
+    ByteBufAllocator byteBufAllocatorMock = mock(ByteBufAllocator.class);
+    assertTrue(sslContext.newEngine(byteBufAllocatorMock).getSSLParameters().getProtocols().length
+        > 0);
+  }
+
+  @TestTemplate
+  public void testCreateSSLContextForServerUsesRoleScopedKeystoreWhenSet() throws Exception {
+    String location = conf.get(X509Util.TLS_CONFIG_KEYSTORE_LOCATION);
+    String password = conf.get(X509Util.TLS_CONFIG_KEYSTORE_PASSWORD);
+    String type = conf.get(X509Util.TLS_CONFIG_KEYSTORE_TYPE);
+    conf.set(X509Util.TLS_CONFIG_SERVER_KEYSTORE_LOCATION, location);
+    conf.set(X509Util.TLS_CONFIG_SERVER_KEYSTORE_PASSWORD, password);
+    conf.set(X509Util.TLS_CONFIG_SERVER_KEYSTORE_TYPE, type);
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_PASSWORD);
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_TYPE);
+
+    SslContext sslContext = X509Util.createSslContextForServer(conf);
+    ByteBufAllocator byteBufAllocatorMock = mock(ByteBufAllocator.class);
+    assertTrue(sslContext.newEngine(byteBufAllocatorMock).getSSLParameters().getProtocols().length
+        > 0);
+  }
+
+  @TestTemplate
+  public void testCreateSSLContextForServerFallsBackToLegacyKeystore() throws Exception {
+    // Base setUp() only sets legacy keys. Assert server-side context still builds; backward
+    // compatibility for existing deployments that only know about the legacy key namespace.
+    conf.unset(X509Util.TLS_CONFIG_SERVER_KEYSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_SERVER_KEYSTORE_PASSWORD);
+    conf.unset(X509Util.TLS_CONFIG_SERVER_KEYSTORE_TYPE);
+    conf.unset(X509Util.TLS_CONFIG_SERVER_TRUSTSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_SERVER_TRUSTSTORE_PASSWORD);
+    conf.unset(X509Util.TLS_CONFIG_SERVER_TRUSTSTORE_TYPE);
+
+    SslContext sslContext = X509Util.createSslContextForServer(conf);
+    ByteBufAllocator byteBufAllocatorMock = mock(ByteBufAllocator.class);
+    assertTrue(sslContext.newEngine(byteBufAllocatorMock).getSSLParameters().getProtocols().length
+        > 0);
+  }
+
+  @TestTemplate
+  public void testCreateSSLContextForServerThrowsWhenNeitherKeystoreSet() {
+    conf.unset(X509Util.TLS_CONFIG_KEYSTORE_LOCATION);
+    conf.unset(X509Util.TLS_CONFIG_SERVER_KEYSTORE_LOCATION);
+    SSLContextException ex =
+      assertThrows(SSLContextException.class, () -> X509Util.createSslContextForServer(conf));
+    // The error should name both keys so the operator knows what to set.
+    assertTrue(ex.getMessage().contains(X509Util.TLS_CONFIG_SERVER_KEYSTORE_LOCATION),
+      "message should mention role-scoped key, got: " + ex.getMessage());
+    assertTrue(ex.getMessage().contains(X509Util.TLS_CONFIG_KEYSTORE_LOCATION),
+      "message should mention legacy key, got: " + ex.getMessage());
+  }
+
 }
diff --git a/hbase-http/src/main/java/org/apache/hadoop/hbase/http/HttpServer.java b/hbase-http/src/main/java/org/apache/hadoop/hbase/http/HttpServer.java
index 49af0939d5a0..bc10d4c03f1c 100644
--- a/hbase-http/src/main/java/org/apache/hadoop/hbase/http/HttpServer.java
+++ b/hbase-http/src/main/java/org/apache/hadoop/hbase/http/HttpServer.java
@@ -227,6 +227,7 @@ public static class Builder {
     private String usernameConfKey;
     private String keytabConfKey;
     private boolean needsClientAuth;
+    private boolean wantsClientAuth;
     private String includeCiphers;
     private String excludeCiphers;
     private String includeProtocols;
@@ -311,13 +312,27 @@ public Builder keyPassword(String password) {
     }
 
     /**
-     * Specify whether the server should authorize the client in SSL connections.
+     * Specify whether the server should require a client certificate during the SSL handshake
+     * (mTLS). When true, clients that do not present a valid certificate are rejected.
+     * <p>
+     * Takes precedence over {@link #wantsClientAuth(boolean)} in Jetty when both are set.
      */
     public Builder needsClientAuth(boolean value) {
       this.needsClientAuth = value;
       return this;
     }
 
+    /**
+     * Specify whether the server should request a client certificate during the SSL handshake but
+     * still accept clients that do not present one. Weaker than
+     * {@link #needsClientAuth(boolean)}: use this to signal "opportunistic mTLS" where a client
+     * cert is validated when supplied but its absence is tolerated.
+     */
+    public Builder wantsClientAuth(boolean value) {
+      this.wantsClientAuth = value;
+      return this;
+    }
+
     /**
      * @see #setAppDir(String)
      * @deprecated Since 0.99.0. Use {@link #setAppDir(String)} instead.
@@ -476,6 +491,7 @@ public HttpServer build() throws IOException {
           httpsConfig.addCustomizer(new SecureRequestCustomizer());
           SslContextFactory.Server sslCtxFactory = new SslContextFactory.Server();
           sslCtxFactory.setNeedClientAuth(needsClientAuth);
+          sslCtxFactory.setWantClientAuth(wantsClientAuth);
           sslCtxFactory.setKeyManagerPassword(keyPassword);
 
           if (keyStore != null) {
diff --git a/hbase-http/src/main/java/org/apache/hadoop/hbase/http/InfoServer.java b/hbase-http/src/main/java/org/apache/hadoop/hbase/http/InfoServer.java
index 5a09315ed774..d3cfd1e79155 100644
--- a/hbase-http/src/main/java/org/apache/hadoop/hbase/http/InfoServer.java
+++ b/hbase-http/src/main/java/org/apache/hadoop/hbase/http/InfoServer.java
@@ -25,6 +25,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.io.crypto.tls.X509Util;
 import org.apache.hadoop.security.authorize.AccessControlList;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -44,6 +45,16 @@ public class InfoServer {
 
   private static final String HADOOP_WEB_TLS_CONFIG_PREFIX = "ssl.server.";
   private static final String HBASE_WEB_TLS_CONFIG_PREFIX = "hbase.ui.ssl.";
+  // Role-scoped prefix for single-EKU certificate support. When set, takes precedence over both
+  // HBASE_WEB_TLS_CONFIG_PREFIX and HADOOP_WEB_TLS_CONFIG_PREFIX. The UI process only ever plays
+  // the TLS-server role, so no parallel .client. prefix is defined.
+  private static final String HBASE_WEB_TLS_SERVER_CONFIG_PREFIX = "hbase.ui.ssl.server.";
+  /**
+   * Config key controlling whether the UI's TLS connector requests or requires a client
+   * certificate. Valid values: {@code NONE} (default), {@code WANT}, {@code NEED}.
+   */
+  static final String HBASE_UI_SSL_CLIENT_AUTH_MODE =
+    HBASE_WEB_TLS_SERVER_CONFIG_PREFIX + "client.auth.mode";
 
   /**
    * Create a status server on the given port. The jsp scripts are taken from
@@ -83,6 +94,16 @@ public InfoServer(String name, String bindAddress, int port, boolean findPort,
         .setExcludeProtocols(getTLSProperty(c, "exclude.protocols"))
         .setIncludeCiphers(getTLSProperty(c, "include.cipher.list"))
         .setExcludeCiphers(getTLSProperty(c, "exclude.cipher.list"));
+
+      // Activate mutual TLS if configured. Default is NONE, which preserves today's behavior of
+      // never requesting a client certificate on the UI connector (leaving any configured
+      // truststore inert for peer verification). Set hbase.ui.ssl.server.client.auth.mode to
+      // WANT or NEED to opt in. The client.auth.mode key is looked up directly on
+      // HBASE_WEB_TLS_SERVER_CONFIG_PREFIX; there is no legacy or Hadoop-prefixed fallback.
+      X509Util.ClientAuth clientAuth = X509Util.ClientAuth
+        .fromPropertyValue(c.get(HBASE_UI_SSL_CLIENT_AUTH_MODE, X509Util.ClientAuth.NONE.name()));
+      builder.needsClientAuth(clientAuth == X509Util.ClientAuth.NEED)
+        .wantsClientAuth(clientAuth == X509Util.ClientAuth.WANT);
     }
 
     final String httpAuthType = c.get(HttpServer.HTTP_UI_AUTHENTICATION, "").toLowerCase();
@@ -104,18 +125,31 @@ public InfoServer(String name, String bindAddress, int port, boolean findPort,
     this.httpServer = builder.build();
   }
 
-  private String getTLSPassword(Configuration c, String postfix) throws IOException {
-    return HBaseConfiguration.getPassword(c, HBASE_WEB_TLS_CONFIG_PREFIX + postfix,
-      HBaseConfiguration.getPassword(c, HADOOP_WEB_TLS_CONFIG_PREFIX + postfix, null));
+  /**
+   * Resolves a TLS-related password with a 3-tier fallback: the role-scoped
+   * {@code hbase.ui.ssl.server.<postfix>} key is checked first, then the unscoped
+   * {@code hbase.ui.ssl.<postfix>} key, and finally Hadoop's {@code ssl.server.<postfix>}.
+   * Package-private for direct testing.
+   */
+  static String getTLSPassword(Configuration c, String postfix) throws IOException {
+    return HBaseConfiguration.getPassword(c, HBASE_WEB_TLS_SERVER_CONFIG_PREFIX + postfix,
+      HBaseConfiguration.getPassword(c, HBASE_WEB_TLS_CONFIG_PREFIX + postfix,
+        HBaseConfiguration.getPassword(c, HADOOP_WEB_TLS_CONFIG_PREFIX + postfix, null)));
   }
 
-  private String getTLSProperty(Configuration c, String postfix) {
+  static String getTLSProperty(Configuration c, String postfix) {
     return getTLSProperty(c, postfix, null);
   }
 
-  private String getTLSProperty(Configuration c, String postfix, String defaultValue) {
-    return c.get(HBASE_WEB_TLS_CONFIG_PREFIX + postfix,
-      c.get(HADOOP_WEB_TLS_CONFIG_PREFIX + postfix, defaultValue));
+  /**
+   * Resolves a TLS-related property with a 3-tier fallback: role-scoped
+   * {@code hbase.ui.ssl.server.<postfix>} → {@code hbase.ui.ssl.<postfix>} →
+   * {@code ssl.server.<postfix>} → {@code defaultValue}. Package-private for direct testing.
+   */
+  static String getTLSProperty(Configuration c, String postfix, String defaultValue) {
+    return c.get(HBASE_WEB_TLS_SERVER_CONFIG_PREFIX + postfix,
+      c.get(HBASE_WEB_TLS_CONFIG_PREFIX + postfix,
+        c.get(HADOOP_WEB_TLS_CONFIG_PREFIX + postfix, defaultValue)));
   }
 
   /**
diff --git a/hbase-http/src/test/java/org/apache/hadoop/hbase/http/TestInfoServerTLSConfig.java b/hbase-http/src/test/java/org/apache/hadoop/hbase/http/TestInfoServerTLSConfig.java
new file mode 100644
index 000000000000..12f836dab791
--- /dev/null
+++ b/hbase-http/src/test/java/org/apache/hadoop/hbase/http/TestInfoServerTLSConfig.java
@@ -0,0 +1,100 @@
+/*
+ * 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.http;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.testclassification.MiscTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link InfoServer}'s TLS-config resolution: the 3-tier fallback chain
+ * ({@code hbase.ui.ssl.server.*} → {@code hbase.ui.ssl.*} → {@code ssl.server.*}) that
+ * underpins single-EKU certificate support on the UI surface, and the client-auth-mode key.
+ */
+@Tag(MiscTests.TAG)
+@Tag(SmallTests.TAG)
+public class TestInfoServerTLSConfig {
+
+  private static final String POSTFIX = "keystore.location";
+  private static final String ROLE_SCOPED_KEY = "hbase.ui.ssl.server." + POSTFIX;
+  private static final String HBASE_KEY = "hbase.ui.ssl." + POSTFIX;
+  private static final String HADOOP_KEY = "ssl.server." + POSTFIX;
+
+  @Test
+  public void testRoleScopedTakesPrecedenceOverHBasePrefixed() {
+    Configuration c = new Configuration(false);
+    c.set(ROLE_SCOPED_KEY, "role-scoped-value");
+    c.set(HBASE_KEY, "hbase-prefixed-value");
+    c.set(HADOOP_KEY, "hadoop-prefixed-value");
+    assertEquals("role-scoped-value", InfoServer.getTLSProperty(c, POSTFIX));
+  }
+
+  @Test
+  public void testHBasePrefixedTakesPrecedenceOverHadoopPrefixed() {
+    Configuration c = new Configuration(false);
+    // No role-scoped key set — hbase-prefixed key must win.
+    c.set(HBASE_KEY, "hbase-prefixed-value");
+    c.set(HADOOP_KEY, "hadoop-prefixed-value");
+    assertEquals("hbase-prefixed-value", InfoServer.getTLSProperty(c, POSTFIX));
+  }
+
+  @Test
+  public void testFallsBackToHadoopPrefixedWhenNoOthersSet() {
+    Configuration c = new Configuration(false);
+    c.set(HADOOP_KEY, "hadoop-prefixed-value");
+    assertEquals("hadoop-prefixed-value", InfoServer.getTLSProperty(c, POSTFIX));
+  }
+
+  @Test
+  public void testReturnsDefaultWhenNoneSet() {
+    Configuration c = new Configuration(false);
+    assertEquals("jks", InfoServer.getTLSProperty(c, POSTFIX, "jks"));
+    assertNull(InfoServer.getTLSProperty(c, POSTFIX));
+  }
+
+  @Test
+  public void testGetTLSPasswordHonorsThreeTierFallback() throws Exception {
+    Configuration c = new Configuration(false);
+    // Configuration.getPassword returns null when unset; verifying the fallback picks each tier.
+    c.set(ROLE_SCOPED_KEY.replace("keystore.location", "keystore.password"), "role-pw");
+    c.set(HBASE_KEY.replace("keystore.location", "keystore.password"), "hbase-pw");
+    c.set(HADOOP_KEY.replace("keystore.location", "keystore.password"), "hadoop-pw");
+    assertEquals("role-pw", InfoServer.getTLSPassword(c, "keystore.password"));
+
+    c.unset(ROLE_SCOPED_KEY.replace("keystore.location", "keystore.password"));
+    assertEquals("hbase-pw", InfoServer.getTLSPassword(c, "keystore.password"));
+
+    c.unset(HBASE_KEY.replace("keystore.location", "keystore.password"));
+    assertEquals("hadoop-pw", InfoServer.getTLSPassword(c, "keystore.password"));
+
+    c.unset(HADOOP_KEY.replace("keystore.location", "keystore.password"));
+    assertNull(InfoServer.getTLSPassword(c, "keystore.password"));
+  }
+
+  @Test
+  public void testClientAuthModeKeyIsRoleScoped() {
+    // Guard against a "double server.server." regression: the client-auth-mode config key must
+    // resolve to the single role-scoped key, not to a nested/prefixed form.
+    assertEquals("hbase.ui.ssl.server.client.auth.mode", InfoServer.HBASE_UI_SSL_CLIENT_AUTH_MODE);
+  }
+}
\ No newline at end of file
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/Constants.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/Constants.java
index b3d603c660ee..473effa3635b 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/Constants.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/Constants.java
@@ -60,6 +60,31 @@ public interface Constants {
   String REST_SSL_EXCLUDE_PROTOCOLS = "hbase.rest.ssl.exclude.protocols";
   String REST_SSL_INCLUDE_PROTOCOLS = "hbase.rest.ssl.include.protocols";
 
+  // ---------------------------------------------------------------------------
+  // Role-scoped SSL configuration for single-EKU certificate support.
+  //
+  // The server-scoped keys are what RESTServer.java actually reads: hbase.rest.ssl.server.*
+  // takes precedence over the corresponding unscoped hbase.rest.ssl.* keys; when unset the
+  // unscoped keys are used as a fallback so existing deployments keep working. RESTServer
+  // only ever plays the TLS-server role on its port, so no parallel .client.* configuration
+  // is defined.
+  // ---------------------------------------------------------------------------
+  String REST_SSL_SERVER_KEYSTORE_STORE = "hbase.rest.ssl.server.keystore.store";
+  String REST_SSL_SERVER_KEYSTORE_PASSWORD = "hbase.rest.ssl.server.keystore.password";
+  String REST_SSL_SERVER_KEYSTORE_KEYPASSWORD = "hbase.rest.ssl.server.keystore.keypassword";
+  String REST_SSL_SERVER_KEYSTORE_TYPE = "hbase.rest.ssl.server.keystore.type";
+  String REST_SSL_SERVER_TRUSTSTORE_STORE = "hbase.rest.ssl.server.truststore.store";
+  String REST_SSL_SERVER_TRUSTSTORE_PASSWORD = "hbase.rest.ssl.server.truststore.password";
+  String REST_SSL_SERVER_TRUSTSTORE_TYPE = "hbase.rest.ssl.server.truststore.type";
+
+  /**
+   * Client-auth mode for the REST server's TLS connector. Valid values: {@code NONE},
+   * {@code WANT}, {@code NEED} (see {@code X509Util.ClientAuth}). Default is {@code NONE}, which
+   * preserves the historical behavior of never requesting client certificates. Set to
+   * {@code NEED} to enforce mTLS.
+   */
+  String REST_SSL_CLIENT_AUTH_MODE = "hbase.rest.ssl.server.client.auth.mode";
+
   String REST_THREAD_POOL_THREADS_MAX = "hbase.rest.threads.max";
   String REST_THREAD_POOL_THREADS_MIN = "hbase.rest.threads.min";
   String REST_THREAD_POOL_TASK_QUEUE_SIZE = "hbase.rest.task.queue.size";
diff --git a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RESTServer.java b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RESTServer.java
index 149d634cbd5f..4ade1324788c 100644
--- a/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RESTServer.java
+++ b/hbase-rest/src/main/java/org/apache/hadoop/hbase/rest/RESTServer.java
@@ -37,6 +37,7 @@
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.http.HttpServerUtil;
 import org.apache.hadoop.hbase.http.InfoServer;
+import org.apache.hadoop.hbase.io.crypto.tls.X509Util;
 import org.apache.hadoop.hbase.log.HBaseMarkers;
 import org.apache.hadoop.hbase.rest.filter.AuthFilter;
 import org.apache.hadoop.hbase.rest.filter.GzipFilter;
@@ -305,11 +306,16 @@ public synchronized void run() throws Exception {
       httpsConfig.addCustomizer(new SecureRequestCustomizer());
 
       SslContextFactory.Server sslCtxFactory = new SslContextFactory.Server();
-      String keystore = conf.get(REST_SSL_KEYSTORE_STORE);
-      String keystoreType = conf.get(REST_SSL_KEYSTORE_TYPE);
-      String password = HBaseConfiguration.getPassword(conf, REST_SSL_KEYSTORE_PASSWORD, null);
-      String keyPassword =
-        HBaseConfiguration.getPassword(conf, REST_SSL_KEYSTORE_KEYPASSWORD, password);
+      // Prefer the role-scoped hbase.rest.ssl.server.* keys, falling back to the historical
+      // unscoped hbase.rest.ssl.* keys for backward compatibility with existing deployments.
+      String keystore =
+        X509Util.resolveConfig(conf, REST_SSL_SERVER_KEYSTORE_STORE, REST_SSL_KEYSTORE_STORE, null);
+      String keystoreType =
+        X509Util.resolveConfig(conf, REST_SSL_SERVER_KEYSTORE_TYPE, REST_SSL_KEYSTORE_TYPE, null);
+      String password = HBaseConfiguration.getPassword(conf, REST_SSL_SERVER_KEYSTORE_PASSWORD,
+        HBaseConfiguration.getPassword(conf, REST_SSL_KEYSTORE_PASSWORD, null));
+      String keyPassword = HBaseConfiguration.getPassword(conf, REST_SSL_SERVER_KEYSTORE_KEYPASSWORD,
+        HBaseConfiguration.getPassword(conf, REST_SSL_KEYSTORE_KEYPASSWORD, password));
       sslCtxFactory.setKeyStorePath(keystore);
       if (StringUtils.isNotBlank(keystoreType)) {
         sslCtxFactory.setKeyStoreType(keystoreType);
@@ -317,20 +323,41 @@ public synchronized void run() throws Exception {
       sslCtxFactory.setKeyStorePassword(password);
       sslCtxFactory.setKeyManagerPassword(keyPassword);
 
-      String trustStore = conf.get(REST_SSL_TRUSTSTORE_STORE);
+      String trustStore = X509Util.resolveConfig(conf, REST_SSL_SERVER_TRUSTSTORE_STORE,
+        REST_SSL_TRUSTSTORE_STORE, null);
       if (StringUtils.isNotBlank(trustStore)) {
         sslCtxFactory.setTrustStorePath(trustStore);
       }
       String trustStorePassword =
-        HBaseConfiguration.getPassword(conf, REST_SSL_TRUSTSTORE_PASSWORD, null);
+        HBaseConfiguration.getPassword(conf, REST_SSL_SERVER_TRUSTSTORE_PASSWORD,
+          HBaseConfiguration.getPassword(conf, REST_SSL_TRUSTSTORE_PASSWORD, null));
       if (StringUtils.isNotBlank(trustStorePassword)) {
         sslCtxFactory.setTrustStorePassword(trustStorePassword);
       }
-      String trustStoreType = conf.get(REST_SSL_TRUSTSTORE_TYPE);
+      String trustStoreType = X509Util.resolveConfig(conf, REST_SSL_SERVER_TRUSTSTORE_TYPE,
+        REST_SSL_TRUSTSTORE_TYPE, null);
       if (StringUtils.isNotBlank(trustStoreType)) {
         sslCtxFactory.setTrustStoreType(trustStoreType);
       }
 
+      // Activate mTLS if configured. Default is NONE, which preserves today's behavior of never
+      // requesting a client certificate — even when a truststore is configured. Set
+      // hbase.rest.ssl.server.client.auth.mode to WANT or NEED to opt in.
+      X509Util.ClientAuth clientAuth = X509Util.ClientAuth
+        .fromPropertyValue(conf.get(REST_SSL_CLIENT_AUTH_MODE, X509Util.ClientAuth.NONE.name()));
+      switch (clientAuth) {
+        case NEED:
+          sslCtxFactory.setNeedClientAuth(true);
+          break;
+        case WANT:
+          sslCtxFactory.setWantClientAuth(true);
+          break;
+        case NONE:
+        default:
+          // no-op; both flags default to false on SslContextFactory.Server
+          break;
+      }
+
       String[] excludeCiphers = servlet.getConfiguration()
         .getStrings(REST_SSL_EXCLUDE_CIPHER_SUITES, ArrayUtils.EMPTY_STRING_ARRAY);
       if (excludeCiphers.length != 0) {
diff --git a/hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/TestRESTServerSSL.java b/hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/TestRESTServerSSL.java
index a123f714a019..f665a53d5eaa 100644
--- a/hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/TestRESTServerSSL.java
+++ b/hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/TestRESTServerSSL.java
@@ -160,6 +160,101 @@ public void testSslConnectionUsingKeystoreFormatPKCS12() throws Exception {
     assertEquals(200, response.getCode());
   }
 
+  // ---------------------------------------------------------------------------
+  // Role-scoped configuration + mTLS (single-EKU certificate support).
+  // ---------------------------------------------------------------------------
+
+  /**
+   * Server started with only the role-scoped hbase.rest.ssl.server.* keys (and legacy keys
+   * unset). A successful SSL connection proves the server-scoped keys were consulted.
+   */
+  @Test
+  public void testSslConnectionUsingRoleScopedServerKeys() throws Exception {
+    // Move the legacy passwords set in beforeEachTest onto the role-scoped keys and clear the
+    // legacy passwords so a successful start proves the server-scoped keys are what the code
+    // actually picked up.
+    conf.unset(Constants.REST_SSL_KEYSTORE_PASSWORD);
+    conf.unset(Constants.REST_SSL_KEYSTORE_KEYPASSWORD);
+    conf.unset(Constants.REST_SSL_TRUSTSTORE_PASSWORD);
+    conf.set(Constants.REST_SSL_SERVER_KEYSTORE_PASSWORD, KEY_STORE_PASSWORD);
+    conf.set(Constants.REST_SSL_SERVER_KEYSTORE_KEYPASSWORD, KEY_STORE_PASSWORD);
+    conf.set(Constants.REST_SSL_SERVER_TRUSTSTORE_PASSWORD, TRUST_STORE_PASSWORD);
+    conf.set(Constants.REST_SSL_SERVER_KEYSTORE_STORE, getKeystoreFilePath("jks"));
+    conf.set(Constants.REST_SSL_SERVER_TRUSTSTORE_STORE, getTruststoreFilePath("jks"));
+
+    REST_TEST_UTIL.startServletContainer(conf);
+    Cluster localCluster = new Cluster().add("localhost", REST_TEST_UTIL.getServletPort());
+    sslClient = new Client(localCluster, getTruststoreFilePath("jks"),
+      Optional.of(TRUST_STORE_PASSWORD), Optional.empty());
+
+    Response response = sslClient.get("/version", Constants.MIMETYPE_TEXT);
+    assertEquals(200, response.getCode());
+  }
+
+  /**
+   * Backward-compatibility regression: existing deployments that know only about the legacy
+   * unscoped keys must continue to work exactly as before. This mirrors {@link #testSslConnection}
+   * but names the intent explicitly.
+   */
+  @Test
+  public void testSslConnectionFallsBackToLegacyKeystoreKeys() throws Exception {
+    // Make sure no role-scoped key is set — the beforeEachTest configures only legacy passwords,
+    // so this is a fresh state check.
+    conf.unset(Constants.REST_SSL_SERVER_KEYSTORE_STORE);
+    conf.unset(Constants.REST_SSL_SERVER_KEYSTORE_PASSWORD);
+    conf.unset(Constants.REST_SSL_SERVER_KEYSTORE_KEYPASSWORD);
+    conf.unset(Constants.REST_SSL_SERVER_KEYSTORE_TYPE);
+    conf.unset(Constants.REST_SSL_SERVER_TRUSTSTORE_STORE);
+    conf.unset(Constants.REST_SSL_SERVER_TRUSTSTORE_PASSWORD);
+    conf.unset(Constants.REST_SSL_SERVER_TRUSTSTORE_TYPE);
+
+    startRESTServerWithDefaultKeystoreType();
+
+    Response response = sslClient.get("/version", Constants.MIMETYPE_TEXT);
+    assertEquals(200, response.getCode());
+  }
+
+  /**
+   * With {@code client.auth.mode=NONE} (the default), a client that presents no client
+   * certificate is accepted — matching today's behavior.
+   */
+  @Test
+  public void testClientAuthNoneAcceptsClientWithoutCert() throws Exception {
+    conf.set(Constants.REST_SSL_CLIENT_AUTH_MODE, "NONE");
+    startRESTServerWithDefaultKeystoreType();
+
+    Response response = sslClient.get("/version", Constants.MIMETYPE_TEXT);
+    assertEquals(200, response.getCode());
+  }
+
+  /**
+   * With {@code client.auth.mode=WANT}, an anonymous client (no client cert) is still accepted;
+   * the server requests a cert but does not require it.
+   */
+  @Test
+  public void testClientAuthWantAllowsAnonymousClient() throws Exception {
+    conf.set(Constants.REST_SSL_CLIENT_AUTH_MODE, "WANT");
+    startRESTServerWithDefaultKeystoreType();
+
+    Response response = sslClient.get("/version", Constants.MIMETYPE_TEXT);
+    assertEquals(200, response.getCode());
+  }
+
+  /**
+   * With {@code client.auth.mode=NEED}, an anonymous client (no client cert) is rejected during
+   * the TLS handshake. The base {@link Client} configures truststore-only, so it presents no key
+   * material to the server.
+   */
+  @Test
+  public void testClientAuthNeedRejectsClientWithoutCert() throws Exception {
+    conf.set(Constants.REST_SSL_CLIENT_AUTH_MODE, "NEED");
+    startRESTServerWithDefaultKeystoreType();
+
+    // The mTLS handshake fails before any HTTP-level status is returned; Apache HttpClient
+    // surfaces this as ClientProtocolException (same failure mode as testNonSslClientDenied).
+    assertThrows(ClientProtocolException.class, () -> sslClient.get("/version"));
+  }
+
   private static File initKeystoreDir() {
     String dataTestDir = TEST_UTIL.getDataTestDir().toString();
     File keystoreDir = new File(dataTestDir, TestRESTServerSSL.class.getSimpleName() + "_keys");
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServer.java
index b21b6e19c78e..718279cc800e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServer.java
@@ -465,7 +465,7 @@ SslContext getSslContext() throws X509Exception, IOException {
         keyStoreWatcher.get() == null && trustStoreWatcher.get() == null
           && conf.getBoolean(X509Util.TLS_CERT_RELOAD, false)
       ) {
-        X509Util.enableCertFileReloading(conf, keyStoreWatcher, trustStoreWatcher,
+        X509Util.enableCertFileReloadingForServer(conf, keyStoreWatcher, trustStoreWatcher,
           () -> sslContextForServer.set(null));
       }
     }
diff --git a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/Constants.java b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/Constants.java
index 83bb90eecefa..7c683ae09a45 100644
--- a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/Constants.java
+++ b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/Constants.java
@@ -71,6 +71,38 @@ private Constants() {
   public static final String THRIFT_SSL_KEYSTORE_TYPE_KEY = "hbase.thrift.ssl.keystore.type";
   public static final String THRIFT_SSL_KEYSTORE_TYPE_DEFAULT = "jks";
 
+  /**
+   * Role-scoped SSL configuration for single-EKU certificate support.
+   *
+   * The server-scoped hbase.thrift.ssl.server.* keys are what ThriftServer reads on the
+   * Thrift-over-HTTP transport; when set they take precedence over the corresponding
+   * unscoped hbase.thrift.ssl.* keys; when unset the unscoped keys are used as a fallback
+   * so existing deployments keep working.
+   */
+  public static final String THRIFT_SSL_SERVER_KEYSTORE_STORE_KEY =
+    "hbase.thrift.ssl.server.keystore.store";
+  public static final String THRIFT_SSL_SERVER_KEYSTORE_PASSWORD_KEY =
+    "hbase.thrift.ssl.server.keystore.password";
+  public static final String THRIFT_SSL_SERVER_KEYSTORE_KEYPASSWORD_KEY =
+    "hbase.thrift.ssl.server.keystore.keypassword";
+  public static final String THRIFT_SSL_SERVER_KEYSTORE_TYPE_KEY =
+    "hbase.thrift.ssl.server.keystore.type";
+  public static final String THRIFT_SSL_SERVER_TRUSTSTORE_STORE_KEY =
+    "hbase.thrift.ssl.server.truststore.store";
+  public static final String THRIFT_SSL_SERVER_TRUSTSTORE_PASSWORD_KEY =
+    "hbase.thrift.ssl.server.truststore.password";
+  public static final String THRIFT_SSL_SERVER_TRUSTSTORE_TYPE_KEY =
+    "hbase.thrift.ssl.server.truststore.type";
+
+  /**
+   * Client-auth mode for the Thrift server's HTTP-transport TLS connector. Valid values:
+   * {@code NONE}, {@code WANT}, {@code NEED} (see {@code X509Util.ClientAuth}). Default is
+   * {@code NONE}, which preserves the historical behavior of never requesting client
+   * certificates.
+   */
+  public static final String THRIFT_SSL_CLIENT_AUTH_MODE_KEY =
+    "hbase.thrift.ssl.server.client.auth.mode";
+
   public static final String THRIFT_SUPPORT_PROXYUSER_KEY = "hbase.thrift.support.proxyuser";
 
   // kerberos related configs
diff --git a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServer.java b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServer.java
index e72090f22957..6d8116bbe7ac 100644
--- a/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServer.java
+++ b/hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServer.java
@@ -68,11 +68,19 @@
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_EXCLUDE_PROTOCOLS_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_INCLUDE_CIPHER_SUITES_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_INCLUDE_PROTOCOLS_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_CLIENT_AUTH_MODE_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_PASSWORD_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_STORE_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_TYPE_DEFAULT;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_KEYSTORE_TYPE_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_KEYSTORE_KEYPASSWORD_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_KEYSTORE_PASSWORD_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_KEYSTORE_STORE_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_KEYSTORE_TYPE_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_TRUSTSTORE_PASSWORD_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_TRUSTSTORE_STORE_KEY;
+import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SSL_SERVER_TRUSTSTORE_TYPE_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.THRIFT_SUPPORT_PROXYUSER_KEY;
 import static org.apache.hadoop.hbase.thrift.Constants.USE_HTTP_CONF_KEY;
 
@@ -93,6 +101,7 @@
 import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.SaslServer;
 import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.hbase.HBaseConfiguration;
@@ -100,6 +109,7 @@
 import org.apache.hadoop.hbase.filter.ParseFilter;
 import org.apache.hadoop.hbase.http.HttpServerUtil;
 import org.apache.hadoop.hbase.http.InfoServer;
+import org.apache.hadoop.hbase.io.crypto.tls.X509Util;
 import org.apache.hadoop.hbase.log.HBaseMarkers;
 import org.apache.hadoop.hbase.security.SaslUtil;
 import org.apache.hadoop.hbase.security.SecurityUtil;
@@ -413,16 +423,56 @@ protected void setupHTTPServer() throws IOException {
       httpsConfig.addCustomizer(new SecureRequestCustomizer());
 
       SslContextFactory.Server sslCtxFactory = new SslContextFactory.Server();
-      String keystore = conf.get(THRIFT_SSL_KEYSTORE_STORE_KEY);
-      String password =
-        HBaseConfiguration.getPassword(conf, THRIFT_SSL_KEYSTORE_PASSWORD_KEY, null);
-      String keyPassword =
-        HBaseConfiguration.getPassword(conf, THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY, password);
+      // Prefer the role-scoped hbase.thrift.ssl.server.* keys, falling back to the historical
+      // unscoped hbase.thrift.ssl.* keys for backward compatibility with existing deployments.
+      String keystore = X509Util.resolveConfig(conf, THRIFT_SSL_SERVER_KEYSTORE_STORE_KEY,
+        THRIFT_SSL_KEYSTORE_STORE_KEY, null);
+      String password = HBaseConfiguration.getPassword(conf, THRIFT_SSL_SERVER_KEYSTORE_PASSWORD_KEY,
+        HBaseConfiguration.getPassword(conf, THRIFT_SSL_KEYSTORE_PASSWORD_KEY, null));
+      String keyPassword = HBaseConfiguration.getPassword(conf,
+        THRIFT_SSL_SERVER_KEYSTORE_KEYPASSWORD_KEY,
+        HBaseConfiguration.getPassword(conf, THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY, password));
       sslCtxFactory.setKeyStorePath(keystore);
       sslCtxFactory.setKeyStorePassword(password);
       sslCtxFactory.setKeyManagerPassword(keyPassword);
-      sslCtxFactory
-        .setKeyStoreType(conf.get(THRIFT_SSL_KEYSTORE_TYPE_KEY, THRIFT_SSL_KEYSTORE_TYPE_DEFAULT));
+      sslCtxFactory.setKeyStoreType(X509Util.resolveConfig(conf, THRIFT_SSL_SERVER_KEYSTORE_TYPE_KEY,
+        THRIFT_SSL_KEYSTORE_TYPE_KEY, THRIFT_SSL_KEYSTORE_TYPE_DEFAULT));
+
+      // Truststore is entirely new for Thrift — no legacy fallback because there is no historical
+      // hbase.thrift.ssl.truststore.* configuration. When left unset, no truststore is configured
+      // on the connector and any hbase.thrift.ssl.server.client.auth.mode = WANT/NEED setting
+      // will fail the handshake for lack of a peer-cert trust root.
+      String trustStore = conf.get(THRIFT_SSL_SERVER_TRUSTSTORE_STORE_KEY);
+      if (StringUtils.isNotBlank(trustStore)) {
+        sslCtxFactory.setTrustStorePath(trustStore);
+        String trustStorePassword =
+          HBaseConfiguration.getPassword(conf, THRIFT_SSL_SERVER_TRUSTSTORE_PASSWORD_KEY, null);
+        if (StringUtils.isNotBlank(trustStorePassword)) {
+          sslCtxFactory.setTrustStorePassword(trustStorePassword);
+        }
+        String trustStoreType = conf.get(THRIFT_SSL_SERVER_TRUSTSTORE_TYPE_KEY);
+        if (StringUtils.isNotBlank(trustStoreType)) {
+          sslCtxFactory.setTrustStoreType(trustStoreType);
+        }
+      }
+
+      // Activate mTLS if configured. Default is NONE, which preserves today's behavior of never
+      // requesting a client certificate on the Thrift-over-HTTP connector. Set
+      // hbase.thrift.ssl.server.client.auth.mode to WANT or NEED to opt in.
+      X509Util.ClientAuth clientAuth = X509Util.ClientAuth
+        .fromPropertyValue(conf.get(THRIFT_SSL_CLIENT_AUTH_MODE_KEY, X509Util.ClientAuth.NONE.name()));
+      switch (clientAuth) {
+        case NEED:
+          sslCtxFactory.setNeedClientAuth(true);
+          break;
+        case WANT:
+          sslCtxFactory.setWantClientAuth(true);
+          break;
+        case NONE:
+        default:
+          // no-op; both flags default to false on SslContextFactory.Server
+          break;
+      }
 
       String[] excludeCiphers =
         conf.getStrings(THRIFT_SSL_EXCLUDE_CIPHER_SUITES_KEY, ArrayUtils.EMPTY_STRING_ARRAY);
diff --git a/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerSSLMutualAuth.java b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerSSLMutualAuth.java
new file mode 100644
index 000000000000..ccca393e0fd7
--- /dev/null
+++ b/hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerSSLMutualAuth.java
@@ -0,0 +1,361 @@
+/*
+ * 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.thrift;
+
+import static org.apache.hadoop.hbase.thrift.TestThriftServerCmdLine.createBoundServer;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Method;
+import java.net.HttpURLConnection;
+import java.nio.file.Files;
+import java.security.KeyPair;
+import java.security.KeyStore;
+import java.security.cert.X509Certificate;
+import javax.net.ssl.SSLContext;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.thrift.generated.Hbase;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
+import org.apache.hadoop.hbase.util.IncrementingEnvironmentEdge;
+import org.apache.hadoop.hbase.util.TableDescriptorChecker;
+import org.apache.hadoop.security.ssl.KeyStoreTestUtil;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.ssl.SSLContexts;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.hbase.thirdparty.org.apache.thrift.protocol.TProtocol;
+import org.apache.hbase.thirdparty.org.apache.thrift.transport.TMemoryBuffer;
+
+/**
+ * Exercises the role-scoped {@code hbase.thrift.ssl.server.*} configuration and the new
+ * mutual-TLS activation on the Thrift-over-HTTP transport. Complements
+ * {@link TestThriftHttpServerSSL}, which covers plain (server-only) TLS termination.
+ */
+@Tag(ClientTests.TAG)
+@Tag(LargeTests.TAG)
+public class TestThriftServerSSLMutualAuth {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TestThriftServerSSLMutualAuth.class);
+  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+  private static final String KEY_STORE_PASSWORD = "myKSPassword";
+  private static final String TRUST_STORE_PASSWORD = "myTSPassword";
+  private static final String CLIENT_KEY_STORE_PASSWORD = "myClientKSPassword";
+
+  private File keyDir;
+  private ThriftServerRunner tsr;
+  private HttpPost httpPost;
+
+  @BeforeAll
+  public static void setUpBeforeClass() throws Exception {
+    TEST_UTIL.getConfiguration().setBoolean(Constants.USE_HTTP_CONF_KEY, true);
+    TEST_UTIL.getConfiguration().setBoolean(TableDescriptorChecker.TABLE_SANITY_CHECKS, false);
+    TEST_UTIL.startMiniCluster();
+    EnvironmentEdgeManagerTestHelper.injectEdge(new IncrementingEnvironmentEdge());
+  }
+
+  @AfterAll
+  public static void tearDownAfterClass() throws Exception {
+    TEST_UTIL.shutdownMiniCluster();
+    EnvironmentEdgeManager.reset();
+  }
+
+  @BeforeEach
+  public void setUp() throws Exception {
+    initializeAlgorithmId();
+    keyDir = initKeystoreDir();
+    keyDir.deleteOnExit();
+
+    // Server identity + a truststore holding the server's cert (used by the test client to
+    // trust the server).
+    KeyPair serverKeyPair = KeyStoreTestUtil.generateKeyPair("RSA");
+    X509Certificate serverCertificate = KeyStoreTestUtil.generateCertificate(
+      "CN=localhost, O=server", serverKeyPair, 30, "SHA1withRSA");
+    generateTrustStore(getServerTruststoreFilePath(), serverCertificate);
+    generateKeyStore(getServerKeystoreFilePath(), serverKeyPair, serverCertificate);
+
+    // Distinct client cert (single-EKU clientAuth in spirit) and a truststore holding that cert
+    // — this is what the server uses to validate presented client certificates.
+    KeyPair clientKeyPair = KeyStoreTestUtil.generateKeyPair("RSA");
+    X509Certificate clientCertificate = KeyStoreTestUtil.generateCertificate("CN=client, O=client",
+      clientKeyPair, 30, "SHA1withRSA");
+    generateTrustStore(getClientCaTruststoreFilePath(), clientCertificate);
+    generateKeyStoreWithPassword(getClientKeystoreFilePath(), clientKeyPair, clientCertificate,
+      CLIENT_KEY_STORE_PASSWORD);
+  }
+
+  @AfterEach
+  public void tearDown() throws IOException {
+    if (httpPost != null) {
+      httpPost.releaseConnection();
+    }
+    if (tsr != null) {
+      tsr.close();
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+  // Role-scoped keystore configuration.
+  // ---------------------------------------------------------------------------
+
+  /**
+   * With only the role-scoped {@code hbase.thrift.ssl.server.keystore.*} keys set (legacy keys
+   * unset), a plain HTTPS request succeeds — proving the server-scoped keys are the ones the
+   * bootstrap actually reads.
+   */
+  @Test
+  public void testServerUsesRoleScopedKeystoreWhenSet() throws Exception {
+    Configuration conf = baseConf();
+    conf.set(Constants.THRIFT_SSL_SERVER_KEYSTORE_STORE_KEY, getServerKeystoreFilePath());
+    conf.set(Constants.THRIFT_SSL_SERVER_KEYSTORE_PASSWORD_KEY, KEY_STORE_PASSWORD);
+    conf.set(Constants.THRIFT_SSL_SERVER_KEYSTORE_KEYPASSWORD_KEY, KEY_STORE_PASSWORD);
+
+    startServer(conf);
+    doRequestExpectingSuccess(clientBuilderWithServerTrust());
+  }
+
+  /**
+   * Backward-compatibility regression: existing deployments that know only about the legacy
+   * unscoped keys must continue to work.
+   */
+  @Test
+  public void testServerFallsBackToLegacyKeystore() throws Exception {
+    Configuration conf = baseConf();
+    conf.set(Constants.THRIFT_SSL_KEYSTORE_STORE_KEY, getServerKeystoreFilePath());
+    conf.set(Constants.THRIFT_SSL_KEYSTORE_PASSWORD_KEY, KEY_STORE_PASSWORD);
+    conf.set(Constants.THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY, KEY_STORE_PASSWORD);
+
+    startServer(conf);
+    doRequestExpectingSuccess(clientBuilderWithServerTrust());
+  }
+
+  // ---------------------------------------------------------------------------
+  // Mutual TLS (client auth mode).
+  // ---------------------------------------------------------------------------
+
+  /**
+   * With {@code client.auth.mode=NONE} (default), a client presenting no certificate is
+   * accepted — matches today's behavior.
+   */
+  @Test
+  public void testClientAuthNoneAcceptsClientWithoutCert() throws Exception {
+    Configuration conf = baseConfWithLegacyKeystore();
+    conf.set(Constants.THRIFT_SSL_CLIENT_AUTH_MODE_KEY, "NONE");
+
+    startServer(conf);
+    doRequestExpectingSuccess(clientBuilderWithServerTrust());
+  }
+
+  /** {@code WANT} lets anonymous clients through. */
+  @Test
+  public void testClientAuthWantAllowsAnonymousClient() throws Exception {
+    Configuration conf = baseConfWithLegacyKeystore();
+    // WANT needs the server to know which CAs it would trust if a client did present a cert.
+    conf.set(Constants.THRIFT_SSL_SERVER_TRUSTSTORE_STORE_KEY, getClientCaTruststoreFilePath());
+    conf.set(Constants.THRIFT_SSL_SERVER_TRUSTSTORE_PASSWORD_KEY, TRUST_STORE_PASSWORD);
+    conf.set(Constants.THRIFT_SSL_CLIENT_AUTH_MODE_KEY, "WANT");
+
+    startServer(conf);
+    doRequestExpectingSuccess(clientBuilderWithServerTrust());
+  }
+
+  /**
+   * {@code NEED} rejects clients that do not present a valid client certificate. The handshake
+   * fails before any HTTP response is produced.
+   */
+  @Test
+  public void testClientAuthNeedRejectsClientWithoutCert() throws Exception {
+    Configuration conf = baseConfWithLegacyKeystore();
+    conf.set(Constants.THRIFT_SSL_SERVER_TRUSTSTORE_STORE_KEY, getClientCaTruststoreFilePath());
+    conf.set(Constants.THRIFT_SSL_SERVER_TRUSTSTORE_PASSWORD_KEY, TRUST_STORE_PASSWORD);
+    conf.set(Constants.THRIFT_SSL_CLIENT_AUTH_MODE_KEY, "NEED");
+
+    startServer(conf);
+    // The Apache HttpClient surface may raise either SSLHandshakeException directly or wrap it
+    // in an IOException — both are acceptable evidence that the handshake was refused.
+    assertThrows(IOException.class, () -> doRequest(clientBuilderWithServerTrust()));
+  }
+
+  /** {@code NEED} accepts a client that presents a certificate trusted by the server. */
+  @Test
+  public void testClientAuthNeedAcceptsClientWithValidCert() throws Exception {
+    Configuration conf = baseConfWithLegacyKeystore();
+    conf.set(Constants.THRIFT_SSL_SERVER_TRUSTSTORE_STORE_KEY, getClientCaTruststoreFilePath());
+    conf.set(Constants.THRIFT_SSL_SERVER_TRUSTSTORE_PASSWORD_KEY, TRUST_STORE_PASSWORD);
+    conf.set(Constants.THRIFT_SSL_CLIENT_AUTH_MODE_KEY, "NEED");
+
+    startServer(conf);
+    doRequestExpectingSuccess(clientBuilderWithMutualTrust());
+  }
+
+  // ---------------------------------------------------------------------------
+  // Helpers
+  // ---------------------------------------------------------------------------
+
+  private Configuration baseConf() throws Exception {
+    Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
+    conf.setBoolean(Constants.THRIFT_SSL_ENABLED_KEY, true);
+    return conf;
+  }
+
+  private Configuration baseConfWithLegacyKeystore() throws Exception {
+    Configuration conf = baseConf();
+    conf.set(Constants.THRIFT_SSL_KEYSTORE_STORE_KEY, getServerKeystoreFilePath());
+    conf.set(Constants.THRIFT_SSL_KEYSTORE_PASSWORD_KEY, KEY_STORE_PASSWORD);
+    conf.set(Constants.THRIFT_SSL_KEYSTORE_KEYPASSWORD_KEY, KEY_STORE_PASSWORD);
+    return conf;
+  }
+
+  private void startServer(Configuration conf) throws Exception {
+    tsr = createBoundServer(() -> new ThriftServer(conf));
+    String url = "https://" + HConstants.LOCALHOST + ":" + tsr.getThriftServer().listenPort;
+    httpPost = new HttpPost(url);
+    httpPost.setHeader("Content-Type", "application/x-thrift");
+    httpPost.setHeader("Accept", "application/x-thrift");
+    httpPost.setHeader("User-Agent", "Java/THttpClient/HC");
+  }
+
+  private HttpClientBuilder clientBuilderWithServerTrust() throws Exception {
+    KeyStore trustStore = loadJksTrustStore(getServerTruststoreFilePath(), TRUST_STORE_PASSWORD);
+    SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(trustStore, null).build();
+    return HttpClients.custom().setSSLContext(sslContext);
+  }
+
+  /**
+   * Builds a client that (a) trusts the server's cert and (b) presents the client keystore. Used
+   * to satisfy {@code client.auth.mode=NEED}.
+   */
+  private HttpClientBuilder clientBuilderWithMutualTrust() throws Exception {
+    KeyStore trustStore = loadJksTrustStore(getServerTruststoreFilePath(), TRUST_STORE_PASSWORD);
+    KeyStore clientKs;
+    try (InputStream in = new BufferedInputStream(
+      Files.newInputStream(new File(getClientKeystoreFilePath()).toPath()))) {
+      clientKs = KeyStore.getInstance("JKS");
+      clientKs.load(in, CLIENT_KEY_STORE_PASSWORD.toCharArray());
+    }
+    SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(trustStore, null)
+      .loadKeyMaterial(clientKs, CLIENT_KEY_STORE_PASSWORD.toCharArray()).build();
+    return HttpClients.custom().setSSLContext(sslContext);
+  }
+
+  private void doRequestExpectingSuccess(HttpClientBuilder builder) throws Exception {
+    try (CloseableHttpClient httpClient = builder.build()) {
+      CloseableHttpResponse response = doOneRoundTrip(httpClient);
+      assertEquals(HttpURLConnection.HTTP_OK, response.getStatusLine().getStatusCode());
+    }
+  }
+
+  private void doRequest(HttpClientBuilder builder) throws Exception {
+    try (CloseableHttpClient httpClient = builder.build()) {
+      doOneRoundTrip(httpClient);
+    }
+  }
+
+  private CloseableHttpResponse doOneRoundTrip(CloseableHttpClient httpClient) throws Exception {
+    TMemoryBuffer memoryBuffer = new TMemoryBuffer(100);
+    TProtocol prot = new TBinaryProtocol(memoryBuffer);
+    Hbase.Client client = new Hbase.Client(prot);
+    client.send_getClusterId();
+    httpPost.setEntity(new ByteArrayEntity(memoryBuffer.getArray()));
+    return httpClient.execute(httpPost);
+  }
+
+  private static KeyStore loadJksTrustStore(String path, String password) throws Exception {
+    try (InputStream in = new BufferedInputStream(Files.newInputStream(new File(path).toPath()))) {
+      KeyStore ks = KeyStore.getInstance("JKS");
+      ks.load(in, password.toCharArray());
+      return ks;
+    }
+  }
+
+  // Workaround for jdk8 292 bug. See https://github.com/bcgit/bc-java/issues/941
+  // Below is a workaround described in above URL. Issue fingered first in comments in
+  // HBASE-25920 Support Hadoop 3.3.1
+  private static void initializeAlgorithmId() {
+    try {
+      Class<?> algoId = Class.forName("sun.security.x509.AlgorithmId");
+      Method method = algoId.getMethod("get", String.class);
+      method.setAccessible(true);
+      method.invoke(null, "PBEWithSHA1AndDESede");
+    } catch (Exception e) {
+      LOG.warn("failed to initialize AlgorithmId", e);
+    }
+  }
+
+  private File initKeystoreDir() {
+    String dataTestDir = TEST_UTIL.getDataTestDir().toString();
+    File keystoreDir =
+      new File(dataTestDir, TestThriftServerSSLMutualAuth.class.getSimpleName() + "_keys");
+    keystoreDir.mkdirs();
+    return keystoreDir;
+  }
+
+  private static void generateKeyStore(String keyStorePath, KeyPair keyPair, X509Certificate cert)
+    throws Exception {
+    KeyStoreTestUtil.createKeyStore(keyStorePath, KEY_STORE_PASSWORD, KEY_STORE_PASSWORD,
+      "serverKS", keyPair.getPrivate(), cert);
+  }
+
+  private static void generateKeyStoreWithPassword(String keyStorePath, KeyPair keyPair,
+    X509Certificate cert, String password) throws Exception {
+    KeyStoreTestUtil.createKeyStore(keyStorePath, password, password, "clientKS",
+      keyPair.getPrivate(), cert);
+  }
+
+  private static void generateTrustStore(String path, X509Certificate cert) throws Exception {
+    KeyStoreTestUtil.createTrustStore(path, TRUST_STORE_PASSWORD, "ts", cert);
+  }
+
+  private String getServerKeystoreFilePath() {
+    return String.format("%s/serverKS.jks", keyDir.getAbsolutePath());
+  }
+
+  private String getServerTruststoreFilePath() {
+    return String.format("%s/serverTS.jks", keyDir.getAbsolutePath());
+  }
+
+  private String getClientKeystoreFilePath() {
+    return String.format("%s/clientKS.jks", keyDir.getAbsolutePath());
+  }
+
+  private String getClientCaTruststoreFilePath() {
+    return String.format("%s/clientCA.jks", keyDir.getAbsolutePath());
+  }
+
+}
\ No newline at end of file