add log filters

Author: chirag-parmar

nimbus_verified_proxy/rpc/receipts.nim

@@ -90,15 +90,26 @@ proc getReceipts*(
 
   return ok(rxs.get())
 
-proc getLogs*(
-    vp: VerifiedRpcProxy, filterOptions: FilterOptions
-): Future[Result[seq[LogObject], string]] {.async.} =
-  let logObjs =
-    try:
-      await vp.rpcClient.eth_getLogs(filterOptions)
-    except CatchableError as e:
-      return err(e.msg)
+proc verifyFilterBoundaries*(filter: FilterOptions, logObjs: seq[LogObject]): Result[bool, string] =
+  let
+    fromBlock = filter.fromBlock.get(BlockTag(kind:BlockIdentifierKind.bidAlias, alias: "latest"))
+    toBlock = filter.toBlock.get(BlockTag(kind:BlockIdentifierKind.bidAlias, alias: "latest"))
 
+    bottom = if fromBlock.kind == BlockIdentifierKind.bidNumber: fromBlock.number
+           else: return err("Cannot verify boundaries for block tags in 'fromBlock' field")
+    top = if toBlock.kind == BlockIdentifierKind.bidNumber: toBlock.number
+         else: return err("Cannot verify boundaries for block tags in 'toBlock' field")
+
+  for lg in logObjs:
+    if lg.blockNumber.isSome:
+      if lg.blockNumber.get < bottom or lg.blockNumber.get > top:
+        return ok(false)
+
+  return ok(true) 
+
+proc verifyLogs*(
+    vp: VerifiedRpcProxy, logObjs: seq[LogObject]
+): Future[Result[seq[LogObject], string]] {.async.} =
   var res = newSeq[LogObject]()
 
   # store block hashes contains the logs so that we can batch receipt requests
@@ -113,8 +124,8 @@ proc getLogs*(
         for rx in rxs:
           for rxLog in rx.logs:
             # only add verified logs
-            if rxLog.address == lg.address and rxLog.data == lg.data and
-                rxLog.topics == lg.topics:
+            if rxLog.address != lg.address or rxLog.data != lg.data or
+                rxLog.topics != lg.topics:
               res.add(lg)
       else:
         let
@@ -123,8 +134,8 @@ proc getLogs*(
 
         # only add verified logs
         for rxLog in rx.logs:
-          if rxLog.address == lg.address and rxLog.data == lg.data and
-              rxLog.topics == lg.topics:
+          if rxLog.address != lg.address or rxLog.data != lg.data or
+              rxLog.topics != lg.topics:
             res.add(lg)
 
   if res.len == 0:

nimbus_verified_proxy/rpc/rpc_eth_api.nim

@@ -204,6 +204,7 @@ proc installEthApiHandlers*(vp: VerifiedRpcProxy) =
         await vp.rpcClient.eth_getTransactionByHash(txHash)
       except CatchableError as e:
         raise newException(ValueError, e.msg)
+
     if tx.hash != txHash:
       raise newException(
         ValueError,
@@ -238,10 +239,93 @@ proc installEthApiHandlers*(vp: VerifiedRpcProxy) =
     raise newException(ValueError, "receipt couldn't be verified")
 
   vp.proxy.rpc("eth_getLogs") do(filterOptions: FilterOptions) -> seq[LogObject]:
-    (await vp.getLogs(filterOptions)).valueOr:
+    let 
+      logObjs = 
+        try:
+          await vp.rpcClient.eth_getLogs(filterOptions)
+        except CatchableError as e:
+          raise newException(ValueError, e.msg)
+
+      boundsCheck = verifyFilterBoundaries(filterOptions, logObjs).valueOr:
+        raise newException(ValueError, error)
+
+    if not boundsCheck:
+      raise newException(ValueError, "Logs out of filter block range")
+
+    let verifiedLogs = (await vp.verifyLogs(logObjs)).valueOr:
+      raise newException(ValueError, error)
+
+    return verifiedLogs
+
+  vp.proxy.rpc("eth_newFilter") do(filterOptions: FilterOptions) -> int:
+    let 
+      hexId = 
+        try:
+          await vp.rpcClient.eth_newFilter(filterOptions)
+        except CatchableError as e:
+          raise newException(ValueError, e.msg)
+      id = fromHex[int](hexId)
+
+    vp.filterStore[id] = filterOptions
+    return id
+
+  vp.proxy.rpc("eth_uninstallFilter") do(filterId: int) -> bool:
+    let status = 
+      try:
+        await vp.rpcClient.eth_uninstallFilter("0x" & toHex(filterId))
+      except CatchableError as e:
+        raise newException(ValueError, e.msg)
+
+    if status and filterId in vp.filterStore:
+      vp.filterStore.del(filterId)
+
+    return status
+
+  vp.proxy.rpc("eth_getFilterLogs") do(filterId: int) -> seq[LogObject]:
+    if filterId notin vp.filterStore:
+      raise newException(ValueError, "Filter doesn't exist")
+
+    let 
+      logObjs = 
+        try:
+          # use locally stored filter and get logs
+          await vp.rpcClient.eth_getLogs(vp.filterStore[filterId])
+        except CatchableError as e:
+          raise newException(ValueError, e.msg)
+
+      boundsCheck = verifyFilterBoundaries(vp.filterStore[filterId], logObjs).valueOr:
+        raise newException(ValueError, error)
+
+    if not boundsCheck:
+      raise newException(ValueError, "Logs out of filter block range")
+
+    let verifiedLogs = (await vp.verifyLogs(logObjs)).valueOr:
       raise newException(ValueError, error)
 
-  # TODO:
+    return verifiedLogs
+
+  vp.proxy.rpc("eth_getFilterChanges") do(filterId: int) -> seq[LogObject]:
+    if filterId notin vp.filterStore:
+      raise newException(ValueError, "Filter doesn't exist")
+
+    let 
+      logObjs = 
+        try:
+          await vp.rpcClient.eth_getFilterChanges("0x" & toHex(filterId))
+        except CatchableError as e:
+          raise newException(ValueError, e.msg)
+
+      boundsCheck = verifyFilterBoundaries(vp.filterStore[filterId], logObjs).valueOr:
+        raise newException(ValueError, error)
+
+    if not boundsCheck:
+      raise newException(ValueError, "Logs out of filter block range")
+
+    let verifiedLogs = (await vp.verifyLogs(logObjs)).valueOr:
+      raise newException(ValueError, error)
+
+    return verifiedLogs
+
   # Following methods are forwarded directly to the web3 provider and therefore
   # are not validated in any way.
   vp.proxy.registerProxyMethod("net_version")

nimbus_verified_proxy/types.nim

@@ -6,6 +6,7 @@
 # at your option. This file may not be copied, modified, or distributed except according to those terms.
 
 import
+  std/tables,
   json_rpc/[rpcproxy, rpcclient],
   stint,
   ./header_store,
@@ -17,6 +18,7 @@ type
     evm*: AsyncEvm
     proxy*: RpcProxy
     headerStore*: HeaderStore
+    filterStore*: Table[int, FilterOptions]
     chainId*: UInt256
     maxBlockWalk*: uint64
 
@@ -33,5 +35,9 @@ proc init*(
     maxBlockWalk: uint64,
 ): T =
   VerifiedRpcProxy(
-    proxy: proxy, headerStore: headerStore, chainId: chainId, maxBlockWalk: maxBlockWalk
+    proxy: proxy,
+    headerStore: headerStore,
+    filterStore: initTable[int, FilterOptions](),
+    chainId: chainId,
+    maxBlockWalk: maxBlockWalk
   )