diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,51 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [2.2.0] - 2026-11-04 + +### Changed + +- Breaking: `Peer` has a `pending` field, `wrapPeer` takes its `TVar Int`, + and `peer` takes the `Peer` that `wrapPeer` built instead of the busy + `TVar`. +- Messages from a peer apply backpressure. A peer may have at most + `maxPendingMessages` messages waiting for the node; after that the node + stops reading its socket until it catches up. Address, header, version, + verack, ping and pong messages are handled to completion before the next + one is taken. A fast peer could fill the node's unbounded queues before. +- Outgoing messages are logged by command at debug level, not in full. + +### Fixed + +- Bitcoin Cash headers sync again. This package now requires haskoin-core + 1.5.1, which evaluates the ASERT difficulty rule at the parent block. With + haskoin-core 1.5.0 the node stopped at mainnet block 975899 with + `Incorrect difficulty bits`, because that release checked the bits of each + header with a formula that used the header's own timestamp. +- Network magic and payload size are checked right after the message header, + before the payload is read. The size limit is the network's + `maxMessageSize` instead of a fixed 32 MiB. An oversized message used to be + logged and its payload then parsed as the next header. +- Peers that send more than 2000 headers or more than 1000 addresses in one + message are disconnected before the message is decoded. +- Address and header messages from a peer are ignored until it completes the + handshake. +- A second version or verack from a peer disconnects it. Each one used to + announce the peer as connected again. +- Relayed addresses that are not public unicast addresses (loopback, + private, link-local, multicast, documentation and similar ranges, or port + 0) are not added to the address pool. `isRoutable` is exported. +- A peer that does not answer a ping within the timeout is disconnected, even + if it keeps sending other messages. +- Only headers from the syncing peer reset its sync timeout. Another peer's + headers could keep a silent syncing peer from ever timing out. +- The peer manager rechecks `maxPeers` when it handles a queued connection, + so the connect loop can no longer go over the limit while the manager is + busy. +- Purging an old chain database with a column family deleted the same keys + from the default column family too. The purge now deletes keys as it walks + them instead of building a list of all of them in memory. + ## [2.1.2] - 2026-11-04 ### Fixed diff --git a/haskoin-node.cabal b/haskoin-node.cabal --- a/haskoin-node.cabal +++ b/haskoin-node.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: haskoin-node -version: 2.1.2 +version: 2.2.0 synopsis: P2P library for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -47,7 +47,7 @@ , data-default , fast-logger , hashable - , haskoin-core >=1.5.0 + , haskoin-core >=1.5.1 , mtl , network , nqe >=0.6.3 @@ -84,7 +84,7 @@ , data-default , fast-logger , hashable - , haskoin-core >=1.5.0 + , haskoin-core >=1.5.1 , haskoin-node , hspec , mtl diff --git a/src/Haskoin/Node.hs b/src/Haskoin/Node.hs --- a/src/Haskoin/Node.hs +++ b/src/Haskoin/Node.hs @@ -105,6 +105,11 @@ _ -> return () publish (ChainEvent event) output +-- | Route events from peers to the peer manager and chain. A message from a +-- peer is handled to completion before the peer may send another one past its +-- small allowance of pending messages (see 'maxPendingMessages'), so that the +-- node's queues cannot grow faster than it can work. Address and header +-- messages are only taken from peers that completed the handshake. peerEvents :: (MonadIO m) => Chain -> PeerMgr -> Inbox PeerEvent -> Publisher NodeEvent -> m () @@ -118,20 +123,26 @@ PeerMessage p msg -> do case msg of MVersion v -> - peerMgrVersion mgr p v + peerMgrVersion mgr p v >> peerMgrSync mgr MVerAck -> - peerMgrVerAck mgr p + peerMgrVerAck mgr p >> peerMgrSync mgr MPing (Ping n) -> - peerMgrPing mgr p n + peerMgrPing mgr p n >> peerMgrSync mgr MPong (Pong n) -> - peerMgrPong mgr p n + peerMgrPong mgr p n >> peerMgrSync mgr MAddr (Addr ns) -> - peerMgrAddrs mgr p (map snd ns) + whenOnline p $ peerMgrAddrs mgr p (map snd ns) >> peerMgrSync mgr MHeaders (Headers hs) -> - chainHeaders ch p (map fst hs) + whenOnline p $ chainHeaders ch p (map fst hs) >> chainSync ch _ -> return () ticklePeer mgr p + peerMessageDone p publish (PeerEvent event) output + where + whenOnline p act = + getOnlinePeer mgr p >>= \case + Just OnlinePeer {online = True} -> act + _ -> return () -- | Launch node process in the foreground. withNode :: diff --git a/src/Haskoin/Node/Chain.hs b/src/Haskoin/Node/Chain.hs --- a/src/Haskoin/Node/Chain.hs +++ b/src/Haskoin/Node/Chain.hs @@ -32,6 +32,7 @@ chainIsSynced, chainBlockMain, chainHeaders, + chainSync, ) where @@ -80,6 +81,7 @@ | ChainPeerConnected !Peer | ChainPeerDisconnected !Peer | ChainPing + | ChainSyncMsg !(Listen ()) -- | Events originating from chain syncing process. data ChainEvent @@ -245,7 +247,7 @@ ("Could not connect headers from peer " <> p.label) killPeer p Just done -> do - setLastReceived ch.state + setLastReceived ch.state p best <- runReaderT getBestBlockHeader ch.config when (pbest.header /= best.header) $ chainEvent ch.config (ChainBestBlock best) @@ -315,6 +317,8 @@ logDebug ch.config.logger ("Disconnected peer " <> p.label) finishPeer ch p syncNewPeer ch +chainConfigMessage _ (ChainSyncMsg r) = + atomically (r ()) chainConfigMessage ch ChainPing = do logDebug ch.config.logger "Internal clock event" let to = ch.config.timeout @@ -348,7 +352,7 @@ initChainDB :: (MonadIO m) => ChainConfig -> m () initChainDB cfg@ChainConfig {db, cf, net} = liftIO $ do ver <- retrieveCommon db cf ChainDataVersionKey - when (ver /= Just dataVersion) $ purgeChainDB cfg >>= writeBatch db + when (ver /= Just dataVersion) $ purgeChainDB cfg case cf of Nothing -> insert db ChainDataVersionKey dataVersion Just cf' -> insertCF db cf' ChainDataVersionKey dataVersion @@ -358,26 +362,29 @@ setBestBlockHeader (genesisNode net) -- | Purge database of elements having keys that may conflict with those used in --- this module. -purgeChainDB :: (MonadIO m) => ChainConfig -> m [R.BatchOp] +-- this module. Keys are deleted from the column family that holds the chain, +-- one at a time as the iterator walks them, so that memory use does not grow +-- with the size of the database. +purgeChainDB :: (MonadIO m) => ChainConfig -> m () purgeChainDB ChainConfig {db, cf} = liftIO $ do with_iter $ \it -> do R.iterSeek it (B.singleton 0x90) - recurse_delete it + go it where with_iter = case cf of Nothing -> R.withIter db Just cf' -> R.withIterCF db cf' - recurse_delete it = + del k = case cf of + Nothing -> R.delete db k + Just cf' -> R.deleteCF db cf' k + go it = R.iterKey it >>= \case Just k - | B.head k == 0x90 || B.head k == 0x91 -> do - case cf of - Nothing -> R.delete db k - Just cf' -> R.deleteCF db cf' k + | B.take 1 k `elem` [B.singleton 0x90, B.singleton 0x91] -> do + del k R.iterNext it - (R.Del k :) <$> recurse_delete it - _ -> return [] + go it + _ -> return () -- | Import a bunch of continuous headers. Returns 'True' if the number of -- headers is less than 2000, which means that there are no more headers to sync @@ -465,11 +472,15 @@ where z = "0000000000000000000000000000000000000000000000000000000000000000" --- | Set the time of last received data to now if a syncing peer is active. -setLastReceived :: (MonadIO m) => TVar ChainState -> m () -setLastReceived st = do +-- | Set the time of last received data to now if the given peer is the one +-- being synced against. Headers from other peers must not keep a silent +-- syncing peer from timing out. +setLastReceived :: (MonadIO m) => TVar ChainState -> Peer -> m () +setLastReceived st p = do now <- liftIO getCurrentTime - let f ChainSync {..} = ChainSync {timestamp = now, ..} + let f c@ChainSync {..} + | peer == p = ChainSync {timestamp = now, ..} + | otherwise = c atomically (modifyTVar st (\s -> s {syncing = f <$> s.syncing})) -- | Add a new peer to the queue of peers to sync against. @@ -608,3 +619,7 @@ chainHeaders :: (MonadIO m) => Chain -> Peer -> [BlockHeader] -> m () chainHeaders ch p hs = ChainHeaders p hs `send` ch.mailbox + +-- | Wait until the chain process has handled every message sent to it before. +chainSync :: (MonadIO m) => Chain -> m () +chainSync ch = ChainSyncMsg `query` ch.mailbox diff --git a/src/Haskoin/Node/Peer.hs b/src/Haskoin/Node/Peer.hs --- a/src/Haskoin/Node/Peer.hs +++ b/src/Haskoin/Node/Peer.hs @@ -28,6 +28,10 @@ getBusy, setBusy, setFree, + peerMessageDone, + maxPendingMessages, + maxHeadersCount, + maxAddrCount, ) where @@ -41,10 +45,10 @@ import Data.Function (on) import Data.List (union) import Data.Maybe (isJust) -import Data.Serialize (decode, runGet, runPut) +import Data.Serialize (decode, get, runGet, runPut) import Data.String.Conversions (cs) import Data.Text (Text) -import Data.Word (Word32) +import Data.Word (Word32, Word64) import Haskoin import Haskoin.Node.Logger import Haskoin.Node.Loop (loopForever) @@ -78,7 +82,9 @@ { mailbox :: !(Mailbox PeerMessage), pub :: !(Publisher PeerEvent), label :: !Text, - busy :: !(TVar Bool) + busy :: !(TVar Bool), + -- | messages from this peer that the node has not finished handling + pending :: !(TVar Int) } instance Eq Peer where @@ -95,44 +101,73 @@ wrapPeer :: PeerConfig -> TVar Bool -> + TVar Int -> Mailbox PeerMessage -> Peer -wrapPeer cfg busy mbox = +wrapPeer cfg busy pending mbox = Peer { mailbox = mbox, pub = cfg.pub, label = cfg.label, - busy = busy + busy = busy, + pending = pending } --- | Run peer process in current thread. +-- | Most messages from one peer that may wait for the node to handle them. +-- When a peer gets this far ahead, the node stops reading from its socket +-- until it catches up, so that a peer that sends faster than the node can +-- handle fills its own TCP window instead of the node's memory. +maxPendingMessages :: Int +maxPendingMessages = 16 + +-- | Most block headers a @headers@ message may carry. Bitcoin Core never +-- sends more and disconnects peers that do. +maxHeadersCount :: Word64 +maxHeadersCount = 2000 + +-- | Most addresses an @addr@ or @addrv2@ message may carry. +maxAddrCount :: Word64 +maxAddrCount = 1000 + +-- | Tell the peer process that the node has finished handling one of the +-- messages that it published, so that it may read another one. +peerMessageDone :: (MonadIO m) => Peer -> m () +peerMessageDone p = atomically $ modifyTVar' p.pending (max 0 . subtract 1) + +-- | Run peer process in current thread. The 'Peer' must be the one that +-- 'wrapPeer' built with the mailbox of this inbox. peer :: (MonadUnliftIO m) => PeerConfig -> - TVar Bool -> + Peer -> Inbox PeerMessage -> m () -peer cfg@PeerConfig {label, net, connect, pub, logger} busy inbox = do - let p = wrapPeer cfg busy (inboxToMailbox inbox) - withRunInIO $ \run -> connect (run . peer_session p) +peer cfg@PeerConfig {label, net, connect, pub, logger} p inbox = + withRunInIO $ \run -> connect (run . peer_session) where go = do logDebug logger $ label <> " awaiting event..." msg <- receive inbox dispatchMessage cfg msg >>= bool (return ()) go - peer_session p ad = do + peer_session ad = do let ins = transPipe liftIO ad.inboundConduit ons = transPipe liftIO ad.outboundConduit src = runConduit $ ins .| inPeerConduit cfg - .| mapM_C (send_msg p) + .| mapM_C send_msg snk = outPeerConduit net .| ons withAsync src $ \as -> do link as runConduit (go .| snk) - send_msg p msg = publish (PeerMessage p msg) pub + -- Wait until the node has room for another message from this peer. + send_msg msg = do + atomically $ do + n <- readTVar p.pending + checkSTM (n < maxPendingMessages) + writeTVar p.pending (n + 1) + publish (PeerMessage p msg) pub -- | Internal function to dispatch peer messages. dispatchMessage :: @@ -141,48 +176,71 @@ PeerMessage -> ConduitT i Message m Bool dispatchMessage PeerConfig {label, logger} (SendMessage msg) = do - logDebug logger (label <> " sending: " <> cs (show msg)) + logDebug logger (label <> " sending cmd " <> cs (show (msgType msg))) yield msg return True dispatchMessage PeerConfig {label, logger} KillPeer = do logInfo logger (label <> " disconnecting") return False --- | Internal conduit to parse messages coming from peer. +-- | Internal conduit to parse messages coming from peer. The header is checked +-- before the payload is read, so that a peer cannot make the node buffer a +-- payload that it will reject anyway. A peer that sends anything invalid is +-- disconnected. inPeerConduit :: (MonadIO m) => PeerConfig -> ConduitT ByteString Message m () -inPeerConduit pc@PeerConfig {label, net, logger} = loopForever $ do +inPeerConduit PeerConfig {label, net, logger} = loopForever $ do logDebug logger (label <> " awaiting message...") x <- takeCE 24 .| foldC when (B.null x) $ do logError logger (label <> " sent empty message header") error "Peer sent empty message header" - case decode x of - Left e -> do - logError logger (label <> " sent invalid message header") - error "Peer sent invalid message header" - Right (MessageHeader _ cmd len _) - | len > 32 * 2 ^ (20 :: Int) -> - logWarn logger $ - label - <> " wants to send " - <> cs (show len) - <> " bytes (too large) for cmd " - <> cs (show cmd) - | otherwise -> do - logDebug logger (label <> " sent cmd " <> cs (show cmd)) - y <- takeCE (fromIntegral len) .| foldC - case runGet (getMessage net) $ x `B.append` y of - Left e -> do - logError - logger - (label <> " sent invalid payload for cmd " <> cs (show cmd)) - error "Peer sent invalid payload" - Right msg -> do - logDebug - logger - (label <> " sent payload for cmd " <> cs (show cmd)) - yield msg + MessageHeader mgc cmd len _ <- case decode x of + Left _ -> bad "invalid message header" + Right h -> return h + when (mgc /= net.magic) $ + bad ("wrong network magic " <> cs (show mgc)) + when (len > net.maxMessageSize) $ + bad + ( "too large payload of " + <> cs (show len) + <> " bytes for cmd " + <> cs (show cmd) + ) + logDebug logger (label <> " sent cmd " <> cs (show cmd)) + y <- takeCE (fromIntegral len) .| foldC + when (B.length y /= fromIntegral len) $ + bad ("truncated payload for cmd " <> cs (show cmd)) + forM_ (countLimit cmd) $ \limit -> + case runGet getCount y of + Right n + | n > limit -> + bad + ( "too many items (" + <> cs (show n) + <> ") in cmd " + <> cs (show cmd) + ) + _ -> return () + case runGet (getMessage net) $ x `B.append` y of + Left _ -> bad ("invalid payload for cmd " <> cs (show cmd)) + Right msg -> do + logDebug + logger + (label <> " sent payload for cmd " <> cs (show cmd)) + yield msg + where + bad :: (MonadIO m) => Text -> m a + bad e = do + logError logger (label <> " sent " <> e) + error ("Peer sent " <> cs e) + getCount = do + VarInt n <- get + return n + countLimit MCHeaders = Just maxHeadersCount + countLimit MCAddr = Just maxAddrCount + countLimit MCAddrV2 = Just maxAddrCount + countLimit _ = Nothing -- | Outgoing peer conduit to serialize and send messages. outPeerConduit :: (Monad m) => Network -> ConduitT Message ByteString m () diff --git a/src/Haskoin/Node/PeerMgr.hs b/src/Haskoin/Node/PeerMgr.hs --- a/src/Haskoin/Node/PeerMgr.hs +++ b/src/Haskoin/Node/PeerMgr.hs @@ -23,6 +23,7 @@ peerMgrPong, peerMgrAddrs, peerMgrVerAck, + peerMgrSync, getPeers, getOnlinePeer, ticklePeer, @@ -30,6 +31,7 @@ myVersion, toSockAddr, toHostService, + isRoutable, ) where @@ -45,7 +47,7 @@ import Data.String.Conversions (cs) import Data.Time.Clock import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds) -import Data.Word (Word32, Word64) +import Data.Word (Word32, Word64, Word8) import Haskoin import Haskoin.Node.Logger import Haskoin.Node.Loop (loopForever) @@ -88,6 +90,7 @@ | PeerPing !Peer !Word64 | PeerPong !Peer !Word64 | PeerAddrs !Peer ![NetworkAddress] + | Sync !(Listen ()) -- | Data structure representing an online peer. data OnlinePeer = OnlinePeer @@ -222,18 +225,32 @@ Nothing -> do logWarn mgr.config.logger - ("Received verack from unknown peer " <> p.label) + ("Verack rejected for peer " <> p.label) killPeer p dispatch mgr (PeerAddrs p nas) - | mgr.config.discover = do - let sas = map (hostToSockAddr . (.address)) nas - len = length sas - logDebug - mgr.config.logger - ("Received " <> cs (show len) <> " addresses from peer " <> p.label) - forM_ sas (newPeer mgr) + | mgr.config.discover = + getOnlinePeer mgr p >>= \case + Just OnlinePeer {online = True} -> do + let sas = filter isRoutable $ map (hostToSockAddr . (.address)) nas + logDebug + mgr.config.logger + ( "Received " + <> cs (show (length nas)) + <> " addresses from peer " + <> p.label + <> ", " + <> cs (show (length sas)) + <> " routable" + ) + forM_ sas (newPeer mgr) + _ -> + logDebug + mgr.config.logger + ("Ignoring addresses from peer before handshake " <> p.label) | otherwise = logDebug mgr.config.logger ("Ignoring addresses from peer " <> p.label) +dispatch _ (Sync r) = + atomically (r ()) dispatch mgr (PeerPong p n) = do logDebug mgr.config.logger @@ -268,7 +285,9 @@ Just o -> do now <- liftIO getCurrentTime let expired = mgr.config.maxPeerLife `addUTCTime` o.connected - let deadline = mgr.config.timeout `addUTCTime` o.tickled + deadline = mgr.config.timeout `addUTCTime` o.tickled + -- Other traffic must not keep a peer that ignores our ping. + pingDeadline = (\(t, _) -> mgr.config.timeout `addUTCTime` t) <$> o.ping if | now > expired -> do logError mgr.config.logger ("Peer " <> p.label <> " is too old") @@ -276,6 +295,9 @@ | now > deadline -> do logError mgr.config.logger ("Peer " <> p.label <> " timed out") killPeer p + | maybe False (now >) pingDeadline -> do + logError mgr.config.logger ("Peer " <> p.label <> " did not answer ping") + killPeer p | isNothing o.ping -> sendPing mgr p | otherwise -> return () Nothing -> do @@ -344,7 +366,15 @@ connectPeer :: (MonadUnliftIO m) => PeerMgr -> SockAddr -> m () connectPeer mgr sa = do + n <- length <$> getOnlinePeers mgr atomically (findPeerAddress mgr.peers sa) >>= \case + _ + | n >= mgr.config.maxPeers -> + -- The connect loop may have queued this before the last free slot + -- was taken. + logDebug + mgr.config.logger + ("Not connecting to peer " <> cs (show sa) <> ": no free slots") Just _ -> logWarn mgr.config.logger ("Duplicate connection to peer " <> cs (show sa)) Nothing -> do @@ -366,9 +396,10 @@ logger = withSource "Peer" mgr.config.logger } busy <- newTVarIO False - let p = wrapPeer pc busy mailbox + pending <- newTVarIO 0 + let p = wrapPeer pc busy pending mailbox a <- withRunInIO $ \io -> - mgr.supervisor `addChild` io (launch pc busy inbox p) + mgr.supervisor `addChild` io (launch pc inbox p) MVersion ver `sendMessage` p atomically $ insertPeer @@ -390,8 +421,8 @@ srv net | net.segWit = 8 | otherwise = 0 - launch pc busy inbox p = - withPeerLoop mgr p (\a -> link a >> peer pc busy inbox) + launch pc inbox p = + withPeerLoop mgr p (\a -> link a >> peer pc p inbox) withPeerLoop :: (MonadUnliftIO m) => PeerMgr -> Peer -> (Async a -> m a) -> m a withPeerLoop mgr p = @@ -452,6 +483,8 @@ else findPeer b p >>= \case Nothing -> return Nothing -- peer not found + Just OnlinePeer {version = Just _} -> + return Nothing -- duplicate version Just o -> do let n = o {version = Just v, online = o.verack} insertPeer b n @@ -460,6 +493,7 @@ setPeerVerAck :: TVar [OnlinePeer] -> Peer -> STM (Maybe OnlinePeer) setPeerVerAck b p = findPeer b p >>= \case + Just OnlinePeer {verack = True} -> return Nothing -- duplicate verack Just o -> do let o' = o {verack = True, online = isJust o.version} insertPeer b o' @@ -519,6 +553,54 @@ peerMgrAddrs :: (MonadIO m) => PeerMgr -> Peer -> [NetworkAddress] -> m () peerMgrAddrs mgr p addrs = PeerAddrs p addrs `send` mgr.mailbox + +-- | Wait until the peer manager has handled every message sent to it before. +peerMgrSync :: (MonadIO m) => PeerMgr -> m () +peerMgrSync mgr = Sync `query` mgr.mailbox + +-- | Whether an address that a peer relayed is worth connecting to: a public +-- unicast address with a port. Loopback, private, link-local, shared, +-- documentation, multicast and reserved ranges are refused, so that a peer +-- cannot make the node connect to services on its own host or network. +-- Static peers in the configuration are not checked. +isRoutable :: SockAddr -> Bool +isRoutable (SockAddrInet port h) = + port /= 0 && routable4 (hostAddressToTuple h) +isRoutable (SockAddrInet6 port _ h _) = + port /= 0 && case hostAddress6ToTuple h of + (0, 0, 0, 0, 0, 0xffff, hi, lo) -> + routable4 + ( fromIntegral (hi `div` 256), + fromIntegral (hi `mod` 256), + fromIntegral (lo `div` 256), + fromIntegral (lo `mod` 256) + ) + (a, b, _, _, _, _, _, _) + | a == 0 -> False -- unspecified, loopback, IPv4-compatible + | a .&. 0xfe00 == 0xfc00 -> False -- unique local + | a .&. 0xffc0 == 0xfe80 -> False -- link-local + | a .&. 0xffc0 == 0xfec0 -> False -- site-local + | a .&. 0xff00 == 0xff00 -> False -- multicast + | a == 0x2001 && b == 0x0db8 -> False -- documentation + | otherwise -> True +isRoutable _ = False + +routable4 :: (Word8, Word8, Word8, Word8) -> Bool +routable4 (a, b, c, _) + | a == 0 = False -- this network + | a == 10 = False -- private + | a == 100 && b .&. 0xc0 == 64 = False -- shared address space + | a == 127 = False -- loopback + | a == 169 && b == 254 = False -- link-local + | a == 172 && b .&. 0xf0 == 16 = False -- private + | a == 192 && b == 0 && c == 0 = False -- protocol assignments + | a == 192 && b == 0 && c == 2 = False -- documentation + | a == 192 && b == 168 = False -- private + | a == 198 && b .&. 0xfe == 18 = False -- benchmarking + | a == 198 && b == 51 && c == 100 = False -- documentation + | a == 203 && b == 0 && c == 113 = False -- documentation + | a >= 224 = False -- multicast and reserved + | otherwise = True toHostService :: String -> (Maybe String, Maybe String) toHostService str = diff --git a/test/Haskoin/NodeSpec.hs b/test/Haskoin/NodeSpec.hs --- a/test/Haskoin/NodeSpec.hs +++ b/test/Haskoin/NodeSpec.hs @@ -6,6 +6,7 @@ {-# LANGUAGE OverloadedRecordDot #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TupleSections #-} {-# LANGUAGE NoFieldSelectors #-} module Haskoin.NodeSpec (spec) where @@ -22,12 +23,13 @@ import Data.List (find) import Data.Maybe (isJust, listToMaybe, mapMaybe) import Data.Serialize (decode, get, runGet, runPut) +import Data.Time.Clock (NominalDiffTime) import Data.Time.Clock.POSIX (getPOSIXTime) import Database.RocksDB qualified as R import Haskoin import Haskoin.Node import NQE -import Network.Socket (AddrInfo (addrAddress), SockAddr (..)) +import Network.Socket (AddrInfo (addrAddress), SockAddr (..), tupleToHostAddress, tupleToHostAddress6) import System.Random (randomIO) import Test.Hspec import Test.Hspec.QuickCheck @@ -38,19 +40,40 @@ { testMgr :: PeerMgr, testChain :: Chain, nodeEvents :: Inbox NodeEvent, + testDB :: R.DB, -- | how many getheaders messages the mock peer has received getHeadersCount :: TVar Int } --- | Connect to a mock peer that serves these blocks. +-- | What the mock peer sends back for each message: a message to serialize +-- for the test network, or raw bytes. +type React = Message -> [Either ByteString Message] + +-- | How to set up a test node. +data TestOpts = TestOpts + { react :: React, + -- | runs on the database before the node starts + setup :: R.DB -> IO (), + timeout :: NominalDiffTime + } + +defOpts :: [Block] -> TestOpts +defOpts blocks = + TestOpts + { react = map Right . mockPeerReact blocks, + setup = \_ -> return (), + timeout = 120 + } + +-- | Connect to a mock peer that reacts to messages like this. dummyPeerConnect :: - [Block] -> + React -> TVar Int -> Network -> NetworkAddress -> SockAddr -> WithConnection -dummyPeerConnect blocks count net ad sa f = do +dummyPeerConnect react count net ad sa f = do r <- newInbox s <- newInbox let s' = inboxToMailbox s @@ -70,10 +93,10 @@ forever (receive r >>= yield) .| inc .| iterMC count_getheaders - .| concatMapC (mockPeerReact blocks) + .| concatMapC react .| outc .| awaitForever (`send` s) - outc = mapMC $ \msg -> return $ runPut (putMessage net msg) + outc = mapMC $ return . either id (runPut . putMessage net) count_getheaders = \case MGetHeaders _ -> atomically (modifyTVar' count (+ 1)) _ -> return () @@ -149,6 +172,114 @@ testMerkle b = buildMerkleRoot (ths b) `shouldBe` Right b.header.merkle testMerkle b1 testMerkle b2 + describe "untrusted peers" $ do + it "imports a full batch of 2000 headers" $ + withTestNodeOpts + (defOpts allBlocks) {react = reactAfterVerAck [Right (manyHeaders net 2000)]} + net + "full-batch" + $ \TestNode {..} -> do + bn <- + receiveMatch nodeEvents $ \case + ChainEvent (ChainBestBlock bn) | bn.height > 0 -> Just bn + _ -> Nothing + bn.height `shouldBe` 2000 + it "disconnects a peer that sends more than 2000 headers" $ + withTestNodeOpts + (defOpts allBlocks) {react = reactAfterVerAck [Right (manyHeaders net 2001)]} + net + "too-many-headers" + $ \TestNode {..} -> do + p <- waitForPeer nodeEvents + waitForDisconnect nodeEvents p + (.height) <$> chainGetBest testChain `shouldReturn` 0 + it "disconnects a peer that sends more than 1000 addresses" $ do + let na = NetworkAddress 0 (sockToHostAddress (SockAddrInet 8333 0x08080808)) + addrs = MAddr (Addr (replicate 1001 (0, na))) + withTestNodeOpts + (defOpts allBlocks) {react = reactAfterVerAck [Right addrs]} + net + "too-many-addrs" + $ \TestNode {..} -> do + p <- waitForPeer nodeEvents + waitForDisconnect nodeEvents p + it "disconnects a peer that sends the wrong network magic" $ do + let bad = runPut (putMessage btc (MPing (Ping 1))) + withTestNodeOpts + (defOpts allBlocks) {react = reactAfterVerAck [Left bad]} + net + "wrong-magic" + $ \TestNode {..} -> do + p <- waitForPeer nodeEvents + waitForDisconnect nodeEvents p + it "disconnects a peer that sends a second verack" $ + withTestNodeOpts + (defOpts allBlocks) {react = reactAfterVerAck [Right MVerAck]} + net + "duplicate-verack" + $ \TestNode {..} -> do + p <- waitForPeer nodeEvents + waitForDisconnect nodeEvents p + it "ignores headers sent before the handshake" $ do + let hs = MHeaders (Headers [(b.header, VarInt 0) | b <- allBlocks]) + react = \case + MVersion _ -> [Right hs, Right MVerAck] + MPing (Ping n) -> [Right (MPong (Pong n))] + _ -> [] + withTestNodeOpts (defOpts allBlocks) {react = react} net "early-headers" $ + \TestNode {..} -> do + _ <- waitForPeer nodeEvents + threadDelay (500 * 1000) + (.height) <$> chainGetBest testChain `shouldReturn` 0 + it "disconnects a peer that chatters instead of answering a ping" $ do + -- Answer the node's ping with a ping of our own, and its pong with + -- another ping, so that the peer is never silent. + let react = \case + MVersion _ -> [Right MVerAck] + MPing _ -> [Right (MPing (Ping 1))] + MPong _ -> [Right (MPing (Ping 1))] + -- end the header sync so its own timeout does not apply + MGetHeaders _ -> [Right (MHeaders (Headers []))] + _ -> [] + withTestNodeOpts + (defOpts allBlocks) {react = react, timeout = 2} + net + "ignored-ping" + $ \TestNode {..} -> do + p <- waitForPeer nodeEvents + waitForDisconnect nodeEvents p + it "does not connect to private addresses that peers relay" $ do + let v4 a b c d p = SockAddrInet p (tupleToHostAddress (a, b, c, d)) + v6 t p = SockAddrInet6 p 0 (tupleToHostAddress6 t) 0 + isRoutable (v4 8 8 8 8 8333) `shouldBe` True + isRoutable (v4 8 8 8 8 0) `shouldBe` False + isRoutable (v4 127 0 0 1 8333) `shouldBe` False + isRoutable (v4 10 1 2 3 8333) `shouldBe` False + isRoutable (v4 172 16 0 1 8333) `shouldBe` False + isRoutable (v4 172 32 0 1 8333) `shouldBe` True + isRoutable (v4 192 168 1 1 8333) `shouldBe` False + isRoutable (v4 169 254 1 1 8333) `shouldBe` False + isRoutable (v4 100 64 0 1 8333) `shouldBe` False + isRoutable (v4 224 0 0 1 8333) `shouldBe` False + isRoutable (v6 (0x2a01, 0x4f8, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` True + isRoutable (v6 (0, 0, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` False + isRoutable (v6 (0xfd00, 0, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` False + isRoutable (v6 (0xfe80, 0, 0, 0, 0, 0, 0, 1) 8333) `shouldBe` False + isRoutable (v6 (0, 0, 0, 0, 0, 0xffff, 0x7f00, 1) 8333) `shouldBe` False + isRoutable (v6 (0, 0, 0, 0, 0, 0xffff, 0x0808, 0x0808) 8333) `shouldBe` True + isRoutable (hostToSockAddr (sockToHostAddress (v4 127 0 0 1 8333))) + `shouldBe` False + describe "chain database" $ do + it "leaves the default column family alone when purging" $ do + -- The same key in both families: the purge must delete it only from + -- the family that holds the chain. + let k = B.pack [0x90, 1, 2, 3] + setup db = do + R.put db k "keep" + forM_ (R.columnFamilies db) $ \cf -> R.putCF db cf k "stale" + withTestNodeOpts (defOpts allBlocks) {setup = setup} net "purge-cf" $ + \TestNode {..} -> do + R.get testDB k `shouldReturn` Just "keep" describe "chain on test network" $ do it "stops asking a peer that has no more headers" $ -- The peer answers getheaders with an empty headers message, as a @@ -195,6 +326,33 @@ forM_ (zip ps hs) $ \(p, h) -> headerHash p.header `shouldBe` h +-- | Wait for the node to disconnect from this peer, failing after a while. +waitForDisconnect :: Inbox NodeEvent -> Peer -> IO () +waitForDisconnect inbox p = + timeout (20 * 1000 * 1000) wait >>= \case + Just () -> return () + Nothing -> expectationFailure "peer was not disconnected" + where + wait = + receiveMatch inbox $ \case + PeerEvent (PeerDisconnected p') | p == p' -> Just () + _ -> Nothing + +-- | Mock peer that answers the handshake and pings, and sends these right +-- after its version ack. +reactAfterVerAck :: [Either ByteString Message] -> React +reactAfterVerAck extra = \case + MVersion _ -> Right MVerAck : extra + MPing (Ping n) -> [Right (MPong (Pong n))] + _ -> [] + +-- | Headers message with a valid chain of this many headers on top of +-- genesis. +manyHeaders :: Network -> Int -> Message +manyHeaders net n = + MHeaders . Headers $ + map (,VarInt 0) (appendBlocks net 0 net.genesisHeader n) + waitForPeer :: (MonadIO m) => Inbox NodeEvent -> m Peer waitForPeer inbox = receiveMatch inbox $ \case @@ -213,12 +371,18 @@ withTestNodeBlocks :: (MonadUnliftIO m) => [Block] -> Network -> String -> (TestNode -> m a) -> m a -withTestNodeBlocks blocks net str f = flip runContT return $ do +withTestNodeBlocks blocks = withTestNodeOpts (defOpts blocks) + +withTestNodeOpts :: + (MonadUnliftIO m) => + TestOpts -> Network -> String -> (TestNode -> m a) -> m a +withTestNodeOpts opts net str f = flip runContT return $ do w <- ContT $ withSystemTempDirectory ("haskoin-node-test-" <> str <> "-") count <- newTVarIO 0 pub <- ContT withPublisher sub <- ContT $ withSubscription pub db <- ContT $ withLifted (R.withDBCF w cfg cols) + liftIO (opts.setup db) let ad = NetworkAddress nodeNetwork @@ -238,9 +402,9 @@ address = na, net = net, pub = pub, - timeout = 120, + timeout = opts.timeout, maxPeerLife = 48 * 3600, - connect = dummyPeerConnect blocks count net ad, + connect = dummyPeerConnect opts.react count net ad, logger = Logger {level = LevelError, source = "", push = \_ -> return ()} } Node mgr ch <- ContT $ withNode cfg' @@ -250,6 +414,7 @@ { testMgr = mgr, testChain = ch, nodeEvents = sub, + testDB = db, getHeadersCount = count } where