Cope with clients that don't implement EIP-706 (#1917)

Not all clients implement EIP-706, even though they may advertise a
version 5 in the hello packet.  To cope with this if we are expecting 
compression but haven't had a compressed message yet and a new message
fails to decompress we turn off compression and try again.

Signed-off-by: Danno Ferrin <danno.ferrin@gmail.com>

ethereum/p2p/src/main/java/org/hyperledger/besu/ethereum/p2p/rlpx/framing/Framer.java

@@ -30,6 +30,8 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.errorprone.annotations.FormatMethod;
 import io.netty.buffer.ByteBuf;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.apache.tuweni.bytes.Bytes;
 import org.apache.tuweni.bytes.MutableBytes;
 import org.bouncycastle.crypto.BlockCipher;
@@ -53,6 +55,8 @@ import org.bouncycastle.crypto.params.ParametersWithIV;
  * @see <a href="https://github.com/ethereum/devp2p/blob/master/rlpx.md#framing">RLPx framing</a>
  */
 public class Framer {
+  private static final Logger LOG = LogManager.getLogger();
+
   private static final int LENGTH_HEADER_DATA = 16;
   private static final int LENGTH_MAC = 16;
   private static final int LENGTH_FULL_HEADER = LENGTH_HEADER_DATA + LENGTH_MAC;
@@ -79,6 +83,8 @@ public class Framer {
   private boolean headerProcessed;
   private int frameSize;
   private boolean compressionEnabled = false;
+  // have we ever successfully uncompressed a packet?
+  private boolean compressionSuccessful = false;
 
   /**
    * Creates a new framer out of the handshake secrets derived during the cryptographic handshake.
@@ -109,6 +115,14 @@ public class Framer {
     this.compressionEnabled = false;
   }
 
+  boolean isCompressionEnabled() {
+    return compressionEnabled;
+  }
+
+  boolean isCompressionSuccessful() {
+    return compressionSuccessful;
+  }
+
   /**
    * Deframes a full message from the byte buffer, if possible.
    *
@@ -269,8 +283,24 @@ public class Framer {
       if (uncompressedLength >= LENGTH_MAX_MESSAGE_FRAME) {
         throw error("Message size %s in excess of maximum length.", uncompressedLength);
       }
-      final byte[] decompressedMessageData = compressor.decompress(compressedMessageData);
+      Bytes _data;
-      data = Bytes.wrap(decompressedMessageData);
+      try {
+        final byte[] decompressedMessageData = compressor.decompress(compressedMessageData);
+        _data = Bytes.wrap(decompressedMessageData);
+        compressionSuccessful = true;
+      } catch (final FramingException fe) {
+        if (compressionSuccessful) {
+          throw fe;
+        } else {
+          // OpenEthereum/Parity does not implement EIP-706
+          // If failing on the first packet downgrade to uncompressed
+          compressionEnabled = false;
+          LOG.debug("Snappy decompression failed: downgrading to uncompressed");
+          final int messageLength = frameSize - LENGTH_MESSAGE_ID;
+          _data = Bytes.wrap(frameData, 1, messageLength);
+        }
+      }
+      data = _data;
     } else {
       // Move data to a ByteBuf
       final int messageLength = frameSize - LENGTH_MESSAGE_ID;

ethereum/p2p/src/test/java/org/hyperledger/besu/ethereum/p2p/rlpx/framing/FramerTest.java

@@ -218,7 +218,7 @@ public class FramerTest {
   }
 
   @Test
-  public void shouldThrowFramingExceptionWhenMessageIsNotCompressedButShouldBe() {
+  public void downgradesToUncompressed() {
     final HandshakeSecrets secrets =
         new HandshakeSecrets(
             Bytes.fromHexString(
@@ -239,7 +239,36 @@ public class FramerTest {
 
     // Then read it with compression enabled.
     receivingFramer.enableCompression();
-    assertThatThrownBy(() -> receivingFramer.deframe(out)).isInstanceOf(FramingException.class);
+    assertThat(receivingFramer.deframe(out)).isNotNull();
+    assertThat(receivingFramer.isCompressionEnabled()).isFalse();
+  }
+
+  @Test
+  public void compressionWorks() {
+    final HandshakeSecrets secrets =
+        new HandshakeSecrets(
+            Bytes.fromHexString(
+                    "0x75b3ee95adff0c529a05efd7612aa1dbe5057eb9facdde0dfc837ad143da1d43")
+                .toArray(),
+            Bytes.fromHexString(
+                    "0x030dfd1566f4800c4842c177f7d476b64ae2b99a2aa0ab5600aa2f41a8710575")
+                .toArray(),
+            Bytes.fromHexString(
+                    "0xc9d3385b1588a5969cba312f8c29bedb4cb9d56ec0cf825436addc1ec644f1d6")
+                .toArray());
+    final Framer receivingFramer = new Framer(secrets);
+    final Framer sendingFramer = new Framer(secrets);
+
+    // Write a disconnect message with compression enabled.
+    sendingFramer.enableCompression();
+    final ByteBuf out = Unpooled.buffer();
+    sendingFramer.frame(DisconnectMessage.create(DisconnectReason.TIMEOUT), out);
+
+    // Then read it with compression enabled.
+    receivingFramer.enableCompression();
+    assertThat(receivingFramer.deframe(out)).isNotNull();
+    assertThat(receivingFramer.isCompressionEnabled()).isTrue();
+    assertThat(receivingFramer.isCompressionSuccessful()).isTrue();
   }
 
   private HandshakeSecrets secretsFrom(final JsonNode td, final boolean swap) {