add snappy compression

eth_p2p.nim

@@ -34,7 +34,8 @@ proc newEthereumNode*(keys: KeyPair,
                       networkId: uint,
                       chain: AbstractChainDB,
                       clientId = "nim-eth-p2p/0.2.0", # TODO: read this value from nimble somehow
-                      addAllCapabilities = true): EthereumNode =
+                      addAllCapabilities = true,
+                      useSnappyCompression: bool = false): EthereumNode =
   new result
   result.keys = keys
   result.networkId = networkId
@@ -44,6 +45,12 @@ proc newEthereumNode*(keys: KeyPair,
   result.address = address
   result.connectionState = ConnectionState.None
 
+  when defined(useSnappy):
+    if useSnappyCompression:
+      result.protocolVersion = devp2pSnappyVersion
+    else:
+      result.protocolVersion = devp2pVersion
+
   if addAllCapabilities:
     for p in rlpxProtocols:
       result.addCapability(p)
@@ -141,4 +148,3 @@ proc randomPeer*(node: EthereumNode): Peer =
   for peer in node.peers:
     if i == peerIdx: return peer
     inc i
-

eth_p2p.nimble

@@ -18,14 +18,15 @@ requires "nim > 0.18.0",
          "chronicles",
          "asyncdispatch2",
          "eth_common",
-         "package_visible_types"
+         "package_visible_types",
+         "https://github.com/jangko/snappy"
 
-proc runTest(name: string, lang = "c") =
-  exec "nim " & lang & " -d:testing --experimental:ForLoopMacros -r tests/" & name
+proc runTest(name: string, defs = "", lang = "c") =
+  exec "nim " & lang & " " & defs & " -d:testing --experimental:ForLoopMacros -r tests/" & name
 
 task test, "Runs the test suite":
   runTest "testenode"
   runTest "tdiscovery"
   runTest "tserver"
+  runTest "tserver", "-d:useSnappy"
   runTest "all_tests"
-

eth_p2p/private/types.nim

@@ -19,6 +19,8 @@ packageTypes:
       protocolStates: seq[RootRef]
       discovery: DiscoveryProtocol
       peerPool*: PeerPool
+      when defined(useSnappy):
+        protocolVersion: uint
 
     Peer* = ref object
       transport: StreamTransport
@@ -31,6 +33,8 @@ packageTypes:
       protocolStates: seq[RootRef]
       outstandingRequests: seq[Deque[OutstandingRequest]]
       awaitedMessages: seq[FutureBase]
+      when defined(useSnappy):
+        snappyEnabled: bool
 
     OutstandingRequest = object
       id: int

eth_p2p/rlpx.nim

@@ -3,6 +3,11 @@ import
   chronicles, nimcrypto, asyncdispatch2, rlp, eth_common, eth_keys,
   private/types, kademlia, auth, rlpxcrypt, enode
 
+when defined(useSnappy):
+  import snappy
+  const
+    devp2pSnappyVersion* = 5
+
 logScope:
   topics = "rlpx"
 
@@ -257,7 +262,16 @@ proc linkSendFailureToReqFuture[S, R](sendFut: Future[S], resFut: Future[R]) =
 proc sendMsg*(peer: Peer, data: Bytes) {.async.} =
   trace "sending msg", peer, msg = getMsgName(peer, rlpFromBytes(data).read(int))
 
-  var cipherText = encryptMsg(data, peer.secretsState)
+  when defined(useSnappy):
+    var cipherText: Bytes
+    if peer.snappyEnabled:
+      let compressed = snappy.compress(data)
+      cipherText = encryptMsg(compressed, peer.secretsState)
+    else:
+      cipherText = encryptMsg(data, peer.secretsState)
+  else:
+    var cipherText = encryptMsg(data, peer.secretsState)
+
   try:
     discard await peer.transport.write(cipherText)
   except:
@@ -407,7 +421,17 @@ proc recvMsg*(peer: Peer): Future[tuple[msgId: int, msgData: Rlp]] {.async.} =
                                   "Cannot decrypt RLPx frame body")
 
   decryptedBytes.setLen(decryptedBytesCount)
-  var rlp = rlpFromBytes(decryptedBytes.toRange)
+
+  when defined(useSnappy):
+    var rlp: Rlp
+    if peer.network.protocolVersion == devp2pSnappyVersion:
+      decryptedBytes = snappy.uncompress(decryptedBytes)
+      if decryptedBytes.len == 0:
+        await peer.disconnectAndRaise(BreachOfProtocol,
+                                  "Snappy uncompress encountered malformed data")
+    rlp = rlpFromBytes(decryptedBytes.toRange)
+  else:
+    var rlp = rlpFromBytes(decryptedBytes.toRange)
 
   try:
     let msgid = rlp.read(int)
@@ -1173,7 +1197,11 @@ proc rlpxConnect*(node: EthereumNode, remote: Node): Future[Peer] {.async.} =
   try:
     result.transport = await connect(ta)
 
-    var handshake = newHandshake({Initiator, EIP8})
+    when defined(useSnappy):
+      var handshake = newHandshake({Initiator, EIP8}, int(node.protocolVersion))
+    else:
+      var handshake = newHandshake({Initiator, EIP8})
+
     handshake.host = node.keys
 
     var authMsg: array[AuthMessageMaxEIP8, byte]
@@ -1195,16 +1223,26 @@ proc rlpxConnect*(node: EthereumNode, remote: Node): Future[Peer] {.async.} =
       ret = handshake.decodeAckMessage(ackMsg)
     check ret
 
+    when defined(useSnappy):
+      result.snappyEnabled = handshake.version.int >= devp2pSnappyVersion
+
     initSecretState(handshake, ^authMsg, ackMsg, result)
 
     # if handshake.remoteHPubkey != remote.node.pubKey:
     #   raise newException(Exception, "Remote pubkey is wrong")
+    when defined(useSnappy):
+      asyncCheck result.hello(handshake.version,
+                              node.clientId,
+                              node.rlpxCapabilities,
+                              uint(node.address.tcpPort),
+                              node.keys.pubkey.getRaw())
 
-    asyncCheck result.hello(devp2pVersion,
-                            node.clientId,
-                            node.rlpxCapabilities,
-                            uint(node.address.tcpPort),
-                            node.keys.pubkey.getRaw())
+    else:
+      asyncCheck result.hello(devp2pVersion,
+                              node.clientId,
+                              node.rlpxCapabilities,
+                              uint(node.address.tcpPort),
+                              node.keys.pubkey.getRaw())
 
     var response = await result.waitSingleMsg(p2p.hello)
 
@@ -1246,6 +1284,7 @@ proc rlpxAccept*(node: EthereumNode,
   try:
     let initialSize = handshake.expectedLength
     var authMsg = newSeqOfCap[byte](1024)
+
     authMsg.setLen(initialSize)
     await transport.readExactly(addr authMsg[0], len(authMsg))
     var ret = handshake.decodeAuthMessage(authMsg)
@@ -1256,6 +1295,10 @@ proc rlpxAccept*(node: EthereumNode,
       ret = handshake.decodeAuthMessage(authMsg)
     check ret
 
+    when defined(useSnappy):
+      result.snappyEnabled = handshake.version >= devp2pSnappyVersion.uint
+      handshake.version = uint8(node.protocolVersion)
+
     var ackMsg: array[AckMessageMaxEIP8, byte]
     var ackMsgLen: int
     check handshake.ackMessage(ackMsg, ackMsgLen)
@@ -1264,9 +1307,18 @@ proc rlpxAccept*(node: EthereumNode,
     initSecretState(handshake, authMsg, ^ackMsg, result)
 
     let listenPort = transport.localAddress().port
-    await result.hello(devp2pVersion, node.clientId,
-                       node.rlpxCapabilities, listenPort.uint,
-                       node.keys.pubkey.getRaw())
+
+    when defined(useSnappy):
+      let peerProtocolVersion = if result.snappyEnabled: devp2pSnappyVersion.uint
+        else: devp2pVersion.uint
+
+      await result.hello(peerProtocolVersion, node.clientId,
+                        node.rlpxCapabilities, listenPort.uint,
+                        node.keys.pubkey.getRaw())
+    else:
+      await result.hello(devp2pVersion, node.clientId,
+                        node.rlpxCapabilities, listenPort.uint,
+                        node.keys.pubkey.getRaw())
 
     var response = await result.waitSingleMsg(p2p.hello)
     if not validatePubKeyInHello(response, handshake.remoteHPubkey):

tests/tserver.nim

@@ -89,10 +89,13 @@ template asyncTest(name, body: untyped) =
     proc scenario {.async.} = body
     waitFor scenario()
 
+import typetraits
+
 asyncTest "network with 3 peers using custom protocols":
+  const usecompression = defined(useSnappy)
   let localKeys = newKeyPair()
   let localAddress = localAddress(30303)
-  var localNode = newEthereumNode(localKeys, localAddress, 1, nil)
+  var localNode = newEthereumNode(localKeys, localAddress, 1, nil, useSnappyCompression = useCompression)
   localNode.initProtocolStates()
   localNode.startListening()
 
@@ -130,4 +133,3 @@ asyncTest "network with 3 peers using custom protocols":
     abcNetState.peers.len == 2
     "Alice" in abcNetState.peers
     "John Doe" in abcNetState.peers
-