feat(dynamic-transaction-model): fee header

hathor/transaction/exceptions.py

@@ -258,3 +258,15 @@ class VerifyFailed(ScriptError):
 
 class TimeLocked(ScriptError):
     """Transaction is invalid because it is time locked"""
+
+
+class InvalidFeeHeader(TxValidationError):
+    """Invalid fee header"""
+
+
+class FeeHeaderTokenNotFound(InvalidFeeHeader):
+    """Token not found in the transaction tokens list"""
+
+
+class FeeHeaderInvalidAmount(InvalidFeeHeader):
+    """Invalid fee amount"""

hathor/transaction/headers/__init__.py

@@ -13,11 +13,13 @@
 # limitations under the License.
 
 from hathor.transaction.headers.base import VertexBaseHeader
+from hathor.transaction.headers.fee_header import FeeHeader
 from hathor.transaction.headers.nano_header import NanoHeader
 from hathor.transaction.headers.types import VertexHeaderId
 
 __all__ = [
     'VertexBaseHeader',
     'VertexHeaderId',
     'NanoHeader',
+    'FeeHeader',
 ]

hathor/transaction/headers/fee_header.py

@@ -0,0 +1,142 @@
+# 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 hathor.serialization import Deserializer, Serializer
+from hathor.serialization.encoding.output_value import decode_output_value
+from hathor.transaction.exceptions import FeeHeaderInvalidAmount
+from hathor.transaction.headers.base import VertexBaseHeader
+from hathor.transaction.headers.types import VertexHeaderId
+from hathor.transaction.util import (
+    VerboseCallback,
+    get_deposit_token_withdraw_amount,
+    int_to_bytes,
+    output_value_to_bytes,
+)
+from hathor.types import TokenUid
+
+if TYPE_CHECKING:
+    from hathor.conf.settings import HathorSettings
+    from hathor.transaction.base_transaction import BaseTransaction
+    from hathor.transaction.transaction import Transaction
+
+
+@dataclass(slots=True, kw_only=True, frozen=True)
+class FeeHeaderEntry:
+    token_index: int
+    amount: int
+
+
+@dataclass(slots=True, kw_only=True, frozen=True)
+class FeeEntry:
+    token_uid: TokenUid
+    amount: int
+
+    def __post_init__(self) -> None:
+        """Validate the amount."""
+        from hathor.conf.settings import HATHOR_TOKEN_UID
+
+        if self.amount <= 0:
+            raise FeeHeaderInvalidAmount(f'fees should be a positive integer, got {self.amount}')
+
+        if self.token_uid != HATHOR_TOKEN_UID and self.amount % 100 != 0:
+            raise FeeHeaderInvalidAmount(f'fees using deposit custom tokens should be a multiple of 100,'
+                                         f' got {self.amount}')
+
+
+@dataclass(slots=True, kw_only=True)
+class FeeHeader(VertexBaseHeader):
+    # transaction that contains the fee header
+    tx: 'Transaction'
+    # list of tokens and amounts that will be used to pay fees in the transaction
+    fees: list[FeeHeaderEntry]
+    _settings: HathorSettings
+
+    def __init__(self, settings: HathorSettings, tx: 'Transaction', fees: list[FeeHeaderEntry]):
+        self.tx = tx
+        self.fees = fees
+        self._settings = settings
+
+    @classmethod
+    def deserialize(
+        cls,
+        tx: BaseTransaction,
+        buf: bytes,
+        *,
+        verbose: VerboseCallback = None
+    ) -> tuple[FeeHeader, bytes]:
+        deserializer = Deserializer.build_bytes_deserializer(buf)
+
+        header_id = deserializer.read_bytes(1)
+        if verbose:
+            verbose('header_id', header_id)
+        assert header_id == VertexHeaderId.FEE_HEADER.value
+
+        fees: list[FeeHeaderEntry] = []
+        fees_len = deserializer.read_byte()
+        if verbose:
+            verbose('fees_len', fees_len)
+        for _ in range(fees_len):
+            token_index = deserializer.read_byte()
+            amount = decode_output_value(deserializer)
+            fees.append(FeeHeaderEntry(
+                token_index=token_index,
+                amount=amount,
+            ))
+
+        from hathor.transaction import Transaction
+        assert isinstance(tx, Transaction)
+        remaining_bytes = bytes(deserializer.read_all())
+        return cls(
+            settings=tx._settings,
+            tx=tx,
+            fees=fees,
+        ), remaining_bytes
+
+    def serialize(self) -> bytes:
+        serializer = Serializer.build_bytes_serializer()
+        serializer.write_bytes(VertexHeaderId.FEE_HEADER.value)
+        serializer.write_bytes(int_to_bytes(len(self.fees), 1))
+
+        for fee in self.fees:
+            serializer.write_bytes(int_to_bytes(fee.token_index, 1))
+            serializer.write_bytes(output_value_to_bytes(fee.amount))
+
+        return bytes(serializer.finalize())
+
+    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
+        ]
+
+    def total_fee_amount(self) -> int:
+        """Sum fees amounts in this header and return as HTR"""
+        total_fee = 0
+        for fee in self.get_fees():
+            if fee.token_uid == self._settings.HATHOR_TOKEN_UID:
+                total_fee += fee.amount
+            else:
+                total_fee += get_deposit_token_withdraw_amount(self._settings, fee.amount)
+        return total_fee

hathor/transaction/headers/types.py

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

hathor/transaction/transaction.py

@@ -16,7 +16,7 @@
 
 import hashlib
 from struct import pack
-from typing import TYPE_CHECKING, Any, NamedTuple, Optional
+from typing import TYPE_CHECKING, Any, NamedTuple, Optional, TypeVar
 
 from typing_extensions import Self, override
 
@@ -26,12 +26,15 @@
 from hathor.transaction import TxInput, TxOutput, TxVersion
 from hathor.transaction.base_transaction import TX_HASH_SIZE, GenericVertex
 from hathor.transaction.exceptions import InvalidToken
-from hathor.transaction.headers import NanoHeader
+from hathor.transaction.headers import NanoHeader, VertexBaseHeader
+from hathor.transaction.headers.fee_header import FeeHeader
 from hathor.transaction.static_metadata import TransactionStaticMetadata
 from hathor.transaction.token_info import TokenInfo, TokenInfoDict, TokenVersion
 from hathor.transaction.util import VerboseCallback, unpack, unpack_len
 from hathor.types import TokenUid, VertexId
 
+T = TypeVar('T', bound=VertexBaseHeader)
+
 if TYPE_CHECKING:
     from hathor.conf.settings import HathorSettings
     from hathor.transaction.storage import TransactionStorage  # noqa: F401
@@ -112,12 +115,29 @@ def is_nano_contract(self) -> bool:
         else:
             return True
 
+    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:
         """Return the NanoHeader or raise ValueError."""
+        return self._get_header(NanoHeader)
+
+    def get_fee_header(self) -> 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):
+            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, storage: Optional['TransactionStorage'] = None,

hathor/verification/fee_header_verifier.py

@@ -0,0 +1,58 @@
+#  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 typing import Sequence
+
+from hathor.transaction.exceptions import FeeHeaderTokenNotFound, InvalidFeeHeader
+from hathor.transaction.headers import FeeHeader
+
+MAX_FEES_LEN: int = 16
+
+
+class FeeHeaderVerifier:
+
+    @staticmethod
+    def verify_fee_list(fee_header: 'FeeHeader', tx_tokens_len: int) -> None:
+        """Perform FeeHeader verifications that do not depend on the tx storage."""
+        fees = fee_header.fees
+        FeeHeaderVerifier._verify_fee_list_size('fees', len(fees))
+
+        # Check for duplicate token indices in fees
+        token_indices = [fee.token_index for fee in fees]
+        FeeHeaderVerifier._verify_duplicate_indexes('fees', token_indices)
+
+        for fee in fees:
+            FeeHeaderVerifier._verify_token_index('fees', fee.token_index, tx_tokens_len)
+
+    @staticmethod
+    def _verify_token_index(prop: str, token_index: int, tx_token_len: int) -> None:
+        if token_index > tx_token_len:
+            raise FeeHeaderTokenNotFound(
+                f'{prop} contains token index {token_index} which is not in tokens list'
+            )
+
+    @staticmethod
+    def _verify_duplicate_indexes(list_name: str, indexes: Sequence[int]) -> None:
+        if len(indexes) != len(set(indexes)):
+            raise InvalidFeeHeader(f'duplicate token indexes in {list_name} list')
+
+    @staticmethod
+    def _verify_fee_list_size(prop: str, list_len: int) -> None:
+        if list_len == 0:
+            raise InvalidFeeHeader(f'{prop} cannot be empty')
+
+        if list_len > MAX_FEES_LEN:
+            raise InvalidFeeHeader(f'more {prop} than the max allowed: {list_len} > {MAX_FEES_LEN}')