LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases282 Packages Wiki Activity

wallet server tracks claim signature

Browse source
This commit is contained in:
Lex Berezhny 2019-05-25 23:06:22 -04:00
parent 9d97b90ff4
commit 29bd936181
6 changed files with 427 additions and 275 deletions
tests/unit/wallet/server
test_sqldb.py

99
tests/unit/wallet/server/test_sqldb.py
View file

@ -1,7 +1,7 @@
import unittest
import ecdsa
import hashlib
from binascii import hexlify, unhexlify
from binascii import hexlify
from torba.client.constants import COIN, NULL_HASH32
from lbrynet.schema.claim import Claim
@ -54,22 +54,21 @@ class TestSQLDB(unittest.TestCase):
self._txos[output.ref.hash] = output
return OldWalletServerTransaction(tx), tx.hash
def get_channel(self, title, amount, name='@foo'):
claim = Claim()
claim.channel.title = title
channel = Output.pay_claim_name_pubkey_hash(amount, name, claim, b'abc')
# deterministic private key
private_key = ecdsa.SigningKey.from_string(b'c'*32, curve=ecdsa.SECP256k1, hashfunc=hashlib.sha256)
channel.private_key = private_key.to_pem().decode()
channel.claim.channel.public_key_bytes = private_key.get_verifying_key().to_der()
channel.script.generate()
return self._make_tx(channel)
def get_channel_update(self, channel, amount, key=b'd'):
def _set_channel_key(self, channel, key):
private_key = ecdsa.SigningKey.from_string(key*32, curve=ecdsa.SECP256k1, hashfunc=hashlib.sha256)
channel.private_key = private_key.to_pem().decode()
channel.claim.channel.public_key_bytes = private_key.get_verifying_key().to_der()
channel.script.generate()
def get_channel(self, title, amount, name='@foo', key=b'a'):
claim = Claim()
claim.channel.title = title
channel = Output.pay_claim_name_pubkey_hash(amount, name, claim, b'abc')
self._set_channel_key(channel, key)
return self._make_tx(channel)
def get_channel_update(self, channel, amount, key=b'a'):
self._set_channel_key(channel, key)
return self._make_tx(
Output.pay_update_claim_pubkey_hash(
amount, channel.claim_name, channel.claim_id, channel.claim, b'abc'
@ -313,57 +312,81 @@ class TestSQLDB(unittest.TestCase):
@staticmethod
def _get_x_with_claim_id_prefix(getter, prefix, cached_iteration=None, **kwargs):
iterations = 100
iterations = cached_iteration+1 if cached_iteration else 100
for i in range(cached_iteration or 1, iterations):
stream = getter(f'claim #{i}', COIN, **kwargs)
if stream[0].tx.outputs[0].claim_id.startswith(prefix):
cached_iteration is None and print(f'Found "{prefix}" in {i} iterations.')
return stream
raise ValueError(f'Failed to find "{prefix}" in {iterations} iterations.')
if cached_iteration:
raise ValueError(f'Failed to find "{prefix}" at cached iteration, run with None to find iteration.')
raise ValueError(f'Failed to find "{prefix}" in {iterations} iterations, try different values.')
def get_channel_with_claim_id_prefix(self, prefix, cached_iteration=None):
return self._get_x_with_claim_id_prefix(self.get_channel, prefix, cached_iteration)
def get_channel_with_claim_id_prefix(self, prefix, cached_iteration=None, **kwargs):
return self._get_x_with_claim_id_prefix(self.get_channel, prefix, cached_iteration, **kwargs)
def get_stream_with_claim_id_prefix(self, prefix, cached_iteration=None, **kwargs):
return self._get_x_with_claim_id_prefix(self.get_stream, prefix, cached_iteration, **kwargs)
def test_canonical_name(self):
def test_canonical_url_and_channel_validation(self):
advance = self.advance
tx_chan_a = self.get_channel_with_claim_id_prefix('a', 1)
tx_chan_ab = self.get_channel_with_claim_id_prefix('ab', 72)
tx_chan_a = self.get_channel_with_claim_id_prefix('a', 1, key=b'c')
tx_chan_ab = self.get_channel_with_claim_id_prefix('ab', 72, key=b'c')
txo_chan_a = tx_chan_a[0].tx.outputs[0]
advance(1, [tx_chan_a])
advance(2, [tx_chan_ab])
r_ab, r_a = self.sql._search(order_by=['height'], limit=2)
self.assertEqual("@foo#a", r_a['canonical'])
self.assertEqual("@foo#ab", r_ab['canonical'])
r_ab, r_a = self.sql._search(order_by=['creation_height'], limit=2)
self.assertEqual("@foo#a", r_a['short_url'])
self.assertEqual("@foo#ab", r_ab['short_url'])
self.assertIsNone(r_a['canonical_url'])
self.assertIsNone(r_ab['canonical_url'])
self.assertEqual(0, r_a['claims_in_channel'])
self.assertEqual(0, r_ab['claims_in_channel'])
tx_a = self.get_stream_with_claim_id_prefix('a', 2)
tx_ab = self.get_stream_with_claim_id_prefix('ab', 42)
tx_abc = self.get_stream_with_claim_id_prefix('abc', 65)
advance(3, [tx_a])
advance(4, [tx_ab])
advance(5, [tx_abc])
r_abc, r_ab, r_a = self.sql._search(order_by=['height'], limit=3)
self.assertEqual("foo#a", r_a['canonical'])
self.assertEqual("foo#ab", r_ab['canonical'])
self.assertEqual("foo#abc", r_abc['canonical'])
advance(4, [tx_ab, tx_abc])
r_abc, r_ab, r_a = self.sql._search(order_by=['creation_height', 'tx_position'], limit=3)
self.assertEqual("foo#a", r_a['short_url'])
self.assertEqual("foo#ab", r_ab['short_url'])
self.assertEqual("foo#abc", r_abc['short_url'])
self.assertIsNone(r_a['canonical_url'])
self.assertIsNone(r_ab['canonical_url'])
self.assertIsNone(r_abc['canonical_url'])
tx_a2 = self.get_stream_with_claim_id_prefix('a', 7, channel=txo_chan_a)
tx_ab2 = self.get_stream_with_claim_id_prefix('ab', 23, channel=txo_chan_a)
a2_claim_id = tx_a2[0].tx.outputs[0].claim_id
ab2_claim_id = tx_ab2[0].tx.outputs[0].claim_id
advance(6, [tx_a2])
advance(7, [tx_ab2])
r_ab2, r_a2 = self.sql._search(order_by=['height'], limit=2)
self.assertEqual("@foo#a/foo#a", r_a2['canonical'])
self.assertEqual("@foo#a/foo#ab", r_ab2['canonical'])
r_ab2, r_a2 = self.sql._search(order_by=['creation_height'], limit=2)
self.assertEqual(f"foo#{a2_claim_id[:2]}", r_a2['short_url'])
self.assertEqual(f"foo#{ab2_claim_id[:4]}", r_ab2['short_url'])
self.assertEqual("@foo#a/foo#a", r_a2['canonical_url'])
self.assertEqual("@foo#a/foo#ab", r_ab2['canonical_url'])
self.assertEqual(2, self.sql._search(claim_id=txo_chan_a.claim_id, limit=1)[0]['claims_in_channel'])
advance(8, [self.get_channel_update(txo_chan_a, COIN)])
_, r_ab2, r_a2 = self.sql._search(order_by=['height'], limit=3)
a2_claim_id = hexlify(r_a2['claim_hash'][::-1]).decode()
ab2_claim_id = hexlify(r_ab2['claim_hash'][::-1]).decode()
self.assertEqual(f"foo#{a2_claim_id[:2]}", r_a2['canonical'])
self.assertEqual(f"foo#{ab2_claim_id[:4]}", r_ab2['canonical'])
# invalidate channel signature
advance(8, [self.get_channel_update(txo_chan_a, COIN, key=b'a')])
r_ab2, r_a2 = self.sql._search(order_by=['creation_height'], limit=2)
self.assertEqual(f"foo#{a2_claim_id[:2]}", r_a2['short_url'])
self.assertEqual(f"foo#{ab2_claim_id[:4]}", r_ab2['short_url'])
self.assertIsNone(r_a2['canonical_url'])
self.assertIsNone(r_ab2['canonical_url'])
self.assertEqual(0, self.sql._search(claim_id=txo_chan_a.claim_id, limit=1)[0]['claims_in_channel'])
# re-validate signature (reverts signature to original one)
advance(9, [self.get_channel_update(txo_chan_a, COIN, key=b'c')])
r_ab2, r_a2 = self.sql._search(order_by=['creation_height'], limit=2)
self.assertEqual(f"foo#{a2_claim_id[:2]}", r_a2['short_url'])
self.assertEqual(f"foo#{ab2_claim_id[:4]}", r_ab2['short_url'])
self.assertEqual("@foo#a/foo#a", r_a2['canonical_url'])
self.assertEqual("@foo#a/foo#ab", r_ab2['canonical_url'])
self.assertEqual(2, self.sql._search(claim_id=txo_chan_a.claim_id, limit=1)[0]['claims_in_channel'])
def test_canonical_find_shortest_id(self):
new_hash = 'abcdef0123456789beef'