DRILL-5943: Avoid the strong check introduced by DRILL-5582 for PLAIN mechanism

This closes #1028

6 files changed, 75 insertions, 31 deletions

diff --git a/contrib/native/client/src/clientlib/drillClientImpl.cpp b/contrib/native/client/src/clientlib/drillClientImpl.cpp
index 343556185..60ae7a409 100644
--- a/contrib/native/client/src/clientlib/drillClientImpl.cpp
+++ b/contrib/native/client/src/clientlib/drillClientImpl.cpp
@@ -520,17 +520,23 @@ bool DrillClientImpl::clientNeedsEncryption(const DrillUserProperties* userPrope
 
 /*
  * Checks if the client has explicitly expressed interest in authenticated connections only.
- * If the USERPROP_PASSWORD or USERPROP_AUTH_MECHANISM connection string properties are
- * non-empty, then it is implied that the client wants authentication.
+ * If the USERPROP_AUTH_MECHANISM connection string properties is non-empty and not equal to PLAIN,
+ * then it is implied that the client wants authentication.
+ *
+ * Explicitly skipping PLAIN to maintain forward compatibility with 1.9 Drillbit and it doesn't matter
+ * if this security check is not there for PLAIN mechanism
  */
 bool DrillClientImpl::clientNeedsAuthentication(const DrillUserProperties* userProperties) {
     bool needsAuthentication = false;
-    if(!userProperties) { return false; }
-    std::string passwd = "";
-    userProperties->getProp(USERPROP_PASSWORD, passwd);
+    if(!userProperties) { return needsAuthentication; }
     std::string authMech = "";
     userProperties->getProp(USERPROP_AUTH_MECHANISM, authMech);
-    return !passwd.empty() || !authMech.empty();
+    boost::algorithm::to_lower(authMech);
+
+    if(!authMech.empty() && SaslAuthenticatorImpl::PLAIN_NAME != authMech) {
+        needsAuthentication = true;
+    }
+    return needsAuthentication;
 }
 
 connectionStatus_t DrillClientImpl::validateHandshake(DrillUserProperties* properties){
diff --git a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp
index c03cb6c0b..1f32b5e77 100644
--- a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp
+++ b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp
@@ -28,16 +28,16 @@
 
 namespace Drill {
 
-static const std::string DEFAULT_SERVICE_NAME = "drill";
+const std::string DEFAULT_SERVICE_NAME = "drill";
+const int PREFERRED_MIN_SSF = 56;
+const std::string SaslAuthenticatorImpl::KERBEROS_SIMPLE_NAME = "kerberos";
+const std::string SaslAuthenticatorImpl::PLAIN_NAME = "plain";
 
-static const std::string KERBEROS_SIMPLE_NAME = "kerberos";
-static const std::string KERBEROS_SASL_NAME = "gssapi";
-static const std::string PLAIN_NAME = "plain";
-static const int PREFERRED_MIN_SSF = 56;
+const std::string KERBEROS_SASL_NAME = "gssapi";
 
 const std::map<std::string, std::string> SaslAuthenticatorImpl::MECHANISM_MAPPING = boost::assign::map_list_of
-    (KERBEROS_SIMPLE_NAME, KERBEROS_SASL_NAME)
-    (PLAIN_NAME, PLAIN_NAME)
+    (SaslAuthenticatorImpl::KERBEROS_SIMPLE_NAME, KERBEROS_SASL_NAME)
+    (SaslAuthenticatorImpl::PLAIN_NAME, SaslAuthenticatorImpl::PLAIN_NAME)
 ;
 
 boost::mutex SaslAuthenticatorImpl::s_mutex;
@@ -138,15 +138,17 @@ int SaslAuthenticatorImpl::init(const std::vector<std::string>& mechanisms, exec
             m_ppwdSecret = (sasl_secret_t *) malloc(sizeof(sasl_secret_t) + length);
             std::memcpy(m_ppwdSecret->data, value.c_str(), length);
             m_ppwdSecret->len = length;
-            authMechanismToUse = PLAIN_NAME;
+            authMechanismToUse = SaslAuthenticatorImpl::PLAIN_NAME;
         } else if (USERPROP_USERNAME == key) {
             m_username = value;
         } else if (USERPROP_AUTH_MECHANISM == key) {
             authMechanismToUse = value;
         }
     }
-    // clientNeedsAuthentication() cannot be false if the code above picks an authMechanism
-    assert (authMechanismToUse.empty() || DrillClientImpl::clientNeedsAuthentication(m_pUserProperties));
+    // clientNeedsAuthentication() cannot be false if the code above picks an authMechanism other than PLAIN
+    assert (authMechanismToUse.empty() || authMechanismToUse == SaslAuthenticatorImpl::PLAIN_NAME ||
+            DrillClientImpl::clientNeedsAuthentication(m_pUserProperties));
+
     m_authMechanismName = authMechanismToUse;
     if (authMechanismToUse.empty()) return SASL_NOMECH;
 
diff --git a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp
index bf61e9dc8..a0ee5a5c9 100644
--- a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp
+++ b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp
@@ -59,6 +59,10 @@ public:
 
     const char *getErrorMessage(int errorCode);
 
+    static const std::string KERBEROS_SIMPLE_NAME;
+
+    static const std::string PLAIN_NAME;
+
 private:
 
     static const std::map<std::string, std::string> MECHANISM_MAPPING;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
index 093cc3dba..2c63cbd2b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java
@@ -251,7 +251,7 @@ public class UserClient
     }
 
     // Check if client needs authentication and server doesn't support any security mechanisms.
-    if (clientNeedsAuth(properties) && serverAuthMechanisms == null) {
+    if (clientNeedsAuthExceptPlain(properties) && serverAuthMechanisms == null) {
       throw new NonTransientRpcException(
           "Client needs authentication but server doesn't support any security mechanisms." +
               " Please contact your administrator. [Warn: It may be due to wrong config or a security attack in" +
@@ -262,19 +262,22 @@ public class UserClient
   /**
    * Determine if client needs authenticated connection or not. It checks for following as an indication of
    * requiring authentication from client side:
-   * 1) Any auth mechanism is provided in connection properties with DrillProperties.AUTH_MECHANISM parameter.
+   * 1) Auth mechanism except PLAIN is provided in connection properties with DrillProperties.AUTH_MECHANISM parameter.
    * 2) A service principal is provided in connection properties in which case we treat AUTH_MECHANISM as Kerberos
-   * 3) Username and Password is provided in connection properties in which case we treat AUTH_MECHANISM as Plain
    * @param props - DrillClient connection parameters
-   * @return - true  - If any of above 3 checks is successful.
-   *         - false - If all the above 3 checks failed.
+   * @return - true  - If any of above 2 checks is successful.
+   *         - false - If all the above 2 checks fails.
    */
-  private boolean clientNeedsAuth(DrillProperties props) {
+  private boolean clientNeedsAuthExceptPlain(DrillProperties props) {
 
-    return !Strings.isNullOrEmpty(props.getProperty(DrillProperties.AUTH_MECHANISM)) ||
-        !Strings.isNullOrEmpty(props.getProperty(DrillProperties.SERVICE_PRINCIPAL)) ||
-        (props.containsKey(DrillProperties.USER) && !Strings.isNullOrEmpty(props.getProperty(DrillProperties.PASSWORD)));
+    boolean clientNeedsAuth = false;
+    final String authMechanism =  props.getProperty(DrillProperties.AUTH_MECHANISM);
+    if (!Strings.isNullOrEmpty(authMechanism) && !authMechanism.equalsIgnoreCase(PlainFactory.SIMPLE_NAME)) {
+      clientNeedsAuth = true;
+    }
 
+    clientNeedsAuth |= !Strings.isNullOrEmpty(props.getProperty(DrillProperties.SERVICE_PRINCIPAL));
+    return clientNeedsAuth;
   }
 
   private CheckedFuture<Void, RpcException> connect(final UserToBitHandshake handshake,
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
index 9e00b83f3..2b50e5040 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserServer.java
@@ -321,13 +321,17 @@ public class UserServer extends BasicServer<RpcType, BitToUserConnection> {
             return respBuilder.build();
           }
 
-          final boolean clientSupportsSasl = inbound.hasSaslSupport() &&
-              (inbound.getSaslSupport().ordinal() > SaslSupport.UNKNOWN_SASL_SUPPORT.ordinal());
+          // If sasl_support field is absent in handshake message then treat the client as < 1.10 client
+          final boolean clientSupportsSasl = inbound.hasSaslSupport();
 
+          // saslSupportOrdinal will be set to UNKNOWN_SASL_SUPPORT, if sasl_support field in handshake is set to a
+          // value which is unknown to this server. We will treat those clients as one which knows SASL protocol.
           final int saslSupportOrdinal = (clientSupportsSasl) ? inbound.getSaslSupport().ordinal()
                                                               : SaslSupport.UNKNOWN_SASL_SUPPORT.ordinal();
 
-          if (saslSupportOrdinal <= SaslSupport.SASL_AUTH.ordinal() && config.isEncryptionEnabled()) {
+          // Check if client doesn't support SASL or only supports SASL_AUTH and server has encryption enabled
+          if ((!clientSupportsSasl || saslSupportOrdinal == SaslSupport.SASL_AUTH.ordinal())
+              && config.isEncryptionEnabled()) {
             throw new UserAuthenticationException("The server doesn't allow client without encryption support." +
                 " Please upgrade your client or talk to your system administrator.");
           }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/user/security/TestUserBitSaslCompatibility.java b/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/user/security/TestUserBitSaslCompatibility.java
index 1ca38f522..3f524b368 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/user/security/TestUserBitSaslCompatibility.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/rpc/user/security/TestUserBitSaslCompatibility.java
@@ -62,12 +62,12 @@ public class TestUserBitSaslCompatibility extends BaseTestQuery {
   }
 
   /**
-   * Test showing failure before SASL handshake when Drillbit is not configured for authentication whereas client
-   * explicitly requested for authentication.
+   * Test showing when Drillbit is not configured for authentication whereas client explicitly requested for PLAIN
+   * authentication then connection succeeds without authentication.
    * @throws Exception
    */
   @Test
-  public void testDisableDrillbitAuth_EnableClientAuth() throws Exception {
+  public void testDisableDrillbitAuth_EnableClientAuthPlain() throws Exception {
 
     final DrillConfig newConfig = new DrillConfig(DrillConfig.create(cloneDefaultTestConfigProperties())
         .withValue(ExecConstants.USER_AUTHENTICATION_ENABLED,
@@ -80,6 +80,29 @@ public class TestUserBitSaslCompatibility extends BaseTestQuery {
 
     try {
       updateTestCluster(1, newConfig, connectionProps);
+    } catch (Exception ex) {
+      fail();
+    }
+  }
+
+  /**
+   * Test showing when Drillbit is not configured for authentication whereas client explicitly requested for Kerberos
+   * authentication then connection fails due to new check before SASL Handshake.
+   * @throws Exception
+   */
+  @Test
+  public void testDisableDrillbitAuth_EnableClientAuthKerberos() throws Exception {
+
+    final DrillConfig newConfig = new DrillConfig(DrillConfig.create(cloneDefaultTestConfigProperties())
+        .withValue(ExecConstants.USER_AUTHENTICATION_ENABLED,
+            ConfigValueFactory.fromAnyRef(false)),
+        false);
+
+    final Properties connectionProps = new Properties();
+    connectionProps.setProperty(DrillProperties.AUTH_MECHANISM, "kerberos");
+
+    try {
+      updateTestCluster(1, newConfig, connectionProps);
       fail();
     } catch (Exception ex) {
       assertTrue(ex.getCause() instanceof NonTransientRpcException);
@@ -193,6 +216,8 @@ public class TestUserBitSaslCompatibility extends BaseTestQuery {
         false);
 
     final Properties connectionProps = new Properties();
+    connectionProps.setProperty(DrillProperties.USER, "anonymous");
+    connectionProps.setProperty(DrillProperties.PASSWORD, "anything works!");
 
     try {
       updateTestCluster(1, newConfig, connectionProps);