Feature/accept mtls keys instead of paths

parameter-sets/credential/parameter-set.json

@@ -130,15 +130,15 @@
         {
             "name": "mtls_certificate_path",
             "label": "Path to certificate",
-            "description": "",
-            "type": "STRING",
+            "description": "or full certificate starting with -----BEGIN and ending with END CERTIFICATE-----",
+            "type": "PASSWORD",
             "visibilityCondition": "model.use_mtls==true"
         },
         {
             "name": "mtls_key_path",
             "label": "Path to key",
-            "description": "",
-            "type": "STRING",
+            "description": "or full key starting with -----BEGIN and ending with END PRIVATE KEY-----",
+            "type": "PASSWORD",
             "visibilityCondition": "model.use_mtls==true"
         }
     ]

parameter-sets/secure-basic/parameter-set.json

@@ -55,15 +55,15 @@
         {
             "name": "mtls_certificate_path",
             "label": "Path to certificate",
-            "description": "",
-            "type": "STRING",
+            "description": "or full certificate starting with -----BEGIN and ending with END CERTIFICATE-----",
+            "type": "PASSWORD",
             "visibilityCondition": "model.use_mtls==true"
         },
         {
             "name": "mtls_key_path",
             "label": "Path to key",
-            "description": "",
-            "type": "STRING",
+            "description": "or full key starting with -----BEGIN and ending with END PRIVATE KEY-----",
+            "type": "PASSWORD",
             "visibilityCondition": "model.use_mtls==true"
         }
     ]

parameter-sets/secure-oauth/parameter-set.json

@@ -64,15 +64,15 @@
         {
             "name": "mtls_certificate_path",
             "label": "Path to certificate",
-            "description": "",
-            "type": "STRING",
+            "description": "or full certificate starting with -----BEGIN and ending with END CERTIFICATE-----",
+            "type": "PASSWORD",
             "visibilityCondition": "model.use_mtls==true"
         },
         {
             "name": "mtls_key_path",
             "label": "Path to key",
-            "description": "",
-            "type": "STRING",
+            "description": "or full key starting with -----BEGIN and ending with END PRIVATE KEY-----",
+            "type": "PASSWORD",
             "visibilityCondition": "model.use_mtls==true"
         }
     ]

python-lib/dku_constants.py

@@ -1,7 +1,7 @@
 class DKUConstants(object):
     API_RESPONSE_KEY = "api_response"
-    FORBIDDEN_KEYS = ["token", "password", "api_key_value", "secure_token"]
+    FORBIDDEN_KEYS = ["token", "password", "api_key_value", "secure_token", "mtls_key_path", "mtls_certificate_path"]
     FORM_DATA_BODY_FORMAT = "FORM_DATA"
-    PLUGIN_VERSION = "1.2.7"
+    PLUGIN_VERSION = "1.2.7-beta.5"
     RAW_BODY_FORMAT = "RAW"
     REPONSE_ERROR_KEY = "dku_error"

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 "):
+                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,20 @@ def get_headers(response):
     if isinstance(response, requests.Response):
         return response.headers
     return None
+
+
+def normalize_key(key):
+    PROTECTED_EXPRESSIONS = [
+        "BEGIN CERTIFICATE", "END CERTIFICATE",
+        "BEGIN PRIVATE KEY", "END PRIVATE KEY",
+        "BEGIN RSA PRIVATE KEY", "END RSA PRIVATE KEY"
+    ]
+    tempo_text = str(key)
+    for expression_to_protect in PROTECTED_EXPRESSIONS:
+        protected_form = expression_to_protect.replace(" ", "")
+        tempo_text = tempo_text.replace(expression_to_protect, protected_form)
+    tempo_text = tempo_text.replace(" ", "\n")
+    for expression_to_protect in PROTECTED_EXPRESSIONS:
+        protected_form = expression_to_protect.replace(" ", "")
+        tempo_text = tempo_text.replace(protected_form, expression_to_protect)
+    return tempo_text