v5.0.1

ddcdatabases/__init__.py

@@ -1,4 +1,5 @@
 import logging
+from .core.certs import SSLCertificateError
 from .core.operations import DBUtils, DBUtilsAsync
 from .core.persistent import PersistentConnectionConfig, close_all_persistent_connections
 from importlib.metadata import version
@@ -7,6 +8,7 @@
     "DBUtils",
     "DBUtilsAsync",
     "PersistentConnectionConfig",
+    "SSLCertificateError",
     "close_all_persistent_connections",
 ]
 

ddcdatabases/core/base.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 import logging
 import sqlalchemy as sa
+from .certs import verify_cert_paths
 from .configs import BaseOperationRetryConfig, BaseRetryConfig
 from .retry import retry_operation, retry_operation_async
 from collections.abc import AsyncGenerator, Generator
@@ -19,6 +20,7 @@ class BaseConnection:
     __slots__ = (
         "connection_url",
         "engine_args",
+        "driver_cert_paths",
         "autoflush",
         "expire_on_commit",
         "sync_driver",
@@ -42,9 +44,11 @@ def __init__(
         connection_retry_config: BaseRetryConfig | None = None,
         operation_retry_config: BaseOperationRetryConfig | None = None,
         logger: Any = None,
+        driver_cert_paths: tuple[tuple[str | None, str], ...] = (),
     ) -> None:
         self.connection_url = connection_url
         self.engine_args = engine_args
+        self.driver_cert_paths = driver_cert_paths
         self.autoflush = autoflush
         self.expire_on_commit = expire_on_commit
         self.sync_driver = sync_driver
@@ -111,13 +115,15 @@ async def __aexit__(
 
     @contextmanager
     def _get_engine(self) -> Generator[Engine, None, None]:
+        verify_cert_paths(self.driver_cert_paths)
         _connection_url = URL.create(drivername=self.sync_driver, **self.connection_url)
         _engine = create_engine(url=_connection_url, **self.engine_args)
         yield _engine
         _engine.dispose()
 
     @asynccontextmanager
     async def _get_async_engine(self) -> AsyncGenerator[AsyncEngine, None]:
+        verify_cert_paths(self.driver_cert_paths)
         _connection_url = URL.create(drivername=self.async_driver, **self.connection_url)
         _engine = create_async_engine(url=_connection_url, **self.engine_args)
         yield _engine

ddcdatabases/core/certs.py

@@ -0,0 +1,56 @@
+"""
+Construction of the client SSL contexts used by the async drivers.
+"""
+
+import os
+import ssl
+from .constants import CA_CERT_LABEL, MINIMUM_TLS_VERSION
+from collections.abc import Iterable
+
+
+class SSLCertificateError(OSError):
+    """A configured certificate path could not be loaded"""
+
+
+def build_client_ssl_context(
+    ca_cert_path: str,
+    client_cert_path: str | None = None,
+    client_key_path: str | None = None,
+    minimum_version: ssl.TLSVersion = MINIMUM_TLS_VERSION,
+) -> ssl.SSLContext:
+    """Build a client SSL context, naming the file when one cannot be loaded"""
+
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)  # noqa: S4423
+    context.minimum_version = minimum_version
+
+    try:
+        context.load_verify_locations(cafile=ca_cert_path)
+    except OSError as err:
+        raise SSLCertificateError(f"{CA_CERT_LABEL} unusable: {ca_cert_path} | {err}") from err
+
+    if client_cert_path and client_key_path:
+        try:
+            context.load_cert_chain(certfile=client_cert_path, keyfile=client_key_path)
+        except OSError as err:
+            raise SSLCertificateError(
+                f"client certificate/key unusable: {client_cert_path}, {client_key_path} | {err}"
+            ) from err
+
+    return context
+
+
+def verify_cert_paths(entries: Iterable[tuple[str | None, str]]) -> None:
+    """Raise SSLCertificateError naming every configured path that is not usable"""
+
+    problems: list[str] = []
+    for path, label in entries:
+        if not path:
+            continue
+        if not os.path.exists(path):
+            problems.append(f"{label} missing: {path}")
+            continue
+        required = os.R_OK | os.X_OK if os.path.isdir(path) else os.R_OK
+        if not os.access(path, required):
+            problems.append(f"{label} not readable: {path}")
+    if problems:
+        raise SSLCertificateError("; ".join(problems))

ddcdatabases/core/configs.py

@@ -34,6 +34,7 @@ def merge_config_with_settings[C](
         field_map: Dict mapping config field names to settings attribute names.
                    If None, config field names must match settings attribute names.
     """
+
     override = override or config_cls()
     field_map = field_map or {}
     kwargs = {}
@@ -47,10 +48,11 @@ def merge_config_with_settings[C](
 
 def _validate_retry_config(
     max_retries: int | None,
-    initial_retry_delay: float | None,
-    max_retry_delay: float | None,
+    initial_retry_delay: float | int | None,
+    max_retry_delay: float | int | None,
 ) -> None:
     """Validation for retry configs to avoid super() issues with frozen slotted dataclasses"""
+
     if max_retries is not None and max_retries < 0:
         raise ValueError("max_retries must be non-negative")
     if initial_retry_delay is not None and initial_retry_delay < 0:
@@ -109,16 +111,16 @@ class BaseSessionConfig:
 class BaseRetryConfig:
     enable_retry: bool | None = None
     max_retries: int | None = None
-    initial_retry_delay: float | None = None
-    max_retry_delay: float | None = None
+    initial_retry_delay: float | int | None = None
+    max_retry_delay: float | int | None = None
 
     def __post_init__(self) -> None:
         _validate_retry_config(self.max_retries, self.initial_retry_delay, self.max_retry_delay)
 
 
 @dataclass(frozen=True, slots=True)
 class BaseOperationRetryConfig(BaseRetryConfig):
-    jitter: float | None = None
+    jitter: float | int | None = None
 
     def __post_init__(self) -> None:
         _validate_retry_config(self.max_retries, self.initial_retry_delay, self.max_retry_delay)

ddcdatabases/core/constants.py

@@ -1,7 +1,11 @@
+import ssl
 from typing import Final
 
+# Lowest TLS version
+MINIMUM_TLS_VERSION: Final = ssl.TLSVersion.TLSv1_3
+
 # SSL Modes
-POSTGRESQL_SSL_MODES: Final[frozenset[str]] = frozenset(
+POSTGRESQL_SSL_MODES: Final[frozenset[str]] = frozenset[str](
     {
         "disable",
         "allow",
@@ -12,7 +16,7 @@
     }
 )
 
-MYSQL_SSL_MODES: Final[frozenset[str]] = frozenset(
+MYSQL_SSL_MODES: Final[frozenset[str]] = frozenset[str](
     {
         "DISABLED",
         "PREFERRED",
@@ -22,8 +26,13 @@
     }
 )
 
+# Labels naming each certificate path in SSLCertificateError messages
+CA_CERT_LABEL: Final = "CA certificate"
+CLIENT_CERT_LABEL: Final = "client certificate"
+CLIENT_KEY_LABEL: Final = "client key"
+
 # Connection error keywords for retry logic
-CONNECTION_ERROR_KEYWORDS: Final[frozenset[str]] = frozenset(
+CONNECTION_ERROR_KEYWORDS: Final[frozenset[str]] = frozenset[str](
     {
         "connection",
         "connect",

ddcdatabases/core/operations.py

@@ -30,6 +30,7 @@ def __init__(self, session: Session, retry_config: BaseOperationRetryConfig | No
 
     def _execute_with_retry[T](self, operation: Callable[[], T], operation_name: str) -> T:
         """Execute an operation with retry logic if enabled."""
+
         if self.retry_config.enable_retry:
             return retry_operation(operation, self.retry_config, operation_name)
         return operation()
@@ -45,7 +46,7 @@ def _fetchall_impl(self, stmt: Any, as_dict: bool = False) -> list[RowMapping] |
             else:
                 result = cursor.mappings().all()
                 cursor.close()
-                return list(result)
+                return list[RowMapping](result)
         except Exception as e:
             self.session.rollback()
             _logger.exception("fetchall failed")
@@ -65,6 +66,7 @@ def fetchall(self, stmt: Any, as_dict: bool = False) -> list[RowMapping] | list[
         Raises:
             DBFetchAllException: If query execution fails
         """
+
         return self._execute_with_retry(lambda: self._fetchall_impl(stmt, as_dict), "fetchall")
 
     def _fetchvalue_impl(self, stmt: Any) -> Any:
@@ -95,6 +97,7 @@ def fetchvalue(self, stmt: Any) -> Any:
         Raises:
             DBFetchValueException: If query execution fails
         """
+
         return self._execute_with_retry(lambda: self._fetchvalue_impl(stmt), "fetchvalue")
 
     def _insert_impl(self, stmt: Any) -> Any:
@@ -121,6 +124,7 @@ def insert(self, stmt: Any) -> Any:
         Raises:
             DBInsertSingleException: If insert operation fails
         """
+
         return self._execute_with_retry(lambda: self._insert_impl(stmt), "insert")
 
     def _insertbulk_impl[T](self, model: type[T], list_data: Sequence[dict[str, Any]], batch_size: int = 1000) -> None:
@@ -153,6 +157,7 @@ def insertbulk[T](self, model: type[T], list_data: Sequence[dict[str, Any]], bat
         Raises:
             DBInsertBulkException: If bulk insert operation fails
         """
+
         return self._execute_with_retry(lambda: self._insertbulk_impl(model, list_data, batch_size), "insertbulk")
 
     def _deleteall_impl[T](self, model: type[T]) -> None:
@@ -176,6 +181,7 @@ def deleteall[T](self, model: type[T]) -> None:
         Raises:
             DBDeleteAllDataException: If delete operation fails
         """
+
         return self._execute_with_retry(lambda: self._deleteall_impl(model), "deleteall")
 
     def _execute_impl(self, stmt: Any) -> None:
@@ -197,6 +203,7 @@ def execute(self, stmt: Any) -> None:
         Raises:
             DBExecuteException: If statement execution fails
         """
+
         return self._execute_with_retry(lambda: self._execute_impl(stmt), "execute")
 
 
@@ -209,6 +216,7 @@ def __init__(self, session: AsyncSession, retry_config: BaseOperationRetryConfig
 
     async def _execute_with_retry(self, operation: Callable[[], Any], operation_name: str) -> Any:
         """Execute an async operation with retry logic if enabled."""
+
         if self.retry_config.enable_retry:
             return await retry_operation_async(operation, self.retry_config, operation_name)
         return await operation()
@@ -224,7 +232,7 @@ async def _fetchall_impl(self, stmt: Any, as_dict: bool = False) -> list[RowMapp
             else:
                 result = cursor.mappings().all()
                 cursor.close()
-                return list(result)
+                return list[RowMapping](result)
         except Exception as e:
             await self.session.rollback()
             _logger.exception("async fetchall failed")
@@ -244,6 +252,7 @@ async def fetchall(self, stmt: Any, as_dict: bool = False) -> list[RowMapping] |
         Raises:
             DBFetchAllException: If query execution fails
         """
+
         return await self._execute_with_retry(lambda: self._fetchall_impl(stmt, as_dict), "fetchall")
 
     async def _fetchvalue_impl(self, stmt: Any) -> Any:
@@ -274,6 +283,7 @@ async def fetchvalue(self, stmt: Any) -> Any:
         Raises:
             DBFetchValueException: If query execution fails
         """
+
         return await self._execute_with_retry(lambda: self._fetchvalue_impl(stmt), "fetchvalue")
 
     async def _insert_impl(self, stmt: Any) -> Any:
@@ -300,10 +310,14 @@ async def insert(self, stmt: Any) -> Any:
         Raises:
             DBInsertSingleException: If insert operation fails
         """
+
         return await self._execute_with_retry(lambda: self._insert_impl(stmt), "insert")
 
     async def _insertbulk_impl[T](
-        self, model: type[T], list_data: Sequence[dict[str, Any]], batch_size: int = 1000
+        self,
+        model: type[T],
+        list_data: Sequence[dict[str, Any]],
+        batch_size: int = 1000,
     ) -> None:
         try:
             if not list_data:
@@ -338,6 +352,7 @@ async def insertbulk[T](self, model: type[T], list_data: Sequence[dict[str, Any]
         Raises:
             DBInsertBulkException: If bulk insert operation fails
         """
+
         return await self._execute_with_retry(lambda: self._insertbulk_impl(model, list_data, batch_size), "insertbulk")
 
     async def _deleteall_impl[T](self, model: type[T]) -> None:
@@ -362,6 +377,7 @@ async def deleteall[T](self, model: type[T]) -> None:
         Raises:
             DBDeleteAllDataException: If delete operation fails
         """
+
         return await self._execute_with_retry(lambda: self._deleteall_impl(model), "deleteall")
 
     async def _execute_impl(self, stmt: Any) -> None:
@@ -383,4 +399,5 @@ async def execute(self, stmt: Any) -> None:
         Raises:
             DBExecuteException: If statement execution fails
         """
+
         return await self._execute_with_retry(lambda: self._execute_impl(stmt), "execute")