Feature/val 1404 eip 7251 head watcher alerts for new el requests

.env.example

@@ -7,7 +7,7 @@ KEYS_API_URI=URL_TO_KEYS_API
 LIDO_LOCATOR_ADDRESS=ETHEREUM_ADDRESS
 EXECUTION_CLIENT_URI=URL_TO_EL_API
 
-# For option when KEYS_SOURCE is 'keys_file'
+# For option when KEYS_SOURCE is 'file'
 # CONSENSUS_CLIENT_URI: URL_TO_CL_API
-# KEYS_SOURCE: keys_file
+# KEYS_SOURCE: file
 # KEYS_FILE_PATH: path/to/keys.yml

.gitignore

@@ -131,3 +131,6 @@ dmypy.json
 
 # IDE
 .idea/
+
+# Docker
+.volumes

Dockerfile

@@ -1,11 +1,11 @@
-FROM python:3.11.3-slim as base
+FROM python:3.11.3-slim AS base
 
 RUN apt-get update && apt-get install -y --no-install-recommends -qq \
     gcc=4:10.2.1-1 \
     libffi-dev=3.3-6 \
     g++=4:10.2.1-1 \
     git=1:2.30.2-1+deb11u2 \
-    curl=7.74.0-1.3+deb11u12 \
+    curl=7.74.0-1.3+deb11u14 \
  && apt-get clean \
  && rm -rf /var/lib/apt/lists/*
 
@@ -20,7 +20,7 @@ ENV PYTHONUNBUFFERED=1 \
 
 ENV PATH="$VENV_PATH/bin:$PATH"
 
-FROM base as builder
+FROM base AS builder
 
 ENV POETRY_VERSION=1.3.2
 RUN pip install --no-cache-dir poetry==$POETRY_VERSION
@@ -30,16 +30,16 @@ COPY pyproject.toml poetry.lock ./
 RUN poetry install --only main --no-root
 
 
-FROM base as production
+FROM base AS production
 
 COPY --from=builder $VENV_PATH $VENV_PATH
 WORKDIR /app
 COPY . .
 
 RUN apt-get clean && find /var/lib/apt/lists/ -type f -delete && chown -R www-data /app/
 
-ENV PROMETHEUS_PORT 9000
-ENV HEALTHCHECK_SERVER_PORT 9010
+ENV PROMETHEUS_PORT=9000
+ENV HEALTHCHECK_SERVER_POR=9010
 
 EXPOSE $PROMETHEUS_PORT
 USER www-data

README.md

@@ -27,7 +27,7 @@ Currently it supports:
 > All exits will be handled as unexpected for specified keys
 
 1. Fill `docker/validators/keys.yml` with your values
-2. Set `KEYS_SOURCE=keys_file` in `.env`
+2. Set `KEYS_SOURCE=file` in `.env`
 
 > If you want to use another path, specify it in `KEYS_FILE_PATH` env variable
 
@@ -42,12 +42,12 @@ Currently it supports:
 * **Required:** false
 * **Default:** false
 ---
-`KEYS_SOURCE` - Keys source. If `keys_api` - application will fetch keys from Keys API, if `keys_file` - application will fetch keys from `KEYS_FILE_PATH`
+`KEYS_SOURCE` - Keys source. If `keys_api` - application will fetch keys from Keys API, if `file` - application will fetch keys from `KEYS_FILE_PATH`
 * **Required:** false
 * **Default:** keys_api
 ---
 `KEYS_FILE_PATH` - Path to file with keys
-* **Required:** if `KEYS_SOURCE` is `keys_file`
+* **Required:** if `KEYS_SOURCE` is `file`
 * **Default:** ./docker/validators/keys.yml
 ---
 `CONSENSUS_CLIENT_URI` - Ethereum consensus layer comma separated API urls
@@ -139,6 +139,10 @@ Currently it supports:
 `ALERTMANAGER_REQUEST_SLEEP_BEFORE_RETRY_IN_SECONDS` - Alertmanager request retry timeout in seconds
 * **Required:** false
 * **Default:** 1
+---
+`VALID_WITHDRAWAL_ADDRESSES` - A comma-separated list of addresses. Triggers a critical alert if a monitored execution_request contains a source_address matching any of these addresses 
+* **Required:** false
+* **Default:** []
 
 ## Application metrics
 

docker-compose.yml

@@ -52,7 +52,7 @@ services:
       - 9090
 
   alertmanager:
-    image: prom/alertmanager:latest
+    image: prom/alertmanager:v0.25.0
     container_name: alertmanager
     restart: unless-stopped
     networks:

src/handlers/consolidation.py

@@ -0,0 +1,42 @@
+import logging
+
+from unsync import unsync
+
+from src.alerts.common import CommonAlert
+from src.handlers.handler import WatcherHandler
+from src.handlers.helpers import beaconchain
+from src.metrics.prometheus.duration_meter import duration_meter
+from src.providers.consensus.typings import FullBlockInfo
+
+logger = logging.getLogger()
+
+
+class ConsolidationHandler(WatcherHandler):
+    @unsync
+    @duration_meter()
+    def handle(self, watcher, head: FullBlockInfo):
+        if not head.message.body.execution_requests or not head.message.body.execution_requests.consolidations:
+            logger.info({"msg": f"No consolidation requests in block [{head.message.slot}]"})
+            return
+
+        slot = head.message.slot
+        for consolidation in head.message.body.execution_requests.consolidations:
+            alert, summary = None, ""
+            if consolidation.source_address in watcher.valid_withdrawal_addresses:
+                alert = CommonAlert(name="HeadWatcherConsolidationSourceWithdrawalAddress", severity="critical")
+                summary = "‼️⛔️Validator consolidation was requested from Withdrawal Vault source address"
+            elif consolidation.source_pubkey in watcher.user_keys:
+                alert = CommonAlert(name="HeadWatcherConsolidationUserSourcePubkey", severity="info")
+                summary = "⚠️Consolidation was requested for our validators"
+            elif consolidation.target_pubkey in watcher.user_keys:
+                alert = CommonAlert(name="HeadWatcherConsolidationUserTargetPubkey", severity="info")
+                summary = "⚠️Someone attempts to consolidate their validators to our validators"
+            # in the future we should check the type of validator WC:
+            # if it is 0x02 and source_address == WCs of source validator - It's donation!
+
+            if alert:
+                description = (f"EL consolidation request source_address='{consolidation.source_address}', "
+                               f"source_pubkey={consolidation.source_pubkey}, "
+                               f"target_pubkey='{consolidation.target_pubkey}'\n"
+                               f"Slot: {beaconchain(slot)}")
+                self.send_alert(watcher, alert.build_body(summary, description))

src/handlers/el_triggered_exit.py

@@ -0,0 +1,37 @@
+import logging
+
+from unsync import unsync
+
+from src.alerts.common import CommonAlert
+from src.handlers.handler import WatcherHandler
+from src.handlers.helpers import beaconchain
+from src.metrics.prometheus.duration_meter import duration_meter
+from src.providers.consensus.typings import FullBlockInfo
+
+logger = logging.getLogger()
+
+
+class ElTriggeredExitHandler(WatcherHandler):
+    @unsync
+    @duration_meter()
+    def handle(self, watcher, head: FullBlockInfo):
+        if not head.message.body.execution_requests or not head.message.body.execution_requests.withdrawals:
+            logger.debug({"msg": f"No withdrawals requests in block [{head.message.slot}]"})
+            return
+
+        slot = head.message.slot
+        for withdrawal in head.message.body.execution_requests.withdrawals:
+            alert, summary = None, ""
+            if withdrawal.source_address in watcher.valid_withdrawal_addresses:
+                alert = CommonAlert(name="HeadWatcherELWithdrawalFromUserWithdrawalAddress", severity="critical")
+                summary = "🔗‍🏃🚪Our validator triggered withdrawal was requested from our Withdrawal Vault address"
+            elif withdrawal.validator_pubkey in watcher.user_keys:
+                alert = CommonAlert(name="HeadWatcherUserELWithdrawal", severity="info")
+                summary = "🔗‍🏃🚪Our validator triggered withdrawal was requested"
+
+            if alert:
+                description = (f"EL withdrawals request source_address='{withdrawal.source_address}', "
+                               f"validator_pubkey={withdrawal.validator_pubkey}, "
+                               f"amount='{withdrawal.amount}'\n"
+                               f"Slot: {beaconchain(slot)}")
+                self.send_alert(watcher, alert.build_body(summary, description))

src/handlers/fork.py

@@ -4,11 +4,9 @@
 
 from src.alerts.common import CommonAlert
 from src.handlers.handler import WatcherHandler
+from src.handlers.helpers import beaconchain
 from src.metrics.prometheus.duration_meter import duration_meter
 from src.providers.consensus.typings import BlockHeaderResponseData, ChainReorgEvent
-from src.variables import NETWORK_NAME
-
-BEACONCHAIN_URL_TEMPLATE = "[{0}](https://{1}.beaconcha.in/slot/{0})"
 
 
 class ForkHandler(WatcherHandler):
@@ -43,7 +41,7 @@ def _send_reorg_alert(self, watcher, chain_reorg: ChainReorgEvent):
         alert = CommonAlert(name="UnhandledChainReorg", severity="info")
         links = "\n".join(
             [
-                BEACONCHAIN_URL_TEMPLATE.format(s, NETWORK_NAME)
+                beaconchain(s)
                 for s in range(int(chain_reorg.slot) - int(chain_reorg.depth), int(chain_reorg.slot) + 1)
             ]
         )
@@ -59,5 +57,5 @@ def _send_unhandled_head_alert(self, watcher, head: BlockHeaderResponseData):
         if diff > 0:
             additional_msg = f"\nAnd {diff} slot(s) before it"
         parent_root = head.header.message.parent_root
-        description = f"Please, check unhandled slot: {BEACONCHAIN_URL_TEMPLATE.format(parent_root, NETWORK_NAME)}{additional_msg}"
+        description = f"Please, check unhandled slot: {beaconchain(parent_root)}{additional_msg}"
         self.send_alert(watcher, alert.build_body(summary, description))

src/handlers/helpers.py

@@ -0,0 +1,7 @@
+from src.variables import NETWORK_NAME
+
+BEACONCHAIN_URL_TEMPLATE = "[{0}](https://{1}.beaconcha.in/slot/{0})"
+
+
+def beaconchain(slot) -> str:
+    return BEACONCHAIN_URL_TEMPLATE.format(slot, NETWORK_NAME)

src/main.py

@@ -2,9 +2,11 @@
 from web3.middleware import simple_cache_middleware
 
 from src import variables
+from src.handlers.el_triggered_exit import ElTriggeredExitHandler
 from src.handlers.exit import ExitsHandler
 from src.handlers.fork import ForkHandler
 from src.handlers.slashing import SlashingHandler
+from src.handlers.consolidation import ConsolidationHandler
 from src.keys_source.base_source import SourceType
 from src.keys_source.file_source import FileSource
 from src.keys_source.keys_api_source import KeysApiSource
@@ -60,6 +62,8 @@ def main():
         ForkHandler(),
         ExitsHandler(),
         # FinalityHandler(), ???
+        ConsolidationHandler(),
+        ElTriggeredExitHandler()
     ]
     Watcher(handlers, keys_source, web3).run()
 

src/providers/consensus/typings.py

@@ -74,12 +74,39 @@ class BlockVoluntaryExit(Nested, FromResponse):
     signature: str
 
 
+@dataclass
+class ConsolidationRequest(FromResponse):
+    source_address: str
+    source_pubkey: str
+    target_pubkey: str
+
+@dataclass
+class WithdrawalRequest(FromResponse):
+    source_address: str
+    validator_pubkey: str
+    amount: str
+
+@dataclass
+class DepositRequest(FromResponse):
+    pubkey: str
+    withdrawal_credentials: str
+    amount: str
+    signature: str
+    index: int
+
+@dataclass
+class ExecutionRequests(Nested, FromResponse):
+    deposits: list[DepositRequest]
+    withdrawals: list[WithdrawalRequest]
+    consolidations: list[ConsolidationRequest]
+
 @dataclass
 class BlockBody(Nested, FromResponse):
     execution_payload: BlockExecutionPayload
     voluntary_exits: list[BlockVoluntaryExit]
     proposer_slashings: list
     attester_slashings: list
+    execution_requests: Optional[ExecutionRequests] = None
 
 
 @dataclass

src/utils/dataclass.py

@@ -1,13 +1,21 @@
 import functools
 from dataclasses import dataclass, fields, is_dataclass
 from types import GenericAlias
-from typing import Callable, Self, Sequence, TypeVar
+from typing import Callable, Self, Sequence, TypeVar, get_origin, Union, get_args
 
 
 class DecodeToDataclassException(Exception):
     pass
 
 
+def try_extract_underlying_type_from_optional(field):
+    args = get_args(field)
+    types = [x for x in args if x != type(None)]
+    if get_origin(field) is Union and type(None) in args and len(types) == 1:
+        return types[0]
+    return None
+
+
 @dataclass
 class Nested:
     """
@@ -31,6 +39,9 @@ def __post_init__(self):
             elif is_dataclass(field.type) and not is_dataclass(getattr(self, field.name)):
                 factory = self.__get_dataclass_factory(field.type)
                 setattr(self, field.name, factory(**getattr(self, field.name)))
+            elif getattr(self, field.name) and (underlying := try_extract_underlying_type_from_optional(field.type)):
+                factory = self.__get_dataclass_factory(underlying)
+                setattr(self, field.name, factory(**getattr(self, field.name)))
 
     @staticmethod
     def __get_dataclass_factory(field_type):

src/variables.py

@@ -43,6 +43,8 @@
 
 LIDO_LOCATOR_ADDRESS = os.getenv('LIDO_LOCATOR_ADDRESS', '')
 
+VALID_WITHDRAWAL_ADDRESSES = os.getenv('VALID_WITHDRAWAL_ADDRESSES', '').split(',')
+
 # - Metrics -
 PROMETHEUS_PORT = int(os.getenv('PROMETHEUS_PORT', 9000))
 PROMETHEUS_PREFIX = os.getenv("PROMETHEUS_PREFIX", "ethereum_head_watcher")

src/watcher.py

@@ -53,6 +53,11 @@ def __init__(self, handlers: list[WatcherHandler], keys_source: BaseSource, web3
         self.indexed_validators_keys: dict[str, str] = {}
         self.chain_reorgs: dict[str, ChainReorgEvent] = {}
         self.handled_headers: list[BlockHeaderResponseData] = []
+        self.valid_withdrawal_addresses = set(variables.VALID_WITHDRAWAL_ADDRESSES)
+        if not self.valid_withdrawal_addresses and self.execution:
+            self.valid_withdrawal_addresses = {
+                self.execution.lido_contracts.lido_locator.functions.withdrawalVault().call()
+            }
 
     def run(self, slots_range: Optional[str] = SLOTS_RANGE):
         def _run(slot_to_handle='head'):
@@ -161,11 +166,12 @@ def force_use_fallback_callback(result) -> bool:
             return False
 
         slot = slot or 'head'
+
         current_head = self.consensus.get_block_header(
             slot, force_use_fallback_callback if slot == 'head' else lambda _: False
         )
         if len(self.handled_headers) > 0 and int(current_head.header.message.slot) == int(
-            self.handled_headers[-1].header.message.slot
+                self.handled_headers[-1].header.message.slot
         ):
             return None
         current_block = self.consensus.get_block_details(current_head.root)

tests/execution_requests/conftest.py

@@ -0,0 +1,36 @@
+import pytest
+
+from src.keys_source.base_source import NamedKey
+from tests.execution_requests.helpers import gen_random_address
+from tests.execution_requests.stubs import TestValidator, WatcherStub
+
+
+@pytest.fixture
+def user_keys() -> dict[str, NamedKey]:
+    return {}
+
+
+@pytest.fixture
+def validator():
+    return TestValidator.random()
+
+
+@pytest.fixture
+def user_validator(user_keys):
+    random_validator = TestValidator.random()
+    user_keys[random_validator.pubkey] = NamedKey(
+        key=random_validator.pubkey, operatorName='Test operator', operatorIndex='1', moduleIndex='1'
+    )
+    return random_validator
+
+
+@pytest.fixture
+def watcher(user_keys) -> WatcherStub:
+    return WatcherStub(user_keys=user_keys)
+
+
+@pytest.fixture
+def withdrawal_address(watcher: WatcherStub) -> str:
+    address = gen_random_address()
+    watcher.valid_withdrawal_addresses.add(address)
+    return address