feat: implement TLS support

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,84 @@
+"""
+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 _load_operator_credentials() -> tuple[AccountId, PrivateKey]:
+    """Load operator credentials from the environment."""
+    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")
+
+    operator_id = AccountId.from_string(operator_id_str)
+    operator_key = PrivateKey.from_string(operator_key_str)
+    return operator_id, operator_key
+
+
+def setup_client() -> Client:
+    """Create and configure a client with TLS enabled using env settings."""
+    network_name = os.getenv("NETWORK", "testnet")
+    verify_certs = _bool_env("VERIFY_CERTS", True)
+
+    network = Network(network_name)
+    client = Client(network)
+
+    # Disable TLS for consensus nodes. Mirror nodes already require TLS.
+    client.set_transport_security(False)
+    client.set_verify_certificates(verify_certs)
+    return client
+
+
+def query_account_balance(client: Client, account_id: AccountId):
+    """Execute a CryptoGetAccountBalanceQuery for the given account."""
+    query = CryptoGetAccountBalanceQuery().set_account_id(account_id)
+    balance = query.execute(client)
+    print(f"Operator account {account_id} balance: {balance.hbars.to_hbars()} hbars")
+
+
+def main():
+    load_dotenv()
+
+    operator_id, operator_key = _load_operator_credentials()
+    client = setup_client()
+    client.set_operator(operator_id, operator_key)
+
+    query_account_balance(client, operator_id)
+
+
+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

@@ -1,6 +1,6 @@
 """Network module for managing Hedera SDK connections."""
 import secrets
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, Tuple
 
 import requests
 
@@ -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,130 @@ 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 _parse_mirror_address(self) -> Tuple[str, int]:
+        """
+        Parse mirror_address into host and port.
+
+        Returns:
+            Tuple[str, int]: (host, port) tuple
+        """
+        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
+        return (host, port)
+
+    def _determine_scheme_and_port(self, host: str, port: int) -> Tuple[str, int]:
+        """
+        Determine the scheme (http/https) and port for the REST URL.
+
+        Args:
+            host: The hostname
+            port: The port number
+
+        Returns:
+            Tuple[str, int]: (scheme, port) tuple
+        """
+        is_localhost = host in ('localhost', '127.0.0.1')
+
+        if is_localhost:
+            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
+
+        return (scheme, port)
+
+    def _build_rest_url(self, scheme: str, host: str, port: int) -> str:
+        """
+        Build the final REST URL with optional port.
+
+        Args:
+            scheme: URL scheme (http or https)
+            host: Hostname
+            port: Port number
+
+        Returns:
+            str: Complete REST URL with /api/v1 suffix
+        """
+        is_default_port = (scheme == 'https' and port == 443) or (scheme == 'http' and port == 80)
+
+        if is_default_port:
+            return f"{scheme}://{host}/api/v1"
+        return f"{scheme}://{host}:{port}/api/v1"
+
+    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
+        host, port = self._parse_mirror_address()
+        scheme, port = self._determine_scheme_and_port(host, port)
+        return self._build_rest_url(scheme, host, port)
+
+    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