diff --git a/src/aleph/chains/tezos.py b/src/aleph/chains/tezos.py index 765f1ddc7..f061378c9 100644 --- a/src/aleph/chains/tezos.py +++ b/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) diff --git a/tests/chains/test_tezos.py b/tests/chains/test_tezos.py index 4c6be4c37..14ca20402 100644 --- a/tests/chains/test_tezos.py +++ b/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)