Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Varint implementation #7

Merged
merged 3 commits into from
Aug 9, 2022
Merged

Varint implementation #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
aaed31c
implement varint
herissondev Jul 31, 2022
d6f525c
test varint and reformat
herissondev Jul 31, 2022
17206e6
reformat
herissondev Jul 31, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
168 changes: 128 additions & 40 deletions src/archethic/transaction_builder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,11 @@
"token": 251,
# Network based transaction types
"code_proposal": 7,
"code_approval": 8
"code_approval": 8,
}


class TransactionBuilder:

def __init__(self, tx_type: str) -> None:
"""
Create a new instance of the transaction builder by specifying firstly the type of transaction :param
Expand All @@ -29,8 +28,10 @@ def __init__(self, tx_type: str) -> None:
self.address: bytes = bytes()
self.previous_public_key: bytes = bytes()
self.previous_signature: bytes = bytes()
assert tx_type in TX_TYPES, "Invalid transaction type. \n Transaction type must be 'transfer', 'hosting', " \
"'keychain_access', 'keychain', 'token', 'code_proposal', 'code_approval' "
assert tx_type in TX_TYPES, (
"Invalid transaction type. \n Transaction type must be 'transfer', 'hosting', "
"'keychain_access', 'keychain', 'token', 'code_proposal', 'code_approval' "
)

self.version = VERSION
self.tx_type = tx_type
Expand Down Expand Up @@ -70,7 +71,9 @@ def set_content(self, content: Union[str, bytes]) -> None:
raise TypeError("Content must be string or bytes")
return

def add_ownership(self, secret_key: Union[str, bytes], authorized_keys: list) -> None:
def add_ownership(
self, secret_key: Union[str, bytes], authorized_keys: list
) -> None:
"""
Add an ownership with a secret and its authorized public keys to the transaction
:param secret_key: The secret key of the ownership (str or bytes)
Expand Down Expand Up @@ -119,7 +122,9 @@ def add_ownership(self, secret_key: Union[str, bytes], authorized_keys: list) ->
}
)

self.data["ownerships"].append({"secret": secret_key, "authorizedKeys": new_authorized_keys})
self.data["ownerships"].append(
{"secret": secret_key, "authorizedKeys": new_authorized_keys}
)
return

def add_uco_transfer(self, send_to: Union[str, bytes], amount: float):
Expand All @@ -141,10 +146,18 @@ def add_uco_transfer(self, send_to: Union[str, bytes], amount: float):
isinstance(amount, float or int), "Amount must be float or int"
assert amount > 0, "Amount must be greater than 0"

self.data["ledger"]["uco"]["transfers"].append({"to": send_to, "amount": utils.to_big_int(amount)})
self.data["ledger"]["uco"]["transfers"].append(
{"to": send_to, "amount": utils.to_big_int(amount)}
)
return

def add_token_transfer(self, send_to: Union[str, bytes], amount: float, token_adress: Union[str, bytes], token_id: int):
def add_token_transfer(
self,
send_to: Union[str, bytes],
amount: float,
token_adress: Union[str, bytes],
token_id: int,
):
"""
Add a token transfer to the transaction
:param send_to: The public key of the receiver
Expand Down Expand Up @@ -179,7 +192,14 @@ def add_token_transfer(self, send_to: Union[str, bytes], amount: float, token_ad
isinstance(token_id, int), "Token id must be int"
assert token_id >= 0, "Token id must be greater or equal to 0"

self.data["ledger"]["token"]["transfers"].append({"to": send_to, "amount": utils.to_big_int(amount), "token": token_adress, "token_id": token_id})
self.data["ledger"]["token"]["transfers"].append(
{
"to": send_to,
"amount": utils.to_big_int(amount),
"token": token_adress,
"token_id": token_id,
}
)
return

def add_recipient(self, send_to: Union[str, bytes]):
Expand All @@ -200,7 +220,9 @@ def add_recipient(self, send_to: Union[str, bytes]):
self.data["recipients"].append(send_to)
return

def set_previous_signature_and_previous_public_key(self, signature: Union[str, bytes], public_key: Union[str, bytes]) -> None:
def set_previous_signature_and_previous_public_key(
self, signature: Union[str, bytes], public_key: Union[str, bytes]
) -> None:
"""
Set the transaction builder with Previous Publickey and Previous Signature
:param signature: The previous signature of the transaction
Expand Down Expand Up @@ -248,14 +270,22 @@ def set_address(self, address: Union[str, bytes]):
self.address = address
return

def build(self, seed: Union[str, bytes], index: int, curve: str = "ed25519", hash_algo: str = "sha256") -> None:
def build(
self,
seed: Union[str, bytes],
index: int,
curve: str = "ed25519",
hash_algo: str = "sha256",
) -> None:
private_key, public_key = crypto.derive_keypair(seed, index, curve)
address = crypto.derive_address(seed, index+1, curve, hash_algo)
address = crypto.derive_address(seed, index + 1, curve, hash_algo)
self.set_address(address)
self.previous_public_key = bytes.fromhex(public_key)

payload_for_previous_signature = self.previous_signature_payload()
self.previous_signature = crypto.sign(payload_for_previous_signature, private_key)
self.previous_signature = crypto.sign(
payload_for_previous_signature, private_key
)
return

def origin_sign(self, private_key: Union[str, bytes]) -> None:
Expand All @@ -273,7 +303,9 @@ def origin_sign(self, private_key: Union[str, bytes]) -> None:
else:
raise TypeError("Private key must be hex string or bytes")

self.origin_signature = crypto.sign(self.origin_signature_payload(), private_key)
self.origin_signature = crypto.sign(
self.origin_signature_payload(), private_key
)
return

def previous_signature_payload(self):
Expand All @@ -290,38 +322,72 @@ def previous_signature_payload(self):
for ownership in self.data["ownerships"]:
authorizedKeys = ownership.get("authorizedKeys")
secret = ownership.get("secret")
authorized_keys_buffer = [bytearray([len(authorizedKeys)])]

buff_auth_key_length = bytearray([len(authorizedKeys)])
authorized_keys_buffer = [
bytearray([len(buff_auth_key_length)]),
buff_auth_key_length,
]

for _authorizedKey in authorizedKeys:
public_key = _authorizedKey.get("publicKey")
encrypted_secret_key = _authorizedKey.get("encryptedSecretKey")
authorized_keys_buffer.append(public_key)
authorized_keys_buffer.append(encrypted_secret_key)

ownerships_buffer.append(
utils.int_to_32(len(bytes(secret))) +
secret +
b''.join(authorized_keys_buffer)
utils.int_to_32(len(bytes(secret)))
+ secret
+ b"".join(authorized_keys_buffer)
)

uco_transfers_buffers = [transfer['to'] + utils.int_to_64(transfer['amount']) for transfer in self.data["ledger"]["uco"]["transfers"]]
token_transfers_buffers = [transfer['token'] + transfer['to'] + utils.int_to_64(transfer['amount']) + bytearray([transfer['token_id']]) for transfer in self.data["ledger"]["token"]["transfers"]]
uco_transfers_buffers = [
transfer["to"] + utils.int_to_64(transfer["amount"])
for transfer in self.data["ledger"]["uco"]["transfers"]
]
token_transfers_buffers = [
transfer["token"]
+ transfer["to"]
+ utils.int_to_64(transfer["amount"])
+ bytearray([transfer["token_id"]])
for transfer in self.data["ledger"]["token"]["transfers"]
]

buf_ownership_length = bytearray([len(self.data["ownerships"])])
buf_uco_transfer_length = bytearray(
[len(self.data["ledger"]["uco"]["transfers"])]
)
buf_token_transfer_length = bytearray(
[len(self.data["ledger"]["token"]["transfers"])]
)
buf_recipient_length = bytearray([len(self.data["recipients"])])

return (
utils.int_to_32(VERSION)
+ self.address
+ bytearray([TX_TYPES[self.tx_type]])
# code
+ buff_code_size
+ self.data["code"]
# content
+ buf_content_size
+ self.data["content"]
+ bytearray([len(self.data["ownerships"])])
+ b''.join(ownerships_buffer)
+ bytearray([len(self.data["ledger"]["uco"]["transfers"])])
+ b''.join(uco_transfers_buffers)
+ bytearray([len(self.data["ledger"]["token"]["transfers"])])
+ b''.join(token_transfers_buffers)
+ bytearray([len(self.data["recipients"])])
+ b''.join(self.data["recipients"])
# ownerships
+ bytearray([len(buf_ownership_length)])
+ buf_ownership_length
+ b"".join(ownerships_buffer)
# uco transfers
+ bytearray([len(buf_uco_transfer_length)])
+ buf_uco_transfer_length
+ b"".join(uco_transfers_buffers)
# token transfers
+ bytearray([len(buf_token_transfer_length)])
+ buf_token_transfer_length
+ b"".join(token_transfers_buffers)
# recipients
+ bytearray([len(buf_recipient_length)])
+ buf_recipient_length
+ b"".join(self.data["recipients"])
)

def set_origin_sign(self, signature: Union[str, bytes]) -> None:
Expand Down Expand Up @@ -353,32 +419,54 @@ def origin_signature_payload(self):

def json(self):

data = {
data = {
"version": VERSION,
"address": self.address.hex(),
"type": self.tx_type,
"data": {
"content": self.data["content"].hex(),
"code": self.data["code"].decode("utf-8"),
"ownerships": [{"secret": _ownership.get('secret').hex(), "authorizedKeys": [{"publicKey": _authorizedKey.get('publicKey').hex(), "encryptedSecretKey": _authorizedKey.get('encryptedSecretKey').hex()} for _authorizedKey in _ownership.get('authorizedKeys')]} for _ownership in self.data["ownerships"]],
"ownerships": [
{
"secret": _ownership.get("secret").hex(),
"authorizedKeys": [
{
"publicKey": _authorizedKey.get("publicKey").hex(),
"encryptedSecretKey": _authorizedKey.get(
"encryptedSecretKey"
).hex(),
}
for _authorizedKey in _ownership.get("authorizedKeys")
],
}
for _ownership in self.data["ownerships"]
],
"ledger": {
"uco": {
"transfers": [{"to": transfer["to"].hex(), "amount": transfer["amount"]} for transfer in self.data["ledger"]["uco"]["transfers"]]
"transfers": [
{"to": transfer["to"].hex(), "amount": transfer["amount"]}
for transfer in self.data["ledger"]["uco"]["transfers"]
]
},
"token": {
"transfers": [{"to": transfer["to"].hex(), "token": transfer["token"].hex(), "amount": transfer["amount"], "token_id": transfer["token_id"]} for transfer in self.data["ledger"]["token"]["transfers"]]
}
"transfers": [
{
"to": transfer["to"].hex(),
"token": transfer["token"].hex(),
"amount": transfer["amount"],
"token_id": transfer["token_id"],
}
for transfer in self.data["ledger"]["token"]["transfers"]
]
},
},
"recipients": [recipient.hex() for recipient in self.data["recipients"]],

"recipients": [
recipient.hex() for recipient in self.data["recipients"]
],
},
"previousPublicKey": self.previous_public_key.hex(),
"previousSignature": self.previous_signature.hex(),
"originSignature": self.origin_signature.hex()
"originSignature": self.origin_signature.hex(),
}

return json.dumps(data)




Loading