Merge bitcoin/bitcoin#31907: qa: clarify and document one assumeutxo test case with malleated snapshot

e5ff4e4 qa: use a clearer and documented amount error in malleated snapshot (Antoine Poinsot)
b34fdb5 test: introduce output amount (de)compression routines (Sebastian Falbesoner)
a7911ed test: introduce VARINT (de)serialization routines (Sebastian Falbesoner)

Pull request description:

  The `feature_assumeutxo.py` functional test checks various errors with malleated snapshots. Some of these cases are brittle or use confusing and undocumented values. Fix one of those by using a clear, documented and forward-compatible value.

  I ran across those when working on an unrelated changeset which affected the snapshot. It took me a while to understand where the seemingly magic byte string was coming from, so i figured it was worth proposing this patch on its own for the sake of making the test more maintainable.

  See commit messages for details.

ACKs for top commit:
  janb84:
    re ACK [e5ff4e4](bitcoin/bitcoin@e5ff4e4)
  theStack:
    ACK e5ff4e4
  fjahr:
    Code review ACK e5ff4e4
  i-am-yuvi:
    tACK e5ff4e4

Tree-SHA512: 60f022b7176836ce05e8f287b436329d7ca6460f3fcd95f78cd24e07a95a7d4d9cbbb68a117916a113fe451732b09a012d300fe860ff33d61823eca797ceddaf

Author: ryanofsky

test/functional/feature_assumeutxo.py

@@ -16,11 +16,16 @@
         create_block,
         create_coinbase
 )
+from test_framework.compressor import (
+    compress_amount,
+)
 from test_framework.messages import (
     CBlockHeader,
     from_hex,
     msg_headers,
-    tx_from_hex
+    tx_from_hex,
+    ser_varint,
+    MAX_MONEY,
 )
 from test_framework.p2p import (
     P2PInterface,
@@ -139,7 +144,14 @@ def expected_error(msg):
             [b"\x81", 34, "3da966ba9826fb6d2604260e01607b55ba44e1a5de298606b08704bc62570ea8", None],  # wrong coin code VARINT
             [b"\x80", 34, "091e893b3ccb4334378709578025356c8bcb0a623f37c7c4e493133c988648e5", None],  # another wrong coin code
             [b"\x84\x58", 34, None, "Bad snapshot data after deserializing 0 coins"],  # wrong coin case with height 364 and coinbase 0
-            [b"\xCA\xD2\x8F\x5A", 39, None, "Bad snapshot data after deserializing 0 coins - bad tx out value"],  # Amount exceeds MAX_MONEY
+            [
+                # compressed txout value + scriptpubkey
+                ser_varint(compress_amount(MAX_MONEY + 1)) + ser_varint(0),
+                # txid + coins per txid + vout + coin height
+                32 + 1 + 1 + 2,
+                None,
+                "Bad snapshot data after deserializing 0 coins - bad tx out value"
+            ],  # Amount exceeds MAX_MONEY
         ]
 
         for content, offset, wrong_hash, custom_message in cases:

test/functional/feature_framework_unit_tests.py

@@ -18,6 +18,7 @@
     "address",
     "crypto.bip324_cipher",
     "blocktools",
+    "compressor",
     "crypto.chacha20",
     "crypto.ellswift",
     "key",

test/functional/test_framework/compressor.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+# Copyright (c) 2025-present The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Routines for compressing transaction output amounts and scripts."""
+import unittest
+
+from .messages import COIN
+
+
+def compress_amount(n):
+    if n == 0:
+        return 0
+    e = 0
+    while ((n % 10) == 0) and (e < 9):
+        n //= 10
+        e += 1
+    if e < 9:
+        d = n % 10
+        assert (d >= 1 and d <= 9)
+        n //= 10
+        return 1 + (n*9 + d - 1)*10 + e
+    else:
+        return 1 + (n - 1)*10 + 9
+
+
+def decompress_amount(x):
+    if x == 0:
+        return 0
+    x -= 1
+    e = x % 10
+    x //= 10
+    n = 0
+    if e < 9:
+        d = (x % 9) + 1
+        x //= 9
+        n = x * 10 + d
+    else:
+        n = x + 1
+    while e > 0:
+        n *= 10
+        e -= 1
+    return n
+
+
+class TestFrameworkCompressor(unittest.TestCase):
+    def test_amount_compress_decompress(self):
+        def check_amount(amount, expected_compressed):
+            self.assertEqual(compress_amount(amount), expected_compressed)
+            self.assertEqual(decompress_amount(expected_compressed), amount)
+
+        # test cases from compress_tests.cpp:compress_amounts
+        check_amount(0, 0x0)
+        check_amount(1, 0x1)
+        check_amount(1000000, 0x7)
+        check_amount(COIN, 0x9)
+        check_amount(50*COIN, 0x32)
+        check_amount(21000000*COIN, 0x1406f40)

test/functional/test_framework/messages.py

@@ -120,6 +120,26 @@ def deser_compact_size(f):
     return nit
 
 
+def ser_varint(l):
+    r = b""
+    while True:
+        r = bytes([(l & 0x7f) | (0x80 if len(r) > 0 else 0x00)]) + r
+        if l <= 0x7f:
+            return r
+        l = (l >> 7) - 1
+
+
+def deser_varint(f):
+    n = 0
+    while True:
+        dat = f.read(1)[0]
+        n = (n << 7) | (dat & 0x7f)
+        if (dat & 0x80) > 0:
+            n += 1
+        else:
+            return n
+
+
 def deser_string(f):
     nit = deser_compact_size(f)
     return f.read(nit)
@@ -1913,3 +1933,20 @@ def check_addrv2(ip, net):
         check_addrv2("2bqghnldu6mcug4pikzprwhtjjnsyederctvci6klcwzepnjd46ikjyd.onion", CAddress.NET_TORV3)
         check_addrv2("255fhcp6ajvftnyo7bwz3an3t4a4brhopm3bamyh2iu5r3gnr2rq.b32.i2p", CAddress.NET_I2P)
         check_addrv2("fc32:17ea:e415:c3bf:9808:149d:b5a2:c9aa", CAddress.NET_CJDNS)
+
+    def test_varint_encode_decode(self):
+        def check_varint(num, expected_encoding_hex):
+            expected_encoding = bytes.fromhex(expected_encoding_hex)
+            self.assertEqual(ser_varint(num), expected_encoding)
+            self.assertEqual(deser_varint(BytesIO(expected_encoding)), num)
+
+        # test cases from serialize_tests.cpp:varint_bitpatterns
+        check_varint(0, "00")
+        check_varint(0x7f, "7f")
+        check_varint(0x80, "8000")
+        check_varint(0x1234, "a334")
+        check_varint(0xffff, "82fe7f")
+        check_varint(0x123456, "c7e756")
+        check_varint(0x80123456, "86ffc7e756")
+        check_varint(0xffffffff, "8efefefe7f")
+        check_varint(0xffffffffffffffff, "80fefefefefefefefe7f")