Feature: support Micheline-style signatures for Tezos

src/aleph/chains/tezos.py

@@ -1,5 +1,7 @@
+import datetime as dt
 import json
 import logging
+from enum import Enum
 
 from aleph_pytezos.crypto.key import Key
 
@@ -10,13 +12,83 @@
 LOGGER = logging.getLogger(__name__)
 CHAIN_NAME = "TEZOS"
 
+# Default dApp URL for Micheline-style signatures
+DEFAULT_DAPP_URL = "aleph.im"
+
+
+class TezosSignatureType(str, Enum):
+    RAW = "raw"
+    MICHELINE = "micheline"
+
+
+def timestamp_to_iso_8601(timestamp: float) -> str:
+    """
+    Returns the timestamp formatted to ISO-8601, JS-style.
+
+    Compared to the regular `isoformat()`, this function only provides precision down
+    to milliseconds and prints a "Z" instead of +0000 for UTC.
+    This format is typically used by JavaScript applications, like our TS SDK.
+
+    Example: 2022-09-23T14:41:19.029Z
+
+    :param timestamp: The timestamp to format.
+    :return: The formatted timestamp.
+    """
+
+    return (
+        dt.datetime.utcfromtimestamp(timestamp).isoformat(timespec="milliseconds") + "Z"
+    )
+
+
+def micheline_verification_buffer(
+    verification_buffer: bytes,
+    timestamp: float,
+    dapp_url: str,
+) -> bytes:
+    """
+    Computes the verification buffer for Micheline-type signatures.
+
+    This verification buffer is used when signing data with a Tezos web wallet.
+    See https://tezostaquito.io/docs/signing/#generating-a-signature-with-beacon-sdk.
+
+    :param verification_buffer: The original (non-Tezos) verification buffer for the Aleph message.
+    :param timestamp: Timestamp of the message.
+    :param dapp_url: The URL of the dApp, for use as part of the verification buffer.
+    :return: The verification buffer used for the signature by the web wallet.
+    """
+
+    prefix = b"Tezos Signed Message:"
+    timestamp = timestamp_to_iso_8601(timestamp).encode("utf-8")
+
+    payload = b" ".join(
+        (prefix, dapp_url.encode("utf-8"), timestamp, verification_buffer)
+    )
+    hex_encoded_payload = payload.hex()
+    payload_size = str(len(hex_encoded_payload)).encode("utf-8")
+
+    return b"\x05" + b"\x01\x00" + payload_size + payload
+
+
+def get_tezos_verification_buffer(
+    message: BasePendingMessage, signature_type: TezosSignatureType, dapp_url: str
+) -> bytes:
+    verification_buffer = get_verification_buffer(message)
+
+    if signature_type == TezosSignatureType.RAW:
+        return verification_buffer
+    elif signature_type == TezosSignatureType.MICHELINE:
+        return micheline_verification_buffer(
+            verification_buffer, message.time, dapp_url
+        )
+
+    raise ValueError(f"Unsupported signature type: {signature_type}")
+
 
 async def verify_signature(message: BasePendingMessage) -> bool:
     """
     Verifies the cryptographic signature of a message signed with a Tezos key.
     """
 
-    verification_buffer = get_verification_buffer(message)
     try:
         signature_dict = json.loads(message.signature)
     except json.JSONDecodeError:
@@ -30,6 +102,9 @@ async def verify_signature(message: BasePendingMessage) -> bool:
         LOGGER.exception("'%s' key missing from Tezos signature dictionary.", e.args[0])
         return False
 
+    signature_type = TezosSignatureType(signature_dict.get("signingType", "raw"))
+    dapp_url = signature_dict.get("dAppUrl", DEFAULT_DAPP_URL)
+
     key = Key.from_encoded_key(public_key)
     # Check that the sender ID is equal to the public key hash
     public_key_hash = key.public_key_hash()
@@ -41,6 +116,10 @@ async def verify_signature(message: BasePendingMessage) -> bool:
             public_key_hash,
         )
 
+    verification_buffer = get_tezos_verification_buffer(
+        message, signature_type, dapp_url
+    )
+
     # Check the signature
     try:
         key.verify(signature, verification_buffer)

tests/chains/test_tezos.py

@@ -2,10 +2,13 @@
 
 from aleph.network import verify_signature
 from aleph.schemas.pending_messages import parse_message
+from aleph.chains import (
+    tezos,
+)  # TODO: this import is currently necessary because of circular dependencies
 
 
 @pytest.mark.asyncio
-async def test_tezos_verify_signature():
+async def test_tezos_verify_signature_raw():
     message_dict = {
         "chain": "TEZOS",
         "channel": "TEST",
@@ -28,7 +31,7 @@ async def test_tezos_verify_signature():
 
 
 @pytest.mark.asyncio
-async def test_tezos_verify_signature_ed25519():
+async def test_tezos_verify_signature_raw_ed25519():
     message_dict = {
         "chain": "TEZOS",
         "sender": "tz1SmGHzna3YhKropa3WudVq72jhTPDBn4r5",
@@ -43,3 +46,20 @@ async def test_tezos_verify_signature_ed25519():
 
     message = parse_message(message_dict)
     await verify_signature(message)
+
+
+@pytest.mark.asyncio
+async def test_tezos_verify_signature_micheline():
+    message_dict = {
+        "chain": "TEZOS",
+        "sender": "tz1VrPqrVdMFsgykWyhGH7SYcQ9avHTjPcdD",
+        "type": "POST",
+        "channel": "ALEPH-TEST",
+        "signature": '{"signingType":"micheline","signature":"sigXD8iT5ivdawgPzE1AbtDwqqAjJhS5sHS1psyE74YjfiaQnxWZsATNjncdsuQw3b9xaK79krxtsC8uQoT5TcUXmo66aovT","publicKey":"edpkvapDnjnasrNcmUdMZXhQZwpX6viPyuGCq6nrP4W7ZJCm7EFTpS"}',
+        "time": 1663944079.029,
+        "item_type": "storage",
+        "item_hash": "72b2722b95582419cfa71f631ff6c6afc56344dc6a4609e772877621813040b7",
+    }
+
+    message = parse_message(message_dict)
+    await verify_signature(message)