feat: fee header support

Author: raul-oliveira

hathorlib/base_transaction.py

@@ -147,6 +147,10 @@ def is_nano_contract(self) -> bool:
         """Return True if this transaction is a nano contract or not."""
         return False
 
+    def has_fees(self) -> bool:
+        """Return True if this transaction has fees 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`.
         """

hathorlib/headers/__init__.py

@@ -14,6 +14,7 @@
 
 from hathorlib.headers.base import VertexBaseHeader
 from hathorlib.headers.deprecated_nano_header import DeprecatedNanoHeader
+from hathorlib.headers.fee_header import FeeEntry, FeeHeader, FeeHeaderEntry
 from hathorlib.headers.nano_header import NC_INITIALIZE_METHOD, NanoHeader
 from hathorlib.headers.types import VertexHeaderId
 
@@ -22,5 +23,8 @@
     'VertexHeaderId',
     'NanoHeader',
     'DeprecatedNanoHeader',
+    'FeeHeader',
+    'FeeHeaderEntry',
+    'FeeEntry',
     'NC_INITIALIZE_METHOD',
 ]

hathorlib/headers/fee_header.py

@@ -0,0 +1,94 @@
+# 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 dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from hathorlib.headers.base import VertexBaseHeader
+from hathorlib.headers.types import VertexHeaderId
+from hathorlib.utils import int_to_bytes, unpack
+
+if TYPE_CHECKING:
+    from hathorlib.base_transaction import BaseTransaction
+    from hathorlib.transaction import Transaction
+
+
+@dataclass(frozen=True)
+class FeeHeaderEntry:
+    token_index: int
+    amount: int
+
+
+@dataclass(frozen=True)
+class FeeEntry:
+    token_uid: bytes
+    amount: int
+
+
+@dataclass(frozen=True)
+class FeeHeader(VertexBaseHeader):
+    tx: Transaction
+    fees: list[FeeHeaderEntry]
+
+    @classmethod
+    def deserialize(cls, tx: BaseTransaction, buf: bytes) -> tuple[FeeHeader, bytes]:
+        from hathorlib.base_transaction import bytes_to_output_value
+
+        header_id, buf = buf[:1], buf[1:]
+        assert header_id == VertexHeaderId.FEE_HEADER.value
+
+        fees: list[FeeHeaderEntry] = []
+        (fees_len,), buf = unpack('!B', buf)
+
+        for _ in range(fees_len):
+            (token_index,), buf = unpack('!B', buf)
+            amount, buf = bytes_to_output_value(buf)
+            fees.append(FeeHeaderEntry(
+                token_index=token_index,
+                amount=amount,
+            ))
+        from hathorlib.transaction import Transaction
+        assert isinstance(tx, Transaction)
+        return cls(
+            tx=tx,
+            fees=fees,
+        ), bytes(buf)
+
+    def serialize(self) -> bytes:
+        from hathorlib.base_transaction import output_value_to_bytes
+
+        ret = [
+            VertexHeaderId.FEE_HEADER.value,
+            int_to_bytes(len(self.fees), 1)
+        ]
+
+        for fee in self.fees:
+            ret.append(int_to_bytes(fee.token_index, 1))
+            ret.append(output_value_to_bytes(fee.amount))
+
+        return b''.join(ret)
+
+    def get_sighash_bytes(self) -> bytes:
+        return self.serialize()
+
+    def get_fees(self) -> list[FeeEntry]:
+        return [
+            FeeEntry(
+                token_uid=self.tx.get_token_uid(fee.token_index),
+                amount=fee.amount
+            )
+            for fee in self.fees
+        ]

hathorlib/headers/nano_header.py

@@ -20,12 +20,11 @@
 
 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 decode_unsigned, encode_unsigned, int_to_bytes, unpack, unpack_len
 
 if TYPE_CHECKING:
     from hathorlib.base_transaction import BaseTransaction
+    from hathorlib.nanocontracts.types import NCActionType
 
 NC_INITIALIZE_METHOD = 'initialize'
 ADDRESS_LEN_BYTES = 25
@@ -35,7 +34,7 @@
 
 @dataclass(frozen=True)
 class NanoHeaderAction:
-    type: NCActionType
+    type: 'NCActionType'
     token_index: int
     amount: int
 
@@ -66,6 +65,8 @@ class NanoHeader(VertexBaseHeader):
     @classmethod
     def _deserialize_action(cls, buf: bytes) -> tuple[NanoHeaderAction, bytes]:
         from hathorlib.base_transaction import bytes_to_output_value
+        from hathorlib.nanocontracts.types import NCActionType
+
         type_bytes, buf = buf[:1], buf[1:]
         action_type = NCActionType.from_bytes(type_bytes)
         (token_index,), buf = unpack('!B', buf)
@@ -78,6 +79,8 @@ def _deserialize_action(cls, buf: bytes) -> tuple[NanoHeaderAction, bytes]:
 
     @classmethod
     def deserialize(cls, tx: BaseTransaction, buf: bytes) -> tuple[NanoHeader, bytes]:
+        from hathorlib.nanocontracts import DeprecatedNanoContract
+
         header_id, buf = buf[:1], buf[1:]
         assert header_id == VertexHeaderId.NANO_HEADER.value
 
@@ -124,6 +127,8 @@ def _serialize_action(action: NanoHeaderAction) -> bytes:
 
     def _serialize_without_header_id(self, *, skip_signature: bool) -> deque[bytes]:
         """Serialize the header with the option to skip the signature."""
+        from hathorlib.nanocontracts import DeprecatedNanoContract
+
         encoded_method = self.nc_method.encode('ascii')
 
         ret: deque[bytes] = deque()

hathorlib/headers/types.py

@@ -18,3 +18,4 @@
 @unique
 class VertexHeaderId(Enum):
     NANO_HEADER = b'\x10'
+    FEE_HEADER = b'\x11'

hathorlib/nanocontracts/nanocontract.py

@@ -49,8 +49,9 @@ def get_funds_fields_from_struct(self, buf: bytes) -> bytes:
         return buf
 
     def get_funds_struct(self) -> bytes:
+        from hathorlib.headers import DeprecatedNanoHeader
         struct_bytes = super().get_funds_struct()
-        nano_header_bytes = self.get_nano_header().serialize()
+        nano_header_bytes = self._get_header(DeprecatedNanoHeader).serialize()
         struct_bytes += nano_header_bytes[1:]
         return struct_bytes
 

hathorlib/transaction.py

@@ -10,15 +10,18 @@
 import struct
 from collections import namedtuple
 from struct import pack
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING, List, TypeVar
 
 from hathorlib.base_transaction import TX_HASH_SIZE, BaseTransaction, TxInput, TxOutput
 from hathorlib.conf import HathorSettings
 from hathorlib.exceptions import InvalidOutputValue, InvalidToken
+from hathorlib.headers import VertexBaseHeader
 from hathorlib.utils import unpack, unpack_len
 
 if TYPE_CHECKING:
-    from hathorlib.headers import DeprecatedNanoHeader, NanoHeader
+    from hathorlib.headers import FeeHeader, NanoHeader
+
+T = TypeVar('T', bound=VertexBaseHeader)
 
 settings = HathorSettings()
 
@@ -62,13 +65,31 @@ def is_nano_contract(self) -> bool:
         else:
             return True
 
-    def get_nano_header(self) -> NanoHeader | DeprecatedNanoHeader:
+    def has_fees(self) -> bool:
+        """Returns true if this transaction has a fee header"""
+        try:
+            self.get_fee_header()
+        except ValueError:
+            return False
+        else:
+            return True
+
+    def get_nano_header(self) -> NanoHeader:
+        from hathorlib.headers import NanoHeader
         """Return the NanoHeader or raise ValueError."""
-        from hathorlib.headers import DeprecatedNanoHeader, NanoHeader
+        return self._get_header(NanoHeader)
+
+    def get_fee_header(self) -> FeeHeader:
+        from hathorlib.headers import FeeHeader
+        """Return the FeeHeader or raise ValueError."""
+        return self._get_header(FeeHeader)
+
+    def _get_header(self, header_type: type[T]) -> T:
+        """Return the header of the given type or raise ValueError."""
         for header in self.headers:
-            if isinstance(header, (NanoHeader, DeprecatedNanoHeader)):
+            if isinstance(header, header_type):
                 return header
-        raise ValueError('nano header not found')
+        raise ValueError(f'{header_type.__name__.lower()} not found')
 
     @classmethod
     def create_from_struct(cls, struct_bytes: bytes) -> 'Transaction':

hathorlib/vertex_parser.py

@@ -26,9 +26,10 @@ class VertexParser:
     @staticmethod
     def get_supported_headers() -> dict[VertexHeaderId, type[VertexBaseHeader]]:
         """Return a dict of supported headers."""
-        from hathorlib.headers import NanoHeader, VertexHeaderId
+        from hathorlib.headers import FeeHeader, NanoHeader, VertexHeaderId
         return {
             VertexHeaderId.NANO_HEADER: NanoHeader,
+            VertexHeaderId.FEE_HEADER: FeeHeader,
         }
 
     @staticmethod

tests/test_basic.py

@@ -107,22 +107,19 @@ def test_token_creation_basics(self):
         tx.update_hash()
         self.assertFalse(tx.verify_pow())
 
-    def test_token_creation_with_fee_version(self):
-        """Test TokenCreationTransaction with token_version=2 (FEE)"""
-        from hathorlib.token_creation_tx import TokenVersion
+    def test_token_creation_with_fee_header(self):
+        """Test TokenCreationTransaction with fee header"""
+        from hathorlib.token_creation_tx import TokenCreationTransaction, TokenVersion
+
+        data = bytes.fromhex(
+            '0002010400000672c17c8fcf7277eece0b8cbe3f0efbdf6205e5e8554ccff5ca85ec8e49000069463044022070c5bfcd3b2f177'
+            'c842de1937c8a089bec64ea2d27754056fb7d7882e731aad7022073b6811313a52f74a88cedbbb2d951ddd5c6d2bba97332eea74'
+            '2e020d7717f04210299138e77a8039c31a112941480231cccefc9e627fef5ff4a391e7a2689b319d40000000900001976a914ba6'
+            'a16b0ab2c2bf132e1cfbdc01ef86a8c749a7188ac0000006401001976a914ba6a16b0ab2c2bf132e1cfbdc01ef86a8c749a7188a'
+            'c0000000181001976a914ba6a16b0ab2c2bf132e1cfbdc01ef86a8c749a7188ac0000000281001976a914ba6a16b0ab2c2bf132e'
+            '1cfbdc01ef86a8c749a7188ac0209546f6b656e4e616d6503544b4e4031b96d6968b53e690472ad000000000011010000000001'
+        )
 
-        # Using the same structure as test_token_creation_basics but with token_version=2
-        data = bytes.fromhex('00020104000005551d7740fd7d3c0acc50b5677fdd844f1225985aa431e1712af2a2fd'
-                             '8900006a473045022100a445edb5cd6c79a0a7b5ed837582fd65b8d511ee60b64fd076'
-                             'e07bd8f63f75a202202dca24320bffc4c3ca2a07cdfff38f7c839bde70ed49ef634ac6'
-                             '588972836cab2103bfa995d676e3c0ed7b863c74cfef9683fab3163b42b6f21442326a'
-                             '023fc57fba0000264800001976a9146876f9578221fdb678d4e8376503098a9228b132'
-                             '88ac00004e2001001976a914031761ef85a24603203c97e75af355b83209f08f88ac00'
-                             '00000181001976a9149f091256cb98649c7c35df0aad44d7805710691e88ac00000002'
-                             '81001976a914b1d7a5ee505ad4d3b93ea1a5162ba83d5049ec4e88ac0209546f546865'
-                             '4d6f6f6e04f09f9a804034a52aec6cece75e0fc0e30200001a72272f48339fcc5d5ec5'
-                             'deaf197855964b0eb912e8c6eefe00928b6cf600001055641c20b71871ed2c5c7d4096'
-                             'a34f40888d79c25bce74421646e732dc01ff730d')
         tx = TokenCreationTransaction.create_from_struct(data)
 
         # Verify the token version is FEE (2)
@@ -134,12 +131,23 @@ def test_token_creation_with_fee_version(self):
         # Verify basic transaction properties
         self.assertTrue(tx.is_transaction)
         self.assertFalse(tx.is_block)
+        self.assertTrue(tx.has_fees())
+
+        # Verify the fee header contains the expected fee entry
+        fee_header = tx.get_fee_header()
+        self.assertEqual(len(fee_header.fees), 1)
+        self.assertEqual(fee_header.fees[0].token_index, 0)
+        self.assertEqual(fee_header.fees[0].amount, 1)
+
+        self.assertEqual(len(fee_header.get_fees()), 1)
+        self.assertEqual(fee_header.get_fees()[0].amount, 1)
+        self.assertEqual(fee_header.get_fees()[0].token_uid, settings.HATHOR_TOKEN_UID)
 
         # Verify the string representation includes token_version=2
         str_repr = str(tx)
         self.assertIn('token_version=2', str_repr)
-        self.assertIn('token_name=ToTheMoon', str_repr)
-        self.assertIn('token_symbol=🚀', str_repr)
+        self.assertIn('token_name=TokenName', str_repr)
+        self.assertIn('token_symbol=TKN', str_repr)
 
     def test_script_basics(self):
         create_output_script(decode_address('HVZjvL1FJ23kH3buGNuttVRsRKq66WHUVZ'))