fix: TLS Hostname Mismatch & Certificate Verification Failure for Nodes (#1038)

Signed-off-by: Manish Dait <[email protected]>

Author: manishdait

CHANGELOG.md

@@ -86,6 +86,7 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.
 - good first issue template yaml rendering
 - Fixed solo workflow defaulting to zero
 - Fix unit test tet_query.py
+- TLS Hostname Mismatch & Certificate Verification Failure for Nodes
 
 ### Breaking Change
 

src/hiero_sdk_python/node.py

@@ -2,7 +2,7 @@
 import socket
 import ssl  # Python's ssl module implements TLS (despite the name)
 import grpc
-from typing import Optional, Callable
+from typing import Optional
 from hiero_sdk_python.account.account_id import AccountId
 from hiero_sdk_python.channels import _Channel
 from hiero_sdk_python.address_book.node_address import NodeAddress
@@ -91,7 +91,7 @@ def __init__(self, account_id: AccountId, address: str, address_book: NodeAddres
         self._address: _ManagedNodeAddress = _ManagedNodeAddress._from_string(address)
         self._verify_certificates: bool = True
         self._root_certificates: Optional[bytes] = None
-        self._authority_override: Optional[str] = self._determine_authority_override()
+        self._node_pem_cert: Optional[bytes] = None
 
     def _close(self):
         """
@@ -115,13 +115,23 @@ def _get_channel(self):
             return self._channel
 
         if self._address._is_transport_security():
+            if self._root_certificates:
+                # Use the certificate that is provided
+                self._node_pem_cert = self._root_certificates
+            else:
+                # Fetch pem_cert for the node
+                self._node_pem_cert = self._fetch_server_certificate_pem()
+
+            if not self._node_pem_cert:
+                raise ValueError("No certificate available.")
+            
             # Validate certificate if verification is enabled
             if self._verify_certificates:
-                self._validate_tls_certificate_with_trust_manager()
+                self._validate_tls_certificate_with_trust_manager()            
 
             options = self._build_channel_options()
             credentials = grpc.ssl_channel_credentials(
-                root_certificates=self._root_certificates,
+                root_certificates=self._node_pem_cert,
                 private_key=None,
                 certificate_chain=None,
             )
@@ -141,7 +151,9 @@ def _apply_transport_security(self, enabled: bool):
             return
         if not enabled and not self._address._is_transport_security():
             return
+        
         self._close()
+        
         if enabled:
             self._address = self._address._to_secure()
         else:
@@ -154,39 +166,48 @@ def _set_root_certificates(self, root_certificates: Optional[bytes]):
         self._root_certificates = root_certificates
         if self._channel and self._address._is_transport_security():
             self._close()
+            
     def _set_verify_certificates(self, verify: bool):
         """
         Set whether TLS certificates should be verified.
         """
         if self._verify_certificates == verify:
             return
+        
         self._verify_certificates = verify
+        
         if verify and self._channel and self._address._is_transport_security():
             # Force channel recreation to ensure certificates are revalidated.
             self._close()
 
-    def _determine_authority_override(self) -> Optional[str]:
-        """
-        Determine the hostname to use for TLS authority override.
-        """
-        if not self._address_book or not self._address_book._addresses:  # pylint: disable=protected-access
-            return None
-        for endpoint in self._address_book._addresses:  # pylint: disable=protected-access
-            domain = endpoint.get_domain_name()
-            if domain:
-                return domain
-        return None
-
     def _build_channel_options(self):
         """
         Build gRPC channel options for TLS connections.
+
+        The options `grpc.default_authority` and `grpc.ssl_target_name_override`
+        are intentionally set to a fixed value ("127.0.0.1") to bypass standard
+        TLS hostname verification.
+
+        This is REQUIRED because Hedera nodes are connected to via IP addresses 
+        from the address book, while their TLS certificates are not issued for 
+        those IPs. As a result, standard hostname verification would fail even 
+        for legitimate nodes.
+
+        Although hostname verification is disabled, transport security is NOT
+        weakened. Instead of relying on hostnames, the SDK validates the server
+        by performing certificate hash pinning. This guarantees the client is 
+        communicating with the correct Hedera node regardless of the hostname 
+        or IP address used to connect.
         """
-        if not self._authority_override:
-            return None
-        host = self._address._get_host()
-        if host == self._authority_override:
-            return None
-        return [('grpc.ssl_target_name_override', self._authority_override)]
+        options = [
+            ("grpc.default_authority", "127.0.0.1"),
+            ("grpc.ssl_target_name_override", "127.0.0.1"),
+            ("grpc.keepalive_time_ms", 100000),
+            ("grpc.keepalive_timeout_ms", 10000),
+            ("grpc.keepalive_permit_without_calls", 1)
+        ]
+
+        return options
 
     def _validate_tls_certificate_with_trust_manager(self):
         """
@@ -197,9 +218,7 @@ def _validate_tls_certificate_with_trust_manager(self):
         Note: If verification is enabled but no cert hash is available (e.g., in unit tests
         without address books), validation is skipped rather than raising an error.
         """
-        if not self._address._is_transport_security():
-            return
-        if not self._verify_certificates:
+        if not self._address._is_transport_security() or not self._verify_certificates:
             return
 
         cert_hash = None
@@ -214,10 +233,7 @@ def _validate_tls_certificate_with_trust_manager(self):
 
         # Create trust manager and validate certificate
         trust_manager = _HederaTrustManager(cert_hash, self._verify_certificates)
-        
-        # Fetch server certificate and validate
-        pem_cert = self._fetch_server_certificate_pem()
-        trust_manager.check_server_trusted(pem_cert)
+        trust_manager.check_server_trusted(self._node_pem_cert)
 
     @staticmethod
     def _normalize_cert_hash(cert_hash: bytes) -> str:
@@ -228,6 +244,7 @@ def _normalize_cert_hash(cert_hash: bytes) -> str:
             decoded = cert_hash.decode('utf-8').strip().lower()
             if decoded.startswith("0x"):
                 decoded = decoded[2:]
+
             return decoded
         except UnicodeDecodeError:
             return cert_hash.hex()
@@ -239,12 +256,22 @@ def _fetch_server_certificate_pem(self) -> bytes:
         Returns:
             bytes: PEM-encoded certificate bytes
         """
+        if not self._address_book:
+            return None
+
         host = self._address._get_host()
         port = self._address._get_port()
-        server_hostname = self._authority_override or host
+        server_hostname = host
 
         # Create TLS context that accepts any certificate (we validate hash ourselves)
         context = ssl.create_default_context()
+        # Restrict SSL/TLS versions to TLSv1.2+ only for security
+        if hasattr(context, 'minimum_version') and hasattr(ssl, 'TLSVersion'):
+            context.minimum_version = ssl.TLSVersion.TLSv1_2
+        else:
+            # Backwards compatibility for Python <3.7 that lacks minimum_version
+            context.options |= ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
+
         context.check_hostname = False
         context.verify_mode = ssl.CERT_NONE
 
@@ -254,4 +281,4 @@ def _fetch_server_certificate_pem(self) -> bytes:
 
         # Convert DER to PEM format (matching Java's PEM encoding)
         pem_cert = ssl.DER_cert_to_PEM_cert(der_cert).encode('utf-8')
-        return pem_cert
+        return pem_cert

tests/unit/conftest.py

@@ -1,15 +1,16 @@
+import hashlib
 import time
 
 import pytest
 
 from hiero_sdk_python.account.account_id import AccountId
+from hiero_sdk_python.address_book.node_address import NodeAddress
 from hiero_sdk_python.client.client import Client
 from hiero_sdk_python.client.network import Network
 from hiero_sdk_python.consensus.topic_id import TopicId
 from hiero_sdk_python.contract.contract_id import ContractId
 from hiero_sdk_python.crypto.private_key import PrivateKey
 from hiero_sdk_python.file.file_id import FileId
-from hiero_sdk_python.hapi.services import timestamp_pb2
 from hiero_sdk_python.logger.log_level import LogLevel
 from hiero_sdk_python.node import _Node
 from hiero_sdk_python.tokens.token_id import TokenId
@@ -18,6 +19,12 @@
 from hiero_sdk_python.transaction.transaction_id import TransactionId
 
 
+FAKE_CERT_PEM = b"""-----BEGIN CERTIFICATE-----
+MIIBszCCAVmgAwIBAgIUQFakeFakeFakeFakeFakeFakeFakewCgYIKoZIzj0EAwIw
+-----END CERTIFICATE-----"""
+
+FAKE_CERT_HASH = hashlib.sha384(FAKE_CERT_PEM).hexdigest().encode("utf-8")
+
 @pytest.fixture
 def mock_account_ids():
     """Fixture to provide mock account IDs and token IDs."""
@@ -78,7 +85,16 @@ def contract_id():
 @pytest.fixture
 def mock_client():
     """Fixture to provide a mock client with hardcoded nodes for testing purposes."""
-    nodes = [_Node(AccountId(0, 0, 3), "node1.example.com:50211", None)]
+    # Mock Node
+    node = _Node(
+        AccountId(0, 0, 3),
+        "node1.example.com:50211",
+        address_book=NodeAddress(cert_hash=FAKE_CERT_HASH, addresses=[])
+    )
+    node._fetch_server_certificate_pem = lambda: FAKE_CERT_PEM
+    
+    nodes = [node]
+
     network = Network(nodes=nodes)
     client = Client(network)
     client.logger.set_level(LogLevel.DISABLED)