# # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. # # """Module containing main contact points with Tendermint and MongoDB. """ import logging from collections import namedtuple from uuid import uuid4 import rapidjson try: from hashlib import sha3_256 except ImportError: # NOTE: needed for Python < 3.6 from sha3 import sha3_256 import requests from resdb_validator.models import Transaction from resdb_validator.exceptions import (SchemaValidationError, ValidationError, DoubleSpend) logger = logging.getLogger(__name__) class ResDB(object): """Bigchain API Create, read, sign, write transactions to the database """ backend=None def __init__(self, connection=None): """Initialize the ResDB instance Database connections and other configuration settings can be defined here. As of now the validator does not directly interact with any database. """ def post_transaction(self, transaction, mode): """Submit a valid transaction to the mempool.""" if not mode or mode not in self.mode_list: raise ValidationError('Mode must be one of the following {}.' .format(', '.join(self.mode_list))) tx_dict = transaction.tx_dict if transaction.tx_dict else transaction.to_dict() payload = { 'method': mode, 'jsonrpc': '2.0', 'params': [encode_transaction(tx_dict)], 'id': str(uuid4()) } # TODO: handle connection errors! return requests.post(self.endpoint, json=payload) def write_transaction(self, transaction, mode): # This method offers backward compatibility with the Web API. """Submit a valid transaction to the mempool.""" response = self.post_transaction(transaction, mode) return self._process_post_response(response.json(), mode) def _process_post_response(self, response, mode): logger.debug(response) error = response.get('error') if error: status_code = 500 message = error.get('message', 'Internal Error') data = error.get('data', '') if 'Tx already exists in cache' in data: status_code = 400 return (status_code, message + ' - ' + data) result = response['result'] if mode == self.mode_commit: check_tx_code = result.get('check_tx', {}).get('code', 0) deliver_tx_code = result.get('deliver_tx', {}).get('code', 0) error_code = check_tx_code or deliver_tx_code else: error_code = result.get('code', 0) if error_code: return (500, 'Transaction validation failed') return (202, '') def store_bulk_transactions(self, transactions): txns = [] assets = [] txn_metadatas = [] for t in transactions: transaction = t.tx_dict if t.tx_dict else rapidjson.loads(rapidjson.dumps(t.to_dict())) if transaction['operation'] == t.CREATE: asset = transaction.pop('asset') asset['id'] = transaction['id'] assets.append(asset) metadata = transaction.pop('metadata') txn_metadatas.append({'id': transaction['id'], 'metadata': metadata}) txns.append(transaction) backend.query.store_metadatas(self.connection, txn_metadatas) if assets: backend.query.store_assets(self.connection, assets) return backend.query.store_transactions(self.connection, txns) def delete_transactions(self, txs): return backend.query.delete_transactions(self.connection, txs) def update_utxoset(self, transaction): """Update the UTXO set given ``transaction``. That is, remove the outputs that the given ``transaction`` spends, and add the outputs that the given ``transaction`` creates. Args: transaction (:obj:`~resdb_validator.models.Transaction`): A new transaction incoming into the system for which the UTXO set needs to be updated. """ spent_outputs = [ spent_output for spent_output in transaction.spent_outputs ] if spent_outputs: self.delete_unspent_outputs(*spent_outputs) self.store_unspent_outputs( *[utxo._asdict() for utxo in transaction.unspent_outputs] ) def store_unspent_outputs(self, *unspent_outputs): """Store the given ``unspent_outputs`` (utxos). Args: *unspent_outputs (:obj:`tuple` of :obj:`dict`): Variable length tuple or list of unspent outputs. """ if unspent_outputs: return backend.query.store_unspent_outputs( self.connection, *unspent_outputs) def get_utxoset_merkle_root(self): """Returns the merkle root of the utxoset. This implies that the utxoset is first put into a merkle tree. For now, the merkle tree and its root will be computed each time. This obviously is not efficient and a better approach that limits the repetition of the same computation when unnecesary should be sought. For instance, future optimizations could simply re-compute the branches of the tree that were affected by a change. The transaction hash (id) and output index should be sufficient to uniquely identify a utxo, and consequently only that information from a utxo record is needed to compute the merkle root. Hence, each node of the merkle tree should contain the tuple (txid, output_index). .. important:: The leaves of the tree will need to be sorted in some kind of lexicographical order. Returns: str: Merkle root in hexadecimal form. """ utxoset = backend.query.get_unspent_outputs(self.connection) # TODO Once ready, use the already pre-computed utxo_hash field. # See common/transactions.py for details. hashes = [ sha3_256( '{}{}'.format(utxo['transaction_id'], utxo['output_index']).encode() ).digest() for utxo in utxoset ] # TODO Notice the sorted call! return merkleroot(sorted(hashes)) def get_unspent_outputs(self): """Get the utxoset. Returns: generator of unspent_outputs. """ cursor = backend.query.get_unspent_outputs(self.connection) return (record for record in cursor) def delete_unspent_outputs(self, *unspent_outputs): """Deletes the given ``unspent_outputs`` (utxos). Args: *unspent_outputs (:obj:`tuple` of :obj:`dict`): Variable length tuple or list of unspent outputs. """ if unspent_outputs: return backend.query.delete_unspent_outputs( self.connection, *unspent_outputs) def is_committed(self, transaction_id): transaction = backend.query.get_transaction(self.connection, transaction_id) return bool(transaction) def get_transaction(self, transaction_id): transaction = backend.query.get_transaction(self.connection, transaction_id) if transaction: asset = backend.query.get_asset(self.connection, transaction_id) metadata = backend.query.get_metadata(self.connection, [transaction_id]) if asset: transaction['asset'] = asset if 'metadata' not in transaction: metadata = metadata[0] if metadata else None if metadata: metadata = metadata.get('metadata') transaction.update({'metadata': metadata}) transaction = Transaction.from_dict(transaction) return transaction def get_transactions(self, txn_ids): return backend.query.get_transactions(self.connection, txn_ids) def get_transactions_filtered(self, asset_id, operation=None, last_tx=None): """Get a list of transactions filtered on some criteria """ txids = backend.query.get_txids_filtered(self.connection, asset_id, operation, last_tx) for txid in txids: yield self.get_transaction(txid) def get_outputs_filtered(self, owner, spent=None): """Get a list of output links filtered on some criteria Args: owner (str): base58 encoded public_key. spent (bool): If ``True`` return only the spent outputs. If ``False`` return only unspent outputs. If spent is not specified (``None``) return all outputs. Returns: :obj:`list` of TransactionLink: list of ``txid`` s and ``output`` s pointing to another transaction's condition """ outputs = self.fastquery.get_outputs_by_public_key(owner) if spent is None: return outputs elif spent is True: return self.fastquery.filter_unspent_outputs(outputs) elif spent is False: return self.fastquery.filter_spent_outputs(outputs) def get_spent(self, txid, output, current_transactions=[]): transactions = backend.query.get_spent(self.connection, txid, output) transactions = list(transactions) if transactions else [] if len(transactions) > 1: raise core_exceptions.CriticalDoubleSpend( '`{}` was spent more than once. There is a problem' ' with the chain'.format(txid)) current_spent_transactions = [] for ctxn in current_transactions: for ctxn_input in ctxn.inputs: if ctxn_input.fulfills and\ ctxn_input.fulfills.txid == txid and\ ctxn_input.fulfills.output == output: current_spent_transactions.append(ctxn) transaction = None if len(transactions) + len(current_spent_transactions) > 1: raise DoubleSpend('tx "{}" spends inputs twice'.format(txid)) elif transactions: transaction = Transaction.from_db(self, transactions[0]) elif current_spent_transactions: transaction = current_spent_transactions[0] return transaction def store_block(self, block): """Create a new block.""" return backend.query.store_block(self.connection, block) def get_latest_block(self): """Get the block with largest height.""" return backend.query.get_latest_block(self.connection) def get_block(self, block_id): """Get the block with the specified `block_id`. Returns the block corresponding to `block_id` or None if no match is found. Args: block_id (int): block id of the block to get. """ block = backend.query.get_block(self.connection, block_id) latest_block = self.get_latest_block() latest_block_height = latest_block['height'] if latest_block else 0 if not block and block_id > latest_block_height: return result = {'height': block_id, 'transactions': []} if block: transactions = backend.query.get_transactions(self.connection, block['transactions']) result['transactions'] = [t.to_dict() for t in Transaction.from_db(self, transactions)] return result def get_block_containing_tx(self, txid): """Retrieve the list of blocks (block ids) containing a transaction with transaction id `txid` Args: txid (str): transaction id of the transaction to query Returns: Block id list (list(int)) """ blocks = list(backend.query.get_block_with_transaction(self.connection, txid)) if len(blocks) > 1: logger.critical('Transaction id %s exists in multiple blocks', txid) return [block['height'] for block in blocks] def validate_transaction(self, tx, current_transactions=[]): """Validate a transaction against the current status of the database.""" transaction = tx # CLEANUP: The conditional below checks for transaction in dict format. # It would be better to only have a single format for the transaction # throught the code base. if isinstance(transaction, dict): try: transaction = Transaction.from_dict(tx) except SchemaValidationError as e: logger.warning('Invalid transaction schema: %s', e.__cause__.message) return False except ValidationError as e: logger.warning('Invalid transaction (%s): %s', type(e).__name__, e) return False return transaction.validate(self, current_transactions) def is_valid_transaction(self, tx, current_transactions=[]): # NOTE: the function returns the Transaction object in case # the transaction is valid try: return self.validate_transaction(tx, current_transactions) except ValidationError as e: logger.warning('Invalid transaction (%s): %s', type(e).__name__, e) return False def text_search(self, search, *, limit=0, table='assets'): """Return an iterator of assets that match the text search Args: search (str): Text search string to query the text index limit (int, optional): Limit the number of returned documents. Returns: iter: An iterator of assets that match the text search. """ return backend.query.text_search(self.connection, search, limit=limit, table=table) def get_assets(self, asset_ids): """Return a list of assets that match the asset_ids Args: asset_ids (:obj:`list` of :obj:`str`): A list of asset_ids to retrieve from the database. Returns: list: The list of assets returned from the database. """ return backend.query.get_assets(self.connection, asset_ids) def get_metadata(self, txn_ids): """Return a list of metadata that match the transaction ids (txn_ids) Args: txn_ids (:obj:`list` of :obj:`str`): A list of txn_ids to retrieve from the database. Returns: list: The list of metadata returned from the database. """ return backend.query.get_metadata(self.connection, txn_ids) @property def fastquery(self): return fastquery.FastQuery(self.connection) def get_validator_change(self, height=None): return backend.query.get_validator_set(self.connection, height) def get_validators(self, height=None): result = self.get_validator_change(height) return [] if result is None else result['validators'] def get_election(self, election_id): return backend.query.get_election(self.connection, election_id) def get_pre_commit_state(self): return backend.query.get_pre_commit_state(self.connection) def store_pre_commit_state(self, state): return backend.query.store_pre_commit_state(self.connection, state) def store_validator_set(self, height, validators): """Store validator set at a given `height`. NOTE: If the validator set already exists at that `height` then an exception will be raised. """ return backend.query.store_validator_set(self.connection, {'height': height, 'validators': validators}) def delete_validator_set(self, height): return backend.query.delete_validator_set(self.connection, height) def store_abci_chain(self, height, chain_id, is_synced=True): return backend.query.store_abci_chain(self.connection, height, chain_id, is_synced) def delete_abci_chain(self, height): return backend.query.delete_abci_chain(self.connection, height) def get_latest_abci_chain(self): return backend.query.get_latest_abci_chain(self.connection) def migrate_abci_chain(self): """Generate and record a new ABCI chain ID. New blocks are not accepted until we receive an InitChain ABCI request with the matching chain ID and validator set. Chain ID is generated based on the current chain and height. `chain-X` => `chain-X-migrated-at-height-5`. `chain-X-migrated-at-height-5` => `chain-X-migrated-at-height-21`. If there is no known chain (we are at genesis), the function returns. """ latest_chain = self.get_latest_abci_chain() if latest_chain is None: return block = self.get_latest_block() suffix = '-migrated-at-height-' chain_id = latest_chain['chain_id'] block_height_str = str(block['height']) new_chain_id = chain_id.split(suffix)[0] + suffix + block_height_str self.store_abci_chain(block['height'] + 1, new_chain_id, False) def store_election(self, election_id, height, is_concluded): return backend.query.store_election(self.connection, election_id, height, is_concluded) def store_elections(self, elections): return backend.query.store_elections(self.connection, elections) def delete_elections(self, height): return backend.query.delete_elections(self.connection, height) Block = namedtuple('Block', ('app_hash', 'height', 'transactions'))