refactor(verification): modularize verify_scripts [part 6/7]

hathor/transaction/exceptions.py

@@ -74,10 +74,18 @@ class ConflictingInputs(TxValidationError):
     """Inputs in the tx are spending the same output"""
 
 
+class OutputNotSelected(TxValidationError):
+    """At least one output is not selected for signing by some input."""
+
+
 class TooManyOutputs(TxValidationError):
     """More than 256 outputs"""
 
 
+class TooManySighashSubsets(TxValidationError):
+    """More sighash subsets than the configured maximum."""
+
+
 class InvalidOutputValue(TxValidationError):
     """Value of output is invalid"""
 
@@ -202,3 +210,25 @@ class VerifyFailed(ScriptError):
 
 class TimeLocked(ScriptError):
     """Transaction is invalid because it is time locked"""
+
+
+class InputNotSelectedError(ScriptError):
+    """Raised when an input does not select itself for signing in its script."""
+
+
+class MaxInputsExceededError(ScriptError):
+    """The transaction has more inputs than the maximum configured in the script."""
+
+
+class MaxOutputsExceededError(ScriptError):
+    """The transaction has more outputs than the maximum configured in the script."""
+
+
+class InputsOutputsLimitModelInvalid(ScriptError):
+    """
+    Raised when the inputs outputs limit model could not be constructed from the arguments provided in the script.
+    """
+
+
+class CustomSighashModelInvalid(ScriptError):
+    """Raised when the sighash model could not be constructed from the arguments provided in the script."""

hathor/transaction/scripts/execute.py

@@ -13,17 +13,33 @@
 #  limitations under the License.
 
 import struct
-from typing import NamedTuple, Optional, Union
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, NamedTuple, Optional, Union
 
+from hathor.conf.get_settings import get_global_settings
 from hathor.transaction import BaseTransaction, Transaction, TxInput
 from hathor.transaction.exceptions import DataIndexError, FinalStackInvalid, InvalidScriptError, OutOfData
 
+if TYPE_CHECKING:
+    from hathor.transaction.scripts.script_context import ScriptContext
+
 
-class ScriptExtras(NamedTuple):
+@dataclass(slots=True, frozen=True, kw_only=True)
+class ScriptExtras:
+    """
+    A simple container for auxiliary data that may be used during execution of scripts.
+    """
     tx: Transaction
-    txin: TxInput
+    input_index: int
     spent_tx: BaseTransaction
 
+    @property
+    def txin(self) -> TxInput:
+        return self.tx.inputs[self.input_index]
+
+    def __post_init__(self) -> None:
+        assert self.txin.tx_id == self.spent_tx.hash
+
 
 # XXX: Because the Stack is a heterogeneous list of bytes and int, and some OPs only work for when the stack has some
 #      or the other type, there are many places that require an assert to prevent the wrong type from being used,
@@ -39,7 +55,7 @@ class OpcodePosition(NamedTuple):
     position: int
 
 
-def execute_eval(data: bytes, log: list[str], extras: ScriptExtras) -> None:
+def execute_eval(data: bytes, log: list[str], extras: ScriptExtras) -> 'ScriptContext':
     """ Execute eval from data executing opcode methods
 
         :param data: data to be evaluated that contains data and opcodes
@@ -56,8 +72,9 @@ def execute_eval(data: bytes, log: list[str], extras: ScriptExtras) -> None:
     """
     from hathor.transaction.scripts.opcode import Opcode, execute_op_code
     from hathor.transaction.scripts.script_context import ScriptContext
+    settings = get_global_settings()
     stack: Stack = []
-    context = ScriptContext(stack=stack, logs=log, extras=extras)
+    context = ScriptContext(settings=settings, stack=stack, logs=log, extras=extras)
     data_len = len(data)
     pos = 0
     while pos < data_len:
@@ -70,6 +87,8 @@ def execute_eval(data: bytes, log: list[str], extras: ScriptExtras) -> None:
 
     evaluate_final_stack(stack, log)
 
+    return context
+
 
 def evaluate_final_stack(stack: Stack, log: list[str]) -> None:
     """ Checks the final state of the stack.
@@ -88,7 +107,7 @@ def evaluate_final_stack(stack: Stack, log: list[str]) -> None:
         raise FinalStackInvalid('\n'.join(log))
 
 
-def script_eval(tx: Transaction, txin: TxInput, spent_tx: BaseTransaction) -> None:
+def script_eval(tx: Transaction, spent_tx: BaseTransaction, *, input_index: int) -> 'ScriptContext':
     """Evaluates the output script and input data according to
     a very limited subset of Bitcoin's scripting language.
 
@@ -103,10 +122,10 @@ def script_eval(tx: Transaction, txin: TxInput, spent_tx: BaseTransaction) -> No
 
     :raises ScriptError: if script verification fails
     """
-    input_data = txin.data
-    output_script = spent_tx.outputs[txin.index].script
+    extras = ScriptExtras(tx=tx, input_index=input_index, spent_tx=spent_tx)
+    input_data = extras.txin.data
+    output_script = spent_tx.outputs[extras.txin.index].script
     log: list[str] = []
-    extras = ScriptExtras(tx=tx, txin=txin, spent_tx=spent_tx)
 
     from hathor.transaction.scripts import MultiSig
     if MultiSig.re_match.search(output_script):
@@ -115,17 +134,17 @@ def script_eval(tx: Transaction, txin: TxInput, spent_tx: BaseTransaction) -> No
         # we can't use input_data + output_script because it will end with an invalid stack
         # i.e. the signatures will still be on the stack after ouput_script is executed
         redeem_script_pos = MultiSig.get_multisig_redeem_script_pos(input_data)
-        full_data = txin.data[redeem_script_pos:] + output_script
+        full_data = extras.txin.data[redeem_script_pos:] + output_script
         execute_eval(full_data, log, extras)
 
         # Second, we need to validate that the signatures on the input_data solves the redeem_script
         # we pop and append the redeem_script to the input_data and execute it
         multisig_data = MultiSig.get_multisig_data(extras.txin.data)
-        execute_eval(multisig_data, log, extras)
+        return execute_eval(multisig_data, log, extras)
     else:
         # merge input_data and output_script
         full_data = input_data + output_script
-        execute_eval(full_data, log, extras)
+        return execute_eval(full_data, log, extras)
 
 
 def decode_opn(opcode: int) -> int:

hathor/transaction/scripts/hathor_script.py

@@ -15,7 +15,16 @@
 import struct
 from typing import Union
 
+from typing_extensions import assert_never
+
 from hathor.transaction.scripts.opcode import Opcode
+from hathor.transaction.scripts.sighash import (
+    InputsOutputsLimit,
+    SighashAll,
+    SighashBitmask,
+    SighashRange,
+    SighashType,
+)
 
 
 class HathorScript:
@@ -49,3 +58,38 @@ def pushData(self, data: Union[int, bytes]) -> None:
             self.data += (bytes([len(data)]) + data)
         else:
             self.data += (bytes([Opcode.OP_PUSHDATA1]) + bytes([len(data)]) + data)
+
+    def push_sighash(self, sighash: SighashType) -> None:
+        """Push a custom sighash to the script."""
+        match sighash:
+            case SighashAll():
+                pass
+            case SighashBitmask():
+                self.pushData(sighash.inputs)
+                self.pushData(sighash.outputs)
+                self.addOpcode(Opcode.OP_SIGHASH_BITMASK)
+            case SighashRange():
+                self.pushData(sighash.input_start)
+                self.pushData(sighash.input_end)
+                self.pushData(sighash.output_start)
+                self.pushData(sighash.output_end)
+                self.addOpcode(Opcode.OP_SIGHASH_RANGE)
+            case _:
+                assert_never(sighash)
+
+    def push_max_sighash_subsets(self, max_subsets: int | None) -> None:
+        """Push a maximum limit for custom sighash subsets."""
+        if max_subsets is None:
+            return
+
+        self.pushData(max_subsets)
+        self.addOpcode(Opcode.OP_MAX_SIGHASH_SUBSETS)
+
+    def push_inputs_outputs_limit(self, limit: InputsOutputsLimit | None) -> None:
+        """Push a custom inputs and outputs limit to the script."""
+        if not limit:
+            return
+
+        self.pushData(limit.max_inputs)
+        self.pushData(limit.max_outputs)
+        self.addOpcode(Opcode.OP_MAX_INPUTS_OUTPUTS)

hathor/transaction/scripts/multi_sig.py

@@ -21,6 +21,7 @@
 from hathor.transaction.scripts.execute import Stack, get_script_op
 from hathor.transaction.scripts.hathor_script import HathorScript
 from hathor.transaction.scripts.opcode import Opcode, op_pushdata, op_pushdata1
+from hathor.transaction.scripts.sighash import InputsOutputsLimit, SighashAll, SighashType
 
 
 class MultiSig(BaseScript):
@@ -111,7 +112,15 @@ def create_output_script(cls, address: bytes, timelock: Optional[Any] = None) ->
         return s.data
 
     @classmethod
-    def create_input_data(cls, redeem_script: bytes, signatures: list[bytes]) -> bytes:
+    def create_input_data(
+        cls,
+        redeem_script: bytes,
+        signatures: list[bytes],
+        *,
+        sighash: SighashType = SighashAll(),
+        max_sighash_subsets: int | None = None,
+        inputs_outputs_limit: InputsOutputsLimit | None = None
+    ) -> bytes:
         """
         :param redeem_script: script to redeem the tokens: <M> <pubkey1> ... <pubkeyN> <N> <OP_CHECKMULTISIG>
         :type redeem_script: bytes
@@ -122,6 +131,9 @@ def create_input_data(cls, redeem_script: bytes, signatures: list[bytes]) -> byt
         :rtype: bytes
         """
         s = HathorScript()
+        s.push_sighash(sighash)
+        s.push_max_sighash_subsets(max_sighash_subsets)
+        s.push_inputs_outputs_limit(inputs_outputs_limit)
         for signature in signatures:
             s.pushData(signature)
         s.pushData(redeem_script)