feat: add support for vertex headers (#52)

Author: glevco

hathorlib/base_transaction.py

@@ -19,6 +19,7 @@
 from hathorlib.exceptions import InvalidOutputValue, WeightError
 from hathorlib.scripts import P2PKH, DataScript, MultiSig, parse_address_script
 from hathorlib.utils import int_to_bytes, unpack, unpack_len
+from hathorlib.vertex_parser import VertexParser
 
 settings = HathorSettings()
 
@@ -81,14 +82,14 @@ def _missing_(cls, value: Any) -> None:
 
     def get_cls(self) -> Type['BaseTransaction']:
         from hathorlib import Block, TokenCreationTransaction, Transaction
-        from hathorlib.nanocontracts.nanocontract import NanoContract
+        from hathorlib.nanocontracts.nanocontract import DeprecatedNanoContract
         from hathorlib.nanocontracts.on_chain_blueprint import OnChainBlueprint
 
         cls_map: Dict[TxVersion, Type[BaseTransaction]] = {
             TxVersion.REGULAR_BLOCK: Block,
             TxVersion.REGULAR_TRANSACTION: Transaction,
             TxVersion.TOKEN_CREATION_TRANSACTION: TokenCreationTransaction,
-            TxVersion.NANO_CONTRACT: NanoContract,
+            TxVersion.NANO_CONTRACT: DeprecatedNanoContract,
             TxVersion.ON_CHAIN_BLUEPRINT: OnChainBlueprint,
         }
 
@@ -103,6 +104,10 @@ def get_cls(self) -> Type['BaseTransaction']:
 class BaseTransaction(ABC):
     """Hathor base transaction"""
 
+    __slots__ = (
+        'version', 'signal_bits', 'weight', 'timestamp', 'nonce', 'inputs', 'outputs', 'parents', 'hash', 'headers'
+    )
+
     # Even though nonce is serialized with different sizes for tx and blocks
     # the same size is used for hashes to enable mining algorithm compatibility
     SERIALIZATION_NONCE_SIZE: ClassVar[int]
@@ -116,6 +121,7 @@ class BaseTransaction(ABC):
     signal_bits: int
 
     def __init__(self) -> None:
+        from hathorlib.headers import VertexBaseHeader
         self.nonce: int = 0
         self.timestamp: int = 0
         self.signal_bits: int = 0
@@ -125,6 +131,7 @@ def __init__(self) -> None:
         self.outputs: List['TxOutput'] = []
         self.parents: List[bytes] = []
         self.hash: bytes = b''
+        self.headers: list[VertexBaseHeader] = []
 
     @property
     @abstractmethod
@@ -136,6 +143,10 @@ def is_block(self) -> bool:
     def is_transaction(self) -> bool:
         raise NotImplementedError
 
+    def is_nano_contract(self) -> bool:
+        """Return True if this transaction is a nano contract or not."""
+        return False
+
     def _get_formatted_fields_dict(self, short: bool = True) -> Dict[str, str]:
         """ Used internally on __repr__ and __str__, returns a dict of `field_name: formatted_value`.
         """
@@ -186,6 +197,21 @@ def get_fields_from_struct(self, struct_bytes: bytes) -> bytes:
         buf = self.get_graph_fields_from_struct(buf)
         return buf
 
+    def get_header_from_bytes(self, buf: bytes) -> bytes:
+        """Parse bytes and return the next header in buffer."""
+        if len(self.headers) >= self.get_maximum_number_of_headers():
+            raise ValueError('too many headers')
+
+        header_type = buf[:1]
+        header_class = VertexParser.get_header_parser(header_type)
+        header, buf = header_class.deserialize(self, buf)
+        self.headers.append(header)
+        return buf
+
+    def get_maximum_number_of_headers(self) -> int:
+        """Return the maximum number of headers for this vertex."""
+        return 1
+
     @classmethod
     @abstractmethod
     def create_from_struct(cls, struct_bytes: bytes) -> 'BaseTransaction':
@@ -294,6 +320,10 @@ def get_graph_struct(self) -> bytes:
             struct_bytes += parent
         return struct_bytes
 
+    def get_headers_struct(self) -> bytes:
+        """Return the serialization of the headers only."""
+        return b''.join(h.serialize() for h in self.headers)
+
     def get_struct_without_nonce(self) -> bytes:
         """Return a partial serialization of the transaction, without including the nonce field
 
@@ -321,6 +351,7 @@ def get_struct(self) -> bytes:
         """
         struct_bytes = self.get_struct_without_nonce()
         struct_bytes += self.get_struct_nonce()
+        struct_bytes += self.get_headers_struct()
         return struct_bytes
 
     def verify_pow(self, override_weight: Optional[float] = None) -> bool:
@@ -355,13 +386,22 @@ def get_graph_hash(self) -> bytes:
         graph_hash.update(self.get_graph_struct())
         return graph_hash.digest()
 
-    def get_header_without_nonce(self) -> bytes:
+    def get_headers_hash(self) -> bytes:
+        """Return the sha256 of the headers of the transaction."""
+        if not self.headers:
+            return b''
+
+        h = hashlib.sha256()
+        h.update(self.get_headers_struct())
+        return h.digest()
+
+    def get_mining_header_without_nonce(self) -> bytes:
         """Return the transaction header without the nonce
 
         :return: transaction header without the nonce
         :rtype: bytes
         """
-        return self.get_funds_hash() + self.get_graph_hash()
+        return self.get_funds_hash() + self.get_graph_hash() + self.get_headers_hash()
 
     def calculate_hash1(self) -> HASH:
         """Return the sha256 of the transaction without including the `nonce`
@@ -370,7 +410,7 @@ def calculate_hash1(self) -> HASH:
         :rtype: :py:class:`_hashlib.HASH`
         """
         calculate_hash1 = hashlib.sha256()
-        calculate_hash1.update(self.get_header_without_nonce())
+        calculate_hash1.update(self.get_mining_header_without_nonce())
         return calculate_hash1
 
     def calculate_hash2(self, part1: HASH) -> bytes:

hathorlib/block.py

@@ -42,10 +42,15 @@ def create_from_struct(cls, struct_bytes: bytes) -> 'Block':
         blc = cls()
         buf = blc.get_fields_from_struct(struct_bytes)
 
-        blc.nonce = int.from_bytes(buf, byteorder='big')
-        if len(buf) != cls.SERIALIZATION_NONCE_SIZE:
+        if len(buf) < cls.SERIALIZATION_NONCE_SIZE:
             raise ValueError('Invalid sequence of bytes')
 
+        blc.nonce = int.from_bytes(buf[:cls.SERIALIZATION_NONCE_SIZE], byteorder='big')
+        buf = buf[cls.SERIALIZATION_NONCE_SIZE:]
+
+        while buf:
+            buf = blc.get_header_from_bytes(buf)
+
         blc.hash = blc.calculate_hash()
 
         return blc

hathorlib/headers/__init__.py

@@ -0,0 +1,24 @@
+# Copyright 2023 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from hathorlib.headers.base import VertexBaseHeader
+from hathorlib.headers.nano_header import NC_INITIALIZE_METHOD, NanoHeader
+from hathorlib.headers.types import VertexHeaderId
+
+__all__ = [
+    'VertexBaseHeader',
+    'VertexHeaderId',
+    'NanoHeader',
+    'NC_INITIALIZE_METHOD',
+]

hathorlib/headers/base.py

@@ -0,0 +1,39 @@
+# Copyright 2023 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from hathorlib.base_transaction import BaseTransaction
+
+
+class VertexBaseHeader(ABC):
+    @classmethod
+    @abstractmethod
+    def deserialize(cls, tx: BaseTransaction, buf: bytes) -> tuple[VertexBaseHeader, bytes]:
+        """Deserialize header from `buf` which starts with header id."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def serialize(self) -> bytes:
+        """Serialize header with header id as prefix."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_sighash_bytes(self) -> bytes:
+        """Return sighash bytes to check digital signatures."""
+        raise NotImplementedError

hathorlib/headers/nano_header.py

@@ -0,0 +1,157 @@
+# Copyright 2023 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from hathorlib.headers.base import VertexBaseHeader
+from hathorlib.headers.types import VertexHeaderId
+from hathorlib.nanocontracts import DeprecatedNanoContract
+from hathorlib.nanocontracts.types import NCActionType
+from hathorlib.utils import int_to_bytes, unpack, unpack_len
+
+if TYPE_CHECKING:
+    from hathorlib.base_transaction import BaseTransaction
+
+NC_VERSION = 1
+NC_INITIALIZE_METHOD = 'initialize'
+
+
+@dataclass(frozen=True)
+class NanoHeaderAction:
+    type: NCActionType
+    token_index: int
+    amount: int
+
+
+@dataclass(frozen=True)
+class NanoHeader(VertexBaseHeader):
+    tx: BaseTransaction
+
+    # nc_id equals to the blueprint_id when a Nano Contract is being created.
+    # nc_id equals to the nanocontract_id when a method is being called.
+    nc_id: bytes
+
+    # Name of the method to be called. When creating a new Nano Contract, it must be equal to 'initialize'.
+    nc_method: str
+
+    # Serialized arguments to nc_method.
+    nc_args_bytes: bytes
+
+    nc_actions: list[NanoHeaderAction]
+
+    # Pubkey and signature of the transaction owner / caller.
+    nc_pubkey: bytes
+    nc_signature: bytes
+
+    nc_version: int = NC_VERSION
+
+    @classmethod
+    def _deserialize_action(cls, buf: bytes) -> tuple[NanoHeaderAction, bytes]:
+        from hathorlib.base_transaction import bytes_to_output_value
+        type_bytes, buf = buf[:1], buf[1:]
+        action_type = NCActionType.from_bytes(type_bytes)
+        (token_index,), buf = unpack('!B', buf)
+        amount, buf = bytes_to_output_value(buf)
+        return NanoHeaderAction(
+            type=action_type,
+            token_index=token_index,
+            amount=amount,
+        ), buf
+
+    @classmethod
+    def deserialize(cls, tx: BaseTransaction, buf: bytes) -> tuple[NanoHeader, bytes]:
+        header_id, buf = buf[:1], buf[1:]
+        assert header_id == VertexHeaderId.NANO_HEADER.value
+        (nc_version,), buf = unpack('!B', buf)
+        if nc_version != NC_VERSION:
+            raise ValueError('unknown nanocontract version: {}'.format(nc_version))
+
+        nc_id, buf = unpack_len(32, buf)
+        (nc_method_len,), buf = unpack('!B', buf)
+        nc_method, buf = unpack_len(nc_method_len, buf)
+        (nc_args_bytes_len,), buf = unpack('!H', buf)
+        nc_args_bytes, buf = unpack_len(nc_args_bytes_len, buf)
+
+        nc_actions: list[NanoHeaderAction] = []
+        if not isinstance(tx, DeprecatedNanoContract):
+            (nc_actions_len,), buf = unpack('!B', buf)
+            for _ in range(nc_actions_len):
+                action, buf = cls._deserialize_action(buf)
+                nc_actions.append(action)
+
+        (nc_pubkey_len,), buf = unpack('!B', buf)
+        nc_pubkey, buf = unpack_len(nc_pubkey_len, buf)
+        (nc_signature_len,), buf = unpack('!B', buf)
+        nc_signature, buf = unpack_len(nc_signature_len, buf)
+
+        decoded_nc_method = nc_method.decode('ascii')
+
+        return cls(
+            tx=tx,
+            nc_version=nc_version,
+            nc_id=nc_id,
+            nc_method=decoded_nc_method,
+            nc_args_bytes=nc_args_bytes,
+            nc_actions=nc_actions,
+            nc_pubkey=nc_pubkey,
+            nc_signature=nc_signature,
+        ), bytes(buf)
+
+    def _serialize_action(self, action: NanoHeaderAction) -> bytes:
+        from hathorlib.base_transaction import output_value_to_bytes
+        ret = [
+            action.type.to_bytes(),
+            int_to_bytes(action.token_index, 1),
+            output_value_to_bytes(action.amount),
+        ]
+        return b''.join(ret)
+
+    def _serialize_without_header_id(self, *, skip_signature: bool) -> deque[bytes]:
+        """Serialize the header with the option to skip the signature."""
+        encoded_method = self.nc_method.encode('ascii')
+
+        ret: deque[bytes] = deque()
+        ret.append(int_to_bytes(NC_VERSION, 1))
+        ret.append(self.nc_id)
+        ret.append(int_to_bytes(len(encoded_method), 1))
+        ret.append(encoded_method)
+        ret.append(int_to_bytes(len(self.nc_args_bytes), 2))
+        ret.append(self.nc_args_bytes)
+
+        if not isinstance(self.tx, DeprecatedNanoContract):
+            ret.append(int_to_bytes(len(self.nc_actions), 1))
+            for action in self.nc_actions:
+                ret.append(self._serialize_action(action))
+
+        ret.append(int_to_bytes(len(self.nc_pubkey), 1))
+        ret.append(self.nc_pubkey)
+        if not skip_signature:
+            ret.append(int_to_bytes(len(self.nc_signature), 1))
+            ret.append(self.nc_signature)
+        else:
+            ret.append(int_to_bytes(0, 1))
+        return ret
+
+    def serialize(self) -> bytes:
+        ret = self._serialize_without_header_id(skip_signature=False)
+        ret.appendleft(VertexHeaderId.NANO_HEADER.value)
+        return b''.join(ret)
+
+    def get_sighash_bytes(self) -> bytes:
+        ret = self._serialize_without_header_id(skip_signature=True)
+        return b''.join(ret)

hathorlib/headers/types.py

@@ -0,0 +1,20 @@
+# Copyright 2023 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from enum import Enum, unique
+
+
+@unique
+class VertexHeaderId(Enum):
+    NANO_HEADER = b'\x10'