Add sign_raw() (#50)

`BaseAccount.sign_raw()` accepts raw bytes and outputs a signature as a string in the preferred encoding.

This allows signing of arbitrary data besides Aleph messages, making it a useful tool in dApp development.

Author: MHHukiewitz

setup.cfg

@@ -79,6 +79,8 @@ testing =
     black
     isort
     flake8
+    substrate-interface
+    py-sr25519-bindings
 mqtt =
     aiomqtt<=0.1.3
     certifi
@@ -90,7 +92,8 @@ ethereum =
     # Required to fix a dependency issue with parsimonious and Python3.11
     eth_abi==4.0.0b2; python_version>="3.11"
 polkadot =
-    substrate-interface==1.3.4
+    substrate-interface
+    py-sr25519-bindings
 cosmos =
     cosmospy
 solana =

src/aleph/sdk/chains/common.py

@@ -62,7 +62,6 @@ def _setup_sender(self, message: Dict) -> Dict:
         else:
             raise ValueError("Message sender does not match the account's public key.")
 
-    @abstractmethod
     async def sign_message(self, message: Dict) -> Dict:
         """
         Returns a signed message from an Aleph message.
@@ -71,6 +70,20 @@ async def sign_message(self, message: Dict) -> Dict:
         Returns:
             Dict: Signed message
         """
+        message = self._setup_sender(message)
+        signature = await self.sign_raw(get_verification_buffer(message))
+        message["signature"] = signature.hex()
+        return message
+
+    @abstractmethod
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        """
+        Returns a signed message from a raw buffer.
+        Args:
+            buffer: Buffer to sign
+        Returns:
+            bytes: Signature in preferred format
+        """
         raise NotImplementedError
 
     @abstractmethod
@@ -143,3 +156,10 @@ def get_fallback_private_key(path: Optional[Path] = None) -> bytes:
         if not default_key_path.exists():
             default_key_path.symlink_to(path)
     return private_key
+
+
+def bytes_from_hex(hex_string: str) -> bytes:
+    if hex_string.startswith("0x"):
+        hex_string = hex_string[2:]
+    hex_string = bytes.fromhex(hex_string)
+    return hex_string

src/aleph/sdk/chains/cosmos.py

@@ -51,28 +51,30 @@ def __init__(self, private_key=None, hrp=DEFAULT_HRP):
 
     async def sign_message(self, message):
         message = self._setup_sender(message)
-
         verif = get_verification_string(message)
-
-        privkey = ecdsa.SigningKey.from_string(self.private_key, curve=ecdsa.SECP256k1)
-        signature_compact = privkey.sign_deterministic(
-            verif.encode("utf-8"),
-            hashfunc=hashlib.sha256,
-            sigencode=ecdsa.util.sigencode_string_canonize,
-        )
-        signature_base64_str = base64.b64encode(signature_compact).decode("utf-8")
         base64_pubkey = base64.b64encode(self.get_public_key().encode()).decode("utf-8")
+        signature = await self.sign_raw(verif.encode("utf-8"))
 
         sig = {
-            "signature": signature_base64_str,
+            "signature": signature.decode("utf-8"),
             "pub_key": {"type": "tendermint/PubKeySecp256k1", "value": base64_pubkey},
             "account_number": str(0),
             "sequence": str(0),
         }
         message["signature"] = json.dumps(sig)
         return message
 
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        privkey = ecdsa.SigningKey.from_string(self.private_key, curve=ecdsa.SECP256k1)
+        signature_compact = privkey.sign_deterministic(
+            buffer,
+            hashfunc=hashlib.sha256,
+            sigencode=ecdsa.util.sigencode_string_canonize,
+        )
+        return base64.b64encode(signature_compact)
+
     def get_address(self) -> str:
+        # WARNING: Fails with OpenSSL >= 3.2.0 due to deprecation of ripemd160
         return privkey_to_address(self.private_key)
 
     def get_public_key(self) -> str:

src/aleph/sdk/chains/ethereum.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Dict, Optional, Union
+from typing import Optional, Union
 
 from eth_account import Account
 from eth_account.messages import encode_defunct
@@ -9,9 +9,9 @@
 from ..exceptions import BadSignatureError
 from .common import (
     BaseAccount,
+    bytes_from_hex,
     get_fallback_private_key,
     get_public_key,
-    get_verification_buffer,
 )
 
 
@@ -24,15 +24,11 @@ def __init__(self, private_key: bytes):
         self.private_key = private_key
         self._account = Account.from_key(self.private_key)
 
-    async def sign_message(self, message: Dict) -> Dict:
-        """Sign a message inplace."""
-        message = self._setup_sender(message)
-
-        msghash = encode_defunct(text=get_verification_buffer(message).decode("utf-8"))
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        """Sign a raw buffer."""
+        msghash = encode_defunct(text=buffer.decode("utf-8"))
         sig = self._account.sign_message(msghash)
-
-        message["signature"] = sig["signature"].hex()
-        return message
+        return sig["signature"]
 
     def get_address(self) -> str:
         return self._account.address
@@ -60,19 +56,14 @@ def verify_signature(
         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"))
+        signature = bytes_from_hex(signature)
     if isinstance(public_key, bytes):
         public_key = "0x" + public_key.hex()
     if isinstance(message, bytes):
-        message = message.decode("utf-8")
+        message_hash = encode_defunct(primitive=message)
+    else:
+        message_hash = encode_defunct(text=message)
 
-    message_hash = encode_defunct(text=message)
     try:
         address = Account.recover_message(message_hash, signature=signature)
         if address.casefold() != public_key.casefold():

src/aleph/sdk/chains/nuls1.py

@@ -315,6 +315,10 @@ async def sign_message(self, message):
         message["signature"] = sig.serialize().hex()
         return message
 
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        sig = NulsSignature.sign_data(self.private_key, buffer)
+        return sig.serialize()
+
     def get_address(self):
         return address_from_hash(
             public_key_to_hash(self.get_public_key(), chain_id=self.chain_id)

src/aleph/sdk/chains/nuls2.py

@@ -1,5 +1,5 @@
 import base64
-from typing import Union
+from typing import Dict, Union
 
 from nuls2.model.data import (
     NETWORKS,
@@ -37,17 +37,23 @@ def __init__(self, private_key=None, chain_id=1, prefix=None):
         else:
             self.prefix = prefix
 
-    async def sign_message(self, message):
-        # sig = NulsSignature.sign_message(self.private_key,
-        #                                  get_verification_buffer(message))
+    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
+        """
         message = self._setup_sender(message)
-
-        sig = sign_recoverable_message(
-            self.private_key, get_verification_buffer(message)
-        )
-        message["signature"] = base64.b64encode(sig).decode()
+        signature = await self.sign_raw(get_verification_buffer(message))
+        message["signature"] = signature.decode()
         return message
 
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        sig = sign_recoverable_message(self.private_key, buffer)
+        return base64.b64encode(sig)
+
     def get_address(self):
         return address_from_hash(
             public_key_to_hash(self.get_public_key(), chain_id=self.chain_id),

src/aleph/sdk/chains/remote.py

@@ -77,6 +77,9 @@ async def sign_message(self, message: Dict) -> Dict:
             response.raise_for_status()
             return await response.json()
 
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        raise NotImplementedError()
+
     def get_address(self) -> str:
         return self._address
 

src/aleph/sdk/chains/sol.py

@@ -30,13 +30,19 @@ async def sign_message(self, message: Dict) -> Dict:
         """Sign a message inplace."""
         message = self._setup_sender(message)
         verif = get_verification_buffer(message)
+        signature = await self.sign_raw(verif)
         sig = {
             "publicKey": self.get_address(),
-            "signature": encode(self._signing_key.sign(verif).signature),
+            "signature": encode(signature),
         }
         message["signature"] = json.dumps(sig)
         return message
 
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        """Sign a raw buffer."""
+        sig = self._signing_key.sign(buffer)
+        return sig.signature
+
     def get_address(self) -> str:
         return encode(self._signing_key.verify_key)
 

src/aleph/sdk/chains/substrate.py

@@ -1,47 +1,73 @@
 import json
-from typing import Union
+import logging
+from pathlib import Path
+from typing import Optional, Union
 
+from sr25519 import verify
 from substrateinterface import Keypair
+from substrateinterface.utils.ss58 import ss58_decode
 
 from ..conf import settings
-from .common import BaseAccount, get_verification_buffer
+from ..exceptions import BadSignatureError
+from .common import BaseAccount, bytes_from_hex, get_verification_buffer
+
+logger = logging.getLogger(__name__)
 
 
 class DOTAccount(BaseAccount):
     CHAIN = "DOT"
     CURVE = "sr25519"
 
-    def __init__(self, mnemonics=None, address_type=42):
+    def __init__(self, mnemonics: str, address_type=42):
         self.mnemonics = mnemonics
         self.address_type = address_type
         self._account = Keypair.create_from_mnemonic(
-            self.mnemonics, address_type=address_type
+            self.mnemonics, ss58_format=address_type
         )
 
     async def sign_message(self, message):
         message = self._setup_sender(message)
         verif = get_verification_buffer(message).decode("utf-8")
-        sig = {"curve": self.CURVE, "data": self._account.sign(verif)}
+        signature = await self.sign_raw(verif.encode("utf-8"))
+        sig = {"curve": self.CURVE, "data": signature.hex()}
         message["signature"] = json.dumps(sig)
         return message
 
-    def get_address(self):
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        return self._account.sign(buffer)
+
+    def get_address(self) -> str:
         return self._account.ss58_address
 
-    def get_public_key(self):
-        return self._account.public_key
+    def get_public_key(self) -> str:
+        return "0x" + self._account.public_key.hex()
 
 
-def get_fallback_account():
-    return DOTAccount(mnemonics=get_fallback_mnemonics())
+def get_fallback_account(path: Optional[Path] = None) -> DOTAccount:
+    return DOTAccount(mnemonics=get_fallback_mnemonics(path))
 
 
-def get_fallback_mnemonics():
-    try:
-        mnemonic = settings.PRIVATE_KEY_FILE.read_text()
-    except OSError:
+def get_fallback_mnemonics(path: Optional[Path] = None) -> str:
+    path = path or settings.PRIVATE_MNEMONIC_FILE
+    if path.exists() and path.stat().st_size > 0:
+        mnemonic = path.read_text()
+    else:
         mnemonic = Keypair.generate_mnemonic()
-        settings.PRIVATE_KEY_FILE.write_text(mnemonic)
+        path.parent.mkdir(exist_ok=True, parents=True)
+        path.write_text(mnemonic)
+        default_mnemonic_path = path.parent / "default.mnemonic"
+
+        # If the symlink exists but does not point to a file, delete it.
+        if (
+            default_mnemonic_path.is_symlink()
+            and not default_mnemonic_path.resolve().exists()
+        ):
+            default_mnemonic_path.unlink()
+            logger.warning("The symlink to the mnemonic is broken")
+
+        # Create a symlink to use this mnemonic by default
+        if not default_mnemonic_path.exists():
+            default_mnemonic_path.symlink_to(path)
 
     return mnemonic
 
@@ -50,6 +76,26 @@ 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")
+) -> None:
+    if isinstance(signature, str):
+        signature = bytes_from_hex(signature)
+    if isinstance(public_key, str):
+        public_key = bytes_from_hex(public_key)
+    if isinstance(message, str):
+        message = message.encode()
+
+    try:
+        # Another attempt with the data wrapped, as discussed in https://github.com/polkadot-js/extension/pull/743
+        if not verify(signature, message, public_key) or verify(
+            signature, b"<Bytes>" + message + b"</Bytes>", public_key
+        ):
+            raise BadSignatureError
+    except Exception as e:
+        raise BadSignatureError from e
+
+
+def verify_signature_with_ss58_address(
+    signature: Union[bytes, str], address: str, message: Union[bytes, str]
+) -> None:
+    address_bytes = ss58_decode(address)
+    return verify_signature(signature, address_bytes, message)

src/aleph/sdk/chains/tezos.py

@@ -25,14 +25,18 @@ async def sign_message(self, message: Dict) -> Dict:
         message = self._setup_sender(message)
 
         verif = get_verification_buffer(message)
+        signature = await self.sign_raw(verif)
         sig = {
             "publicKey": self.get_public_key(),
-            "signature": self._account.sign(verif),
+            "signature": signature.decode(),
         }
 
         message["signature"] = json.dumps(sig)
         return message
 
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        return self._account.sign(buffer).encode()
+
     def get_address(self) -> str:
         return self._account.public_key_hash()