Feature: Signature verification for Solana and Ethereum messages (#19)

* Problem: Users cannot verify signatures

Solution: Implement verify_signature() functions for Ethereum and Solana signatures, more are following

* reformat tests with black

* add tezos signature verification function

* remove unused local variable

* raise new BadSignatureError exception instead of boolean check for verify_signature()

Author: MHHukiewitz

src/aleph/sdk/chains/common.py

@@ -9,9 +9,16 @@
 from aleph.sdk.conf import settings
 
 
-def get_verification_buffer(message):
-    """Returns a serialized string to verify the message integrity
-    (this is was it signed)
+def get_verification_buffer(message: Dict) -> bytes:
+    """
+    Returns the verification buffer that Aleph nodes use to verify the signature of a message.
+    Note:
+        The verification buffer is a string of the following format:
+        b"{chain}\\n{sender}\\n{type}\\n{item_hash}"
+    Args:
+        message: Message to get the verification buffer for
+    Returns:
+        bytes: Verification buffer
     """
     return "{chain}\n{sender}\n{type}\n{item_hash}".format(**message).encode("utf-8")
 
@@ -27,8 +34,13 @@ class BaseAccount(ABC):
     private_key: bytes
 
     def _setup_sender(self, message: Dict) -> Dict:
-        """Set the sender of the message as the account's public key.
+        """
+        Set the sender of the message as the account's public key.
         If a sender is already specified, check that it matches the account's public key.
+        Args:
+            message: Message to add the sender to
+        Returns:
+            Dict: Message with the sender set
         """
         if not message.get("sender"):
             message["sender"] = self.get_address()
@@ -40,24 +52,51 @@ def _setup_sender(self, message: Dict) -> Dict:
 
     @abstractmethod
     async def sign_message(self, message: Dict) -> Dict:
+        """
+        Returns a signed message from an Aleph message.
+        Args:
+            message: Message to sign
+        Returns:
+            Dict: Signed message
+        """
         raise NotImplementedError
 
     @abstractmethod
     def get_address(self) -> str:
+        """
+        Returns the account's displayed address.
+        """
         raise NotImplementedError
 
     @abstractmethod
     def get_public_key(self) -> str:
+        """
+        Returns the account's public key.
+        """
         raise NotImplementedError
 
-    async def encrypt(self, content) -> bytes:
+    async def encrypt(self, content: bytes) -> bytes:
+        """
+        Encrypts a message using the account's public key.
+        Args:
+            content: Content bytes to encrypt
+        Returns:
+            bytes: Encrypted content as bytes
+        """
         if self.CURVE == "secp256k1":
             value: bytes = encrypt(self.get_public_key(), content)
             return value
         else:
             raise NotImplementedError
 
-    async def decrypt(self, content) -> bytes:
+    async def decrypt(self, content: bytes) -> bytes:
+        """
+        Decrypts a message using the account's private key.
+        Args:
+            content: Content bytes to decrypt
+        Returns:
+            bytes: Decrypted content as bytes
+        """
         if self.CURVE == "secp256k1":
             value: bytes = decrypt(self.private_key, content)
             return value

src/aleph/sdk/chains/cosmos.py

@@ -1,6 +1,7 @@
 import base64
 import hashlib
 import json
+from typing import Union
 
 import ecdsa
 from cosmospy._wallet import privkey_to_address, privkey_to_pubkey
@@ -80,3 +81,12 @@ def get_public_key(self) -> str:
 
 def get_fallback_account(hrp=DEFAULT_HRP):
     return CSDKAccount(private_key=get_fallback_private_key(), hrp=hrp)
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+) -> bool:
+    """TODO: Implement this"""
+    raise NotImplementedError("Not implemented yet")

src/aleph/sdk/chains/ethereum.py

@@ -1,10 +1,12 @@
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 from eth_account import Account
 from eth_account.messages import encode_defunct
 from eth_account.signers.local import LocalAccount
+from eth_keys.exceptions import BadSignature as EthBadSignatureError
 
+from ..exceptions import BadSignatureError
 from .common import (
     BaseAccount,
     get_fallback_private_key,
@@ -41,3 +43,39 @@ def get_public_key(self) -> str:
 
 def get_fallback_account(path: Optional[Path] = None) -> ETHAccount:
     return ETHAccount(private_key=get_fallback_private_key(path=path))
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+):
+    """
+    Verifies a signature.
+    Args:
+        signature: The signature to verify. Can be a hex encoded string or bytes.
+        public_key: The sender's public key to use for verification. Can be a checksummed, hex encoded string or bytes.
+        message: The message to verify. Can be an utf-8 string or bytes.
+    Raises:
+        BadSignatureError: If the signature is invalid.
+    """
+    if isinstance(signature, str):
+        if signature.startswith("0x"):
+            signature = signature[2:]
+        signature = bytes.fromhex(signature)
+    else:
+        if signature.startswith(b"0x"):
+            signature = signature[2:]
+        signature = bytes.fromhex(signature.decode("utf-8"))
+    if isinstance(public_key, bytes):
+        public_key = "0x" + public_key.hex()
+    if isinstance(message, bytes):
+        message = message.decode("utf-8")
+
+    message_hash = encode_defunct(text=message)
+    try:
+        address = Account.recover_message(message_hash, signature=signature)
+        if address != public_key:
+            raise BadSignatureError
+    except (EthBadSignatureError, BadSignatureError) as e:
+        raise BadSignatureError from e

src/aleph/sdk/chains/nuls1.py

@@ -4,7 +4,7 @@
 import logging
 import struct
 from binascii import hexlify, unhexlify
-from typing import Optional
+from typing import Optional, Union
 
 from coincurve.keys import PrivateKey, PublicKey
 
@@ -324,3 +324,12 @@ def get_public_key(self):
 
 def get_fallback_account(chain_id=8964):
     return NULSAccount(private_key=get_fallback_private_key(), chain_id=chain_id)
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+) -> bool:
+    """TODO: Implement this"""
+    raise NotImplementedError("Not implemented yet")

src/aleph/sdk/chains/nuls2.py

@@ -1,4 +1,5 @@
 import base64
+from typing import Union
 
 from nuls2.model.data import (
     NETWORKS,
@@ -60,3 +61,12 @@ def get_public_key(self):
 def get_fallback_account(chain_id=1):
     acc = NULSAccount(private_key=get_fallback_private_key(), chain_id=chain_id)
     return acc
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+) -> bool:
+    """TODO: Implement this"""
+    raise NotImplementedError("Not implemented yet")

src/aleph/sdk/chains/sol.py

@@ -1,13 +1,15 @@
 import json
 import os
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 import base58
+from nacl.exceptions import BadSignatureError as NaclBadSignatureError
 from nacl.public import PrivateKey, SealedBox
-from nacl.signing import SigningKey
+from nacl.signing import SigningKey, VerifyKey
 
 from ..conf import settings
+from ..exceptions import BadSignatureError
 from .common import BaseAccount, get_verification_buffer
 
 
@@ -81,3 +83,29 @@ def get_fallback_private_key(path: Optional[Path] = None) -> bytes:
             # Create a symlink to use this key by default
             os.symlink(path, default_key_path)
     return private_key
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+):
+    """
+    Verifies a signature.
+    Args:
+        signature: The signature to verify. Can be a base58 encoded string or bytes.
+        public_key: The public key to use for verification. Can be a base58 encoded string or bytes.
+        message: The message to verify. Can be an utf-8 string or bytes.
+    Raises:
+        BadSignatureError: If the signature is invalid.
+    """
+    if isinstance(signature, str):
+        signature = base58.b58decode(signature)
+    if isinstance(message, str):
+        message = message.encode("utf-8")
+    if isinstance(public_key, str):
+        public_key = base58.b58decode(public_key)
+    try:
+        VerifyKey(public_key).verify(message, signature)
+    except NaclBadSignatureError as e:
+        raise BadSignatureError from e

src/aleph/sdk/chains/substrate.py

@@ -1,4 +1,5 @@
 import json
+from typing import Union
 
 from substrateinterface import Keypair
 
@@ -45,3 +46,12 @@ def get_fallback_mnemonics():
             prvfile.write(mnemonic)
 
     return mnemonic
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+) -> bool:
+    """TODO: Implement this"""
+    raise NotImplementedError("Not implemented yet")

src/aleph/sdk/chains/tezos.py

@@ -1,6 +1,6 @@
 import json
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 from aleph_pytezos.crypto.key import Key
 from nacl.public import SealedBox
@@ -48,3 +48,24 @@ async def decrypt(self, content) -> bytes:
 
 def get_fallback_account(path: Optional[Path] = None) -> TezosAccount:
     return TezosAccount(private_key=get_fallback_private_key(path=path))
+
+
+def verify_signature(
+    signature: Union[bytes, str],
+    public_key: Union[bytes, str],
+    message: Union[bytes, str],
+) -> bool:
+    """
+    Verify a signature using the public key (hash) of a tezos account.
+
+    Note: It requires the public key hash (sp, p2, ed-prefix), not the address (tz1, tz2 prefix)!
+    Args:
+        signature: The signature to verify. Can be a base58 encoded string or bytes.
+        public_key: The public key (hash) of the account. Can be a base58 encoded string or bytes.
+        message: The message that was signed. Is a sequence of bytes in raw format or hexadecimal notation.
+    """
+    key = Key.from_encoded_key(public_key)
+    try:
+        return key.verify(signature, message)
+    except ValueError:
+        return False

src/aleph/sdk/exceptions.py

@@ -34,3 +34,11 @@ class InvalidMessageError(BroadcastError):
     """
 
     pass
+
+
+class BadSignatureError(Exception):
+    """
+    The signature of a message is invalid.
+    """
+
+    pass

tests/unit/test_chain_ethereum.py

@@ -4,7 +4,9 @@
 
 import pytest
 
-from aleph.sdk.chains.ethereum import get_fallback_account
+from aleph.sdk.chains.common import get_verification_buffer
+from aleph.sdk.chains.ethereum import get_fallback_account, verify_signature
+from aleph.sdk.exceptions import BadSignatureError
 
 
 @dataclass
@@ -42,6 +44,48 @@ async def test_ETHAccount(ethereum_account):
     assert len(pubkey) == 68
 
 
+@pytest.mark.asyncio
+async def test_verify_signature(ethereum_account):
+    account = ethereum_account
+
+    message = asdict(
+        Message(
+            "ETH",
+            account.get_address(),
+            "POST",
+            "SomeHash",
+        )
+    )
+    await account.sign_message(message)
+    assert message["signature"]
+
+    verify_signature(
+        message["signature"], message["sender"], get_verification_buffer(message)
+    )
+
+
+@pytest.mark.asyncio
+async def test_verify_signature_with_forged_signature(ethereum_account):
+    account = ethereum_account
+
+    message = asdict(
+        Message(
+            "ETH",
+            account.get_address(),
+            "POST",
+            "SomeHash",
+        )
+    )
+    await account.sign_message(message)
+    assert message["signature"]
+
+    forged_signature = "0x" + "0" * 130
+    with pytest.raises(BadSignatureError):
+        verify_signature(
+            forged_signature, message["sender"], get_verification_buffer(message)
+        )
+
+
 @pytest.mark.asyncio
 async def test_decrypt_secp256k1(ethereum_account):
     account = ethereum_account