fix: differentiate legacy and segwit types for p2sh when using Descriptor::scriptpubkey_type

src/embit/bip39.py

@@ -11,7 +11,7 @@ def mnemonic_to_bytes(mnemonic: str, ignore_checksum: bool = False, wordlist=WOR
     # this function is copied from Jimmy Song's HDPrivateKey.from_mnemonic() method
 
     words = mnemonic.strip().split()
-    if len(words) % 3 != 0 or len(words) < 12:
+    if len(words) % 3 != 0 or not 12 <= len(words) <= 24:
         raise ValueError("Invalid recovery phrase")
 
     binary_seed = bytearray()
@@ -97,7 +97,7 @@ def _extract_index(bits, b, n):
 
 
 def mnemonic_from_bytes(entropy, wordlist=WORDLIST):
-    if len(entropy) % 4 != 0:
+    if len(entropy) % 4 != 0 or not 16 <= len(entropy) <= 32:
         raise ValueError("Byte array should be multiple of 4 long (16, 20, ..., 32)")
     total_bits = len(entropy) * 8
     checksum_bits = total_bits // 32

src/embit/bip85.py

@@ -17,7 +17,8 @@ def derive_entropy(root, app_index, path):
     """
     Derive app-specific bip85 entropy using path m/83696968'/app_index'/...path'
     """
-    assert max(path) < HARDENED_INDEX
+    if max(path) >= HARDENED_INDEX:
+        raise ValueError("Path elements must be less than 2^31")
     derivation = [HARDENED_INDEX + 83696968, HARDENED_INDEX + app_index] + [
         p + HARDENED_INDEX for p in path
     ]
@@ -27,7 +28,8 @@ def derive_entropy(root, app_index, path):
 
 def derive_mnemonic(root, num_words=12, index=0, language=LANGUAGES.ENGLISH):
     """Derive a new mnemonic with num_words using language (code, wordlist)"""
-    assert num_words in [12, 18, 24]
+    if num_words not in [12, 18, 24]:
+        raise ValueError("Number of words must be 12, 18 or 24")
     langcode, wordlist = language
     path = [langcode, num_words, index]
     entropy = derive_entropy(root, 39, path)
@@ -49,7 +51,9 @@ def derive_xprv(root, index=0):
 
 def derive_hex(root, num_bytes=32, index=0):
     """Derive raw entropy from 16 to 64 bytes long"""
-    assert num_bytes <= 64
-    assert num_bytes >= 16
+    if num_bytes > 64:
+        raise ValueError("Number of bytes must not exceed 64")
+    if num_bytes < 16:
+        raise ValueError("Number of bytes must be at least 16")
     entropy = derive_entropy(root, 128169, [num_bytes, index])
     return entropy[:num_bytes]

src/embit/descriptor/arguments.py

@@ -15,7 +15,8 @@ def __init__(self, fingerprint: bytes, derivation: list):
     def from_string(cls, s: str):
         arr = s.split("/")
         mfp = unhexlify(arr[0])
-        assert len(mfp) == 4
+        if len(mfp) != 4:
+            raise ArgumentError("Invalid fingerprint length")
         arr[0] = "m"
         path = "/".join(arr)
         derivation = bip32.parse_path(path)
@@ -315,7 +316,8 @@ def xonly(self):
         return self.key.xonly()
 
     def taproot_tweak(self, h=b""):
-        assert self.taproot
+        if not self.taproot:
+            raise ArgumentError("Key is not taproot")
         return self.key.taproot_tweak(h)
 
     def serialize(self):

src/embit/descriptor/descriptor.py

@@ -59,6 +59,9 @@ def script_len(self):
 
     @property
     def num_branches(self):
+        if self.miniscript is not None:
+            return max({k.num_branches for k in self.miniscript.keys})
+
         return max([k.num_branches for k in self.keys])
 
     def branch(self, branch_index=None):
@@ -88,7 +91,7 @@ def is_wildcard(self):
 
     @property
     def is_wrapped(self):
-        return self.sh and self.is_segwit
+        return bool(self.sh and self.wpkh) or bool(self.sh and self.wsh)
 
     @property
     def is_legacy(self):
@@ -97,7 +100,9 @@ def is_legacy(self):
     @property
     def is_segwit(self):
         return (
-            (self.wsh and self.miniscript) or (self.wpkh and self.key) or self.taproot
+            bool(self.wsh and self.miniscript)
+            or bool(self.wpkh and self.key)
+            or bool(self.taproot)
         )
 
     @property
@@ -118,18 +123,28 @@ def is_basic_multisig(self) -> bool:
     def is_sorted(self) -> bool:
         return bool(self.is_basic_multisig and isinstance(self.miniscript, Sortedmulti))
 
+    @property
+    def is_miniscript(self) -> bool:
+        return bool(
+            self.miniscript and not isinstance(self.miniscript, (Multi, Sortedmulti))
+        )
+
     def scriptpubkey_type(self):
         if self.is_taproot:
             return "p2tr"
         if self.sh:
+            if self.wpkh:
+                return "p2sh-p2wpkh"
+            if self.wsh:
+                return "p2sh-p2wsh"
             return "p2sh"
+        if self.wsh:
+            return "p2wsh"
         if self.is_pkh:
             if self.is_legacy:
                 return "p2pkh"
             if self.is_segwit:
                 return "p2wpkh"
-        else:
-            return "p2wsh"
 
     @property
     def brief_policy(self):
@@ -204,7 +219,7 @@ def owns(self, psbt_scope):
         if psbt_scope.script_pubkey is None:
             return False
         # quick check of script_pubkey type
-        if psbt_scope.script_pubkey.script_type() != self.scriptpubkey_type():
+        if psbt_scope.script_pubkey.script_type() not in self.scriptpubkey_type():
             return False
         for pub, der in psbt_scope.bip32_derivations.items():
             # check of the fingerprints
@@ -214,9 +229,30 @@ def owns(self, psbt_scope):
                 res = k.check_derivation(der)
                 if res:
                     idx, branch_idx = res
-                    sc = self.derive(idx, branch_index=branch_idx).script_pubkey()
-                    # if derivation is found but scriptpubkey doesn't match - fail
-                    return sc == psbt_scope.script_pubkey
+
+                    # get the script from the scope,
+                    # it can be either legacy, wrapped segwit or segwit v0
+                    sc = self.derive(idx, branch_index=branch_idx)
+
+                    # If derivation is found but scriptpubkey doesn't match - fail.
+                    # In case of wrapped segwit we need to check redeem script too.
+                    # In case of non-wrapped segwit we can check script_pubkey directly.
+                    if sc.is_wrapped:
+                        redeem_sc = sc.redeem_script()
+
+                        # you can check script_type of both
+                        # script_pubkey and redeem_script
+                        # to make sure they match the psbt_scope
+                        # print("%s-%s" % (sc.script_pubkey().script_type(), redeem_sc.script_type()))
+                        return (
+                            sc.script_pubkey() == psbt_scope.script_pubkey
+                            and redeem_sc.script_type() == psbt_scope.redeem_script.script_type()
+                        )
+                    else:
+                        # you can check script_type
+                        # to make sure they match the psbt_scope
+                        # print("%s" % sc.script_pubkey().script_type())
+                        return sc.script_pubkey() == psbt_scope.script_pubkey
         for pub, (leafs, der) in psbt_scope.taproot_bip32_derivations.items():
             # check of the fingerprints
             for k in self.keys:
@@ -292,7 +328,7 @@ def from_string(cls, desc):
     @classmethod
     def read_from(cls, s):
         # starts with sh(wsh()), sh() or wsh()
-        start = s.read(7)
+        start = s.read(8)
         sh = False
         wsh = False
         wpkh = False
@@ -301,30 +337,30 @@ def read_from(cls, s):
         taptree = TapTree()
         if start.startswith(b"tr("):
             taproot = True
-            s.seek(-4, 1)
+            s.seek(-5, 1)
         elif start.startswith(b"sh(wsh("):
             sh = True
             wsh = True
+            s.seek(-1, 1)
         elif start.startswith(b"wsh("):
             sh = False
             wsh = True
-            s.seek(-3, 1)
-        elif start.startswith(b"sh(wpkh"):
+            s.seek(-4, 1)
+        elif start.startswith(b"sh(wpkh("):
             is_miniscript = False
             sh = True
             wpkh = True
-            assert s.read(1) == b"("
         elif start.startswith(b"wpkh("):
             is_miniscript = False
             wpkh = True
-            s.seek(-2, 1)
+            s.seek(-3, 1)
         elif start.startswith(b"pkh("):
             is_miniscript = False
-            s.seek(-3, 1)
+            s.seek(-4, 1)
         elif start.startswith(b"sh("):
             sh = True
             wsh = False
-            s.seek(-4, 1)
+            s.seek(-5, 1)
         else:
             raise ValueError("Invalid descriptor (starts with '%s')" % start.decode())
         # taproot always has a key, and may have taptree miniscript

src/embit/descriptor/miniscript.py

@@ -54,11 +54,25 @@ def type(self):
 
     @classmethod
     def read_from(cls, s, taproot=False):
-        op, char = read_until(s, b"(")
+        def wrapped(m_script):
+            for w in reversed(wrappers):
+                if w not in WRAPPER_NAMES:
+                    raise MiniscriptError("Unknown wrapper")
+                WrapperCls = WRAPPERS[WRAPPER_NAMES.index(w)]
+                m_script = WrapperCls(m_script, taproot=taproot)
+            return m_script
+
+        op, char = read_until(s, b"(,)")
+        if char in (b",", b")"):
+            s.seek(-1, 1)
         op = op.decode()
         wrappers = ""
         if ":" in op:
             wrappers, op = op.split(":")
+        # handle boolean literals: 0 or 1
+        if op in ("0", "1"):
+            miniscript = JustOne() if op == "1" else JustZero()
+            return wrapped(miniscript)
         if char != b"(":
             raise MiniscriptError("Missing operator")
         if op not in OPERATOR_NAMES:
@@ -67,12 +81,7 @@ def read_from(cls, s, taproot=False):
         MiniscriptCls = OPERATORS[OPERATOR_NAMES.index(op)]
         args = MiniscriptCls.read_arguments(s, taproot=taproot)
         miniscript = MiniscriptCls(*args, taproot=taproot)
-        for w in reversed(wrappers):
-            if w not in WRAPPER_NAMES:
-                raise MiniscriptError("Unknown wrapper")
-            WrapperCls = WRAPPERS[WRAPPER_NAMES.index(w)]
-            miniscript = WrapperCls(miniscript, taproot=taproot)
-        return miniscript
+        return wrapped(miniscript)
 
     @classmethod
     def read_arguments(cls, s, taproot=False):
@@ -119,6 +128,28 @@ def len_args(self):
 ########### Known fragments (miniscript operators) ##############
 
 
+class JustZero(Miniscript):
+    TYPE = "B"
+    PROPS = "zud"
+
+    def inner_compile(self):
+        return Number(0).compile()
+
+    def __str__(self):
+        return "0"
+
+
+class JustOne(Miniscript):
+    TYPE = "B"
+    PROPS = "zu"
+
+    def inner_compile(self):
+        return Number(1).compile()
+
+    def __str__(self):
+        return "1"
+
+
 class OneArg(Miniscript):
     NARGS = 1
 
@@ -870,6 +901,11 @@ def inner_compile(self):
 
     def __len__(self):
         return len(self.arg) + 1
+
+    def verify(self):
+        super().verify()
+        if self.arg.type != "V":
+            raise MiniscriptError("t: X must be of type V")
 
     @property
     def properties(self):

src/embit/ec.py

@@ -26,7 +26,8 @@ def read_from(cls, stream):
 
 class SchnorrSig(EmbitBase):
     def __init__(self, sig):
-        assert len(sig) == 64
+        if len(sig) != 64:
+            raise ECError("Invalid schnorr signature")
         self._sig = sig
 
     def write_to(self, stream) -> int:
@@ -93,7 +94,8 @@ def _xonly(self):
 
     @classmethod
     def from_xonly(cls, data: bytes):
-        assert len(data) == 32
+        if len(data) != 32:
+            raise ECError("Invalid xonly pubkey")
         return cls.parse(b"\x02" + data)
 
     def schnorr_verify(self, sig, msg_hash) -> bool:

src/embit/liquid/pset.py

@@ -96,9 +96,11 @@ def unblind(self, blinding_key):
             return
         # verify
         gen = secp256k1.generator_generate_blinded(asset, in_abf)
-        assert gen == secp256k1.generator_parse(self.utxo.asset)
+        if gen != secp256k1.generator_parse(self.utxo.asset):
+            raise PSBTError("Invalid asset commitment")
         cmt = secp256k1.pedersen_commit(vbf, value, gen)
-        assert cmt == secp256k1.pedersen_commitment_parse(self.utxo.value)
+        if cmt != secp256k1.pedersen_commitment_parse(self.utxo.value):
+            raise PSBTError("Invalid value commitment")
 
         self.asset = asset
         self.value = value
@@ -506,7 +508,8 @@ def unblind(self, blinding_key):
             inp.unblind(blinding_key)
 
     def txseed(self, seed: bytes):
-        assert len(seed) == 32
+        if len(seed) != 32:
+            raise PSBTError("Seed should be 32 bytes")
         # get unique seed for this tx:
         # we use seed + txid:vout + scriptpubkey as unique data for tagged hash
         data = b"".join(

src/embit/liquid/psetview.py

@@ -5,7 +5,8 @@
 
 def skip_commitment(stream):
     c = stream.read(1)
-    assert len(c) == 1
+    if len(c) != 1:
+        raise PSBTError("Unexpected end of stream")
     if c == b"\x00":  # None
         return 1
     if c == b"\x01":  # unconfidential