Merge pull request #913 from carver/verify-light-contract

Light client: validate contract code with account

Author: carver

evm/vm/forks/tangerine_whistle/__init__.py

@@ -12,3 +12,7 @@ class TangerineWhistleVM(HomesteadVM):
 
     # classes
     _state_class = TangerineWhistleState  # type: Type[BaseState]
+
+    # Don't bother with any DAO logic in Tangerine VM or later
+    # This is how we skip DAO logic on Ropsten, for example
+    support_dao_fork = False

p2p/cancel_token.py

@@ -105,7 +105,8 @@ async def wait_first(self,
                          *awaitables: Awaitable[_TReturn],
                          token: CancelToken = None,
                          timeout: float = None) -> _TReturn:
-        """Wait for the first awaitable to complete, unless we timeout or the token chain is triggered.
+        """
+        Wait for the first awaitable to complete, unless we timeout or the token chain is triggered.
 
         The given token is chained with this service's token, so triggering either will cancel
         this.

p2p/constants.py

@@ -35,11 +35,15 @@
 
 # Timeout used when waiting for a reply from a remote node.
 REPLY_TIMEOUT = 3
+MAX_REQUEST_ATTEMPTS = 3
 
 # Timeout used when performing the check to ensure peers are on the same side of chain splits as
 # us.
 CHAIN_SPLIT_CHECK_TIMEOUT = 5 * REPLY_TIMEOUT
 
+# Default timeout before giving up on a caller-initiated interaction
+COMPLETION_TIMEOUT = 5
+
 # Types of LES Announce messages
 LES_ANNOUNCE_SIMPLE = 1
 LES_ANNOUNCE_SIGNED = 2

p2p/exceptions.py

@@ -138,6 +138,9 @@ class BadAckMessage(BaseP2PError):
 class BadLESResponse(BaseP2PError):
     """
     Raised when the response to a LES request doesn't contain the data we asked for.
+
+    The peer can be treated as violating protocol. Often, the repurcussion should be
+    disconnection and blacklisting.
     """
     pass
 

p2p/lightchain.py

@@ -1,4 +1,7 @@
 import asyncio
+from functools import (
+    partial,
+)
 from typing import (
     Any,
     Callable,
@@ -24,6 +27,7 @@
 )
 
 from trie import HexaryTrie
+from trie.exceptions import BadTrieProof
 
 from evm.exceptions import (
     BlockNotFound,
@@ -35,10 +39,20 @@
 
 from p2p.exceptions import (
     BadLESResponse,
+    NoConnectedPeers,
+    NoEligiblePeers,
 )
 from p2p.cancel_token import CancelToken
 from p2p import protocol
-from p2p.constants import REPLY_TIMEOUT
+from p2p.constants import (
+    COMPLETION_TIMEOUT,
+    MAX_REORG_DEPTH,
+    MAX_REQUEST_ATTEMPTS,
+    REPLY_TIMEOUT,
+)
+from p2p.p2p_proto import (
+    DisconnectReason,
+)
 from p2p.peer import (
     LESPeer,
     PeerPool,
@@ -47,6 +61,7 @@
 from p2p.rlp import BlockBody
 from p2p.service import (
     BaseService,
+    service_timeout,
 )
 from p2p.utils import gen_request_id
 
@@ -103,11 +118,22 @@ def callback(r: protocol._DecodedMsgType) -> None:
         return cast(Dict[str, Any], reply)
 
     @alru_cache(maxsize=1024, cache_exceptions=False)
+    @service_timeout(COMPLETION_TIMEOUT)
     async def get_block_header_by_hash(self, block_hash: Hash32) -> BlockHeader:
-        peer = cast(LESPeer, self.peer_pool.highest_td_peer)
-        return await self._get_block_header_by_hash(peer, block_hash)
+        """
+        :param block_hash: hash of the header to retrieve
+
+        :return: header returned by peer
+
+        :raise NoEligiblePeers: if no peers are available to fulfill the request
+        :raise TimeoutError: if an individual request or the overall process times out
+        """
+        return await self._retry_on_bad_response(
+            partial(self._get_block_header_by_hash, block_hash)
+        )
 
     @alru_cache(maxsize=1024, cache_exceptions=False)
+    @service_timeout(COMPLETION_TIMEOUT)
     async def get_block_body_by_hash(self, block_hash: Hash32) -> BlockBody:
         peer = cast(LESPeer, self.peer_pool.highest_td_peer)
         self.logger.debug("Fetching block %s from %s", encode_hex(block_hash), peer)
@@ -121,6 +147,7 @@ async def get_block_body_by_hash(self, block_hash: Hash32) -> BlockBody:
     # TODO add a get_receipts() method to BaseChain API, and dispatch to this, as needed
 
     @alru_cache(maxsize=1024, cache_exceptions=False)
+    @service_timeout(COMPLETION_TIMEOUT)
     async def get_receipts(self, block_hash: Hash32) -> List[Receipt]:
         peer = cast(LESPeer, self.peer_pool.highest_td_peer)
         self.logger.debug("Fetching %s receipts from %s", encode_hex(block_hash), peer)
@@ -135,25 +162,158 @@ async def get_receipts(self, block_hash: Hash32) -> List[Receipt]:
     # request accounts and code (and storage?)
 
     @alru_cache(maxsize=1024, cache_exceptions=False)
+    @service_timeout(COMPLETION_TIMEOUT)
     async def get_account(self, block_hash: Hash32, address: Address) -> Account:
-        peer = cast(LESPeer, self.peer_pool.highest_td_peer)
+        return await self._retry_on_bad_response(
+            partial(self._get_account_from_peer, block_hash, address)
+        )
+
+    async def _get_account_from_peer(
+            self,
+            block_hash: Hash32,
+            address: Address,
+            peer: LESPeer) -> Account:
         key = keccak(address)
         proof = await self._get_proof(peer, block_hash, account_key=b'', key=key)
-        header = await self._get_block_header_by_hash(peer, block_hash)
-        rlp_account = HexaryTrie.get_from_proof(header.state_root, key, proof)
+        header = await self._get_block_header_by_hash(block_hash, peer)
+        try:
+            rlp_account = HexaryTrie.get_from_proof(header.state_root, key, proof)
+        except BadTrieProof as exc:
+            raise BadLESResponse("Peer %s returned an invalid proof for account %s at block %s" % (
+                peer,
+                encode_hex(address),
+                encode_hex(block_hash),
+            )) from exc
         return rlp.decode(rlp_account, sedes=Account)
 
     @alru_cache(maxsize=1024, cache_exceptions=False)
-    async def get_contract_code(self, block_hash: Hash32, key: bytes) -> bytes:
-        peer = cast(LESPeer, self.peer_pool.highest_td_peer)
+    @service_timeout(COMPLETION_TIMEOUT)
+    async def get_contract_code(self, block_hash: Hash32, address: Address) -> bytes:
+        """
+        :param block_hash: find code as of the block with block_hash
+        :param address: which contract to look up
+
+        :return: bytecode of the contract, ``b''`` if no code is set
+
+        :raise NoEligiblePeers: if no peers are available to fulfill the request
+        :raise TimeoutError: if an individual request or the overall process times out
+        """
+        # get account for later verification, and
+        # to confirm that our highest total difficulty peer has the info
+        try:
+            account = await self.get_account(block_hash, address)
+        except HeaderNotFound as exc:
+            raise NoEligiblePeers("Our best peer does not have header %s" % block_hash) from exc
+
+        code_hash = account.code_hash
+
+        return await self._retry_on_bad_response(
+            partial(self._get_contract_code_from_peer, block_hash, address, code_hash)
+        )
+
+    async def _get_contract_code_from_peer(
+            self,
+            block_hash: Hash32,
+            address: Address,
+            code_hash: Hash32,
+            peer: LESPeer) -> bytes:
+        """
+        A single attempt to get the contract code from the given peer
+
+        :raise BadLESResponse: if the peer replies with contract code that does not match the
+            account's code hash
+        """
+        # request contract code
         request_id = gen_request_id()
-        peer.sub_proto.send_get_contract_code(block_hash, key, request_id)
+        peer.sub_proto.send_get_contract_code(block_hash, keccak(address), request_id)
         reply = await self._wait_for_reply(request_id)
+
         if not reply['codes']:
-            return b''
-        return reply['codes'][0]
+            bytecode = b''
+        else:
+            bytecode = reply['codes'][0]
+
+        # validate bytecode against a proven account
+        if code_hash == keccak(bytecode):
+            return bytecode
+        elif bytecode == b'':
+            await self._raise_for_empty_code(block_hash, address, code_hash, peer)
+            # The following is added for mypy linting:
+            raise RuntimeError("Unreachable, _raise_for_empty_code must raise its own exception")
+        else:
+            # a bad-acting peer sent an invalid non-empty bytecode
+            raise BadLESResponse("Peer %s sent code %s that did not match hash %s in account %s" % (
+                peer,
+                encode_hex(bytecode),
+                encode_hex(code_hash),
+                encode_hex(address),
+            ))
+
+    async def _raise_for_empty_code(
+            self,
+            block_hash: Hash32,
+            address: Address,
+            code_hash: Hash32,
+            peer: LESPeer) -> None:
+        """
+        A peer might return b'' if it doesn't have the block at the requested header,
+        or it might maliciously return b'' when the code is non-empty. This method tries to tell the
+        difference.
+
+        This method MUST raise an exception, it's trying to determine the appropriate one.
+
+        :raise BadLESResponse: if peer seems to be maliciously responding with invalid empty code
+        :raise NoEligiblePeers: if peer might simply not have the code available
+        """
+        try:
+            header = await self._get_block_header_by_hash(block_hash, peer)
+        except HeaderNotFound:
+            # We presume that the current peer is the best peer. Because
+            # our best peer doesn't have the header we want, there are no eligible peers.
+            raise NoEligiblePeers("Our best peer does not have the header %s" % block_hash)
 
-    async def _get_block_header_by_hash(self, peer: LESPeer, block_hash: Hash32) -> BlockHeader:
+        head_number = peer.head_info.block_number
+        if head_number - header.block_number > MAX_REORG_DEPTH:
+            # The peer claims to be far ahead of the header we requested
+            if self.headerdb.get_canonical_block_hash(header.block_number) == block_hash:
+                # Our node believes that the header at the reference hash is canonical,
+                # so treat the peer as malicious
+                raise BadLESResponse(
+                    "Peer %s sent empty code that did not match hash %s in account %s" % (
+                        peer,
+                        encode_hex(code_hash),
+                        encode_hex(address),
+                    )
+                )
+            else:
+                # our header isn't canonical, so treat the empty response as missing data
+                raise NoEligiblePeers(
+                    "Our best peer does not have the non-canonical header %s" % block_hash
+                )
+        elif head_number - header.block_number < 0:
+            # The peer claims to be behind the header we requested, but somehow served it to us.
+            # Odd, it might be a race condition. Treat as if there are no eligible peers for now.
+            raise NoEligiblePeers("Our best peer's head does include header %s" % block_hash)
+        else:
+            # The peer is ahead of the current block header, but only by a bit. It might be on
+            # an uncle, or we might be. So we can't tell the difference between missing and
+            # malicious. We don't want to aggressively drop this peer, so treat the code as missing.
+            raise NoEligiblePeers(
+                "Peer %s claims to be ahead of %s, but returned empty code with hash %s. "
+                "It is on number %d, maybe an uncle. Retry with an older block hash." % (
+                    peer,
+                    header,
+                    code_hash,
+                    head_number,
+                )
+            )
+
+    async def _get_block_header_by_hash(self, block_hash: Hash32, peer: LESPeer) -> BlockHeader:
+        """
+        A single attempt to get the block header from the given peer.
+
+        :raise BadLESResponse: if the peer replies with a header that has a different hash
+        """
         self.logger.debug("Fetching header %s from %s", encode_hex(block_hash), peer)
         request_id = gen_request_id()
         max_headers = 1
@@ -178,3 +338,27 @@ async def _get_proof(self,
         peer.sub_proto.send_get_proof(block_hash, account_key, key, from_level, request_id)
         reply = await self._wait_for_reply(request_id)
         return reply['proof']
+
+    async def _retry_on_bad_response(self, make_request_to_peer: Callable[[LESPeer], Any]) -> Any:
+        """
+        Make a call to a peer. If it behaves badly, drop it and retry with a different peer.
+
+        :param make_request_to_peer: an abstract call to a peer that may raise a BadLESResponse
+
+        :raise NoEligiblePeers: if no peers are available to fulfill the request
+        :raise TimeoutError: if an individual request or the overall process times out
+        """
+        for _ in range(MAX_REQUEST_ATTEMPTS):
+            try:
+                peer = cast(LESPeer, self.peer_pool.highest_td_peer)
+            except NoConnectedPeers as exc:
+                raise NoEligiblePeers() from exc
+
+            try:
+                return await make_request_to_peer(peer)
+            except BadLESResponse as exc:
+                self.logger.warn("Disconnecting from peer, because: %s", exc)
+                await peer.disconnect(DisconnectReason.subprotocol_error)
+                # reattempt after removing this peer from our pool
+
+        raise TimeoutError("Could not complete peer request in %d attempts" % MAX_REQUEST_ATTEMPTS)

p2p/peer.py

@@ -199,8 +199,11 @@ async def ensure_same_side_on_dao_fork(
         for start_block, vm_class in vm_configuration:
             if not issubclass(vm_class, HomesteadVM):
                 continue
+            elif not vm_class.support_dao_fork:
+                break
             elif start_block > vm_class.dao_fork_block_number:
-                continue
+                # VM comes after the fork, so stop checking
+                break
 
             fork_block = vm_class.dao_fork_block_number
             try:
@@ -213,12 +216,12 @@ async def ensure_same_side_on_dao_fork(
                 self.logger.info("Timed out waiting for DAO fork header from %s", self)
                 raise
             except ValidationError as e:
-                raise HandshakeFailure("Peer failed DAO fork check: {}".format(e))
+                raise HandshakeFailure("Peer failed DAO fork check retrieval: {}".format(e))
 
             try:
                 vm_class.validate_header(header, parent)
             except ValidationError as e:
-                raise HandshakeFailure("Peer failed DAO fork check: {}".format(e))
+                raise HandshakeFailure("Peer failed DAO fork check validation: {}".format(e))
 
     @abstractmethod
     async def _get_headers_at_chain_split(

p2p/service.py

@@ -1,14 +1,13 @@
 from abc import ABC, abstractmethod
 import asyncio
+import functools
 import logging
-from typing import (  # noqa: #401
+from typing import (
     Any,
-    Awaitable,
     Callable,
-    cast,
     List,
     Optional,
-    TypeVar,
+    cast,
 )
 
 from evm.utils.logging import TraceLogger
@@ -129,6 +128,25 @@ async def _cleanup(self) -> None:
         raise NotImplementedError()
 
 
+def service_timeout(timeout: int) -> Callable[..., Any]:
+    """
+    Decorator to time out a method call.
+
+    :param timeout: seconds to wait before raising a timeout exception
+
+    :raise asyncio.futures.TimeoutError: if the call is not complete before timeout seconds
+    """
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        @functools.wraps(func)
+        async def wrapped(service: BaseService, *args: Any, **kwargs: Any) -> Any:
+            return await service.wait(
+                func(service, *args, **kwargs),
+                timeout=timeout,
+            )
+        return wrapped
+    return decorator
+
+
 class EmptyService(BaseService):
     async def _run(self) -> None:
         pass