import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Creates a SaslPropertiesResolver from the given configuration.  This method
 * works by cloning the configuration, translating configuration properties
 * specific to DataTransferProtocol to what SaslPropertiesResolver expects,
 * and then delegating to SaslPropertiesResolver for initialization.  This
 * method returns null if SASL protection has not been configured for
 * DataTransferProtocol.
 *
 * @param conf configuration to read
 * @return SaslPropertiesResolver for DataTransferProtocol, or null if not
 *   configured
 */
public static SaslPropertiesResolver getSaslPropertiesResolver(
    Configuration conf) {
  String qops = conf.get(DFS_DATA_TRANSFER_PROTECTION_KEY);
  if (qops == null || qops.isEmpty()) {
    LOG.debug("DataTransferProtocol not using SaslPropertiesResolver, no " +
      "QOP found in configuration for {}", DFS_DATA_TRANSFER_PROTECTION_KEY);
    return null;
  }
  Configuration saslPropsResolverConf = new Configuration(conf);
  saslPropsResolverConf.set(HADOOP_RPC_PROTECTION, qops);
  Class<? extends SaslPropertiesResolver> resolverClass = conf.getClass(
    HADOOP_SECURITY_SASL_PROPS_RESOLVER_CLASS,
    SaslPropertiesResolver.class, SaslPropertiesResolver.class);
  resolverClass = conf.getClass(DFS_DATA_TRANSFER_SASL_PROPS_RESOLVER_CLASS_KEY,
    resolverClass, SaslPropertiesResolver.class);
  saslPropsResolverConf.setClass(HADOOP_SECURITY_SASL_PROPS_RESOLVER_CLASS,
    resolverClass, SaslPropertiesResolver.class);
  SaslPropertiesResolver resolver = SaslPropertiesResolver.getInstance(
    saslPropsResolverConf);
  LOG.debug("DataTransferProtocol using SaslPropertiesResolver, configured " +
    "QOP {} = {}, configured class {} = {}", DFS_DATA_TRANSFER_PROTECTION_KEY, qops, 
    DFS_DATA_TRANSFER_SASL_PROPS_RESOLVER_CLASS_KEY, resolverClass);
  return resolver;
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Receives SASL negotiation for general-purpose handshake.
 *
 * @param peer connection peer
 * @param underlyingOut connection output stream
 * @param underlyingIn connection input stream
 * @return new pair of streams, wrapped after SASL negotiation
 * @throws IOException for any error
 */
private IOStreamPair getSaslStreams(Peer peer, OutputStream underlyingOut,
    InputStream underlyingIn) throws IOException {
  if (peer.hasSecureChannel() ||
      dnConf.getTrustedChannelResolver().isTrusted(getPeerAddress(peer))) {
    return new IOStreamPair(underlyingIn, underlyingOut);
  }

  SaslPropertiesResolver saslPropsResolver = dnConf.getSaslPropsResolver();
  Map<String, String> saslProps = saslPropsResolver.getServerProperties(
    getPeerAddress(peer));

  CallbackHandler callbackHandler = new SaslServerCallbackHandler(
    new PasswordFunction() {
      @Override
      public char[] apply(String userName) throws IOException {
        return buildServerPassword(userName);
      }
  });
  return doSaslHandshake(underlyingOut, underlyingIn, saslProps,
      callbackHandler);
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Checks if the DataNode has a secure configuration if security is enabled.
 * There are 2 possible configurations that are considered secure:
 * 1. The server has bound to privileged ports for RPC and HTTP via
 *   SecureDataNodeStarter.
 * 2. The configuration enables SASL on DataTransferProtocol and HTTPS (no
 *   plain HTTP) for the HTTP server.  The SASL handshake guarantees
 *   authentication of the RPC server before a client transmits a secret, such
 *   as a block access token.  Similarly, SSL guarantees authentication of the
 *   HTTP server before a client transmits a secret, such as a delegation
 *   token.
 * It is not possible to run with both privileged ports and SASL on
 * DataTransferProtocol.  For backwards-compatibility, the connection logic
 * must check if the target port is a privileged port, and if so, skip the
 * SASL handshake.
 *
 * @param dnConf DNConf to check
 * @param conf Configuration to check
 * @param resources SecuredResources obtained for DataNode
 * @throws RuntimeException if security enabled, but configuration is insecure
 */
private static void checkSecureConfig(DNConf dnConf, Configuration conf,
    SecureResources resources) throws RuntimeException {
  if (!UserGroupInformation.isSecurityEnabled()) {
    return;
  }
  SaslPropertiesResolver saslPropsResolver = dnConf.getSaslPropsResolver();
  if (resources != null && saslPropsResolver == null) {
    return;
  }
  if (dnConf.getIgnoreSecurePortsForTesting()) {
    return;
  }
  if (saslPropsResolver != null &&
      DFSUtil.getHttpPolicy(conf) == HttpConfig.Policy.HTTPS_ONLY &&
      resources == null) {
    return;
  }
  throw new RuntimeException("Cannot start secure DataNode without " +
    "configuring either privileged resources or SASL RPC data transfer " +
    "protection and SSL for HTTP.  Using privileged resources in " +
    "combination with SASL RPC data transfer protection is not supported.");
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Creates a SaslPropertiesResolver from the given configuration.  This method
 * works by cloning the configuration, translating configuration properties
 * specific to DataTransferProtocol to what SaslPropertiesResolver expects,
 * and then delegating to SaslPropertiesResolver for initialization.  This
 * method returns null if SASL protection has not been configured for
 * DataTransferProtocol.
 *
 * @param conf configuration to read
 * @return SaslPropertiesResolver for DataTransferProtocol, or null if not
 *   configured
 */
public static SaslPropertiesResolver getSaslPropertiesResolver(
    Configuration conf) {
  String qops = conf.get(DFS_DATA_TRANSFER_PROTECTION_KEY);
  if (qops == null || qops.isEmpty()) {
    LOG.debug("DataTransferProtocol not using SaslPropertiesResolver, no " +
        "QOP found in configuration for {}",
        DFS_DATA_TRANSFER_PROTECTION_KEY);
    return null;
  }
  Configuration saslPropsResolverConf = new Configuration(conf);
  saslPropsResolverConf.set(HADOOP_RPC_PROTECTION, qops);
  Class<? extends SaslPropertiesResolver> resolverClass = conf.getClass(
      HADOOP_SECURITY_SASL_PROPS_RESOLVER_CLASS,
      SaslPropertiesResolver.class, SaslPropertiesResolver.class);
  resolverClass =
      conf.getClass(DFS_DATA_TRANSFER_SASL_PROPS_RESOLVER_CLASS_KEY,
      resolverClass, SaslPropertiesResolver.class);
  saslPropsResolverConf.setClass(HADOOP_SECURITY_SASL_PROPS_RESOLVER_CLASS,
      resolverClass, SaslPropertiesResolver.class);
  SaslPropertiesResolver resolver = SaslPropertiesResolver.getInstance(
      saslPropsResolverConf);
  LOG.debug("DataTransferProtocol using SaslPropertiesResolver, configured " +
          "QOP {} = {}, configured class {} = {}",
      DFS_DATA_TRANSFER_PROTECTION_KEY, qops,
      DFS_DATA_TRANSFER_SASL_PROPS_RESOLVER_CLASS_KEY, resolverClass);
  return resolver;
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Checks if the DataNode has a secure configuration if security is enabled.
 * There are 2 possible configurations that are considered secure:
 * 1. The server has bound to privileged ports for RPC and HTTP via
 *   SecureDataNodeStarter.
 * 2. The configuration enables SASL on DataTransferProtocol and HTTPS (no
 *   plain HTTP) for the HTTP server.  The SASL handshake guarantees
 *   authentication of the RPC server before a client transmits a secret, such
 *   as a block access token.  Similarly, SSL guarantees authentication of the
 *   HTTP server before a client transmits a secret, such as a delegation
 *   token.
 * It is not possible to run with both privileged ports and SASL on
 * DataTransferProtocol.  For backwards-compatibility, the connection logic
 * must check if the target port is a privileged port, and if so, skip the
 * SASL handshake.
 *
 * @param dnConf DNConf to check
 * @param conf Configuration to check
 * @param resources SecuredResources obtained for DataNode
 * @throws RuntimeException if security enabled, but configuration is insecure
 */
private static void checkSecureConfig(DNConf dnConf, Configuration conf,
    SecureResources resources) throws RuntimeException {
  if (!UserGroupInformation.isSecurityEnabled()) {
    return;
  }

  // Abort out of inconsistent state if Kerberos is enabled
  // but block access tokens are not enabled.
  boolean isEnabled = conf.getBoolean(
      DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY,
      DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_DEFAULT);
  if (!isEnabled) {
    String errMessage = "Security is enabled but block access tokens " +
        "(via " + DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY + ") " +
        "aren't enabled. This may cause issues " +
        "when clients attempt to connect to a DataNode. Aborting DataNode";
    throw new RuntimeException(errMessage);
  }

  SaslPropertiesResolver saslPropsResolver = dnConf.getSaslPropsResolver();
  if (resources != null && saslPropsResolver == null) {
    return;
  }
  if (dnConf.getIgnoreSecurePortsForTesting()) {
    return;
  }
  if (saslPropsResolver != null &&
      DFSUtil.getHttpPolicy(conf) == HttpConfig.Policy.HTTPS_ONLY &&
      resources == null) {
    return;
  }
  throw new RuntimeException("Cannot start secure DataNode without " +
    "configuring either privileged resources or SASL RPC data transfer " +
    "protection and SSL for HTTP.  Using privileged resources in " +
    "combination with SASL RPC data transfer protection is not supported.");
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
static void trySaslNegotiate(Configuration conf, Channel channel, DatanodeInfo dnInfo,
    int timeoutMs, DFSClient client, Token<BlockTokenIdentifier> accessToken,
    Promise<Void> saslPromise) throws IOException {
  SaslDataTransferClient saslClient = client.getSaslDataTransferClient();
  SaslPropertiesResolver saslPropsResolver = SASL_ADAPTOR.getSaslPropsResolver(saslClient);
  TrustedChannelResolver trustedChannelResolver =
      SASL_ADAPTOR.getTrustedChannelResolver(saslClient);
  AtomicBoolean fallbackToSimpleAuth = SASL_ADAPTOR.getFallbackToSimpleAuth(saslClient);
  InetAddress addr = ((InetSocketAddress) channel.remoteAddress()).getAddress();
  if (trustedChannelResolver.isTrusted() || trustedChannelResolver.isTrusted(addr)) {
    saslPromise.trySuccess(null);
    return;
  }
  DataEncryptionKey encryptionKey = client.newDataEncryptionKey();
  if (encryptionKey != null) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(
        "SASL client doing encrypted handshake for addr = " + addr + ", datanodeId = " + dnInfo);
    }
    doSaslNegotiation(conf, channel, timeoutMs, getUserNameFromEncryptionKey(encryptionKey),
      encryptionKeyToPassword(encryptionKey.encryptionKey),
      createSaslPropertiesForEncryption(encryptionKey.encryptionAlgorithm), saslPromise);
  } else if (!UserGroupInformation.isSecurityEnabled()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("SASL client skipping handshake in unsecured configuration for addr = " + addr
          + ", datanodeId = " + dnInfo);
    }
    saslPromise.trySuccess(null);
  } else if (dnInfo.getXferPort() < 1024) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("SASL client skipping handshake in secured configuration with "
          + "privileged port for addr = " + addr + ", datanodeId = " + dnInfo);
    }
    saslPromise.trySuccess(null);
  } else if (fallbackToSimpleAuth != null && fallbackToSimpleAuth.get()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("SASL client skipping handshake in secured configuration with "
          + "unsecured cluster for addr = " + addr + ", datanodeId = " + dnInfo);
    }
    saslPromise.trySuccess(null);
  } else if (saslPropsResolver != null) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(
        "SASL client doing general handshake for addr = " + addr + ", datanodeId = " + dnInfo);
    }
    doSaslNegotiation(conf, channel, timeoutMs, buildUsername(accessToken),
      buildClientPassword(accessToken), saslPropsResolver.getClientProperties(addr), saslPromise);
  } else {
    // It's a secured cluster using non-privileged ports, but no SASL. The only way this can
    // happen is if the DataNode has ignore.secure.ports.for.testing configured, so this is a rare
    // edge case.
    if (LOG.isDebugEnabled()) {
      LOG.debug("SASL client skipping handshake in secured configuration with no SASL "
          + "protection configured for addr = " + addr + ", datanodeId = " + dnInfo);
    }
    saslPromise.trySuccess(null);
  }
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Creates a new SaslDataTransferClient.
 *
 * @param conf the configuration
 * @param saslPropsResolver for determining properties of SASL negotiation
 * @param trustedChannelResolver for identifying trusted connections that do
 *   not require SASL negotiation
 * @param fallbackToSimpleAuth checked on each attempt at general SASL
 *   handshake, if true forces use of simple auth
 */
public SaslDataTransferClient(Configuration conf, 
    SaslPropertiesResolver saslPropsResolver,
    TrustedChannelResolver trustedChannelResolver,
    AtomicBoolean fallbackToSimpleAuth) {
  this.conf = conf;
  this.fallbackToSimpleAuth = fallbackToSimpleAuth;
  this.saslPropsResolver = saslPropsResolver;
  this.trustedChannelResolver = trustedChannelResolver;
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Creates a new SaslDataTransferClient.
 *
 * @param conf the configuration
 * @param saslPropsResolver for determining properties of SASL negotiation
 * @param trustedChannelResolver for identifying trusted connections that do
 *   not require SASL negotiation
 * @param fallbackToSimpleAuth checked on each attempt at general SASL
 *   handshake, if true forces use of simple auth
 */
public SaslDataTransferClient(Configuration conf,
    SaslPropertiesResolver saslPropsResolver,
    TrustedChannelResolver trustedChannelResolver,
    AtomicBoolean fallbackToSimpleAuth) {
  this.conf = conf;
  this.fallbackToSimpleAuth = fallbackToSimpleAuth;
  this.saslPropsResolver = saslPropsResolver;
  this.trustedChannelResolver = trustedChannelResolver;
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
/**
 * Returns the SaslPropertiesResolver configured for use with
 * DataTransferProtocol, or null if not configured.
 *
 * @return SaslPropertiesResolver configured for use with DataTransferProtocol
 */
public SaslPropertiesResolver getSaslPropsResolver() {
  return saslPropsResolver;
}
 

import org.apache.hadoop.security.SaslPropertiesResolver; //导入依赖的package包/类
SaslPropertiesResolver getSaslPropsResolver(SaslDataTransferClient saslClient);