all: add checks for prev_hash size

Author: matejcik

core/src/apps/wallet/sign_tx/helpers.py

@@ -17,6 +17,7 @@
 from trezor.messages.TxRequest import TxRequest
 
 from .signing import SigningError
+from .writers import TX_HASH_SIZE
 
 from apps.common.coininfo import CoinInfo
 
@@ -213,6 +214,8 @@ def sanitize_tx_input(tx: TransactionType, coin: CoinInfo) -> TxInputType:
         txi.script_type = InputScriptType.SPENDADDRESS
     if txi.sequence is None:
         txi.sequence = 0xFFFFFFFF
+    if txi.prev_hash is None or len(txi.prev_hash) != TX_HASH_SIZE:
+        raise SigningError(FailureType.DataError, "Provided prev_hash is invalid.")
     if txi.multisig and txi.script_type not in MULTISIG_INPUT_SCRIPT_TYPES:
         raise SigningError(
             FailureType.DataError, "Multisig field provided but not expected.",

legacy/firmware/signing.c

@@ -588,6 +588,12 @@ void signing_init(const SignTx *msg, const CoinInfo *_coin,
 #define MIN(a, b) (((a) < (b)) ? (a) : (b))
 
 static bool signing_validate_input(const TxInputType *txinput) {
+  if (txinput->prev_hash.size != 32) {
+    fsm_sendFailure(FailureType_Failure_ProcessError,
+                    _("Encountered invalid prevhash"));
+    signing_abort();
+    return false;
+  }
   if (txinput->has_multisig &&
       (txinput->script_type != InputScriptType_SPENDMULTISIG &&
        txinput->script_type != InputScriptType_SPENDP2SHWITNESS &&

tests/device_tests/test_msg_signtx_prevhash.py

@@ -0,0 +1,164 @@
+from hashlib import sha256
+from io import BytesIO
+
+import pytest
+
+from trezorlib import btc, messages
+from trezorlib.exceptions import TrezorFailure
+
+from ..tx_cache import tx_cache
+
+TXHASH_157041 = bytes.fromhex(
+    "1570416eb4302cf52979afd5e6909e37d8fdd874301f7cc87e547e509cb1caa6"
+)
+
+
+def write_prefixed_bytes(io, data) -> None:
+    assert len(data) < 253
+    io.write(len(data).to_bytes(1, "little"))
+    io.write(data)
+
+
+def serialize_input(tx_input) -> bytes:
+    """serialize for Bitcoin tx format"""
+    b = BytesIO()
+    if tx_input.prev_hash:
+        b.write(tx_input.prev_hash[::-1])
+    b.write(tx_input.prev_index.to_bytes(4, "little"))
+    write_prefixed_bytes(b, tx_input.script_sig)
+    b.write(tx_input.sequence.to_bytes(4, "little"))
+    return b.getvalue()
+
+
+def serialize_bin_output(tx_output) -> bytes:
+    b = BytesIO()
+    b.write(tx_output.amount.to_bytes(8, "little"))
+    write_prefixed_bytes(b, tx_output.script_pubkey)
+    return b.getvalue()
+
+
+def serialize_tx(tx) -> bytes:
+    b = BytesIO()
+    b.write(tx.version.to_bytes(4, "little"))
+    assert len(tx.inputs) < 253
+    b.write(len(tx.inputs).to_bytes(1, "little"))
+    for inp in tx.inputs:
+        b.write(serialize_input(inp))
+    assert len(tx.bin_outputs) < 253
+    b.write(len(tx.bin_outputs).to_bytes(1, "little"))
+    for outp in tx.bin_outputs:
+        b.write(serialize_bin_output(outp))
+    b.write(tx.lock_time.to_bytes(4, "little"))
+    if tx.extra_data:
+        b.write(tx.extra_data)
+    return b.getvalue()
+
+
+def hash_tx(data: bytes) -> bytes:
+    return sha256(sha256(data).digest()).digest()[::-1]
+
+
+def _check_error_message(value: bytes, model: str, message: str):
+    if model != "1":
+        assert message == "Provided prev_hash is invalid."
+        return
+
+    # T1 has several possible errors
+    if value is None:
+        assert message.endswith("missing required field")
+    elif len(value) > 32:
+        assert message.endswith("bytes overflow")
+    else:
+        assert message.endswith("Encountered invalid prevhash")
+
+
+@pytest.mark.skip_ui
+@pytest.mark.parametrize("prev_hash", (None, b"", b"x", b"hello world", b"x" * 33))
+def test_invalid_prev_hash(client, prev_hash):
+    inp1 = messages.TxInputType(
+        address_n=[0],
+        amount=123456789,
+        prev_hash=prev_hash,
+        prev_index=0,
+        script_type=messages.InputScriptType.SPENDP2SHWITNESS,
+    )
+    out1 = messages.TxOutputType(
+        address="mhRx1CeVfaayqRwq5zgRQmD7W5aWBfD5mC",
+        amount=12300000,
+        script_type=messages.OutputScriptType.PAYTOADDRESS,
+    )
+
+    with pytest.raises(TrezorFailure) as e:
+        btc.sign_tx(client, "Testnet", [inp1], [out1])
+    _check_error_message(prev_hash, client.features.model, e.value.failure.message)
+
+
+@pytest.mark.skip_ui
+@pytest.mark.parametrize("prev_hash", (None, b"", b"x", b"hello world", b"x" * 33))
+def test_invalid_prev_hash_attack(client, prev_hash):
+    # prepare input with a valid prev-hash
+    inp1 = messages.TxInputType(
+        address_n=[0],
+        amount=123456789,
+        prev_hash=b"\x00" * 32,
+        prev_index=0,
+        script_type=messages.InputScriptType.SPENDP2SHWITNESS,
+    )
+    out1 = messages.TxOutputType(
+        address="mhRx1CeVfaayqRwq5zgRQmD7W5aWBfD5mC",
+        amount=12300000,
+        script_type=messages.OutputScriptType.PAYTOADDRESS,
+    )
+
+    counter = 1
+
+    def attack_filter(msg):
+        nonlocal counter
+
+        if not msg.tx.inputs:
+            return msg
+
+        # on first attempt, send unmodified input
+        if counter > 0:
+            counter -= 1
+            return msg
+
+        # on second request, send modified input
+        msg.tx.inputs[0].prev_hash = prev_hash
+        return msg
+
+    with client, pytest.raises(TrezorFailure) as e:
+        client.set_filter(messages.TxAck, attack_filter)
+        btc.sign_tx(client, "Testnet", [inp1], [out1])
+
+    # check that injection was performed
+    assert counter == 0
+    _check_error_message(prev_hash, client.features.model, e.value.failure.message)
+
+
+@pytest.mark.skip_ui
+@pytest.mark.parametrize("prev_hash", (None, b"", b"x", b"hello world", b"x" * 33))
+def test_invalid_prev_hash_in_prevtx(client, prev_hash):
+    cache = tx_cache("Bitcoin")
+    prev_tx = cache[TXHASH_157041]
+
+    # smoke check: replace prev_hash with all zeros, reserialize and hash, try to sign
+    prev_tx.inputs[0].prev_hash = b"\x00" * 32
+    tx_hash = hash_tx(serialize_tx(prev_tx))
+
+    inp0 = messages.TxInputType(address_n=[0], prev_hash=tx_hash, prev_index=0)
+    out1 = messages.TxOutputType(
+        address="1MJ2tj2ThBE62zXbBYA5ZaN3fdve5CPAz1",
+        amount=1000,
+        script_type=messages.OutputScriptType.PAYTOADDRESS,
+    )
+    btc.sign_tx(client, "Bitcoin", [inp0], [out1], prev_txes={tx_hash: prev_tx})
+
+    # attack: replace prev_hash with an invalid value
+    prev_tx.inputs[0].prev_hash = prev_hash
+    tx_hash = hash_tx(serialize_tx(prev_tx))
+    inp0.prev_hash = tx_hash
+
+    with pytest.raises(TrezorFailure) as e:
+        btc.sign_tx(client, "Bitcoin", [inp0], [out1], prev_txes={tx_hash: prev_tx})
+    _check_error_message(prev_hash, client.features.model, e.value.failure.message)