diff --git a/src/hiero_sdk_python/crypto/public_key.py b/src/hiero_sdk_python/crypto/public_key.py index 18269f379..f5f97c8f6 100644 --- a/src/hiero_sdk_python/crypto/public_key.py +++ b/src/hiero_sdk_python/crypto/public_key.py @@ -413,6 +413,90 @@ def to_bytes_der(self) -> bytes: encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo ) + @classmethod + def _encode_vlq(cls, value: int) -> bytes: + """ + Encode a value in Variable-Length Quantity (VLQ) format. + """ + if value < 0: + raise ValueError("VLQ value must be non-negative") + if value == 0: + return b"\x00" + + encoded = bytearray() + while value > 0: + encoded.insert(0, value & 0x7F) + value >>= 7 + + # Set the high bit (0x80) on all bytes except the last + for i in range(len(encoded) - 1): + encoded[i] |= 0x80 + + return bytes(encoded) + + @classmethod + def _encode_der_length(cls, length: int) -> bytes: + """Encode a DER length field per X.690 standards.""" + if length < 0: + raise ValueError("DER length must be non-negative") + if length < 0x80: + return bytes([length]) + + length_bytes = length.to_bytes((length.bit_length() + 7) // 8, "big") + return bytes([0x80 | len(length_bytes)]) + length_bytes + + @classmethod + def _encode_der_oid(cls, oid: str) -> bytes: + """ + Encode a dotted OID string into DER OID bytes including tag and length. + """ + parts = [int(part) for part in oid.split(".")] + + is_valid = len(parts) >= 2 and parts[0] in (0, 1, 2) and parts[1] >= 0 and (parts[0] == 2 or parts[1] < 40) + if not is_valid: + raise ValueError(f"Invalid OID structure for '{oid}'") + + first, second = parts[0], parts[1] + + # Encode the combined root using VLQ to handle edge cases correctly + encoded = bytearray(cls._encode_vlq(40 * first + second)) + + for value in parts[2:]: + encoded.extend(cls._encode_vlq(value)) + + return bytes([0x06]) + cls._encode_der_length(len(encoded)) + bytes(encoded) + + @classmethod + def _encode_der_sequence(cls, content: bytes) -> bytes: + """Encode DER SEQUENCE tag, length, and content per X.690 standards.""" + return bytes([0x30]) + cls._encode_der_length(len(content)) + content + + @classmethod + def _encode_der_bit_string(cls, content: bytes) -> bytes: + """ + Encode DER BIT STRING tag, length, and content per X.690 standards. + """ + payload = bytes([0x00]) + content + return bytes([0x03]) + cls._encode_der_length(len(payload)) + payload + + def to_bytes_der_ecdsa_compressed(self) -> bytes: + """ + Returns DER-encoded SubjectPublicKeyInfo for secp256k1 using a compressed SEC1 point. + Raises: + ValueError: If this key is not ECDSA secp256k1. + """ + if not self.is_ecdsa(): + raise ValueError("Compressed ECDSA DER export is only supported for ECDSA keys") + + # id-ecPublicKey + secp256k1 OID algorithm identifier + algorithm_id = self._encode_der_sequence( + self._encode_der_oid("1.2.840.10045.2.1") + self._encode_der_oid("1.3.132.0.10") + ) + compressed_point = self.to_bytes_ecdsa(compressed=True) + subject_public_key = self._encode_der_bit_string(compressed_point) + + return self._encode_der_sequence(algorithm_id + subject_public_key) + # # --------------------------------- # Type-specific (Ed25519, ECDSA secp256k1) to hex string. @@ -440,6 +524,12 @@ def to_string_der(self) -> str: """ return self.to_bytes_der().hex() + def to_string_der_ecdsa_compressed(self) -> str: + """ + Returns DER SPKI hex for ECDSA secp256k1 using a compressed SEC1 point. + """ + return self.to_bytes_der_ecdsa_compressed().hex() + def to_string_raw(self) -> str: """ Catch all ed25519 or ecdsa for convenience. diff --git a/src/hiero_sdk_python/utils/key_utils.py b/src/hiero_sdk_python/utils/key_utils.py index 757a8aa53..6a4549b0e 100644 --- a/src/hiero_sdk_python/utils/key_utils.py +++ b/src/hiero_sdk_python/utils/key_utils.py @@ -7,43 +7,30 @@ from __future__ import annotations -from hiero_sdk_python.crypto.private_key import PrivateKey -from hiero_sdk_python.crypto.public_key import PublicKey +from hiero_sdk_python.crypto.key import Key from hiero_sdk_python.hapi.services import basic_types_pb2 -# Type alias for keys that can be either PrivateKey or PublicKey -Key = PrivateKey | PublicKey - - def key_to_proto(key: Key | None) -> basic_types_pb2.Key | None: """ - Helper function to convert a key (PrivateKey or PublicKey) to protobuf Key format. + Helper function to convert an SDK key to protobuf Key format. - This function handles the conversion of SDK key types to protobuf format: - - If a PrivateKey is provided, its corresponding public key is extracted and converted. - - If a PublicKey is provided, it is converted directly to protobuf. - - If None is provided, None is returned. + This function handles any concrete subclass of Key by delegating to its + to_proto_key() implementation. If None is provided, None is returned. Args: - key (Optional[Key]): The key to convert (PrivateKey or PublicKey), or None + key (Optional[Key]): The key to convert, or None Returns: basic_types_pb2.Key (Optional): The protobuf key or None if key is None Raises: - TypeError: If the provided key is not a PrivateKey, PublicKey, or None. + TypeError: If the provided key is not a Key instance or None. """ - if not key: + if key is None: return None - # If it's a PrivateKey, get the public key first, then convert to proto - if isinstance(key, PrivateKey): - return key.public_key()._to_proto() - - # If it's a PublicKey, convert directly to proto - if isinstance(key, PublicKey): - return key._to_proto() + if isinstance(key, Key): + return key.to_proto_key() - # Safety net: This will fail if a non-key is passed - raise TypeError("Key must be of type PrivateKey or PublicKey") + raise TypeError("Key must be of type PrivateKey or PublicKey, or another SDK Key implementation") diff --git a/tests/unit/keys_public_test.py b/tests/unit/keys_public_test.py index 4dfd8a9ce..7a03c82b9 100644 --- a/tests/unit/keys_public_test.py +++ b/tests/unit/keys_public_test.py @@ -328,6 +328,91 @@ def test_from_bytes_invalid(): PublicKey.from_bytes(data) +# ------------------------------------------------------------------------------ +# Test: DER helper encoders and compressed DER export +# ------------------------------------------------------------------------------ +def test_encode_der_length_short_and_long_forms(): + assert PublicKey._encode_der_length(0) == b"\x00" + assert PublicKey._encode_der_length(0x7F) == b"\x7f" + assert PublicKey._encode_der_length(0x80) == b"\x81\x80" + assert PublicKey._encode_der_length(0x0100) == b"\x82\x01\x00" + assert PublicKey._encode_der_length(0x1000000) == b"\x84\x01\x00\x00\x00" + + +def test_encode_der_length_negative_raises(): + with pytest.raises(ValueError, match="non-negative"): + PublicKey._encode_der_length(-1) + + +def test_encode_der_oid_known_values(): + # id-ecPublicKey + assert PublicKey._encode_der_oid("1.2.840.10045.2.1") == bytes.fromhex("06072a8648ce3d0201") + # secp256k1 + assert PublicKey._encode_der_oid("1.3.132.0.10") == bytes.fromhex("06052b8104000a") + + +def test_encode_der_oid_combined_root_multibyte(): + # "2.999" -> 2*40 + 999 = 1079, encoded as VLQ: 0x88 0x37 + result = PublicKey._encode_der_oid("2.999.1") + assert result == bytes.fromhex("0603883701") + assert result[0] == 0x06 # OID tag + assert result[1] == 0x03 # Length of OID content + + +def test_encode_der_oid_invalid_components_raise(): + for oid in ("1", "3.1.1", "1.40.1", "9.999.1"): + with pytest.raises(ValueError, match=f"Invalid OID structure for '{oid}'"): + PublicKey._encode_der_oid(oid) + + with pytest.raises(ValueError, match="non-negative"): + PublicKey._encode_der_oid("1.2.-1") + + with pytest.raises(ValueError, match="invalid literal for int()"): + PublicKey._encode_der_oid("1.999bit") + with pytest.raises(ValueError): + PublicKey._encode_der_oid("") + with pytest.raises(ValueError): + PublicKey._encode_der_oid("...") + + +def test_encode_der_sequence_and_bit_string(): + assert PublicKey._encode_der_sequence(b"\x01\x02") == b"\x30\x02\x01\x02" + assert PublicKey._encode_der_bit_string(b"\xaa\xbb") == b"\x03\x03\x00\xaa\xbb" + + +def test_to_bytes_der_ecdsa_compressed_structure_and_roundtrip(ecdsa_keypair): + _, pub = ecdsa_keypair + public_key = PublicKey(pub) + + der = public_key.to_bytes_der_ecdsa_compressed() + compressed_point = public_key.to_bytes_ecdsa(compressed=True) + + # Fixed SPKI prefix for secp256k1 compressed-point encoding. + expected_prefix = bytes.fromhex("3036301006072a8648ce3d020106052b8104000a032200") + assert der.startswith(expected_prefix) + assert der[len(expected_prefix) :] == compressed_point + + # Ensure produced DER is parseable and preserves the same public key bytes. + loaded = PublicKey.from_der(der) + assert loaded.is_ecdsa() + assert loaded.to_bytes_ecdsa(compressed=True) == compressed_point + + +def test_to_bytes_der_ecdsa_compressed_rejects_ed25519(ed25519_keypair): + _, pub = ed25519_keypair + public_key = PublicKey(pub) + + with pytest.raises(ValueError, match="only supported for ECDSA"): + public_key.to_bytes_der_ecdsa_compressed() + + +def test_encode_vlq_values(): + assert PublicKey._encode_vlq(0) == b"\x00" + assert PublicKey._encode_vlq(127) == b"\x7f" + assert PublicKey._encode_vlq(128) == b"\x81\x00" + assert PublicKey._encode_vlq(0x4000) == b"\x81\x80\x00" + + # ------------------------------------------------------------------------------ # Test: from_string_xxx # ------------------------------------------------------------------------------ diff --git a/tests/unit/token_create_transaction_test.py b/tests/unit/token_create_transaction_test.py index fba6d30a1..c4f5ea73a 100644 --- a/tests/unit/token_create_transaction_test.py +++ b/tests/unit/token_create_transaction_test.py @@ -64,6 +64,16 @@ def generate_transaction_id(account_id_proto): return TransactionId(valid_start=tx_timestamp, account_id=account_id_proto) +def _mock_private_key(public_key_bytes: bytes, signature: bytes) -> MagicMock: + """Create a PrivateKey mock that returns a stable protobuf key representation.""" + key = MagicMock(spec=PrivateKey) + key.sign.return_value = signature + key.public_key().to_bytes_raw.return_value = public_key_bytes + key.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=public_key_bytes) + key.to_proto_key.return_value = basic_types_pb2.Key(ed25519=public_key_bytes) + return key + + ########### Basic Tests for Building Transactions ########### @@ -311,40 +321,14 @@ def test_sign_transaction(mock_account_ids, mock_client): private_key.sign.return_value = b"signature" private_key.public_key().to_bytes_raw.return_value = b"public_key" - private_key_admin = MagicMock(spec=PrivateKey) - private_key_admin.sign.return_value = b"admin_signature" - private_key_admin.public_key().to_bytes_raw.return_value = b"admin_public_key" - private_key_admin.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"admin_public_key") - - private_key_supply = MagicMock(spec=PrivateKey) - private_key_supply.sign.return_value = b"supply_signature" - private_key_supply.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"supply_public_key") - - private_key_freeze = MagicMock(spec=PrivateKey) - private_key_freeze.sign.return_value = b"freeze_signature" - private_key_freeze.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"freeze_public_key") - - private_key_wipe = MagicMock(spec=PrivateKey) - private_key_wipe.sign.return_value = b"wipe_signature" - private_key_wipe.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"wipe_public_key") - - private_key_metadata = MagicMock(spec=PrivateKey) - private_key_metadata.sign.return_value = b"metadata_signature" - private_key_metadata.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"metadata_public_key") - - private_key_pause = MagicMock(spec=PrivateKey) - private_key_pause.sign.return_value = b"pause_signature" - private_key_pause.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"pause_public_key") - - private_key_kyc = MagicMock(spec=PrivateKey) - private_key_kyc.sign.return_value = b"kyc_signature" - private_key_kyc.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"kyc_public_key") - - private_key_fee_schedule = MagicMock(spec=PrivateKey) - private_key_fee_schedule.sign.return_value = b"fee_schedule_signature" - private_key_fee_schedule.public_key()._to_proto.return_value = basic_types_pb2.Key( - ed25519=b"fee_schedule_public_key" - ) + private_key_admin = _mock_private_key(b"admin_public_key", b"admin_signature") + private_key_supply = _mock_private_key(b"supply_public_key", b"supply_signature") + private_key_freeze = _mock_private_key(b"freeze_public_key", b"freeze_signature") + private_key_wipe = _mock_private_key(b"wipe_public_key", b"wipe_signature") + private_key_metadata = _mock_private_key(b"metadata_public_key", b"metadata_signature") + private_key_pause = _mock_private_key(b"pause_public_key", b"pause_signature") + private_key_kyc = _mock_private_key(b"kyc_public_key", b"kyc_signature") + private_key_fee_schedule = _mock_private_key(b"fee_schedule_public_key", b"fee_schedule_signature") token_tx = TokenCreateTransaction() token_tx.set_token_name("MyToken") @@ -778,40 +762,14 @@ def test_build_and_sign_nft_transaction_to_proto(mock_account_ids, mock_client): private_key_private.sign.return_value = b"private_signature" private_key_private.public_key().to_bytes_raw.return_value = b"private_public_key" - private_key_admin = MagicMock(spec=PrivateKey) - private_key_admin.sign.return_value = b"admin_signature" - private_key_admin.public_key().to_bytes_raw.return_value = b"admin_public_key" - private_key_admin.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"admin_public_key") - - private_key_supply = MagicMock(spec=PrivateKey) - private_key_supply.sign.return_value = b"supply_signature" - private_key_supply.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"supply_public_key") - - private_key_freeze = MagicMock(spec=PrivateKey) - private_key_freeze.sign.return_value = b"freeze_signature" - private_key_freeze.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"freeze_public_key") - - private_key_wipe = MagicMock(spec=PrivateKey) - private_key_wipe.sign.return_value = b"wipe_signature" - private_key_wipe.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"wipe_public_key") - - private_key_metadata = MagicMock(spec=PrivateKey) - private_key_metadata.sign.return_value = b"metadata_signature" - private_key_metadata.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"metadata_public_key") - - private_key_pause = MagicMock(spec=PrivateKey) - private_key_pause.sign.return_value = b"pause_signature" - private_key_pause.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"pause_public_key") - - private_key_kyc = MagicMock(spec=PrivateKey) - private_key_kyc.sign.return_value = b"kyc_signature" - private_key_kyc.public_key()._to_proto.return_value = basic_types_pb2.Key(ed25519=b"kyc_public_key") - - private_key_fee_schedule = MagicMock(spec=PrivateKey) - private_key_fee_schedule.sign.return_value = b"fee_schedule_signature" - private_key_fee_schedule.public_key()._to_proto.return_value = basic_types_pb2.Key( - ed25519=b"fee_schedule_public_key" - ) + private_key_admin = _mock_private_key(b"admin_public_key", b"admin_signature") + private_key_supply = _mock_private_key(b"supply_public_key", b"supply_signature") + private_key_freeze = _mock_private_key(b"freeze_public_key", b"freeze_signature") + private_key_wipe = _mock_private_key(b"wipe_public_key", b"wipe_signature") + private_key_metadata = _mock_private_key(b"metadata_public_key", b"metadata_signature") + private_key_pause = _mock_private_key(b"pause_public_key", b"pause_signature") + private_key_kyc = _mock_private_key(b"kyc_public_key", b"kyc_signature") + private_key_fee_schedule = _mock_private_key(b"fee_schedule_public_key", b"fee_schedule_signature") # Build the transaction token_tx = TokenCreateTransaction()