feat(terraform): add option to add proxy to request (#6916)

* add proxy to request

* add proxy to request

* add envs, in the proxy_url we need to add the GcpIdentityProvider

* .

* .

* .

checkov/common/util/env_vars_config.py

@@ -79,6 +79,8 @@ def __init__(self) -> None:
         self.HCL_PARSE_TIMEOUT_SEC = force_int(os.getenv("HCL_PARSE_TIMEOUT_SEC", 10))
         self.ENABLE_DOTNET_CPM = os.getenv('ENABLE_DOTNET_CPM', False)
         self.JAVA_FULL_DT = os.getenv('JAVA_FULL_DT', False)
+        self.PROXY_CA_PATH = os.getenv('PROXY_CA_PATH', None)
+        self.PROXY_URL = os.getenv('PROXY_URL', None)
 
 
 env_vars_config = EnvVarsConfig()

checkov/terraform/module_loading/loaders/registry_loader.py

@@ -19,6 +19,7 @@
     order_versions_in_descending_order,
     get_version_constraints
 )
+from checkov.terraform.module_loading.proxy_client import call_http_request_with_proxy
 
 if TYPE_CHECKING:
     from checkov.terraform.module_loading.module_params import ModuleParams
@@ -83,11 +84,19 @@ def _load_module(self, module_params: ModuleParams) -> ModuleContent:
         logging.debug(f"Best version for {module_params.module_source} is {best_version} based on the version constraint {module_params.version}.")
         logging.debug(f"Module download url: {request_download_url}")
         try:
-            response = requests.get(
+            request = requests.Request(
+                method='GET',
                 url=request_download_url,
-                headers={"Authorization": f"Bearer {module_params.token}"} if module_params.token else None,
-                timeout=DEFAULT_TIMEOUT
+                headers={"Authorization": f"Bearer {module_params.token}"} if module_params.token else None
             )
+            if os.getenv('PROXY_URL'):
+                logging.info('Send request with proxy')
+                response = call_http_request_with_proxy(request)
+            else:
+                session = requests.Session()
+                prepared_request = session.prepare_request(request)
+                response = session.send(prepared_request, timeout=DEFAULT_TIMEOUT)
+
             response.raise_for_status()
         except HTTPError as e:
             self.logger.warning(e)

checkov/terraform/module_loading/proxy_client.py

@@ -0,0 +1,33 @@
+import os
+from typing import Any
+
+import requests
+
+
+class ProxyClient:
+    def __init__(self) -> None:
+        self.proxy_ca_path = os.getenv('PROXY_CA_PATH', None)
+        if self.proxy_ca_path is None:
+            raise Exception("[ProxyClient] CA certificate path is missing")
+
+    def get_session(self) -> requests.Session:
+        if not os.getenv('PROXY_URL', None):
+            raise Exception('Please provide "PROXY_URL" env var')
+        proxy_url = os.getenv('PROXY_URL')
+        session = requests.Session()
+        proxies = {
+            "http": proxy_url,
+            "https": proxy_url,
+        }
+        session.proxies.update(proxies)  # type: ignore
+        return session
+
+    def send_request(self, request: requests.Request) -> requests.Response:
+        session = self.get_session()
+        prepared_request = session.prepare_request(request)
+        return session.send(prepared_request, verify=self.proxy_ca_path)
+
+
+def call_http_request_with_proxy(request: requests.Request) -> Any:
+    proxy_client = ProxyClient()
+    return proxy_client.send_request(request=request)