You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by cr...@apache.org on 2025/02/18 09:24:34 UTC
(iotdb) branch dev/1.3 updated: Add SSL support to Python client (#14864)
This is an automated email from the ASF dual-hosted git repository.
critas pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new ee503f3cc5a Add SSL support to Python client (#14864)
ee503f3cc5a is described below
commit ee503f3cc5a696f97fcb1edc20ed83a2cb5c1bae
Author: CloudWise-Lukemiao <28...@qq.com>
AuthorDate: Tue Feb 18 17:24:28 2025 +0800
Add SSL support to Python client (#14864)
---
iotdb-client/client-py/iotdb/Session.py | 52 +++++++++++++++++++++++------
iotdb-client/client-py/iotdb/SessionPool.py | 11 ++++--
2 files changed, 50 insertions(+), 13 deletions(-)
diff --git a/iotdb-client/client-py/iotdb/Session.py b/iotdb-client/client-py/iotdb/Session.py
index 8feff10671b..90018e06379 100644
--- a/iotdb-client/client-py/iotdb/Session.py
+++ b/iotdb-client/client-py/iotdb/Session.py
@@ -19,10 +19,12 @@
import logging
import random
import struct
+import sys
+import ssl
import time
import warnings
from thrift.protocol import TBinaryProtocol, TCompactProtocol
-from thrift.transport import TSocket, TTransport
+from thrift.transport import TSocket, TTransport, TSSLSocket
from iotdb.utils.SessionDataSet import SessionDataSet
from .template.Template import Template
@@ -84,6 +86,8 @@ class Session(object):
fetch_size=DEFAULT_FETCH_SIZE,
zone_id=DEFAULT_ZONE_ID,
enable_redirection=True,
+ use_ssl=False,
+ ca_certs=None,
):
self.__host = host
self.__port = port
@@ -104,6 +108,8 @@ class Session(object):
self.__enable_redirection = enable_redirection
self.__device_id_to_endpoint = None
self.__endpoint_to_connection = None
+ self.__use_ssl = use_ssl
+ self.__ca_certs = ca_certs
@classmethod
def init_from_node_urls(
@@ -114,11 +120,21 @@ class Session(object):
fetch_size=DEFAULT_FETCH_SIZE,
zone_id=DEFAULT_ZONE_ID,
enable_redirection=True,
+ use_ssl=False,
+ ca_certs=None,
):
if node_urls is None:
raise RuntimeError("node urls is empty")
session = Session(
- None, None, user, password, fetch_size, zone_id, enable_redirection
+ None,
+ None,
+ user,
+ password,
+ fetch_size,
+ zone_id,
+ enable_redirection,
+ use_ssl=use_ssl,
+ ca_certs=ca_certs,
)
session.__hosts = []
session.__ports = []
@@ -166,15 +182,31 @@ class Session(object):
}
def init_connection(self, endpoint):
- transport = TTransport.TFramedTransport(
- TSocket.TSocket(endpoint.ip, endpoint.port)
- )
+ try:
+ if self.__use_ssl:
+ if sys.version_info >= (3, 10):
+ context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
+ else:
+ context = ssl.SSLContext(ssl.PROTOCOL_TLS)
+ context.verify_mode = ssl.CERT_REQUIRED
+ context.check_hostname = True
+ context.load_verify_locations(cafile=self.__ca_certs)
+ socket = TSSLSocket.TSSLSocket(
+ host=endpoint.ip, port=endpoint.port, ssl_context=context
+ )
+ else:
+ socket = TSocket.TSocket(endpoint.ip, endpoint.port)
+ transport = TTransport.TFramedTransport(socket)
- if not transport.isOpen():
- try:
- transport.open()
- except TTransport.TTransportException as e:
- raise IoTDBConnectionException(e) from None
+ if not transport.isOpen():
+ try:
+ transport.open()
+ except TTransport.TTransportException as e:
+ raise IoTDBConnectionException(e) from None
+ except ssl.SSLError as e:
+ print(f"SSL error occurred: {e}")
+ except Exception as e:
+ print(f"An unexpected error occurred: {e}")
if self.__enable_rpc_compression:
client = Client(TCompactProtocol.TCompactProtocolAccelerated(transport))
diff --git a/iotdb-client/client-py/iotdb/SessionPool.py b/iotdb-client/client-py/iotdb/SessionPool.py
index 6f1d7580796..f6d74e16515 100644
--- a/iotdb-client/client-py/iotdb/SessionPool.py
+++ b/iotdb-client/client-py/iotdb/SessionPool.py
@@ -42,6 +42,8 @@ class PoolConfig(object):
time_zone: str = DEFAULT_TIME_ZONE,
max_retry: int = DEFAULT_MAX_RETRY,
enable_compression: bool = False,
+ use_ssl: bool = False,
+ ca_certs: str = None,
):
self.host = host
self.port = port
@@ -58,6 +60,8 @@ class PoolConfig(object):
self.time_zone = time_zone
self.max_retry = max_retry
self.enable_compression = enable_compression
+ self.use_ssl = use_ssl
+ self.ca_certs = ca_certs
class SessionPool(object):
@@ -80,6 +84,8 @@ class SessionPool(object):
self.__pool_config.password,
self.__pool_config.fetch_size,
self.__pool_config.time_zone,
+ use_ssl=self.__pool_config.use_ssl,
+ ca_certs=self.__pool_config.ca_certs,
)
else:
@@ -90,6 +96,8 @@ class SessionPool(object):
self.__pool_config.password,
self.__pool_config.fetch_size,
self.__pool_config.time_zone,
+ use_ssl=self.__pool_config.use_ssl,
+ ca_certs=self.__pool_config.ca_certs,
)
session.open(self.__pool_config.enable_compression)
@@ -102,7 +110,6 @@ class SessionPool(object):
return q
def get_session(self) -> Session:
-
if self.__closed:
raise ConnectionError("SessionPool has already been closed.")
@@ -137,7 +144,6 @@ class SessionPool(object):
return session
def put_back(self, session: Session):
-
if self.__closed:
raise ConnectionError(
"SessionPool has already been closed, please close the session manually."
@@ -150,7 +156,6 @@ class SessionPool(object):
self.__pool_size -= 1
def close(self):
-
with self.__lock:
while not self.__queue.empty():
session = self.__queue.get(block=False)