Lint with pre-commit hook

RedHatProductSecurity · Sep 26, 2024 · 0b7a686 · 0b7a686
1 parent 1f71682
commit 0b7a686
Show file tree

Hide file tree

Showing 31 changed files with 188 additions and 523 deletions.
diff --git a/config/config-template-zap-long.yaml b/config/config-template-zap-long.yaml
@@ -66,7 +66,7 @@ general:
     #
     # "browser" authentication will use firefox in the background to generate cookies
     #  - verifyUrl must return an error if the user is not logged in
-    #type: "browser"   
+    #type: "browser"
     #parameters:
     #  username: "user"
     #  password: "mypassw0rd"

diff --git a/configmodel/__init__.py b/configmodel/__init__.py
@@ -77,9 +77,7 @@ def delete(self, path):
         except AttributeError:
             pass
         # Failed to iterate until the end: the path does not exist
-        logging.warning(
-            f"RapidastConfigModel.delete(): Config path {path} was not found. No deletion"
-        )
+        logging.warning(f"RapidastConfigModel.delete(): Config path {path} was not found. No deletion")
         return False
 
     def exists(self, path):
@@ -122,9 +120,7 @@ def set(self, path, value, overwrite=True):
                 tmp = walk[key]
             # case 3: not a "dictionary" type: warn and overwrite (if True)
             if not isinstance(tmp, dict):
-                logging.warning(
-                    f"RapidastConfigModel.set: Incompatible {path} at {tmp}"
-                )
+                logging.warning(f"RapidastConfigModel.set: Incompatible {path} at {tmp}")
                 if not overwrite:
                     logging.info("RapidastConfigModel.set: no overwrite: early return")
                     return False
@@ -162,9 +158,7 @@ def merge(self, merge, preserve=False, root=None):
         if not merge:
             return
         if not isinstance(merge, dict):
-            raise TypeError(
-                f"RapidastConfigModel.merge: merge must be a dict (was: {type(merge)})"
-            )
+            raise TypeError(f"RapidastConfigModel.merge: merge must be a dict (was: {type(merge)})")
 
         root = path_to_list(root)
 
@@ -177,7 +171,7 @@ def merge(self, merge, preserve=False, root=None):
         deep_dict_merge(sub_conf, merge, preserve)
 
     def get_official_app_name(self):
-        """ Shortcut: 
+        """Shortcut:
         Return a string corresponding to how the application should be called
         Based on the configuratoin.
         Prefer the full product name, but defer to short name if unavailable

diff --git a/configmodel/converter.py b/configmodel/converter.py
@@ -45,9 +45,7 @@ def dispatch(version):
 def convert_configmodel(conf):
     """This is the base function, attached to error reporting"""
     version = conf.get("config.configVersion", 0)
-    raise RuntimeError(
-        f"There was an error in converting configuration. No convertion available for version {version}"
-    )
+    raise RuntimeError(f"There was an error in converting configuration. No convertion available for version {version}")
 
 
 @convert_configmodel.register(4)
@@ -60,9 +58,7 @@ def convert_from_version_4_to_5(old):
     new = copy.deepcopy(old)
 
     for key in old.conf["scanners"]:
-        if key.startswith("zap") and old.exists(
-            f"scanners.{key}.miscOptions.oauth2OpenapiManualDownload"
-        ):
+        if key.startswith("zap") and old.exists(f"scanners.{key}.miscOptions.oauth2OpenapiManualDownload"):
             new.move(
                 f"scanners.{key}.miscOptions.oauth2OpenapiManualDownload",
                 f"scanners.{key}.miscOptions.oauth2ManualDownload",
@@ -174,29 +170,22 @@ def convert_from_version_0_to_1(old):
     auth_method = old.get("scan.auth_method", default=None)
     if (
         auth_method == "scriptBasedAuthentication"
-        and old.get("scan.scriptAuth.authScriptFilePath", default="")
-        == "scripts/offline-token.js"
+        and old.get("scan.scriptAuth.authScriptFilePath", default="") == "scripts/offline-token.js"
     ):
         # probably OAuth2
         new.set(
             "general.authentication",
             {
                 "type": "oauth2_rtoken",
                 "parameters": {
-                    "client_id": old.get(
-                        "scan.scriptAuth.authClientID", default="cloud-services"
-                    ),
-                    "token_endpoint": old.get(
-                        "scan.scriptAuth.authTokenEndpoint", default=""
-                    ),
+                    "client_id": old.get("scan.scriptAuth.authClientID", default="cloud-services"),
+                    "token_endpoint": old.get("scan.scriptAuth.authTokenEndpoint", default=""),
                     "rtoken_var_name": "RTOKEN",
                 },
             },
         )
     else:
-        logging.warning(
-            "The config version translator does not support this particular authentication"
-        )
+        logging.warning("The config version translator does not support this particular authentication")
 
     # "Scanners.Zap" section
     new.set(
@@ -206,13 +195,9 @@ def convert_from_version_0_to_1(old):
 
     ### OpenAPI
     if old.get("openapi.importFromUrl", default=False):
-        new.set(
-            "scanners.zap.apiScan.apis.apiUrl", old.get("openapi.url", default=None)
-        )
+        new.set("scanners.zap.apiScan.apis.apiUrl", old.get("openapi.url", default=None))
     elif old.get("openapi.directory", default=""):
-        logging.warning(
-            "The config version translator does not support Directory based OpenAPI"
-        )
+        logging.warning("The config version translator does not support Directory based OpenAPI")
 
     ## Passive scan
     new.set("scanners.zap.passiveScan", {})
@@ -225,9 +210,7 @@ def convert_from_version_0_to_1(old):
     ## Active scan
     # Active scanner was always enabled, so we do the same:
     new.set("scanners.zap.activeScan", {})
-    new.set(
-        "scanners.zap.activeScan.policy", old.get("scan.policies.scanPolicyName", None)
-    )
+    new.set("scanners.zap.activeScan.policy", old.get("scan.policies.scanPolicyName", None))
 
     # Finally, set the correct version number
     new.set("config.configVersion", 1)

diff --git a/exports/defect_dojo.py b/exports/defect_dojo.py
@@ -11,9 +11,7 @@ class DefectDojo:
 
     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"
-            )
+            raise ValueError("Defect Dojo invalid configuration: URL is a mandatory value")
         parsed = parse.urlparse(base_url)  # expects to raise exception on invalid URL
         if parsed.scheme not in ["http", "https"]:
             raise ValueError("Defect Dojo invalid configuration: URL is not correct")
@@ -27,9 +25,7 @@ def __init__(self, base_url, login=None, token=None, ssl=None):
                 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: DefectDojo was created with invalid login information...")
                 logging.error("RapiDAST BUG: ...[continuing without credentials]")
 
         self.token = token
@@ -47,9 +43,7 @@ def _auth_and_set_token(self):
         """Force a refresh of the token using the username/password"""
         logging.debug("Defect Dojo: refreshing token")
         if not self.username or not self.password:
-            raise ValueError(
-                "Defect Dojo invalid configuration: A username and a password are required to get a token"
-            )
+            raise ValueError("Defect Dojo invalid configuration: A username and a password are required to get a token")
         url = self.base_url + "/api/v2/api-token-auth/"
         data = {"username": self.username, "password": self.password}
 
@@ -63,9 +57,7 @@ def _auth_and_set_token(self):
             self.headers["Authorization"] = f"Token {self.token}"
             logging.debug("Defect Dojo: successfully refreshed token")
         except requests.exceptions.ConnectTimeout as e:
-            logging.error(
-                f"Getting token failed. Check the URL for defectDojo in config file. err details: {e}"
-            )
+            logging.error(f"Getting token failed. Check the URL for defectDojo in config file. err details: {e}")
             return 1
         except requests.exceptions.HTTPError as e:
             logging.error(
@@ -96,9 +88,7 @@ def engagement_exists(self, engagement_id=None, name=None):
             raise ValueError("Either an engagement name or ID must be provided")
 
         if resp.status_code >= 400:
-            logging.warning(
-                f"Error while looking for engagement ({resp.status_code}, {resp.get('message')})"
-            )
+            logging.warning(f"Error while looking for engagement ({resp.status_code}, {resp.get('message')})")
         counts = resp.json()["counts"]
         if counts > 1:
             logging.warning("Error while looking for engagement: too many hits")
@@ -134,9 +124,7 @@ def _private_import(self, endpoint, data, filename):
             logging.error(f"Error while exporting ({resp.status_code}, {err})")
 
             if "Invalid token" in err["detail"]:
-                logging.error(
-                    "Please check your token in 'config.defectDojo' of the config file"
-                )
+                logging.error("Please check your token in 'config.defectDojo' of the config file")
 
             return 1
 
@@ -146,31 +134,19 @@ def reimport_scan(self, data, filename):
         """Reimport to an existing engagement with an existing compatible scan."""
 
         if not data.get("test") and not (
-            data.get("engagement_name")
-            and data.get("product_name")
-            and data.get("test_title")
+            data.get("engagement_name") and data.get("product_name") and data.get("test_title")
         ):
-            raise ValueError(
-                "Reimport needs to identify an existing test (by ID or names of product+engagement+test)"
-            )
+            raise ValueError("Reimport needs to identify an existing test (by ID or names of product+engagement+test)")
 
-        return self._private_import(
-            f"{self.base_url}/api/v2/reimport-scan/", data, filename
-        )
+        return self._private_import(f"{self.base_url}/api/v2/reimport-scan/", data, filename)
 
     def import_scan(self, data, filename):
         """export to an existing engagement, via the `import-scan` endpoint."""
 
-        if not data.get("engagement") and not (
-            data.get("engagement_name") and data.get("product_name")
-        ):
-            raise ValueError(
-                "Import needs to identify an existing engagement (by ID or names of product+engagement)"
-            )
+        if not data.get("engagement") and not (data.get("engagement_name") and data.get("product_name")):
+            raise ValueError("Import needs to identify an existing engagement (by ID or names of product+engagement)")
 
-        return self._private_import(
-            f"{self.base_url}/api/v2/import-scan/", data, filename
-        )
+        return self._private_import(f"{self.base_url}/api/v2/import-scan/", data, filename)
 
     def export_scan(self, data, filename):
         """Decide wether to import or reimport. Based on:

diff --git a/exports/google_cloud_storage.py b/exports/google_cloud_storage.py
@@ -56,9 +56,7 @@ def export_scan(self, data, filename):
 
         metadata = self.create_metadata(data)
 
-        logging.info(
-            f"GoogleCloudStorage: sending {filename}. UUID: {metadata['uuid']}"
-        )
+        logging.info(f"GoogleCloudStorage: sending {filename}. UUID: {metadata['uuid']}")
 
         # export data as a metadata.json file
         json_stream = StringIO()
@@ -82,11 +80,7 @@ def export_scan(self, data, filename):
         unique_id = "{}-RapiDAST-{}-{}.tgz".format(  # pylint: disable=C0209
             datetime.datetime.now(tz=datetime.timezone.utc).isoformat(),
             self.app_name,
-            "".join(
-                random.choices(
-                    string.ascii_letters + string.ascii_uppercase + string.digits, k=6
-                )
-            ),
+            "".join(random.choices(string.ascii_letters + string.ascii_uppercase + string.digits, k=6)),
         )
         blob_name = self.directory + "/" + unique_id
 

diff --git a/rapidast.py b/rapidast.py
@@ -108,9 +108,7 @@ def run_scanner(name, config, args, scan_exporter):
 
     # Part 5: cleanup
     if not scanner.state == scanners.State.PROCESSED:
-        logging.error(
-            f"Something is wrong. Scanner {name} is not in PROCESSED state: the workdir won't be cleaned up"
-        )
+        logging.error(f"Something is wrong. Scanner {name} is not in PROCESSED state: the workdir won't be cleaned up")
         return 1
 
     if not args.no_cleanup:
@@ -155,19 +153,13 @@ def run():
     args.loglevel = args.loglevel.upper()
     add_logging_level("VERBOSE", logging.DEBUG + 5)
     logging.basicConfig(format="%(levelname)s:%(message)s", level=args.loglevel)
-    logging.debug(
-        f"log level set to debug. Config file: '{parser.parse_args().config_file}'"
-    )
+    logging.debug(f"log level set to debug. Config file: '{parser.parse_args().config_file}'")
 
     # Load config file
     try:
-        config = configmodel.RapidastConfigModel(
-            yaml.safe_load(load_config_file(parser.parse_args().config_file))
-        )
+        config = configmodel.RapidastConfigModel(yaml.safe_load(load_config_file(parser.parse_args().config_file)))
     except yaml.YAMLError as exc:
-        raise RuntimeError(
-            f"YAML error in config {parser.parse_args().config_file}':\n {str(exc)}"
-        ) from exc
+        raise RuntimeError(f"YAML error in config {parser.parse_args().config_file}':\n {str(exc)}") from exc
 
     # Optionally adds default if file exists (will not overwrite existing entries)
     default_conf = os.path.join(os.path.dirname(__file__), "rapidast-defaults.yaml")
@@ -176,18 +168,14 @@ def run():
         try:
             config.merge(yaml.safe_load(load_config_file(default_conf)), preserve=True)
         except yaml.YAMLError as exc:
-            raise RuntimeError(
-                f"YAML error in config {default_conf}':\n {str(exc)}"
-            ) from exc
+            raise RuntimeError(f"YAML error in config {default_conf}':\n {str(exc)}") from exc
 
     # Update to latest config schema if need be
     config = configmodel.converter.update_to_latest_config(config)
 
     config.set("config.results_dir", get_full_result_dir_path(config))
 
-    logging.debug(
-        f"The entire loaded configuration is as follow:\n=====\n{pp.pformat(config)}\n====="
-    )
+    logging.debug(f"The entire loaded configuration is as follow:\n=====\n{pp.pformat(config)}\n=====")
 
     # Do early: load the environment file if one is there
     load_environment(config)
@@ -196,9 +184,7 @@ def run():
     scan_exporter = None
     if config.get("config.googleCloudStorage.bucketName"):
         scan_exporter = GoogleCloudStorage(
-            bucket_name=config.get(
-                "config.googleCloudStorage.bucketName", "default-bucket-name"
-            ),
+            bucket_name=config.get("config.googleCloudStorage.bucketName", "default-bucket-name"),
             app_name=config.get_official_app_name(),
             directory=config.get("config.googleCloudStorage.directory", None),
             keyfile=config.get("config.googleCloudStorage.keyFile", None),
@@ -207,12 +193,8 @@ def run():
         scan_exporter = DefectDojo(
             config.get("config.defectDojo.url"),
             {
-                "username": config.get(
-                    "config.defectDojo.authorization.username", default=""
-                ),
-                "password": config.get(
-                    "config.defectDojo.authorization.password", default=""
-                ),
+                "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),

diff --git a/scanners/__init__.py b/scanners/__init__.py
@@ -25,9 +25,7 @@ def __init__(self, config, ident):
         self.config = config
         self.state = State.UNCONFIGURED
 
-        self.results_dir = os.path.join(
-            self.config.get("config.results_dir", default="results"), self.ident
-        )
+        self.results_dir = os.path.join(self.config.get("config.results_dir", default="results"), self.ident)
 
         # When requested to create a temporary file or directory, it will be a subdir of
         # this temporary directory
@@ -70,8 +68,7 @@ def _should_export_to_defect_dojo(self):
         - this particular scanner's export is not explicitely disabled (`defectDojoExport` is not False)
         """
         return self.my_conf("defectDojoExport") is not False and (
-            self.config.get("config.googleCloudStorage")
-            or self.config.get("config.defectDojo")
+            self.config.get("config.googleCloudStorage") or self.config.get("config.defectDojo")
         )
 
     def _fill_up_data_for_defect_dojo(self, data):
@@ -117,9 +114,7 @@ def _fill_up_data_for_defect_dojo(self, data):
             # A default product name was chosen as part of `self.get_default_defectdojo_data()`
             # Generate an engagement name if none are set
             if not data.get("engagement_name"):
-                data[
-                    "engagement_name"
-                ] = f"RapiDAST-{data['product_name']}-{datetime.date.today()}"
+                data["engagement_name"] = f"RapiDAST-{data['product_name']}-{datetime.date.today()}"
 
         return data