fork_ct_test.py raw

   1  #!/usr/bin/env python3
   2  # Copyright (c) 2026 The Limenka developers
   3  # Distributed under the MIT software license, see the accompanying
   4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5  """End-to-end confidential transaction loop on the fork standalone chain.
   6  
   7  Mints transparent value into CT, pays a stealth (lm2) address
   8  non-interactively, and verifies the wallet scans and recovers the
   9  payment from on-chain data alone.
  10  
  11  Flow:
  12    - mine mature transparent coins
  13    - mintct 10            (transparent -> confidential, kernel balance)
  14    - listctreceipts       (minted output + change are spendable)
  15    - getnewstealthaddress (lm2, view + spend keys)
  16    - sendtostealth lm2 4  (non-interactive confidential payment)
  17    - mine the payments and verify the wallet recovered the stealth output
  18    - kernel outputs carry the "BK" magic in every CT transaction
  19  """
  20  import hashlib
  21  
  22  from test_framework.address import byte_to_base58
  23  from test_framework.descriptors import descsum_create
  24  from test_framework.key import ECKey
  25  from test_framework.segwit_addr import encode_segwit_address
  26  from test_framework.test_framework import LimenkaTestFramework
  27  from test_framework.util import (
  28      assert_equal,
  29      assert_greater_than,
  30  )
  31  
  32  GENESIS_TIME = 1231006505
  33  
  34  
  35  class ForkCTTest(LimenkaTestFramework):
  36      def set_test_params(self):
  37          self.setup_clean_chain = True
  38          self.num_nodes = 1
  39          self.chain = 'limenka'
  40          self.extra_args = [[
  41              '-forkactivationtime=1',
  42              '-forkdelaysteps=1024',
  43              '-forkmineondemand',
  44              '-forkstandalone',
  45              '-fallbackfee=0.0002',
  46              '-addresstype=bech32',
  47          ]]
  48          self.rpc_timeout = 240
  49  
  50      def skip_test_if_missing_module(self):
  51          self.skip_if_no_wallet()
  52  
  53      def add_options(self, parser):
  54          self.add_wallet_options(parser, descriptors=True, legacy=False)
  55  
  56      def init_wallet(self, *, node):
  57          wallet_name = self.default_wallet_name if self.wallet_names is None else self.wallet_names[node] if node < len(self.wallet_names) else False
  58          if wallet_name is not False:
  59              n = self.nodes[node]
  60              if wallet_name is not None:
  61                  n.createwallet(wallet_name=wallet_name, descriptors=True, load_on_startup=True)
  62                  # The fork chain stamps from 2009 genesis, but a fresh wallet
  63                  # birthtime is 2026 - blocks older than the birthtime are
  64                  # skipped.  Import a deterministic funding key with an old
  65                  # timestamp so the wallet tracks the mocktime-mined chain.
  66                  seed = hashlib.sha256(b'fork-ct-funding').digest()
  67                  key = ECKey()
  68                  key.set(seed, compressed=True)
  69                  priv = key.get_bytes()
  70                  wif = byte_to_base58(priv + b'\x01', 128)  # mainnet WIF, compressed
  71                  desc = descsum_create(f"wpkh({wif})")
  72                  import_res = n.importdescriptors([{
  73                      "desc": desc,
  74                      "timestamp": GENESIS_TIME + 600,
  75                      "active": False,
  76                      "internal": False,
  77                  }])
  78                  self.log.info(f"importdescriptors result: {import_res}")
  79                  self.funding_key = priv
  80                  self.funding_pub = key.get_pubkey().get_bytes()
  81  
  82      def mine_blocks(self, node, count, addr, start_mt):
  83          mt = start_mt
  84          hashes = []
  85          for _ in range(count):
  86              node.setmocktime(mt)
  87              h = self.generatetoaddress(node, 1, addr, sync_fun=self.no_op)[0]
  88              hashes.append(h)
  89              mt += 600
  90          return hashes
  91  
  92      def run_test(self):
  93          node = self.nodes[0]
  94          bi = node.getblockchaininfo()
  95          assert_equal(bi['chain'], 'limenka')
  96          self.log.info(f"chain {bi['chain']}, delay steps {bi['fork_delay_steps']}")
  97  
  98          # The fork chain's witness HRP is "bf" (mainnet base58 prefixes).
  99          from test_framework.address import hash160
 100          p2wpkh_program = hash160(self.funding_pub)
 101          addr = encode_segwit_address("bf", 0, p2wpkh_program)
 102  
 103          # 110 blocks: coinbase maturity (100) plus margin, at ~600s mocktime.
 104          self.log.info(f"addr to mine: {addr}")
 105          self.mine_blocks(node, 110, addr, GENESIS_TIME + 600)
 106          self.log.info(f"walletinfo: {node.getwalletinfo()}")
 107          self.log.info(f"listdescriptors: {node.listdescriptors()}")
 108          self.log.info(f"unspent: {node.listunspent()}")
 109          bal = node.getbalance()
 110          self.log.info(f"transparent balance after mining: {bal}")
 111          assert_greater_than(bal, 100)
 112  
 113          # --- Stealth address ---
 114          stealth = node.getnewstealthaddress()
 115          self.log.info(f"stealth address: {stealth['address']}")
 116          assert stealth['address'].startswith('lm2')
 117          assert_equal(len(stealth['view_pubkey']), 66)
 118          assert_equal(len(stealth['spend_pubkey']), 66)
 119  
 120          # --- Mint: transparent value enters the confidential domain ---
 121          # Explicit fee: 1000 sats (1e-5 lambda) - comfortably above min relay.
 122          # Split the minted value into 3 random confidential outputs: every
 123          # output is CT (a transparent change would be rejected and would
 124          # break the privacy).
 125          mint_txid = node.mintct("10", "0.00001", 3)['txid']
 126          self.log.info(f"mint txid: {mint_txid}")
 127          receipts = node.listctreceipts()
 128          self.log.info(f"receipts after mint: {receipts}")
 129          mint_receipts = [r for r in receipts if r['txid'] == mint_txid]
 130          assert_equal(len(mint_receipts), 3)
 131          # All three outputs are confidential and positive (no transparent
 132          # change leaked).
 133          for r in mint_receipts:
 134              assert_greater_than(float(r['amount']), 0)
 135  
 136          # Every CT tx carries the BK kernel.
 137          mint_raw = node.getrawtransaction(mint_txid, True)
 138          kernel_hexes = [v['scriptPubKey']['hex'] for v in mint_raw['vout']]
 139          assert any(h.startswith('6a02424b') for h in kernel_hexes), kernel_hexes
 140  
 141          # --- Non-interactive stealth payment ---
 142          pay_txid = node.sendtostealth(stealth['address'], "4", "0.00001")['txid']
 143          self.log.info(f"stealth payment txid: {pay_txid}")
 144          pay_raw = node.getrawtransaction(pay_txid, True)
 145          kernel_hexes = [v['scriptPubKey']['hex'] for v in pay_raw['vout']]
 146          assert any(h.startswith('6a02424b') for h in kernel_hexes), kernel_hexes
 147  
 148          # Mine both transactions.
 149          self.mine_blocks(node, 1, addr, GENESIS_TIME + 111 * 600)
 150  
 151          # PSBT round-trip: build an unsigned CT mint, finalize it (sign the
 152          # transparent inputs + the kernel), and broadcast the result.
 153          psbt = node.createctpsbt("2", "0.00001", 2)
 154          self.log.info(f"created CT psbt (len {len(psbt)})")
 155          final_tx_hex = node.finalizectpsbt(psbt)
 156          self.log.info(f"finalized CT tx hex len {len(final_tx_hex)}")
 157          accept = node.testmempoolaccept([final_tx_hex])
 158          assert accept[0]['allowed'], accept
 159          node.sendrawtransaction(final_tx_hex)
 160          self.mine_blocks(node, 1, addr, GENESIS_TIME + 112 * 600)
 161          receipts = node.listctreceipts()
 162          self.log.info(f"receipts after psbt mint: {receipts}")
 163  
 164          # PSBT stealth-spend: build an unsigned CT spend to a fresh stealth
 165          # address, finalize it (attach proofs + sign the kernel), and
 166          # broadcast.  The receiver's blinding rides the kernel offset.
 167          stealth2 = node.getnewstealthaddress()
 168          stealth_psbt = node.createstealthpsbt(stealth2['address'], "1", "0.00001", 1)
 169          self.log.info(f"created stealth CT psbt (len {len(stealth_psbt)})")
 170          stealth_hex = node.finalizectpsbt(stealth_psbt)
 171          self.log.info(f"finalized stealth CT tx hex len {len(stealth_hex)}")
 172          accept2 = node.testmempoolaccept([stealth_hex])
 173          assert accept2[0]['allowed'], accept2
 174          node.sendrawtransaction(stealth_hex)
 175          self.mine_blocks(node, 1, addr, GENESIS_TIME + 113 * 600)
 176          receipts = node.listctreceipts()
 177          self.log.info(f"receipts after psbt stealth: {receipts}")
 178  
 179          # lm1 (P2SPKH) spend via PSBT on the fork chain: exercises the
 180          # forkid Schnorr signing (the marker binds in the hash, not the
 181          # BIP341 type byte) and the PSBT signing path forkid.
 182          lm1_addr = node.getnewaddress("", "p2spkh")
 183          assert lm1_addr.startswith('lm1'), lm1_addr
 184          node.sendtoaddress(lm1_addr, "2")
 185          self.mine_blocks(node, 1, addr, GENESIS_TIME + 114 * 600)
 186          unspent = node.listunspent(0, 9999999, [lm1_addr])
 187          assert_equal(len(unspent), 1)
 188          utxo = unspent[0]
 189          psbt = node.walletcreatefundedpsbt(
 190              [{"txid": utxo["txid"], "vout": utxo["vout"]}],
 191              {addr: "1.5"}, 0, {"subtractFeeFromOutputs": [0]})["psbt"]
 192          processed = node.walletprocesspsbt(psbt)
 193          assert processed["complete"], processed
 194          signed_hex = node.finalizepsbt(processed["psbt"])["hex"]
 195          accept = node.testmempoolaccept([signed_hex])
 196          assert accept[0]["allowed"], accept
 197          node.sendrawtransaction(signed_hex)
 198          self.mine_blocks(node, 1, addr, GENESIS_TIME + 115 * 600)
 199          self.log.info("lm1 PSBT spend complete")
 200  
 201          # 128-bit precision: getbalances().mine.precise includes the
 202          # confidential outputs (10 - 4 spent = 6 remaining from the mint
 203          # change, plus the stealth 4 and its ~6 change) on top of the
 204          # transparent balance.
 205          balances = node.getbalances()
 206          self.log.info(f"precise balance: {balances['mine'].get('precise')}")
 207          assert 'precise' in balances['mine']
 208  
 209          # The wallet scans on-chain data and recovers the stealth payment:
 210          # a spendable receipt for exactly 4 lambda must appear.
 211          receipts = node.listctreceipts()
 212          self.log.info(f"receipts after payment: {receipts}")
 213          amounts = {r['amount'] for r in receipts}
 214          assert '4' in amounts, f"stealth payment not recovered: {amounts}"
 215          assert '1' in amounts, f"psbt stealth payment not recovered: {amounts}"
 216  
 217          self.log.info("CT loop complete: mint -> stealth pay -> scan -> spendable")
 218  
 219  if __name__ == '__main__':
 220      ForkCTTest(__file__).main()
 221