diff --git a/pbcabi/__init__.py b/pbcabi/__init__.py new file mode 100644 index 0000000..2d8791d --- /dev/null +++ b/pbcabi/__init__.py @@ -0,0 +1,10 @@ +''' +Utility library for parsing and processing the PBC ABI format. + +Follows the ABI specification quite closely. The full ABI specification can be found at: + + https://partisiablockchain.gitlab.io/documentation/abiv.html +''' + +import pbcabi.model +import pbcabi.binaryreader diff --git a/pbcabi/binaryreader.py b/pbcabi/binaryreader.py new file mode 100644 index 0000000..a07e596 --- /dev/null +++ b/pbcabi/binaryreader.py @@ -0,0 +1,137 @@ +''' +Utility module for reading binary streams of data. + +Follows the ABI specification quite closely. The full ABI specification can be found at: + + https://partisiablockchain.gitlab.io/documentation/abiv.html +''' + +import io + + +class BinaryReader: + ''' + Wrapper for io.BytesIO for iteratively parsing byte streams. + ''' + + def __init__(self, buf): + ''' + Initializes BinaryReader. + ''' + self.buf = buf + if isinstance(self.buf, bytes): + self.buf = io.BytesIO(self.buf) + self.size = len(buf) + self.position = 0 + + def readBytes(self, num_bytes): + ''' + Reads a number of bytes from stream. + ''' + self.position += num_bytes + bts = self.buf.read(num_bytes) + if len(bts) != num_bytes: + raise Exception("Not enough bytes in buffer: Got {}, but expected {}".format(len(bts), num_bytes)) + return bts + + def readDynamicBytes(self, size_reader = None): + ''' + Reads a dynamically sized list of bytes. + ''' + size_reader = size_reader or BinaryReader.readUInt32BigEndian + num_bytes = size_reader(self) + return self.readBytes(num_bytes) + + def readString(self, size_reader = None): + ''' + Reads a string from stream. + ''' + return self.readDynamicBytes(size_reader).decode('utf8') + + def readUInt8(self): + ''' + Reads a unsigned 8-bit integer from stream. + ''' + bytes = self.readBytes(1) + return bytes[0] + + def readUInt32BigEndian(self): + ''' + Reads a unsigned 32-bit integer from stream. + ''' + return self.readUIntBigEndian(4) + + def readUInt32LittleEndian(self): + ''' + Reads a unsigned 32-bit integer from stream. + ''' + return self.readUIntLittleEndian(4) + + def readUIntBigEndian(self, num_bytes): + ''' + Reads an unsigned N-bit integer from stream. + ''' + bytes = self.readBytes(num_bytes) + c = 0 + for i in range(0, num_bytes): + c += bytes[num_bytes - i - 1] * 2**(i * 8) + return c + + def readSignedIntBigEndian(self, num_bytes): + ''' + Reads an signed N-bit integer from stream. + ''' + # TODO: Test! + result = self.readUIntBigEndian(num_bytes) + half = 2**(num_bytes * 8) + if result > half: + result -= half + return result + + def readUIntLittleEndian(self, num_bytes): + ''' + Reads an unsigned N-bit integer from stream. + ''' + bytes = self.readBytes(num_bytes) + c = 0 + for i in range(0, num_bytes): + c += bytes[i] * 2**(i * 8) + return c + + def readSignedIntLittleEndian(self, num_bytes): + ''' + Reads an signed N-bit integer from stream. + ''' + # TODO: Test! + result = self.readUIntLittleEndian(num_bytes) + half = 2**(num_bytes * 8) + if result > half: + result -= half + return result + + def readLeb128(self): + ''' + Reads a LEB-128 integer from stream. + ''' + # TODO: Test! + v = self.readUInt8() + count = 0 + while v & 0x80 != 0: + v = self.readUInt8() + count += 1 + assert count < 6, "TODO: " + str(v) + return v + + def readList(self, fn, size_reader=None): + ''' + Reads a list of elements from stream, by applying the given function + several times. + + Function must take a single argument of type BinaryReader. + ''' + size_reader = size_reader or BinaryReader.readUInt32BigEndian + length = size_reader(self) + ls = [] + for i in range(0, length): + ls.append(fn(self)) + return ls diff --git a/pbcabi/data.py b/pbcabi/data.py new file mode 100644 index 0000000..f278a08 --- /dev/null +++ b/pbcabi/data.py @@ -0,0 +1,208 @@ +''' +Definitions of several useful types commonly seen on Partisia Blockchain. + +Follows the ABI specification quite closely. The full ABI specification can be found at: + + https://partisiablockchain.gitlab.io/documentation/abiv.html +''' + +import datetime +import dataclasses +from typing import Mapping, Collection, Optional, Union +from enum import Enum +import enforce_typing + +def hexformat(bts: bytes, sep=''): + return sep.join('{:02X}'.format(c) for c in bts) + +class AddressType(Enum): + ''' + Type of a PBC BlockchainAddress. + ''' + ACCOUNT = 0x00 + CONTRACT_SYSTEM = 0x01 + CONTRACT_PUBLIC = 0x02 + CONTRACT_ZK = 0x03 + CONTRACT_GOVERANCE = 0x04 + + def human_name(self): + ''' + Produces a human-readable name for self. + ''' + return ADDRESS_TYPES[self] + + +ADDRESS_TYPES = { + AddressType.ACCOUNT: 'Account', + AddressType.CONTRACT_SYSTEM: 'System Contract', + AddressType.CONTRACT_PUBLIC: 'Public Contract', + AddressType.CONTRACT_ZK: 'ZK Contract', + AddressType.CONTRACT_GOVERANCE: 'Governance Contract', +} + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class ByteData(object): + ''' + Supertype for several built-in PBC types. + ''' + data: bytes + + def to_hex(self): + ''' + Formats this byte data as a hex string. + ''' + return hexformat(self.data) + + def short_str(self): + ''' + Formats this byte data as a short hex string. + ''' + long = str(self) + return '{}..{}'.format(long[:4], long[-4:]) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class BlockchainAddress(ByteData): + ''' + Address on PBC. + ''' + + BYTE_LEN = 21 + + def __post_init__(self): + assert len(self.data) == BlockchainAddress.BYTE_LEN + assert self.type() + + @staticmethod + def read_from(reader): + return BlockchainAddress(reader.readBytes(BlockchainAddress.BYTE_LEN)) + + @staticmethod + def from_hex_hash(s): + return BlockchainAddress(bytes.fromhex(s)) + + def type(self) -> AddressType: + return AddressType(self.data[0]) + + def __repr__(self): + return self.to_hex() + + def __str__(self): + return repr(self) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class Hash(ByteData): + ''' + Hash. + ''' + + data: bytes + + BYTE_LEN = 32 + + @staticmethod + def read_from(reader): + return Hash(reader.readBytes(Hash.BYTE_LEN)) + + @classmethod + def from_hex(cls, s): + return Hash(bytes.fromhex(s)) + + def __repr__(self): + return self.to_hex() + + def __str__(self): + return repr(self) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class PublicKey(ByteData): + ''' + PBC public key. + ''' + + data: bytes + + BYTE_LEN = 33 + + @staticmethod + def read_from(reader): + return PublicKey(reader.readBytes(PublicKey.BYTE_LEN)) + + def __repr__(self): + return self.to_hex() + + def __str__(self): + return repr(self) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class Signature(ByteData): + ''' + Message signature on PBC. + ''' + + data: bytes + + BYTE_LEN = 65 + + @staticmethod + def read_from(reader): + return Signature(reader.readBytes(Signature.BYTE_LEN)) + + def __repr__(self): + return self.to_hex() + + def __str__(self): + return repr(self) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class BlsPublicKey(ByteData): + ''' + BLS Public Key. + ''' + + data: bytes + + BYTE_LEN = 96 + + @staticmethod + def read_from(reader): + return BlsPublicKey(reader.readBytes(BlsPublicKey.BYTE_LEN)) + + def __repr__(self): + return self.to_hex() + + def __str__(self): + return repr(self) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, order=True) +class BlsSignature(ByteData): + ''' + BLS signature. + ''' + + data: bytes + + BYTE_LEN = 48 + + @staticmethod + def read_from(reader): + return BlsSignature(reader.readBytes(BlsSignature.BYTE_LEN)) + + def __repr__(self): + return self.to_hex() + + def __str__(self): + return repr(self) diff --git a/pbcabi/model.py b/pbcabi/model.py new file mode 100644 index 0000000..9aa21fb --- /dev/null +++ b/pbcabi/model.py @@ -0,0 +1,680 @@ +''' +Specifies the ABI Spec model and how to deserialize the ABI, and additionally +how specific type specifications can deserialize their elements. + +Follows the ABI specification quite closely. The full ABI specification can be found at: + + https://partisiablockchain.gitlab.io/documentation/abiv.html +''' + +from frozendict import frozendict +import dataclasses +from typing import Mapping +from enum import Enum +import enforce_typing +from .binaryreader import BinaryReader +import pbcabi.data as data + +Identifier = str + + +class SerializeMode(Enum): + RPC = 0x01 + STATE = 0x02 + ZK = 0x03 + + +class SimpleType(Enum): + ''' + The specific simple type. + ''' + + U8 = 0X01 + U16 = 0X02 + U32 = 0X03 + U64 = 0X04 + U128 = 0X05 + U256 = 0X18 + I8 = 0X06 + I16 = 0X07 + I32 = 0X08 + I64 = 0X09 + I128 = 0X0a + STRING = 0X0b + BOOL = 0X0c + ADDRESS = 0X0d + HASH = 0X13 + PUBLIC_KEY = 0X14 + SIGNATURE = 0X15 + BLS_PUBLIC_KEY = 0X16 + BLS_SIGNATURE = 0X17 + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class TypeSpec: + ''' + Supertype of all 'TypeSpec'. + ''' + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + if type_spec_read_from := TYPE_SPEC_SUBTYPE_READ_FROM_BY_DISCRIMINANT.get( + discriminant): + type_spec = type_spec_read_from(reader) + return type_spec + assert False, "Unknown discriminant: 0x{:02x}".format(discriminant) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class NamedTypeRef(TypeSpec): + ''' + Reference to a named type. + ''' + + idx: int + + DISCRIMINANT = 0x00 + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + assert reader.readUInt8() == NamedTypeRef.DISCRIMINANT + return NamedTypeRef.read_from_inner(reader) + + @staticmethod + def read_from_inner(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + return NamedTypeRef(reader.readUInt8()) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + return type_env[self.idx].read_element_from(reader, type_env, mode) + + +READ_SIMPLE_TYPE = {} + +READ_SIMPLE_TYPE[SerializeMode.RPC] = { + (SimpleType.BOOL): lambda r: r.readUInt8() != 0, + (SimpleType.U8): BinaryReader.readUInt8, + (SimpleType.U16): lambda r: r.readUIntBigEndian(2), + (SimpleType.U32): BinaryReader.readUInt32BigEndian, + (SimpleType.U64): lambda r: r.readUIntBigEndian(8), + (SimpleType.U128): lambda r: r.readUIntBigEndian(16), + (SimpleType.U256): lambda r: r.readUIntBigEndian(32), + (SimpleType.I8): lambda r: r.readSignedIntBigEndian(1), + (SimpleType.I16): lambda r: r.readSignedIntBigEndian(2), + (SimpleType.I32): lambda r: r.readSignedIntBigEndian(4), + (SimpleType.I64): lambda r: r.readSignedIntBigEndian(8), + (SimpleType.I128): lambda r: r.readSignedIntBigEndian(16), + (SimpleType.STRING): lambda r: r.readString(BinaryReader.readUInt32BigEndian), + (SimpleType.ADDRESS): data.BlockchainAddress.read_from, + (SimpleType.HASH): data.Hash.read_from, + (SimpleType.PUBLIC_KEY): data.PublicKey.read_from, + (SimpleType.SIGNATURE): data.Signature.read_from, + (SimpleType.BLS_PUBLIC_KEY): data.BlsPublicKey.read_from, + (SimpleType.BLS_SIGNATURE): data.BlsSignature.read_from, +} + +READ_SIMPLE_TYPE[SerializeMode.STATE] = { + (SimpleType.BOOL): lambda r: r.readUInt8() != 0, + (SimpleType.U8): BinaryReader.readUInt8, + (SimpleType.U16): lambda r: r.readUIntLittleEndian(2), + (SimpleType.U32): BinaryReader.readUInt32LittleEndian, + (SimpleType.U64): lambda r: r.readUIntLittleEndian(8), + (SimpleType.U128): lambda r: r.readUIntLittleEndian(16), + (SimpleType.U256): lambda r: r.readUIntLittleEndian(32), + (SimpleType.I8): lambda r: r.readSignedIntLittleEndian(1), + (SimpleType.I16): lambda r: r.readSignedIntLittleEndian(2), + (SimpleType.I32): lambda r: r.readSignedIntLittleEndian(4), + (SimpleType.I64): lambda r: r.readSignedIntLittleEndian(8), + (SimpleType.I128): lambda r: r.readSignedIntLittleEndian(16), + (SimpleType.STRING): lambda r: r.readString(BinaryReader.readUInt32LittleEndian), + (SimpleType.ADDRESS): data.BlockchainAddress.read_from, + (SimpleType.HASH): data.Hash.read_from, + (SimpleType.PUBLIC_KEY): data.PublicKey.read_from, + (SimpleType.SIGNATURE): data.Signature.read_from, + (SimpleType.BLS_PUBLIC_KEY): data.BlsPublicKey.read_from, + (SimpleType.BLS_SIGNATURE): data.BlsSignature.read_from, +} + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class SimpleTypeSpec(TypeSpec): + ''' + Simple type spec. + ''' + + type: SimpleType + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + return SimpleTypeSpec(SimpleType[reader.readUInt8()]) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + return READ_SIMPLE_TYPE[mode][self.type](reader) + + +SIZE_SIMPLE_TYPE_SPEC = SimpleTypeSpec(SimpleType.U32) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class CompositeTypeSpec(TypeSpec): + ''' + Supertype of all composite type specs. + ''' + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class VecTypeSpec(TypeSpec): + ''' + Type spec for a dynamically-sized vector of some some type. + ''' + + type_elements: TypeSpec + + DISCRIMINANT = 0x0e + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + type_elements = TypeSpec.read_from(reader) + return VecTypeSpec(type_elements) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + length = SIZE_SIMPLE_TYPE_SPEC.read_element_from(reader, type_env, mode) + + if self.type_elements == SimpleTypeSpec(SimpleType.U8): + return reader.readBytes(length) + + array = [] + for i in range(0, length): + elem = self.type_elements.read_element_from(reader, type_env, mode) + array.append(elem) + return array + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class MapTypeSpec(TypeSpec): + ''' + Type spec for a dynamically-sized map from some type to another type. + ''' + + type_key: TypeSpec + type_value: TypeSpec + + DISCRIMINANT = 0x0f + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + type_key = TypeSpec.read_from(reader) + type_value = TypeSpec.read_from(reader) + return MapTypeSpec(type_key, type_value) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + out = {} + length = SIZE_SIMPLE_TYPE_SPEC.read_element_from(reader, type_env, mode) + for i in range(0, length): + key = self.type_key.read_element_from(reader, type_env, mode) + val = self.type_value.read_element_from(reader, type_env, mode) + out[key] = val + return out + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class SetTypeSpec(TypeSpec): + ''' + Type spec for a dynamically-sized set. + ''' + + type_elements: TypeSpec + + DISCRIMINANT = 0x10 + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + type_elements = TypeSpec.read_from(reader) + return SetTypeSpec(type_elements) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + array = [] + length = SIZE_SIMPLE_TYPE_SPEC.read_element_from(reader, type_env, mode) + for i in range(0, length): + array.append(self.type_elements.read_element_from(reader, type_env), + mode) + return frozenset(array) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class ArrayTypeSpec(TypeSpec): + ''' + Type spec for byte-array. + ''' + + length: int + + DISCRIMINANT = 0x11 + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + length = reader.readUInt8() + return ArrayTypeSpec(length) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + return reader.readBytes(self.length) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from state 'BinaryReader'. + ''' + return reader.readBytes(self.length) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class OptionTypeSpec(TypeSpec): + ''' + Type spec for some option type containing a specific sub-type. + ''' + + type_elements: TypeSpec + + DISCRIMINANT = 0x12 + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + type_elements = TypeSpec.read_from(reader) + return OptionTypeSpec(type_elements) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + if discriminant == 0: + return None + else: + return self.type_elements.read_element_from(reader, type_env, mode) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from state 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + if discriminant == 0: + return None + else: + return self.type_elements.read_element_from(reader, type_env, mode) + + +def simple_type_spec_read_from(variant): + v = SimpleTypeSpec(variant) + return lambda r: v + + +TYPE_SPEC_SUBTYPE_READ_FROM_BY_DISCRIMINANT = { + 0x00: NamedTypeRef.read_from_inner, +} | { + v.value: simple_type_spec_read_from(v) for v in SimpleType +} | { + t.DISCRIMINANT: t.read_from for t in + [VecTypeSpec, MapTypeSpec, SetTypeSpec, ArrayTypeSpec, OptionTypeSpec] +} + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots = True, order=True) +class Version: + ''' + Version field. + ''' + major: int + minor: int + patch: int + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this struct from 'BinaryReader'. + ''' + major = reader.readUInt8() + minor = reader.readUInt8() + patch = reader.readUInt8() + return Version(major, minor, patch) + + +class FnKind(Enum): + ''' + Function kinds that contract can be invoked with. + ''' + + INIT = 0x01 + ACTION = 0x02 + CALLBACK = 0x03 + ZK_SECRET_INPUT = 0x10 + ZK_VAR_INPUTTED = 0x11 + ZK_VAR_REJECTED = 0x12 + ZK_COMPUTE_COMPLETE = 0x13 + ZK_VAR_OPENED = 0x14 + ZK_USER_VAR_OPENED = 0x15 + ZK_ATTESTATION_COMPLETE = 0x16 + ZK_SECRET_INPUT_WITH_EXPLICIT_TYPE = 0x17 + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this struct from 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + if found := [kind for kind in FnKind if kind.value == discriminant]: + assert len(found) == 1 + return found[0] + assert False, "Unknown discriminant: 0x{:02x}".format(discriminant) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class FieldAbi: + ''' + Field of a struct. + ''' + + name: Identifier + type: TypeSpec + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this struct from 'BinaryReader'. + ''' + name = reader.readString() + type = TypeSpec.read_from(reader) + return FieldAbi(name, type) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class ArgumentAbi: + ''' + Argument of a function. + ''' + name: Identifier + type: TypeSpec + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize this struct from 'BinaryReader'. + ''' + name = reader.readString() + type = TypeSpec.read_from(reader) + return ArgumentAbi(name, type) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class NamedTypeSpec: + ''' + Supertype of named types. + ''' + + name: Identifier + type_index: int + + @staticmethod + def read_from(reader: BinaryReader, type_index: int): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + if discriminant == StructTypeSpec.DISCRIMINANT: + return StructTypeSpec.read_from(reader, type_index) + if discriminant == EnumTypeSpec.DISCRIMINANT: + return EnumTypeSpec.read_from(reader, type_index) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class StructTypeSpec(NamedTypeSpec): + ''' + Struct type specification. + ''' + + fields: list[FieldAbi] + + DISCRIMINANT = 0x01 + + @staticmethod + def read_from(reader: BinaryReader, type_index: int): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + name = reader.readString() + variants = reader.readList(FieldAbi.read_from) + return StructTypeSpec(name, type_index, variants) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + result = {} + for field in self.fields: + result[field.name] = field.type.read_element_from( + reader, type_env, mode) + result['__type'] = self.name + return result + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class EnumVariant: + ''' + Enum variant specification. + ''' + + discriminant: int + definition: NamedTypeRef + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize an 'EnumVariant' from 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + type = NamedTypeRef.read_from_inner(reader) + return EnumVariant(discriminant, type) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class EnumTypeSpec(NamedTypeSpec): + ''' + Enum type specification. + ''' + + variants_by_discriminant: Mapping[int, EnumVariant] + + DISCRIMINANT = 0x02 + + @staticmethod + def read_from(reader: BinaryReader, type_index: int): + ''' + Deserialize this 'TypeSpec' type from 'BinaryReader'. + ''' + name = reader.readString() + variants = reader.readList(EnumVariant.read_from) + variants_by_discriminant = frozendict( + {v.discriminant: v for v in variants}) + return EnumTypeSpec(name, type_index, variants_by_discriminant) + + def read_element_from(self, reader: BinaryReader, type_env, + mode: SerializeMode): + ''' + Deserialize elements for this 'TypeSpec' from RPC 'BinaryReader'. + ''' + discriminant = reader.readUInt8() + variant = self.variants_by_discriminant[discriminant] + return variant.definition.read_element_from(reader, type_env, mode) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class FnAbi: + ''' + Function definition. + ''' + + kind: FnKind + name: Identifier + shortname: int + arguments: list[ArgumentAbi] + secret_argument: ArgumentAbi | None + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize 'FnAbi' from 'BinaryReader'. + ''' + kind = FnKind.read_from(reader) + name = reader.readString() + shortname = reader.readLeb128() + arguments = reader.readList(ArgumentAbi.read_from) + secret_argument = None + if kind == FnKind.ZK_SECRET_INPUT_WITH_EXPLICIT_TYPE: + secret_argument = ArgumentAbi.read_from(reader) + + return FnAbi(kind, name, shortname, arguments, secret_argument) + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class ContractAbi: + ''' + Contract definition. + ''' + + named_types_by_id: Mapping[Identifier, NamedTypeSpec] + named_types_by_idx: Mapping[int, NamedTypeSpec] + hooks: Mapping[Identifier, FnAbi] + state_type: TypeSpec + + @staticmethod + def read_from(reader: BinaryReader | str): + ''' + Deserialize 'ContractAbi' from 'BinaryReader'. + ''' + + num_named_types = reader.readUInt32BigEndian() + named_types = [] + for type_index in range(0, num_named_types): + named_types.append(NamedTypeSpec.read_from(reader, type_index)) + + hooks = reader.readList(FnAbi.read_from) + state_type = TypeSpec.read_from(reader) + return ContractAbi(frozendict({t.name: t for t in named_types}), + frozendict({t.type_index: t for t in named_types}), + frozendict({t.name: t for t in hooks}), state_type) + + def read_type_element_from_rpc(self, type_name: Identifier, + rpc: BinaryReader): + ''' + Reads element of the given type name from RPC 'BinaryReader'. + ''' + return self.named_types_by_id[type_name].read_element_from( + rpc, self.named_types_by_idx, SerializeMode.RPC) + + def read_state(self, reader: BinaryReader): + if not isinstance(reader, BinaryReader): + reader = BinaryReader(reader) + + return self.state_type.read_element_from(reader, self.named_types_by_idx, + SerializeMode.STATE) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class FileAbi: + ''' + File definition. + ''' + + version_binder: Version + version_client: Version + contract: ContractAbi + + @staticmethod + def read_from(reader: BinaryReader): + ''' + Deserialize 'FileAbi' from 'BinaryReader'. + ''' + + if not isinstance(reader, BinaryReader): + reader = BinaryReader(reader) + + header = reader.readBytes(6) + assert header == b'PBCABI' + version_binder = Version.read_from(reader) + version_client = Version.read_from(reader) + + assert version_client >= Version(5, 0, 0), version_client + + contract = ContractAbi.read_from(reader) + return FileAbi(version_binder, version_client, contract) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..75be2f0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +enforce_typing