Script interpreter (#83)

* First commit of script engine

* Add support for some op_codes

* Complete test vector

* Old changes

* Refactor

* Remove proposal

* Fix some tox errors

* Add new op_codes

* Add tests from bitcoin core

* Start veryfing legacy scripts, add new functions and introduce flags

* Pass validation of some edge cases

* Pass tx_valid_legacy.json

* Pass tx_invalid_taproot.json except for segwit

* checksequenceverify and checklocktimeverify

* Pass all tests apart from segwit related ones

* Fix tox errors and rebase

* Refactor code, split script and tapscript

* Pass validation of all tests

* Refactor

* Start testing script_tests.json

* Fix some errors in script validation

* Pass most of tests in script_tests.json

* Simplify script execution

* Simplify op_if

* Pass validation of bitcoin core tests

* Fix tox errors

* Fix rebase

* Fix rebase

* Fix errors in testnet

* Fix op_nop and segwit stack limit

* Fix op count limit

* Refactor, fix warnings

* Fix after rebase

* Fix after rebase

* Fix after rebase

* Fix types

* 'Refactored by Sourcery' (#111)

Co-authored-by: Sourcery AI <>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: 3 people

HISTORY.md

@@ -157,8 +157,8 @@ Major changes includes:
   now it is xkey_dict.index < 0x80000000
 - bip32: local "./" derivation, opposed to absolute "m/" derivation,
   is not available anymore
-- bip32: indexes_from_bip32_path now returns List[int] instead of
-  Tuple[List[bytes], bool] losing the "absolute derivation" bool
+- bip32: indexes_from_bip32_path now returns list[int] instead of
+  Tuple[list[bytes], bool] losing the "absolute derivation" bool
 - bms: serialize/deserialize have been renamed encode/decode as they
   include the base64 (de)encoding, not jut the plain (de)serialization
 - Block: fixed bug in difficulty calculation

btclib/alias.py

@@ -14,7 +14,7 @@
 from __future__ import annotations
 
 from io import BytesIO
-from typing import Any, Callable, Tuple, Union
+from typing import Any, Callable, List, Tuple, Union
 
 # Octets are a sequence of eight-bit bytes or a hex-string (not text string)
 #
@@ -104,3 +104,6 @@
 # TaprootLeaf = Tuple[int, Script]
 # TaprootScriptTree = List[Union[Any, TaprootLeaf]]
 TaprootScriptTree = Any
+
+Command = Union[int, str, bytes]
+ScriptList = List[Command]

btclib/b58.py

@@ -20,7 +20,7 @@
 from btclib.exceptions import BTClibValueError
 from btclib.hashes import hash160, sha256
 from btclib.network import NETWORKS, network_from_key_value
-from btclib.script import serialize
+from btclib.script.script import serialize
 from btclib.to_prv_key import PrvKey, prv_keyinfo_from_prv_key
 from btclib.to_pub_key import Key, pub_keyinfo_from_key
 from btclib.utils import bytes_from_octets

btclib/ecc/borromean.py

@@ -8,6 +8,7 @@
 # No part of btclib including this file, may be copied, modified, propagated,
 # or distributed except according to the terms contained in the LICENSE file.
 """Borromean signature functions."""
+
 from __future__ import annotations
 
 import secrets

btclib/ecc/dsa.py

@@ -49,20 +49,23 @@ def _serialize_scalar(scalar: int) -> bytes:
     return _DER_SCALAR_MARKER + var_bytes.serialize(scalar_bytes)
 
 
-def _deserialize_scalar(sig_data_stream: BytesIO) -> int:
+def _deserialize_scalar(sig_data_stream: BytesIO, strict: bool = True) -> int:
     marker = sig_data_stream.read(1)
     if marker != _DER_SCALAR_MARKER:
         err_msg = f"invalid value header: {marker.hex()}"
         err_msg += f", instead of integer element {_DER_SCALAR_MARKER.hex()}"
         raise BTClibValueError(err_msg)
 
     r_bytes = var_bytes.parse(sig_data_stream, forbid_zero_size=True)
-    if r_bytes[0] == 0 and r_bytes[1] < 0x80:
+    if r_bytes[0] == 0 and r_bytes[1] < 0x80 and strict:
         raise BTClibValueError("invalid 'highest bit set' padding")
-    if r_bytes[0] >= 0x80:
+
+    scalar = int.from_bytes(r_bytes, byteorder="big", signed=False)
+
+    if r_bytes[0] >= 0x80 and strict:
         raise BTClibValueError("invalid negative scalar")
 
-    return int.from_bytes(r_bytes, byteorder="big", signed=False)
+    return abs(scalar)
 
 
 @dataclass(frozen=True)
@@ -161,7 +164,12 @@ def serialize(self, check_validity: bool = True) -> bytes:
         return _DER_SIG_MARKER + var_bytes.serialize(out)
 
     @classmethod
-    def parse(cls: type[Sig], data: BinaryData, check_validity: bool = True) -> Sig:
+    def parse(
+        cls: type[Sig],
+        data: BinaryData,
+        check_validity: bool = True,
+        strict: bool = True,
+    ) -> Sig:
         """Return a Sig by parsing binary data.
 
         Deserialize a strict ASN.1 DER representation of an ECDSA
@@ -182,8 +190,8 @@ def parse(cls: type[Sig], data: BinaryData, check_validity: bool = True) -> Sig:
 
         # [0x02][r-size][r][0x02][s-size][s]
         sig_data_substream = bytesio_from_binarydata(sig_data)
-        r = _deserialize_scalar(sig_data_substream)
-        s = _deserialize_scalar(sig_data_substream)
+        r = _deserialize_scalar(sig_data_substream, strict)
+        s = _deserialize_scalar(sig_data_substream, strict)
 
         # to prevent malleability
         # the sig_data_substream must have been consumed entirely

btclib/hashes.py

@@ -34,6 +34,12 @@ def ripemd160(octets: Octets) -> bytes:
     return hashlib.new("ripemd160", octets).digest()
 
 
+def sha1(octets: Octets) -> bytes:
+    """Return the SHA1(*) of the input octet sequence."""
+    octets = bytes_from_octets(octets)
+    return hashlib.sha1(octets).digest()  # nosec
+
+
 def sha256(octets: Octets) -> bytes:
     """Return the SHA256(*) of the input octet sequence."""
     octets = bytes_from_octets(octets)

btclib/psbt/psbt.py

@@ -17,9 +17,9 @@
 import random
 from copy import deepcopy
 from dataclasses import dataclass
-from typing import Any, Callable, Mapping, Sequence, TypeVar
+from typing import Any, Callable, Mapping, Sequence, TypeVar, cast
 
-from btclib.alias import Octets, String
+from btclib.alias import Octets, ScriptList, String
 from btclib.bip32 import (
     BIP32KeyOrigin,
     HdKeyPaths,
@@ -417,7 +417,9 @@ def finalize_psbt(psbt: Psbt) -> Psbt:
             psbt_in.final_script_sig = serialize([psbt_in.redeem_script])
             psbt_in.final_script_witness = Witness(cmds + [psbt_in.witness_script])
         else:
-            psbt_in.final_script_sig = serialize(cmds + [psbt_in.redeem_script])
+            psbt_in.final_script_sig = serialize(
+                cast(ScriptList, cmds + [psbt_in.redeem_script])
+            )
         psbt_in.partial_sigs = {}
         psbt_in.sig_hash_type = None
         psbt_in.redeem_script = b""

btclib/script/engine/__init__.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python3
+
+# Copyright (C) 2017-2021 The btclib developers
+#
+# This file is part of btclib. It is subject to the license terms in the
+# LICENSE file found in the top-level directory of this distribution.
+#
+# No part of btclib including this file, may be copied, modified, propagated,
+# or distributed except according to the terms contained in the LICENSE file.
+"""Bitcoin Script engine."""
+
+from __future__ import annotations
+
+from typing import cast
+
+from btclib.alias import Command, ScriptList
+from btclib.exceptions import BTClibValueError
+from btclib.hashes import sha256
+from btclib.script.engine import tapscript
+from btclib.script.engine.script import verify_script as verify_script_legacy
+from btclib.script.engine.script_op_codes import _to_num
+from btclib.script.script import parse, serialize
+from btclib.script.script_pub_key import is_segwit, type_and_payload
+from btclib.script.taproot import check_output_pubkey
+from btclib.script.witness import Witness
+from btclib.tx.tx import Tx
+from btclib.tx.tx_out import TxOut
+
+
+def taproot_unwrap_script(
+    script: bytes, stack: list[bytes]
+) -> tuple[bytes, list[bytes], int]:
+    pub_key = type_and_payload(script)[1]
+    script_bytes = stack[-2]
+    control = stack[-1]
+
+    if not check_output_pubkey(pub_key, script_bytes, control):
+        raise BTClibValueError()
+
+    leaf_version = stack[-1][0] & 0xFE
+
+    return script_bytes, stack[:-2], leaf_version
+
+
+def taproot_get_annex(witness: Witness) -> bytes:
+    annex = b""
+    if len(witness.stack) >= 2 and witness.stack[-1][0] == 0x50:
+        annex = witness.stack[-1]
+        witness.stack = witness.stack[:-1]
+    return annex
+
+
+def validate_redeem_script(redeem_script: ScriptList) -> None:
+    for c in redeem_script:
+        if isinstance(c, str):
+            if c == "OP_1NEGATE":
+                continue
+            if c[:2] == "OP" and not c[3:].isdigit():
+                raise BTClibValueError()
+
+
+ALL_FLAGS = [
+    "P2SH",
+    # Bip 62, never finalized
+    # "SIGPUSHONLY",
+    # "LOW_S",
+    # "STRICTENC",
+    # "CONST_SCRIPTCODE",
+    # "CLEANSTACK",
+    # "MINIMALDATA",
+    "DERSIG",
+    # only standard, not consensus
+    # "NULLFAIL",
+    # "MINMALIF",
+    # "DISCOURAGE_UPGRADABLE_NOPS",
+    # "DISCOURAGE_UPGRADABLE_WITNESS_PROGRAM",
+    "CHECKLOCKTIMEVERIFY",
+    "CHECKSEQUENCEVERIFY",
+    "WITNESS",
+    "NULLDUMMY",
+    # only standard, not strictly consensus
+    # "WITNESS_PUBKEYTYPE",
+    "TAPROOT",
+]
+
+
+def verify_input(prevouts: list[TxOut], tx: Tx, i: int, flags: list[str]) -> None:
+    script_sig = tx.vin[i].script_sig
+    parsed_script_sig = parse(script_sig, accept_unknown=True)
+    if "SIGPUSHONLY" in flags:
+        validate_redeem_script(parsed_script_sig)
+    if "CONST_SCRIPTCODE" in flags:
+        for x in parsed_script_sig:
+            op_checks = [
+                "OP_CHECKSIG",
+                "OP_CHECKSIGVERIFY",
+                "OP_CHECKMULTISIG",
+                "OP_CHECKSIGVERIFY",
+            ]
+            if x in op_checks:
+                raise BTClibValueError()
+    stack: list[bytes] = []
+    verify_script_legacy(
+        script_sig, stack, prevouts[i].value, tx, i, flags, False, False
+    )
+    p2sh_script = stack[-1] if stack else b"\x00"
+
+    script = prevouts[i].script_pub_key.script
+    verify_script_legacy(script, stack, prevouts[i].value, tx, i, flags, False, True)
+
+    script_type, payload = type_and_payload(script)
+
+    p2sh = False
+    if script_type == "p2sh" and "P2SH" in flags:
+        p2sh = True
+        validate_redeem_script(parsed_script_sig)  # similar to SIGPUSHONLY
+        script = p2sh_script
+        verify_script_legacy(
+            script, stack, prevouts[i].value, tx, i, flags, False, True
+        )
+        script_type, payload = type_and_payload(script)
+
+    segwit_version = _to_num(stack[-1], []) if is_segwit(script) else -1
+    supported_segwit_version = -1
+    if "WITNESS" in flags:
+        supported_segwit_version = 0
+    if "TAPROOT" in flags:
+        supported_segwit_version = 1
+    if segwit_version + 1 and tx.vin[i].script_sig and not p2sh:
+        raise BTClibValueError()
+    if not (segwit_version + 1) and tx.vin[i].script_witness:
+        raise BTClibValueError()  # witness without witness script
+    if segwit_version > supported_segwit_version:
+        if segwit_version + 1 and "DISCOURAGE_UPGRADABLE_WITNESS_PROGRAM" in flags:
+            raise BTClibValueError()
+        return
+
+    if segwit_version == 1:
+        if script_type == "p2tr":
+            if p2sh:
+                return  # remains unencumbered
+            witness = tx.vin[i].script_witness
+            budget = 50 + len(witness.serialize())
+            annex = taproot_get_annex(witness)
+            stack = witness.stack
+            if len(stack) == 0:
+                raise BTClibValueError()
+            if len(stack) == 1:
+                tapscript.verify_key_path(script, stack, prevouts, tx, i, annex)
+                stack = []
+            else:
+                script_bytes, stack, leaf_version = taproot_unwrap_script(script, stack)
+                if leaf_version == 0xC0:
+                    tapscript.verify_script_path_vc0(
+                        script_bytes, stack, prevouts, tx, i, annex, budget, flags
+                    )
+                else:
+                    return  # unknown program, passes validation
+
+    if segwit_version == 0:
+        if script_type == "p2wpkh":
+            stack = tx.vin[i].script_witness.stack
+            # serialization of ["OP_DUP", "OP_HASH160", payload, "OP_EQUALVERIFY", "OP_CHECKSIG"]
+            script = b"v\xa9\x14" + payload + b"\x88\xac"
+        elif script_type == "p2wsh":
+            stack = tx.vin[i].script_witness.stack
+            if any(len(x) > 520 for x in stack[:-1]):
+                raise BTClibValueError()
+            script = stack[-1]
+            if payload != sha256(script):
+                raise BTClibValueError()
+            stack = stack[:-1]
+        else:
+            raise BTClibValueError()
+
+        if "OP_CODESEPARATOR" in parse(script):
+            return
+
+        verify_script_legacy(script, stack, prevouts[i].value, tx, i, flags, True, True)
+
+    if stack and ("CLEANSTACK" in flags or segwit_version == 0):
+        raise BTClibValueError()
+
+
+def verify_transaction(
+    prevouts: list[TxOut], tx: Tx, flags: list | None = None
+) -> None:
+    if flags is None:
+        flags = ALL_FLAGS[:]
+    if len(prevouts) != len(tx.vin):
+        raise BTClibValueError()
+    for i in range(len(prevouts)):
+        verify_input(prevouts, tx, i, flags)