import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
/**
 * Try to locate the required token for the server.
 * 
 * @param authType of the SASL client
 * @return Token for server, or null if no token available
 * @throws IOException - token selector cannot be instantiated
 */
private Token<?> getServerToken(SaslAuth authType) throws IOException {
  TokenInfo tokenInfo = SecurityUtil.getTokenInfo(protocol, conf);
  LOG.debug("Get token info proto:" + protocol + " info:" + tokenInfo);
  if (tokenInfo == null) { // protocol has no support for tokens
    return null;
  }
  TokenSelector<?> tokenSelector = null;
  try {
    tokenSelector = tokenInfo.value().newInstance();
  } catch (InstantiationException | IllegalAccessException e) {
    throw new IOException(e.toString(), e);
  }
  return tokenSelector.selectToken(
      SecurityUtil.buildTokenService(serverAddr), ugi.getTokens());
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
private RpcSaslProto buildNegotiateResponse(List<AuthMethod> authMethods)
    throws IOException {
  RpcSaslProto.Builder negotiateBuilder = RpcSaslProto.newBuilder();
  if (authMethods.contains(AuthMethod.SIMPLE) && authMethods.size() == 1) {
    // SIMPLE-only servers return success in response to negotiate
    negotiateBuilder.setState(SaslState.SUCCESS);
  } else {
    negotiateBuilder.setState(SaslState.NEGOTIATE);
    for (AuthMethod authMethod : authMethods) {
      SaslRpcServer saslRpcServer = new SaslRpcServer(authMethod);      
      SaslAuth.Builder builder = negotiateBuilder.addAuthsBuilder()
          .setMethod(authMethod.toString())
          .setMechanism(saslRpcServer.mechanism);
      if (saslRpcServer.protocol != null) {
        builder.setProtocol(saslRpcServer.protocol);
      }
      if (saslRpcServer.serverId != null) {
        builder.setServerId(saslRpcServer.serverId);
      }
    }
  }
  return negotiateBuilder.build();
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
private boolean isValidAuthType(SaslAuth authType) {
  AuthMethod authMethod;
  try {
    authMethod = AuthMethod.valueOf(authType.getMethod());
  } catch (IllegalArgumentException iae) { // unknown auth
    authMethod = null;
  }
  // do we know what it is?  is it using our mechanism?
  return authMethod != null &&
         authMethod.getMechanismName().equals(authType.getMechanism());
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
/**
 * Try to create a SaslClient for an authentication type.  May return
 * null if the type isn't supported or the client lacks the required
 * credentials.
 * 
 * @param authType - the requested authentication method
 * @return SaslClient for the authType or null
 * @throws SaslException - error instantiating client
 * @throws IOException - misc errors
 */
private SaslClient createSaslClient(SaslAuth authType)
    throws SaslException, IOException {
  String saslUser = null;
  // SASL requires the client and server to use the same proto and serverId
  // if necessary, auth types below will verify they are valid
  final String saslProtocol = authType.getProtocol();
  final String saslServerName = authType.getServerId();
  Map<String, String> saslProperties =
    saslPropsResolver.getClientProperties(serverAddr.getAddress());  
  CallbackHandler saslCallback = null;
  
  final AuthMethod method = AuthMethod.valueOf(authType.getMethod());
  switch (method) {
    case TOKEN: {
      Token<?> token = getServerToken(authType);
      if (token == null) {
        LOG.debug("tokens aren't supported for this protocol" +
            " or user doesn't have one");
        return null;
      }
      saslCallback = new SaslClientCallbackHandler(token);
      break;
    }
    case KERBEROS: {
      if (ugi.getRealAuthenticationMethod().getAuthMethod() !=
          AuthMethod.KERBEROS) {
        LOG.debug("client isn't using kerberos");
        return null;
      }
      String serverPrincipal = getServerPrincipal(authType);
      if (serverPrincipal == null) {
        LOG.debug("protocol doesn't use kerberos");
        return null;
      }
      if (LOG.isDebugEnabled()) {
        LOG.debug("RPC Server's Kerberos principal name for protocol="
            + protocol.getCanonicalName() + " is " + serverPrincipal);
      }
      break;
    }
    default:
      throw new IOException("Unknown authentication method " + method);
  }

  String mechanism = method.getMechanismName();
  if (LOG.isDebugEnabled()) {
    LOG.debug("Creating SASL " + mechanism + "(" + method + ") "
        + " client to authenticate to service at " + saslServerName);
  }
  return Sasl.createSaslClient(
      new String[] { mechanism }, saslUser, saslProtocol, saslServerName,
      saslProperties, saslCallback);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
/**
 * Process a saslMessge.
 * @param saslMessage received SASL message
 * @return the sasl response to send back to client
 * @throws SaslException if authentication or generating response fails, 
 *                       or SASL protocol mixup
 * @throws IOException if a SaslServer cannot be created
 * @throws AccessControlException if the requested authentication type 
 *         is not supported or trying to re-attempt negotiation.
 * @throws InterruptedException
 */
private RpcSaslProto processSaslMessage(RpcSaslProto saslMessage)
    throws SaslException, IOException, AccessControlException,
    InterruptedException {
  final RpcSaslProto saslResponse;
  final SaslState state = saslMessage.getState(); // required      
  switch (state) {
    case NEGOTIATE: {
      if (sentNegotiate) {
        // FIXME shouldn't this be SaslException?
        throw new AccessControlException(
            "Client already attempted negotiation");
      }
      saslResponse = buildSaslNegotiateResponse();
      // simple-only server negotiate response is success which client
      // interprets as switch to simple
      if (saslResponse.getState() == SaslState.SUCCESS) {
        switchToSimple();
      }
      break;
    }
    case INITIATE: {
      if (saslMessage.getAuthsCount() != 1) {
        throw new SaslException("Client mechanism is malformed");
      }
      // verify the client requested an advertised authType
      SaslAuth clientSaslAuth = saslMessage.getAuths(0);
      if (!negotiateResponse.getAuthsList().contains(clientSaslAuth)) {
        if (sentNegotiate) {
          throw new AccessControlException(
              clientSaslAuth.getMethod() + " authentication is not enabled."
                  + "  Available:" + enabledAuthMethods);
        }
        saslResponse = buildSaslNegotiateResponse();
        break;
      }
      authMethod = AuthMethod.valueOf(clientSaslAuth.getMethod());
      // abort SASL for SIMPLE auth, server has already ensured that
      // SIMPLE is a legit option above.  we will send no response
      if (authMethod == AuthMethod.SIMPLE) {
        switchToSimple();
        saslResponse = null;
        break;
      }
      // sasl server for tokens may already be instantiated
      if (saslServer == null || authMethod != AuthMethod.TOKEN) {
        saslServer = createSaslServer(authMethod);
      }
      saslResponse = processSaslToken(saslMessage);
      break;
    }
    case RESPONSE: {
      saslResponse = processSaslToken(saslMessage);
      break;
    }
    default:
      throw new SaslException("Client sent unsupported state " + state);
  }
  return saslResponse;
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
/**
 * Try to create a SaslClient for an authentication type.  May return
 * null if the type isn't supported or the client lacks the required
 * credentials.
 * 
 * @param authType - the requested authentication method
 * @return SaslClient for the authType or null
 * @throws SaslException - error instantiating client
 * @throws IOException - misc errors
 */
private SaslClient createSaslClient(SaslAuth authType)
    throws SaslException, IOException {
  String saslUser = null;
  // SASL requires the client and server to use the same proto and serverId
  // if necessary, auth types below will verify they are valid
  final String saslProtocol = authType.getProtocol();
  final String saslServerName = authType.getServerId();
  Map<String, String> saslProperties =
    saslPropsResolver.getClientProperties(serverAddr.getAddress());  
  CallbackHandler saslCallback = null;
  
  final AuthMethod method = AuthMethod.valueOf(authType.getMethod());
  switch (method) {
    case TOKEN: {
      Token<?> token = getServerToken(authType);
      if (token == null) {
        return null; // tokens aren't supported or user doesn't have one
      }
      saslCallback = new SaslClientCallbackHandler(token);
      break;
    }
    case KERBEROS: {
      if (ugi.getRealAuthenticationMethod().getAuthMethod() !=
          AuthMethod.KERBEROS) {
        return null; // client isn't using kerberos
      }
      String serverPrincipal = getServerPrincipal(authType);
      if (serverPrincipal == null) {
        return null; // protocol doesn't use kerberos
      }
      if (LOG.isDebugEnabled()) {
        LOG.debug("RPC Server's Kerberos principal name for protocol="
            + protocol.getCanonicalName() + " is " + serverPrincipal);
      }
      break;
    }
    default:
      throw new IOException("Unknown authentication method " + method);
  }
  
  String mechanism = method.getMechanismName();
  if (LOG.isDebugEnabled()) {
    LOG.debug("Creating SASL " + mechanism + "(" + method + ") "
        + " client to authenticate to service at " + saslServerName);
  }
  return Sasl.createSaslClient(
      new String[] { mechanism }, saslUser, saslProtocol, saslServerName,
      saslProperties, saslCallback);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
private RpcSaslProto processSaslMessage(RpcSaslProto saslMessage)
    throws IOException, InterruptedException {
  final RpcSaslProto saslResponse;
  final SaslState state = saslMessage.getState(); // required      
  switch (state) {
    case NEGOTIATE: {
      if (sentNegotiate) {
        throw new AccessControlException(
            "Client already attempted negotiation");
      }
      saslResponse = buildSaslNegotiateResponse();
      // simple-only server negotiate response is success which client
      // interprets as switch to simple
      if (saslResponse.getState() == SaslState.SUCCESS) {
        switchToSimple();
      }
      break;
    }
    case INITIATE: {
      if (saslMessage.getAuthsCount() != 1) {
        throw new SaslException("Client mechanism is malformed");
      }
      // verify the client requested an advertised authType
      SaslAuth clientSaslAuth = saslMessage.getAuths(0);
      if (!negotiateResponse.getAuthsList().contains(clientSaslAuth)) {
        if (sentNegotiate) {
          throw new AccessControlException(
              clientSaslAuth.getMethod() + " authentication is not enabled."
                  + "  Available:" + enabledAuthMethods);
        }
        saslResponse = buildSaslNegotiateResponse();
        break;
      }
      authMethod = AuthMethod.valueOf(clientSaslAuth.getMethod());
      // abort SASL for SIMPLE auth, server has already ensured that
      // SIMPLE is a legit option above.  we will send no response
      if (authMethod == AuthMethod.SIMPLE) {
        switchToSimple();
        saslResponse = null;
        break;
      }
      // sasl server for tokens may already be instantiated
      if (saslServer == null || authMethod != AuthMethod.TOKEN) {
        saslServer = createSaslServer(authMethod);
      }
      saslResponse = processSaslToken(saslMessage);
      break;
    }
    case RESPONSE: {
      saslResponse = processSaslToken(saslMessage);
      break;
    }
    default:
      throw new SaslException("Client sent unsupported state " + state);
  }
  return saslResponse;
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
/**
 * Try to create a SaslClient for an authentication type.  May return
 * null if the type isn't supported or the client lacks the required
 * credentials.
 * 
 * @param authType - the requested authentication method
 * @return SaslClient for the authType or null
 * @throws SaslException - error instantiating client
 * @throws IOException - misc errors
 */
private SaslClient createSaslClient(SaslAuth authType)
    throws SaslException, IOException {
  String saslUser = null;
  // SASL requires the client and server to use the same proto and serverId
  // if necessary, auth types below will verify they are valid
  final String saslProtocol = authType.getProtocol();
  final String saslServerName = authType.getServerId();
  Map<String, String> saslProperties = SaslRpcServer.SASL_PROPS;
  CallbackHandler saslCallback = null;
  
  final AuthMethod method = AuthMethod.valueOf(authType.getMethod());
  switch (method) {
    case TOKEN: {
      Token<?> token = getServerToken(authType);
      if (token == null) {
        return null; // tokens aren't supported or user doesn't have one
      }
      saslCallback = new SaslClientCallbackHandler(token);
      break;
    }
    case KERBEROS: {
      if (ugi.getRealAuthenticationMethod().getAuthMethod() !=
          AuthMethod.KERBEROS) {
        return null; // client isn't using kerberos
      }
      String serverPrincipal = getServerPrincipal(authType);
      if (serverPrincipal == null) {
        return null; // protocol doesn't use kerberos
      }
      if (LOG.isDebugEnabled()) {
        LOG.debug("RPC Server's Kerberos principal name for protocol="
            + protocol.getCanonicalName() + " is " + serverPrincipal);
      }
      break;
    }
    default:
      throw new IOException("Unknown authentication method " + method);
  }
  
  String mechanism = method.getMechanismName();
  if (LOG.isDebugEnabled()) {
    LOG.debug("Creating SASL " + mechanism + "(" + method + ") "
        + " client to authenticate to service at " + saslServerName);
  }
  return Sasl.createSaslClient(
      new String[] { mechanism }, saslUser, saslProtocol, saslServerName,
      saslProperties, saslCallback);
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto.SaslAuth; //导入依赖的package包/类
private RpcSaslProto processSaslMessage(RpcSaslProto saslMessage)
    throws IOException, InterruptedException {
  RpcSaslProto saslResponse = null;
  final SaslState state = saslMessage.getState(); // required      
  switch (state) {
    case NEGOTIATE: {
      if (sentNegotiate) {
        throw new AccessControlException(
            "Client already attempted negotiation");
      }
      saslResponse = buildSaslNegotiateResponse();
      // simple-only server negotiate response is success which client
      // interprets as switch to simple
      if (saslResponse.getState() == SaslState.SUCCESS) {
        switchToSimple();
      }
      break;
    }
    case INITIATE: {
      if (saslMessage.getAuthsCount() != 1) {
        throw new SaslException("Client mechanism is malformed");
      }
      // verify the client requested an advertised authType
      SaslAuth clientSaslAuth = saslMessage.getAuths(0);
      if (!negotiateResponse.getAuthsList().contains(clientSaslAuth)) {
        if (sentNegotiate) {
          throw new AccessControlException(
              clientSaslAuth.getMethod() + " authentication is not enabled."
                  + "  Available:" + enabledAuthMethods);
        }
        saslResponse = buildSaslNegotiateResponse();
        break;
      }
      authMethod = AuthMethod.valueOf(clientSaslAuth.getMethod());
      // abort SASL for SIMPLE auth, server has already ensured that
      // SIMPLE is a legit option above.  we will send no response
      if (authMethod == AuthMethod.SIMPLE) {
        switchToSimple();
        break;
      }
      // sasl server for tokens may already be instantiated
      if (saslServer == null || authMethod != AuthMethod.TOKEN) {
        saslServer = createSaslServer(authMethod);
      }
      // fallthru to process sasl token
    }
    case RESPONSE: {
      if (!saslMessage.hasToken()) {
        throw new SaslException("Client did not send a token");
      }
      byte[] saslToken = saslMessage.getToken().toByteArray();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Have read input token of size " + saslToken.length
            + " for processing by saslServer.evaluateResponse()");
      }
      saslToken = saslServer.evaluateResponse(saslToken);
      saslResponse = buildSaslResponse(
          saslServer.isComplete() ? SaslState.SUCCESS : SaslState.CHALLENGE,
          saslToken);
      break;
    }
    default:
      throw new SaslException("Client sent unsupported state " + state);
  }
  return saslResponse;
}