feat(sighash): implement sighash bitmask [part 1/7]

hathor/transaction/exceptions.py

@@ -74,6 +74,10 @@ 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"""
 
@@ -202,3 +206,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,25 +107,20 @@ 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:
-    """Evaluates the output script and input data according to
-    a very limited subset of Bitcoin's scripting language.
-
-    :param tx: the transaction being validated, the 'owner' of the input data
-    :type tx: :py:class:`hathor.transaction.Transaction`
-
-    :param txin: transaction input being evaluated
-    :type txin: :py:class:`hathor.transaction.TxInput`
-
-    :param spent_tx: the transaction referenced by the input
-    :type spent_tx: :py:class:`hathor.transaction.BaseTransaction`
+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.
+    Raises ScriptError if script verification fails
 
-    :raises ScriptError: if script verification fails
+    Args:
+        tx: the transaction being validated, the 'owner' of the input data
+        spent_tx: the transaction referenced by the input
+        input_index: index of the transaction input being evaluated
     """
-    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 +129,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/opcode.py

@@ -16,6 +16,7 @@
 import struct
 from enum import IntEnum
 
+import pydantic
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec
@@ -28,9 +29,14 @@
     is_pubkey_compressed,
 )
 from hathor.transaction.exceptions import (
+    CustomSighashModelInvalid,
     EqualVerifyFailed,
+    InputNotSelectedError,
+    InputsOutputsLimitModelInvalid,
     InvalidScriptError,
     InvalidStackData,
+    MaxInputsExceededError,
+    MaxOutputsExceededError,
     MissingStackItems,
     OracleChecksigFailed,
     ScriptError,
@@ -39,6 +45,8 @@
 )
 from hathor.transaction.scripts.execute import Stack, binary_to_int, decode_opn, get_data_value, get_script_op
 from hathor.transaction.scripts.script_context import ScriptContext
+from hathor.transaction.scripts.sighash import InputsOutputsLimit, SighashBitmask
+from hathor.transaction.util import bytes_to_int
 
 
 class Opcode(IntEnum):
@@ -72,6 +80,9 @@ class Opcode(IntEnum):
     OP_DATA_GREATERTHAN = 0xC1
     OP_FIND_P2PKH = 0xD0
     OP_DATA_MATCH_VALUE = 0xD1
+    OP_SIGHASH_BITMASK = 0xE0
+    OP_SIGHASH_RANGE = 0xE1
+    OP_MAX_INPUTS_OUTPUTS = 0xE2
 
     @classmethod
     def is_pushdata(cls, opcode: int) -> bool:
@@ -249,7 +260,8 @@ def op_checksig(context: ScriptContext) -> None:
         # pubkey is not compressed public key
         raise ScriptError('OP_CHECKSIG: pubkey is not a public key') from e
     try:
-        public_key.verify(signature, context.extras.tx.get_sighash_all_data(), ec.ECDSA(hashes.SHA256()))
+        sighash_data = context.get_tx_sighash_data(context.extras.tx)
+        public_key.verify(signature, sighash_data, ec.ECDSA(hashes.SHA256()))
         # valid, push true to stack
         context.stack.append(1)
     except InvalidSignature:
@@ -583,7 +595,7 @@ def op_checkmultisig(context: ScriptContext) -> None:
         while pubkey_index < len(pubkeys):
             pubkey = pubkeys[pubkey_index]
             new_stack = [signature, pubkey]
-            op_checksig(ScriptContext(stack=new_stack, logs=context.logs, extras=context.extras))
+            op_checksig(ScriptContext(stack=new_stack, logs=context.logs, extras=context.extras, settings=settings))
             result = new_stack.pop()
             pubkey_index += 1
             if result == 1:
@@ -617,6 +629,59 @@ def op_integer(opcode: int, stack: Stack) -> None:
         raise ScriptError(e) from e
 
 
+def op_sighash_bitmask(context: ScriptContext) -> None:
+    """Pop two items from the stack, constructing a sighash bitmask and setting it in the script context."""
+    if len(context.stack) < 2:
+        raise MissingStackItems(f'OP_SIGHASH_BITMASK: expected 2 elements on stack, has {len(context.stack)}')
+
+    outputs = context.stack.pop()
+    inputs = context.stack.pop()
+    assert isinstance(inputs, bytes)
+    assert isinstance(outputs, bytes)
+
+    try:
+        sighash = SighashBitmask(
+            inputs=bytes_to_int(inputs),
+            outputs=bytes_to_int(outputs)
+        )
+    except pydantic.ValidationError as e:
+        raise CustomSighashModelInvalid('Could not construct sighash bitmask.') from e
+
+    if context.extras.input_index not in sighash.get_input_indexes():
+        raise InputNotSelectedError(
+            f'Input at index {context.extras.input_index} must select itself when using a custom sighash.'
+        )
+
+    context.set_sighash(sighash)
+
+
+def op_max_inputs_outputs(context: ScriptContext) -> None:
+    """Pop two items from the stack, constructing an inputs and outputs limit and setting it in the script context."""
+    if len(context.stack) < 2:
+        raise MissingStackItems(f'OP_MAX_INPUTS_OUTPUTS: expected 2 elements on stack, has {len(context.stack)}')
+
+    max_outputs = context.stack.pop()
+    max_inputs = context.stack.pop()
+    assert isinstance(max_inputs, bytes)
+    assert isinstance(max_outputs, bytes)
+
+    try:
+        limit = InputsOutputsLimit(
+            max_inputs=bytes_to_int(max_inputs),
+            max_outputs=bytes_to_int(max_outputs)
+        )
+    except pydantic.ValidationError as e:
+        raise InputsOutputsLimitModelInvalid("Could not construct inputs and outputs limits.") from e
+
+    tx_inputs_len = len(context.extras.tx.inputs)
+    if tx_inputs_len > limit.max_inputs:
+        raise MaxInputsExceededError(f'Maximum number of inputs exceeded ({tx_inputs_len} > {limit.max_inputs}).')
+
+    tx_outputs_len = len(context.extras.tx.outputs)
+    if tx_outputs_len > limit.max_outputs:
+        raise MaxOutputsExceededError(f'Maximum number of outputs exceeded ({tx_outputs_len} > {limit.max_outputs}).')
+
+
 def execute_op_code(opcode: Opcode, context: ScriptContext) -> None:
     """
     Execute a function opcode.
@@ -625,6 +690,8 @@ def execute_op_code(opcode: Opcode, context: ScriptContext) -> None:
         opcode: the opcode to be executed.
         context: the script context to be manipulated.
     """
+    if not is_opcode_valid(opcode):
+        raise ScriptError(f'Opcode "{opcode.name}" is invalid.')
     context.logs.append(f'Executing function opcode {opcode.name} ({hex(opcode.value)})')
     match opcode:
         case Opcode.OP_DUP: op_dup(context)
@@ -639,4 +706,26 @@ def execute_op_code(opcode: Opcode, context: ScriptContext) -> None:
         case Opcode.OP_DATA_MATCH_VALUE: op_data_match_value(context)
         case Opcode.OP_CHECKDATASIG: op_checkdatasig(context)
         case Opcode.OP_FIND_P2PKH: op_find_p2pkh(context)
+        case Opcode.OP_SIGHASH_BITMASK: op_sighash_bitmask(context)
+        case Opcode.OP_MAX_INPUTS_OUTPUTS: op_max_inputs_outputs(context)
         case _: raise ScriptError(f'unknown opcode: {opcode}')
+
+
+def is_opcode_valid(opcode: Opcode) -> bool:
+    """Return whether an opcode is valid, that is, it's currently enabled."""
+    valid_opcodes = [
+        Opcode.OP_DUP,
+        Opcode.OP_EQUAL,
+        Opcode.OP_EQUALVERIFY,
+        Opcode.OP_CHECKSIG,
+        Opcode.OP_HASH160,
+        Opcode.OP_GREATERTHAN_TIMESTAMP,
+        Opcode.OP_CHECKMULTISIG,
+        Opcode.OP_DATA_STREQUAL,
+        Opcode.OP_DATA_GREATERTHAN,
+        Opcode.OP_DATA_MATCH_VALUE,
+        Opcode.OP_CHECKDATASIG,
+        Opcode.OP_FIND_P2PKH,
+    ]
+
+    return opcode in valid_opcodes

hathor/transaction/scripts/p2pkh.py

@@ -20,6 +20,7 @@
 from hathor.transaction.scripts.construct import get_pushdata, re_compile
 from hathor.transaction.scripts.hathor_script import HathorScript
 from hathor.transaction.scripts.opcode import Opcode
+from hathor.transaction.scripts.sighash import InputsOutputsLimit, SighashBitmask
 
 
 class P2PKH(BaseScript):
@@ -91,16 +92,35 @@ def create_output_script(cls, address: bytes, timelock: Optional[Any] = None) ->
         return s.data
 
     @classmethod
-    def create_input_data(cls, public_key_bytes: bytes, signature: bytes) -> bytes:
+    def create_input_data(
+        cls,
+        public_key_bytes: bytes,
+        signature: bytes,
+        *,
+        sighash: SighashBitmask | None = None,
+        inputs_outputs_limit: InputsOutputsLimit | None = None
+    ) -> bytes:
         """
         :param private_key: key corresponding to the address we want to spend tokens from
         :type private_key: :py:class:`cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKey`
 
         :rtype: bytes
         """
         s = HathorScript()
+
+        if sighash:
+            s.pushData(sighash.inputs)
+            s.pushData(sighash.outputs)
+            s.addOpcode(Opcode.OP_SIGHASH_BITMASK)
+
+        if inputs_outputs_limit:
+            s.pushData(inputs_outputs_limit.max_inputs)
+            s.pushData(inputs_outputs_limit.max_outputs)
+            s.addOpcode(Opcode.OP_MAX_INPUTS_OUTPUTS)
+
         s.pushData(signature)
         s.pushData(public_key_bytes)
+
         return s.data
 
     @classmethod