Reduce log level when invalid messages received from peers (#274)

* Throw FramingException when snappy decompression fails. 
* Handle FramingException in DeFramer as a routine event representing a malformed message.

CHANGELOG.md

@@ -16,6 +16,7 @@
     - ECRec precompile should return empty instead of 32 zero bytes when the input is invalid (PR [#227](https://github.com/PegaSysEng/pantheon/pull/227))
  - File name too long error while building from source ([#215](https://github.com/PegaSysEng/pantheon/issues/215) thanks to [@5chdn](https://github.com/5chdn) for reporting) (PR [#221](https://github.com/PegaSysEng/pantheon/pull/221))
  - Loop syntax in `runPantheonPrivateNetwork.sh` (PR [#237](https://github.com/PegaSysEng/pantheon/pull/237) thanks to [@matt9ucci](https://github.com/matt9ucci))
+ - Fix `CompressionException: Snappy decompression failed` errors ([#251](https://github.com/PegaSysEng/pantheon/issues/251) thanks to [@5chdn](https://github.com/5chdn) for reporting) (PR [#274](https://github.com/PegaSysEng/pantheon/pull/274))
 
 ### Additions and Improvements
  - Added `--ropsten` command line argument to make syncing to Ropsten easier ([#186](https://github.com/PegaSysEng/pantheon/issues/186)) (PR [#197](https://github.com/PegaSysEng/pantheon/pull/197) with thanks to [@jvirtanen](https://github.com/jvirtanen))

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/netty/DeFramer.java

@@ -16,6 +16,7 @@ import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
 import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection;
 import tech.pegasys.pantheon.ethereum.p2p.netty.exceptions.IncompatiblePeerException;
 import tech.pegasys.pantheon.ethereum.p2p.rlpx.framing.Framer;
+import tech.pegasys.pantheon.ethereum.p2p.rlpx.framing.FramingException;
 import tech.pegasys.pantheon.ethereum.p2p.wire.PeerInfo;
 import tech.pegasys.pantheon.ethereum.p2p.wire.SubProtocol;
 import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
@@ -31,6 +32,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.DecoderException;
 import io.netty.handler.timeout.IdleStateHandler;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -117,7 +119,17 @@ final class DeFramer extends ByteToMessageDecoder {
   @Override
   public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable throwable)
       throws Exception {
-    if (throwable instanceof IOException) {
+    final Throwable cause =
+        throwable instanceof DecoderException && throwable.getCause() != null
+            ? throwable.getCause()
+            : throwable;
+    if (cause instanceof FramingException) {
+      LOG.debug("Invalid incoming message", throwable);
+      if (connectFuture.isDone()) {
+        connectFuture.get().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
+        return;
+      }
+    } else if (cause instanceof IOException) {
       // IO failures are routine when communicating with random peers across the network.
       LOG.debug("IO error while processing incoming message", throwable);
     } else {

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/rlpx/framing/CompressionException.java

@@ -1,25 +0,0 @@
-/*
- * Copyright 2018 ConsenSys AG.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package tech.pegasys.pantheon.ethereum.p2p.rlpx.framing;
-
-/** Thrown when an error occurs during compression and decompression of payloads. */
-public class CompressionException extends RuntimeException {
-
-  public CompressionException(final String message) {
-    super(message);
-  }
-
-  public CompressionException(final String message, final Throwable cause) {
-    super(message, cause);
-  }
-}

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/rlpx/framing/Compressor.java

@@ -1,47 +0,0 @@
-/*
- * Copyright 2018 ConsenSys AG.
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package tech.pegasys.pantheon.ethereum.p2p.rlpx.framing;
-
-/** A strategy for compressing and decompressing devp2p subprotocol messages. */
-public interface Compressor {
-
-  /**
-   * Compresses the provided payload.
-   *
-   * @param decompressed The original payload.
-   * @throws CompressionException Thrown if an error occurs during compression; expect to find the
-   *     root cause inside.
-   * @return The compressed payload.
-   */
-  byte[] compress(byte[] decompressed) throws CompressionException;
-
-  /**
-   * Decompresses the provided payload.
-   *
-   * @param compressed The compressed payload.
-   * @throws CompressionException Thrown if an error occurs during decompression; expect to find the
-   *     root cause inside.
-   * @return The original payload.
-   */
-  byte[] decompress(byte[] compressed) throws CompressionException;
-
-  /**
-   * Return the length when uncompressed
-   *
-   * @param compressed The compressed payload.
-   * @return The length of the payload when uncompressed.
-   * @throws CompressionException Thrown if the size cannot be calculated from the available data;
-   *     expect to find the root cause inside;
-   */
-  int uncompressedLength(byte[] compressed) throws CompressionException;
-}

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/rlpx/framing/Framer.java

@@ -34,6 +34,7 @@ import org.bouncycastle.crypto.engines.AESEngine;
 import org.bouncycastle.crypto.modes.SICBlockCipher;
 import org.bouncycastle.crypto.params.KeyParameter;
 import org.bouncycastle.crypto.params.ParametersWithIV;
+import org.iq80.snappy.CorruptionException;
 
 /**
  * This component is responsible for reading and composing RLPx protocol frames, conformant to the
@@ -68,7 +69,7 @@ public class Framer {
           .extractArray();
 
   private final HandshakeSecrets secrets;
-  private static final Compressor compressor = new SnappyCompressor();
+  private static final SnappyCompressor compressor = new SnappyCompressor();
   private final StreamCipher encryptor;
   private final StreamCipher decryptor;
   private final BlockCipher macEncryptor;
@@ -254,14 +255,17 @@ public class Framer {
     // Write message data to ByteBuf, decompressing as necessary
     final BytesValue data;
     if (compressionEnabled) {
-      // Decompress data before writing to ByteBuf
       final byte[] compressedMessageData = Arrays.copyOfRange(frameData, 1, frameData.length - pad);
-      // Check message length
-      Preconditions.checkState(
-          compressor.uncompressedLength(compressedMessageData) < LENGTH_MAX_MESSAGE_FRAME,
-          "Message size in excess of maximum length.");
+      final int uncompressedLength = compressor.uncompressedLength(compressedMessageData);
+      if (uncompressedLength >= LENGTH_MAX_MESSAGE_FRAME) {
+        throw error("Message size %s in excess of maximum length.", uncompressedLength);
+      }
+      try {
         final byte[] decompressedMessageData = compressor.decompress(compressedMessageData);
         data = BytesValue.wrap(decompressedMessageData);
+      } catch (final CorruptionException e) {
+        throw new FramingException("Decompression failed", e);
+      }
     } else {
       // Move data to a ByteBuf
       final int messageLength = frameSize - LENGTH_MESSAGE_ID;

ethereum/p2p/src/main/java/tech/pegasys/pantheon/ethereum/p2p/rlpx/framing/SnappyCompressor.java

@@ -21,21 +21,18 @@ import org.iq80.snappy.Snappy;
  *
  * @see <a href="https://google.github.io/snappy/">Snappy algorithm</a>
  */
-public class SnappyCompressor implements Compressor {
+public class SnappyCompressor {
 
-  @Override
   public byte[] compress(final byte[] uncompressed) {
     checkNotNull(uncompressed, "input data must not be null");
     return Snappy.compress(uncompressed);
   }
 
-  @Override
   public byte[] decompress(final byte[] compressed) {
     checkNotNull(compressed, "input data must not be null");
     return Snappy.uncompress(compressed, 0, compressed.length);
   }
 
-  @Override
   public int uncompressedLength(final byte[] compressed) {
     checkNotNull(compressed, "input data must not be null");
     return Snappy.getUncompressedLength(compressed, 0);

ethereum/p2p/src/test/java/tech/pegasys/pantheon/ethereum/p2p/rlpx/framing/FramerTest.java

@@ -86,9 +86,9 @@ public class FramerTest {
     final Framer deframer = new Framer(deframeSecrets);
     deframer.enableCompression();
 
-    assertThatExceptionOfType(IllegalStateException.class)
+    assertThatExceptionOfType(FramingException.class)
         .isThrownBy(() -> deframer.deframe(framedMessage))
-        .withMessageContaining("Message size in excess of maximum length.");
+        .withMessageContaining("Message size 16777216 in excess of maximum length.");
   }
 
   @Test

pantheon/src/main/java/tech/pegasys/pantheon/Pantheon.java

@@ -16,7 +16,7 @@ import static picocli.CommandLine.defaultExceptionHandler;
 
 import tech.pegasys.pantheon.cli.PantheonCommand;
 import tech.pegasys.pantheon.cli.PantheonControllerBuilder;
-import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration.Builder;
+import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
 import tech.pegasys.pantheon.util.BlockImporter;
 
 import picocli.CommandLine.RunLast;
@@ -32,7 +32,7 @@ public final class Pantheon {
             new BlockImporter(),
             new RunnerBuilder(),
             new PantheonControllerBuilder(),
-            new Builder());
+            new SynchronizerConfiguration.Builder());
 
     pantheonCommand.parse(
         new RunLast().andExit(SUCCESS_EXIT_CODE),