Enforce length bounds for Req/Resp messages

ethereum/datastructures/src/main/java/tech/pegasys/teku/datastructures/util/LengthBounds.java

@@ -38,6 +38,10 @@ public long getMax() {
     return max;
   }
 
+  public boolean isWithinBounds(final long length) {
+    return length >= min && length <= max;
+  }
+
   public LengthBounds add(final LengthBounds other) {
     return new LengthBounds(this.min + other.min, this.max + other.max);
   }

networking/eth2/src/integration-test/java/tech/pegasys/teku/networking/eth2/ErrorConditionsIntegrationTest.java

@@ -21,14 +21,16 @@
 import org.apache.tuweni.bytes.Bytes32;
 import org.apache.tuweni.ssz.SSZ;
 import org.assertj.core.api.Assertions;
+import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 import tech.pegasys.teku.datastructures.networking.libp2p.rpc.StatusMessage;
 import tech.pegasys.teku.networking.eth2.peers.Eth2Peer;
 import tech.pegasys.teku.networking.eth2.rpc.core.Eth2RpcMethod;
-import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.DeserializationFailedException;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.LengthOutOfBoundsException;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.RpcEncoding;
 import tech.pegasys.teku.util.Waiter;
 import tech.pegasys.teku.util.async.SafeFuture;
@@ -58,10 +60,23 @@ public void shouldRejectInvalidRequests(final String encodingName, final RpcEnco
     final SafeFuture<StatusMessage> response =
         peer.requestSingleItem(status, new InvalidStatusMessage());
 
+    final RpcException expected = new LengthOutOfBoundsException();
+
     Assertions.assertThatThrownBy(() -> Waiter.waitFor(response))
         .isInstanceOf(ExecutionException.class)
         .extracting(Throwable::getCause)
-        .isEqualToComparingFieldByField(new DeserializationFailedException());
+        .isInstanceOf(RpcException.class)
+        .is(
+            new Condition<>(
+                error -> {
+                  final RpcException rpcException = (RpcException) error;
+                  return rpcException
+                          .getErrorMessageString()
+                          .equals(expected.getErrorMessageString())
+                      && rpcException.getResponseCode() == expected.getResponseCode();
+                },
+                "Exception did not match expected exception %s",
+                expected));
   }
 
   public static Stream<Arguments> getEncodings() {

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/RpcEncoder.java

@@ -41,6 +41,6 @@ public <T> Bytes encodeSuccessfulResponse(T response) {
 
   public Bytes encodeErrorResponse(RpcException error) {
     return Bytes.concatenate(
-        Bytes.of(error.getResponseCode()), encoding.encodePayload(error.getErrorMessageBytes()));
+        Bytes.of(error.getResponseCode()), encoding.encodePayload(error.getErrorMessage()));
   }
 }

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/RpcException.java

@@ -24,8 +24,10 @@
 import org.apache.tuweni.bytes.Bytes;
 
 public class RpcException extends Exception {
-
   private static final Logger LOG = LogManager.getLogger();
+
+  public static final int MAXIMUM_ERROR_MESSAGE_LENGTH = 256;
+
   // Server errors
   public static class ServerErrorException extends RpcException {
     public ServerErrorException() {
@@ -72,6 +74,12 @@ public ChunkTooLongException() {
     }
   }
 
+  public static class LengthOutOfBoundsException extends RpcException {
+    public LengthOutOfBoundsException() {
+      super(INVALID_REQUEST_CODE, "Chunk length is not within bounds for expected type");
+    }
+  }
+
   private final byte responseCode;
   private final String errorMessage;
 
@@ -81,33 +89,26 @@ public RpcException(final byte responseCode, final String errorMessage) {
     this.errorMessage = errorMessage;
   }
 
-  public RpcException(final byte responseCode, final Bytes errorMessageBytes) {
+  public RpcException(final byte responseCode, final RpcErrorMessage errorMessage) {
     this.responseCode = responseCode;
-    String err;
-    try {
-      err = new String(errorMessageBytes.toArray(), StandardCharsets.UTF_8);
-    } catch (IllegalArgumentException ex) {
-      err = errorMessageBytes.toHexString().toLowerCase();
-      LOG.trace("Error message could not be read as UTF-8: {} ", err);
-    }
-    this.errorMessage = err;
+    this.errorMessage = errorMessage.toString();
   }
 
   public byte getResponseCode() {
     return responseCode;
   }
 
-  public String getErrorMessage() {
+  public String getErrorMessageString() {
     return errorMessage;
   }
 
-  public Bytes getErrorMessageBytes() {
+  public RpcErrorMessage getErrorMessage() {
     Bytes bytes = Bytes.wrap(errorMessage.getBytes(UTF_8));
-    if (bytes.size() > 256) {
-      LOG.debug("Message {} was longer than 256 bytes", errorMessage);
-      return bytes.slice(0, 256);
+    if (bytes.size() > MAXIMUM_ERROR_MESSAGE_LENGTH) {
+      LOG.debug("Message {} was longer than {} bytes", errorMessage, MAXIMUM_ERROR_MESSAGE_LENGTH);
+      return new RpcErrorMessage(bytes.slice(0, MAXIMUM_ERROR_MESSAGE_LENGTH));
     }
-    return bytes;
+    return new RpcErrorMessage(bytes);
   }
 
   @Override
@@ -126,4 +127,26 @@ public boolean equals(final Object o) {
   public int hashCode() {
     return Objects.hash(responseCode, errorMessage);
   }
+
+  public static class RpcErrorMessage {
+    private final Bytes data;
+
+    public RpcErrorMessage(final Bytes data) {
+      this.data = data;
+    }
+
+    public Bytes getData() {
+      return data;
+    }
+
+    @Override
+    public String toString() {
+      try {
+        return new String(data.toArray(), StandardCharsets.UTF_8);
+      } catch (IllegalArgumentException ex) {
+        LOG.trace("Error message could not be read as UTF-8: {} ", data);
+        return data.toHexString().toLowerCase();
+      }
+    }
+  }
 }

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/RpcResponseCallback.java

@@ -42,7 +42,7 @@ public void completeSuccessfully() {
 
   @Override
   public void completeWithErrorResponse(final RpcException error) {
-    LOG.debug("Responding to RPC request with error: {}", error.getErrorMessage());
+    LOG.debug("Responding to RPC request with error: {}", error.getErrorMessageString());
     try {
       rpcStream.writeBytes(rpcEncoder.encodeErrorResponse(error)).reportExceptions();
     } catch (StreamClosedException e) {

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/RpcResponseDecoder.java

@@ -20,8 +20,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import org.apache.tuweni.bytes.Bytes;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.PayloadTruncatedException;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.RpcErrorMessage;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.RpcByteBufDecoder;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.RpcEncoding;
 
@@ -33,7 +33,7 @@
 public class RpcResponseDecoder<T> {
   private Optional<Integer> respCodeMaybe = Optional.empty();
   private Optional<RpcByteBufDecoder<T>> payloadDecoder = Optional.empty();
-  private Optional<RpcByteBufDecoder<Bytes>> errorDecoder = Optional.empty();
+  private Optional<RpcByteBufDecoder<RpcErrorMessage>> errorDecoder = Optional.empty();
   private final Class<T> responseType;
   private final RpcEncoding encoding;
 
@@ -78,7 +78,7 @@ private Optional<T> decodeNextResponse(final ByteBuf data) throws RpcException {
       return ret;
     } else {
       if (errorDecoder.isEmpty()) {
-        errorDecoder = Optional.of(encoding.createDecoder(Bytes.class));
+        errorDecoder = Optional.of(encoding.createDecoder(RpcErrorMessage.class));
       }
       Optional<RpcException> rpcException =
           errorDecoder

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/LengthPrefixedPayloadDecoder.java

@@ -23,6 +23,7 @@
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.ChunkTooLongException;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.DecompressFailedException;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.ExtraDataAppendedException;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.LengthOutOfBoundsException;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.MessageTruncatedException;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.PayloadTruncatedException;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.compression.Compressor;
@@ -59,8 +60,14 @@ public Optional<T> decodeOneMessage(final ByteBuf in) throws RpcException {
     }
 
     if (decompressor.isEmpty()) {
-      readLengthPrefixHeader(in)
-          .ifPresent(len -> decompressor = Optional.of(compressor.createDecompressor(len)));
+      final Optional<Integer> maybeLength = readLengthPrefixHeader(in);
+      if (maybeLength.isPresent()) {
+        final int length = maybeLength.get();
+        if (!payloadEncoder.isLengthWithinBounds(length)) {
+          throw new LengthOutOfBoundsException();
+        }
+        decompressor = Optional.of(compressor.createDecompressor(length));
+      }
     }
     if (decompressor.isPresent()) {
       final Optional<ByteBuf> ret;

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/RpcPayloadEncoder.java

@@ -20,4 +20,6 @@ public interface RpcPayloadEncoder<T> {
   Bytes encode(T message);
 
   T decode(Bytes message) throws RpcException;
+
+  boolean isLengthWithinBounds(long length);
 }

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/RpcPayloadEncoders.java

@@ -20,9 +20,10 @@
 import java.util.Map;
 import java.util.function.Function;
 import tech.pegasys.teku.datastructures.networking.libp2p.rpc.BeaconBlocksByRootRequestMessage;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.RpcErrorMessage;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.ssz.BeaconBlocksByRootRequestMessageEncoder;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.ssz.DefaultRpcPayloadEncoder;
-import tech.pegasys.teku.networking.eth2.rpc.core.encodings.ssz.StringSszEncoder;
+import tech.pegasys.teku.networking.eth2.rpc.core.encodings.ssz.RpcErrorMessagePayloadEncoder;
 
 public class RpcPayloadEncoders {
 
@@ -40,7 +41,7 @@ public static RpcPayloadEncoders createSszEncoders() {
     return RpcPayloadEncoders.builder()
         .withEncoder(
             BeaconBlocksByRootRequestMessage.class, new BeaconBlocksByRootRequestMessageEncoder())
-        .withEncoder(String.class, new StringSszEncoder())
+        .withEncoder(RpcErrorMessage.class, new RpcErrorMessagePayloadEncoder())
         .defaultEncoderProvider(DefaultRpcPayloadEncoder::new)
         .build();
   }
@@ -56,7 +57,7 @@ public <T> RpcPayloadEncoder<T> getEncoder(final Class<T> clazz) {
   }
 
   public static class Builder {
-    private Map<Class<?>, RpcPayloadEncoder<?>> encoders = new HashMap<>();
+    private final Map<Class<?>, RpcPayloadEncoder<?>> encoders = new HashMap<>();
     private Function<Class<?>, RpcPayloadEncoder<?>> defaultEncoderProvider;
 
     private Builder() {}

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/ssz/BeaconBlocksByRootRequestMessageEncoder.java

@@ -24,6 +24,7 @@
 import org.apache.tuweni.bytes.Bytes32;
 import org.apache.tuweni.ssz.SSZ;
 import tech.pegasys.teku.datastructures.networking.libp2p.rpc.BeaconBlocksByRootRequestMessage;
+import tech.pegasys.teku.datastructures.util.LengthBounds;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException;
 import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.DeserializationFailedException;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.RpcPayloadEncoder;
@@ -32,6 +33,9 @@ public class BeaconBlocksByRootRequestMessageEncoder
     implements RpcPayloadEncoder<BeaconBlocksByRootRequestMessage> {
   private static final Logger LOG = LogManager.getLogger();
 
+  private static final LengthBounds VALID_LENGTH_BOUNDS =
+      new LengthBounds(Bytes32.SIZE, MAX_REQUEST_BLOCKS * Bytes32.SIZE);
+
   @Override
   public Bytes encode(final BeaconBlocksByRootRequestMessage message) {
     return SSZ.encode(writer -> writer.writeFixedBytesVector(message.getBlockRoots().asList()));
@@ -54,4 +58,9 @@ public BeaconBlocksByRootRequestMessage decode(final Bytes message) throws RpcEx
 
     return new BeaconBlocksByRootRequestMessage(blockRoots);
   }
+
+  @Override
+  public boolean isLengthWithinBounds(final long length) {
+    return VALID_LENGTH_BOUNDS.isWithinBounds(length);
+  }
 }

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/ssz/DefaultRpcPayloadEncoder.java

@@ -33,18 +33,12 @@ public DefaultRpcPayloadEncoder(final Class<T> clazz) {
 
   @Override
   public Bytes encode(final T message) {
-    return message instanceof Bytes
-        ? (Bytes) message
-        : SimpleOffsetSerializer.serialize((SimpleOffsetSerializable) message);
+    return SimpleOffsetSerializer.serialize((SimpleOffsetSerializable) message);
   }
 
-  @SuppressWarnings("unchecked")
   @Override
   public T decode(final Bytes message) throws RpcException {
     try {
-      if (clazz.equals(Bytes.class)) {
-        return (T) message;
-      }
       return SimpleOffsetSerializer.deserialize(message, clazz);
     } catch (final InvalidSSZTypeException e) {
       if (LOG.isTraceEnabled()) {
@@ -53,4 +47,9 @@ public T decode(final Bytes message) throws RpcException {
       throw new DeserializationFailedException();
     }
   }
+
+  @Override
+  public boolean isLengthWithinBounds(final long length) {
+    return SimpleOffsetSerializer.getLengthBounds(clazz).isWithinBounds(length);
+  }
 }

networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/ssz/StringSszEncoder.java → networking/eth2/src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/encodings/ssz/RpcErrorMessagePayloadEncoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 ConsenSys AG.
+ * Copyright 2020 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
@@ -13,19 +13,25 @@
 
 package tech.pegasys.teku.networking.eth2.rpc.core.encodings.ssz;
 
-import java.nio.charset.StandardCharsets;
 import org.apache.tuweni.bytes.Bytes;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException;
+import tech.pegasys.teku.networking.eth2.rpc.core.RpcException.RpcErrorMessage;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.RpcPayloadEncoder;
 
-public class StringSszEncoder implements RpcPayloadEncoder<String> {
+public class RpcErrorMessagePayloadEncoder implements RpcPayloadEncoder<RpcErrorMessage> {
 
   @Override
-  public Bytes encode(final String message) {
-    return Bytes.wrap(message.getBytes(StandardCharsets.UTF_8));
+  public Bytes encode(final RpcErrorMessage message) {
+    return message.getData();
   }
 
   @Override
-  public String decode(final Bytes message) {
-    return new String(message.toArray(), StandardCharsets.UTF_8);
+  public RpcErrorMessage decode(final Bytes message) {
+    return new RpcErrorMessage(message);
+  }
+
+  @Override
+  public boolean isLengthWithinBounds(final long length) {
+    return length <= RpcException.MAXIMUM_ERROR_MESSAGE_LENGTH;
   }
 }