Adjust for parity

cli_isolated/main.py

@@ -1,18 +1,20 @@
 from pathlib import Path
 from safety_schemas.models import FileType
-
 from read_dependency_files import read_dependency_files
 from vulnerability_checker import check_vulnerabilities
-from rich.console import Console
-
-console = Console()
 
+# Initialize input files
 file_path = Path("/Users/dylanpulver/Repos/pyup/safety/test_requirements.txt")
 file_type = FileType.REQUIREMENTS_TXT
 
-# Call the function and consume the generator
-# Process and check vulnerabilities
+# String buffer to accumulate results
+output_buffer = []
+
+# Process files and check for vulnerabilities
 for path, inspectable_file in read_dependency_files([file_path], [file_type]):
-    console.print(f"Processed file: {path}")
-    file_model = check_vulnerabilities(path, inspectable_file, console)
-    console.print(f"File Model: {file_model}")
+    output_buffer.append(f"Processed file: {path}")
+    file_model, file_output = check_vulnerabilities(path, inspectable_file)
+    output_buffer.append(file_output)
+
+# Print the final results
+print("\n".join(output_buffer))

cli_isolated/read_dependency_files.py

@@ -24,13 +24,10 @@ def read_dependency_files(file_paths: List[Path], file_types: List[FileType], co
         if not file_path.exists() or not file_path.is_file():
             raise FileNotFoundError(f"File not found: {file_path}")
 
-        print(file_path, file_type)
-        # Wrap the file in InspectableFileContext
         with InspectableFileContext(file_path, file_type=file_type) as inspectable_file:
-            if inspectable_file:  # Ensure the file was successfully wrapped
+            if inspectable_file:
                 inspectable_file.inspect(config=config)
                 inspectable_file.remediate()
                 yield file_path, inspectable_file
             else:
                 print(f"Unable to process file: {file_path}")
-

cli_isolated/vulnerability_checker.py

@@ -1,67 +1,114 @@
-from typing import Generator, Tuple
 from pathlib import Path
+from typing import Tuple
 from safety_schemas.models import FileModel, Vulnerability, VulnerabilitySeverityLabels
-from rich.console import Console
 
-# Sort vulnerabilities by severity
-def sort_vulns_by_score(vuln: Vulnerability) -> int:
-    if vuln.severity and vuln.severity.cvssv3:
-        return vuln.severity.cvssv3.get("base_score", 0)
-    return 0
+
+EXIT_CODE_VULNERABILITIES_FOUND = 64
+
+
+def pluralize(word: str, count: int = 0) -> str:
+    """
+    Pluralize a word based on the count.
+
+    Args:
+        word (str): The word to pluralize.
+        count (int): The count.
+
+    Returns:
+        str: The pluralized word.
+    """
+    if count == 1:
+        return word
+
+    default = {"was": "were", "this": "these", "has": "have"}
+
+    if word in default:
+        return default[word]
+
+    if word.endswith("s") or word.endswith("x") or word.endswith("z") \
+        or word.endswith("ch") or word.endswith("sh"):
+        return word + "es"
+
+    if word.endswith("y"):
+        if word[-2] in "aeiou":
+            return word + "s"
+        else:
+            return word[:-1] + "ies"
+
+    return word + "s"
 
 def check_vulnerabilities(
     file_path: Path,
     inspectable_file,
-    console: Console
-) -> FileModel:
+) -> Tuple[FileModel, str]:
     """
-    Checks for vulnerabilities in the given InspectableFile and prepares a FileModel.
+    Checks for vulnerabilities in a dependency file and returns a report.
 
     Args:
         file_path (Path): The path to the processed file.
         inspectable_file: The processed InspectableFile.
-        console (Console): The console for logging.
 
     Returns:
-        FileModel: A structured model with vulnerabilities and results.
+        Tuple[FileModel, str]: A FileModel and a string summary of vulnerabilities.
     """
+    exit_code = 0
+    count_vulns_scanned = 0
+    affected_count = 0
+    dependency_vuln_detected = False
+    count_vulns_scanned += len(inspectable_file.dependency_results.dependencies)
+
+    if exit_code == 0 and inspectable_file.dependency_results.failed:
+                exit_code = EXIT_CODE_VULNERABILITIES_FOUND
+
+
     dependency_results = inspectable_file.dependency_results
     affected_specifications = dependency_results.get_affected_specifications()
+    affected_count += len(affected_specifications)
+
+    # Initialize the string buffer for the output
+    output = [""]
 
-    # Sort vulnerabilities by severity
-    def sort_vulns_by_score(vuln: Vulnerability) -> int:
-        if vuln.severity and vuln.severity.cvssv3:
-            return vuln.severity.cvssv3.get("base_score", 0)
-        return 0
+    if any(affected_specifications):
+        if not dependency_vuln_detected:
+            output.append("Dependency vulnerabilities detected:")
+            dependency_vuln_detected = True
 
-    if affected_specifications:
-        console.print(f"Vulnerabilities found in {file_path}:")
+        msg = f":pencil: [file_title]{file_path}:[/file_title]"
+        output.append(msg)
 
         for spec in affected_specifications:
+            # Sort vulnerabilities by severity
             vulns_to_report = sorted(
                 [v for v in spec.vulnerabilities if not v.ignored],
-                key=sort_vulns_by_score,
+                key=lambda v: v.severity.cvssv3.get("base_score", 0) if v.severity and v.severity.cvssv3 else 0,
                 reverse=True,
             )
-            critical_vulns = sum(
+            critical_vulns_count = sum(
                 1
                 for v in vulns_to_report
                 if v.severity
                 and v.severity.cvssv3
                 and v.severity.cvssv3.get("base_severity", "").lower()
                 == VulnerabilitySeverityLabels.CRITICAL.value.lower()
             )
-            console.print(f"- {spec.name}: {len(vulns_to_report)} vulnerabilities")
-            if critical_vulns:
-                console.print(f"  Critical: {critical_vulns}")
+
+            vulns_found = len(vulns_to_report)
+            vuln_word = pluralize("vulnerability", vulns_found)
+
+            msg = f"[dep_name]{spec.name}[/dep_name][specifier]{spec.raw.replace(spec.name, '')}[/specifier]  [{vulns_found} {vuln_word} found"
+
+            if vulns_found > 3 and critical_vulns_count > 0:
+                msg += f", [brief_severity]including {critical_vulns_count} critical severity {pluralize('vulnerability', critical_vulns_count)}[/brief_severity]"
+
+            output.append(msg)
 
     else:
-        console.print(f"No vulnerabilities found in {file_path}.")
+        output.append(f"No vulnerabilities found in {file_path}.")
 
-    # Prepare and return a FileModel
-    return FileModel(
+    # Prepare and return the FileModel
+    file_model = FileModel(
         location=file_path,
         file_type=inspectable_file.file_type,
         results=dependency_results,
     )
-
+    return file_model, "\n".join(output)