fix(sdk): use system trust store for push-to-cloud

- Combine native and configured certificate authorities
- Scope TLS changes to the ingestion session
- Add regression tests and documentation

docs/user-guide/tutorials/prowler-import-findings.mdx

@@ -153,6 +153,16 @@ export PROWLER_CLOUD_API_KEY="pk_your_api_key_here"
 prowler aws --push-to-cloud
 ```
 
+### TLS Certificate Trust
+
+For `--push-to-cloud` uploads, Prowler CLI creates one ingestion-scoped TLS context and validates HTTPS certificates with one handshake and one POST request. The upload does not retry or fall back to another TLS configuration. Redirect responses are rejected.
+
+The ingestion context combines the operating system roots with the previous Requests default certificate authority (CA) roots. If `REQUESTS_CA_BUNDLE` is configured, Prowler CLI adds that file or directory to the same context. Otherwise, Prowler CLI adds `CURL_CA_BUNDLE` when configured. These variables add roots only for this ingestion session and are not required when the CA is already installed in the operating system store.
+
+For Prowler Private Cloud deployments that use an organization CA or a TLS-intercepting corporate proxy, installing the required root CA in the operating system or container store remains the recommended approach. Containers have an isolated system CA store, so add the organization or proxy CA to the container image or runtime, then run the operating system's CA update command, such as `update-ca-certificates`, before starting Prowler CLI. Installing a CA on the container host does not automatically install it inside the container.
+
+This trust configuration applies only to the temporary `push-to-cloud` ingestion session. It does not change API, provider, integration, global SSL, environment variable, or unrelated Requests session behavior.
+
 ### Combining with Output Formats
 
 When using `--push-to-cloud` with custom output formats that exclude OCSF, Prowler generates a temporary OCSF file for upload:

prowler/__main__.py

@@ -126,7 +126,10 @@ from prowler.lib.outputs.compliance.prowler_threatscore.prowler_threatscore_m365
 from prowler.lib.outputs.csv.csv import CSV
 from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.html.html import HTML
-from prowler.lib.outputs.ocsf.ingestion import send_ocsf_to_api
+from prowler.lib.outputs.ocsf.ingestion import (
+    SystemTrustStoreError,
+    send_ocsf_to_api,
+)
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.lib.outputs.outputs import extract_findings_statistics, report
 from prowler.lib.outputs.sarif.sarif import SARIF
@@ -162,6 +165,60 @@ from prowler.providers.stackit.models import StackITOutputOptions
 from prowler.providers.vercel.models import VercelOutputOptions
 
 
+def _send_ocsf_to_cloud(file_path: str) -> dict | None:
+    """Upload OCSF findings and report safe, actionable CLI failures."""
+    try:
+        return send_ocsf_to_api(file_path)
+    except ValueError:
+        print(
+            f"{Style.BRIGHT}{Fore.YELLOW}\nPush to Prowler Cloud skipped: no API key configured. "
+            "Set the PROWLER_CLOUD_API_KEY environment variable to enable it. "
+            f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+        )
+    except SystemTrustStoreError:
+        print(
+            f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: the operating system trust store "
+            "could not be initialized. Check the configured CA bundle paths, verify the host certificate "
+            "hostname, validity period, and certificate chain, and install the organization or TLS-intercepting "
+            "proxy CA in the operating system trust store. In containers, configure and update the container "
+            "system CA store. "
+            f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+        )
+    except requests.exceptions.SSLError:
+        print(
+            f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: TLS certificate validation failed. "
+            "Verify the hostname, validity period, and certificate chain, and install the organization or "
+            "TLS-intercepting proxy CA in the operating system trust store. In containers, configure and update "
+            "the container system CA store. "
+            f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+        )
+    except requests.ConnectionError:
+        print(
+            f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: could not reach the Prowler Cloud API at "
+            f"{cloud_api_base_url}. Check the URL and your network connection. "
+            f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+        )
+    except requests.HTTPError as http_err:
+        if http_err.response.status_code == 402:
+            print(
+                f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: "
+                "this feature is only available with a Prowler Cloud subscription. "
+                f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+            )
+        else:
+            print(
+                f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: the API returned HTTP "
+                f"{http_err.response.status_code}. Verify your API key is valid and has the right permissions. "
+                f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+            )
+    except Exception:
+        print(
+            f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed unexpectedly. "
+            f"Scan results were saved to {file_path}{Style.RESET_ALL}"
+        )
+    return None
+
+
 def prowler():
     # Parse Arguments
     # Refactor(CLI)
@@ -641,39 +698,8 @@ def prowler():
         print(
             f"{Style.BRIGHT}\nPushing findings to Prowler Cloud, please wait...{Style.RESET_ALL}"
         )
-        try:
-            response = send_ocsf_to_api(ocsf_output.file_path)
-        except ValueError:
-            print(
-                f"{Style.BRIGHT}{Fore.YELLOW}\nPush to Prowler Cloud skipped: no API key configured. "
-                "Set the PROWLER_CLOUD_API_KEY environment variable to enable it. "
-                f"Scan results were saved to {ocsf_output.file_path}{Style.RESET_ALL}"
-            )
-        except requests.ConnectionError:
-            print(
-                f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: could not reach the Prowler Cloud API at "
-                f"{cloud_api_base_url}. Check the URL and your network connection. "
-                f"Scan results were saved to {ocsf_output.file_path}{Style.RESET_ALL}"
-            )
-        except requests.HTTPError as http_err:
-            if http_err.response.status_code == 402:
-                print(
-                    f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: "
-                    "this feature is only available with a Prowler Cloud subscription. "
-                    f"Scan results were saved to {ocsf_output.file_path}{Style.RESET_ALL}"
-                )
-            else:
-                print(
-                    f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed: the API returned HTTP {http_err.response.status_code}. "
-                    "Verify your API key is valid and has the right permissions. "
-                    f"Scan results were saved to {ocsf_output.file_path}{Style.RESET_ALL}"
-                )
-        except Exception as error:
-            print(
-                f"{Style.BRIGHT}{Fore.RED}\nPush to Prowler Cloud failed unexpectedly: {error}. "
-                f"Scan results were saved to {ocsf_output.file_path}{Style.RESET_ALL}"
-            )
-        else:
+        response = _send_ocsf_to_cloud(ocsf_output.file_path)
+        if response is not None:
             job_id = response.get("data", {}).get("id") if response else None
             if job_id:
                 print(

prowler/changelog.d/push-to-cloud-system-trust.fixed.md

@@ -0,0 +1 @@
+`push-to-cloud` now validates Private Cloud TLS certificates with the operating system trust store without changing provider HTTP clients

prowler/lib/outputs/ocsf/ingestion.py

@@ -1,7 +1,9 @@
 import os
+import ssl
 from typing import Any, Dict, Optional
 
 import requests
+from requests.adapters import HTTPAdapter
 
 from prowler.config.config import (
     cloud_api_base_url,
@@ -10,6 +12,101 @@ from prowler.config.config import (
 )
 
 
+class SystemTrustStoreError(RuntimeError):
+    """Raised when the operating system trust store cannot be initialized."""
+
+
+def _load_ca_bundle(ssl_context: ssl.SSLContext, ca_bundle_path: str) -> None:
+    """Add a CA bundle file or directory to an existing TLS context."""
+    if os.path.isfile(ca_bundle_path):
+        ssl_context.load_verify_locations(cafile=ca_bundle_path)
+    elif os.path.isdir(ca_bundle_path):
+        ssl_context.load_verify_locations(capath=ca_bundle_path)
+    else:
+        raise FileNotFoundError
+
+
+class _SystemTrustHTTPAdapter(HTTPAdapter):
+    """Use one combined trust context for HTTPS origin connection pools."""
+
+    def __init__(self, ssl_context: ssl.SSLContext) -> None:
+        self._ssl_context = ssl_context
+        super().__init__()
+
+    def build_connection_pool_key_attributes(
+        self,
+        request: requests.PreparedRequest,
+        verify: Any,
+        cert: Any = None,
+    ) -> tuple[Dict[str, Any], Dict[str, Any]]:
+        verify = True
+        host_params, pool_kwargs = super().build_connection_pool_key_attributes(
+            request, verify, cert
+        )
+        pool_kwargs["ssl_context"] = self._ssl_context
+        pool_kwargs["cert_reqs"] = "CERT_REQUIRED"
+        pool_kwargs.pop("ca_certs", None)
+        pool_kwargs.pop("ca_cert_dir", None)
+        return host_params, pool_kwargs
+
+    def cert_verify(self, conn: Any, url: str, verify: Any, cert: Any) -> None:
+        """Keep certificate verification in the supplied native context."""
+        if verify is not True:
+            raise ValueError("TLS certificate verification is required for ingestion.")
+        conn.cert_reqs = "CERT_REQUIRED"
+        conn.ca_certs = None
+        conn.ca_cert_dir = None
+
+    def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any:
+        """Use native trust for HTTPS proxies without changing SOCKS support."""
+        if proxy.lower().startswith("https://"):
+            proxy_kwargs.setdefault("proxy_ssl_context", self._ssl_context)
+        return super().proxy_manager_for(proxy, **proxy_kwargs)
+
+
+class _SystemTrustSession(requests.Session):
+    """Requests session using native trust augmented with compatibility roots."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        try:
+            import truststore
+
+            ssl_context = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+            _load_ca_bundle(ssl_context, requests.certs.where())
+            configured_ca_bundle = os.environ.get(
+                "REQUESTS_CA_BUNDLE"
+            ) or os.environ.get("CURL_CA_BUNDLE")
+            if configured_ca_bundle:
+                _load_ca_bundle(ssl_context, configured_ca_bundle)
+            ssl_context.check_hostname = True
+            ssl_context.verify_mode = ssl.CERT_REQUIRED
+        except Exception as error:
+            self.close()
+            raise SystemTrustStoreError(
+                "Could not initialize the operating system trust store. "
+                "Check the configured CA bundle paths."
+            ) from error
+
+        self.mount("https://", _SystemTrustHTTPAdapter(ssl_context))
+
+    def merge_environment_settings(
+        self,
+        url: str,
+        proxies: Optional[Dict[str, str]],
+        stream: Optional[bool],
+        verify: Any,
+        cert: Any,
+    ) -> Dict[str, Any]:
+        """Preserve environment proxies without replacing the prepared context."""
+        verify = False
+        settings = super().merge_environment_settings(
+            url, proxies, stream, verify, cert
+        )
+        settings["verify"] = True
+        return settings
+
+
 def send_ocsf_to_api(
     file_path: str,
     *,
@@ -32,6 +129,7 @@ def send_ocsf_to_api(
     Raises:
         FileNotFoundError: If the OCSF file does not exist.
         ValueError: If no API key is available.
+        SystemTrustStoreError: If the operating system trust store cannot initialize.
         requests.HTTPError: If the API returns an error status.
     """
     if not file_path:
@@ -53,15 +151,31 @@ def send_ocsf_to_api(
 
     url = f"{base_url}{cloud_api_ingestion_path}"
 
-    with open(file_path, "rb") as fh:
-        response = requests.post(
-            url,
-            headers={
-                "Authorization": f"Api-Key {api_key}",
-                "Accept": "application/vnd.api+json",
-            },
-            files={"file": (os.path.basename(file_path), fh, "application/json")},
-            timeout=timeout,
-        )
-    response.raise_for_status()
-    return response.json() if response.text else {}
+    session = _SystemTrustSession()
+    try:
+        with open(file_path, "rb") as fh:
+            response = session.post(
+                url,
+                headers={
+                    "Authorization": f"Api-Key {api_key}",
+                    "Accept": "application/vnd.api+json",
+                },
+                files={
+                    "file": (
+                        os.path.basename(file_path),
+                        fh,
+                        "application/json",
+                    )
+                },
+                timeout=timeout,
+                allow_redirects=False,
+            )
+        if 300 <= response.status_code < 400:
+            raise requests.HTTPError(
+                f"Prowler Cloud ingestion refused HTTP redirect {response.status_code}.",
+                response=response,
+            )
+        response.raise_for_status()
+        return response.json() if response.text else {}
+    finally:
+        session.close()

pyproject.toml

@@ -103,6 +103,7 @@ dependencies = [
   "stackit-objectstorage==1.4.0",
   "stackit-resourcemanager==0.8.0",
   "tabulate==0.9.0",
+  "truststore==0.10.4",
   "tzlocal==5.3.1",
   "uuid6==2024.7.10",
   "py-iam-expand==0.3.0",

tests/lib/outputs/ocsf/ingestion_cli_test.py

@@ -0,0 +1,95 @@
+from unittest.mock import MagicMock
+
+import pytest
+import requests
+
+from prowler import __main__ as cli
+from prowler.lib.outputs.ocsf.ingestion import SystemTrustStoreError
+
+
+def test_ssl_error_uses_actionable_tls_message_before_connection_error(
+    monkeypatch, capsys
+):
+    upload = MagicMock(
+        side_effect=requests.exceptions.SSLError("secret transport detail")
+    )
+    monkeypatch.setattr(cli, "send_ocsf_to_api", upload)
+
+    response = cli._send_ocsf_to_cloud("/tmp/saved-findings.ocsf.json")
+
+    output = capsys.readouterr().out
+    assert response is None
+    assert "TLS certificate validation failed" in output
+    assert "hostname, validity period, and certificate chain" in output
+    assert "operating system trust store" in output
+    assert "TLS-intercepting proxy CA" in output
+    assert "container system CA store" in output
+    assert "Scan results were saved to /tmp/saved-findings.ocsf.json" in output
+    assert "secret transport detail" not in output
+
+
+def test_truststore_initialization_error_uses_safe_actionable_message(
+    monkeypatch, capsys
+):
+    monkeypatch.setattr(
+        cli,
+        "send_ocsf_to_api",
+        MagicMock(side_effect=SystemTrustStoreError("secret initialization detail")),
+    )
+
+    response = cli._send_ocsf_to_cloud("/tmp/saved-findings.ocsf.json")
+
+    output = capsys.readouterr().out
+    assert response is None
+    assert "operating system trust store could not be initialized" in output
+    assert "configured CA bundle paths" in output
+    assert "TLS-intercepting proxy CA" in output
+    assert "container system CA store" in output
+    assert "Scan results were saved to /tmp/saved-findings.ocsf.json" in output
+    assert "secret initialization detail" not in output
+
+
+@pytest.mark.parametrize(
+    ("error", "expected_message"),
+    [
+        (
+            ValueError("missing"),
+            "Push to Prowler Cloud skipped: no API key configured",
+        ),
+        (
+            requests.ConnectionError("offline"),
+            "could not reach the Prowler Cloud API",
+        ),
+        (
+            requests.HTTPError(response=MagicMock(status_code=402)),
+            "only available with a Prowler Cloud subscription",
+        ),
+        (
+            requests.HTTPError(response=MagicMock(status_code=403)),
+            "the API returned HTTP 403",
+        ),
+    ],
+)
+def test_existing_upload_errors_keep_specific_messages(
+    monkeypatch, capsys, error, expected_message
+):
+    monkeypatch.setattr(cli, "send_ocsf_to_api", MagicMock(side_effect=error))
+
+    response = cli._send_ocsf_to_cloud("/tmp/saved-findings.ocsf.json")
+
+    output = capsys.readouterr().out
+    assert response is None
+    assert expected_message in output
+    assert "Scan results were saved to /tmp/saved-findings.ocsf.json" in output
+
+
+def test_successful_upload_returns_response(monkeypatch, capsys):
+    expected_response = {"data": {"id": "job-id"}}
+    monkeypatch.setattr(
+        cli, "send_ocsf_to_api", MagicMock(return_value=expected_response)
+    )
+
+    response = cli._send_ocsf_to_cloud("/tmp/saved-findings.ocsf.json")
+
+    assert response == expected_response
+    assert capsys.readouterr().out == ""

tests/lib/outputs/ocsf/ingestion_test.py

@@ -0,0 +1,486 @@
+import ssl
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock, call
+
+import pytest
+import requests
+from requests import PreparedRequest
+from requests.adapters import HTTPAdapter
+
+from prowler.lib.outputs.ocsf import ingestion
+
+
+@pytest.fixture
+def system_trust_session(monkeypatch):
+    monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False)
+    monkeypatch.delenv("CURL_CA_BUNDLE", raising=False)
+    native_context = MagicMock()
+    ssl_context = MagicMock(return_value=native_context)
+    inject_into_ssl = MagicMock()
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(
+            SSLContext=ssl_context,
+            inject_into_ssl=inject_into_ssl,
+        ),
+    )
+
+    session = ingestion._SystemTrustSession()
+    yield session, native_context, ssl_context, inject_into_ssl
+    session.close()
+
+
+def test_https_origin_pool_uses_native_trust_context(system_trust_session):
+    session, native_context, ssl_context, _ = system_trust_session
+    adapter = session.get_adapter("https://private.prowler.example")
+    request = PreparedRequest()
+    request.prepare(method="POST", url="https://private.prowler.example/api/v1")
+
+    _, pool_kwargs = adapter.build_connection_pool_key_attributes(request, verify=True)
+
+    ssl_context.assert_called_once_with(ssl.PROTOCOL_TLS_CLIENT)
+    assert pool_kwargs["ssl_context"] is native_context
+    assert pool_kwargs["cert_reqs"] == "CERT_REQUIRED"
+    assert "ca_certs" not in pool_kwargs
+    assert "ca_cert_dir" not in pool_kwargs
+
+
+def test_native_context_loads_requests_default_ca_bundle(monkeypatch, tmp_path):
+    monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False)
+    monkeypatch.delenv("CURL_CA_BUNDLE", raising=False)
+    default_ca_bundle = tmp_path / "requests-default-ca.pem"
+    default_ca_bundle.write_text("default CA")
+    native_context = MagicMock()
+    monkeypatch.setattr(requests.certs, "where", lambda: str(default_ca_bundle))
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(return_value=native_context)),
+    )
+
+    session = ingestion._SystemTrustSession()
+
+    native_context.load_verify_locations.assert_called_once_with(
+        cafile=str(default_ca_bundle)
+    )
+    session.close()
+
+
+def test_native_context_adds_requests_ca_bundle_file(monkeypatch, tmp_path):
+    default_ca_bundle = tmp_path / "requests-default-ca.pem"
+    configured_ca_bundle = tmp_path / "configured-ca.pem"
+    default_ca_bundle.write_text("default CA")
+    configured_ca_bundle.write_text("configured CA")
+    native_context = MagicMock()
+    monkeypatch.setattr(requests.certs, "where", lambda: str(default_ca_bundle))
+    monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(configured_ca_bundle))
+    monkeypatch.delenv("CURL_CA_BUNDLE", raising=False)
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(return_value=native_context)),
+    )
+
+    session = ingestion._SystemTrustSession()
+
+    assert native_context.load_verify_locations.call_args_list == [
+        call(cafile=str(default_ca_bundle)),
+        call(cafile=str(configured_ca_bundle)),
+    ]
+    session.close()
+
+
+def test_native_context_uses_curl_ca_bundle_as_fallback(monkeypatch, tmp_path):
+    default_ca_bundle = tmp_path / "requests-default-ca.pem"
+    configured_ca_bundle = tmp_path / "curl-ca.pem"
+    default_ca_bundle.write_text("default CA")
+    configured_ca_bundle.write_text("configured CA")
+    native_context = MagicMock()
+    monkeypatch.setattr(requests.certs, "where", lambda: str(default_ca_bundle))
+    monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False)
+    monkeypatch.setenv("CURL_CA_BUNDLE", str(configured_ca_bundle))
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(return_value=native_context)),
+    )
+
+    session = ingestion._SystemTrustSession()
+
+    assert native_context.load_verify_locations.call_args_list == [
+        call(cafile=str(default_ca_bundle)),
+        call(cafile=str(configured_ca_bundle)),
+    ]
+    session.close()
+
+
+def test_requests_ca_bundle_takes_precedence_over_curl_ca_bundle(monkeypatch, tmp_path):
+    default_ca_bundle = tmp_path / "requests-default-ca.pem"
+    requests_ca_bundle = tmp_path / "requests-ca.pem"
+    curl_ca_bundle = tmp_path / "curl-ca.pem"
+    for ca_bundle in (default_ca_bundle, requests_ca_bundle, curl_ca_bundle):
+        ca_bundle.write_text("CA")
+    native_context = MagicMock()
+    monkeypatch.setattr(requests.certs, "where", lambda: str(default_ca_bundle))
+    monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(requests_ca_bundle))
+    monkeypatch.setenv("CURL_CA_BUNDLE", str(curl_ca_bundle))
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(return_value=native_context)),
+    )
+
+    session = ingestion._SystemTrustSession()
+
+    assert native_context.load_verify_locations.call_args_list == [
+        call(cafile=str(default_ca_bundle)),
+        call(cafile=str(requests_ca_bundle)),
+    ]
+    session.close()
+
+
+def test_native_context_adds_configured_ca_directory(monkeypatch, tmp_path):
+    default_ca_bundle = tmp_path / "requests-default-ca.pem"
+    configured_ca_directory = tmp_path / "configured-ca-directory"
+    default_ca_bundle.write_text("default CA")
+    configured_ca_directory.mkdir()
+    native_context = MagicMock()
+    monkeypatch.setattr(requests.certs, "where", lambda: str(default_ca_bundle))
+    monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(configured_ca_directory))
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(return_value=native_context)),
+    )
+
+    session = ingestion._SystemTrustSession()
+
+    assert native_context.load_verify_locations.call_args_list == [
+        call(cafile=str(default_ca_bundle)),
+        call(capath=str(configured_ca_directory)),
+    ]
+    session.close()
+
+
+@pytest.mark.parametrize("invalid_bundle_source", ["default", "configured"])
+def test_invalid_ca_bundle_raises_safe_initialization_error(
+    monkeypatch, tmp_path, invalid_bundle_source
+):
+    default_ca_bundle = tmp_path / "requests-default-ca.pem"
+    invalid_ca_bundle = tmp_path / "secret-missing-ca.pem"
+    default_ca_bundle.write_text("default CA")
+    if invalid_bundle_source == "default":
+        monkeypatch.setattr(requests.certs, "where", lambda: str(invalid_ca_bundle))
+        monkeypatch.delenv("REQUESTS_CA_BUNDLE", raising=False)
+    else:
+        monkeypatch.setattr(requests.certs, "where", lambda: str(default_ca_bundle))
+        monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(invalid_ca_bundle))
+    monkeypatch.delenv("CURL_CA_BUNDLE", raising=False)
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(return_value=MagicMock())),
+    )
+
+    with pytest.raises(
+        ingestion.SystemTrustStoreError,
+        match="Check the configured CA bundle paths",
+    ) as error:
+        ingestion._SystemTrustSession()
+
+    assert str(invalid_ca_bundle) not in str(error.value)
+
+
+def test_session_does_not_inject_global_ssl_or_change_default_adapters(
+    system_trust_session,
+):
+    session, _, _, inject_into_ssl = system_trust_session
+    unrelated_session = requests.Session()
+
+    assert type(session.get_adapter("http://private.prowler.example")) is HTTPAdapter
+    assert type(unrelated_session.get_adapter("https://example.com")) is HTTPAdapter
+    assert type(requests.Session().get_adapter("https://example.com")) is HTTPAdapter
+    inject_into_ssl.assert_not_called()
+
+    unrelated_session.close()
+
+
+def test_environment_proxy_is_preserved_without_replacing_prepared_context(
+    monkeypatch, system_trust_session
+):
+    session, _, _, _ = system_trust_session
+    for variable in (
+        "ALL_PROXY",
+        "CURL_CA_BUNDLE",
+        "HTTP_PROXY",
+        "HTTPS_PROXY",
+        "NO_PROXY",
+        "all_proxy",
+        "http_proxy",
+        "https_proxy",
+        "no_proxy",
+    ):
+        monkeypatch.delenv(variable, raising=False)
+    monkeypatch.setenv("HTTPS_PROXY", "https://proxy.example:8443")
+
+    settings = session.merge_environment_settings(
+        "https://private.prowler.example", {}, None, None, None
+    )
+
+    assert settings["verify"] is True
+    assert settings["proxies"]["https"] == "https://proxy.example:8443"
+
+
+def test_https_proxy_uses_native_trust_context(monkeypatch, system_trust_session):
+    session, native_context, _, _ = system_trust_session
+    adapter = session.get_adapter("https://private.prowler.example")
+    proxy_manager = MagicMock()
+    proxy_from_url = MagicMock(return_value=proxy_manager)
+    monkeypatch.setattr(requests.adapters, "proxy_from_url", proxy_from_url)
+
+    result = adapter.proxy_manager_for("https://proxy.example:8443")
+
+    assert result is proxy_manager
+    assert proxy_from_url.call_args.kwargs["proxy_ssl_context"] is native_context
+
+
+def test_socks_proxy_does_not_receive_proxy_ssl_context(
+    monkeypatch, system_trust_session
+):
+    session, _, _, _ = system_trust_session
+    adapter = session.get_adapter("https://private.prowler.example")
+    socks_proxy_manager = MagicMock()
+    socks_proxy_factory = MagicMock(return_value=socks_proxy_manager)
+    monkeypatch.setattr(requests.adapters, "SOCKSProxyManager", socks_proxy_factory)
+
+    result = adapter.proxy_manager_for("socks5h://proxy.example:1080")
+
+    assert result is socks_proxy_manager
+    assert "proxy_ssl_context" not in socks_proxy_factory.call_args.kwargs
+
+
+def test_adapter_keeps_native_verification_without_certifi(system_trust_session):
+    session, _, _, _ = system_trust_session
+    adapter = session.get_adapter("https://private.prowler.example")
+    connection = SimpleNamespace(
+        cert_reqs=None,
+        ca_certs=None,
+        ca_cert_dir=None,
+        cert_file=None,
+        key_file=None,
+    )
+
+    adapter.cert_verify(
+        connection,
+        "https://private.prowler.example",
+        verify=True,
+        cert=None,
+    )
+
+    assert connection.cert_reqs == "CERT_REQUIRED"
+    assert connection.ca_certs is None
+    assert connection.ca_cert_dir is None
+
+
+def test_adapter_rejects_disabled_tls_verification(system_trust_session):
+    session, _, _, _ = system_trust_session
+    adapter = session.get_adapter("https://private.prowler.example")
+
+    with pytest.raises(ValueError, match="verification is required"):
+        adapter.cert_verify(
+            MagicMock(),
+            "https://private.prowler.example",
+            verify=False,
+            cert=None,
+        )
+
+
+def test_adapter_does_not_retry_posts(system_trust_session):
+    session, _, _, _ = system_trust_session
+    adapter = session.get_adapter("https://private.prowler.example")
+
+    assert adapter.max_retries.total == 0
+
+
+def test_truststore_is_loaded_lazily_and_initialization_errors_are_safe(
+    monkeypatch,
+):
+    assert "truststore" not in ingestion.__dict__
+    monkeypatch.setitem(
+        sys.modules,
+        "truststore",
+        SimpleNamespace(SSLContext=MagicMock(side_effect=OSError("secret detail"))),
+    )
+
+    with pytest.raises(
+        ingestion.SystemTrustStoreError,
+        match="Could not initialize the operating system trust store",
+    ) as error:
+        ingestion._SystemTrustSession()
+
+    assert "secret detail" not in str(error.value)
+
+
+def test_truststore_import_failure_becomes_custom_error(monkeypatch):
+    monkeypatch.setitem(sys.modules, "truststore", None)
+
+    with pytest.raises(ingestion.SystemTrustStoreError):
+        ingestion._SystemTrustSession()
+
+
+def test_upload_preserves_request_and_response_behavior(tmp_path, monkeypatch):
+    ocsf_file = tmp_path / "findings.ocsf.json"
+    ocsf_file.write_text('{"finding": true}')
+    response = MagicMock(status_code=200, text='{"data": {"id": "job-id"}}')
+    response.json.return_value = {"data": {"id": "job-id"}}
+    session = MagicMock()
+    uploaded_file = {}
+
+    def capture_upload(*args, **kwargs):
+        filename, file_handle, content_type = kwargs["files"]["file"]
+        uploaded_file.update(
+            filename=filename,
+            content=file_handle.read(),
+            content_type=content_type,
+        )
+        return response
+
+    session.post.side_effect = capture_upload
+    monkeypatch.setattr(
+        ingestion, "_SystemTrustSession", MagicMock(return_value=session)
+    )
+
+    result = ingestion.send_ocsf_to_api(
+        str(ocsf_file),
+        base_url="private.prowler.example/",
+        api_key="safe-api-key",
+        timeout=17,
+    )
+
+    assert result == {"data": {"id": "job-id"}}
+    assert uploaded_file == {
+        "filename": "findings.ocsf.json",
+        "content": b'{"finding": true}',
+        "content_type": "application/json",
+    }
+    session.post.assert_called_once_with(
+        f"https://private.prowler.example{ingestion.cloud_api_ingestion_path}",
+        headers={
+            "Authorization": "Api-Key safe-api-key",
+            "Accept": "application/vnd.api+json",
+        },
+        files=session.post.call_args.kwargs["files"],
+        timeout=17,
+        allow_redirects=False,
+    )
+    response.raise_for_status.assert_called_once_with()
+    response.json.assert_called_once_with()
+    session.close.assert_called_once_with()
+
+
+@pytest.mark.parametrize(
+    "error",
+    [requests.exceptions.SSLError("TLS failed"), requests.ConnectionError("offline")],
+)
+def test_upload_closes_session_when_request_fails(tmp_path, monkeypatch, error):
+    ocsf_file = tmp_path / "findings.ocsf.json"
+    ocsf_file.write_text("[]")
+    session = MagicMock()
+    session.post.side_effect = error
+    monkeypatch.setattr(
+        ingestion, "_SystemTrustSession", MagicMock(return_value=session)
+    )
+
+    with pytest.raises(type(error)):
+        ingestion.send_ocsf_to_api(
+            str(ocsf_file),
+            base_url="https://private.prowler.example",
+            api_key="safe-api-key",
+        )
+
+    session.post.assert_called_once()
+    session.close.assert_called_once_with()
+
+
+def test_empty_response_returns_empty_object_and_closes_session(tmp_path, monkeypatch):
+    ocsf_file = tmp_path / "findings.ocsf.json"
+    ocsf_file.write_text("[]")
+    response = MagicMock(status_code=200, text="")
+    session = MagicMock()
+    session.post.return_value = response
+    monkeypatch.setattr(
+        ingestion, "_SystemTrustSession", MagicMock(return_value=session)
+    )
+
+    result = ingestion.send_ocsf_to_api(str(ocsf_file), api_key="safe-api-key")
+
+    assert result == {}
+    response.json.assert_not_called()
+    session.close.assert_called_once_with()
+
+
+def test_upload_rejects_redirect_response(tmp_path, monkeypatch):
+    ocsf_file = tmp_path / "findings.ocsf.json"
+    ocsf_file.write_text("[]")
+    response = requests.Response()
+    response.status_code = 307
+    session = MagicMock()
+    session.post.return_value = response
+    monkeypatch.setattr(
+        ingestion, "_SystemTrustSession", MagicMock(return_value=session)
+    )
+
+    with pytest.raises(requests.HTTPError, match="redirect") as error:
+        ingestion.send_ocsf_to_api(str(ocsf_file), api_key="safe-api-key")
+
+    assert error.value.response is response
+
+
+@pytest.mark.parametrize("failure_stage", ["status", "json"])
+def test_upload_closes_session_when_response_processing_fails(
+    tmp_path, monkeypatch, failure_stage
+):
+    ocsf_file = tmp_path / "findings.ocsf.json"
+    ocsf_file.write_text("[]")
+    response = MagicMock(status_code=200, text="invalid-json")
+    if failure_stage == "status":
+        response.raise_for_status.side_effect = requests.HTTPError("forbidden")
+    else:
+        response.json.side_effect = ValueError("invalid response")
+    session = MagicMock()
+    session.post.return_value = response
+    monkeypatch.setattr(
+        ingestion, "_SystemTrustSession", MagicMock(return_value=session)
+    )
+
+    with pytest.raises((requests.HTTPError, ValueError)):
+        ingestion.send_ocsf_to_api(str(ocsf_file), api_key="safe-api-key")
+
+    session.close.assert_called_once_with()
+
+
+def test_missing_api_key_fails_before_creating_session(tmp_path, monkeypatch):
+    ocsf_file = tmp_path / "findings.ocsf.json"
+    ocsf_file.write_text("[]")
+    session_factory = MagicMock()
+    monkeypatch.setattr(ingestion, "cloud_api_key", None)
+    monkeypatch.setattr(ingestion, "_SystemTrustSession", session_factory)
+
+    with pytest.raises(ValueError, match="Missing API key"):
+        ingestion.send_ocsf_to_api(str(ocsf_file))
+
+    session_factory.assert_not_called()
+
+
+def test_missing_file_fails_before_creating_session(tmp_path, monkeypatch):
+    session_factory = MagicMock()
+    monkeypatch.setattr(ingestion, "_SystemTrustSession", session_factory)
+
+    with pytest.raises(FileNotFoundError, match="OCSF file not found"):
+        ingestion.send_ocsf_to_api(
+            str(tmp_path / "missing.ocsf.json"), api_key="safe-api-key"
+        )
+
+    session_factory.assert_not_called()

uv.lock

@@ -3846,6 +3846,7 @@ dependencies = [
     { name = "stackit-objectstorage" },
     { name = "stackit-resourcemanager" },
     { name = "tabulate" },
+    { name = "truststore" },
     { name = "tzlocal" },
     { name = "uuid6" },
     { name = "zstandard" },
@@ -3967,6 +3968,7 @@ requires-dist = [
     { name = "stackit-objectstorage", specifier = "==1.4.0" },
     { name = "stackit-resourcemanager", specifier = "==0.8.0" },
     { name = "tabulate", specifier = "==0.9.0" },
+    { name = "truststore", specifier = "==0.10.4" },
     { name = "tzlocal", specifier = "==5.3.1" },
     { name = "uuid6", specifier = "==2024.7.10" },
     { name = "zstandard", specifier = "==0.25.0" },
@@ -5220,6 +5222,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/6a/43/8bd850ee71a191bf072e31302c73a66be413fecdd98fdcd111ecbcce13ca/tomlkit-0.15.0-py3-none-any.whl", hash = "sha256:4dbc8f0fc024412b57ced8757ac7461305126a648ff8c2c807fcb8e133a78738", size = 41328, upload-time = "2026-05-10T07:38:23.517Z" },
 ]
 
+[[package]]
+name = "truststore"
+version = "0.10.4"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.73Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" },
+]
+
 [[package]]
 name = "typing-extensions"
 version = "4.15.0"