Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_validate_tx_threshold_duplicated_pk(user_pub, user_priv,
asset_definition):
from cryptoconditions import Ed25519Sha256, ThresholdSha256
from bigchaindb.common.transaction import Input, Output, Transaction
threshold = ThresholdSha256(threshold=2)
threshold.add_subfulfillment(
Ed25519Sha256(public_key=b58decode(user_pub)))
threshold.add_subfulfillment(
Ed25519Sha256(public_key=b58decode(user_pub)))
threshold_input = Input(threshold, [user_pub, user_pub])
threshold_output = Output(threshold, [user_pub, user_pub])
tx = Transaction(Transaction.CREATE, asset_definition,
[threshold_input], [threshold_output])
tx_dict = tx.to_dict()
tx_dict['inputs'][0]['fulfillment'] = None
serialized_tx = json.dumps(tx_dict, sort_keys=True,
separators=(',', ':'), ensure_ascii=True)
message = sha3_256(serialized_tx.encode()).digest()
expected = deepcopy(threshold_input)
expected.fulfillment.subconditions[0]['body'].sign(
message, b58decode(user_priv))
def test_generate_outputs_flat_ownage(user_pub, user2_pub, user3_pub):
from bigchaindb.common.transaction import Output
from cryptoconditions import Ed25519Sha256, ThresholdSha256
expected_simple1 = Ed25519Sha256(public_key=b58decode(user_pub))
expected_simple2 = Ed25519Sha256(public_key=b58decode(user2_pub))
expected_simple3 = Ed25519Sha256(public_key=b58decode(user3_pub))
expected = ThresholdSha256(threshold=3)
expected.add_subfulfillment(expected_simple1)
expected.add_subfulfillment(expected_simple2)
expected.add_subfulfillment(expected_simple3)
cond = Output.generate([user_pub, user2_pub, expected_simple3], 1)
assert cond.fulfillment.to_dict() == expected.to_dict()
def test_generate_outputs_split_half_single_owner(user_pub,
user2_pub, user3_pub):
from bigchaindb.common.transaction import Output
from cryptoconditions import Ed25519Sha256, ThresholdSha256
expected_simple1 = Ed25519Sha256(public_key=b58decode(user_pub))
expected_simple2 = Ed25519Sha256(public_key=b58decode(user2_pub))
expected_simple3 = Ed25519Sha256(public_key=b58decode(user3_pub))
expected = ThresholdSha256(threshold=2)
expected_threshold = ThresholdSha256(threshold=2)
expected_threshold.add_subfulfillment(expected_simple2)
expected_threshold.add_subfulfillment(expected_simple3)
expected.add_subfulfillment(expected_threshold)
expected.add_subfulfillment(expected_simple1)
cond = Output.generate([[expected_simple2, user3_pub], user_pub], 1)
assert cond.fulfillment.to_dict() == expected.to_dict()
def test_generate_output_split_half_recursive(user_pub, user2_pub, user3_pub):
from bigchaindb.common.transaction import Output
from cryptoconditions import Ed25519Sha256, ThresholdSha256
expected_simple1 = Ed25519Sha256(public_key=b58decode(user_pub))
expected_simple2 = Ed25519Sha256(public_key=b58decode(user2_pub))
expected_simple3 = Ed25519Sha256(public_key=b58decode(user3_pub))
expected = ThresholdSha256(threshold=2)
expected.add_subfulfillment(expected_simple1)
expected_threshold = ThresholdSha256(threshold=2)
expected_threshold.add_subfulfillment(expected_simple2)
expected_threshold.add_subfulfillment(expected_simple3)
expected.add_subfulfillment(expected_threshold)
cond = Output.generate([user_pub, [user2_pub, expected_simple3]], 1)
assert cond.fulfillment.to_dict() == expected.to_dict()
def test_fulfill(self, driver, alice_keypair, unsigned_transaction):
signed_transaction = driver.transactions.fulfill(
unsigned_transaction, private_keys=alice_keypair.sk)
unsigned_transaction['inputs'][0]['fulfillment'] = None
message = json.dumps(
unsigned_transaction,
sort_keys=True,
separators=(',', ':'),
ensure_ascii=False,
)
message = sha3_256(message.encode())
ed25519 = Ed25519Sha256(public_key=base58.b58decode(alice_keypair.vk))
ed25519.sign(message.digest(), base58.b58decode(alice_keypair.sk))
fulfillment_uri = ed25519.serialize_uri()
assert signed_transaction['inputs'][0]['fulfillment'] == fulfillment_uri # noqa
assert len(unspent_outputs) == 3
assert all(utxo.transaction_id == tx.id for utxo in unspent_outputs)
assert all(utxo.asset_id == tx.id for utxo in unspent_outputs)
assert all(
utxo.output_index == i for i, utxo in enumerate(unspent_outputs))
unspent_output_0 = unspent_outputs[0]
assert unspent_output_0.amount == 1
assert unspent_output_0.condition_uri == Ed25519Sha256(
public_key=b58decode(alice.public_key)).condition_uri
unspent_output_1 = unspent_outputs[1]
assert unspent_output_1.amount == 2
assert unspent_output_1.condition_uri == Ed25519Sha256(
public_key=b58decode(bob.public_key)).condition_uri
unspent_output_2 = unspent_outputs[2]
assert unspent_output_2.amount == 3
assert unspent_output_2.condition_uri == Ed25519Sha256(
public_key=b58decode(carol.public_key)).condition_uri
def test_generate_outputs_flat_ownage(user_pub, user2_pub, user3_pub):
from bigchaindb.common.transaction import Output
from cryptoconditions import Ed25519Sha256, ThresholdSha256
expected_simple1 = Ed25519Sha256(public_key=b58decode(user_pub))
expected_simple2 = Ed25519Sha256(public_key=b58decode(user2_pub))
expected_simple3 = Ed25519Sha256(public_key=b58decode(user3_pub))
expected = ThresholdSha256(threshold=3)
expected.add_subfulfillment(expected_simple1)
expected.add_subfulfillment(expected_simple2)
expected.add_subfulfillment(expected_simple3)
cond = Output.generate([user_pub, user2_pub, expected_simple3], 1)
assert cond.fulfillment.to_dict() == expected.to_dict()
def test_validate_tx_threshold_duplicated_pk(user_pub, user_priv,
asset_definition):
from cryptoconditions import Ed25519Sha256, ThresholdSha256
from bigchaindb.common.transaction import Input, Output, Transaction
threshold = ThresholdSha256(threshold=2)
threshold.add_subfulfillment(
Ed25519Sha256(public_key=b58decode(user_pub)))
threshold.add_subfulfillment(
Ed25519Sha256(public_key=b58decode(user_pub)))
threshold_input = Input(threshold, [user_pub, user_pub])
threshold_output = Output(threshold, [user_pub, user_pub])
tx = Transaction(Transaction.CREATE, asset_definition,
[threshold_input], [threshold_output])
tx_dict = tx.to_dict()
tx_dict['inputs'][0]['fulfillment'] = None
serialized_tx = json.dumps(tx_dict, sort_keys=True,
separators=(',', ':'), ensure_ascii=True)
message = sha3_256(serialized_tx.encode()).digest()
expected = deepcopy(threshold_input)
"""
threshold = len(public_keys)
if not isinstance(amount, int):
raise TypeError('`amount` must be a int')
if amount < 1:
raise AmountError('`amount` needs to be greater than zero')
if not isinstance(public_keys, list):
raise TypeError('`public_keys` must be an instance of list')
if len(public_keys) == 0:
raise ValueError('`public_keys` needs to contain at least one'
'owner')
elif len(public_keys) == 1 and not isinstance(public_keys[0], list):
if isinstance(public_keys[0], Fulfillment):
ffill = public_keys[0]
else:
ffill = Ed25519Sha256(
public_key=base58.b58decode(public_keys[0]))
return cls(ffill, public_keys, amount=amount)
else:
initial_cond = ThresholdSha256(threshold=threshold)
threshold_cond = reduce(cls._gen_condition, public_keys,
initial_cond)
return cls(threshold_cond, public_keys, amount=amount)
raise ValueError('Sublist cannot contain single owner')
else:
try:
new_public_keys = new_public_keys.pop()
except AttributeError:
pass
# NOTE: Instead of submitting base58 encoded addresses, a user
# of this class can also submit fully instantiated
# Cryptoconditions. In the case of casting
# `new_public_keys` to a Ed25519Fulfillment with the
# result of a `TypeError`, we're assuming that
# `new_public_keys` is a Cryptocondition then.
if isinstance(new_public_keys, Fulfillment):
ffill = new_public_keys
else:
ffill = Ed25519Sha256(
public_key=base58.b58decode(new_public_keys))
initial.add_subfulfillment(ffill)
return initial