feat: Add custom CA file support for verify_ssl (#292)

* Add custom CA file support for `api.verify_ssl` config option

* Fix Union types in config docs

Author: pederhan

CHANGELOG

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Support for custom CA file bundles for Zabbix API connections. The config option `api.verify_ssl` now accepts a path to a custom CA file bundle.
+
 ### Fixed
 
 - `create_maintenance_definition` with multiple host groups only including the first group in the maintenance definition for Zabbix >=6.0.

docs/scripts/gen_config_data.py

@@ -165,6 +165,8 @@ def validate_default(cls, value: Any) -> Optional[Any]:
             return None
         if isinstance(value, SecretStr):
             return value.get_secret_value()
+        if isinstance(value, bool):
+            return str(value).lower()
         return value
 
     @field_validator("type", mode="before")
@@ -175,6 +177,24 @@ def validate_type(cls, value: Any) -> str:
 
         origin = get_origin(value)
         args = get_args(value)
+
+        def type_to_str(t: type[Any]) -> str:
+            if lenient_issubclass(value, str):
+                return "str"
+            if lenient_issubclass(value, Enum):
+                # Get the name of the first enum member type
+                # Will fail if enum has no members
+                return str(list(value)[0])  # pyright: ignore[reportUnknownArgumentType]
+            # Types that are represented as strings in config (paths, secrets, etc.)
+            if typ := TYPE_MAP.get(t):
+                return typ
+            # Primitives and built-in generics (str, int, list[str], dict[str, int], etc.)
+            if origin in TYPE_CAN_STR:
+                return str(value)
+            # Fall back on the string representation of the type
+            return getattr(value, "__name__", str(value))
+
+        # Handle generics, literals, etc.
         if origin and args:
             # Get the name of the first type in the Literal type
             # NOTE: we expect that Literal is only used with a single type
@@ -183,26 +203,11 @@ def validate_type(cls, value: Any) -> str:
             # Get first non-None type in Union
             # NOTE: we expect that the config does not have unions of more than 2 types
             elif origin is Union and args:
-                return next(a.__name__ for a in args if a is not type(None))
-
-        if lenient_issubclass(value, str):
-            return "str"
-
-        if lenient_issubclass(value, Enum):
-            # Get the name of the first enum member type
-            # Will fail if enum has no members
-            return str(list(value)[0])  # pyright: ignore[reportUnknownArgumentType]
-
-        # Types that are represented as strings in config (paths, secrets, etc.)
-        if t := TYPE_MAP.get(value):
-            return t
-
-        # Primitives and built-in generics (str, int, list[str], dict[str, int], etc.)
-        if origin in TYPE_CAN_STR:
-            return str(value)
+                # Strip None from the Union
+                ar = (type_to_str(a) for a in args if a is not type(None))
+                return " | ".join(ar)
 
-        # Fall back on the string representation of the type
-        return getattr(value, "__name__", str(value))
+        return type_to_str(value)
 
     @field_validator("description", mode="before")
     @classmethod

zabbix_cli/config/model.py

@@ -128,11 +128,11 @@ class APIConfig(BaseModel):
         description="API auth token.",
         examples=["API_TOKEN_123"],
     )
-    verify_ssl: bool = Field(
+    verify_ssl: Union[bool, Path] = Field(
         default=True,
         # Changed in V3: cert_verify -> verify_ssl
         validation_alias=AliasChoices("verify_ssl", "cert_verify"),
-        description="Verify SSL certificate of the Zabbix API host.",
+        description="Verify SSL certificate of the Zabbix API host. Can also be a path to a CA bundle.",
     )
     timeout: Optional[int] = Field(
         default=0,

zabbix_cli/pyzabbix/client.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import logging
+import ssl
 from collections.abc import MutableMapping
 from datetime import datetime
 from functools import cached_property
@@ -242,12 +243,12 @@ def __init__(
         server: str = "http://localhost/zabbix",
         *,
         timeout: Optional[int] = None,
-        verify_ssl: bool = True,
+        verify_ssl: Union[bool, Path] = True,
     ) -> None:
         """Parameters:
         server: Base URI for zabbix web interface (omitting /api_jsonrpc.php)
-        session: optional pre-configured requests.Session instance
-        timeout: optional connect and read timeout in seconds.
+        timeout: Read and connect timeout for HTTP requests in seconds.
+        verify_ssl: Verify SSL certificates. Can be a boolean or a path to a CA bundle.
         """
         self.timeout = timeout if timeout else None
         self.session = self._get_client(verify_ssl=verify_ssl, timeout=timeout)
@@ -279,14 +280,28 @@ def from_config(cls, config: Config) -> ZabbixAPI:
         )
         return client
 
+    def _get_ssl_context(
+        self, verify_ssl: Union[bool, Path]
+    ) -> Union[ssl.SSLContext, bool]:
+        if isinstance(verify_ssl, Path):
+            if not verify_ssl.exists():
+                raise ValueError(f"CA bundle not found: {verify_ssl}")
+            if verify_ssl.is_dir():
+                ctx = ssl.create_default_context(capath=verify_ssl)
+            else:
+                ctx = ssl.create_default_context(cafile=verify_ssl)
+        else:
+            ctx = verify_ssl
+        return ctx
+
     def _get_client(
-        self, *, verify_ssl: bool, timeout: Union[float, int, None] = None
+        self, *, verify_ssl: Union[bool, Path], timeout: Union[float, int, None] = None
     ) -> httpx.Client:
         kwargs: HTTPXClientKwargs = {}
         if timeout is not None:
             kwargs["timeout"] = timeout
         client = httpx.Client(
-            verify=verify_ssl,
+            verify=self._get_ssl_context(verify_ssl),
             # Default headers for all requests
             headers={
                 "Content-Type": "application/json-rpc",