add log filters

Author: chirag-parmar

nimbus_verified_proxy/rpc/receipts.nim

@@ -5,11 +5,8 @@
 #   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
 # at your option. This file may not be copied, modified, or distributed except according to those terms.
 
-<<<<<<< HEAD
 {.push gcsafe, raises: [].}
 
-=======
->>>>>>> 5e3712b73 (add copyright)
 import
   std/sequtils,
   results,
@@ -87,14 +84,17 @@ proc getReceipts*(
 
   await vp.getReceipts(header, numberTag)
 
-proc getLogs*(
-    vp: VerifiedRpcProxy, filterOptions: FilterOptions
+proc verifyLogs*(
+    vp: VerifiedRpcProxy, logObjs: seq[LogObject]
 ): Future[Result[seq[LogObject], string]] {.async.} =
-  let logObjs =
-    try:
-      await vp.rpcClient.eth_getLogs(filterOptions)
-    except CatchableError as e:
-      return err(e.msg)
+  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")
 
   # store block hashes contains the logs so that we can batch receipt requests
   var
@@ -103,7 +103,7 @@ proc getLogs*(
 
   for lg in logObjs:
     # none only for pending logs before block is built
-    if lg.blockHash.isSome() and lg.transactionIndex.isSome() and lg.logIndex.isSome():
+    if lg.blockNumber.isSome() and lg.blockHash.isSome() and lg.transactionIndex.isSome() and lg.logIndex.isSome():
       # exploit sequentiality of logs 
       if prevBlockHash != lg.blockHash.get():
         # TODO: a cache will solve downloading the same block receipts for multiple logs
@@ -119,7 +119,7 @@ proc getLogs*(
 
       if rxLog.address != lg.address or rxLog.data != lg.data or
           rxLog.topics != lg.topics or
-          (not match(toLog(lg), filterOptions.address, filterOptions.topics)):
+          (not match(toLog(lg), filterOptions.address, filterOptions.topics)) or lg.blockNumber.get < bottom or lg.blockNumber.get > top:
         return err("one of the returned logs is invalid")
 
   return ok(logObjs)

nimbus_verified_proxy/rpc/rpc_eth_api.nim

@@ -237,6 +237,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,
@@ -271,10 +272,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],
   web3/[eth_api, eth_api_types],
   stint,
@@ -72,6 +73,7 @@ type
 
     # TODO: when the list grows big add a config object instead
     # config parameters
+    filterStore*: Table[int, FilterOptions]
     chainId*: UInt256
     maxBlockWalk*: uint64
 
@@ -88,6 +90,7 @@ proc init*(
     accountsCache: AccountsCache.init(ACCOUNTS_CACHE_SIZE),
     codeCache: CodeCache.init(CODE_CACHE_SIZE),
     storageCache: StorageCache.init(STORAGE_CACHE_SIZE),
+    filterStore: initTable[int, FilterOptions](),
     chainId: chainId,
-    maxBlockWalk: maxBlockWalk,
+    maxBlockWalk: maxBlockWalk
   )