# Copyright © 2020 Interplanetary Database Association e.V.,
# BigchainDB and IPDB software contributors.
# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)
# Code is Apache-2.0 and docs are CC-BY-4.0
"""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
import bigchaindb
from bigchaindb import backend, config_utils, fastquery
from bigchaindb.models import Transaction
from bigchaindb.common.exceptions import (SchemaValidationError,
ValidationError,
DoubleSpend)
from bigchaindb.common.transaction_mode_types import (BROADCAST_TX_COMMIT,
BROADCAST_TX_ASYNC,
BROADCAST_TX_SYNC)
from bigchaindb.tendermint_utils import encode_transaction, merkleroot
from bigchaindb import exceptions as core_exceptions
from bigchaindb.validation import BaseValidationRules
logger = logging.getLogger(__name__)
[docs]class BigchainDB(object):
"""Bigchain API
Create, read, sign, write transactions to the database
"""
def __init__(self, connection=None):
"""Initialize the Bigchain instance
A Bigchain instance has several configuration parameters (e.g. host).
If a parameter value is passed as an argument to the Bigchain
__init__ method, then that is the value it will have.
Otherwise, the parameter value will come from an environment variable.
If that environment variable isn't set, then the value
will come from the local configuration file. And if that variable
isn't in the local configuration file, then the parameter will have
its default value (defined in bigchaindb.__init__).
Args:
connection (:class:`~bigchaindb.backend.connection.Connection`):
A connection to the database.
"""
config_utils.autoconfigure()
self.mode_commit = BROADCAST_TX_COMMIT
self.mode_list = (BROADCAST_TX_ASYNC,
BROADCAST_TX_SYNC,
self.mode_commit)
self.tendermint_host = bigchaindb.config['tendermint']['host']
self.tendermint_port = bigchaindb.config['tendermint']['port']
self.endpoint = 'http://{}:{}/'.format(self.tendermint_host, self.tendermint_port)
validationPlugin = bigchaindb.config.get('validation_plugin')
if validationPlugin:
self.validation = config_utils.load_validation_plugin(validationPlugin)
else:
self.validation = BaseValidationRules
self.connection = connection if connection else backend.connect(**bigchaindb.config['database'])
[docs] 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)
[docs] 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)
[docs] 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:`~bigchaindb.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]
)
[docs] 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)
[docs] 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))
[docs] 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)
[docs] 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)
[docs] 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)
[docs] 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
[docs] def store_block(self, block):
"""Create a new block."""
return backend.query.store_block(self.connection, block)
[docs] def get_latest_block(self):
"""Get the block with largest height."""
return backend.query.get_latest_block(self.connection)
[docs] 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
[docs] 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]
[docs] 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
[docs] 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)
[docs] 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)
@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)
[docs] 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)
[docs] 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'))