Add ssv operator key check

src/app.py

@@ -24,7 +24,11 @@ async def app() -> None:
     if not is_checks_ok:
         return
 
-    await create_tasks()
+    try:
+        await create_tasks()
+    except Exception as e:
+        logger.exception_verbose(e)
+        return
 
     logger.info('DVT Sidecar service started')
 

src/common/setup_logging.py

@@ -30,6 +30,10 @@ def error_verbose(self, msg, *args, **kwargs):  # type: ignore
     def exception_verbose(self, e: Exception):  # type: ignore
         self.error_verbose('%s', e)
 
+    def warning(self, msg, *args, **kwargs):  # type: ignore
+        args = [format_error(arg) if isinstance(arg, Exception) else arg for arg in args]
+        super().warning(msg, *args, **kwargs)
+
 
 def setup_logging() -> None:
     if settings.log_format == LOG_JSON:

src/validators/keystores/ssv.py

@@ -16,6 +16,7 @@
 from src.common.setup_logging import ExtendedLogger
 from src.common.utils import to_chunks
 from src.config import settings
+from src.validators.keystores import ssv_api
 from src.validators.keystores.base import BaseKeystore
 from src.validators.keystores.typings import BLSPrivkey, Keys
 
@@ -47,21 +48,22 @@ async def load() -> 'SSVKeystore':
         if not settings.ssv_operator_password_file:
             raise RuntimeError('SSV_OPERATOR_PASSWORD_FILE must be set')
 
-        return SSVKeystore.load_as_operator(
+        return await SSVKeystore.load_as_operator(
             settings.ssv_operator_id,
             settings.ssv_operator_key_file,
             settings.ssv_operator_password_file,
         )
 
     @staticmethod
-    def load_as_operator(
+    async def load_as_operator(
         ssv_operator_id: int, ssv_operator_key_file: str, ssv_operator_password_file: str
     ) -> 'SSVKeystore':
         """
         Loads validator keys from keyshares json file,
         filters key shares related to a given operator.
         """
         operator_key = SSVOperator.load_key(ssv_operator_key_file, ssv_operator_password_file)
+        await SSVOperator.check_operator_key(ssv_operator_id, operator_key)
 
         if not settings.ssv_keyshares_file:
             raise RuntimeError('SSV_KEYSHARES_FILE must be set')
@@ -295,6 +297,29 @@ def public_key_from_string(s: str) -> RSA.RsaKey:
         """
         return RSA.import_key(base64.b64decode(s))
 
+    @staticmethod
+    async def check_operator_key(ssv_operator_id: int, operator_key: RSA.RsaKey) -> None:
+        """
+        Checks that `operator_key` belongs to operator `ssv_operator_id`.
+        SSV API is used to fetch operator public key by id.
+
+        :param ssv_operator_id: SSV Operator id
+        :param operator_key: SSV Operator private key
+        """
+        logger.info('Checking SSV operator key for operator id %d...', ssv_operator_id)
+        try:
+            operator_data = await ssv_api.get_operator(ssv_operator_id)
+        except Exception as e:
+            # skip checks in case of ssv api error
+            logger.warning('SSV api error. Skip checking operator key. Error detail: %s', e)
+            return
+
+        public_key = operator_key.public_key()
+        api_public_key = SSVOperator.public_key_from_string(operator_data['public_key'])
+
+        if public_key != api_public_key:
+            raise ValueError(f'Operator key does not belong to operator {ssv_operator_id}')
+
 
 @dataclass
 class SSVSharesData:

src/validators/keystores/ssv_api.py

@@ -0,0 +1,15 @@
+import aiohttp
+from aiohttp import ClientTimeout
+
+from src.config import settings
+
+base_url = 'https://api.ssv.network/api/v4'
+timeout = 10
+
+
+async def get_operator(operator_id: int) -> dict:
+    url = f'{base_url}/{settings.network}/operators/{operator_id}'
+    async with aiohttp.ClientSession(timeout=ClientTimeout(timeout)) as session:
+        async with session.get(url) as res:
+            res.raise_for_status()
+            return await res.json()

src/validators/tasks.py

@@ -22,18 +22,10 @@
 
 async def create_tasks() -> None:
     if settings.cluster_type == OBOL:
-        try:
-            await obol_create_tasks()
-        except Exception as e:
-            logger.exception_verbose(e)
-            return
+        await obol_create_tasks()
 
     if settings.cluster_type == SSV:
-        try:
-            await ssv_create_tasks()
-        except Exception as e:
-            logger.exception_verbose(e)
-            return
+        await ssv_create_tasks()
 
 
 async def obol_create_tasks() -> None:
@@ -79,7 +71,7 @@ async def ssv_create_tasks() -> None:
             ssv_operator_password_file = settings.ssv_operator_password_file_template.format(
                 operator_id=ssv_operator_id
             )
-            keystore = SSVKeystore.load_as_operator(
+            keystore = await SSVKeystore.load_as_operator(
                 ssv_operator_id, ssv_operator_key_file, ssv_operator_password_file
             )