lbry-sdk/lbrynet/extras/wallet/manager.py
2018-12-04 12:12:08 -05:00

604 lines
25 KiB
Python

import os
import json
import asyncio
import logging
from binascii import unhexlify
from aiorpcx.jsonrpc import CodeMessageError
from datetime import datetime
from typing import Optional
from twisted.internet import defer
from lbrynet.schema.schema import SECP256k1
from torba.client.basemanager import BaseWalletManager
from lbrynet.schema.claim import ClaimDict
from lbrynet.extras.wallet.ledger import MainNetLedger
from lbrynet.extras.wallet.account import BaseAccount, generate_certificate
from lbrynet.extras.wallet.transaction import Transaction
from lbrynet.extras.wallet.database import WalletDatabase
from lbrynet.extras.wallet.dewies import dewies_to_lbc
log = logging.getLogger(__name__)
class ReservedPoints:
def __init__(self, identifier, amount):
self.identifier = identifier
self.amount = amount
class BackwardsCompatibleNetwork:
def __init__(self, manager):
self.manager = manager
def get_local_height(self):
for ledger in self.manager.ledgers.values():
assert isinstance(ledger, MainNetLedger)
return ledger.headers.height
def get_server_height(self):
return self.get_local_height()
class LbryWalletManager(BaseWalletManager):
@property
def ledger(self) -> MainNetLedger:
return self.default_account.ledger
@property
def db(self) -> WalletDatabase:
return self.ledger.db
@property
def wallet(self):
return self
@property
def network(self):
return BackwardsCompatibleNetwork(self)
@property
def use_encryption(self):
return self.default_account.serialize_encrypted
@property
def is_first_run(self):
return True
@property
def is_wallet_unlocked(self):
return not self.default_account.encrypted
def check_locked(self):
return self.default_account.encrypted
def decrypt_account(self, account):
assert account.password is not None, "account is not unlocked"
assert not account.encrypted, "account is not unlocked"
account.serialize_encrypted = False
self.save()
return not account.encrypted and not account.serialize_encrypted
def encrypt_account(self, password, account):
assert not account.encrypted, "account is already encrypted"
account.encrypt(password)
account.serialize_encrypted = True
self.save()
self.unlock_account(password, account)
return account.serialize_encrypted
def unlock_account(self, password, account):
assert account.encrypted, "account is not locked"
account.decrypt(password)
return not account.encrypted
def lock_account(self, account):
assert account.password is not None, "account is already locked"
assert not account.encrypted and account.serialize_encrypted, "account is not encrypted"
account.encrypt(account.password)
return account.encrypted
@staticmethod
def migrate_lbryum_to_torba(path):
if not os.path.exists(path):
return None, None
with open(path, 'r') as f:
unmigrated_json = f.read()
unmigrated = json.loads(unmigrated_json)
# TODO: After several public releases of new torba based wallet, we can delete
# this lbryum->torba conversion code and require that users who still
# have old structured wallets install one of the earlier releases that
# still has the below conversion code.
if 'master_public_keys' not in unmigrated:
return None, None
total = unmigrated.get('addr_history')
receiving_addresses, change_addresses = set(), set()
for _, unmigrated_account in unmigrated.get('accounts', {}).items():
receiving_addresses.update(map(unhexlify, unmigrated_account.get('receiving', [])))
change_addresses.update(map(unhexlify, unmigrated_account.get('change', [])))
log.info("Wallet migrator found %s receiving addresses and %s change addresses. %s in total on history.",
len(receiving_addresses), len(change_addresses), len(total))
migrated_json = json.dumps({
'version': 1,
'name': 'My Wallet',
'accounts': [{
'version': 1,
'name': 'Main Account',
'ledger': 'lbc_mainnet',
'encrypted': unmigrated['use_encryption'],
'seed': unmigrated['seed'],
'seed_version': unmigrated['seed_version'],
'private_key': unmigrated['master_private_keys']['x/'],
'public_key': unmigrated['master_public_keys']['x/'],
'certificates': unmigrated.get('claim_certificates', {}),
'address_generator': {
'name': 'deterministic-chain',
'receiving': {'gap': 20, 'maximum_uses_per_address': 1},
'change': {'gap': 6, 'maximum_uses_per_address': 1}
}
}]
}, indent=4, sort_keys=True)
mode = os.stat(path).st_mode
i = 1
backup_path_template = os.path.join(os.path.dirname(path), "old_lbryum_wallet") + "_%i"
while os.path.isfile(backup_path_template % i):
i += 1
os.rename(path, backup_path_template % i)
temp_path = "{}.tmp.{}".format(path, os.getpid())
with open(temp_path, "w") as f:
f.write(migrated_json)
f.flush()
os.fsync(f.fileno())
os.rename(temp_path, path)
os.chmod(path, mode)
return receiving_addresses, change_addresses
@classmethod
async def from_lbrynet_config(cls, settings, db):
ledger_id = {
'lbrycrd_main': 'lbc_mainnet',
'lbrycrd_testnet': 'lbc_testnet',
'lbrycrd_regtest': 'lbc_regtest'
}[settings['blockchain_name']]
ledger_config = {
'auto_connect': True,
'default_servers': settings['lbryum_servers'],
'data_path': settings['lbryum_wallet_dir'],
'use_keyring': settings['use_keyring'],
#'db': db
}
wallets_directory = os.path.join(settings['lbryum_wallet_dir'], 'wallets')
if not os.path.exists(wallets_directory):
os.mkdir(wallets_directory)
wallet_file_path = os.path.join(wallets_directory, 'default_wallet')
receiving_addresses, change_addresses = cls.migrate_lbryum_to_torba(wallet_file_path)
manager = cls.from_config({
'ledgers': {ledger_id: ledger_config},
'wallets': [wallet_file_path]
})
ledger = manager.get_or_create_ledger(ledger_id)
if manager.default_account is None:
log.info('Wallet at %s is empty, generating a default account.', wallet_file_path)
manager.default_wallet.generate_account(ledger)
manager.default_wallet.save()
if receiving_addresses or change_addresses:
if not os.path.exists(ledger.path):
os.mkdir(ledger.path)
await ledger.db.open()
try:
await manager._migrate_addresses(receiving_addresses, change_addresses)
finally:
await ledger.db.close()
return manager
async def _migrate_addresses(self, receiving_addresses: set, change_addresses: set):
async with self.default_account.receiving.address_generator_lock:
migrated_receiving = set((await self.default_account.receiving._generate_keys(0, len(receiving_addresses))))
async with self.default_account.change.address_generator_lock:
migrated_change = set((await self.default_account.change._generate_keys(0, len(change_addresses))))
receiving_addresses = set(map(self.default_account.ledger.public_key_to_address, receiving_addresses))
change_addresses = set(map(self.default_account.ledger.public_key_to_address, change_addresses))
if not any(change_addresses.difference(migrated_change)):
log.info("Successfully migrated %s change addresses.", len(change_addresses))
else:
log.warning("Failed to migrate %s change addresses!",
len(set(change_addresses).difference(set(migrated_change))))
if not any(receiving_addresses.difference(migrated_receiving)):
log.info("Successfully migrated %s receiving addresses.", len(receiving_addresses))
else:
log.warning("Failed to migrate %s receiving addresses!",
len(set(receiving_addresses).difference(set(migrated_receiving))))
def get_best_blockhash(self):
return self.ledger.headers.hash(self.ledger.headers.height).decode()
def get_unused_address(self):
return self.default_account.receiving.get_or_create_usable_address()
def get_new_address(self):
return self.get_unused_address()
def reserve_points(self, address, amount: int):
# TODO: check if we have enough to cover amount
return ReservedPoints(address, amount)
async def send_amount_to_address(self, amount: int, destination_address: bytes, account=None):
account = account or self.default_account
tx = await Transaction.pay(amount, destination_address, [account], account)
await account.ledger.broadcast(tx)
return tx
async def send_claim_to_address(self, claim_id: str, destination_address: str, amount: Optional[int],
account=None):
account = account or self.default_account
claims = await account.get_claims(
claim_name_type__any={'is_claim': 1, 'is_update': 1}, # exclude is_supports
claim_id=claim_id
)
if not claims:
raise NameError(f"Claim not found: {claim_id}")
if not amount:
amount = claims[0].get_estimator(self.ledger).effective_amount
tx = await Transaction.update(
claims[0], ClaimDict.deserialize(claims[0].script.values['claim']), amount,
destination_address.encode(), [account], account
)
await self.ledger.broadcast(tx)
return tx
def send_points_to_address(self, reserved: ReservedPoints, amount: int, account=None):
destination_address: bytes = reserved.identifier.encode('latin1')
return self.send_amount_to_address(amount, destination_address, account)
def get_wallet_info_query_handler_factory(self):
return LBRYcrdAddressQueryHandlerFactory(self)
def get_info_exchanger(self):
return LBRYcrdAddressRequester(self)
async def resolve(self, *uris, **kwargs):
page = kwargs.get('page', 0)
page_size = kwargs.get('page_size', 10)
check_cache = kwargs.get('check_cache', False) # TODO: put caching back (was force_refresh parameter)
ledger: MainNetLedger = self.default_account.ledger
results = await ledger.resolve(page, page_size, *uris)
if 'error' not in results:
await self.old_db.save_claims_for_resolve(
(value for value in results.values() if 'error' not in value)
).asFuture(asyncio.get_event_loop())
return results
async def get_claims_for_name(self, name: str):
response = await self.ledger.network.get_claims_for_name(name)
if 'claims' in response:
to_resolve = [(claim['name'] + '#' + claim['claim_id']) for claim in response['claims']]
response['claims'] = [resolution['claim'] for resolution in (await self.resolve(*to_resolve)).values()]
return response
async def address_is_mine(self, unknown_address, account):
match = await self.ledger.db.get_address(address=unknown_address, account=account)
if match is not None:
return True
return False
async def get_transaction(self, txid):
tx = await self.db.get_transaction(txid=txid)
if not tx:
try:
_raw = await self.ledger.network.get_transaction(txid)
except CodeMessageError as e:
return {'success': False, 'code': e.code, 'message': e.message}
# this is a workaround for the current protocol. Should be fixed when lbryum support is over and we
# are able to use the modern get_transaction call, which accepts verbose to show height and other fields
height = await self.ledger.network.get_transaction_height(txid)
tx = self.ledger.transaction_class(unhexlify(_raw))
if tx and height > 0:
await self.ledger.maybe_verify_transaction(tx, height + 1) # off by one from server side, yes...
return tx
@staticmethod
async def get_history(account: BaseAccount, **constraints):
headers = account.ledger.headers
txs = await account.get_transactions(**constraints)
history = []
for tx in txs:
ts = headers[tx.height]['timestamp'] if tx.height > 0 else None
item = {
'txid': tx.id,
'timestamp': ts,
'date': datetime.fromtimestamp(ts).isoformat(' ')[:-3] if tx.height > 0 else None,
'confirmations': headers.height - tx.height if tx.height > 0 else 0,
'claim_info': [],
'update_info': [],
'support_info': [],
'abandon_info': []
}
is_my_inputs = all([txi.is_my_account for txi in tx.inputs])
if is_my_inputs:
# fees only matter if we are the ones paying them
item['value'] = dewies_to_lbc(tx.net_account_balance+tx.fee)
item['fee'] = dewies_to_lbc(-tx.fee)
else:
# someone else paid the fees
item['value'] = dewies_to_lbc(tx.net_account_balance)
item['fee'] = '0.0'
for txo in tx.my_claim_outputs:
item['claim_info'].append({
'address': txo.get_address(account.ledger),
'balance_delta': dewies_to_lbc(-txo.amount),
'amount': dewies_to_lbc(txo.amount),
'claim_id': txo.claim_id,
'claim_name': txo.claim_name,
'nout': txo.position
})
for txo in tx.my_update_outputs:
if is_my_inputs: # updating my own claim
previous = None
for txi in tx.inputs:
if txi.txo_ref.txo is not None:
other_txo = txi.txo_ref.txo
if other_txo.is_claim and other_txo.claim_id == txo.claim_id:
previous = other_txo
break
assert previous is not None,\
"Invalid claim update state, expected to find previous claim in input."
item['update_info'].append({
'address': txo.get_address(account.ledger),
'balance_delta': dewies_to_lbc(previous.amount-txo.amount),
'amount': dewies_to_lbc(txo.amount),
'claim_id': txo.claim_id,
'claim_name': txo.claim_name,
'nout': txo.position
})
else: # someone sent us their claim
item['update_info'].append({
'address': txo.get_address(account.ledger),
'balance_delta': dewies_to_lbc(0),
'amount': dewies_to_lbc(txo.amount),
'claim_id': txo.claim_id,
'claim_name': txo.claim_name,
'nout': txo.position
})
for txo in tx.my_support_outputs:
item['support_info'].append({
'address': txo.get_address(account.ledger),
'balance_delta': dewies_to_lbc(txo.amount if not is_my_inputs else -txo.amount),
'amount': dewies_to_lbc(txo.amount),
'claim_id': txo.claim_id,
'claim_name': txo.claim_name,
'is_tip': not is_my_inputs,
'nout': txo.position
})
for txo in tx.other_support_outputs:
item['support_info'].append({
'address': txo.get_address(account.ledger),
'balance_delta': dewies_to_lbc(-txo.amount),
'amount': dewies_to_lbc(txo.amount),
'claim_id': txo.claim_id,
'claim_name': txo.claim_name,
'is_tip': is_my_inputs,
'nout': txo.position
})
for txo in tx.my_abandon_outputs:
item['abandon_info'].append({
'address': txo.get_address(account.ledger),
'balance_delta': dewies_to_lbc(txo.amount),
'amount': dewies_to_lbc(txo.amount),
'claim_id': txo.claim_id,
'claim_name': txo.claim_name,
'nout': txo.position
})
history.append(item)
return history
@staticmethod
def get_utxos(account: BaseAccount):
return account.get_utxos()
async def claim_name(self, account, name, amount, claim_dict, certificate=None, claim_address=None):
claim = ClaimDict.load_dict(claim_dict)
if not claim_address:
claim_address = await account.receiving.get_or_create_usable_address()
if certificate:
claim = claim.sign(
certificate.private_key, claim_address, certificate.claim_id, curve=SECP256k1
)
existing_claims = await account.get_claims(
claim_name_type__any={'is_claim': 1, 'is_update': 1}, # exclude is_supports
claim_name=name
)
if len(existing_claims) == 0:
tx = await Transaction.claim(
name, claim, amount, claim_address, [account], account
)
elif len(existing_claims) == 1:
tx = await Transaction.update(
existing_claims[0], claim, amount, claim_address, [account], account
)
else:
raise NameError(f"More than one other claim exists with the name '{name}'.")
await account.ledger.broadcast(tx)
await self.old_db.save_claims([self._old_get_temp_claim_info(
tx, tx.outputs[0], claim_address, claim_dict, name, amount
)]).asFuture(asyncio.get_event_loop())
# TODO: release reserved tx outputs in case anything fails by this point
return tx
async def support_claim(self, claim_name, claim_id, amount, account):
holding_address = await account.receiving.get_or_create_usable_address()
tx = await Transaction.support(claim_name, claim_id, amount, holding_address, [account], account)
await account.ledger.broadcast(tx)
return tx
async def tip_claim(self, amount, claim_id, account):
claim_to_tip = await self.get_claim_by_claim_id(claim_id)
tx = await Transaction.support(
claim_to_tip['name'], claim_id, amount, claim_to_tip['address'], [account], account
)
await account.ledger.broadcast(tx)
return tx
async def abandon_claim(self, claim_id, txid, nout, account):
claim = await account.get_claim(claim_id=claim_id, txid=txid, nout=nout)
if not claim:
raise Exception('No claim found for the specified claim_id or txid:nout')
tx = await Transaction.abandon(claim, [account], account)
await account.ledger.broadcast(tx)
# TODO: release reserved tx outputs in case anything fails by this point
return tx
async def claim_new_channel(self, channel_name, amount, account):
address = await account.receiving.get_or_create_usable_address()
cert, key = generate_certificate()
tx = await Transaction.claim(channel_name, cert, amount, address, [account], account)
await account.ledger.broadcast(tx)
account.add_certificate_private_key(tx.outputs[0].ref, key.decode())
# TODO: release reserved tx outputs in case anything fails by this point
return tx
def _old_get_temp_claim_info(self, tx, txo, address, claim_dict, name, bid):
return {
"claim_id": txo.claim_id,
"name": name,
"amount": bid,
"address": address,
"txid": tx.id,
"nout": txo.position,
"value": claim_dict,
"height": -1,
"claim_sequence": -1,
}
def get_certificates(self, private_key_accounts, exclude_without_key=True, **constraints):
return self.db.get_certificates(
private_key_accounts=private_key_accounts,
exclude_without_key=exclude_without_key,
**constraints
)
def update_peer_address(self, peer, address):
pass # TODO: Data payments is disabled
def get_unused_address_for_peer(self, peer):
# TODO: Data payments is disabled
return self.get_unused_address()
def add_expected_payment(self, peer, amount):
pass # TODO: Data payments is disabled
def send_points(self, reserved_points, amount):
defer.succeed(True) # TODO: Data payments is disabled
def cancel_point_reservation(self, reserved_points):
pass # fixme: disabled for now.
def save(self):
for wallet in self.wallets:
wallet.save()
def get_block(self, block_hash=None, height=None):
if height is None:
height = self.ledger.headers.height
if block_hash is None:
block_hash = self.ledger.headers.hash(height).decode()
return self.ledger.network.get_block(block_hash)
def get_claim_by_claim_id(self, claim_id):
return self.ledger.get_claim_by_claim_id(claim_id)
def get_claim_by_outpoint(self, txid, nout):
return self.ledger.get_claim_by_outpoint(txid, nout)
class ClientRequest:
def __init__(self, request_dict, response_identifier=None):
self.request_dict = request_dict
self.response_identifier = response_identifier
class LBRYcrdAddressRequester:
def __init__(self, wallet):
self.wallet = wallet
self._protocols = []
def send_next_request(self, peer, protocol):
if not protocol in self._protocols:
r = ClientRequest({'lbrycrd_address': True}, 'lbrycrd_address')
d = protocol.add_request(r)
d.addCallback(self._handle_address_response, peer, r, protocol)
d.addErrback(self._request_failed, peer)
self._protocols.append(protocol)
return defer.succeed(True)
else:
return defer.succeed(False)
def _handle_address_response(self, response_dict, peer, request, protocol):
if request.response_identifier not in response_dict:
raise ValueError(
f"Expected {request.response_identifier} in response but did not get it")
assert protocol in self._protocols, "Responding protocol is not in our list of protocols"
address = response_dict[request.response_identifier]
self.wallet.update_peer_address(peer, address)
def _request_failed(self, error, peer):
raise Exception(
"A peer failed to send a valid public key response. Error: {}, peer: {}".format(
error.getErrorMessage(), str(peer)
)
)
class LBRYcrdAddressQueryHandlerFactory:
def __init__(self, wallet):
self.wallet = wallet
def build_query_handler(self):
q_h = LBRYcrdAddressQueryHandler(self.wallet)
return q_h
def get_primary_query_identifier(self):
return 'lbrycrd_address'
def get_description(self):
return "LBRYcrd Address - an address for receiving payments via LBRYcrd"
class LBRYcrdAddressQueryHandler:
def __init__(self, wallet):
self.wallet = wallet
self.query_identifiers = ['lbrycrd_address']
self.address = None
self.peer = None
def register_with_request_handler(self, request_handler, peer):
self.peer = peer
request_handler.register_query_handler(self, self.query_identifiers)
@defer.inlineCallbacks
def handle_queries(self, queries):
if self.query_identifiers[0] in queries:
future = self.wallet.get_unused_address_for_peer(self.peer)
address = yield defer.Deferred.fromFuture(asyncio.ensure_future(future))
self.address = address
fields = {'lbrycrd_address': address}
return fields
if self.address is None:
raise Exception("Expected a request for an address, but did not receive one")
else:
return {}