HDDS-15168. Support configurable SASL mechanism

hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/Server.java

@@ -97,7 +97,9 @@
 import org.apache.hadoop.ipc_.protobuf.RpcHeaderProtos.RpcSaslProto.SaslState;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security_.SaslMechanismFactory;
 import org.apache.hadoop.security.SaslPropertiesResolver;
+import org.apache.hadoop.security_.SaslRpcClient;
 import org.apache.hadoop.security_.SaslRpcServer;
 import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
 import org.apache.hadoop.security.SecurityUtil;
@@ -1916,6 +1918,10 @@ public Server getServer() {
       return Server.this;
     }
 
+    public Configuration getConf() {
+      return Server.this.getConf();
+    }
+
     /* Return true if the connection has no outstanding rpc */
     private boolean isIdle() {
       return rpcCount.get() == 0;
@@ -2383,7 +2389,8 @@ private RpcSaslProto buildSaslNegotiateResponse()
       RpcSaslProto negotiateMessage = negotiateResponse;
       // accelerate token negotiation by sending initial challenge
       // in the negotiation response
-      if (enabledAuthMethods.contains(AuthMethod.TOKEN)) {
+      if (enabledAuthMethods.contains(AuthMethod.TOKEN)
+          && SaslMechanismFactory.isDigestMechanism(AuthMethod.TOKEN)) {
         saslServer = createSaslServer(AuthMethod.TOKEN);
         byte[] challenge = saslServer.evaluateResponse(new byte[0]);
         RpcSaslProto.Builder negotiateBuilder =

hadoop-hdds/common/src/main/java/org/apache/hadoop/security_/CustomizedCallbackHandler.java

@@ -0,0 +1,121 @@
+/*
+ * 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.security_;
+
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** For handling customized {@link Callback}. */
+public interface CustomizedCallbackHandler {
+  Logger LOG = LoggerFactory.getLogger(CustomizedCallbackHandler.class);
+
+  class Cache {
+    private static final Map<String, CustomizedCallbackHandler> MAP = new HashMap<>();
+
+    private static synchronized CustomizedCallbackHandler getSynchronously(
+        String key, Configuration conf) {
+      //check again synchronously
+      final CustomizedCallbackHandler cached = MAP.get(key);
+      if (cached != null) {
+        return cached; //cache hit
+      }
+
+      //cache miss
+      final Class<?> clazz = conf.getClass(key, DefaultHandler.class);
+      LOG.debug("{} = {}", key, clazz);
+      if (clazz == DefaultHandler.class) {
+        return DefaultHandler.INSTANCE;
+      }
+
+      final Object created;
+      try {
+        created = clazz.newInstance();
+      } catch (Exception e) {
+        LOG.warn("Failed to create a new instance of {}, fallback to {}",
+            clazz, DefaultHandler.class, e);
+        return DefaultHandler.INSTANCE;
+      }
+
+      final CustomizedCallbackHandler handler = created instanceof CustomizedCallbackHandler ?
+          (CustomizedCallbackHandler) created : CustomizedCallbackHandler.delegate(created);
+      MAP.put(key, handler);
+      return handler;
+    }
+
+    private static CustomizedCallbackHandler get(String key, Configuration conf) {
+      final CustomizedCallbackHandler cached = MAP.get(key);
+      return cached != null ? cached : getSynchronously(key, conf);
+    }
+
+    public static synchronized void clear() {
+      MAP.clear();
+    }
+
+    private Cache() { }
+  }
+
+  class DefaultHandler implements CustomizedCallbackHandler {
+    private static final DefaultHandler INSTANCE = new DefaultHandler();
+
+    @Override
+    public void handleCallbacks(List<Callback> callbacks, String username, char[] password)
+        throws UnsupportedCallbackException {
+      if (!callbacks.isEmpty()) {
+        final Callback cb = callbacks.get(0);
+        throw new UnsupportedCallbackException(callbacks.get(0),
+            "Unsupported callback: " + (cb == null ? null : cb.getClass()));
+      }
+    }
+  }
+
+  static CustomizedCallbackHandler delegate(Object delegated) {
+    final String methodName = "handleCallbacks";
+    final Class<?> clazz = delegated.getClass();
+    final Method method;
+    try {
+      method = clazz.getMethod(methodName, List.class, String.class, char[].class);
+    } catch (NoSuchMethodException e) {
+      throw new IllegalStateException("Failed to get method " + methodName + " from " + clazz, e);
+    }
+
+    return (callbacks, name, password) -> {
+      try {
+        method.invoke(delegated, callbacks, name, password);
+      } catch (IllegalAccessException | InvocationTargetException e) {
+        throw new IOException("Failed to invoke " + method, e);
+      }
+    };
+  }
+
+  static CustomizedCallbackHandler get(String key, Configuration conf) {
+    return Cache.get(key, conf);
+  }
+
+  void handleCallbacks(List<Callback> callbacks, String name, char[] password)
+      throws UnsupportedCallbackException, IOException;
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/security_/SaslMechanismFactory.java

@@ -0,0 +1,90 @@
+/*
+ * 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.security_;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SASL related constants.
+ */
+public final class SaslMechanismFactory {
+  static final Logger LOG = LoggerFactory.getLogger(SaslMechanismFactory.class);
+
+  public static final String HADOOP_SECURITY_SASL_MECHANISM_KEY
+      = "hadoop.security.sasl.mechanism";
+  public static final String HADOOP_SECURITY_SASL_MECHANISM_DEFAULT
+      = "DIGEST-MD5";
+  public static final String HADOOP_SECURITY_SASL_CUSTOMIZEDCALLBACKHANDLER_CLASS_KEY
+      = "hadoop.security.sasl.CustomizedCallbackHandler.class";
+
+  private static final String SASL_MECHANISM_ENV = "HADOOP_SASL_MECHANISM";
+  private static volatile String mechanism;
+
+  private static synchronized String getSynchronously() {
+    // env
+    final String envValue = System.getenv(SASL_MECHANISM_ENV);
+    LOG.debug("{} = {} (env)", SASL_MECHANISM_ENV, envValue);
+
+    // conf
+    final Configuration conf = new Configuration();
+    final String confValue = conf.get(HADOOP_SECURITY_SASL_MECHANISM_KEY,
+        HADOOP_SECURITY_SASL_MECHANISM_DEFAULT);
+    LOG.debug("{} = {} (conf)", HADOOP_SECURITY_SASL_MECHANISM_KEY, confValue);
+
+    mechanism = envValue != null ? envValue
+        : confValue != null ? confValue
+        : HADOOP_SECURITY_SASL_MECHANISM_DEFAULT;
+    LOG.debug("SASL_MECHANISM = {} (effective)", mechanism);
+    return mechanism;
+  }
+
+  public static String getMechanism() {
+    final String value = mechanism;
+    return value != null ? value : getSynchronously();
+  }
+
+  public static boolean isDefaultMechanism(AuthMethod authMethod) {
+    return HADOOP_SECURITY_SASL_MECHANISM_DEFAULT.equals(getMechanismName(authMethod));
+  }
+
+  public static boolean isDigestMechanism(AuthMethod authMethod) {
+    return getMechanismName(authMethod).startsWith("DIGEST-");
+  }
+
+  private SaslMechanismFactory() {}
+
+  public static void main(String[] args) {
+    System.out.println("SASL_MECHANISM = " + getMechanism());
+  }
+
+  /** Helper to get actual mechanism name from config.  Required because {@code AuthMethod} is from Hadoop,
+   * not forked (because it is used in UGI, etc.). */
+  public static String getMechanismName(AuthMethod authMethod) {
+    switch (authMethod) {
+    case DIGEST:
+    case TOKEN:
+      return getMechanism();
+    default:
+      return authMethod.getMechanismName();
+
+    }
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/security_/SaslRpcClient.java

@@ -39,6 +39,7 @@
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.RealmCallback;
 import javax.security.sasl.RealmChoiceCallback;
 import javax.security.sasl.Sasl;
@@ -185,7 +186,7 @@ private boolean isValidAuthType(SaslAuth authType) {
     }
     // do we know what it is?  is it using our mechanism?
     return authMethod != null &&
-           authMethod.getMechanismName().equals(authType.getMechanism());
+           SaslMechanismFactory.getMechanismName(authMethod).equals(authType.getMechanism());
   }
 
   /**
@@ -242,7 +243,7 @@ private SaslClient createSaslClient(SaslAuth authType)
         throw new IOException("Unknown authentication method " + method);
     }
 
-    String mechanism = method.getMechanismName();
+    String mechanism = SaslMechanismFactory.getMechanismName(method);
     if (LOG.isDebugEnabled()) {
       LOG.debug("Creating SASL " + mechanism + "(" + method + ") "
           + " client to authenticate to service at " + saslServerName);
@@ -664,9 +665,17 @@ public void handle(Callback[] callbacks)
           pc = (PasswordCallback) callback;
         } else if (callback instanceof RealmCallback) {
           rc = (RealmCallback) callback;
+        } else if (callback instanceof AuthorizeCallback) {
+          final AuthorizeCallback ac = (AuthorizeCallback) callback;
+          final String authId = ac.getAuthenticationID();
+          final String authzId = ac.getAuthorizationID();
+          ac.setAuthorized(authId.equals(authzId));
+          if (ac.isAuthorized()) {
+            ac.setAuthorizedID(authzId);
+          }
         } else {
           throw new UnsupportedCallbackException(callback,
-              "Unrecognized SASL client callback");
+              "Unrecognized SASL client callback " + callback.getClass());
         }
       }
       if (nc != null) {
@@ -712,4 +721,5 @@ public static String getHostName(String name) {
       }
     }
   }
+
 }