Status ClientNegotiation::HandleNegotiate()

in src/kudu/rpc/client_negotiation.cc [351:469]
86 lines of code
26 McCabe index (conditional complexity)

Status ClientNegotiation::HandleNegotiate(const NegotiatePB& response) {
  if (PREDICT_FALSE(response.step() != NegotiatePB::NEGOTIATE)) {
    return Status::NotAuthorized("expected NEGOTIATE step",
                                 NegotiatePB::NegotiateStep_Name(response.step()));
  }
  TRACE("Received NEGOTIATE response from server");

  // Fill in the set of features supported by the server.
  for (int flag : response.supported_features()) {
    // We only add the features that our local build knows about.
    RpcFeatureFlag feature_flag = RpcFeatureFlag_IsValid(flag) ?
                                  static_cast<RpcFeatureFlag>(flag) : UNKNOWN;
    if (feature_flag != UNKNOWN) {
      server_features_.insert(feature_flag);
    }
  }

  if (encryption_ == RpcEncryption::REQUIRED &&
      !ContainsKey(server_features_, RpcFeatureFlag::TLS)) {
    return Status::NotAuthorized("server does not support required TLS encryption");
  }

  // Get the authentication type which the server would like to use.
  DCHECK_LE(response.authn_types().size(), 1);
  if (response.authn_types().empty()) {
    // If the server doesn't send back an authentication type, default to SASL
    // in order to maintain backwards compatibility.
    negotiated_authn_ = AuthenticationType::SASL;
  } else {
    const auto& authn_type = response.authn_types(0);
    switch (authn_type.type_case()) {
      case AuthenticationTypePB::kSasl:
        negotiated_authn_ = AuthenticationType::SASL;
        break;
      case AuthenticationTypePB::kToken:
        // TODO(todd): we should also be checking tls_context_->has_trusted_cert()
        // here to match the original logic we used to advertise TOKEN support,
        // or perhaps just check explicitly whether we advertised TOKEN.
        if (!authn_token_) {
          return Status::RuntimeError(
              "server chose token authentication, but client has no token");
        }
        negotiated_authn_ = AuthenticationType::TOKEN;
        return Status::OK();
      case AuthenticationTypePB::kCertificate:
        if (!tls_context_->has_signed_cert()) {
          return Status::RuntimeError(
              "server chose certificate authentication, but client has no certificate");
        }
        negotiated_authn_ = AuthenticationType::CERTIFICATE;
        return Status::OK();
      case AuthenticationTypePB::TYPE_NOT_SET:
        return Status::RuntimeError("server chose an unknown authentication type");
    }
  }

  DCHECK_EQ(negotiated_authn_, AuthenticationType::SASL);

  // Build a map of the SASL mechanisms offered by the server.
  set<SaslMechanism::Type> client_mechs(helper_.EnabledMechs());
  set<SaslMechanism::Type> server_mechs;
  for (const NegotiatePB::SaslMechanism& sasl_mech : response.sasl_mechanisms()) {
    auto mech = SaslMechanism::value_of(sasl_mech.mechanism());
    if (mech == SaslMechanism::INVALID) {
      continue;
    }
    server_mechs.insert(mech);
  }

  // Determine which SASL mechanism to use for authenticating the connection.
  // We pick the most preferred mechanism which is supported by both parties.
  // The preference list in order of most to least preferred:
  //  * GSSAPI
  //  * PLAIN
  //
  // TODO(KUDU-1921): allow the client to require authentication.
  if (ContainsKey(client_mechs, SaslMechanism::GSSAPI) &&
      ContainsKey(server_mechs, SaslMechanism::GSSAPI)) {

    // Check that the client has local Kerberos credentials, and if not fall
    // back to an alternate mechanism.
    Status s = CheckGSSAPI();
    if (s.ok()) {
      negotiated_mech_ = SaslMechanism::GSSAPI;
      return Status::OK();
    }

    TRACE("Kerberos authentication credentials are not available: $0", s.ToString());
    client_mechs.erase(SaslMechanism::GSSAPI);
  }

  if (ContainsKey(client_mechs, SaslMechanism::PLAIN) &&
      ContainsKey(server_mechs, SaslMechanism::PLAIN)) {
    negotiated_mech_ = SaslMechanism::PLAIN;
    return Status::OK();
  }

  // There are no mechanisms in common.
  if (ContainsKey(server_mechs, SaslMechanism::GSSAPI) &&
      !ContainsKey(client_mechs, SaslMechanism::GSSAPI)) {
    return Status::NotAuthorized("server requires authentication, "
                                  "but client does not have Kerberos credentials available");
  }
  if (!ContainsKey(server_mechs, SaslMechanism::GSSAPI) &&
      ContainsKey(client_mechs, SaslMechanism::GSSAPI)) {
    return Status::NotAuthorized("client requires authentication, "
                                  "but server does not have Kerberos enabled");
  }
  string msg = Substitute("client/server supported SASL mechanism mismatch; "
                          "client mechanisms: [$0], server mechanisms: [$1]",
                          JoinMapped(client_mechs, SaslMechanism::name_of, ", "),
                          JoinMapped(server_mechs, SaslMechanism::name_of, ", "));

  // For now, there should never be a SASL mechanism mismatch that isn't due
  // to one of the sides requiring Kerberos and the other not having it, so
  // lets sanity check that.
  DCHECK(STLSetIntersection(client_mechs, server_mechs).empty()) << msg;
  return Status::NotAuthorized(msg);
}