Repository: ethereum/py-evm
Issue: rlp decoding error crashes `DiscoveryService`

### What is wrong?

The discovery service crashed because of an `rlp.DecodingError`

```
Traceback (most recent call last):
  File "/home/piper/projects/py-evm/p2p/service.py", line 60, in run
    await self._run()
  File "/home/piper/projects/py-evm/p2p/discovery.py", line 339, in _run
    await self.maybe_connect_to_more_peers()
  File "/home/piper/projects/py-evm/p2p/discovery.py", line 351, in maybe_connect_to_more_peers
    self.proto.get_nodes_to_connect(self.peer_pool.max_peers))
  File "/home/piper/projects/py-evm/p2p/peer.py", line 937, in connect_to_nodes
    await self.start_peer(peer)
  File "/home/piper/projects/py-evm/p2p/peer.py", line 852, in start_peer
    msgs = await self.ensure_same_side_on_dao_fork(peer)
  File "/home/piper/projects/py-evm/p2p/peer.py", line 965, in ensure_same_side_on_dao_fork
    while True:
  File "/home/piper/projects/py-evm/p2p/cancel_token.py", line 102, in wait
    return await self.wait_first(awaitable, token=token, timeout=timeout)
  File "/home/piper/projects/py-evm/p2p/cancel_token.py", line 124, in wait_first
    return await wait_with_token(*awaitables, token=token_chain, timeout=timeout)
  File "/home/piper/projects/py-evm/p2p/cancel_token.py", line 161, in wait_with_token
    return done.pop().result()
  File "/home/piper/projects/py-evm/p2p/peer.py", line 448, in read_msg
    decoded_msg = cast(Dict[str, Any], cmd.decode(msg))
  File "/home/piper/projects/py-evm/p2p/protocol.py", line 97, in decode
    return self.decode_payload(data[1:])
  File "/home/piper/projects/py-evm/p2p/protocol.py", line 84, in decode_payload
    data = rlp.decode(rlp_data, sedes=decoder)
  File "/home/piper/python-environments/evm/lib/python3.6/site-packages/rlp/codec.py", line 213, in decode
    item, end = consume_item(rlp, 0)
  File "/home/piper/python-environments/evm/lib/python3.6/site-packages/rlp/codec.py", line 191, in consume_item
    return consume_payload(rlp, s, t, l)
  File "/home/piper/python-environments/evm/lib/python3.6/site-packages/rlp/codec.py", line 172, in consume_payload
    item, next_item_start = consume_payload(rlp, s, t, l)
  File "/home/piper/python-environments/evm/lib/python3.6/site-packages/rlp/codec.py", line 172, in consume_payload
    item, next_item_start = consume_payload(rlp, s, t, l)
  File "/home/piper/python-environments/evm/lib/python3.6/site-packages/rlp/codec.py", line 171, in consume_payload
    t, l, s = consume_length_prefix(rlp, next_item_start)
  File "/home/piper/python-environments/evm/lib/python3.6/site-packages/rlp/codec.py", line 131, in consume_length_prefix
    raise DecodingError('Encoded as short string although single byte was possible', rlp)
rlp.exceptions.DecodingError: Encoded as short string although single byte was possible
```


### How can it be fixed

I think this should be handled in the `_run` loop of the `BasePeer` class.  Would like a second opinion.

Changed files:
p2p/exceptions.py, p2p/peer.py, p2p/protocol.py

diff --git a/p2p/exceptions.py b/p2p/exceptions.py
index 05a3ff1c..dbe6256b 100644
--- a/p2p/exceptions.py
+++ b/p2p/exceptions.py
@@ -42,6 +42,7 @@ class MalformedMessage(BaseP2PError):
     """
     Raised when a p2p command is received with a malformed message
     """
+    pass
 
 
 class UnknownProtocolCommand(BaseP2PError):
diff --git a/p2p/peer.py b/p2p/peer.py
index aeae5e12..187971f8 100644
--- a/p2p/peer.py
+++ b/p2p/peer.py
@@ -355,10 +355,18 @@ class BasePeer(BaseService):
         # too much time is being spent on this again, we need to consider running this in a
         # ProcessPoolExecutor(). Need to make sure we don't use all CPUs in the machine for that,
         # though, otherwise asyncio's event loop can't run and we can't keep up with other peers.
-        decoded_msg = cast(Dict[str, Any], cmd.decode(msg))
-        self.logger.trace("Successfully decoded %s msg: %s", cmd, decoded_msg)
-        self.received_msgs[cmd] += 1
-        return cmd, decoded_msg
+        try:
+            decoded_msg = cast(Dict[str, Any], cmd.decode(msg))
+        except MalformedMessage as err:
+            self.logger.debug(
+                "Malformed message from peer %s: CMD:%s Error: %r",
+                self, type(cmd).__name__, err,
+            )
+            raise
+        else:
+            self.logger.trace("Successfully decoded %s msg: %s", cmd, decoded_msg)
+            self.received_msgs[cmd] += 1
+            return cmd, decoded_msg
 
     def handle_p2p_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgType) -> None:
         """Handle the base protocol (P2P) messages."""
@@ -837,8 +845,8 @@ class PeerPool(BaseService, AsyncIterable[BasePeer]):
             # check, we do it here because we want to perform it for incoming peer connections as
             # well.
             msgs = await self.ensure_same_side_on_dao_fork(peer)
-        except DAOForkCheckFailure as e:
-            self.logger.debug("DAO fork check with %s failed: %s", peer, e)
+        except DAOForkCheckFailure as err:
+            self.logger.debug("DAO fork check with %s failed: %s", peer, err)
             await peer.disconnect(DisconnectReason.useless_peer)
             return
         asyncio.ensure_future(peer.run(finished_callback=self._peer_finished))
@@ -962,9 +970,15 @@ class PeerPool(BaseService, AsyncIterable[BasePeer]):
                     else:
                         msgs.append((cmd, msg))
                         continue
-            except (TimeoutError, PeerConnectionLost) as e:
+            except (TimeoutError, PeerConnectionLost) as err:
+                raise DAOForkCheckFailure(
+                    "Timed out waiting for DAO fork header from {}: {}".format(peer, err))
+            except MalformedMessage as err:
                 raise DAOForkCheckFailure(
-                    "Timed out waiting for DAO fork header from {}: {}".format(peer, e))
+                    "Malformed message while doing DAO fork check with {0}: {1}".format(
+                        peer, err,
+                    )
+                ) from err
 
             try:
                 request.validate_headers(headers)
diff --git a/p2p/protocol.py b/p2p/protocol.py
index 6928a3f4..ab73e8fb 100644
--- a/p2p/protocol.py
+++ b/p2p/protocol.py
@@ -22,7 +22,10 @@ from eth_typing import BlockIdentifier, BlockNumber
 from eth.constants import NULL_BYTE
 from eth.rlp.headers import BlockHeader
 
-from p2p.exceptions import ValidationError
+from p2p.exceptions import (
+    MalformedMessage,
+    ValidationError,
+)
 from p2p.utils import get_devp2p_cmd_id
 
 
@@ -87,7 +90,13 @@ class Command:
         else:
             decoder = sedes.List(
                 [type_ for _, type_ in self.structure], strict=self.decode_strict)
-        data = rlp.decode(rlp_data, sedes=decoder)
+        try:
+            data = rlp.decode(rlp_data, sedes=decoder)
+        except rlp.DecodingError as err:
+            raise MalformedMessage(
+                "Malformed %s message: %r".format(type(self).__name__, err)
+            ) from err
+
         if isinstance(self.structure, sedes.CountableList):
             return data
         return {
@@ -99,7 +108,7 @@ class Command:
     def decode(self, data: bytes) -> _DecodedMsgType:
         packet_type = get_devp2p_cmd_id(data)
         if packet_type != self.cmd_id:
-            raise ValueError("Wrong packet type: {}".format(packet_type))
+            raise MalformedMessage("Wrong packet type: {}".format(packet_type))
         return self.decode_payload(data[1:])
 
     def encode(self, data: _DecodedMsgType) -> Tuple[bytes, bytes]: