58 lines
2 KiB
Python
58 lines
2 KiB
Python
from binascii import hexlify
|
|
from twisted.internet import defer
|
|
|
|
from torba.baseaccount import BaseAccount
|
|
|
|
from lbryschema.claim import ClaimDict
|
|
from lbryschema.signer import SECP256k1, get_signer
|
|
|
|
from .transaction import Transaction
|
|
|
|
|
|
def generate_certificate():
|
|
secp256k1_private_key = get_signer(SECP256k1).generate().private_key.to_pem()
|
|
return ClaimDict.generate_certificate(secp256k1_private_key, curve=SECP256k1), secp256k1_private_key
|
|
|
|
|
|
def get_certificate_lookup(tx_or_hash, nout):
|
|
if isinstance(tx_or_hash, Transaction):
|
|
return '{}:{}'.format(tx_or_hash.hex_id.decode(), nout)
|
|
else:
|
|
return '{}:{}'.format(hexlify(tx_or_hash[::-1]).decode(), nout)
|
|
|
|
|
|
class Account(BaseAccount):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(Account, self).__init__(*args, **kwargs)
|
|
self.certificates = {}
|
|
|
|
def add_certificate(self, tx, nout, private_key):
|
|
lookup_key = '{}:{}'.format(tx.hex_id.decode(), nout)
|
|
assert lookup_key not in self.certificates, 'Trying to add a duplicate certificate.'
|
|
self.certificates[lookup_key] = private_key
|
|
|
|
def get_certificate_private_key(self, tx_or_hash, nout):
|
|
return self.certificates.get(get_certificate_lookup(tx_or_hash, nout))
|
|
|
|
@defer.inlineCallbacks
|
|
def maybe_migrate_certificates(self):
|
|
for lookup_key in self.certificates.keys():
|
|
if ':' not in lookup_key:
|
|
claim = self.ledger.
|
|
def get_balance(self, include_claims=False):
|
|
if include_claims:
|
|
return super(Account, self).get_balance()
|
|
else:
|
|
return super(Account, self).get_balance(
|
|
is_claim=0, is_update=0, is_support=0
|
|
)
|
|
|
|
def get_unspent_outputs(self, include_claims=False):
|
|
if include_claims:
|
|
return super(Account, self).get_unspent_outputs()
|
|
else:
|
|
return super(Account, self).get_unspent_outputs(
|
|
is_claim=0, is_update=0, is_support=0
|
|
)
|