Geth support for inspect

README.md

@@ -41,7 +41,7 @@ Example:
 export RPC_URL="http://111.111.111.111:8546"
 ```
 
-**Note**: mev-inspect-py currently requires an RPC of a full archive node with support for Erigon traces and receipts (not geth 😔).
+**Note**: mev-inspect-py currently requires an RPC of a full archive node with support for Erigon traces and receipts. Geth additions have been added to translate geth traces and receipts to Erigon ones and can be accessed using `--geth` flag.
 
 Next, start all services with:
 
@@ -62,6 +62,7 @@ kubectl exec deploy/mev-inspect -- alembic upgrade head
 ### Inspect a single block
 
 Inspecting block [12914944](https://twitter.com/mevalphaleak/status/1420416437575901185):
+**Note**: Add `--geth` at the end if RPC_URL points to a geth / geth like node.
 
 ```
 kubectl exec deploy/mev-inspect -- poetry run inspect-block 12914944
@@ -70,6 +71,7 @@ kubectl exec deploy/mev-inspect -- poetry run inspect-block 12914944
 ### Inspect many blocks
 
 Inspecting blocks 12914944 to 12914954:
+**Note**: Add `--geth` at the end if RPC_URL points to a geth / geth like node.
 
 ```
 kubectl exec deploy/mev-inspect -- poetry run inspect-many-blocks 12914944 12914954

cli.py

@@ -4,6 +4,7 @@
 
 import click
 from web3 import Web3
+from web3.middleware import geth_poa_middleware
 
 from mev_inspect.classifiers.trace import TraceClassifier
 from mev_inspect.db import get_inspect_session, get_trace_session
@@ -26,12 +27,15 @@ def cli():
 @click.argument("block_number", type=int)
 @click.option("--rpc", default=lambda: os.environ.get(RPC_URL_ENV, ""))
 @click.option("--cache/--no-cache", default=True)
-def inspect_block_command(block_number: int, rpc: str, cache: bool):
+@click.option("--geth/--no-geth", default=False)
+def inspect_block_command(block_number: int, rpc: str, cache: bool, geth: bool):
     inspect_db_session = get_inspect_session()
     trace_db_session = get_trace_session()
 
     base_provider = get_base_provider(rpc)
     w3 = Web3(base_provider)
+    if geth:
+        w3.middleware_onion.inject(geth_poa_middleware, layer=0)
     trace_classifier = TraceClassifier()
 
     if not cache:
@@ -41,6 +45,7 @@ def inspect_block_command(block_number: int, rpc: str, cache: bool):
         inspect_db_session,
         base_provider,
         w3,
+        geth,
         trace_classifier,
         block_number,
         trace_db_session=trace_db_session,
@@ -52,15 +57,18 @@ def inspect_block_command(block_number: int, rpc: str, cache: bool):
 @click.argument("before_block", type=int)
 @click.option("--rpc", default=lambda: os.environ.get(RPC_URL_ENV, ""))
 @click.option("--cache/--no-cache", default=True)
+@click.option("--geth/--no-geth", default=False)
 def inspect_many_blocks_command(
-    after_block: int, before_block: int, rpc: str, cache: bool
+    after_block: int, before_block: int, rpc: str, cache: bool, geth: bool
 ):
 
     inspect_db_session = get_inspect_session()
     trace_db_session = get_trace_session()
 
     base_provider = get_base_provider(rpc)
     w3 = Web3(base_provider)
+    if geth:
+        w3.middleware_onion.inject(geth_poa_middleware, layer=0)
     trace_classifier = TraceClassifier()
 
     if not cache:
@@ -79,6 +87,7 @@ def inspect_many_blocks_command(
             inspect_db_session,
             base_provider,
             w3,
+            geth,
             trace_classifier,
             block_number,
             trace_db_session=trace_db_session,

mev_inspect/block.py

@@ -1,5 +1,8 @@
 from pathlib import Path
 from typing import List, Optional
+import json
+import asyncio
+import aiohttp
 
 from sqlalchemy import orm
 from web3 import Web3
@@ -19,6 +22,7 @@ def get_latest_block_number(w3: Web3) -> int:
 def create_from_block_number(
     base_provider,
     w3: Web3,
+    geth: bool,
     block_number: int,
     trace_db_session: Optional[orm.Session],
 ) -> Block:
@@ -28,25 +32,37 @@ def create_from_block_number(
         block = _find_block(trace_db_session, block_number)
 
     if block is None:
-        return _fetch_block(w3, base_provider, block_number)
+        return _fetch_block(w3, base_provider, geth, block_number)
     else:
         return block
 
 
 def _fetch_block(
     w3,
     base_provider,
+    geth,
     block_number: int,
 ) -> Block:
     block_json = w3.eth.get_block(block_number)
-    receipts_json = base_provider.make_request("eth_getBlockReceipts", [block_number])
-    traces_json = w3.parity.trace_block(block_number)
 
-    receipts: List[Receipt] = [
-        Receipt(**receipt) for receipt in receipts_json["result"]
-    ]
-    traces = [Trace(**trace_json) for trace_json in traces_json]
-    base_fee_per_gas = fetch_base_fee_per_gas(w3, block_number)
+    if not geth:
+        receipts_json = base_provider.make_request(
+            "eth_getBlockReceipts", [block_number]
+        )
+        traces_json = w3.parity.trace_block(block_number)
+
+        receipts: List[Receipt] = [
+            Receipt(**receipt) for receipt in receipts_json["result"]
+        ]
+        traces = [Trace(**trace_json) for trace_json in traces_json]
+        base_fee_per_gas = fetch_base_fee_per_gas(w3, block_number)
+    else:
+        traces = geth_get_tx_traces_parity_format(base_provider, block_json)
+        geth_tx_receipts = geth_get_tx_receipts(
+            base_provider, block_json["transactions"]
+        )
+        receipts = geth_receipts_translator(block_json, geth_tx_receipts)
+        base_fee_per_gas = 0
 
     return Block(
         block_number=block_number,
@@ -164,3 +180,144 @@ def cache_block(cache_path: Path, block: Block):
 def _get_cache_path(block_number: int) -> Path:
     cache_directory_path = Path(cache_directory)
     return cache_directory_path / f"{block_number}.json"
+
+
+# Geth specific additions
+
+
+def geth_get_tx_traces_parity_format(base_provider, block_json):
+    block_hash = block_json["hash"]
+    block_trace = geth_get_tx_traces(base_provider, block_hash)
+    parity_traces = []
+    for idx, trace in enumerate(block_trace["result"]):
+        if "result" in trace:
+            parity_traces.extend(
+                unwrap_tx_trace_for_parity(block_json, idx, trace["result"])
+            )
+    return parity_traces
+
+
+def geth_get_tx_traces(base_provider, block_hash):
+    block_trace = base_provider.make_request(
+        "debug_traceBlockByHash", [block_hash.hex(), {"tracer": "callTracer"}]
+    )
+    return block_trace
+
+
+def unwrap_tx_trace_for_parity(
+    block_json, tx_pos_in_block, tx_trace, position=[]
+) -> List[Trace]:
+    response_list = []
+    _calltype_mapping = {
+        "CALL": "call",
+        "DELEGATECALL": "delegateCall",
+        "CREATE": "create",
+        "SUICIDE": "suicide",
+        "REWARD": "reward",
+    }
+    try:
+        if tx_trace["type"] == "STATICCALL":
+            return []
+        action_dict = dict()
+        action_dict["callType"] = _calltype_mapping[tx_trace["type"]]
+        if action_dict["callType"] == "call":
+            action_dict["value"] = tx_trace["value"]
+        for key in ["from", "to", "gas", "input"]:
+            action_dict[key] = tx_trace[key]
+
+        result_dict = dict()
+        for key in ["gasUsed", "output"]:
+            result_dict[key] = tx_trace[key]
+
+        response_list.append(
+            Trace(
+                action=action_dict,
+                block_hash=str(block_json["hash"]),
+                block_number=int(block_json["number"]),
+                result=result_dict,
+                subtraces=len(tx_trace["calls"]) if "calls" in tx_trace.keys() else 0,
+                trace_address=position,
+                transaction_hash=block_json["transactions"][tx_pos_in_block].hex(),
+                transaction_position=tx_pos_in_block,
+                type=TraceType(_calltype_mapping[tx_trace["type"]]),
+            )
+        )
+    except Exception:
+        return []
+
+    if "calls" in tx_trace.keys():
+        for idx, subcall in enumerate(tx_trace["calls"]):
+            response_list.extend(
+                unwrap_tx_trace_for_parity(
+                    block_json, tx_pos_in_block, subcall, position + [idx]
+                )
+            )
+    return response_list
+
+
+async def geth_get_tx_receipts_task(session, endpoint_uri, tx):
+    data = {
+        "jsonrpc": "2.0",
+        "id": "0",
+        "method": "eth_getTransactionReceipt",
+        "params": [tx.hex()],
+    }
+    async with session.post(endpoint_uri, json=data) as response:
+        if response.status != 200:
+            response.raise_for_status()
+        return await response.text()
+
+
+async def geth_get_tx_receipts_async(endpoint_uri, transactions):
+    geth_tx_receipts = []
+    async with aiohttp.ClientSession() as session:
+        tasks = [
+            asyncio.create_task(geth_get_tx_receipts_task(session, endpoint_uri, tx))
+            for tx in transactions
+        ]
+        geth_tx_receipts = await asyncio.gather(*tasks)
+    return [json.loads(tx_receipts) for tx_receipts in geth_tx_receipts]
+
+
+def geth_get_tx_receipts(base_provider, transactions):
+    return asyncio.run(
+        geth_get_tx_receipts_async(base_provider.endpoint_uri, transactions)
+    )
+
+
+def geth_receipts_translator(block_json, geth_tx_receipts) -> List[Receipt]:
+    json_decoded_receipts = [
+        tx_receipt["result"]
+        if tx_receipt != None and ("result" in tx_receipt.keys())
+        else None
+        for tx_receipt in geth_tx_receipts
+    ]
+    results = []
+    for idx, tx_receipt in enumerate(json_decoded_receipts):
+        if tx_receipt != None:
+            results.append(unwrap_tx_receipt_for_parity(block_json, idx, tx_receipt))
+    return results
+
+
+def unwrap_tx_receipt_for_parity(block_json, tx_pos_in_block, tx_receipt) -> Receipt:
+    try:
+        if tx_pos_in_block != int(tx_receipt["transactionIndex"], 16):
+            print(
+                "Alert the position of transaction in block is mismatched ",
+                tx_pos_in_block,
+                tx_receipt["transactionIndex"],
+            )
+        return Receipt(
+            block_number=block_json["number"],
+            transaction_hash=tx_receipt["transactionHash"],
+            transaction_index=tx_pos_in_block,
+            gas_used=tx_receipt["gasUsed"],
+            effective_gas_price=tx_receipt["effectiveGasPrice"],
+            cumulative_gas_used=tx_receipt["cumulativeGasUsed"],
+            to=tx_receipt["to"],
+        )
+
+    except Exception as e:
+        print("error while decoding receipt", tx_receipt, e)
+
+    return Receipt()

mev_inspect/inspect_block.py

@@ -39,6 +39,7 @@ def inspect_block(
     inspect_db_session: orm.Session,
     base_provider,
     w3: Web3,
+    geth: bool,
     trace_clasifier: TraceClassifier,
     block_number: int,
     trace_db_session: Optional[orm.Session],
@@ -47,6 +48,7 @@ def inspect_block(
     block = create_from_block_number(
         base_provider,
         w3,
+        geth,
         block_number,
         trace_db_session,
     )

pyproject.toml

@@ -11,6 +11,8 @@ pydantic = "^1.8.2"
 hexbytes = "^0.2.1"
 click = "^8.0.1"
 psycopg2 = "^2.9.1"
+aiohttp = "^3.7.4"
+asyncio = "^3.4.3"
 
 [tool.poetry.dev-dependencies]
 pre-commit = "^2.13.0"