feat(sighash): implement sighash bitmask

Author: glevco

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"""
 
@@ -194,3 +198,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,16 +13,21 @@
 #  limitations under the License.
 
 import struct
-from typing import NamedTuple, Optional, Union
+from typing import TYPE_CHECKING, NamedTuple, Optional, Union
 
+from hathor.conf.get_settings import get_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):
     tx: Transaction
     txin: TxInput
     spent_tx: BaseTransaction
+    input_index: int
 
 
 # XXX: Because the Stack is a heterogeneous list of bytes and int, and some OPs only work for when the stack has some
@@ -39,7 +44,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 +61,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_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 +76,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 +96,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, txin: TxInput, 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.
 
@@ -106,7 +114,7 @@ def script_eval(tx: Transaction, txin: TxInput, spent_tx: BaseTransaction) -> No
     input_data = txin.data
     output_script = spent_tx.outputs[txin.index].script
     log: list[str] = []
-    extras = ScriptExtras(tx=tx, txin=txin, spent_tx=spent_tx)
+    extras = ScriptExtras(tx=tx, txin=txin, spent_tx=spent_tx, input_index=input_index)
 
     from hathor.transaction.scripts import MultiSig
     if MultiSig.re_match.search(output_script):
@@ -121,11 +129,11 @@ def script_eval(tx: Transaction, txin: TxInput, spent_tx: BaseTransaction) -> No
         # 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

@@ -28,9 +28,14 @@
     is_pubkey_compressed,
 )
 from hathor.transaction.exceptions import (
+    CustomSighashModelInvalid,
     EqualVerifyFailed,
+    InputNotSelectedError,
+    InputsOutputsLimitModelInvalid,
     InvalidScriptError,
     InvalidStackData,
+    MaxInputsExceededError,
+    MaxOutputsExceededError,
     MissingStackItems,
     OracleChecksigFailed,
     ScriptError,
@@ -39,6 +44,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 +79,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 +259,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 +594,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 +628,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 Exception 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 Exception 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 +689,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 +705,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

hathor/transaction/scripts/script_context.py

@@ -12,14 +12,58 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+import hashlib
+
+from typing_extensions import assert_never
+
+from hathor.conf.settings import HathorSettings
+from hathor.transaction import Transaction
+from hathor.transaction.exceptions import ScriptError
 from hathor.transaction.scripts.execute import ScriptExtras, Stack
+from hathor.transaction.scripts.sighash import SighashAll, SighashBitmask, SighashType
 
 
 class ScriptContext:
     """A context to be manipulated during script execution. A separate instance must be used for each script."""
-    __slots__ = ('stack', 'logs', 'extras')
+    __slots__ = ('stack', 'logs', 'extras', '_settings', '_sighash')
 
-    def __init__(self, *, stack: Stack, logs: list[str], extras: ScriptExtras) -> None:
+    def __init__(self, *, stack: Stack, logs: list[str], extras: ScriptExtras, settings: HathorSettings) -> None:
         self.stack = stack
         self.logs = logs
         self.extras = extras
+        self._settings = settings
+        self._sighash: SighashType = SighashAll()
+
+    def set_sighash(self, sighash: SighashType) -> None:
+        """
+        Set a Sighash type in this context.
+        It can only be set once, that is, a script cannot use more than one sighash type.
+        """
+        if type(self._sighash) is not SighashAll:
+            raise ScriptError('Cannot modify sighash after it is already set.')
+
+        self._sighash = sighash
+
+    def get_tx_sighash_data(self, tx: Transaction) -> bytes:
+        """
+        Return the sighash data for a tx, depending on the sighash type set in this context.
+        Must be used when verifying signatures during script execution.
+        """
+        match self._sighash:
+            case SighashAll():
+                return tx.get_sighash_all_data()
+            case SighashBitmask():
+                data = tx.get_custom_sighash_data(self._sighash)
+                return hashlib.sha256(data).digest()
+            case _:
+                assert_never(self._sighash)
+
+    def get_selected_outputs(self) -> set[int]:
+        """Get a set with all output indexes selected (that is, signed) in this context."""
+        match self._sighash:
+            case SighashAll():
+                return set(range(self._settings.MAX_NUM_OUTPUTS))
+            case SighashBitmask():
+                return set(self._sighash.get_output_indexes())
+            case _:
+                assert_never(self._sighash)