Text-protocol authentication

src/main/java/net/spy/memcached/DefaultConnectionFactory.java

@@ -214,20 +214,21 @@ public MemcachedNode createMemcachedNode(SocketAddress sa, SocketChannel c,
       int bufSize) {
 
     OperationFactory of = getOperationFactory();
+    boolean doAuth = false;
+    if (getAuthDescriptor() != null) {
+      doAuth = true;
+    }
     if (of instanceof AsciiOperationFactory) {
       return new AsciiMemcachedNodeImpl(sa, c, bufSize,
           createReadOperationQueue(),
           createWriteOperationQueue(),
           createOperationQueue(),
           getOpQueueMaxBlockTime(),
+          doAuth,
           getOperationTimeout(),
           getAuthWaitTime(),
           this);
     } else if (of instanceof BinaryOperationFactory) {
-      boolean doAuth = false;
-      if (getAuthDescriptor() != null) {
-        doAuth = true;
-      }
       return new BinaryMemcachedNodeImpl(sa, c, bufSize,
           createReadOperationQueue(),
           createWriteOperationQueue(),

src/main/java/net/spy/memcached/auth/AsciiAuthThread.java

@@ -0,0 +1,127 @@
+/**
+ * Copyright (C) 2006-2009 Dustin Sallings
+ * Copyright (C) 2009-2014 Couchbase, Inc.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALING
+ * IN THE SOFTWARE.
+ */
+
+package net.spy.memcached.auth;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+
+import net.spy.memcached.MemcachedConnection;
+import net.spy.memcached.MemcachedNode;
+import net.spy.memcached.OperationFactory;
+import net.spy.memcached.compat.SpyThread;
+import net.spy.memcached.compat.log.Level;
+import net.spy.memcached.ops.Operation;
+import net.spy.memcached.ops.OperationStatus;
+import net.spy.memcached.ops.StoreOperation;
+import net.spy.memcached.ops.StoreType;
+
+/**
+ * A thread that does text authentication.
+ */
+public class AsciiAuthThread extends SpyThread {
+
+  /**
+   * If the total AUTH steps take longer than this period in milliseconds, a
+   * warning will be issued instead of a debug message.
+   */
+  public static final int AUTH_TOTAL_THRESHOLD = 250;
+
+  private final MemcachedConnection conn;
+  private final AuthDescriptor authDescriptor;
+  private final OperationFactory opFact;
+  private final MemcachedNode node;
+
+  public AsciiAuthThread(MemcachedConnection c, OperationFactory o,
+                         AuthDescriptor a, MemcachedNode n) {
+    conn = c;
+    opFact = o;
+    authDescriptor = a;
+    node = n;
+    start();
+  }
+
+  @Override
+  public void run() {
+    final AtomicBoolean done = new AtomicBoolean();
+    long start = System.nanoTime();
+      final CountDownLatch latch = new CountDownLatch(1);
+      NameCallback nameCallback = new NameCallback("memcached");
+      PasswordCallback passwordCallback = new PasswordCallback("memcached", false);
+      try {
+        authDescriptor.getCallback().handle(new Callback[]{nameCallback, passwordCallback});
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+      String credentials = nameCallback.getName() + ' ' + new String(passwordCallback.getPassword());
+      Operation op = opFact.store(StoreType.set, "ignore", 0, 0, credentials.getBytes(StandardCharsets.US_ASCII),
+            new StoreOperation.Callback() {
+              @Override
+              public void receivedStatus(OperationStatus val) {
+                if (val.isSuccess()) {
+                  done.set(true);
+                  node.authComplete();
+                }
+              }
+
+              @Override
+              public void gotData(String key, long cas) {
+              }
+
+              @Override
+              public void complete() {
+                latch.countDown();
+              }
+            });
+      conn.insertOperation(node, op);
+
+      try {
+        if (!conn.isShutDown()) {
+          latch.await();
+        } else {
+          done.set(true); // Connection is shutting down, tear.down.
+        }
+        Thread.sleep(100);
+      } catch (InterruptedException e) {
+        // we can be interrupted if we were in the
+        // process of auth'ing and the connection is
+        // lost or dropped due to bad auth
+        Thread.currentThread().interrupt();
+        if (op != null) {
+          op.cancel();
+        }
+        done.set(true); // If we were interrupted, tear down.
+      } finally {
+        long diff = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+        Level level = diff >= AUTH_TOTAL_THRESHOLD ? Level.WARN : Level.DEBUG;
+        getLogger().log(level, String.format("ASCII authentication step took %dms on %s",
+              diff, node.toString()));
+      }
+    }
+}

src/main/java/net/spy/memcached/auth/AuthThreadMonitor.java

@@ -29,17 +29,18 @@
 import net.spy.memcached.MemcachedNode;
 import net.spy.memcached.OperationFactory;
 import net.spy.memcached.compat.SpyObject;
+import net.spy.memcached.protocol.binary.BinaryOperationFactory;
 
 /**
  * This will ensure no more than one AuthThread will exist for a given
  * MemcachedNode.
  */
 public class AuthThreadMonitor extends SpyObject {
 
-  private final Map<Object, AuthThread> nodeMap;
+  private final Map<Object, Thread> nodeMap;
 
   public AuthThreadMonitor() {
-    nodeMap = new HashMap<Object, AuthThread>();
+    nodeMap = new HashMap<Object, Thread>();
   }
 
   /**
@@ -58,20 +59,21 @@ public synchronized void authConnection(MemcachedConnection conn,
       OperationFactory opFact, AuthDescriptor authDescriptor,
       MemcachedNode node) {
     interruptOldAuth(node);
-    AuthThread newSASLAuthenticator =
-        new AuthThread(conn, opFact, authDescriptor, node);
-    nodeMap.put(node, newSASLAuthenticator);
+    Thread authenticator = opFact instanceof BinaryOperationFactory ?
+          new BinaryAuthThread(conn, opFact, authDescriptor, node) :
+          new AsciiAuthThread(conn, opFact, authDescriptor, node);
+    nodeMap.put(node, authenticator);
   }
 
   /**
-   * Interrupt all pending {@link AuthThread}s.
+   * Interrupt all pending {@link BinaryAuthThread}s.
    *
-   * While shutting down a connection, if there are any {@link AuthThread}s
+   * While shutting down a connection, if there are any {@link BinaryAuthThread}s
    * running, terminate them so that the java process can exit gracefully (
    * otherwise it will wait infinitely).
    */
   public synchronized void interruptAllPendingAuth(){
-    for (AuthThread toStop : nodeMap.values()) {
+    for (Thread toStop : nodeMap.values()) {
       if (toStop.isAlive()) {
         getLogger().warn("Connection shutdown in progress - interrupting "
           + "waiting authentication thread.");
@@ -81,7 +83,7 @@ public synchronized void interruptAllPendingAuth(){
   }
 
   private void interruptOldAuth(MemcachedNode nodeToStop) {
-    AuthThread toStop = nodeMap.get(nodeToStop);
+    Thread toStop = nodeMap.get(nodeToStop);
     if (toStop != null) {
       if (toStop.isAlive()) {
         getLogger().warn(
@@ -99,7 +101,7 @@ private void interruptOldAuth(MemcachedNode nodeToStop) {
    * from anywhere else.
    * @return
    */
-  protected Map<Object, AuthThread> getNodeMap() {
+  protected Map<Object, Thread> getNodeMap() {
     return nodeMap;
   }
 }

src/main/java/net/spy/memcached/auth/AuthThread.java → src/main/java/net/spy/memcached/auth/BinaryAuthThread.java

@@ -41,7 +41,7 @@
 /**
  * A thread that does SASL authentication.
  */
-public class AuthThread extends SpyThread {
+public class BinaryAuthThread extends SpyThread {
 
   /**
    * If a SASL step takes longer than this period in milliseconds, a warning
@@ -62,8 +62,8 @@ public class AuthThread extends SpyThread {
   private final OperationFactory opFact;
   private final MemcachedNode node;
 
-  public AuthThread(MemcachedConnection c, OperationFactory o,
-      AuthDescriptor a, MemcachedNode n) {
+  public BinaryAuthThread(MemcachedConnection c, OperationFactory o,
+                          AuthDescriptor a, MemcachedNode n) {
     conn = c;
     opFact = o;
     authDescriptor = a;

src/main/java/net/spy/memcached/protocol/ascii/AsciiMemcachedNodeImpl.java

@@ -41,10 +41,10 @@ public final class AsciiMemcachedNodeImpl extends TCPMemcachedNodeImpl {
 
   public AsciiMemcachedNodeImpl(SocketAddress sa, SocketChannel c, int bufSize,
       BlockingQueue<Operation> rq, BlockingQueue<Operation> wq,
-      BlockingQueue<Operation> iq, Long opQueueMaxBlockTimeNs, long dt,
-      long at, ConnectionFactory fa) {
+      BlockingQueue<Operation> iq, Long opQueueMaxBlockTimeNs,
+      boolean waitForAuth, long dt, long at, ConnectionFactory fa) {
     // ASCII never does auth
-    super(sa, c, bufSize, rq, wq, iq, opQueueMaxBlockTimeNs, false, dt, at, fa);
+    super(sa, c, bufSize, rq, wq, iq, opQueueMaxBlockTimeNs, waitForAuth, dt, at, fa);
   }
 
   @Override

src/test/manual/net/spy/memcached/test/AsciiAuthTest.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright (C) 2006-2009 Dustin Sallings
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALING
+ * IN THE SOFTWARE.
+ */
+
+package net.spy.memcached.test;
+
+import net.spy.memcached.AddrUtil;
+import net.spy.memcached.ConnectionFactoryBuilder;
+import net.spy.memcached.ConnectionFactoryBuilder.Protocol;
+import net.spy.memcached.MemcachedClient;
+import net.spy.memcached.auth.AuthDescriptor;
+import net.spy.memcached.compat.SpyObject;
+
+/**
+ * Authentication functional test.
+ */
+public class AsciiAuthTest extends SpyObject implements Runnable {
+
+  private final String username;
+  private final String password;
+  private MemcachedClient client;
+
+  public AsciiAuthTest(String u, String p) {
+    username = u;
+    password = p;
+  }
+
+  public void init() throws Exception {
+    client = new MemcachedClient(new ConnectionFactoryBuilder()
+        .setProtocol(Protocol.TEXT)
+        .setAuthDescriptor(AuthDescriptor.typical(username, password))
+        .build(), AddrUtil.getAddresses("localhost:11212"));
+  }
+
+  public void shutdown() throws Exception {
+    client.shutdown();
+  }
+
+  public void run() {
+    client.getVersions();
+  }
+
+  public static void main(String[] a) throws Exception {
+    AsciiAuthTest lt = new AsciiAuthTest("testuser", "testpass");
+    lt.init();
+    long start = System.currentTimeMillis();
+    try {
+      lt.run();
+    } finally {
+      lt.shutdown();
+    }
+    long end = System.currentTimeMillis();
+    System.out.println("Runtime:  " + (end - start) + "ms");
+  }
+}

src/test/manual/net/spy/memcached/test/AuthTest.java → src/test/manual/net/spy/memcached/test/BinaryAuthTest.java

@@ -32,13 +32,13 @@
 /**
  * Authentication functional test.
  */
-public class AuthTest extends SpyObject implements Runnable {
+public class BinaryAuthTest extends SpyObject implements Runnable {
 
   private final String username;
   private final String password;
   private MemcachedClient client;
 
-  public AuthTest(String u, String p) {
+  public BinaryAuthTest(String u, String p) {
     username = u;
     password = p;
   }
@@ -65,7 +65,7 @@ public void run() {
   }
 
   public static void main(String[] a) throws Exception {
-    AuthTest lt = new AuthTest("testuser", "testpass");
+    BinaryAuthTest lt = new BinaryAuthTest("testuser", "testpass");
     lt.init();
     long start = System.currentTimeMillis();
     try {