Pool certificates

.gitignore

@@ -6,4 +6,6 @@ dist
 
 # IDE
 .idea
-.code
+.code
+/integration-test/.env
+/integration-test/tmp_configs/*

pycardano/backend/cardano_cli.py

@@ -31,7 +31,7 @@
 from pycardano.hash import DatumHash, ScriptHash
 from pycardano.nativescript import NativeScript
 from pycardano.network import Network
-from pycardano.plutus import PlutusV1Script, PlutusV2Script, RawPlutusData, Datum
+from pycardano.plutus import Datum, PlutusV1Script, PlutusV2Script, RawPlutusData
 from pycardano.serialization import RawCBOR
 from pycardano.transaction import (
     Asset,

pycardano/certificate.py

@@ -1,17 +1,26 @@
+from __future__ import annotations
+
 from dataclasses import dataclass, field
-from typing import Optional, Union
+from typing import Optional, Tuple, Type, Union
 
+from pycardano.exception import DeserializeException
 from pycardano.hash import PoolKeyHash, ScriptHash, VerificationKeyHash
-from pycardano.serialization import ArrayCBORSerializable
+from pycardano.serialization import ArrayCBORSerializable, limit_primitive_type
 
 __all__ = [
     "Certificate",
     "StakeCredential",
     "StakeRegistration",
     "StakeDeregistration",
     "StakeDelegation",
+    "PoolRegistration",
+    "PoolRetirement",
 ]
 
+from pycardano.pool_params import PoolParams
+
+unit_interval = Tuple[int, int]
+
 
 @dataclass(repr=False)
 class StakeCredential(ArrayCBORSerializable):
@@ -25,20 +34,55 @@ def __post_init__(self):
         else:
             self._CODE = 1
 
+    @classmethod
+    @limit_primitive_type(list)
+    def from_primitive(
+        cls: Type[StakeCredential], values: Union[list, tuple]
+    ) -> StakeCredential:
+        if values[0] == 0:
+            return cls(VerificationKeyHash(values[1]))
+        elif values[0] == 1:
+            return cls(ScriptHash(values[1]))
+        else:
+            raise DeserializeException(f"Invalid StakeCredential type {values[0]}")
+
 
 @dataclass(repr=False)
 class StakeRegistration(ArrayCBORSerializable):
     _CODE: int = field(init=False, default=0)
 
     stake_credential: StakeCredential
 
+    def __post_init__(self):
+        self._CODE = 0
+
+    @classmethod
+    @limit_primitive_type(list)
+    def from_primitive(
+        cls: Type[StakeRegistration], values: Union[list, tuple]
+    ) -> StakeRegistration:
+        return cls(stake_credential=StakeCredential.from_primitive(values[1]))
+
 
 @dataclass(repr=False)
 class StakeDeregistration(ArrayCBORSerializable):
     _CODE: int = field(init=False, default=1)
 
     stake_credential: StakeCredential
 
+    def __post_init__(self):
+        self._CODE = 1
+
+    @classmethod
+    @limit_primitive_type(list)
+    def from_primitive(
+        cls: Type[StakeDeregistration], values: Union[list, tuple]
+    ) -> StakeDeregistration:
+        if values[0] == 1:
+            return cls(StakeCredential.from_primitive(values[1]))
+        else:
+            raise DeserializeException(f"Invalid StakeDeregistration type {values[0]}")
+
 
 @dataclass(repr=False)
 class StakeDelegation(ArrayCBORSerializable):
@@ -48,5 +92,83 @@ class StakeDelegation(ArrayCBORSerializable):
 
     pool_keyhash: PoolKeyHash
 
+    def __post_init__(self):
+        self._CODE = 2
+
+    @classmethod
+    @limit_primitive_type(list)
+    def from_primitive(
+        cls: Type[StakeDelegation], values: Union[list, tuple]
+    ) -> StakeDelegation:
+        if values[0] == 2:
+            return cls(
+                stake_credential=StakeCredential.from_primitive(values[1]),
+                pool_keyhash=PoolKeyHash.from_primitive(values[2]),
+            )
+        else:
+            raise DeserializeException(f"Invalid StakeDelegation type {values[0]}")
+
+
+@dataclass(repr=False)
+class PoolRegistration(ArrayCBORSerializable):
+    _CODE: int = field(init=False, default=3)
+
+    pool_params: PoolParams
+
+    def __post_init__(self):
+        self._CODE = 3
+
+    def to_primitive(self):
+        pool_params = self.pool_params.to_primitive()
+        if isinstance(pool_params, list):
+            return [self._CODE, *pool_params]
+        return super().to_primitive()
+
+    @classmethod
+    @limit_primitive_type(list)
+    def from_primitive(
+        cls: Type[PoolRegistration], values: Union[list, tuple]
+    ) -> PoolRegistration:
+        if values[0] == 3:
+            if isinstance(values[1], list):
+                return cls(
+                    pool_params=PoolParams.from_primitive(values[1]),
+                )
+            else:
+                return cls(
+                    pool_params=PoolParams.from_primitive(values[1:]),
+                )
+        else:
+            raise DeserializeException(f"Invalid PoolRegistration type {values[0]}")
+
+
+@dataclass(repr=False)
+class PoolRetirement(ArrayCBORSerializable):
+    _CODE: int = field(init=False, default=4)
+
+    pool_keyhash: PoolKeyHash
+    epoch: int
+
+    def __post_init__(self):
+        self._CODE = 4
+
+    @classmethod
+    @limit_primitive_type(list)
+    def from_primitive(
+        cls: Type[PoolRetirement], values: Union[list, tuple]
+    ) -> PoolRetirement:
+        if values[0] == 4:
+            return cls(
+                pool_keyhash=PoolKeyHash.from_primitive(values[1]), epoch=values[2]
+            )
+        else:
+            raise DeserializeException(f"Invalid PoolRetirement type {values[0]}")
+
 
-Certificate = Union[StakeRegistration, StakeDeregistration, StakeDelegation]
+Certificate = Union[
+    StakeRegistration,
+    StakeDeregistration,
+    StakeDelegation,
+    PoolRegistration,
+    PoolRetirement,
+]

pycardano/cip/cip14.py

@@ -2,6 +2,7 @@
 
 from nacl.encoding import RawEncoder
 from nacl.hash import blake2b
+
 from pycardano.crypto.bech32 import encode
 from pycardano.hash import ScriptHash
 from pycardano.transaction import AssetName

pycardano/hash.py

@@ -12,6 +12,9 @@
     "AUXILIARY_DATA_HASH_SIZE",
     "POOL_KEY_HASH_SIZE",
     "SCRIPT_DATA_HASH_SIZE",
+    "VRF_KEY_HASH_SIZE",
+    "POOL_METADATA_HASH_SIZE",
+    "REWARD_ACCOUNT_HASH_SIZE",
     "ConstrainedBytes",
     "VerificationKeyHash",
     "ScriptHash",
@@ -20,6 +23,9 @@
     "DatumHash",
     "AuxiliaryDataHash",
     "PoolKeyHash",
+    "PoolMetadataHash",
+    "VrfKeyHash",
+    "RewardAccountHash",
 ]
 
 VERIFICATION_KEY_HASH_SIZE = 28
@@ -29,6 +35,9 @@
 DATUM_HASH_SIZE = 32
 AUXILIARY_DATA_HASH_SIZE = 32
 POOL_KEY_HASH_SIZE = 28
+POOL_METADATA_HASH_SIZE = 32
+VRF_KEY_HASH_SIZE = 32
+REWARD_ACCOUNT_HASH_SIZE = 29
 
 
 T = TypeVar("T", bound="ConstrainedBytes")
@@ -124,7 +133,25 @@ class AuxiliaryDataHash(ConstrainedBytes):
     MAX_SIZE = MIN_SIZE = AUXILIARY_DATA_HASH_SIZE
 
 
-class PoolKeyHash(ConstrainedBytes):
+class PoolKeyHash(VerificationKeyHash):
     """Hash of a stake pool"""
 
     MAX_SIZE = MIN_SIZE = POOL_KEY_HASH_SIZE
+
+
+class PoolMetadataHash(ConstrainedBytes):
+    """Hash of a stake pool metadata"""
+
+    MAX_SIZE = MIN_SIZE = POOL_METADATA_HASH_SIZE
+
+
+class VrfKeyHash(ConstrainedBytes):
+    """Hash of a Cardano VRF key."""
+
+    MAX_SIZE = MIN_SIZE = VRF_KEY_HASH_SIZE
+
+
+class RewardAccountHash(ConstrainedBytes):
+    """Hash of a Cardano VRF key."""
+
+    MAX_SIZE = MIN_SIZE = REWARD_ACCOUNT_HASH_SIZE

pycardano/key.py

@@ -32,6 +32,9 @@
     "StakeSigningKey",
     "StakeVerificationKey",
     "StakeKeyPair",
+    "StakePoolSigningKey",
+    "StakePoolVerificationKey",
+    "StakePoolKeyPair",
 ]
 
 
@@ -314,3 +317,37 @@ def from_signing_key(
         cls: Type[StakeKeyPair], signing_key: SigningKey
     ) -> StakeKeyPair:
         return cls(signing_key, StakeVerificationKey.from_signing_key(signing_key))
+
+
+class StakePoolSigningKey(SigningKey):
+    KEY_TYPE = "StakePoolSigningKey_ed25519"
+    DESCRIPTION = "Stake Pool Operator Signing Key"
+
+
+class StakePoolVerificationKey(VerificationKey):
+    KEY_TYPE = "StakePoolVerificationKey_ed25519"
+    DESCRIPTION = "Stake Pool Operator Verification Key"
+
+
+class StakePoolKeyPair:
+    def __init__(self, signing_key: SigningKey, verification_key: VerificationKey):
+        self.signing_key = signing_key
+        self.verification_key = verification_key
+
+    @classmethod
+    def generate(cls: Type[StakePoolKeyPair]) -> StakePoolKeyPair:
+        signing_key = StakePoolSigningKey.generate()
+        return cls.from_signing_key(signing_key)
+
+    @classmethod
+    def from_signing_key(
+        cls: Type[StakePoolKeyPair], signing_key: SigningKey
+    ) -> StakePoolKeyPair:
+        return cls(signing_key, StakePoolVerificationKey.from_signing_key(signing_key))
+
+    def __eq__(self, other):
+        if isinstance(other, StakePoolKeyPair):
+            return (
+                other.signing_key == self.signing_key
+                and other.verification_key == self.verification_key
+            )