using temp file for using key+certificats stored in presets

Author: alexbourret

python-lib/rest_api_client.py

@@ -1,6 +1,7 @@
 import requests
 import time
 import copy
+import tempfile
 from pagination import Pagination
 from safe_logger import SafeLogger
 from loop_detector import LoopDetector
@@ -184,14 +185,35 @@ def request(self, method, url, can_raise_exeption=True, **kwargs):
     def request_with_redirect_retry(self, method, url, **kwargs):
         # In case of redirection to another domain, the authorization header is not kept
         # If redirect_auth_header is true, another attempt is made with initial headers to the redirected url
-        response = self.session.request(method, url, **kwargs)
+        response = self.request_with_cert(method, url, **kwargs)
         if self.redirect_auth_header and not response.url.startswith(url):
             redirection_kwargs = copy.deepcopy(kwargs)
             redirection_kwargs.pop("params", None)  # params are contained in the redirected url
             logger.warning("Redirection ! Accessing endpoint {} with initial authorization headers".format(response.url))
-            response = self.session.request(method, response.url, **redirection_kwargs)
+            response = self.request_with_cert(method, response.url, **redirection_kwargs)
         return response
 
+    def request_with_cert(self, method, url, **kwargs):
+        cert = kwargs.get("cert", None)
+        if cert and len(cert) == 2:
+            if cert[0].startswith("-----BEGIN CERTIFICATE") and cert[1].startswith("-----BEGIN PRIVATE KEY"):
+                logger.info("mTLS certificate and key are strings")
+                response = None
+                with tempfile.NamedTemporaryFile(mode="w", suffix=".crt") as tmp_certificate:
+                    with tempfile.NamedTemporaryFile(mode="w", suffix=".key") as tmp_key:
+                        tmp_certificate.write(
+                            normalize_key(cert[0])
+                        )
+                        tmp_certificate.seek(0)
+                        tmp_key.write(
+                            normalize_key(cert[1])
+                        )
+                        tmp_key.seek(0)
+                        kwargs["cert"] = (tmp_certificate.name, tmp_key.name)
+                        response = self.session.request(method, url, **kwargs)
+                return response
+        return self.session.request(method, url, **kwargs)
+
     def paginated_api_call(self, can_raise_exeption=True):
         if self.pagination.params_must_be_blanked:
             self.requests_kwargs["params"] = {}
@@ -278,3 +300,17 @@ def get_headers(response):
     if isinstance(response, requests.Response):
         return response.headers
     return None
+
+
+def normalize_key(key):
+    tempo_text = str(key)
+    tempo_text = tempo_text.replace("BEGIN CERTIFICATE", "BEGINCERTIFICATE")
+    tempo_text = tempo_text.replace("END CERTIFICATE", "ENDCERTIFICATE")
+    tempo_text = tempo_text.replace("-----BEGIN PRIVATE KEY-----", "-----BEGINPRIVATEKEY-----")
+    tempo_text = tempo_text.replace("-----END PRIVATE KEY-----", "-----ENDPRIVATEKEY-----")
+    tempo_text = tempo_text.replace(" ", "\n")
+    tempo_text = tempo_text.replace("BEGINCERTIFICATE", "BEGIN CERTIFICATE")
+    tempo_text = tempo_text.replace("ENDCERTIFICATE", "END CERTIFICATE")
+    tempo_text = tempo_text.replace("-----BEGINPRIVATEKEY-----", "-----BEGIN PRIVATE KEY-----")
+    tempo_text = tempo_text.replace("-----ENDPRIVATEKEY-----", "-----END PRIVATE KEY-----")
+    return tempo_text