Add vulture

.pre-commit-config.yaml

@@ -82,3 +82,11 @@ repos:
         files: no-files
         args: ["lock", "--check"]
         always_run: true
+
+  - repo: https://github.com/jendrikseipp/vulture
+    rev: 'v2.13'
+    hooks:
+      - id: vulture
+        args: ["src/"]
+        files: no-files
+        always_run: true

pyproject.toml

@@ -116,3 +116,15 @@ fail_under = 70
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
+
+[tool.vulture]
+exclude = ["*/test*", "conftest.py", "networks.py"]
+ignore_names = [
+    "default_account",  # execution client
+    "_async_session_pool",  # credentials.py
+    "eth_typing_metadata", "ssz_metadata",  # pyinstaller
+    "DATA_DIR",  # settings
+    "contract_event", "get_from_block", "process_events",  # event processor
+    "validators_root",  # ApprovalRequest
+    "previous_version", "current_version", "genesis_validators_root", "fork_info", "voluntary_exit"  # remote.py
+]

src/common/contracts.py

@@ -245,14 +245,6 @@ async def get_exit_signatures_updated_event(
 
         return last_event
 
-    async def get_rewards_min_oracles(self) -> int:
-        """Fetches the last oracles config updated event."""
-        return await self.contract.functions.rewardsMinOracles().call()
-
-    async def get_validators_min_oracles(self) -> int:
-        """Fetches the last oracles config updated event."""
-        return await self.contract.functions.validatorsMinOracles().call()
-
     async def can_harvest(self, vault_address: ChecksumAddress) -> bool:
         return await self.contract.functions.canHarvest(vault_address).call()
 
@@ -261,10 +253,6 @@ class DepositDataRegistryContract(ContractWrapper):
     abi_path = 'abi/IDepositDataRegistry.json'
     settings_key = 'DEPOSIT_DATA_REGISTRY_CONTRACT_ADDRESS'
 
-    async def get_deposit_data_manager(self) -> ChecksumAddress:
-        """Fetches the vault deposit data manager address."""
-        return await self.contract.functions.getDepositDataManager(settings.vault).call()
-
     async def get_validators_root(self) -> Bytes32:
         """Fetches vault's validators root."""
         return await self.contract.functions.depositDataRoots(settings.vault).call()

src/validators/signing/key_shares.py

@@ -13,38 +13,13 @@
 from py_ecc.optimized_bls12_381.optimized_curve import (
     G1 as P1,  # don't confuse group name (G1) with primitive element name (P1)
 )
-from py_ecc.optimized_bls12_381.optimized_curve import (
-    Z1,
-    Z2,
-    add,
-    curve_order,
-    multiply,
-)
+from py_ecc.optimized_bls12_381.optimized_curve import add, curve_order, multiply
 from py_ecc.typing import Optimized_Field, Optimized_Point3D
-from py_ecc.utils import prime_field_inv
-
-from src.validators.typings import BLSPrivkey
 
 # element of G1 or G2
 G12: TypeAlias = Optimized_Point3D[Optimized_Field]
 
 
-def get_polynomial_points(coefficients: list[int], num_points: int) -> list[int]:
-    """Calculates polynomial points."""
-    points = []
-    for x in range(1, num_points + 1):
-        # start with x=1 and calculate the value of y
-        y = coefficients[0]
-        # calculate each term and add it to y, using modular math
-        for i in range(1, len(coefficients)):
-            exponentiation = (x**i) % curve_order
-            term = (coefficients[i] * exponentiation) % curve_order
-            y = (y + term) % curve_order
-        # add the point to the list of points
-        points.append(y)
-    return points
-
-
 def get_G12_polynomial_points(coefficients: list, num_points: int) -> list:
     """Calculates polynomial points in G1 or G2."""
     points = []
@@ -62,21 +37,6 @@ def get_G12_polynomial_points(coefficients: list, num_points: int) -> list:
     return points
 
 
-def private_key_to_private_key_shares(
-    private_key: BLSPrivkey,
-    threshold: int,
-    total: int,
-) -> list[BLSPrivkey]:
-    coefficients: list[int] = [int.from_bytes(private_key, 'big')]
-
-    for _ in range(threshold - 1):
-        coefficients.append(secrets.randbelow(curve_order))
-
-    points = get_polynomial_points(coefficients, total)
-
-    return [BLSPrivkey(p.to_bytes(32, 'big')) for p in points]
-
-
 def bls_signature_to_shares(
     bls_signature: BLSSignature,
     coefficients_G2: list[G12],
@@ -121,31 +81,3 @@ def bls_signature_and_public_key_to_shares(
     public_key_shards = bls_public_key_to_shares(public_key, coefficients_G1, total)
 
     return bls_signature_shards, public_key_shards
-
-
-def reconstruct_shared_bls_signature(signatures: dict[int, BLSSignature]) -> BLSSignature:
-    """
-    Reconstructs shared BLS private key signature.
-    Copied from https://github.com/dankrad/python-ibft/blob/master/bls_threshold.py
-    """
-    r = Z2
-    for i, sig in signatures.items():
-        sig_point = signature_to_G2(sig)
-        coef = 1
-        for j in signatures:
-            if j != i:
-                coef = -coef * (j + 1) * prime_field_inv(i - j, curve_order) % curve_order
-        r = add(r, multiply(sig_point, coef))
-    return G2_to_signature(r)
-
-
-def get_aggregate_key(keyshares: dict[int, BLSPubkey]) -> BLSPubkey:
-    r = Z1
-    for i, key in keyshares.items():
-        key_point = pubkey_to_G1(key)
-        coef = 1
-        for j in keyshares:
-            if j != i:
-                coef = -coef * (j + 1) * prime_field_inv(i - j, curve_order) % curve_order
-        r = add(r, multiply(key_point, coef))
-    return G1_to_pubkey(r)

src/validators/signing/tests/key_shares.py

@@ -0,0 +1,37 @@
+from eth_typing import BLSPubkey, BLSSignature
+from py_ecc.bls.g2_primitives import (
+    G1_to_pubkey,
+    G2_to_signature,
+    pubkey_to_G1,
+    signature_to_G2,
+)
+from py_ecc.optimized_bls12_381 import Z1, Z2, add, curve_order, multiply
+from py_ecc.utils import prime_field_inv
+
+
+def reconstruct_shared_bls_signature(signatures: dict[int, BLSSignature]) -> BLSSignature:
+    """
+    Reconstructs shared BLS private key signature.
+    Copied from https://github.com/dankrad/python-ibft/blob/master/bls_threshold.py
+    """
+    r = Z2
+    for i, sig in signatures.items():
+        sig_point = signature_to_G2(sig)
+        coef = 1
+        for j in signatures:
+            if j != i:
+                coef = -coef * (j + 1) * prime_field_inv(i - j, curve_order) % curve_order
+        r = add(r, multiply(sig_point, coef))
+    return G2_to_signature(r)
+
+
+def get_aggregate_key(keyshares: dict[int, BLSPubkey]) -> BLSPubkey:
+    r = Z1
+    for i, key in keyshares.items():
+        key_point = pubkey_to_G1(key)
+        coef = 1
+        for j in keyshares:
+            if j != i:
+                coef = -coef * (j + 1) * prime_field_inv(i - j, curve_order) % curve_order
+        r = add(r, multiply(key_point, coef))
+    return G1_to_pubkey(r)

src/validators/signing/tests/oracle_functions.py

@@ -8,7 +8,7 @@
 from web3 import Web3
 
 from src.config.settings import settings
-from src.validators.signing.key_shares import (
+from src.validators.signing.tests.key_shares import (
     get_aggregate_key,
     reconstruct_shared_bls_signature,
 )

src/validators/typings.py

@@ -4,7 +4,6 @@
 
 from eth_typing import BlockNumber, BLSSignature, ChecksumAddress, HexStr
 from multiproof import MultiProof, StandardMerkleTree
-from sw_utils.typings import Bytes32
 
 BLSPrivkey = NewType('BLSPrivkey', bytes)
 
@@ -74,14 +73,6 @@ class ApprovalRequest:
     validators_manager_signature: HexStr | None = None
 
 
-@dataclass
-class KeeperApprovalParams:
-    validatorsRegistryRoot: HexStr | Bytes32
-    validators: HexStr | bytes
-    signatures: HexStr | bytes
-    exitSignaturesIpfsHash: str
-
-
 class ValidatorsRegistrationMode(Enum):
     """
     AUTO mode: validators are registered automatically when vault assets are enough.