diff --git a/tests/integration/wallet/test_transactions.py b/tests/integration/wallet/test_transactions.py index 3ce121ff8..9951f18d5 100644 --- a/tests/integration/wallet/test_transactions.py +++ b/tests/integration/wallet/test_transactions.py @@ -1,7 +1,11 @@ +import binascii import logging import asyncio +from lbrynet.schema.address import decode_address from lbrynet.schema.schema import SECP256k1 +from lbrynet.schema.signature import Signature, NAMED_SECP256K1 +from lbrynet.schema.signer import get_signer from torba.testcase import IntegrationTestCase from lbrynet.schema.claim import ClaimDict from lbrynet.extras.wallet.transaction import Transaction @@ -124,3 +128,45 @@ class BasicTransactionTest(IntegrationTestCase): response = await self.ledger.resolve(0, 10, 'lbry://@bar/foo') self.assertIn('lbry://@bar/foo', response) self.assertIn('claim', response['lbry://@bar/foo']) + + async def test_new_signature_model_from_unserializable_claim(self): + address1, address2 = await self.account.receiving.get_addresses(limit=2, only_usable=True) + sendtxid1 = await self.blockchain.send_to_address(address1, 5) + sendtxid2 = await self.blockchain.send_to_address(address2, 5) + await self.blockchain.generate(1) + await asyncio.wait([ + self.on_transaction_id(sendtxid1), + self.on_transaction_id(sendtxid2) + ]) + + self.assertEqual(d2l(await self.account.get_balance()), '10.0') + + cert, key = generate_certificate() + cert_tx = await Transaction.claim('@bar', cert, l2d('1.0'), address1, [self.account], self.account) + original = ClaimDict.load_dict(example_claim_dict).serialized + altered = original + b'\x00\x01\x02\x30\x50\x80\x99' # pretend this extra trash is from some unknown protobuf + + # manually sign + signer = get_signer(SECP256k1).load_pem(key) + signature = signer.sign( + b'foo', + decode_address(address1), + altered, + binascii.unhexlify(cert_tx.outputs[0].claim_id), + ) + detached_sig = Signature(NAMED_SECP256K1( + signature, + binascii.unhexlify(cert_tx.outputs[0].claim_id), + altered + )) + claim_tx = await Transaction.claim('foo', detached_sig, l2d('1.0'), address1, [self.account], self.account) + + await self.broadcast(cert_tx) + await self.broadcast(claim_tx) + await self.ledger.wait(claim_tx) + await self.blockchain.generate(1) + await self.ledger.wait(claim_tx) + + response = await self.ledger.resolve(0, 10, 'lbry://@bar/foo') + self.assertIn('lbry://@bar/foo', response) + self.assertIn('claim', response['lbry://@bar/foo'])