Add support for custom ca certificate

[PedroMSantosD]

Is your feature request related to a problem? Please describe.

I would like the connection to trust custom root certificate bundles passed to it, as opposed to current contructor : 'disable_ssl=True'

Describe the solution you'd like

A clear and concise description of what you want to happen.
I'd like the client constructor to support accetping the path to the certificate bundle (trusted root Ca being self-signed).
i.e.

class PrometheusConnect:
    def __init__(
        self,
        url: str = "http://127.0.0.1:9090",
        headers: dict = None,
        disable_ssl: bool = False,
        custom_ca_cert: str = None,  # Custom CA certificate path

Describe alternatives you've considered

A clear and concise description of any alternative solutions or features you've considered.
I'd like to avoid injecting the ca certificate bundle on the container/computer trusted store (ie. sudo update-ca-certificates).

I wouldn't mind adding said path as an environment varible if it is considered (ie. REQUESTS_CA_BUNDLE = 'path_to_bundle.pem').

Additional context

N/A just trust a self-signed certificate is passed to the constructor/env variable

chatgpt suggests something like

import ssl
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.poolmanager import PoolManager
from requests import Session

class SSLAdapter(HTTPAdapter):
    """Custom adapter to pass an SSL context for custom CA certificates."""
    def __init__(self, ssl_context=None, **kwargs):
        self.ssl_context = ssl_context
        super().__init__(**kwargs)

    def init_poolmanager(self, *args, **kwargs):
        kwargs['ssl_context'] = self.ssl_context
        return super().init_poolmanager(*args, **kwargs)

class PrometheusConnect:
    def __init__(
        self,
        url: str = "http://127.0.0.1:9090",
        headers: dict = None,
        disable_ssl: bool = False,
        custom_ca_cert: str = None,  # Custom CA certificate path
        retry: Retry = None,
        auth: tuple = None,
        proxy: dict = None,
        session: Session = None,
        timeout: int = None,
    ):
        self.url = url
        self.headers = headers
        self.disable_ssl = disable_ssl
        self.custom_ca_cert = custom_ca_cert
        self.retry = retry
        self.auth = auth
        self.proxy = proxy
        self.session = session if session else Session()
        self.timeout = timeout

        # Set custom SSL configuration if CA cert is provided
        if custom_ca_cert:
            self._set_custom_ssl_context(custom_ca_cert)

    def _set_custom_ssl_context(self, ca_cert_path: str):
        """Sets up the custom SSL context to trust the provided CA certificate."""
        ssl_context = ssl.create_default_context(cafile=ca_cert_path)

        # Mount the custom adapter to the session
        ssl_adapter = SSLAdapter(ssl_context=ssl_context)
        self.session.mount('https://', ssl_adapter)

    def _make_request(self, url: str):
        """Example method to make a request to the Prometheus instance."""
        try:
            response = self.session.get(url, headers=self.headers, timeout=self.timeout, proxies=self.proxy)
            response.raise_for_status()  # Will raise an error for bad status codes
            return response.json()
        except requests.RequestException as e:
            print(f"Error: {e}")
            return None

# Example usage:
prometheus = PrometheusConnect(
    url="https://your-prometheus-url",
    custom_ca_cert="/path/to/your/custom-ca-cert.pem",  # Path to custom CA certificate
)

response = prometheus._make_request("https://your-prometheus-url/api/v1/query")
print(response)

But I'm not sure how should the library be improved.

Thanks in advance.

[s3714110]

Hi, not sure if you have found a solution for this, but there's a much simpler way to add a CA cert to verify all requests using this library without modifying the source code.

All you need to do is create your own Session object, add the CA cert path to it, and pass that Session when initialising a new PrometheusConnect object

from requests import Session

ca_path = "path/to/your/ca/cert"

session = Session()
session.verify = ca_path 

pc = PrometheusConnect(
       url="https://your-prometheus-url",
       session=session
)