Compare commits

..

No commits in common. "main" and "v0.1.1" have entirely different histories.
main ... v0.1.1

10 changed files with 22 additions and 422 deletions

View File

@ -1,10 +1,3 @@
# Conventions
When contributing code to this project, you MUST follow the requirements
specified here.
## Code Conventions
When contributing code to this project, you MUST follow these principles: When contributing code to this project, you MUST follow these principles:
- Code should be easy to read and understand. - Code should be easy to read and understand.
@ -20,13 +13,3 @@ When contributing code to this project, you MUST follow these principles:
- Documentation should document semantics, not syntax. - Documentation should document semantics, not syntax.
- Prefer importing modules, not individual items from modules. - Prefer importing modules, not individual items from modules.
- Do not use f-strings in logging statements. - Do not use f-strings in logging statements.
- Loop variables and walrus-expression-variables should be deleted when
unneeded to keep scope clean, and to avoid accidental use.
## Testing
When contributing test to this project, you MUST follow these principles:
- Do not use any testing libraries other than `pytest`.
- Mocking is the root of all evil. Writing your own stubs is much more
preferable.

View File

@ -4,29 +4,19 @@
# Partisia Blockchain Client #
![Test program/library](https://gitfub.space/Jmaa/pbc-client/actions/workflows/python-test.yml/badge.svg) ![Test program/library](https://gitfub.space/Jmaa/pbc-client/actions/workflows/python-test.yml/badge.svg)
Unofficial library for reading contract states from Partisia Blockchain.
This library is not officially associated with Partisia Blockchain nor Partisia
Group ApS.
## Dependencies ## Dependencies
This project requires [Python](https://www.python.org/) 3.8 or newer. This project requires [Python](https://www.python.org/) 3.8 or newer.
All required libraries can be installed easily using: This project does not have any library requirements 😎
```bash
pip install -r requirements.txt
```
Full list of requirements:
- [coincurve](https://pypi.org/project/coincurve/)
- [pbcabi](https://gitfub.space/Jmaa/pbcabi)
## Contributing ## Contributing

View File

@ -1,58 +1,38 @@
"""# Partisia Blockchain Client
Unofficial library for reading contract states from Partisia Blockchain.
This library is not officially associated with Partisia Blockchain nor Partisia
Group ApS.
"""
import base64
import dataclasses import dataclasses
import datetime import datetime
import email.utils import email.utils
import json import json
import logging import logging
import time
from collections.abc import Mapping from collections.abc import Mapping
from decimal import Decimal from decimal import Decimal
from typing import Any from typing import Any
import pbcabi
import requests import requests
from frozendict import frozendict from frozendict import frozendict
from ._version import __version__ # noqa: F401
from .crypto import SenderAuthentication, sign_transaction
from .pbc_types import (
Address,
SignedTransaction,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
HOSTNAME_MAINNET = 'reader.partisiablockchain.com' # mainnet: https://reader.partisiablockchain.com
HOSTNAME_TESTNET = 'node1.testnet.partisiablockchain.com' # testnet: https://node1.testnet.partisiablockchain.com
HOSTNAME = 'reader.partisiablockchain.com'
URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local' URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local'
URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global' URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global'
URL_TRANSACTION = 'https://{hostname}/{shard}blockchain/transaction'
URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false' URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false'
URL_CHAIN_ID = 'https://{hostname}/blockchain/chainId'
URL_NONCE = 'https://{hostname}/{shard}blockchain/account/{address}'
MPC_DECIMALS = 10000 MPC_DECIMALS = 10000
TRANSACTION_VALIDITY_DURATION = 60
def shard_id_for_address(address: str) -> str:
def shard_id_for_address(address: str | Address) -> str:
address = str(address)
# Very rough implementation # Very rough implementation
if address is None: if address is None:
msg = 'Address must not be None' msg = 'Address must not be None'
raise TypeError(msg) raise TypeError(msg)
if address.endswith(('a', 'd', 'c')): if address.endswith('a'):
return 'shards/Shard0/' return 'shards/Shard0/'
if address.endswith('2'): if address.endswith('2'):
return 'shards/Shard1/' return 'shards/Shard1/'
@ -69,57 +49,8 @@ class Balances:
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class PbcClient: class PbcClient:
session: requests.Session session: requests.Session
sender_authentication: SenderAuthentication | None = None
hostname: str = HOSTNAME_MAINNET
def __post_init__(self): def get_json(
assert isinstance(self.session, requests.Session)
assert isinstance(self.hostname, str)
if self.sender_authentication is not None:
assert isinstance(self.sender_authentication, SenderAuthentication)
def send_transaction(self, contract_address: str, rpc: bytes, gas_cost: int):
if self.sender_authentication is None:
msg = 'PbcClient.sender_authentication required for send_transaction'
raise Exception(msg)
valid_to_time: int = int(time.time() + TRANSACTION_VALIDITY_DURATION) * 1000
signed_transaction = sign_transaction(
self.sender_authentication,
self.get_sender_authentication_nonce(),
valid_to_time,
gas_cost,
self.get_chain_id(),
contract_address,
rpc,
)
assert len(signed_transaction.inner.transaction.rpc_serialize()) == 25
assert len(signed_transaction.signature.rpc_serialize()) == 65
assert len(signed_transaction.inner.rpc_serialize()) == 49
assert len(signed_transaction.rpc_serialize()) == 114
return self.send_signed_transaction(signed_transaction)
def send_signed_transaction(self, signed_transaction: SignedTransaction):
url = URL_TRANSACTION.format(
hostname=self.hostname,
shard=shard_id_for_address(
signed_transaction.inner.transaction.contract_address,
),
)
transaction_payload: str = base64.b64encode(
signed_transaction.rpc_serialize(),
).decode('utf8')
self._get_json(
url,
data={'transactionPayload': transaction_payload},
method='PUT',
)
def _get_json(
self, self,
url: str, url: str,
data: Mapping[str, Any] = frozendict(), data: Mapping[str, Any] = frozendict(),
@ -141,8 +72,6 @@ class PbcClient:
'date', 'date',
) )
date = email.utils.parsedate_to_datetime(date_text) date = email.utils.parsedate_to_datetime(date_text)
if response.text == '':
return (None, date)
json_data = response.json() json_data = response.json()
if json_data is None: if json_data is None:
msg = 'No result data for ' + url msg = 'No result data for ' + url
@ -153,18 +82,18 @@ class PbcClient:
data: dict = {'path': []} data: dict = {'path': []}
url: str = URL_ACCOUNT_PLUGIN_GLOBAL.format( url: str = URL_ACCOUNT_PLUGIN_GLOBAL.format(
hostname=self.hostname, hostname=HOSTNAME,
shard='', shard='',
) )
json_data, date = self._get_json(url, data=data) json_data, date = self.get_json(url, data=data)
return json_data['coins']['coins'] return json_data['coins']['coins']
def get_account_balances(self, address: str) -> Balances: def get_account_balances(self, address: str) -> Balances:
coins = self.determine_coins() coins = self.determine_coins()
url = URL_ACCOUNT_PLUGIN.format( url = URL_ACCOUNT_PLUGIN.format(
hostname=self.hostname, hostname=HOSTNAME,
shard=shard_id_for_address(address), shard=shard_id_for_address(address),
) )
@ -174,7 +103,7 @@ class PbcClient:
{'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address}, {'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address},
], ],
} }
account_data, date = self._get_json(url, data=data) account_data, date = self.get_json(url, data=data)
byoc: dict[str, Decimal] = {} byoc: dict[str, Decimal] = {}
mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS
@ -189,81 +118,12 @@ class PbcClient:
return Balances(date, mpc, byoc) return Balances(date, mpc, byoc)
def get_chain_id(self) -> str:
url = URL_CHAIN_ID.format(
hostname=self.hostname,
)
return self._get_json(url, method='GET')[0]['chainId']
def get_sender_authentication_nonce(self) -> int:
return self.get_nonce_for_account(self.sender_authentication.sender_address())
def get_nonce_for_account(self, address: str) -> int:
url = URL_NONCE.format(
hostname=self.hostname,
shard=shard_id_for_address(address),
address=address,
)
return self._get_json(url, method='GET')[0]['nonce']
def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]: def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]:
# TODO: Rename to get_contract_state_json
url = URL_CONTRACT_STATE.format( url = URL_CONTRACT_STATE.format(
hostname=self.hostname, hostname=HOSTNAME,
shard=shard_id_for_address(address), shard=shard_id_for_address(address),
address=address, address=address,
) )
data: dict = {'path': []} data: dict = {'path': []}
return self._get_json(url, data=data) return self.get_json(url, data=data)
def get_typed_contract_state(self, address: str) -> tuple[dict, datetime.datetime]:
"""
Only suitable for non-governance WASM contracts.
"""
file_abi = self.get_contract_abi(address)
state, server_time = self.get_contract_state(address)
state_bytes = base64.b64decode(state['state']['data'])
state_deserialized = file_abi.contract.read_state(state_bytes)
return state_deserialized, server_time
def get_typed_contract_avl_tree(
self,
address: str,
avl_tree_id: pbcabi.model.AvlTreeId,
) -> tuple[dict, datetime.datetime]:
file_abi = self.get_contract_abi(address)
state, server_time = self.get_contract_state(address)
for avl_tree in state['avlTrees']:
if avl_tree['key'] == avl_tree_id.avl_tree_id:
break
data = {}
for key_and_value in avl_tree['value']['avlTree']:
key_bytes = base64.b64decode(key_and_value['key']['data']['data'])
value_bytes = base64.b64decode(key_and_value['value']['data'])
key = file_abi.contract.read_state(
key_bytes,
avl_tree_id.type_spec.type_key,
)
value = file_abi.contract.read_state(
value_bytes,
avl_tree_id.type_spec.type_value,
)
data[key] = value
# TODO: Type
return data, server_time
def get_contract_abi(self, address: str) -> pbcabi.model.FileAbi:
url = URL_CONTRACT_STATE.format(
hostname=self.hostname,
shard=shard_id_for_address(address),
address=address,
)
meta, _ = self._get_json(url, method='GET')
abi_bytes = base64.b64decode(meta['abi'])
return pbcabi.model.FileAbi.read_from(abi_bytes)

View File

@ -1 +1 @@
__version__ = '0.1.5' __version__ = '0.1.1'

View File

@ -1,84 +0,0 @@
import dataclasses
import coincurve
from .pbc_types import (
Address,
HashSha256,
Signature,
SignedTransaction,
SignedTransactionInnerPart,
Transaction,
)
def find_recovery_id(
der_sig: bytes,
message_hash: HashSha256,
expected_public_key: coincurve.PublicKey,
) -> int:
r, s = coincurve.der.parse_signature(der_sig)
for recovery_id in range(4):
recovered_public_key = coincurve.PublicKey.from_signature_and_message(
signature=r + s + bytes([recovery_id]),
message=message_hash._bytes,
hasher=None,
)
if recovered_public_key.format() == expected_public_key.format():
return recovery_id
return None
@dataclasses.dataclass(frozen=True)
class SenderAuthentication:
_private_key_hex: str
def sender_address(self) -> Address:
verifying_key_repr = self._public_key().format(False)
assert len(verifying_key_repr) == 65
hashed = HashSha256.of_bytes(verifying_key_repr)
return Address(b'\0' + hashed._bytes[-20:])
def sign_hash(self, hash_to_sign: HashSha256) -> Signature:
private_key = self._private_key()
signature_wrong_location = private_key.sign_recoverable(
hash_to_sign._bytes,
hasher=lambda x: x,
)
signature = signature_wrong_location[64:] + signature_wrong_location[:64]
return Signature(signature)
def _public_key(self) -> coincurve.PublicKey:
return self._private_key().public_key
def _private_key(self) -> coincurve.PrivateKey:
return coincurve.PrivateKey.from_hex(
self._private_key_hex,
)
def sign_transaction(
sender_authentication: SenderAuthentication,
nonce: int,
valid_to_time: int,
gas_cost: int,
chain_id: str,
contract_address: Address | str,
transaction_rpc: bytes,
) -> SignedTransaction:
sender = sender_authentication.sender_address
inner: SignedTransactionInnerPart = SignedTransactionInnerPart(
nonce,
valid_to_time,
gas_cost,
Transaction(Address.from_string(contract_address), transaction_rpc),
)
signature: Signature = sender_authentication.sign_hash(inner.hash(chain_id))
return SignedTransaction(inner, signature)

View File

@ -1,106 +0,0 @@
import dataclasses
import hashlib
def size_prefixed(_bytes: bytes) -> bytes:
return len(_bytes).to_bytes(4, 'big') + _bytes
@dataclasses.dataclass(frozen=True)
class Address:
_bytes: bytes
def __post_init__(self):
assert len(self._bytes) == 21, len(self._bytes)
def rpc_serialize(self) -> bytes:
return self._bytes
def __str__(self) -> str:
return self._bytes.hex()
@staticmethod
def from_string(s: 'Address | str') -> 'Address':
if isinstance(s, Address):
return s
return Address(bytes.fromhex(s))
@dataclasses.dataclass(frozen=True)
class Signature:
_bytes: bytes
def __post_init__(self):
assert len(self._bytes) == 65, len(self._bytes)
assert self._bytes[0] < 4, 'First byte is the recovery id. Must be less than 4'
def rpc_serialize(self) -> bytes:
return self._bytes
def __str__(self) -> str:
return self._bytes.hex()
@dataclasses.dataclass(frozen=True)
class HashSha256:
_bytes: bytes
def __post_init__(self):
assert len(self._bytes) == 32, len(self._bytes)
@staticmethod
def of_bytes(b: bytes) -> 'HashSha256':
return HashSha256(hashlib.sha256(b).digest())
def rpc_serialize(self) -> bytes:
return self._bytes
def __str__(self) -> str:
return self._bytes.hex()
@dataclasses.dataclass(frozen=True)
class Transaction:
contract_address: Address
transaction_rpc: bytes
def __post_init__(self):
assert isinstance(self.contract_address, Address), self.contract_address
def rpc_serialize(self) -> bytes:
return self.contract_address.rpc_serialize() + size_prefixed(
self.transaction_rpc,
)
@dataclasses.dataclass(frozen=True)
class SignedTransactionInnerPart:
nonce: int
valid_to_time: int
gas_cost: int
transaction: Transaction
def rpc_serialize(self) -> bytes:
return (
self.nonce.to_bytes(8, 'big')
+ self.valid_to_time.to_bytes(8, 'big')
+ self.gas_cost.to_bytes(8, 'big')
+ self.transaction.rpc_serialize()
)
def hash(self, chain_id: str) -> HashSha256:
return HashSha256.of_bytes(
self.rpc_serialize() + size_prefixed(chain_id.encode('utf8')),
)
@dataclasses.dataclass(frozen=True)
class SignedTransaction:
inner: SignedTransactionInnerPart
signature: Signature
def hash(self, chain_id: str) -> HashSha256:
return self.inner.hash(chain_id)
def rpc_serialize(self) -> bytes:
return self.signature.rpc_serialize() + self.inner.rpc_serialize()

View File

@ -1,2 +0,0 @@
coincurve
pbcabi @ git+https://gitfub.space/Jmaa/pbcabi

View File

@ -11,16 +11,10 @@ from setuptools import setup
PACKAGE_NAME = 'pbc_client' PACKAGE_NAME = 'pbc_client'
PACKAGE_DESCRIPTION = """ PACKAGE_DESCRIPTION = """
# Partisia Blockchain Client
Unofficial library for reading contract states from Partisia Blockchain.
This library is not officially associated with Partisia Blockchain nor Partisia
Group ApS.
""".strip() """.strip()
PACKAGE_DESCRIPTION_SHORT = """ PACKAGE_DESCRIPTION_SHORT = """
Unofficial library for reading contract states from Partisia Blockchain.""".strip() """.strip()
def parse_version_file(text: str) -> str: def parse_version_file(text: str) -> str:
@ -35,10 +29,7 @@ with open(PACKAGE_NAME + '/_version.py') as f:
version = parse_version_file(f.read()) version = parse_version_file(f.read())
REQUIREMENTS_MAIN = [ REQUIREMENTS_MAIN = []
'coincurve',
'pbcabi @ git+https://gitfub.space/Jmaa/pbcabi',
]
REQUIREMENTS_TEST = [] REQUIREMENTS_TEST = []

View File

@ -1,30 +0,0 @@
from pbc_client.crypto import SenderAuthentication, sign_transaction
def test_sign():
sender_authentication = SenderAuthentication('01')
chain_id = 'SpecificChainIDLocatable'
contract_address = '000000000000000000000000000000000000000001'
signed = sign_transaction(
sender_authentication=sender_authentication,
nonce=2,
valid_to_time=3,
gas_cost=2,
chain_id=chain_id,
contract_address=contract_address,
transaction_rpc=bytes([0 for i in range(99)]),
)
print(signed)
assert (
str(signed.inner.rpc_serialize().hex())
== '00000000000000020000000000000003000000000000000200000000000000000000000000000000000000000100000063000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000'
)
assert (
str(signed.hash(chain_id))
== '1be2895a85862e8dd3cc75b290cc28a22e960ae02dd5c07a73b93716f9adbee8'
)
assert (
str(signed.signature)
== '01f0f7d8f44919466eedbebc76bcb369bbb4c6f4f076e25c0ffe8ec9285890e53b4e39098540d088878a019c345ad73963543ce813fb9ccf4b84b0c25770452bd1'
)

View File

@ -1,2 +0,0 @@
def test_init():
import pbc_client # noqa: F401