feat(centreon): support TLS verification and mTLS client certificates

docs/snippets/providers/centreon-snippet-autogenerated.mdx

@@ -1,10 +1,14 @@
-{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py 
+{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py
 Do not edit it manually, as it will be overwritten */}
 
 ## Authentication
 This provider requires authentication.
 - **host_url**: Centreon Host URL (required: True, sensitive: False)
 - **api_token**: Centreon API Token (required: True, sensitive: True)
+- **verify**: Verify SSL certificates (required: False, sensitive: False)
+- **ca_certificate**: CA certificate (PEM) used to verify the Centreon server (required: False, sensitive: True)
+- **client_certificate**: Client certificate (PEM) for mutual TLS (required: False, sensitive: True)
+- **client_key**: Client private key (PEM) for mutual TLS (required: False, sensitive: True)
 
 Certain scopes may be required to perform specific actions or queries via the provider. Below is a summary of relevant scopes and their use cases:
 - **authenticated**: User is authenticated  

keep/providers/centreon_provider/centreon_provider.py

@@ -2,8 +2,11 @@
 Centreon is a class that provides a set of methods to interact with the Centreon API.
 """
 
+import contextlib
 import dataclasses
 import datetime
+import os
+import tempfile
 
 import pydantic
 import requests
@@ -39,6 +42,50 @@ class CentreonProviderAuthConfig:
         default=None,
     )
 
+    verify: bool = dataclasses.field(
+        default=True,
+        metadata={
+            "required": False,
+            "description": "Verify SSL certificates",
+            "hint": "Set to false to allow self-signed certificates",
+            "sensitive": False,
+            "type": "switch",
+        },
+    )
+
+    ca_certificate: str = dataclasses.field(
+        default="",
+        metadata={
+            "name": "ca_certificate",
+            "description": "CA certificate (PEM) used to verify the Centreon server",
+            "required": False,
+            "sensitive": True,
+            "type": "file",
+        },
+    )
+
+    client_certificate: str = dataclasses.field(
+        default="",
+        metadata={
+            "name": "client_certificate",
+            "description": "Client certificate (PEM) for mutual TLS",
+            "required": False,
+            "sensitive": True,
+            "type": "file",
+        },
+    )
+
+    client_key: str = dataclasses.field(
+        default="",
+        metadata={
+            "name": "client_key",
+            "description": "Client private key (PEM) for mutual TLS",
+            "required": False,
+            "sensitive": True,
+            "type": "file",
+        },
+    )
+
 
 class CentreonProvider(BaseProvider):
     PROVIDER_DISPLAY_NAME = "Centreon"
@@ -93,15 +140,54 @@ def __get_headers(self):
             "centreon-auth-token": self.authentication_config.api_token,
         }
 
+    @contextlib.contextmanager
+    def __request_kwargs(self):
+        """Yield the TLS keyword arguments for a requests call.
+
+        The CA/client cert fields hold PEM content, so they are written to
+        temporary files for the duration of the request (requests expects paths)
+        and removed afterwards.
+        """
+        temp_paths = []
+
+        def _materialize(content: str) -> str:
+            temp = tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False)
+            temp.write(content)
+            temp.close()
+            temp_paths.append(temp.name)
+            return temp.name
+
+        kwargs = {}
+        auth = self.authentication_config
+        if not auth.verify:
+            kwargs["verify"] = False
+        elif auth.ca_certificate:
+            kwargs["verify"] = _materialize(auth.ca_certificate)
+        if auth.client_certificate and auth.client_key:
+            kwargs["cert"] = (
+                _materialize(auth.client_certificate),
+                _materialize(auth.client_key),
+            )
+        try:
+            yield kwargs
+        finally:
+            for path in temp_paths:
+                try:
+                    os.unlink(path)
+                except OSError:
+                    pass
+
     def validate_scopes(self) -> dict[str, bool | str]:
         """
         Validate the scopes of the provider.
         """
         try:
-            response = requests.get(
-                self.__get_url("object=centreon_realtime_hosts&action=list"),
-                headers=self.__get_headers(),
-            )
+            with self.__request_kwargs() as request_kwargs:
+                response = requests.get(
+                    self.__get_url("object=centreon_realtime_hosts&action=list"),
+                    headers=self.__get_headers(),
+                    **request_kwargs,
+                )
             if response.ok:
                 scopes = {"authenticated": True}
             else:
@@ -118,7 +204,10 @@ def validate_scopes(self) -> dict[str, bool | str]:
     def __get_host_status(self) -> list[AlertDto]:
         try:
             url = self.__get_url("object=centreon_realtime_hosts&action=list")
-            response = requests.get(url, headers=self.__get_headers())
+            with self.__request_kwargs() as request_kwargs:
+                response = requests.get(
+                    url, headers=self.__get_headers(), **request_kwargs
+                )
 
             if not response.ok:
                 self.logger.error(
@@ -154,7 +243,10 @@ def __get_host_status(self) -> list[AlertDto]:
     def __get_service_status(self) -> list[AlertDto]:
         try:
             url = self.__get_url("object=centreon_realtime_services&action=list")
-            response = requests.get(url, headers=self.__get_headers())
+            with self.__request_kwargs() as request_kwargs:
+                response = requests.get(
+                    url, headers=self.__get_headers(), **request_kwargs
+                )
 
             if not response.ok:
                 self.logger.error(

tests/providers/centreon_provider/test_centreon_tls.py

@@ -0,0 +1,127 @@
+"""Tests for TLS/mTLS options on the Centreon provider (issue #3340).
+
+The provider talks to an on-prem Centreon over HTTPS, which commonly uses a
+self-signed or internal-CA certificate. These tests cover the verify switch and
+the CA / client-certificate wiring through to the underlying requests call.
+"""
+
+import os
+from unittest.mock import MagicMock, patch
+
+from keep.contextmanager.contextmanager import ContextManager
+from keep.providers.centreon_provider.centreon_provider import CentreonProvider
+from keep.providers.models.provider_config import ProviderConfig
+
+REQUESTS_GET = "keep.providers.centreon_provider.centreon_provider.requests.get"
+
+CA_PEM = "-----BEGIN CERTIFICATE-----\nca-content\n-----END CERTIFICATE-----\n"
+CLIENT_PEM = "-----BEGIN CERTIFICATE-----\nclient-content\n-----END CERTIFICATE-----\n"
+KEY_PEM = "-----BEGIN PRIVATE KEY-----\nkey-content\n-----END PRIVATE KEY-----\n"
+
+
+def _make_provider(**auth):
+    context_manager = ContextManager(tenant_id="test", workflow_id="test")
+    config = ProviderConfig(
+        authentication={
+            "host_url": "https://centreon.internal",
+            "api_token": "token",
+            **auth,
+        },
+        name="test-centreon",
+    )
+    return CentreonProvider(context_manager, "centreon", config)
+
+
+def _capturing_get():
+    """A requests.get replacement that records the TLS kwargs and reads the
+    temporary cert files while they still exist (they are removed after the call)."""
+    captured = {}
+
+    def fake_get(url, headers=None, **kwargs):
+        captured["verify"] = kwargs.get("verify", "UNSET")
+        captured["cert"] = kwargs.get("cert")
+        verify = kwargs.get("verify")
+        if isinstance(verify, str):
+            captured["verify_path"] = verify
+            with open(verify) as fh:
+                captured["ca_content"] = fh.read()
+        cert = kwargs.get("cert")
+        if cert:
+            captured["cert_paths"] = cert
+            with open(cert[0]) as fh:
+                captured["client_cert_content"] = fh.read()
+            with open(cert[1]) as fh:
+                captured["client_key_content"] = fh.read()
+        response = MagicMock()
+        response.ok = True
+        response.status_code = 200
+        response.json.return_value = []
+        return response
+
+    return fake_get, captured
+
+
+def test_verify_true_by_default():
+    provider = _make_provider()
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    # requests default verify is True, so no explicit override is passed
+    assert captured["verify"] == "UNSET"
+    assert captured["cert"] is None
+
+
+def test_verify_false_skips_tls_verification():
+    provider = _make_provider(verify=False)
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    assert captured["verify"] is False
+
+
+def test_ca_certificate_passed_as_file_path():
+    provider = _make_provider(ca_certificate=CA_PEM)
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    assert isinstance(captured["verify"], str)
+    assert captured["ca_content"] == CA_PEM
+
+
+def test_verify_false_takes_precedence_over_ca_certificate():
+    provider = _make_provider(verify=False, ca_certificate=CA_PEM)
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    assert captured["verify"] is False
+
+
+def test_client_certificate_enables_mutual_tls():
+    provider = _make_provider(client_certificate=CLIENT_PEM, client_key=KEY_PEM)
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    assert captured["cert"] is not None
+    assert captured["client_cert_content"] == CLIENT_PEM
+    assert captured["client_key_content"] == KEY_PEM
+
+
+def test_client_certificate_without_key_is_ignored():
+    provider = _make_provider(client_certificate=CLIENT_PEM)
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    assert captured["cert"] is None
+
+
+def test_temp_cert_files_are_cleaned_up():
+    provider = _make_provider(
+        ca_certificate=CA_PEM, client_certificate=CLIENT_PEM, client_key=KEY_PEM
+    )
+    fake_get, captured = _capturing_get()
+    with patch(REQUESTS_GET, side_effect=fake_get):
+        provider.validate_scopes()
+    # paths existed during the call but must be removed afterwards
+    assert not os.path.exists(captured["verify_path"])
+    for path in captured["cert_paths"]:
+        assert not os.path.exists(path)