Ruff: Add PTH123, merge PTH, fix in /dojo (#12025)

Author: kiblik

dojo/api_v2/views.py

@@ -2,6 +2,7 @@
 import logging
 import mimetypes
 from datetime import datetime
+from pathlib import Path
 
 import tagulous
 from crum import get_current_user
@@ -724,12 +725,12 @@ def download_proof(self, request, pk=None):
                 status=status.HTTP_404_NOT_FOUND,
             )
         # Get the path of the file in media root
-        file_path = f"{settings.MEDIA_ROOT}/{file_object.name}"
-        file_handle = open(file_path, "rb")
+        file_path = Path(settings.MEDIA_ROOT) / file_object.name
+        file_handle = file_path.open("rb")
         # send file
         response = FileResponse(
             file_handle,
-            content_type=f"{mimetypes.guess_type(file_path)}",
+            content_type=f"{mimetypes.guess_type(str(file_path))}",
             status=status.HTTP_200_OK,
         )
         response["Content-Length"] = file_object.size

dojo/engagement/views.py

@@ -5,6 +5,7 @@
 import re
 from datetime import datetime
 from functools import reduce
+from pathlib import Path
 from tempfile import NamedTemporaryFile
 from time import strftime
 
@@ -1462,7 +1463,7 @@ def download_risk_acceptance(request, eid, raid):
         raise PermissionDenied
     response = StreamingHttpResponse(
         FileIterWrapper(
-            open(settings.MEDIA_ROOT + "/" + risk_acceptance.path.name, mode="rb")))
+            (Path(settings.MEDIA_ROOT) / "risk_acceptance.path.name").open(mode="rb")))
     response["Content-Disposition"] = f'attachment; filename="{risk_acceptance.filename()}"'
     mimetype, _encoding = mimetypes.guess_type(risk_acceptance.path.name)
     response["Content-Type"] = mimetype

dojo/finding/views.py

@@ -7,6 +7,7 @@
 import mimetypes
 from collections import OrderedDict, defaultdict
 from itertools import chain
+from pathlib import Path
 
 from django.conf import settings
 from django.contrib import messages
@@ -2460,7 +2461,7 @@ class Original(ImageSpec):
     except Exception:
         raise PermissionDenied
 
-    with open(access_token.file.file.file.name, "rb") as file:
+    with Path(access_token.file.file.file.name).open("rb") as file:
         file_name = file.name
         image = size_map[size](source=file).generate()
         response = StreamingHttpResponse(FileIterWrapper(image))

dojo/jira_link/helper.py

@@ -1225,7 +1225,7 @@ def jira_attachment(finding, jira, issue, file, jira_filename=None):
                     issue=issue, attachment=attachment, filename=jira_filename)
             else:
                 # read and upload a file
-                with open(file, "rb") as f:
+                with Path(file).open("rb") as f:
                     jira.add_attachment(issue=issue, attachment=f)
         except JIRAError as e:
             logger.exception("Unable to add attachment")

dojo/management/commands/csv_findings_export.py

@@ -1,4 +1,5 @@
 import csv
+from pathlib import Path
 
 from django.core.management.base import BaseCommand
 from pytz import timezone
@@ -21,12 +22,12 @@ def add_arguments(self, parser):
         parser.add_argument("file_path")
 
     def handle(self, *args, **options):
-        file_path = options["file_path"]
+        file_path = Path(options["file_path"])
 
         findings = Finding.objects.filter(verified=True,
                                           active=True).select_related(
             "test__engagement__product")
-        writer = csv.writer(open(file_path, "w", encoding="utf-8"))
+        writer = csv.writer(file_path.open("w", encoding="utf-8"))
 
         headers = [
             "product_name",

dojo/management/commands/import_surveys.py

@@ -27,9 +27,8 @@ def handle(self, *args, **options):
             row = cursor.fetchone()
             ctype_id = row[0]
         # Find the current id in the surveys file
-        path = Path(__file__).parent.absolute()
-        path = path[:-19] + "fixtures/initial_surveys.json"
-        contents = open(path, encoding="utf-8").readlines()
+        path = Path(__file__).parent.parent.parent / "fixtures" / "initial_surveys.json"
+        contents = path.open(encoding="utf-8").readlines()
         for line in contents:
             if '"polymorphic_ctype": ' in line:
                 matchedLine = line
@@ -38,7 +37,7 @@ def handle(self, *args, **options):
         old_id = "".join(c for c in matchedLine if c.isdigit())
         new_line = matchedLine.replace(old_id, str(ctype_id))
         # Replace the all lines in the file
-        with open(path, "w", encoding="utf-8") as fout:
+        with path.open("w", encoding="utf-8") as fout:
             fout.writelines(line.replace(matchedLine, new_line) for line in contents)
         # Delete the temp question
         created_question.delete()

dojo/tools/blackduck/importer.py

@@ -26,13 +26,13 @@ def parse_findings(self, report: Path) -> Iterable[BlackduckFinding]:
             return self._process_zipfile(report)
         return self._process_csvfile(report)
 
-    def _process_csvfile(self, report):
+    def _process_csvfile(self, report: Path):
         """
         If passed in a regular security.csv, process it.
         No file information then.
         """
         security_issues = {}
-        with open(str(report), encoding="utf-8") as f:
+        with report.open(encoding="utf-8") as f:
             security_issues = self.__partition_by_key(f)
 
         project_ids = set(security_issues.keys())

dojo/tools/blackduck_binary_analysis/importer.py

@@ -22,10 +22,10 @@ def parse_findings(self, report: Path) -> Iterable[BlackduckBinaryAnalysisFindin
 
         return self._process_csvfile(report, orig_report_name)
 
-    def _process_csvfile(self, report, orig_report_name):
+    def _process_csvfile(self, report: Path, orig_report_name):
         """If passed a CSV file, process."""
         vulnerabilities = {}
-        with open(str(report), encoding="utf-8") as f:
+        with report.open(encoding="utf-8") as f:
             vulnerabilities = self.__partition_by_key(f)
 
         sha1_hash_keys = set(vulnerabilities.keys())

dojo/utils.py

@@ -1377,23 +1377,24 @@ def handle_uploaded_threat(f, eng):
     path = Path(f.name)
     extension = path.suffix
     # Check if threat folder exist.
-    if not Path(settings.MEDIA_ROOT + "/threat/").is_dir():
+    threat_dir = Path(settings.MEDIA_ROOT) / "threat"
+    if not threat_dir.is_dir():
         # Create the folder
-        Path(settings.MEDIA_ROOT + "/threat/").mkdir()
-    with open(settings.MEDIA_ROOT + f"/threat/{eng.id}{extension}",
-              "wb+") as destination:
+        threat_dir.mkdir()
+    eng_path = threat_dir / f"{eng.id}{extension}"
+    with eng_path.open("wb+") as destination:
         destination.writelines(chunk for chunk in f.chunks())
-    eng.tmodel_path = settings.MEDIA_ROOT + f"/threat/{eng.id}{extension}"
+    eng.tmodel_path = str(eng_path)
     eng.save()
 
 
 def handle_uploaded_selenium(f, cred):
     path = Path(f.name)
     extension = path.suffix
-    with open(settings.MEDIA_ROOT + f"/selenium/{cred.id}{extension}",
-              "wb+") as destination:
+    sel_path = Path(settings.MEDIA_ROOT) / "selenium" / f"{cred.id}{extension}"
+    with sel_path.open("wb+") as destination:
         destination.writelines(chunk for chunk in f.chunks())
-    cred.selenium_script = settings.MEDIA_ROOT + f"/selenium/{cred.id}{extension}"
+    cred.selenium_script = str(sel_path)
     cred.save()
 
 
@@ -2694,7 +2695,7 @@ def generate_file_response_from_file_path(
     # Generate the FileResponse
     full_file_name = f"{file_name}{file_extension}"
     response = FileResponse(
-        open(file_path, "rb"),
+        path.open("rb"),
         filename=full_file_name,
         content_type=f"{mimetypes.guess_type(file_path)}",
     )

ruff.toml

@@ -68,7 +68,7 @@ select = [
    "TC",
    "INT",
    "ARG003", "ARG004", "ARG005",
-   "PTH2", "PTH10", "PTH11", "PTH120", "PTH121", "PTH122", "PTH124",
+   "PTH",
    "TD001", "TD004", "TD005", "TD007",
    "FIX",
    "PD",
@@ -113,6 +113,7 @@ preview = true
 "unittests/**" = [
     "S105",  # hardcoded passwords in tests are fine
     "S108",  # tmp paths mentioned in tests are fine
+    "PTH123", # Fix is needed in unittests as well but there are too many problematic files, let's do it per partes
 ]
 
 [lint.flake8-boolean-trap]