implement TLS support

Signed-off-by: emiliyank <e.kadiyski@gmail.com>

Author: emiliyank

CHANGELOG.md

@@ -8,6 +8,7 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.
 
 ### Added
 
+- TLS support with two-stage control (`set_transport_security()` and `set_verify_certificates()`) for encrypted connections to Hedera networks. TLS is enabled by default for hosted networks (mainnet, testnet, previewnet) and disabled for local networks (solo, localhost) (#855)
 - Add detail to `token_airdrop.py` and `token_airdrop_cancel.py`
 - Add workflow: github bot to respond to unverified PR commits (#750)
 - Add workflow: bot workflow which notifies developers of workflow failures in their pull requests.

docs/sdk_developers/setup.md

@@ -183,10 +183,13 @@ FREEZE_KEY=...
 RECIPIENT_ID=...
 TOKEN_ID=...
 TOPIC_ID=...
+VERIFY_CERTS=true  # Enable certificate verification for TLS (default: true)
 ```
 
 These are only needed if you're customizing example scripts.
 
+**Note on TLS:** The SDK uses TLS by default for hosted networks (testnet, mainnet, previewnet). For local networks (solo, localhost), TLS is disabled by default.
+
 ### Verify Your Setup
 
 Run the test suite to ensure everything is working:

examples/tls_query_balance.py

@@ -0,0 +1,65 @@
+"""
+TLS Query Balance Example
+
+Demonstrates how to connect to the Hedera network with TLS enabled.
+
+Required environment variables:
+  - OPERATOR_ID
+  - OPERATOR_KEY
+Optional:
+  - NETWORK (defaults to testnet)
+  - VERIFY_CERTS (set to \"true\" to enforce certificate hash checks)
+
+Run with:
+  uv run examples/tls_query_balance.py
+"""
+
+import os
+from dotenv import load_dotenv
+
+from hiero_sdk_python import (
+    Network,
+    Client,
+    AccountId,
+    PrivateKey,
+    CryptoGetAccountBalanceQuery,
+)
+
+
+def _bool_env(name: str, default: bool = False) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes"}
+
+
+def main():
+    load_dotenv()
+
+    network_name = os.getenv("NETWORK", "testnet")
+    operator_id_str = os.getenv("OPERATOR_ID")
+    operator_key_str = os.getenv("OPERATOR_KEY")
+
+    if not operator_id_str or not operator_key_str:
+        raise ValueError("OPERATOR_ID and OPERATOR_KEY must be set in the environment")
+
+    network = Network(network_name)
+    client = Client(network)
+
+    # Enable TLS for consensus nodes. Mirror nodes already require TLS.
+    client.set_transport_security(False)
+
+    verify_certs = _bool_env("VERIFY_CERTS", True)
+    client.set_verify_certificates(verify_certs)
+
+    operator_id = AccountId.from_string(operator_id_str)
+    operator_key = PrivateKey.from_string(operator_key_str)
+    client.set_operator(operator_id, operator_key)
+
+    balance = CryptoGetAccountBalanceQuery().set_account_id(operator_id).execute(client)
+    print(f"Operator account {operator_id} balance: {balance.hbars.to_hbars()} hbars")
+
+
+if __name__ == "__main__":
+    main()
+

src/hiero_sdk_python/client/client.py

@@ -2,7 +2,7 @@
 Client module for interacting with the Hedera network.
 """
 
-from typing import NamedTuple, List, Union
+from typing import NamedTuple, List, Union, Optional
 
 import grpc
 
@@ -50,9 +50,11 @@ def __init__(self, network: Network = None) -> None:
     def _init_mirror_stub(self) -> None:
         """
         Connect to a mirror node for topic message subscriptions.
-        We now use self.network.get_mirror_address() for a configurable mirror address.
+        Mirror nodes always use TLS (mandatory). We use self.network.get_mirror_address()
+        for a configurable mirror address, which should use port 443 for HTTPS connections.
         """
         mirror_address = self.network.get_mirror_address()
+        # Mirror nodes always require TLS - secure_channel is mandatory
         self.mirror_channel = grpc.secure_channel(mirror_address, grpc.ssl_channel_credentials())
         self.mirror_stub = mirror_consensus_grpc.ConsensusServiceStub(self.mirror_channel)
 
@@ -103,6 +105,54 @@ def close(self) -> None:
 
         self.mirror_stub = None
 
+    def set_transport_security(self, enabled: bool) -> "Client":
+        """
+        Enable or disable TLS for consensus node connections.
+        
+        Note:
+            TLS is enabled by default for hosted networks (mainnet, testnet, previewnet).
+            For local networks (solo, localhost) and custom networks, TLS is disabled by default.
+            Use this method to override the default behavior.
+        """
+        self.network.set_transport_security(enabled)
+        return self
+
+    def is_transport_security(self) -> bool:
+        """
+        Determine if TLS is enabled for consensus node connections.
+        """
+        return self.network.is_transport_security()
+
+    def set_verify_certificates(self, verify: bool) -> "Client":
+        """
+        Enable or disable verification of server certificates when TLS is enabled.
+        
+        Note:
+            Certificate verification is enabled by default for all networks.
+            Use this method to disable verification (e.g., for testing with self-signed certificates).
+        """
+        self.network.set_verify_certificates(verify)
+        return self
+
+    def is_verify_certificates(self) -> bool:
+        """
+        Determine if certificate verification is enabled.
+        """
+        return self.network.is_verify_certificates()
+
+    def set_tls_root_certificates(self, root_certificates: Optional[bytes]) -> "Client":
+        """
+        Provide custom root certificates for TLS connections.
+        """
+        self.network.set_tls_root_certificates(root_certificates)
+        return self
+
+    def get_tls_root_certificates(self) -> Optional[bytes]:
+        """
+        Retrieve the configured root certificates for TLS connections.
+        """
+        return self.network.get_tls_root_certificates()
+
     def __enter__(self) -> "Client":
         """
         Allows the Client to be used in a 'with' statement for automatic resource management.

src/hiero_sdk_python/client/network.py

@@ -15,18 +15,20 @@ class Network:
     Manages the network configuration for connecting to the Hedera network.
     """
 
+    # Mirror node gRPC addresses (always use TLS, port 443 for HTTPS)
     MIRROR_ADDRESS_DEFAULT: Dict[str,str] = {
         'mainnet': 'mainnet.mirrornode.hedera.com:443',
         'testnet': 'testnet.mirrornode.hedera.com:443',
         'previewnet': 'previewnet.mirrornode.hedera.com:443',
-        'solo': 'localhost:5600'
+        'solo': 'localhost:5600'  # Local development only
     }
 
+    # Mirror node REST API base URLs (HTTPS for production networks, HTTP for localhost)
     MIRROR_NODE_URLS: Dict[str,str] = {
         'mainnet': 'https://mainnet-public.mirrornode.hedera.com',
         'testnet': 'https://testnet.mirrornode.hedera.com',
         'previewnet': 'https://previewnet.mirrornode.hedera.com',
-        'solo': 'http://localhost:8080'
+        'solo': 'http://localhost:8080'  # Local development only
     }
 
     DEFAULT_NODES: Dict[str,List[_Node]] = {
@@ -92,13 +94,25 @@ def __init__(
             mirror_address (str, optional): A mirror node address (host:port) for topic queries.
                             If not provided,
                             we'll use a default from MIRROR_ADDRESS_DEFAULT[network].
+        
+        Note:
+            TLS is enabled by default for hosted networks (mainnet, testnet, previewnet).
+            For local networks (solo, localhost) and custom networks, TLS is disabled by default.
+            Certificate verification is enabled by default for all networks.
+            Use Client.set_transport_security() and Client.set_verify_certificates() to customize.
         """
         self.network: str = network or 'testnet'
         self.mirror_address: str = mirror_address or self.MIRROR_ADDRESS_DEFAULT.get(
             network, 'localhost:5600'
         )
 
         self.ledger_id = ledger_id or self.LEDGER_ID.get(network, bytes.fromhex('03'))
+        
+        # Default TLS configuration: enabled for hosted networks, disabled for local/custom
+        hosted_networks = ('mainnet', 'testnet', 'previewnet')
+        self._transport_security: bool = self.network in hosted_networks
+        self._verify_certificates: bool = True  # Always enabled by default
+        self._root_certificates: Optional[bytes] = None
 
         if nodes is not None:
             final_nodes = nodes
@@ -114,6 +128,12 @@ def __init__(
                 raise ValueError(f"No default nodes for network='{self.network}'")
 
         self.nodes: List[_Node] = final_nodes
+        
+        # Apply TLS configuration to all nodes
+        for node in self.nodes:
+            node._apply_transport_security(self._transport_security)  # pylint: disable=protected-access
+            node._set_verify_certificates(self._verify_certificates)  # pylint: disable=protected-access
+            node._set_root_certificates(self._root_certificates)  # pylint: disable=protected-access
 
         self._node_index: int = secrets.randbelow(len(self.nodes))
         self.current_node: _Node = self.nodes[self._node_index]
@@ -180,6 +200,91 @@ def _select_node(self) -> _Node:
 
     def get_mirror_address(self) -> str:
         """
-        Return the configured mirror node address used for mirror queries.
+        Return the configured mirror node address used for mirror gRPC queries.
+        Mirror nodes always use TLS, so addresses should use port 443 for HTTPS.
         """
         return self.mirror_address
+    
+    def get_mirror_rest_url(self) -> str:
+        """
+        Get the REST API base URL for the mirror node.
+        Returns the URL in format: scheme://host[:port]/api/v1
+        For non-localhost networks, defaults to https:// with port 443.
+        """
+        base_url = self.MIRROR_NODE_URLS.get(self.network)
+        if base_url:
+            # MIRROR_NODE_URLS contains base URLs, append /api/v1
+            return f"{base_url}/api/v1"
+        
+        # Fallback: construct from mirror_address
+        mirror_addr = self.mirror_address
+        if ':' in mirror_addr:
+            host, port_str = mirror_addr.rsplit(':', 1)
+            try:
+                port = int(port_str)
+            except ValueError:
+                port = 443
+        else:
+            host = mirror_addr
+            port = 443
+        
+        # Use HTTPS for non-localhost, HTTP for localhost
+        if host in ('localhost', '127.0.0.1'):
+            scheme = 'http'
+            if port == 443:
+                port = 8080  # Default REST port for localhost
+        else:
+            scheme = 'https'
+            if port == 5600:  # gRPC port, use 443 for REST
+                port = 443
+        
+        # Omit default ports
+        if (scheme == 'https' and port == 443) or (scheme == 'http' and port == 80):
+            return f"{scheme}://{host}/api/v1"
+        return f"{scheme}://{host}:{port}/api/v1"
+
+    def set_transport_security(self, enabled: bool) -> None:
+        """
+        Enable or disable TLS for consensus node connections.
+        """
+        if self._transport_security == enabled:
+            return
+        for node in self.nodes:
+            node._apply_transport_security(enabled)  # pylint: disable=protected-access
+        self._transport_security = enabled
+
+    def is_transport_security(self) -> bool:
+        """
+        Determine if TLS is enabled for consensus node connections.
+        """
+        return self._transport_security
+
+    def set_verify_certificates(self, verify: bool) -> None:
+        """
+        Enable or disable server certificate verification when TLS is enabled.
+        """
+        if self._verify_certificates == verify:
+            return
+        for node in self.nodes:
+            node._set_verify_certificates(verify)  # pylint: disable=protected-access
+        self._verify_certificates = verify
+
+    def set_tls_root_certificates(self, root_certificates: Optional[bytes]) -> None:
+        """
+        Provide custom root certificates to use when establishing TLS channels.
+        """
+        self._root_certificates = root_certificates
+        for node in self.nodes:
+            node._set_root_certificates(root_certificates)  # pylint: disable=protected-access
+
+    def get_tls_root_certificates(self) -> Optional[bytes]:
+        """
+        Retrieve the configured root certificates used for TLS channels.
+        """
+        return self._root_certificates
+
+    def is_verify_certificates(self) -> bool:
+        """
+        Determine if certificate verification is enabled.
+        """
+        return self._verify_certificates

src/hiero_sdk_python/managed_node_address.py

@@ -5,6 +5,12 @@ class _ManagedNodeAddress:
     Represents a managed node address with a host and port.
     This class is used to handle node addresses in the Hedera network.
     """
+    PORT_NODE_PLAIN = 50211
+    PORT_NODE_TLS = 50212
+    PORT_MIRROR_TLS = 443
+    PORT_MIRROR_PLAIN = 5600
+    TLS_PORTS = {PORT_NODE_TLS, PORT_MIRROR_TLS}
+    PLAIN_PORTS = {PORT_NODE_PLAIN, PORT_MIRROR_PLAIN}
 
     # Regular expression to parse a host:port string
     HOST_PORT_PATTERN = re.compile(r'^(\S+):(\d+)$')
@@ -53,8 +59,48 @@ def _is_transport_security(self):
         Returns:
             bool: True if the port is a secure port (50212 or 443), False otherwise.
         """
-        return self._port == 50212 or self._port == 443
+        return self._port in self.TLS_PORTS
 
+    def _to_secure(self):
+        """
+        Return a new ManagedNodeAddress that uses the secure port when possible.
+        """
+        if self._is_transport_security():
+            return self
+        
+        port = self._port
+        if port == self.PORT_NODE_PLAIN:
+            port = self.PORT_NODE_TLS
+        elif port == self.PORT_MIRROR_PLAIN:
+            port = self.PORT_MIRROR_TLS
+        return _ManagedNodeAddress(self._address, port)
+    
+    def _to_insecure(self):
+        """
+        Return a new ManagedNodeAddress that uses the plaintext port when possible.
+        """
+        if not self._is_transport_security():
+            return self
+        
+        port = self._port
+        if port == self.PORT_NODE_TLS:
+            port = self.PORT_NODE_PLAIN
+        elif port == self.PORT_MIRROR_TLS:
+            port = self.PORT_MIRROR_PLAIN
+        return _ManagedNodeAddress(self._address, port)
+    
+    def _get_host(self):
+        """
+        Return the host component of the address.
+        """
+        return self._address
+
+    def _get_port(self):
+        """
+        Return the port component of the address.
+        """
+        return self._port
+
     def __str__(self):
         """
         Get a string representation of the ManagedNodeAddress.