test

tck/handlers/__init__.py

@@ -5,5 +5,7 @@
 
 # Import all handler modules to trigger @register_handler decorators
 from . import sdk      # setup, reset
+from . import consensus  # createTopic
+from . import key      # generateKey
 
 __all__ = ["get_handler", "get_all_handlers", "safe_dispatch", "validate_request_params"]

tck/handlers/consensus.py

@@ -0,0 +1,163 @@
+"""TCK handlers for Consensus Service operations (Topics)."""
+
+from typing import Any, Dict, Optional, List
+from hiero_sdk_python import (
+    Client, PrivateKey, PublicKey, TopicCreateTransaction, Duration,
+    AccountId, ResponseCode, TopicId
+)
+from hiero_sdk_python.utils.key_utils import Key
+from .registry import register_handler, validate_request_params
+from ..client_manager import get_client
+from ..errors import JsonRpcError, INTERNAL_ERROR, INVALID_PARAMS, HIERO_ERROR
+
+
+def _parse_key(key_string: Optional[str]) -> Optional[Key]:
+    """Parse a key from a hex string.
+
+    Args:
+        key_string: Hex-encoded key string
+
+    Returns:
+        PublicKey or PrivateKey instance, or None if key_string is None
+
+    Raises:
+        JsonRpcError: If the key cannot be parsed
+    """
+    if key_string is None:
+        return None
+
+    try:
+        # Try to parse as a private key first
+        try:
+            return PrivateKey.from_string(key_string)
+        except Exception:
+            # If it fails, try as a public key
+            return PublicKey.from_string(key_string)
+    except Exception as e:
+        raise JsonRpcError(INTERNAL_ERROR, f"Failed to parse key: {str(e)}")
+
+
+def _get_signers(signers_list: Optional[List[str]]) -> List[PrivateKey]:
+    """Parse and return list of private keys for signing.
+
+    Args:
+        signers_list: List of hex-encoded private key strings
+
+    Returns:
+        List of PrivateKey instances
+    """
+    if not signers_list:
+        return []
+
+    keys = []
+    for signer_str in signers_list:
+        try:
+            key = PrivateKey.from_string(signer_str)
+            keys.append(key)
+        except Exception as e:
+            raise JsonRpcError(INTERNAL_ERROR, f"Failed to parse signer key: {str(e)}")
+    return keys
+
+
+@register_handler("createTopic")
+def create_topic_handler(params: Dict[str, Any], session_id: Optional[str] = None) -> Dict[str, Any]:
+    """Create a new topic using TopicCreateTransaction.
+
+    Params:
+        memo (str, optional): Topic memo
+        adminKey (str, optional): Admin key (hex-encoded)
+        submitKey (str, optional): Submit key (hex-encoded)
+        autoRenewPeriod (str, optional): Auto-renew period in seconds
+        autoRenewAccountId (str, optional): Account ID for auto-renewal
+        commonTransactionParams (dict, optional): Common params including signers
+
+    Returns:
+        Dict with status and topicId
+    """
+    # Get client from session
+    client = get_client(session_id or "default")
+    if client is None:
+        raise JsonRpcError(INTERNAL_ERROR, "No client initialized for this session")
+
+    try:
+        # Create the transaction
+        tx = TopicCreateTransaction()
+
+        # Set memo if provided
+        if "memo" in params and params["memo"] is not None:
+            memo = params["memo"]
+            # Validate memo length (max 100 bytes)
+            if len(memo.encode('utf-8')) > 100:
+                raise JsonRpcError(HIERO_ERROR, "Hiero error", {"status": "MEMO_TOO_LONG"})
+            # Check for null bytes
+            if '\0' in memo:
+                raise JsonRpcError(HIERO_ERROR, "Hiero error", {"status": "INVALID_ZERO_BYTE_IN_STRING"})
+            tx.set_memo(memo)
+
+        # Set admin key if provided
+        if "adminKey" in params and params["adminKey"] is not None:
+            admin_key = _parse_key(params["adminKey"])
+            if admin_key:
+                tx.set_admin_key(admin_key)
+
+        # Set submit key if provided
+        if "submitKey" in params and params["submitKey"] is not None:
+            submit_key = _parse_key(params["submitKey"])
+            if submit_key:
+                tx.set_submit_key(submit_key)
+
+        # Set auto-renew period if provided
+        if "autoRenewPeriod" in params and params["autoRenewPeriod"] is not None:
+            try:
+                period = int(params["autoRenewPeriod"])
+                # Validate auto-renew period (approximate range: 6999999 to 8000001)
+                if period <= 0 or period < 6999999 or period > 8000001:
+                    raise JsonRpcError(HIERO_ERROR, "Hiero error", {"status": "AUTORENEW_DURATION_NOT_IN_RANGE"})
+                tx.set_auto_renew_period(period)
+            except ValueError:
+                raise JsonRpcError(INVALID_PARAMS, "Invalid params", "autoRenewPeriod must be a string representing an integer")
+
+        # Set auto-renew account if provided
+        if "autoRenewAccountId" in params and params["autoRenewAccountId"] is not None:
+            try:
+                auto_renew_account = AccountId.from_string(params["autoRenewAccountId"])
+                tx.set_auto_renew_account(auto_renew_account)
+            except Exception as e:
+                raise JsonRpcError(INVALID_PARAMS, "Invalid params", f"Invalid autoRenewAccountId: {str(e)}")
+
+        # Freeze the transaction with the client
+        tx.freeze_with(client)
+
+        # Get signers from commonTransactionParams if provided
+        signers = []
+        if "commonTransactionParams" in params:
+            common_params = params["commonTransactionParams"]
+            if "signers" in common_params and isinstance(common_params["signers"], list):
+                signers = _get_signers(common_params["signers"])
+
+        # Sign the transaction with any additional signers
+        for signer in signers:
+            tx.sign(signer)
+
+        # Execute the transaction
+        receipt = tx.execute(client)
+
+        # Check the receipt status
+        if receipt.status != ResponseCode.SUCCESS:
+            status_name = ResponseCode(receipt.status).name
+            raise JsonRpcError(HIERO_ERROR, "Hiero error", {"status": status_name})
+
+        # Extract the topic ID
+        topic_id = receipt.topic_id
+        if not topic_id:
+            raise JsonRpcError(INTERNAL_ERROR, "Transaction succeeded but no topic ID returned")
+
+        return {
+            "status": ResponseCode(receipt.status).name,
+            "topicId": str(topic_id)
+        }
+
+    except JsonRpcError:
+        raise
+    except Exception as e:
+        raise JsonRpcError(HIERO_ERROR, "Hiero error", str(e))

tck/handlers/key.py

@@ -0,0 +1,185 @@
+"""TCK handlers for key generation and manipulation."""
+
+from typing import Any, Dict, Optional, List
+from hiero_sdk_python import PrivateKey, PublicKey
+from hiero_sdk_python.hapi.services import basic_types_pb2
+from hiero_sdk_python.utils.key_utils import key_to_proto
+from .registry import register_handler
+from ..errors import JsonRpcError, INVALID_PARAMS, INTERNAL_ERROR
+
+
+def _key_to_hex(key: Any) -> str:
+    """Convert a key to its DER-encoded hex representation.
+
+    Args:
+        key: PrivateKey or PublicKey instance
+
+    Returns:
+        Hex string representation of the key
+    """
+    proto_key = key_to_proto(key)
+    if proto_key is None:
+        raise JsonRpcError(INTERNAL_ERROR, "Failed to convert key to protobuf")
+
+    # Serialize to bytes and convert to hex
+    key_bytes = proto_key.SerializeToString()
+    return key_bytes.hex().upper()
+
+
+@register_handler("generateKey")
+def generate_key_handler(params: Dict[str, Any], session_id: Optional[str] = None) -> Dict[str, Any]:
+    """Generate cryptographic keys for testing.
+
+    Params:
+        type (str, required): Type of key to generate
+            - ed25519PrivateKey: Generate a new Ed25519 private key
+            - ecdsaSecp256k1PrivateKey: Generate a new ECDSA (secp256k1) private key
+            - ed25519PublicKey: Get public key from private key or generate public key
+            - ecdsaSecp256k1PublicKey: Get public key from private key or generate public key
+            - keyList: Generate a keyList with multiple keys
+            - thresholdKey: Generate a thresholdKey
+        fromKey (str, optional): Private key to derive public key from (hex string)
+        threshold (int, optional): Threshold for thresholdKey type
+        keys (list, optional): List of key specs for keyList/thresholdKey
+
+    Returns:
+        Dict with key information and privateKeys list
+    """
+    try:
+        key_type = params.get("type")
+        if not key_type:
+            raise JsonRpcError(INVALID_PARAMS, "Missing required parameter: type")
+
+        if key_type == "ed25519PrivateKey":
+            # Generate new Ed25519 private key
+            private_key = PrivateKey.generate_ed25519()
+            return {
+                "key": _key_to_hex(private_key),
+                "privateKeys": [_key_to_hex(private_key)]
+            }
+
+        elif key_type == "ecdsaSecp256k1PrivateKey":
+            # Generate new ECDSA (secp256k1) private key
+            private_key = PrivateKey.generate_ecdsa_secp256k1()
+            return {
+                "key": _key_to_hex(private_key),
+                "privateKeys": [_key_to_hex(private_key)]
+            }
+
+        elif key_type == "ed25519PublicKey":
+            # Get or generate Ed25519 public key
+            if "fromKey" in params and params["fromKey"]:
+                # Derive public key from private key
+                private_key = PrivateKey.from_string(params["fromKey"])
+                public_key = private_key.public_key()
+            else:
+                # Generate new key pair and return public key
+                private_key = PrivateKey.generate_ed25519()
+                public_key = private_key.public_key()
+
+            return {
+                "key": _key_to_hex(public_key)
+            }
+
+        elif key_type == "ecdsaSecp256k1PublicKey":
+            # Get or generate ECDSA (secp256k1) public key
+            if "fromKey" in params and params["fromKey"]:
+                # Derive public key from private key
+                private_key = PrivateKey.from_string(params["fromKey"])
+                public_key = private_key.public_key()
+            else:
+                # Generate new key pair and return public key
+                private_key = PrivateKey.generate_ecdsa_secp256k1()
+                public_key = private_key.public_key()
+
+            return {
+                "key": _key_to_hex(public_key)
+            }
+
+        elif key_type == "keyList":
+            # Generate a KeyList with multiple keys
+            keys_spec = params.get("keys", [])
+            if not isinstance(keys_spec, list):
+                raise JsonRpcError(INVALID_PARAMS, "keys parameter must be a list")
+
+            key_list_proto = basic_types_pb2.KeyList()
+            private_keys_hex = []
+
+            for key_spec in keys_spec:
+                if not isinstance(key_spec, dict):
+                    raise JsonRpcError(INVALID_PARAMS, "Each key spec must be a dict")
+
+                spec_type = key_spec.get("type")
+                if spec_type == "ed25519PrivateKey":
+                    private_key = PrivateKey.generate_ed25519()
+                    private_keys_hex.append(_key_to_hex(private_key))
+                    proto_key = key_to_proto(private_key)
+                elif spec_type == "ecdsaSecp256k1PrivateKey":
+                    private_key = PrivateKey.generate_ecdsa_secp256k1()
+                    private_keys_hex.append(_key_to_hex(private_key))
+                    proto_key = key_to_proto(private_key)
+                else:
+                    raise JsonRpcError(INVALID_PARAMS, f"Unsupported key type in keyList: {spec_type}")
+
+                if proto_key:
+                    key_list_proto.keys.append(proto_key)
+
+            # Serialize KeyList to bytes and convert to hex
+            key_list_bytes = key_list_proto.SerializeToString()
+            key_list_hex = key_list_bytes.hex().upper()
+
+            return {
+                "key": key_list_hex,
+                "privateKeys": private_keys_hex
+            }
+
+        elif key_type == "thresholdKey":
+            # Generate a ThresholdKey with threshold and keys
+            threshold = params.get("threshold")
+            if threshold is None:
+                raise JsonRpcError(INVALID_PARAMS, "threshold parameter required for thresholdKey")
+
+            keys_spec = params.get("keys", [])
+            if not isinstance(keys_spec, list):
+                raise JsonRpcError(INVALID_PARAMS, "keys parameter must be a list")
+
+            threshold_key_proto = basic_types_pb2.ThresholdKey()
+            threshold_key_proto.threshold = int(threshold)
+
+            private_keys_hex = []
+
+            for key_spec in keys_spec:
+                if not isinstance(key_spec, dict):
+                    raise JsonRpcError(INVALID_PARAMS, "Each key spec must be a dict")
+
+                spec_type = key_spec.get("type")
+                if spec_type == "ed25519PrivateKey":
+                    private_key = PrivateKey.generate_ed25519()
+                    private_keys_hex.append(_key_to_hex(private_key))
+                    proto_key = key_to_proto(private_key)
+                elif spec_type == "ecdsaSecp256k1PrivateKey":
+                    private_key = PrivateKey.generate_ecdsa_secp256k1()
+                    private_keys_hex.append(_key_to_hex(private_key))
+                    proto_key = key_to_proto(private_key)
+                else:
+                    raise JsonRpcError(INVALID_PARAMS, f"Unsupported key type in thresholdKey: {spec_type}")
+
+                if proto_key:
+                    threshold_key_proto.keys.keys.append(proto_key)
+
+            # Serialize ThresholdKey to bytes and convert to hex
+            threshold_key_bytes = threshold_key_proto.SerializeToString()
+            threshold_key_hex = threshold_key_bytes.hex().upper()
+
+            return {
+                "key": threshold_key_hex,
+                "privateKeys": private_keys_hex
+            }
+
+        else:
+            raise JsonRpcError(INVALID_PARAMS, f"Unsupported key type: {key_type}")
+
+    except JsonRpcError:
+        raise
+    except Exception as e:
+        raise JsonRpcError(INTERNAL_ERROR, f"Failed to generate key: {str(e)}")