[security] Add per-connection nonce for Kerberos replay resistance

Kerberos is susceptible to replay attacks, which it attempts to mitigate
by using a server-side replay cache. The cache is not 100% effective,
and is extremely slow. This commit introduces an effective and efficient
method of mitigating replay attacks by using a server-generated nonce
which the client must send back to the server, wrapped in SASL integrity
protection. This will allow Kudu to disable the replay cache without
negatively affecting security.

No tests are provided, but the codepath is well covered by existing
Kerberos negotiation tests. I intend to write simulated mitm tests to
check this and the channel binding protections soon.

Change-Id: If0fb433896963be5e81d349ebf3a044a458e6627
Reviewed-on: http://gerrit.cloudera.org:8080/6137
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <[email protected]>
Reviewed-by: Todd Lipcon <[email protected]>

docs/design-docs/rpc.md

@@ -531,13 +531,18 @@ keytab. Kerberos authentication is handled through the SASL `GSSAPI` mechanism.
 When using Kerberos authentication TLS certificates are not verified.
 
 When the SASL `GSSAPI` negotiation is complete, the server returns a special
-channel binding token to the client as part of the `SASL_SUCCESS` message. The
-channel binding token contains a hash of the server's certificate, wrapped in a
-SASL integrity protected envelope. The client must check the channel binding
-token against the certificate presented by the server during the TLS handshake
-before continuing to use the connection. See RFC 5056 for more information on
-channel binding and why it is necessary, and RFC 5929 for a description of the
-specific 'tls-server-end-point' channel binding type used.
+channel binding token and a random nonce to the client as part of the
+`SASL_SUCCESS` message. The channel binding token contains a hash of the
+server's certificate, wrapped in a SASL integrity protected envelope. The client
+must check the channel binding token against the certificate presented by the
+server during the TLS handshake before continuing to use the connection. See RFC
+5056 for more information on channel binding and why it is necessary, and RFC
+5929 for a description of the specific 'tls-server-end-point' channel binding
+type used. The client must return the nonce, wrapped in a SASL integrity
+protected envelope, to the server as part of the connection context. The nonce
+protects the server against Kerberos replay attacks. Kerberos's built-in replay
+attack mitigation is extremely slow, so this allows much faster connection
+negotiation.
 
 #### Certificate Authentication
 

java/kudu-client/src/main/java/org/apache/kudu/client/Negotiator.java

@@ -28,6 +28,8 @@
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.security.PrivilegedExceptionAction;
 import java.security.cert.Certificate;
 import java.util.List;
@@ -149,6 +151,12 @@ private static enum State {
   /** True if we have negotiated TLS with the server */
   private boolean negotiatedTls;
 
+  /**
+   * The nonce sent from the server to the client, or null if negotiation has
+   * not yet taken place, or the server does not send a nonce.
+   */
+  private byte[] nonce;
+
   /**
    * Future indicating whether the embedded handshake has completed.
    * Only non-null once TLS is initiated.
@@ -338,9 +346,10 @@ private void chooseAndInitializeSaslMech(NegotiatePB response) throws SaslExcept
         continue;
       }
       Map<String, String> props = Maps.newHashMap();
-      // If we are using GSSAPI with TLS, enable integrity protection, which we use
-      // to securely transmit the channel bindings.
-      if ("GSSAPI".equals(clientMech) && negotiatedTls) {
+      // If the negotiated mechanism is GSSAPI (Kerberos), configure SASL to use
+      // integrity protection so that the channel bindings and nonce can be
+      // verified.
+      if ("GSSAPI".equals(clientMech)) {
         props.put(Sasl.QOP, "auth-int");
       }
       try {
@@ -536,7 +545,8 @@ private void sendTokenExchange(Channel chan) {
     sendSaslMessage(chan, builder.build());
   }
 
-  private void handleTokenExchangeResponse(Channel chan, NegotiatePB response) {
+  private void handleTokenExchangeResponse(Channel chan,
+                                           NegotiatePB response) throws SaslException {
     Preconditions.checkArgument(response.getStep() == NegotiateStep.TOKEN_EXCHANGE,
         "expected TOKEN_EXCHANGE, got step: {}", response.getStep());
 
@@ -595,11 +605,20 @@ private void verifyChannelBindings(NegotiatePB response) {
     }
   }
 
-  private void handleSuccessResponse(Channel chan, NegotiatePB response) {
+  private void handleSuccessResponse(Channel chan, NegotiatePB response) throws SaslException {
     Preconditions.checkState(saslClient.isComplete(),
                              "server sent SASL_SUCCESS step, but SASL negotiation is not complete");
-    if (peerCert != null && "GSSAPI".equals(chosenMech)) {
-      verifyChannelBindings(response);
+    if (chosenMech.equals("GSSAPI")) {
+      if (response.hasNonce()) {
+        // Grab the nonce from the server, if it has sent one. We'll send it back
+        // later with SASL integrity protection as part of the connection context.
+        nonce = response.getNonce().toByteArray();
+      }
+
+      if (peerCert != null) {
+        // Check the channel bindings provided by the server against the expected channel bindings.
+        verifyChannelBindings(response);
+      }
     }
 
     finish(chan);
@@ -609,15 +628,15 @@ private void handleSuccessResponse(Channel chan, NegotiatePB response) {
    * Marks the negotiation as finished, and sends the connection context to the server.
    * @param chan the connection channel
    */
-  private void finish(Channel chan) {
+  private void finish(Channel chan) throws SaslException {
     state = State.FINISHED;
     chan.getPipeline().remove(this);
 
     Channels.write(chan, makeConnectionContext());
     Channels.fireMessageReceived(chan, new Result(serverFeatures));
   }
 
-  private RpcOutboundMessage makeConnectionContext() {
+  private RpcOutboundMessage makeConnectionContext() throws SaslException {
     RpcHeader.ConnectionContextPB.Builder builder = RpcHeader.ConnectionContextPB.newBuilder();
 
     // The UserInformationPB is deprecated, but used by servers prior to Kudu 1.1.
@@ -626,6 +645,19 @@ private RpcOutboundMessage makeConnectionContext() {
     userBuilder.setEffectiveUser(user);
     userBuilder.setRealUser(user);
     builder.setDEPRECATEDUserInfo(userBuilder.build());
+
+    if (nonce != null) {
+      // Reply with the SASL-protected nonce. We only set the nonce when using SASL GSSAPI.
+      // The Java SASL client does not automatically add the length header,
+      // so we have to do it ourselves.
+      byte[] encodedNonce = saslClient.wrap(nonce, 0, nonce.length);
+      ByteBuffer buf = ByteBuffer.allocate(encodedNonce.length + 4);
+      buf.order(ByteOrder.BIG_ENDIAN);
+      buf.putInt(encodedNonce.length);
+      buf.put(encodedNonce);
+      builder.setEncodedNonce(ZeroCopyLiteralByteString.wrap(buf.array()));
+    }
+
     RpcHeader.ConnectionContextPB pb = builder.build();
     RpcHeader.RequestHeader.Builder header =
         RpcHeader.RequestHeader.newBuilder().setCallId(CONNECTION_CTX_CALL_ID);

src/kudu/rpc/client_negotiation.cc

@@ -19,7 +19,6 @@
 
 #include <string.h>
 
-#include <algorithm>
 #include <map>
 #include <memory>
 #include <set>
@@ -558,10 +557,10 @@ Status ClientNegotiation::SendSaslInitiate() {
   // Check that the SASL library is using the mechanism that we picked.
   DCHECK_EQ(SaslMechanism::value_of(negotiated_mech), negotiated_mech_);
 
-  // If we are speaking TLS and the negotiated mechanism is GSSAPI (Kerberos),
-  // configure SASL to use integrity protection so that the channel bindings
-  // can be verified.
-  if (tls_negotiated_ && negotiated_mech_ == SaslMechanism::GSSAPI) {
+  // If the negotiated mechanism is GSSAPI (Kerberos), configure SASL to use
+  // integrity protection so that the channel bindings and nonce can be
+  // verified.
+  if (negotiated_mech_ == SaslMechanism::GSSAPI) {
     RETURN_NOT_OK(EnableIntegrityProtection(sasl_conn_.get()));
   }
 
@@ -608,34 +607,41 @@ Status ClientNegotiation::HandleSaslSuccess(const NegotiatePB& response) {
   }
   TRACE("Received SASL_SUCCESS response from server");
 
-  if (tls_negotiated_ &&
-      negotiated_authn_ == AuthenticationType::SASL &&
-      negotiated_mech_ == SaslMechanism::GSSAPI) {
-    // Check the channel bindings provided by the server against the expected
-    // channel bindings.
-    security::Cert cert;
-    RETURN_NOT_OK(tls_handshake_.GetRemoteCert(&cert));
+  if (negotiated_mech_ == SaslMechanism::GSSAPI) {
+    if (response.has_nonce()) {
+      // Grab the nonce from the server, if it has sent one. We'll send it back
+      // later with SASL integrity protection as part of the connection context.
+      nonce_ = response.nonce();
+    }
+
+    if (tls_negotiated_) {
+      // Check the channel bindings provided by the server against the expected channel bindings.
+      if (!response.has_channel_bindings()) {
+        return Status::NotAuthorized("no channel bindings provided by server");
+      }
 
-    string expected_channel_bindings;
-    RETURN_NOT_OK_PREPEND(cert.GetServerEndPointChannelBindings(&expected_channel_bindings),
-                          "failed to generate channel bindings");
+      security::Cert cert;
+      RETURN_NOT_OK(tls_handshake_.GetRemoteCert(&cert));
 
-    if (!response.has_channel_bindings()) {
-      return Status::NotAuthorized("no channel bindings provided by server");
-    }
+      string expected_channel_bindings;
+      RETURN_NOT_OK_PREPEND(cert.GetServerEndPointChannelBindings(&expected_channel_bindings),
+                            "failed to generate channel bindings");
 
-    string recieved_channel_bindings;
-    RETURN_NOT_OK_PREPEND(SaslDecode(response.channel_bindings(), &recieved_channel_bindings),
-                          "failed to decode channel bindings");
+      string received_channel_bindings;
+      RETURN_NOT_OK_PREPEND(SaslDecode(sasl_conn_.get(),
+                                       response.channel_bindings(),
+                                       &received_channel_bindings),
+                            "failed to decode channel bindings");
 
-    if (expected_channel_bindings != recieved_channel_bindings) {
-      Sockaddr addr;
-      ignore_result(socket_->GetPeerAddress(&addr));
+      if (expected_channel_bindings != received_channel_bindings) {
+        Sockaddr addr;
+        ignore_result(socket_->GetPeerAddress(&addr));
 
-      LOG(WARNING) << "Recieved unexpected channel bindings from server "
-                   << addr.ToString()
-                   << ", this could indicate an active network man-in-the-middle";
-      return Status::NotAuthorized("channel bindings do not match");
+        LOG(WARNING) << "Received invalid channel bindings from server "
+                    << addr.ToString()
+                    << ", this could indicate an active network man-in-the-middle";
+        return Status::NotAuthorized("channel bindings do not match");
+      }
     }
   }
 
@@ -650,27 +656,6 @@ Status ClientNegotiation::DoSaslStep(const string& in, const char** out, unsigne
   });
 }
 
-Status ClientNegotiation::SaslDecode(const string& encoded, string* plaintext) {
-  size_t offset = 0;
-  const char* out;
-  unsigned out_len;
-
-  // The SASL library can only decode up to a maximum amount at a time, so we
-  // have to call decode multiple times if our input is larger than this max.
-  while (offset < encoded.size()) {
-    size_t len = std::min(kSaslMaxOutBufLen, encoded.size() - offset);
-
-    RETURN_NOT_OK(WrapSaslCall(sasl_conn_.get(), [&]() {
-        return sasl_decode(sasl_conn_.get(), encoded.data() + offset, len, &out, &out_len);
-    }));
-
-    plaintext->append(out, out_len);
-    offset += len;
-  }
-
-  return Status::OK();
-}
-
 Status ClientNegotiation::SendConnectionContext() {
   TRACE("Sending connection context");
   RequestHeader header;
@@ -681,6 +666,12 @@ Status ClientNegotiation::SendConnectionContext() {
   // this and use the SASL-provided username instead.
   conn_context.mutable_deprecated_user_info()->set_real_user(
       plain_auth_user_.empty() ? "cpp-client" : plain_auth_user_);
+
+  if (nonce_) {
+    // Reply with the SASL-protected nonce. We only set the nonce when using SASL GSSAPI.
+    RETURN_NOT_OK(SaslEncode(sasl_conn_.get(), *nonce_, conn_context.mutable_encoded_nonce()));
+  }
+
   return SendFramedMessageBlocking(socket(), header, conn_context, deadline_);
 }
 

src/kudu/rpc/client_negotiation.h

@@ -199,9 +199,6 @@ class ClientNegotiation {
   // otherwise returns an appropriate error status.
   Status DoSaslStep(const std::string& in, const char** out, unsigned* out_len) WARN_UNUSED_RESULT;
 
-  // Decode the provided SASL-encoded data and append it to 'plaintext'.
-  Status SaslDecode(const std::string& encoded, std::string* plaintext) WARN_UNUSED_RESULT;
-
   Status SendConnectionContext() WARN_UNUSED_RESULT;
 
   // The socket to the remote server.
@@ -211,6 +208,7 @@ class ClientNegotiation {
   std::vector<sasl_callback_t> callbacks_;
   std::unique_ptr<sasl_conn_t, SaslDeleter> sasl_conn_;
   SaslHelper helper_;
+  boost::optional<std::string> nonce_;
 
   // TLS state.
   const security::TlsContext* tls_context_;

src/kudu/rpc/negotiation-test.cc

@@ -32,6 +32,7 @@
 
 #include "kudu/gutil/gscoped_ptr.h"
 #include "kudu/gutil/map-util.h"
+#include "kudu/gutil/ref_counted.h"
 #include "kudu/gutil/strings/join.h"
 #include "kudu/gutil/strings/substitute.h"
 #include "kudu/gutil/walltime.h"
@@ -51,6 +52,7 @@
 #include "kudu/util/net/sockaddr.h"
 #include "kudu/util/net/socket.h"
 #include "kudu/util/subprocess.h"
+#include "kudu/util/trace.h"
 #include "kudu/util/user.h"
 
 // HACK: MIT Kerberos doesn't have any way of determining its version number,
@@ -68,6 +70,7 @@ DEFINE_bool(is_test_child, false,
             "Used by tests which require clean processes. "
             "See TestDisableInit.");
 DECLARE_bool(rpc_encrypt_loopback_connections);
+DECLARE_bool(rpc_trace_negotiation);
 
 using std::string;
 using std::thread;
@@ -261,9 +264,20 @@ TEST_P(TestNegotiation, TestNegotiation) {
       client_negotiation.socket()->Close();
   });
   thread server_thread([&] () {
+      scoped_refptr<Trace> t(new Trace());
+      ADOPT_TRACE(t.get());
       server_status = server_negotiation.Negotiate();
       // Close the socket so that the client will not block forever on error.
       server_negotiation.socket()->Close();
+
+      if (FLAGS_rpc_trace_negotiation || !server_status.ok()) {
+        string msg = Trace::CurrentTrace()->DumpToString();
+        if (!server_status.ok()) {
+          LOG(WARNING) << "Failed RPC negotiation. Trace:\n" << msg;
+        } else {
+          LOG(INFO) << "RPC negotiation tracing enabled. Trace:\n" << msg;
+        }
+      }
   });
   client_thread.join();
   server_thread.join();

src/kudu/rpc/rpc_header.proto

@@ -52,6 +52,12 @@ message ConnectionContextPB {
   // Impersonation (effective user) was never supported, so we'll have
   // to add that back at some point later.
   optional UserInformationPB DEPRECATED_user_info = 2;
+
+  // If the server sends a nonce to the client during the SASL_SUCCESS
+  // negotiation step, the client is required to encode it with SASL integrity
+  // protection and return it in this field. The nonce protects the server
+  // against a Kerberos replay attack.
+  optional bytes encoded_nonce = 3 [(REDACT) = true];
 }
 
 // Features supported by the RPC system itself.
@@ -162,6 +168,12 @@ message NegotiatePB {
   // value matches the expected value.
   optional bytes channel_bindings = 6 [(REDACT) = true];
 
+  // A random nonce sent from the server to the client during the SASL_SUCCESS
+  // step when the Kerberos (GSSAPI) SASL mechanism is used with TLS. The nonce
+  // must be sent back to the server, wrapped in SASL integrity protection, as
+  // part of the connection context.
+  optional bytes nonce = 9 [(REDACT) = true];
+
   // During the NEGOTIATE step, contains the supported SASL mechanisms.
   // During the SASL_INITIATE step, contains the single chosen SASL mechanism.
   repeated SaslMechanism sasl_mechanisms = 4;