[defectdojo] Adding SSL verification management (RedHatProductSecurit…

…y#144)

* [defectdojo] Adding SSL verification management

Either:
- REQUESTS_CA_BUNDLE points to the CA bundle
- `config.defectDojo.ssl` is:
  + False: verification is optional
  + True: verification is mandatory (default behavior)
  + path to CA bundle: verification is mandatory, and this bundle is
    used

* [DD] pytests: added new test, fixed broken test

README.md

@@ -147,12 +147,19 @@ config:
   # Defect dojo configuration
   defectDojo:
     url: "https://mydefectdojo.example.com/"
+    ssl: [True | False | "/path/to/CA"]
     authorization:
       username: "rapidast_productname"
       password: "password"
       # alternatively, a `token` entry can be set in place of username/password
 ```
 
+the `ssl` parameter is provided as Requests' `verify` parameter. It can be either:
+- True: SSL verification is mandatory, against the default CA bundle
+- False: SSL verification is not mandatory (but prints a warning if it fails)
+- /path/to/CA: a bundle of CAs to verify from
+Alternatively, the `REQUESTS_CA_BUNDLE` environment variable can be used to select a CA bundle file. If nothing is provided, the default value will be `True`
+
 You can either authenticate using a username/password combination, or a token (make sure it is not expired). In either case, you can use the `_from_var` method described in previous chapter to avoid hardcoding the value in the configuration.
 
 Then, RapiDAST needs to know, for each scanner, sufficient information such that it can identify which product/engagement/test to match.

exports/defect_dojo.py

@@ -9,7 +9,7 @@ class DefectDojo:
 
     DD_CONNECT_TIMEOUT = 10  # in seconds
 
-    def __init__(self, base_url, username=None, password=None, token=None):
+    def __init__(self, base_url, login=None, token=None, ssl=None):
         if not base_url:
             raise ValueError(
                 "Defect Dojo invalid configuration: URL is a mandatory value"
@@ -19,13 +19,30 @@ def __init__(self, base_url, username=None, password=None, token=None):
             raise ValueError("Defect Dojo invalid configuration: URL is not correct")
 
         self.base_url = base_url
-        self.username = username
-        self.password = password
+
+        self.username = None
+        self.password = None
+        if login:
+            try:
+                self.username = login["username"]
+                self.password = login["password"]
+            except KeyError:
+                logging.error(
+                    "RapiDAST BUG: DefectDojo was created with invalid login information..."
+                )
+                logging.error("RapiDAST BUG: ...[continuing without credentials]")
+
         self.token = token
         self.headers = {}
         if token:
             self.headers["Authorization"] = f"Token {token}"
 
+        # params is injected as a **kwargs to each request
+        # this is to prevent `verify` to override REQUESTS_CA_BUNDLE
+        self.params = {"timeout": self.DD_CONNECT_TIMEOUT}
+        if ssl is not None:
+            self.params["verify"] = ssl
+
     def _auth_and_set_token(self):
         """Force a refresh of the token using the username/password"""
         logging.debug("Defect Dojo: refreshing token")
@@ -37,7 +54,7 @@ def _auth_and_set_token(self):
         data = {"username": self.username, "password": self.password}
 
         try:
-            resp = requests.post(url, timeout=self.DD_CONNECT_TIMEOUT, data=data)
+            resp = requests.post(url, data=data, **self.params)
             resp.raise_for_status()
 
             logging.debug(f"resp: {resp.json()}")
@@ -66,14 +83,14 @@ def engagement_exists(self, engagement_id=None, name=None):
         if engagement_id:
             resp = requests.get(
                 f"{self.base_url}/api/v2/engagements/?engagment={engagement_id}",
-                timeout=self.DD_CONNECT_TIMEOUT,
                 headers=self.headers,
+                **self.params,
             )
         elif name:
             resp = requests.get(
                 f"{self.base_url}/api/v2/engagements/?name={parse.quote_plus(name)}",
-                timeout=self.DD_CONNECT_TIMEOUT,
                 headers=self.headers,
+                **self.params,
             )
         else:
             raise ValueError("Either an engagement name or ID must be provided")
@@ -106,10 +123,10 @@ def _private_import(self, endpoint, data, filename):
 
         resp = requests.post(
             endpoint,
-            timeout=self.DD_CONNECT_TIMEOUT,
             headers=self.headers,
             data=data,
             files={"file": open(filename, "rb")},  # pylint: disable=consider-using-with
+            **self.params,
         )
         if resp.status_code >= 400:
             logging.debug(vars(resp))

rapidast.py

@@ -196,9 +196,16 @@ def run():
     if config.get("config.defectDojo.url"):
         defect_d = DefectDojo(
             config.get("config.defectDojo.url"),
-            config.get("config.defectDojo.authorization.username"),
-            config.get("config.defectDojo.authorization.password"),
+            {
+                "username": config.get(
+                    "config.defectDojo.authorization.username", default=""
+                ),
+                "password": config.get(
+                    "config.defectDojo.authorization.password", default=""
+                ),
+            },
             config.get("config.defectDojo.authorization.token"),
+            config.get("config.defectDojo.ssl", default=True),
         )
 
     # Run all scanners

tests/test_defectdojo_integration.py

@@ -21,7 +21,23 @@ def test_dd_auth_and_set_token_no_username():
 def test_dd_auth_and_set_token_non_existent_url():
     # assuming 127.0.0.1:12345 is non-existent
     defect_d = DefectDojo(
-        "https://127.0.0.1:12345", "random_username", "random_password", "random_token"
+        "https://127.0.0.1:12345",
+        {"username": "random_username", "password": "random_password"},
+        "random_token",
     )
     with pytest.raises(requests.exceptions.ConnectionError):
         defect_d._auth_and_set_token()
+
+
+def test_dd_parameters():
+    defect_d = DefectDojo("https://127.0.0.1:12345", token="random_token")
+
+    assert defect_d.params["timeout"] == DefectDojo.DD_CONNECT_TIMEOUT
+    with pytest.raises(KeyError):
+        defect_d.params["verify"]
+
+    defect_d = DefectDojo(
+        "https://127.0.0.1:12345", token="random_token", ssl="CAbundle"
+    )
+    assert defect_d.params["timeout"] == DefectDojo.DD_CONNECT_TIMEOUT
+    assert defect_d.params["verify"] == "CAbundle"