HBASE-11149 Wire encryption is broken (Devaraj Das)

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClient.java

@@ -50,6 +50,7 @@ import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.UserInformation;
 import org.apache.hadoop.hbase.protobuf.generated.TracingProtos.RPCTInfo;
 import org.apache.hadoop.hbase.security.AuthMethod;
 import org.apache.hadoop.hbase.security.HBaseSaslRpcClient;
+import org.apache.hadoop.hbase.security.SaslUtil.QualityOfProtection;
 import org.apache.hadoop.hbase.security.SecurityInfo;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.UserProvider;
@@ -804,7 +805,9 @@ public class RpcClient {
 
     private synchronized boolean setupSaslConnection(final InputStream in2,
         final OutputStream out2) throws IOException {
-      saslRpcClient = new HBaseSaslRpcClient(authMethod, token, serverPrincipal, fallbackAllowed);
+      saslRpcClient = new HBaseSaslRpcClient(authMethod, token, serverPrincipal, fallbackAllowed,
+          conf.get("hbase.rpc.protection", 
+              QualityOfProtection.AUTHENTICATION.name().toLowerCase()));
       return saslRpcClient.saslConnect(in2, out2);
     }
 

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

@@ -57,7 +57,6 @@ public class HBaseSaslRpcClient {
 
   private final SaslClient saslClient;
   private final boolean fallbackAllowed;
-
   /**
    * Create a HBaseSaslRpcClient for an authentication method
    * 
@@ -65,11 +64,37 @@ public class HBaseSaslRpcClient {
    *          the requested authentication method
    * @param token
    *          token to use if needed by the authentication method
+   * @param serverPrincipal
+   *          the server principal that we are trying to set the connection up to
+   * @param fallbackAllowed
+   *          does the client allow fallback to simple authentication
+   * @throws IOException
    */
   public HBaseSaslRpcClient(AuthMethod method,
       Token<? extends TokenIdentifier> token, String serverPrincipal, boolean fallbackAllowed)
       throws IOException {
+    this(method, token, serverPrincipal, fallbackAllowed, "authentication"); 
+  }
+  /**
+   * Create a HBaseSaslRpcClient for an authentication method
+   * 
+   * @param method
+   *          the requested authentication method
+   * @param token
+   *          token to use if needed by the authentication method
+   * @param serverPrincipal
+   *          the server principal that we are trying to set the connection up to
+   * @param fallbackAllowed
+   *          does the client allow fallback to simple authentication
+   * @param rpcProtection
+   *          the protection level ("authentication", "integrity" or "privacy")
+   * @throws IOException
+   */
+  public HBaseSaslRpcClient(AuthMethod method,
+      Token<? extends TokenIdentifier> token, String serverPrincipal, boolean fallbackAllowed,
+      String rpcProtection) throws IOException {
     this.fallbackAllowed = fallbackAllowed;
+    SaslUtil.initSaslProperties(rpcProtection);
     switch (method) {
     case DIGEST:
       if (LOG.isDebugEnabled())

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

@@ -23,12 +23,30 @@ import org.apache.commons.codec.binary.Base64;
 import java.util.Map;
 import java.util.TreeMap;
 
+import javax.security.sasl.Sasl;
+
 public class SaslUtil {
   public static final String SASL_DEFAULT_REALM = "default";
   public static final Map<String, String> SASL_PROPS =
       new TreeMap<String, String>();
   public static final int SWITCH_TO_SIMPLE_AUTH = -88;
 
+  public static enum QualityOfProtection {
+    AUTHENTICATION("auth"),
+    INTEGRITY("auth-int"),
+    PRIVACY("auth-conf");
+
+    public final String saslQop;
+
+    private QualityOfProtection(String saslQop) {
+      this.saslQop = saslQop;
+    }
+
+    public String getSaslQop() {
+      return saslQop;
+    }
+  }
+
   /** Splitting fully qualified Kerberos name into parts */
   public static String[] splitKerberosName(String fullName) {
     return fullName.split("[/@]");
@@ -45,4 +63,18 @@ public class SaslUtil {
   static char[] encodePassword(byte[] password) {
     return new String(Base64.encodeBase64(password)).toCharArray();
   }
+
+  static void initSaslProperties(String rpcProtection) {
+    QualityOfProtection saslQOP = QualityOfProtection.AUTHENTICATION;
+    if (QualityOfProtection.INTEGRITY.name().toLowerCase()
+        .equals(rpcProtection)) {
+      saslQOP = QualityOfProtection.INTEGRITY;
+    } else if (QualityOfProtection.PRIVACY.name().toLowerCase().equals(
+        rpcProtection)) {
+      saslQOP = QualityOfProtection.PRIVACY;
+    }
+
+    SaslUtil.SASL_PROPS.put(Sasl.QOP, saslQOP.getSaslQop());
+    SaslUtil.SASL_PROPS.put(Sasl.SERVER_AUTH, "true");
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/security/HBaseSaslRpcServer.java

@@ -29,12 +29,12 @@ import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.RealmCallback;
-import javax.security.sasl.Sasl;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ipc.RpcServer;
+import org.apache.hadoop.hbase.security.SaslUtil.QualityOfProtection;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.SecretManager;
 import org.apache.hadoop.security.token.TokenIdentifier;
@@ -46,36 +46,9 @@ import org.apache.hadoop.security.token.SecretManager.InvalidToken;
 public class HBaseSaslRpcServer {
   public static final Log LOG = LogFactory.getLog(HBaseSaslRpcServer.class);
 
-  public static enum QualityOfProtection {
-    AUTHENTICATION("auth"),
-    INTEGRITY("auth-int"),
-    PRIVACY("auth-conf");
-
-    public final String saslQop;
-
-    private QualityOfProtection(String saslQop) {
-      this.saslQop = saslQop;
-    }
-
-    public String getSaslQop() {
-      return saslQop;
-    }
-  }
-
   public static void init(Configuration conf) {
-    QualityOfProtection saslQOP = QualityOfProtection.AUTHENTICATION;
+    SaslUtil.initSaslProperties(conf.get("hbase.rpc.protection", 
-    String rpcProtection = conf.get("hbase.rpc.protection",
+          QualityOfProtection.AUTHENTICATION.name().toLowerCase()));
-        QualityOfProtection.AUTHENTICATION.name().toLowerCase());
-    if (QualityOfProtection.INTEGRITY.name().toLowerCase()
-        .equals(rpcProtection)) {
-      saslQOP = QualityOfProtection.INTEGRITY;
-    } else if (QualityOfProtection.PRIVACY.name().toLowerCase().equals(
-        rpcProtection)) {
-      saslQOP = QualityOfProtection.PRIVACY;
-    }
-
-    SaslUtil.SASL_PROPS.put(Sasl.QOP, saslQOP.getSaslQop());
-    SaslUtil.SASL_PROPS.put(Sasl.SERVER_AUTH, "true");
   }
 
   public static <T extends TokenIdentifier> T getIdentifier(String id,

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

@@ -38,6 +38,7 @@ import javax.security.auth.callback.NameCallback;
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.TextOutputCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.Sasl;
 import javax.security.sasl.RealmCallback;
 import javax.security.sasl.RealmChoiceCallback;
 import javax.security.sasl.SaslClient;
@@ -75,6 +76,23 @@ public class TestHBaseSaslRpcClient {
     Logger.getRootLogger().setLevel(Level.DEBUG);
   }
 
+  @Test
+  public void testSaslQOPNotEmpty() throws Exception {
+    // default QOP is authentication
+    new HBaseSaslRpcClient(AuthMethod.KERBEROS, createTokenMock(), "principal/host@DOMAIN.COM", false);
+    assertTrue(SaslUtil.SASL_PROPS.get(Sasl.QOP).equals(SaslUtil.QualityOfProtection.AUTHENTICATION.getSaslQop()));
+
+    // check with specific QOPs
+    new HBaseSaslRpcClient(AuthMethod.KERBEROS, createTokenMock(), "principal/host@DOMAIN.COM", false, "authentication");
+    assertTrue(SaslUtil.SASL_PROPS.get(Sasl.QOP).equals(SaslUtil.QualityOfProtection.AUTHENTICATION.getSaslQop()));
+
+    new HBaseSaslRpcClient(AuthMethod.KERBEROS, createTokenMock(), "principal/host@DOMAIN.COM", false, "privacy");
+    assertTrue(SaslUtil.SASL_PROPS.get(Sasl.QOP).equals(SaslUtil.QualityOfProtection.PRIVACY.getSaslQop()));
+
+    new HBaseSaslRpcClient(AuthMethod.KERBEROS, createTokenMock(), "principal/host@DOMAIN.COM", false, "integrity");
+    assertTrue(SaslUtil.SASL_PROPS.get(Sasl.QOP).equals(SaslUtil.QualityOfProtection.INTEGRITY.getSaslQop()));
+  }
+
   @Test
   public void testSaslClientCallbackHandler() throws UnsupportedCallbackException {
     final Token<? extends TokenIdentifier> token = createTokenMock();