Merge pull request #4526 from opsmill/pog-http-adapter-custom-ca

Add support for custom CA bundles to http.adapter

backend/infrahub/config.py

@@ -2,6 +2,7 @@
 
 import os
 import os.path
+import ssl
 import sys
 from dataclasses import dataclass
 from enum import Enum
@@ -329,6 +330,45 @@ class HTTPSettings(BaseSettings):
         default=False,
         description="Indicates if Infrahub will validate server certificates or if the validation is ignored.",
     )
+    tls_ca_bundle: str | None = Field(
+        default=None,
+        description="Custom CA bundle in PEM format. The value should either be the CA bundle as a string, alternatively as a file path.",
+    )
+
+    @model_validator(mode="after")
+    def set_tls_context(self) -> Self:
+        try:
+            # Validate that the context can be created, we want to raise this error during application start
+            # instead of running into issues later when we first try to use the tls context.
+            self.get_tls_context()
+        except ssl.SSLError as exc:
+            raise ValueError(f"Unable load CA bundle from {self.tls_ca_bundle}: {exc}") from exc
+
+        return self
+
+    def get_tls_context(self) -> ssl.SSLContext:
+        if self.tls_insecure:
+            return ssl._create_unverified_context()
+
+        if not self.tls_ca_bundle:
+            return ssl.create_default_context()
+
+        tls_ca_path = Path(self.tls_ca_bundle)
+
+        try:
+            possibly_file = tls_ca_path.exists()
+        except OSError:
+            # Raised if the filename is too long which can indicate
+            # that the value is a PEM certificate in string form.
+            possibly_file = False
+
+        if possibly_file and tls_ca_path.is_file():
+            context = ssl.create_default_context(cafile=str(tls_ca_path))
+        else:
+            context = ssl.create_default_context()
+            context.load_verify_locations(cadata=self.tls_ca_bundle)
+
+        return context
 
 
 class InitialSettings(BaseSettings):

backend/infrahub/services/adapters/http/httpx.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import ssl
+from functools import cached_property
 from typing import TYPE_CHECKING, Any
 
 import httpx
@@ -22,14 +23,19 @@ async def initialize(self, service: InfrahubServices) -> None:
         self.service = service
         self.settings = config.SETTINGS.http
 
-    def verify_tls(self, verify: bool | None = None) -> bool:
-        if verify is not None:
-            return verify
+        # Cache the context during init, this is to avoid issue when a CA bundle might be accessible
+        # when Infrahub initializes but then removed before the first external HTTP call is made.
+        _ = self.tls_context
 
-        if self.settings.tls_insecure is True:
+    @cached_property
+    def tls_context(self) -> ssl.SSLContext:
+        return self.settings.get_tls_context()
+
+    def verify_tls(self, verify: bool | None = None) -> bool | ssl.SSLContext:
+        if verify is False:
             return False
 
-        return True
+        return self.tls_context
 
     async def _request(
         self,

pyproject.toml

@@ -605,6 +605,10 @@ allow-dunder-method-names = [
     "ANN401", # Dynamically typed expressions (typing.Any) are disallowed
 ]
 
+"backend/infrahub/config.py" = [
+    "S323",   # Allow users to create an SSL context that doesn't validate certificates
+]
+
 "backend/infrahub/graphql/mutations/**.py" = [
     ##################################################################################################
     # Review and change the below later                                                              #