Fix race conditions in auth process

network/src/main/java/io/bitsquare/p2p/network/Connection.java

@@ -258,7 +258,7 @@ public class Connection implements MessageListener {
                     Thread.currentThread().setName("Connection:SendCloseConnectionMessage-" + this.uid);
                     Log.traceCall("sendCloseConnectionMessage");
                     try {
-                        sendMessage(new CloseConnectionMessage());
+                        sendMessage(new CloseConnectionMessage(peerAddressOptional));
                         setStopFlags();
 
                         Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
@@ -558,6 +558,8 @@ public class Connection implements MessageListener {
 
                         sharedSpace.updateLastActivityDate();
                         if (message instanceof CloseConnectionMessage) {
+                            log.info("Close connection message received from peer {}",
+                                    ((CloseConnectionMessage) message).peerAddressOptional);
                             stopped = true;
                             sharedSpace.shutDown(false);
                         } else if (!stopped) {

network/src/main/java/io/bitsquare/p2p/network/TorNetworkNode.java

@@ -92,7 +92,6 @@ public class TorNetworkNode extends NetworkNode {
     @Override
     @Nullable
     public Address getAddress() {
-        Log.traceCall();
         if (hiddenServiceDescriptor != null)
             return new Address(hiddenServiceDescriptor.getFullAddress());
         else

network/src/main/java/io/bitsquare/p2p/network/messages/CloseConnectionMessage.java

@@ -1,16 +1,32 @@
 package io.bitsquare.p2p.network.messages;
 
 import io.bitsquare.app.Version;
+import io.bitsquare.p2p.Address;
 import io.bitsquare.p2p.Message;
 
+import java.util.Optional;
+
 public final class CloseConnectionMessage implements Message {
     // That object is sent over the wire, so we need to take care of version compatibility.
     private static final long serialVersionUID = Version.NETWORK_PROTOCOL_VERSION;
 
     private final int networkId = Version.NETWORK_ID;
+    public Optional<Address> peerAddressOptional;
+
+    public CloseConnectionMessage(Optional<Address> peerAddressOptional) {
+        this.peerAddressOptional = peerAddressOptional;
+    }
 
     @Override
     public int networkId() {
         return networkId;
     }
+
+    @Override
+    public String toString() {
+        return "CloseConnectionMessage{" +
+                "peerAddressOptional=" + peerAddressOptional +
+                ", networkId=" + networkId +
+                '}';
+    }
 }

network/src/main/java/io/bitsquare/p2p/peers/AuthenticationHandshake.java

@@ -12,9 +12,9 @@ import io.bitsquare.p2p.network.ConnectionPriority;
 import io.bitsquare.p2p.network.MessageListener;
 import io.bitsquare.p2p.network.NetworkNode;
 import io.bitsquare.p2p.peers.messages.auth.AuthenticationChallenge;
+import io.bitsquare.p2p.peers.messages.auth.AuthenticationFinalResponse;
 import io.bitsquare.p2p.peers.messages.auth.AuthenticationMessage;
 import io.bitsquare.p2p.peers.messages.auth.AuthenticationRequest;
-import io.bitsquare.p2p.peers.messages.auth.AuthenticationResponse;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,9 +45,10 @@ public class AuthenticationHandshake implements MessageListener {
     private final BiConsumer<HashSet<ReportedPeer>, Connection> addReportedPeersConsumer;
 
     private final long startAuthTs;
-    private long nonce;
+    private long nonce = 0;
     private boolean stopped;
-    private Optional<SettableFuture<Connection>> resultFutureOptional;
+    private Optional<SettableFuture<Connection>> resultFutureOptional = Optional.empty();
+    private boolean ownRequestCanceled;
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -67,10 +68,6 @@ public class AuthenticationHandshake implements MessageListener {
         this.peerAddress = peerAddress;
 
         startAuthTs = System.currentTimeMillis();
-        stopped = false;
-        nonce = 0;
-        resultFutureOptional = Optional.empty();
-
         networkNode.addMessageListener(this);
     }
 
@@ -94,6 +91,11 @@ public class AuthenticationHandshake implements MessageListener {
                 Log.traceCall(message.toString());
                 if (message instanceof AuthenticationChallenge) {
                     // Requesting peer
+                    if (ownRequestCanceled) {
+                        log.info("Our own request has been canceled because of a race condition. " +
+                                "\nWe ignore that message and go on with the protocol from the other peers request. " +
+                                "\nThat might happen in rare cases.");
+                    } else {
                         AuthenticationChallenge authenticationChallenge = (AuthenticationChallenge) message;
                         // We need to set the address to the connection, otherwise we will not find the connection when sending
                         // the next message and we would create a new outbound connection instead using the inbound.
@@ -103,11 +105,11 @@ public class AuthenticationHandshake implements MessageListener {
                         log.trace("Received authenticationResponse from " + peerAddress);
                         boolean verified = nonce != 0 && nonce == authenticationChallenge.requesterNonce;
                         if (verified) {
-                        AuthenticationResponse authenticationResponse = new AuthenticationResponse(myAddress,
+                            AuthenticationFinalResponse authenticationFinalResponse = new AuthenticationFinalResponse(myAddress,
                                     authenticationChallenge.responderNonce,
                                     new HashSet<>(authenticatedAndReportedPeersSupplier.get()));
-                        SettableFuture<Connection> future = networkNode.sendMessage(peerAddress, authenticationResponse);
+                            SettableFuture<Connection> future = networkNode.sendMessage(peerAddress, authenticationFinalResponse);
-                        log.trace("Sent GetPeersAuthRequest {} to {}", authenticationResponse, peerAddress);
+                            log.trace("Sent GetPeersAuthRequest {} to {}", authenticationFinalResponse, peerAddress);
                             Futures.addCallback(future, new FutureCallback<Connection>() {
                                 @Override
                                 public void onSuccess(Connection connection) {
@@ -129,23 +131,24 @@ public class AuthenticationHandshake implements MessageListener {
                             // now we add the reported peers to our list 
                             addReportedPeersConsumer.accept(authenticationChallenge.reportedPeers, connection);
                         } else {
-                        log.warn("Verification of nonce failed. AuthenticationResponse=" + authenticationChallenge + " / nonce=" + nonce);
+                            log.warn("Verification of nonce failed. AuthenticationChallenge=" + authenticationChallenge + " / nonce=" + nonce);
-                        failed(new Exception("Verification of nonce failed. AuthenticationResponse=" + authenticationChallenge + " / nonceMap=" + nonce));
+                            failed(new Exception("Verification of nonce failed. AuthenticationChallenge=" + authenticationChallenge + " / nonceMap=" + nonce));
                         }
-                } else if (message instanceof AuthenticationResponse) {
+                    }
+                } else if (message instanceof AuthenticationFinalResponse) {
                     // Responding peer
-                    AuthenticationResponse authenticationResponse = (AuthenticationResponse) message;
+                    AuthenticationFinalResponse authenticationFinalResponse = (AuthenticationFinalResponse) message;
                     log.trace("Received GetPeersAuthRequest from " + peerAddress + " at " + myAddress);
-                    boolean verified = nonce != 0 && nonce == authenticationResponse.responderNonce;
+                    boolean verified = nonce != 0 && nonce == authenticationFinalResponse.responderNonce;
                     if (verified) {
-                        addReportedPeersConsumer.accept(authenticationResponse.reportedPeers, connection);
+                        addReportedPeersConsumer.accept(authenticationFinalResponse.reportedPeers, connection);
                         log.info("AuthenticationComplete: Peer with address " + peerAddress
                                 + " authenticated (" + connection.getUid() + "). Took "
                                 + (System.currentTimeMillis() - startAuthTs) + " ms.");
                         completed(connection);
                     } else {
-                        log.warn("Verification of nonce failed. authenticationResponse=" + authenticationResponse + " / nonce=" + nonce);
+                        log.warn("Verification of nonce failed. authenticationResponse=" + authenticationFinalResponse + " / nonce=" + nonce);
-                        failed(new Exception("Verification of nonce failed. getPeersMessage=" + authenticationResponse + " / nonce=" + nonce));
+                        failed(new Exception("Verification of nonce failed. getPeersMessage=" + authenticationFinalResponse + " / nonce=" + nonce));
                     }
                 }
             }
@@ -251,9 +254,11 @@ public class AuthenticationHandshake implements MessageListener {
     // Cancel if we send reject message
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    public void cancel() {
+    public void setOwnRequestCanceled() {
         Log.traceCall();
-        failed(new CancelAuthenticationException());
+        nonce = 0;
+        stopped = false;
+        ownRequestCanceled = true;
     }
 
 

network/src/main/java/io/bitsquare/p2p/peers/CancelAuthenticationException.java

@@ -1,4 +0,0 @@
-package io.bitsquare.p2p.peers;
-
-public class CancelAuthenticationException extends Exception {
-}

network/src/main/java/io/bitsquare/p2p/peers/PeerGroup.java

@@ -98,7 +98,13 @@ public class PeerGroup implements MessageListener, ConnectionListener {
     @Override
     public void onDisconnect(Reason reason, Connection connection) {
         log.debug("onDisconnect connection=" + connection + " / reason=" + reason);
-        connection.getPeerAddress().ifPresent(peerAddress -> removePeer(peerAddress));
+
+        connection.getPeerAddress().ifPresent(peerAddress -> {
+            // We only remove it if we are nto in the authentication process
+            // Connection shut down is a step in the authentication process.
+            if (!authenticationHandshakes.containsKey(peerAddress))
+                removePeer(peerAddress);
+        });
     }
 
     @Override
@@ -207,8 +213,8 @@ public class PeerGroup implements MessageListener, ConnectionListener {
                 log.info("We got an incoming AuthenticationRequest but we have started ourselves already " +
                         "an authentication handshake for that peerAddress ({})", peerAddress);
                 log.debug("We avoid such race conditions by rejecting the request if the hashCode of our address ({}) is " +
-                                "smaller then the hashCode of the peers address ({}).", getMyAddress().hashCode(),
+                                "smaller then the hashCode of the peers address ({}). Result = {}", getMyAddress().hashCode(),
-                        message.senderAddress.hashCode());
+                        message.senderAddress.hashCode(), (getMyAddress().hashCode() < peerAddress.hashCode()));
 
                 authenticationHandshake = authenticationHandshakes.get(peerAddress);
 
@@ -233,7 +239,6 @@ public class PeerGroup implements MessageListener, ConnectionListener {
     private void processAuthenticationRejection(AuthenticationRejection message) {
         Log.traceCall(message.toString());
         Address peerAddress = message.senderAddress;
-
         cancelOwnAuthenticationRequest(peerAddress);
     }
 
@@ -261,8 +266,7 @@ public class PeerGroup implements MessageListener, ConnectionListener {
     private void cancelOwnAuthenticationRequest(Address peerAddress) {
         Log.traceCall();
         if (authenticationHandshakes.containsKey(peerAddress)) {
-            authenticationHandshakes.get(peerAddress).cancel();
+            authenticationHandshakes.get(peerAddress).setOwnRequestCanceled();
-            authenticationHandshakes.remove(peerAddress);
         }
     }
 
@@ -301,9 +305,9 @@ public class PeerGroup implements MessageListener, ConnectionListener {
 
                 @Override
                 public void onFailure(@NotNull Throwable throwable) {
-                    log.info("Authentication to " + peerAddress + " failed." +
+                    log.info("Authentication to " + peerAddress + " failed at authenticateToFirstSeedNode." +
                             "\nThat is expected if seed nodes are offline." +
-                            "\nException:" + throwable.getMessage());
+                            "\nException:" + throwable.toString());
 
                     removePeer(peerAddress);
 
@@ -340,9 +344,9 @@ public class PeerGroup implements MessageListener, ConnectionListener {
 
                             @Override
                             public void onFailure(@NotNull Throwable throwable) {
-                                log.info("Authentication to " + peerAddress + " failed." +
+                                log.info("Authentication to " + peerAddress + " failed at authenticateToRemainingSeedNode." +
                                         "\nThat is expected if the seed node is offline." +
-                                        "\nException:" + throwable.getMessage());
+                                        "\nException:" + throwable.toString());
 
                                 removePeer(peerAddress);
 
@@ -394,9 +398,9 @@ public class PeerGroup implements MessageListener, ConnectionListener {
 
                         @Override
                         public void onFailure(@NotNull Throwable throwable) {
-                            log.info("Authentication to " + peerAddress + " failed." +
+                            log.info("Authentication to " + peerAddress + " failed at authenticateToRemainingReportedPeer." +
                                     "\nThat is expected if the peer is offline." +
-                                    "\nException:" + throwable.getMessage());
+                                    "\nException:" + throwable.toString());
 
                             removePeer(peerAddress);
 
@@ -473,9 +477,9 @@ public class PeerGroup implements MessageListener, ConnectionListener {
 
                 @Override
                 public void onFailure(@NotNull Throwable throwable) {
-                    log.error("Authentication to " + peerAddress + " for sending a private message failed." +
+                    log.error("Authentication to " + peerAddress + " for sending a private message failed at authenticateToDirectMessagePeer." +
                             "\nSeems that the peer is offline." +
-                            "\nException:" + throwable.getMessage());
+                            "\nException:" + throwable.toString());
                     removePeer(peerAddress);
                     if (faultHandler != null)
                         faultHandler.run();

network/src/main/java/io/bitsquare/p2p/peers/messages/auth/AuthenticationFinalResponse.java

@@ -6,14 +6,14 @@ import io.bitsquare.p2p.peers.ReportedPeer;
 
 import java.util.HashSet;
 
-public final class AuthenticationResponse extends AuthenticationMessage {
+public final class AuthenticationFinalResponse extends AuthenticationMessage {
     // That object is sent over the wire, so we need to take care of version compatibility.
     private static final long serialVersionUID = Version.NETWORK_PROTOCOL_VERSION;
 
     public final long responderNonce;
     public final HashSet<ReportedPeer> reportedPeers;
 
-    public AuthenticationResponse(Address senderAddress, long responderNonce, HashSet<ReportedPeer> reportedPeers) {
+    public AuthenticationFinalResponse(Address senderAddress, long responderNonce, HashSet<ReportedPeer> reportedPeers) {
         super(senderAddress);
         this.responderNonce = responderNonce;
         this.reportedPeers = reportedPeers;