feat: implement block decoding and refactor encoding layer

pactus/block/__init__.py

@@ -0,0 +1,5 @@
+from .block import Block
+from .certificate import Certificate
+from .header import Header
+
+__all__ = ["Block", "Certificate", "Header"]

pactus/block/block.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from pactus.encoding import encoding
+from pactus.transaction import Transaction
+
+from .certificate import Certificate
+from .header import Header
+
+
+class Block:
+    def __init__(
+        self,
+        header: Header,
+        prev_cert: Certificate | None,
+        transactions: list,
+    ) -> None:
+        self.header = header
+        self.prev_cert = prev_cert
+        self.transactions = transactions
+
+    @classmethod
+    def decode(cls, data: bytes) -> Block:
+        """Decode a Block from raw bytes."""
+        header, data = Header.decode(data)
+
+        # Genesis block has no certificate
+        is_genesis = header.prev_block_hash.is_undef()
+
+        prev_cert = None
+        if not is_genesis:
+            prev_cert, data = Certificate.decode(data)
+
+        num_txs, data = encoding.read_var_int(data)
+
+        transactions = []
+        for _ in range(num_txs):
+            tx, data = Transaction.decode(data)
+            transactions.append(tx)
+
+        return cls(header, prev_cert, transactions)

pactus/block/certificate.py

@@ -0,0 +1,46 @@
+from pactus.crypto.bls.signature import Signature
+from pactus.encoding import encoding
+from pactus.types.height import Height
+from pactus.types.round import Round
+
+
+class Certificate:
+    def __init__(
+        self,
+        height: Height,
+        round_: Round,
+        committers: list,
+        absentees: list,
+        signature: Signature,
+    ) -> None:
+        self.height = height
+        self.round = round_
+        self.committers = committers
+        self.absentees = absentees
+        self.signature = signature
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        """
+        Decode a Certificate from bytes.
+        Returns (Certificate, remaining_buf).
+        """
+        height, buf = Height.decode(buf)
+        round_, buf = Round.decode(buf)
+
+        num_committers, buf = encoding.read_var_int(buf)
+        committers = []
+        for _ in range(num_committers):
+            n, buf = encoding.read_var_int(buf)
+            committers.append(n)
+
+        num_absentees, buf = encoding.read_var_int(buf)
+        absentees = []
+        for _ in range(num_absentees):
+            n, buf = encoding.read_var_int(buf)
+            absentees.append(n)
+
+        signature, buf = Signature.decode(buf)
+
+        cert = cls(height, round_, committers, absentees, signature)
+        return cert, buf

pactus/block/header.py

@@ -0,0 +1,44 @@
+from pactus.crypto.address import Address
+from pactus.crypto.hash import Hash
+from pactus.encoding import encoding
+
+
+class Header:
+    def __init__(
+        self,
+        version: int,
+        unix_time: int,
+        prev_block_hash: Hash,
+        state_root: Hash,
+        sortition_seed: bytes,
+        proposer_address: Address,
+    ) -> None:
+        self.version = version
+        self.unix_time = unix_time
+        self.prev_block_hash = prev_block_hash
+        self.state_root = state_root
+        self.sortition_seed = sortition_seed
+        self.proposer_address = proposer_address
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        """
+        Decode a Header from bytes.
+        Returns (Header, remaining_buf).
+        """
+        version, buf = encoding.read_uint8(buf)
+        unix_time, buf = encoding.read_uint32(buf)
+        prev_block_hash, buf = Hash.decode(buf)
+        state_root, buf = Hash.decode(buf)
+        sortition_seed, buf = encoding.read_fixed_bytes(buf, 48)
+        proposer_address, buf = Address.decode(buf)
+
+        header = cls(
+            version,
+            unix_time,
+            prev_block_hash,
+            state_root,
+            sortition_seed,
+            proposer_address,
+        )
+        return header, buf

pactus/crypto/__init__.py

@@ -1,6 +1,7 @@
 from .address import Address, AddressType
+from .hash import Hash
 from .hrp import HRP
 from .private_key import PrivateKey
 from .public_key import PublicKey
 
-__all__ = ["HRP", "Address", "AddressType", "PrivateKey", "PublicKey"]
+__all__ = ["HRP", "Address", "AddressType", "Hash", "PrivateKey", "PublicKey"]

pactus/crypto/address.py

@@ -3,6 +3,7 @@
 from enum import Enum
 
 from pactus.crypto.hrp import HRP
+from pactus.encoding import encoding
 from pactus.utils import utils
 
 # Address format: hrp + `1` + type + data + checksum
@@ -22,7 +23,7 @@ class AddressType(Enum):
 class Address:
     def __init__(self, address_type: AddressType, data: bytes) -> None:
         if len(data) != ADDRESS_SIZE - 1:
-            msg = "Data must be 21 bytes long"
+            msg = "Data must be 20 bytes long"
             raise ValueError(msg)
 
         self.data = bytearray()
@@ -32,7 +33,7 @@ def __init__(self, address_type: AddressType, data: bytes) -> None:
     @classmethod
     def from_string(cls, text: str) -> Address:
         if text == TREASURY_ADDRESS_STRING:
-            return bytes([0])
+            return cls(AddressType.TREASURY, bytes(20))
 
         hrp, typ, data = utils.decode_to_base256_with_type(text)
         if hrp != HRP.ADDRESS_HRP:
@@ -59,7 +60,7 @@ def raw_bytes(self) -> bytes:
         return bytes(self.data)
 
     def string(self) -> str:
-        if self.data == bytes([0]):
+        if self.is_treasury_address():
             return TREASURY_ADDRESS_STRING
 
         return utils.encode_from_base256_with_type(
@@ -76,7 +77,30 @@ def is_treasury_address(self) -> bool:
 
     def is_account_address(self) -> bool:
         t = self.address_type()
-        return t in (AddressType.TREASURY, AddressType.BLS_ACCOUNT, AddressType.ED25519_ACCOUNT)
+        return t in (
+            AddressType.TREASURY,
+            AddressType.BLS_ACCOUNT,
+            AddressType.ED25519_ACCOUNT,
+        )
 
     def is_validator_address(self) -> bool:
         return self.address_type() == AddressType.VALIDATOR
+
+    def encode(self) -> bytes:
+        buf = b""
+        if self.is_treasury_address():
+            return encoding.append_uint8(buf, AddressType.TREASURY.value)
+        return encoding.append_fixed_bytes(buf, self.raw_bytes())
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        """
+        Decode an Address from bytes.
+        Returns (Address, remaining_buf).
+        """
+        addr_type, buf = encoding.read_uint8(buf)
+        if addr_type == AddressType.TREASURY.value:
+            return cls(AddressType.TREASURY, bytes(20)), buf
+
+        data, buf = encoding.read_fixed_bytes(buf, 20)
+        return cls(AddressType(addr_type), data), buf

pactus/crypto/bls/public_key.py

@@ -6,6 +6,7 @@
 
 from pactus.crypto.address import Address, AddressType
 from pactus.crypto.hrp import HRP
+from pactus.encoding import encoding
 from pactus.utils import utils
 
 from .bls12_381.bls_sig_g1 import aggregate_pubs, verify
@@ -60,6 +61,14 @@ def string(self) -> str:
             self.raw_bytes(),
         )
 
+    def encode(self) -> bytes:
+        return encoding.append_fixed_bytes(b"", self.raw_bytes())
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        data, buf = encoding.read_fixed_bytes(buf, PUBLIC_KEY_SIZE)
+        return cls.from_bytes(data), buf
+
     def account_address(self) -> Address:
         return self._make_address(AddressType.BLS_ACCOUNT)
 

pactus/crypto/bls/signature.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pactus.encoding import encoding
+
 from .bls12_381.bls_sig_g1 import aggregate_sigs
 from .bls12_381.serdesZ import deserialize, serialize
 
@@ -36,3 +38,12 @@ def raw_bytes(self) -> bytes:
 
     def string(self) -> str:
         return self.raw_bytes().hex()
+
+    def encode(self) -> bytes:
+        return encoding.append_fixed_bytes(b"", self.raw_bytes())
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        data, buf = encoding.read_fixed_bytes(buf, SIGNATURE_SIZE)
+        point_g1 = deserialize(data, is_ell2=False)
+        return cls(point_g1), buf

pactus/crypto/ed25519/public_key.py

@@ -8,6 +8,7 @@
 
 from pactus.crypto.address import Address, AddressType
 from pactus.crypto.hrp import HRP
+from pactus.encoding import encoding
 from pactus.utils import utils
 
 from .signature import SIGNATURE_TYPE_ED25519, Signature
@@ -49,6 +50,14 @@ def string(self) -> str:
             self.raw_bytes(),
         )
 
+    def encode(self) -> bytes:
+        return encoding.append_fixed_bytes(b"", self.raw_bytes())
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        data, buf = encoding.read_fixed_bytes(buf, PUBLIC_KEY_SIZE)
+        return cls(ed25519.Ed25519PublicKey.from_public_bytes(data)), buf
+
     def account_address(self) -> Address:
         blake2b = hashlib.blake2b(digest_size=32)
         blake2b.update(self.raw_bytes())

pactus/crypto/ed25519/signature.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pactus.encoding import encoding
+
 SIGNATURE_SIZE = 64
 SIGNATURE_TYPE_ED25519 = 3
 
@@ -23,3 +25,11 @@ def raw_bytes(self) -> bytes:
 
     def string(self) -> str:
         return self.sig.hex()
+
+    def encode(self) -> bytes:
+        return encoding.append_fixed_bytes(b"", self.sig)
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        data, buf = encoding.read_fixed_bytes(buf, SIGNATURE_SIZE)
+        return cls(data), buf

pactus/crypto/hash.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from pactus.encoding import encoding
+
+HASH_SIZE = 32
+
+
+class Hash:
+    """Hash represents a 32-byte hash in the Pactus blockchain."""
+
+    def __init__(self, data: bytes) -> None:
+        if len(data) != HASH_SIZE:
+            msg = f"Hash must be {HASH_SIZE} bytes, got {len(data)}"
+            raise ValueError(msg)
+        self.data = data
+
+    def __eq__(self, other: Hash) -> bool:
+        if isinstance(other, Hash):
+            return self.data == other.data
+        return False
+
+    def __hash__(self) -> int:
+        return hash(self.data)
+
+    def __str__(self) -> str:
+        return self.data.hex()
+
+    def is_undef(self) -> bool:
+        return self.data == bytes(HASH_SIZE)
+
+    def encode(self) -> bytes:
+        return encoding.append_fixed_bytes(b"", self.data)
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple[Hash, bytes]:
+        """
+        Decode a Hash from bytes.
+        Returns (Hash, remaining_buf).
+        """
+        data, buf = encoding.read_fixed_bytes(buf, HASH_SIZE)
+        return cls(data), buf

pactus/crypto/public_key.py

@@ -21,6 +21,15 @@ def raw_bytes(self) -> bytes:
     def string(self) -> str:
         pass
 
+    @abstractmethod
+    def encode(self) -> bytes:
+        pass
+
+    @classmethod
+    @abstractmethod
+    def decode(cls, buf: bytes) -> tuple:
+        pass
+
     @abstractmethod
     def verify(self, msg: str, sig: Signature) -> bool:
         pass

pactus/crypto/secp256k1/public_key.py

@@ -8,6 +8,7 @@
 
 from pactus.crypto.address import Address, AddressType
 from pactus.crypto.hrp import HRP
+from pactus.encoding import encoding
 from pactus.utils import utils
 
 from .signature import SIGNATURE_TYPE_SECP256K1, Signature
@@ -49,6 +50,14 @@ def string(self) -> str:
             self.raw_bytes(),
         )
 
+    def encode(self) -> bytes:
+        return encoding.append_fixed_bytes(b"", self.raw_bytes())
+
+    @classmethod
+    def decode(cls, buf: bytes) -> tuple:
+        data, buf = encoding.read_fixed_bytes(buf, PUBLIC_KEY_SIZE)
+        return cls(secp256k1.PublicKey(data, raw=True)), buf
+
     def account_address(self) -> Address:
         blake2b = hashlib.blake2b(digest_size=32)
         blake2b.update(self.raw_bytes())