Lint code

setup.py

@@ -8,16 +8,16 @@
 
 setup(
     name='fafnir',  # Replace with your project name
-    version = version,
-    author = 'syn-4ck',
-    author_email = 'repoJFM@protonmail.com',
-    url = 'https://github.com/syn-4ck/fafnir',
-    description = 'Software supply chain security tool to automate appsec vulnerability detection',
-    long_description = 'Fafnir is an open-source tool that allows for the complete automation ' +
-        'of launching different security tools detecting vulnerabilities in the application''s, code',
-    license = "MIT license",
-    packages = find_packages(exclude=["tests",".github",".github"]),
-    install_requires = requirements,
+    version=version,
+    author='syn-4ck',
+    author_email='repoJFM@protonmail.com',
+    url='https://github.com/syn-4ck/fafnir',
+    description='Software supply chain security tool to automate appsec vulnerability detection',
+    long_description='Fafnir is an open-source tool that allows for the complete automation ' +
+    'of launching different security tools detecting vulnerabilities in the application''s, code',
+    license="MIT license",
+    packages=find_packages(exclude=["tests", ".github", ".github"]),
+    install_requires=requirements,
     entry_points={
         'console_scripts': [
             'fafnir = src.main:main',
@@ -28,4 +28,4 @@
         "License :: OSI Approved :: Apache License 2.0",
         "Operating System :: OS Independent",
     ]
-)
+)

src/core/run_containers.py

@@ -9,13 +9,15 @@ def _setup_environment_vars(configuration, tool):
         if configuration.get('tools-config').get('semgrep') is not None and configuration.get('tools-config').get('semgrep').get('api-key') is not None:
             return ["SEMGREP_APP_TOKEN={}".format(configuration.get('tools-config').get('semgrep').get('api-key'))]
         else:
-            logging.warning('Semgrep api-key is not setted properly. Please, review the documentation of the fafnir configuration.')
+            logging.warning(
+                'Semgrep api-key is not setted properly. Please, review the documentation of the fafnir configuration.')
             return []
     elif tool == "checkov":
         if configuration.get('tools-config').get('checkov') is not None and configuration.get('tools-config').get('checkov').get('api-key') is not None:
             return ["BC_API_KEY={}".format(configuration.get('tools-config').get('checkov').get('api-key'))]
         else:
-            logging.warning('Checkov api-key is not setted properly. The tool will evaluate the vulnerabilities, but the data will not be completed.')
+            logging.warning(
+                'Checkov api-key is not setted properly. The tool will evaluate the vulnerabilities, but the data will not be completed.')
             return []
     else:
         return []
@@ -77,7 +79,8 @@ def run_tools(client, config, scan_fullpath, verbose, configuration, async_optio
 
     # Run Continer tools
     if configuration is None or configuration.get("containers") is None or configuration.get("containers").get("image") is None or configuration.get("containers").get("image") == "":
-        logging.warning("Container security not enabled. Please, set up the image name in Fafnir configuration to evaluate it")
+        logging.warning(
+            "Container security not enabled. Please, set up the image name in Fafnir configuration to evaluate it")
     else:
         logging.info("Running the Container security analysis")
         for tool in config.get("containers").get("container-security"):

src/main.py

@@ -21,16 +21,18 @@
 @click.option("-a", "--asynchronous", is_flag=True, show_default=True, default=False, help="Asynchronous mode")
 @click.option("-t", "--output-type", type=click.Choice(['json', 'sarif']), default="json", help="Report type")
 @click.option("-o", "--output-path", default=os.path.join(os.path.abspath("."), "reports"), help="Path to store the tools/Fafnir report")
-@click.option("-x","--disable-apis", is_flag=True, show_default=True, default=True, help="Disable API requests")
+@click.option("-x", "--disable-apis", is_flag=True, show_default=True, default=True, help="Disable API requests")
 def main(scan_fullpath: str, verbose: bool, configuration: Optional[str],
          asynchronous: bool, output_type: str, output_path: str, disable_apis: bool) -> None:
 
     print_banner(VERSION)
 
     if verbose:
-        logging.basicConfig(level=logging.DEBUG, format='%(asctime)s: %(levelname)s - %(message)s')
+        logging.basicConfig(level=logging.DEBUG,
+                            format='%(asctime)s: %(levelname)s - %(message)s')
     else:
-        logging.basicConfig(level=logging.INFO, format='%(asctime)s: %(levelname)s - %(message)s')
+        logging.basicConfig(level=logging.INFO,
+                            format='%(asctime)s: %(levelname)s - %(message)s')
 
     client = docker.from_env()
 

src/report/generate_report/generate_sarif.py

@@ -27,17 +27,17 @@ def generate_report_sarif(scan_fullpath: str, report: dict) -> dict:
                         },
                         "locations": [
                             {
-                            "physicalLocation": {
-                                "artifactLocation": {
-                                "uri": vuln.get("file"),
-                                "index": 0
-                                },
-                                "region": {
-                                "startLine": vuln.get("location"),
-                                "startColumn": 1
+                                "physicalLocation": {
+                                    "artifactLocation": {
+                                        "uri": vuln.get("file"),
+                                        "index": 0
+                                    },
+                                    "region": {
+                                        "startLine": vuln.get("location"),
+                                        "startColumn": 1
+                                    }
                                 }
                             }
-                            }
                         ],
                         "ruleId": vuln.get("rule"),
                         "ruleIndex": 0
@@ -63,7 +63,6 @@ def generate_report_sarif(scan_fullpath: str, report: dict) -> dict:
                 "results": value
             })
 
-
     sarif_report["runs"] = runs
 
-    return sarif_report
+    return sarif_report

src/report/parsers/checkov_parser.py

@@ -45,7 +45,8 @@ def parse_checkov_vulns(report: Dict[str, Dict[str, Dict[str, List[Dict[str, str
                     vulnerability.set_severity(vuln['severity'])
                     vulnerability.set_cvss(None)
                     vulnerability.set_epss(None)
-                    vulnerability.set_category(_get_category(r.get("check_type")))
+                    vulnerability.set_category(
+                        _get_category(r.get("check_type")))
                     vulnerability.set_rule(vuln['check_id'])
                     vulnerability.set_file(vuln['file_path'])
                     vulnerability.set_location(vuln['resource'])

src/report/parsers/osv_scanner_parser.py

@@ -54,12 +54,12 @@ def parse_osv_scanner_vulns(report: Dict[str, List[Dict[str, List[Dict[str, str]
     if results:
         for result in results:
             vulns = [vuln
-                    for package in result.get('packages')
-                    for vuln in package.get('vulnerabilities')]
+                     for package in result.get('packages')
+                     for vuln in package.get('vulnerabilities')]
 
             for vuln in vulns:
                 vuln_id = next((alias for alias in vuln.get('aliases')
-                            if alias.startswith('CVE-')), vuln.get('id'))
+                                if alias.startswith('CVE-')), vuln.get('id'))
                 affected = vuln.get('affected')
                 fix_version = _get_fix_version(affected)
                 db_specific = vuln.get('database_specific')
@@ -71,7 +71,8 @@ def parse_osv_scanner_vulns(report: Dict[str, List[Dict[str, List[Dict[str, str]
                 severity = vuln['severity'][0]['score']
 
                 vulnerability = Vulnerability()
-                vulnerability.set_name(f"{vuln_id} ({package_name}): {summary}")
+                vulnerability.set_name(
+                    f"{vuln_id} ({package_name}): {summary}")
                 vulnerability.set_description(details)
                 vulnerability.set_identifier(vuln_id)
                 vulnerability.set_severity(vuln_severity)

src/report/process/group_vulnerabilities.py

@@ -23,7 +23,8 @@ def group_sast_vulnerabilities(vulnerabilities: List[Vulnerability]) -> List[Dic
                 and grouped_vuln.location == vuln.location
             ):
                 if vuln.tools[0] not in grouped_vuln.tools:
-                    grouped_vuln.set_tools(grouped_vuln.get_tools().append((vuln.tools[0])))
+                    grouped_vuln.set_tools(
+                        grouped_vuln.get_tools().append((vuln.tools[0])))
                 added = True
                 break
         if not added:
@@ -51,7 +52,8 @@ def group_sca_vulnerabilities(vulnerabilities: List[Vulnerability]) -> List[Dict
                 and grouped_vuln.location == vuln.location
             ):
                 if vuln.tools[0] not in grouped_vuln.tools:
-                    grouped_vuln.set_tools(grouped_vuln.get_tools().append((vuln.tools[0])))
+                    grouped_vuln.set_tools(
+                        grouped_vuln.get_tools().append((vuln.tools[0])))
                 added = True
                 break
         if not added:
@@ -79,7 +81,8 @@ def group_container_vulnerabilities(vulnerabilities: List[Vulnerability]) -> Lis
                 and grouped_vuln.location == vuln.location
             ):
                 if vuln.tools[0] not in grouped_vuln.tools:
-                    grouped_vuln.set_tools(grouped_vuln.get_tools().append((vuln.tools[0])))
+                    grouped_vuln.set_tools(
+                        grouped_vuln.get_tools().append((vuln.tools[0])))
                 added = True
                 break
         if not added:
@@ -107,7 +110,8 @@ def group_iac_vulnerabilities(vulnerabilities: List[Vulnerability]) -> List[Dict
                 and grouped_vuln.location == vuln.location
             ):
                 if vuln.tools[0] not in grouped_vuln.tools:
-                    grouped_vuln.set_tools(grouped_vuln.get_tools().append((vuln.tools[0])))
+                    grouped_vuln.set_tools(
+                        grouped_vuln.get_tools().append((vuln.tools[0])))
                 added = True
                 break
         if not added:
@@ -134,9 +138,10 @@ def group_secrets_vulnerabilities(vulnerabilities: List[Vulnerability]) -> List[
                 and grouped_vuln.location == vuln.location
             ):
                 if vuln.tools[0] not in grouped_vuln.tools:
-                    grouped_vuln.set_tools(grouped_vuln.get_tools().append((vuln.tools[0])))
+                    grouped_vuln.set_tools(
+                        grouped_vuln.get_tools().append((vuln.tools[0])))
                 added = True
                 break
         if not added:
             grouped_vulnerabilities.append(vuln)
-    return [vuln.__dict__ for vuln in grouped_vulnerabilities]
+    return [vuln.__dict__ for vuln in grouped_vulnerabilities]