import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的package包/类
private void saslReadAndProcess(DataInputStream dis) throws
WrappedRpcServerException, IOException, InterruptedException {
  final RpcSaslProto saslMessage =
      decodeProtobufFromStream(RpcSaslProto.newBuilder(), dis);
  switch (saslMessage.getState()) {
    case WRAP: {
      if (!saslContextEstablished || !useWrap) {
        throw new WrappedRpcServerException(
            RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
            new SaslException("Server is not wrapping data"));
      }
      // loops over decoded data and calls processOneRpc
      unwrapPacketAndProcessRpcs(saslMessage.getToken().toByteArray());
      break;
    }
    default:
      saslProcess(saslMessage);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的package包/类
/**
 * Evaluate the server provided challenge.  The server must send a token
 * if it's not done.  If the server is done, the challenge token is
 * optional because not all mechanisms send a final token for the client to
 * update its internal state.  The client must also be done after
 * evaluating the optional token to ensure a malicious server doesn't
 * prematurely end the negotiation with a phony success.
 *  
 * @param saslResponse - client response to challenge
 * @param serverIsDone - server negotiation state
 * @throws SaslException - any problems with negotiation
 */
private byte[] saslEvaluateToken(RpcSaslProto saslResponse,
    boolean serverIsDone) throws SaslException {
  byte[] saslToken = null;
  if (saslResponse.hasToken()) {
    saslToken = saslResponse.getToken().toByteArray();
    saslToken = saslClient.evaluateChallenge(saslToken);
  } else if (!serverIsDone) {
    // the server may only omit a token when it's done
    throw new SaslException("Server challenge contains no token");
  }
  if (serverIsDone) {
    // server tried to report success before our client completed
    if (!saslClient.isComplete()) {
      throw new SaslException("Client is out of sync with server");
    }
    // a client cannot generate a response to a success message
    if (saslToken != null) {
      throw new SaslException("Client generated spurious response");        
    }
  }
  return saslToken;
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的package包/类
@Override
public void write(byte[] buf, int off, int len) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("wrapping token of length:" + len);
  }
  buf = saslClient.wrap(buf, off, len);
  RpcSaslProto saslMessage = RpcSaslProto.newBuilder()
      .setState(SaslState.WRAP)
      .setToken(ByteString.copyFrom(buf, 0, buf.length))
      .build();
  RpcRequestMessageWrapper request =
      new RpcRequestMessageWrapper(saslHeader, saslMessage);
  DataOutputStream dob = new DataOutputStream(out);
  dob.writeInt(request.getLength());
  request.write(dob);
 }
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的package包/类
/**
 * Process the Sasl's Negotiate request, including the optimization of 
 * accelerating token negotiation.
 * @return the response to Negotiate request - the list of enabled 
 *         authMethods and challenge if the TOKENS are supported. 
 * @throws SaslException - if attempt to generate challenge fails.
 * @throws IOException - if it fails to create the SASL server for Tokens
 */
private RpcSaslProto buildSaslNegotiateResponse()
    throws InterruptedException, SaslException, IOException {
  RpcSaslProto negotiateMessage = negotiateResponse;
  // accelerate token negotiation by sending initial challenge
  // in the negotiation response
  if (enabledAuthMethods.contains(AuthMethod.TOKEN)) {
    saslServer = createSaslServer(AuthMethod.TOKEN);
    byte[] challenge = saslServer.evaluateResponse(new byte[0]);
    RpcSaslProto.Builder negotiateBuilder =
        RpcSaslProto.newBuilder(negotiateResponse);
    negotiateBuilder.getAuthsBuilder(0)  // TOKEN is always first
        .setChallenge(ByteString.copyFrom(challenge));
    negotiateMessage = negotiateBuilder.build();
  }
  sentNegotiate = true;
  return negotiateMessage;
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的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; //导入依赖的package包/类
private RpcSaslProto buildSaslNegotiateResponse()
    throws IOException, InterruptedException {
  RpcSaslProto negotiateMessage = negotiateResponse;
  // accelerate token negotiation by sending initial challenge
  // in the negotiation response
  if (enabledAuthMethods.contains(AuthMethod.TOKEN)) {
    saslServer = createSaslServer(AuthMethod.TOKEN);
    byte[] challenge = saslServer.evaluateResponse(new byte[0]);
    RpcSaslProto.Builder negotiateBuilder =
        RpcSaslProto.newBuilder(negotiateResponse);
    negotiateBuilder.getAuthsBuilder(0)  // TOKEN is always first
        .setChallenge(ByteString.copyFrom(challenge));
    negotiateMessage = negotiateBuilder.build();
  }
  sentNegotiate = true;
  return negotiateMessage;
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的package包/类
private void saslReadAndProcess(RpcWritable.Buffer buffer) throws
    RpcServerException, IOException, InterruptedException {
  final RpcSaslProto saslMessage =
      getMessage(RpcSaslProto.getDefaultInstance(), buffer);
  switch (saslMessage.getState()) {
    case WRAP: {
      if (!saslContextEstablished || !useWrap) {
        throw new FatalRpcServerException(
            RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,
            new SaslException("Server is not wrapping data"));
      }
      // loops over decoded data and calls processOneRpc
      unwrapPacketAndProcessRpcs(saslMessage.getToken().toByteArray());
      break;
    }
    default:
      saslProcess(saslMessage);
  }
}
 

import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcSaslProto; //导入依赖的package包/类
private void wrapWithSasl(RpcCall call) throws IOException {
  if (call.connection.saslServer != null) {
    byte[] token = call.rpcResponse.array();
    // synchronization may be needed since there can be multiple Handler
    // threads using saslServer to wrap responses.
    synchronized (call.connection.saslServer) {
      token = call.connection.saslServer.wrap(token, 0, token.length);
    }
    if (LOG.isDebugEnabled())
      LOG.debug("Adding saslServer wrapped token of size " + token.length
          + " as call response.");
    // rebuild with sasl header and payload
    RpcResponseHeaderProto saslHeader = RpcResponseHeaderProto.newBuilder()
        .setCallId(AuthProtocol.SASL.callId)
        .setStatus(RpcStatusProto.SUCCESS)
        .build();
    RpcSaslProto saslMessage = RpcSaslProto.newBuilder()
        .setState(SaslState.WRAP)
        .setToken(ByteString.copyFrom(token))
        .build();
    setupResponse(call, saslHeader, RpcWritable.wrap(saslMessage));
  }
}