chore: update crypto public key (#2042)

Signed-off-by: MonaaEid <monaa_eid@hotmail.com>

Author: MonaaEid

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.

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")

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
 # ------------------------------------------------------------------------------

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()