PR Feedback

Author: pipermerriam

trinity/protocol/common/managers.py

@@ -12,8 +12,10 @@
 from cancel_token import CancelToken
 
 from p2p.exceptions import (
+    MalformedMessage,
     ValidationError,
 )
+from p2p.p2p_proto import DisconnectReason
 from p2p.peer import BasePeer, PeerSubscriber
 from p2p.protocol import (
     Command,
@@ -83,13 +85,14 @@ async def _handle_msg(self, msg: TResponse) -> None:
 
         try:
             response = await self._normalize_response(msg)
-        except ValidationError as err:
-            self.logger.debug(
-                "Malformed response for pending %s request from peer %s: %s",
+        except MalformedMessage as err:
+            self.logger.warn(
+                "Malformed response for pending %s request from peer %s, disconnecting: %s",
                 self.response_msg_name,
                 self._peer,
                 err,
             )
+            await self._peer.disconnect(DisconnectReason.bad_protocol)
             return
 
         try:

trinity/protocol/common/requests.py

@@ -32,12 +32,6 @@ class BaseHeaderRequest(BaseRequest):
     skip: int
     reverse: bool
 
-    def validate_response(self, response: Tuple[BlockHeader, ...]) -> None:
-        """
-        Core `Request` API used for validation.
-        """
-        return self.validate_headers(response)
-
     @property
     @abstractmethod
     def max_size(self) -> int:
@@ -73,13 +67,12 @@ def generate_block_numbers(self,
     def is_numbered(self) -> bool:
         return isinstance(self.block_number_or_hash, int)
 
-    def validate_headers(self,
-                         headers: Tuple[BlockHeader, ...]) -> None:
-        if not headers:
+    def validate_response(self, response: Tuple[BlockHeader, ...]) -> None:
+        if not response:
             # An empty response is always valid
             return
         elif not self.is_numbered:
-            first_header = headers[0]
+            first_header = response[0]
             if first_header.hash != self.block_number_or_hash:
                 raise ValidationError(
                     "Returned headers cannot be matched to header request. "
@@ -91,7 +84,7 @@ def validate_headers(self,
                 )
 
         block_numbers: Tuple[BlockNumber, ...] = tuple(
-            header.block_number for header in headers
+            header.block_number for header in response
         )
         return self.validate_sequence(block_numbers)
 

trinity/protocol/eth/managers.py

@@ -13,7 +13,7 @@
 
 from eth.rlp.headers import BlockHeader
 
-from p2p.exceptions import ValidationError
+from p2p.exceptions import MalformedMessage
 from p2p.protocol import (
     Command,
 )
@@ -100,9 +100,9 @@ async def _normalize_response(self,
                                   msg: Tuple[bytes, ...]
                                   ) -> Tuple[Tuple[Hash32, bytes], ...]:
         if not isinstance(msg, tuple):
-            raise ValidationError("Invalid msg, must be tuple of byte strings")
+            raise MalformedMessage("Invalid msg, must be tuple of byte strings")
         elif not all(isinstance(item, bytes) for item in msg):
-            raise ValidationError("Invalid msg, must be tuple of byte strings")
+            raise MalformedMessage("Invalid msg, must be tuple of byte strings")
 
         node_keys = await self._run_in_executor(tuple, map(keccak, msg))
         return tuple(zip(node_keys, msg))

trinity/protocol/eth/requests.py

@@ -40,17 +40,11 @@ def __init__(self, node_hashes: Tuple[Hash32, ...]) -> None:
         self.node_hashes = node_hashes
 
     def validate_response(self, response: Tuple[Tuple[Hash32, bytes], ...]) -> None:
-        """
-        Core `Request` API used for validation.
-        """
-        return self.validate_node_data(response)
-
-    def validate_node_data(self, node_data: Tuple[Tuple[Hash32, bytes], ...]) -> None:
-        if not node_data:
+        if not response:
             # an empty response is always valid
             return
 
-        node_keys = tuple(node_key for node_key, node in node_data)
+        node_keys = tuple(node_key for node_key, node in response)
         node_key_set = set(node_keys)
 
         if len(node_keys) != len(node_key_set):

trinity/protocol/les/managers.py

@@ -11,7 +11,7 @@
 from eth.rlp.headers import BlockHeader
 
 from p2p.exceptions import (
-    ValidationError,
+    MalformedMessage,
 )
 from p2p.protocol import (
     Command,
@@ -72,11 +72,11 @@ async def _normalize_response(self,
                                   msg: Dict[str, Any]
                                   ) -> Tuple[BlockHeader, ...]:
         if not isinstance(msg, dict):
-            raise ValidationError("msg must be a dictionary")
+            raise MalformedMessage("msg must be a dictionary")
         elif 'headers' not in msg:
-            raise ValidationError("No 'headers' key found in response")
+            raise MalformedMessage("No 'headers' key found in response")
         elif not all(isinstance(item, BlockHeader) for item in msg['headers']):
-            raise ValidationError(
+            raise MalformedMessage(
                 "`headers` key must be a tuple of `BlockHeader` instances"
             )
 

trinity/protocol/les/requests.py

@@ -1,10 +1,5 @@
-from typing import (
-    Tuple,
-)
-
 from eth_typing import BlockIdentifier
 
-from eth.rlp.headers import BlockHeader
 
 from trinity.protocol.common.requests import (
     BaseHeaderRequest,
@@ -13,9 +8,6 @@
 from .constants import MAX_HEADERS_FETCH
 
 
-BlockHeaders_R = Tuple[BlockHeader, ...]
-
-
 class HeaderRequest(BaseHeaderRequest):
     request_id: int
 

trinity/sync/full/state.py

@@ -229,7 +229,10 @@ async def request_nodes(self, node_keys: Iterable[Hash32]) -> None:
     async def _request_and_process_nodes(self, peer: ETHPeer, batch: Tuple[Hash32, ...]) -> None:
         self.logger.debug("Requesting %d trie nodes from %s", len(batch), peer)
         node_data = await peer.requests.get_node_data(batch)
-        self.request_tracker.active_requests.pop(peer)
+        try:
+            self.request_tracker.active_requests.pop(peer)
+        except KeyError:
+            self.logger.warn("Unexpected error removing peer from active requests: %s", peer)
 
         self.logger.debug("Got %d NodeData entries from %s", len(node_data), peer)
 
@@ -241,7 +244,7 @@ async def _request_and_process_nodes(self, peer: ETHPeer, batch: Tuple[Hash32, .
         # check for missing nodes and re-schedule them
         missing = set(batch).difference(node_keys)
 
-        # TODO: this doesn't actually mean the peer doesn't have them, just
+        # TODO: this doesn't necessarily mean the peer doesn't have them, just
         # that they didn't respond with them this time.  We should explore
         # alternate ways to do this since a false negative here will result in
         # not requesting this node from this peer again.