HBASE-14700 Support a permissive mode for secure clusters to allow SIMPLE auth clients

hbase-common/src/main/resources/hbase-default.xml

@@ -922,6 +922,17 @@ possible configurations would overwhelm and obscure the important.
       When false (the default), the client will not allow the fallback to SIMPLE
       authentication, and will abort the connection.</description>
   </property>
+  <property>
+    <name>hbase.ipc.server.fallback-to-simple-auth-allowed</name>
+    <value>false</value>
+    <description>When a server is configured to require secure connections, it will
+      reject connection attempts from clients using SASL SIMPLE (unsecure) authentication.
+      This setting allows secure servers to accept SASL SIMPLE connections from clients
+      when the client requests.  When false (the default), the server will not allow the fallback
+      to SIMPLE authentication, and will reject the connection.  WARNING: This setting should ONLY
+      be used as a temporary measure while converting clients over to secure authentication.  It
+      MUST BE DISABLED for secure operation.</description>
+  </property>
   <property>
     <name>hbase.coprocessor.enabled</name>
     <value>true</value>

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServerSource.java

@@ -34,6 +34,9 @@ public interface MetricsHBaseServerSource extends BaseSource {
   String AUTHENTICATION_FAILURES_NAME = "authenticationFailures";
   String AUTHENTICATION_FAILURES_DESC =
       "Number of authentication failures.";
+  String AUTHENTICATION_FALLBACKS_NAME = "authenticationFallbacks";
+  String AUTHENTICATION_FALLBACKS_DESC =
+      "Number of fallbacks to insecure authentication.";
   String SENT_BYTES_NAME = "sentBytes";
   String SENT_BYTES_DESC = "Number of bytes sent.";
   String RECEIVED_BYTES_NAME = "receivedBytes";
@@ -80,6 +83,8 @@ public interface MetricsHBaseServerSource extends BaseSource {
 
   void authenticationFailure();
 
+  void authenticationFallback();
+
   void exception();
 
   /**

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServerSourceImpl.java

@@ -36,6 +36,7 @@ public class MetricsHBaseServerSourceImpl extends BaseSourceImpl
   private final MutableCounterLong authorizationFailures;
   private final MutableCounterLong authenticationSuccesses;
   private final MutableCounterLong authenticationFailures;
+  private final MutableCounterLong authenticationFallbacks;
   private final MutableCounterLong sentBytes;
   private final MutableCounterLong receivedBytes;
 
@@ -85,6 +86,8 @@ public class MetricsHBaseServerSourceImpl extends BaseSourceImpl
         AUTHENTICATION_SUCCESSES_NAME, AUTHENTICATION_SUCCESSES_DESC, 0L);
     this.authenticationFailures = this.getMetricsRegistry().newCounter(AUTHENTICATION_FAILURES_NAME,
         AUTHENTICATION_FAILURES_DESC, 0L);
+    this.authenticationFallbacks = this.getMetricsRegistry().newCounter(
+        AUTHENTICATION_FALLBACKS_NAME, AUTHENTICATION_FALLBACKS_DESC, 0L);
     this.sentBytes = this.getMetricsRegistry().newCounter(SENT_BYTES_NAME,
         SENT_BYTES_DESC, 0L);
     this.receivedBytes = this.getMetricsRegistry().newCounter(RECEIVED_BYTES_NAME,
@@ -116,6 +119,11 @@ public class MetricsHBaseServerSourceImpl extends BaseSourceImpl
     authenticationFailures.incr();
   }
 
+  @Override
+  public void authenticationFallback() {
+    authenticationFallbacks.incr();
+  }
+
   @Override
   public void exception() {
     exceptions.incr();

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServer.java

@@ -53,6 +53,10 @@ public class MetricsHBaseServer {
     source.authenticationSuccess();
   }
 
+  void authenticationFallback() {
+    source.authenticationFallback();
+  }
+
   void sentBytes(long count) {
     source.sentBytes(count);
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java

@@ -81,6 +81,7 @@ import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.NeedUnmanagedConnectionException;
 import org.apache.hadoop.hbase.client.Operation;
 import org.apache.hadoop.hbase.codec.Codec;
+import org.apache.hadoop.hbase.conf.ConfigurationObserver;
 import org.apache.hadoop.hbase.exceptions.RegionMovedException;
 import org.apache.hadoop.hbase.io.ByteBufferOutputStream;
 import org.apache.hadoop.hbase.io.BoundedByteBufferPool;
@@ -157,7 +158,7 @@ import com.google.protobuf.TextFormat;
  */
 @InterfaceAudience.LimitedPrivate({HBaseInterfaceAudience.COPROC, HBaseInterfaceAudience.PHOENIX})
 @InterfaceStability.Evolving
-public class RpcServer implements RpcServerInterface {
+public class RpcServer implements RpcServerInterface, ConfigurationObserver {
   // LOG is being used in CallRunner and the log level is being changed in tests
   public static final Log LOG = LogFactory.getLog(RpcServer.class);
   private static final CallQueueTooBigException CALL_QUEUE_TOO_BIG_EXCEPTION
@@ -168,6 +169,12 @@ public class RpcServer implements RpcServerInterface {
 
   public static final byte CURRENT_VERSION = 0;
 
+  /**
+   * Whether we allow a fallback to SIMPLE auth for insecure clients when security is enabled.
+   */
+  public static final String FALLBACK_TO_INSECURE_CLIENT_AUTH =
+          "hbase.ipc.server.fallback-to-simple-auth-allowed";
+
   /**
    * How many calls/handler are allowed in the queue.
    */
@@ -276,6 +283,7 @@ public class RpcServer implements RpcServerInterface {
 
   private final BoundedByteBufferPool reservoir;
 
+  private volatile boolean allowFallbackToSimpleAuth;
 
   /**
    * Datastructure that holds all necessary to a method invocation and then afterward, carries
@@ -1237,6 +1245,9 @@ public class RpcServer implements RpcServerInterface {
     private final Call saslCall =
       new Call(SASL_CALLID, this.service, null, null, null, null, this, null, 0, null, null);
 
+    // was authentication allowed with a fallback to simple auth
+    private boolean authenticatedWithFallback;
+
     public UserGroupInformation attemptingUser = null; // user name before auth
     protected User user = null;
     protected UserGroupInformation ugi = null;
@@ -1313,19 +1324,21 @@ public class RpcServer implements RpcServerInterface {
 
     private UserGroupInformation getAuthorizedUgi(String authorizedId)
         throws IOException {
+      UserGroupInformation authorizedUgi;
       if (authMethod == AuthMethod.DIGEST) {
         TokenIdentifier tokenId = HBaseSaslRpcServer.getIdentifier(authorizedId,
             secretManager);
-        UserGroupInformation ugi = tokenId.getUser();
+        authorizedUgi = tokenId.getUser();
-        if (ugi == null) {
+        if (authorizedUgi == null) {
           throw new AccessDeniedException(
               "Can't retrieve username from tokenIdentifier.");
         }
-        ugi.addTokenIdentifier(tokenId);
+        authorizedUgi.addTokenIdentifier(tokenId);
-        return ugi;
       } else {
-        return UserGroupInformation.createRemoteUser(authorizedId);
+        authorizedUgi = UserGroupInformation.createRemoteUser(authorizedId);
       }
+      authorizedUgi.setAuthenticationMethod(authMethod.authenticationMethod.getAuthMethod());
+      return authorizedUgi;
     }
 
     private void saslReadAndProcess(byte[] saslToken) throws IOException,
@@ -1504,10 +1517,15 @@ public class RpcServer implements RpcServerInterface {
         return doBadPreambleHandling(msg, new BadAuthException(msg));
       }
       if (isSecurityEnabled && authMethod == AuthMethod.SIMPLE) {
-        AccessDeniedException ae = new AccessDeniedException("Authentication is required");
+        if (allowFallbackToSimpleAuth) {
-        setupResponse(authFailedResponse, authFailedCall, ae, ae.getMessage());
+          metrics.authenticationFallback();
-        responder.doRespond(authFailedCall);
+          authenticatedWithFallback = true;
-        throw ae;
+        } else {
+          AccessDeniedException ae = new AccessDeniedException("Authentication is required");
+          setupResponse(authFailedResponse, authFailedCall, ae, ae.getMessage());
+          responder.doRespond(authFailedCall);
+          throw ae;
+        }
       }
       if (!isSecurityEnabled && authMethod != AuthMethod.SIMPLE) {
         doRawSaslReply(SaslStatus.SUCCESS, new IntWritable(
@@ -1654,6 +1672,12 @@ public class RpcServer implements RpcServerInterface {
         if (ugi != null) {
           ugi.setAuthenticationMethod(AuthMethod.SIMPLE.authenticationMethod);
         }
+        // audit logging for SASL authenticated users happens in saslReadAndProcess()
+        if (authenticatedWithFallback) {
+          LOG.warn("Allowed fallback to SIMPLE auth for " + ugi
+              + " connecting from " + getHostAddress());
+        }
+        AUDITLOG.info(AUTH_SUCCESSFUL_FOR + ugi);
       } else {
         // user is authenticated
         ugi.setAuthenticationMethod(authMethod.authenticationMethod);
@@ -2010,10 +2034,31 @@ public class RpcServer implements RpcServerInterface {
     if (isSecurityEnabled) {
       HBaseSaslRpcServer.init(conf);
     }
+    initReconfigurable(conf);
+
     this.scheduler = scheduler;
     this.scheduler.init(new RpcSchedulerContext(this));
   }
 
+  @Override
+  public void onConfigurationChange(Configuration newConf) {
+    initReconfigurable(newConf);
+  }
+
+  private void initReconfigurable(Configuration confToLoad) {
+    this.allowFallbackToSimpleAuth = confToLoad.getBoolean(FALLBACK_TO_INSECURE_CLIENT_AUTH, false);
+    if (isSecurityEnabled && allowFallbackToSimpleAuth) {
+      LOG.warn("********* WARNING! *********");
+      LOG.warn("This server is configured to allow connections from INSECURE clients");
+      LOG.warn("(" + FALLBACK_TO_INSECURE_CLIENT_AUTH + " = true).");
+      LOG.warn("While this option is enabled, client identities cannot be secured, and user");
+      LOG.warn("impersonation is possible!");
+      LOG.warn("For secure operation, please disable SIMPLE authentication as soon as possible,");
+      LOG.warn("by setting " + FALLBACK_TO_INSECURE_CLIENT_AUTH + " = false in hbase-site.xml");
+      LOG.warn("****************************");
+    }
+  }
+
   /**
    * Subclasses of HBaseServer can override this to provide their own
    * Connection implementations.

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -876,6 +876,7 @@ public class HRegionServer extends HasThread implements
   private void registerConfigurationObservers() {
     // Registering the compactSplitThread object with the ConfigurationManager.
     configurationManager.registerObserver(this.compactSplitThread);
+    configurationManager.registerObserver(this.rpcServices);
   }
 
   /**

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

@@ -67,6 +67,7 @@ import org.apache.hadoop.hbase.client.RegionReplicaUtil;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.RowMutations;
 import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.conf.ConfigurationObserver;
 import org.apache.hadoop.hbase.coordination.CloseRegionCoordination;
 import org.apache.hadoop.hbase.coordination.OpenRegionCoordination;
 import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
@@ -192,7 +193,8 @@ import com.google.protobuf.TextFormat;
 @InterfaceAudience.Private
 @SuppressWarnings("deprecation")
 public class RSRpcServices implements HBaseRPCErrorHandler,
-    AdminService.BlockingInterface, ClientService.BlockingInterface, PriorityFunction {
+    AdminService.BlockingInterface, ClientService.BlockingInterface, PriorityFunction,
+    ConfigurationObserver {
   protected static final Log LOG = LogFactory.getLog(RSRpcServices.class);
 
   /** RPC scheduler to use for the region server. */
@@ -900,6 +902,13 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
     rs.setName(name);
   }
 
+  @Override
+  public void onConfigurationChange(Configuration newConf) {
+    if (rpcServer instanceof ConfigurationObserver) {
+      ((ConfigurationObserver)rpcServer).onConfigurationChange(newConf);
+    }
+  }
+
   protected PriorityFunction createPriority() {
     return new AnnotationReadingPriorityFunction(this);
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/security/TestSecureRPC.java

@@ -22,6 +22,7 @@ import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getKeytabFileF
 import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getPrincipalForTesting;
 import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getSecuredConfiguration;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertSame;
 
 import java.io.File;
@@ -99,21 +100,27 @@ public class TestSecureRPC {
     testRpcCallWithEnabledKerberosSaslAuth(RpcClientImpl.class);
   }
 
+  @Test
+  public void testRpcWithInsecureFallback() throws Exception {
+    testRpcFallbackToSimpleAuth(RpcClientImpl.class);
+  }
+
   @Test
   public void testAsyncRpc() throws Exception {
     testRpcCallWithEnabledKerberosSaslAuth(AsyncRpcClient.class);
   }
 
+  @Test
+  public void testAsyncRpcWithInsecureFallback() throws Exception {
+    testRpcFallbackToSimpleAuth(AsyncRpcClient.class);
+  }
+
   private void testRpcCallWithEnabledKerberosSaslAuth(Class<? extends RpcClient> rpcImplClass)
       throws Exception {
     String krbKeytab = getKeytabFileForTesting();
     String krbPrincipal = getPrincipalForTesting();
 
-    Configuration cnf = new Configuration();
+    UserGroupInformation ugi = loginKerberosPrincipal(krbKeytab, krbPrincipal);
-    cnf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "kerberos");
-    UserGroupInformation.setConfiguration(cnf);
-    UserGroupInformation.loginUserFromKeytab(krbPrincipal, krbKeytab);
-    UserGroupInformation ugi = UserGroupInformation.getLoginUser();
     UserGroupInformation ugi2 = UserGroupInformation.getCurrentUser();
 
     // check that the login user is okay:
@@ -121,8 +128,28 @@ public class TestSecureRPC {
     assertEquals(AuthenticationMethod.KERBEROS, ugi.getAuthenticationMethod());
     assertEquals(krbPrincipal, ugi.getUserName());
 
+    Configuration clientConf = getSecuredConfiguration();
+    callRpcService(rpcImplClass, User.create(ugi2), clientConf, false);
+  }
+
+  private UserGroupInformation loginKerberosPrincipal(String krbKeytab, String krbPrincipal)
+      throws Exception {
+    Configuration cnf = new Configuration();
+    cnf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "kerberos");
+    UserGroupInformation.setConfiguration(cnf);
+    UserGroupInformation.loginUserFromKeytab(krbPrincipal, krbKeytab);
+    return UserGroupInformation.getLoginUser();
+  }
+
+  private void callRpcService(Class<? extends RpcClient> rpcImplClass, User clientUser,
+                              Configuration clientConf, boolean allowInsecureFallback)
+      throws Exception {
+    Configuration clientConfCopy = new Configuration(clientConf);
+    clientConfCopy.set(RpcClientFactory.CUSTOM_RPC_CLIENT_IMPL_CONF_KEY, rpcImplClass.getName());
+
     Configuration conf = getSecuredConfiguration();
-    conf.set(RpcClientFactory.CUSTOM_RPC_CLIENT_IMPL_CONF_KEY, rpcImplClass.getName());
+    conf.setBoolean(RpcServer.FALLBACK_TO_INSECURE_CLIENT_AUTH, allowInsecureFallback);
+
     SecurityInfo securityInfoMock = Mockito.mock(SecurityInfo.class);
     Mockito.when(securityInfoMock.getServerPrincipal())
         .thenReturn(HBaseKerberosUtils.KRB_PRINCIPAL);
@@ -140,7 +167,7 @@ public class TestSecureRPC {
             conf, new FifoRpcScheduler(conf, 1));
     rpcServer.start();
     RpcClient rpcClient =
-        RpcClientFactory.createClient(conf, HConstants.DEFAULT_CLUSTER_ID.toString());
+        RpcClientFactory.createClient(clientConfCopy, HConstants.DEFAULT_CLUSTER_ID.toString());
     try {
       InetSocketAddress address = rpcServer.getListenerAddress();
       if (address == null) {
@@ -149,7 +176,7 @@ public class TestSecureRPC {
       BlockingRpcChannel channel =
           rpcClient.createBlockingRpcChannel(
             ServerName.valueOf(address.getHostName(), address.getPort(),
-            System.currentTimeMillis()), User.getCurrent(), 5000);
+            System.currentTimeMillis()), clientUser, 5000);
       TestDelayedRpcProtos.TestDelayedService.BlockingInterface stub =
           TestDelayedRpcProtos.TestDelayedService.newBlockingStub(channel);
       List<Integer> results = new ArrayList<Integer>();
@@ -163,4 +190,26 @@ public class TestSecureRPC {
       rpcServer.stop();
     }
   }
+
+  public void testRpcFallbackToSimpleAuth(Class<? extends RpcClient> rpcImplClass) throws Exception {
+    String krbKeytab = getKeytabFileForTesting();
+    String krbPrincipal = getPrincipalForTesting();
+
+    UserGroupInformation ugi = loginKerberosPrincipal(krbKeytab, krbPrincipal);
+    assertEquals(AuthenticationMethod.KERBEROS, ugi.getAuthenticationMethod());
+    assertEquals(krbPrincipal, ugi.getUserName());
+
+    String clientUsername = "testuser";
+    UserGroupInformation clientUgi = UserGroupInformation.createUserForTesting(clientUsername,
+        new String[]{clientUsername});
+
+    // check that the client user is insecure
+    assertNotSame(ugi, clientUgi);
+    assertEquals(AuthenticationMethod.SIMPLE, clientUgi.getAuthenticationMethod());
+    assertEquals(clientUsername, clientUgi.getUserName());
+
+    Configuration clientConf = new Configuration();
+    clientConf.set(User.HBASE_SECURITY_CONF_KEY, "simple");
+    callRpcService(rpcImplClass, User.create(clientUgi), clientConf, true);
+  }
 }