update ruff; enable more rules; bump codebase to py39 syntax

Author: MarshalX

CONTRIBUTING.md

@@ -11,7 +11,7 @@ But it’s fine to use a higher version without using new features from these ve
 The project is under Poetry project management.
 To deal with it, you should install it on your system:
 
-Install Poetry (feel free to use Brew, etc):
+Install Poetry (feel free to use Brew, etc.):
 
 ```shell
 curl -sSL https://install.python-poetry.org | python - -y
@@ -70,6 +70,8 @@ poetry run ruff format .
 
 Many rules support auto-fixing. You can run it with the `--fix` flag.
 
+Plugin for JB IDEs with auto formatting on save is available [here](https://plugins.jetbrains.com/plugin/20574-ruff).
+
 ### Branching and versioning
 
 We use the `main` branch as the main one.

cycode/cli/apps/auth/auth_command.py

@@ -16,7 +16,6 @@ def auth_command(ctx: typer.Context) -> None:
     * `cycode auth`: Start interactive authentication
     * `cycode auth --help`: View authentication options
     """
-
     add_breadcrumb('auth')
     printer = ctx.obj.get('console_printer')
 

cycode/cli/apps/auth/auth_manager.py

@@ -1,6 +1,6 @@
 import time
 import webbrowser
-from typing import TYPE_CHECKING, Tuple
+from typing import TYPE_CHECKING
 
 from cycode.cli.exceptions.custom_exceptions import AuthProcessError
 from cycode.cli.user_settings.configuration_manager import ConfigurationManager
@@ -78,7 +78,7 @@ def get_api_token_polling(self, session_id: str, code_verifier: str) -> 'ApiToke
     def save_api_token(self, api_token: 'ApiToken') -> None:
         self.credentials_manager.update_credentials(api_token.client_id, api_token.secret)
 
-    def _generate_pkce_code_pair(self) -> Tuple[str, str]:
+    def _generate_pkce_code_pair(self) -> tuple[str, str]:
         code_verifier = generate_random_string(self.CODE_VERIFIER_LENGTH)
         code_challenge = hash_string_to_sha256(code_verifier)
         return code_challenge, code_verifier

cycode/cli/apps/scan/code_scanner.py

@@ -3,7 +3,7 @@
 import sys
 import time
 from platform import platform
-from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Callable, Optional
 from uuid import UUID, uuid4
 
 import click
@@ -84,7 +84,7 @@ def scan_sca_commit_range(ctx: typer.Context, path: str, commit_range: str) -> N
     scan_commit_range_documents(ctx, from_commit_documents, to_commit_documents, scan_parameters=scan_parameters)
 
 
-def scan_disk_files(ctx: typer.Context, paths: Tuple[str]) -> None:
+def scan_disk_files(ctx: typer.Context, paths: tuple[str, ...]) -> None:
     scan_type = ctx.obj['scan_type']
     progress_bar = ctx.obj['progress_bar']
 
@@ -96,7 +96,7 @@ def scan_disk_files(ctx: typer.Context, paths: Tuple[str]) -> None:
         handle_scan_exception(ctx, e)
 
 
-def set_issue_detected_by_scan_results(ctx: typer.Context, scan_results: List[LocalScanResult]) -> None:
+def set_issue_detected_by_scan_results(ctx: typer.Context, scan_results: list[LocalScanResult]) -> None:
     set_issue_detected(ctx, any(scan_result.issue_detected for scan_result in scan_results))
 
 
@@ -110,6 +110,7 @@ def _should_use_sync_flow(command_scan_type: str, scan_type: str, sync_option: b
     - for IAC scan, sync flow is always used
     - for SAST scan, sync flow is not supported
     - for SCA and Secrets scan, sync flow is supported only for path/repository scan
+
     """
     if not sync_option and scan_type != consts.IAC_SCAN_TYPE:
         return False
@@ -161,14 +162,14 @@ def _enrich_scan_result_with_data_from_detection_rules(
 
 def _get_scan_documents_thread_func(
     ctx: typer.Context, is_git_diff: bool, is_commit_range: bool, scan_parameters: dict
-) -> Callable[[List[Document]], Tuple[str, CliError, LocalScanResult]]:
+) -> Callable[[list[Document]], tuple[str, CliError, LocalScanResult]]:
     cycode_client = ctx.obj['client']
     scan_type = ctx.obj['scan_type']
     severity_threshold = ctx.obj['severity_threshold']
     sync_option = ctx.obj['sync']
     command_scan_type = ctx.info_name
 
-    def _scan_batch_thread_func(batch: List[Document]) -> Tuple[str, CliError, LocalScanResult]:
+    def _scan_batch_thread_func(batch: list[Document]) -> tuple[str, CliError, LocalScanResult]:
         local_scan_result = error = error_message = None
         detections_count = relevant_detections_count = zip_file_size = 0
 
@@ -297,7 +298,7 @@ def scan_commit_range(
 
 def scan_documents(
     ctx: typer.Context,
-    documents_to_scan: List[Document],
+    documents_to_scan: list[Document],
     scan_parameters: dict,
     is_git_diff: bool = False,
     is_commit_range: bool = False,
@@ -335,13 +336,12 @@ def scan_documents(
 
 def scan_commit_range_documents(
     ctx: typer.Context,
-    from_documents_to_scan: List[Document],
-    to_documents_to_scan: List[Document],
+    from_documents_to_scan: list[Document],
+    to_documents_to_scan: list[Document],
     scan_parameters: Optional[dict] = None,
     timeout: Optional[int] = None,
 ) -> None:
-    """Used by SCA only"""
-
+    """In use by SCA only."""
     cycode_client = ctx.obj['client']
     scan_type = ctx.obj['scan_type']
     severity_threshold = ctx.obj['severity_threshold']
@@ -424,13 +424,13 @@ def scan_commit_range_documents(
     )
 
 
-def should_scan_documents(from_documents_to_scan: List[Document], to_documents_to_scan: List[Document]) -> bool:
+def should_scan_documents(from_documents_to_scan: list[Document], to_documents_to_scan: list[Document]) -> bool:
     return len(from_documents_to_scan) > 0 or len(to_documents_to_scan) > 0
 
 
 def create_local_scan_result(
     scan_result: ZippedFileScanResult,
-    documents_to_scan: List[Document],
+    documents_to_scan: list[Document],
     command_scan_type: str,
     scan_type: str,
     severity_threshold: str,
@@ -568,15 +568,15 @@ def print_debug_scan_details(scan_details_response: 'ScanDetailsResponse') -> No
 
 
 def print_results(
-    ctx: typer.Context, local_scan_results: List[LocalScanResult], errors: Optional[Dict[str, 'CliError']] = None
+    ctx: typer.Context, local_scan_results: list[LocalScanResult], errors: Optional[dict[str, 'CliError']] = None
 ) -> None:
     printer = ctx.obj.get('console_printer')
     printer.print_scan_results(local_scan_results, errors)
 
 
 def get_document_detections(
-    scan_result: ZippedFileScanResult, documents_to_scan: List[Document]
-) -> List[DocumentDetections]:
+    scan_result: ZippedFileScanResult, documents_to_scan: list[Document]
+) -> list[DocumentDetections]:
     logger.debug('Getting document detections')
 
     document_detections = []
@@ -595,11 +595,11 @@ def get_document_detections(
 
 
 def exclude_irrelevant_document_detections(
-    document_detections_list: List[DocumentDetections],
+    document_detections_list: list[DocumentDetections],
     scan_type: str,
     command_scan_type: str,
     severity_threshold: str,
-) -> List[DocumentDetections]:
+) -> list[DocumentDetections]:
     relevant_document_detections_list = []
     for document_detections in document_detections_list:
         relevant_detections = exclude_irrelevant_detections(
@@ -614,8 +614,7 @@ def exclude_irrelevant_document_detections(
 
 
 def parse_pre_receive_input() -> str:
-    """
-    Parsing input to pushed branch update details
+    """Parse input to pushed branch update details.
 
     Example input:
     old_value new_value refname
@@ -624,7 +623,7 @@ def parse_pre_receive_input() -> str:
     973a96d3e925b65941f7c47fa16129f1577d499f 0000000000000000000000000000000000000000 refs/heads/feature-branch
     59564ef68745bca38c42fc57a7822efd519a6bd9 3378e52dcfa47fb11ce3a4a520bea5f85d5d0bf3 refs/heads/develop
 
-    :return: first branch update details (input's first line)
+    :return: First branch update details (input's first line)
     """
     # FIXME(MarshalX): this blocks main thread forever if called outside of pre-receive hook
     pre_receive_input = sys.stdin.read().strip()
@@ -649,7 +648,7 @@ def _get_default_scan_parameters(ctx: typer.Context) -> dict:
     }
 
 
-def get_scan_parameters(ctx: typer.Context, paths: Optional[Tuple[str]] = None) -> dict:
+def get_scan_parameters(ctx: typer.Context, paths: Optional[tuple[str, ...]] = None) -> dict:
     scan_parameters = _get_default_scan_parameters(ctx)
 
     if not paths:
@@ -684,15 +683,14 @@ def try_get_git_remote_url(path: str) -> Optional[str]:
 
 
 def _get_plastic_repository_name(path: str) -> Optional[str]:
-    """Gets the name of the Plastic repository from the current working directory.
+    """Get the name of the Plastic repository from the current working directory.
 
     The command to execute is:
         cm status --header --machinereadable --fieldseparator=":::"
 
     Example of status header in machine-readable format:
         STATUS:::0:::Project/RepoName:::OrgName@ServerInfo
     """
-
     try:
         command = [
             'cm',
@@ -718,8 +716,8 @@ def _get_plastic_repository_name(path: str) -> Optional[str]:
         return None
 
 
-def _get_plastic_repository_list(working_dir: Optional[str] = None) -> Dict[str, str]:
-    """Gets the list of Plastic repositories and their GUIDs.
+def _get_plastic_repository_list(working_dir: Optional[str] = None) -> dict[str, str]:
+    """Get the list of Plastic repositories and their GUIDs.
 
     The command to execute is:
         cm repo list --format="{repname}:::{repguid}"
@@ -729,7 +727,6 @@ def _get_plastic_repository_list(working_dir: Optional[str] = None) -> Dict[str,
 
     Each line represents an individual repository.
     """
-
     repo_name_to_guid = {}
 
     try:
@@ -771,14 +768,14 @@ def try_to_get_plastic_remote_url(path: str) -> Optional[str]:
 
 
 def exclude_irrelevant_detections(
-    detections: List[Detection], scan_type: str, command_scan_type: str, severity_threshold: str
-) -> List[Detection]:
+    detections: list[Detection], scan_type: str, command_scan_type: str, severity_threshold: str
+) -> list[Detection]:
     relevant_detections = _exclude_detections_by_exclusions_configuration(detections, scan_type)
     relevant_detections = _exclude_detections_by_scan_type(relevant_detections, scan_type, command_scan_type)
     return _exclude_detections_by_severity(relevant_detections, severity_threshold)
 
 
-def _exclude_detections_by_severity(detections: List[Detection], severity_threshold: str) -> List[Detection]:
+def _exclude_detections_by_severity(detections: list[Detection], severity_threshold: str) -> list[Detection]:
     relevant_detections = []
     for detection in detections:
         severity = detection.severity
@@ -795,8 +792,8 @@ def _exclude_detections_by_severity(detections: List[Detection], severity_thresh
 
 
 def _exclude_detections_by_scan_type(
-    detections: List[Detection], scan_type: str, command_scan_type: str
-) -> List[Detection]:
+    detections: list[Detection], scan_type: str, command_scan_type: str
+) -> list[Detection]:
     if command_scan_type == consts.PRE_COMMIT_COMMAND_SCAN_TYPE:
         return exclude_detections_in_deleted_lines(detections)
 
@@ -811,16 +808,16 @@ def _exclude_detections_by_scan_type(
     return detections
 
 
-def exclude_detections_in_deleted_lines(detections: List[Detection]) -> List[Detection]:
+def exclude_detections_in_deleted_lines(detections: list[Detection]) -> list[Detection]:
     return [detection for detection in detections if detection.detection_details.get('line_type') != 'Removed']
 
 
-def _exclude_detections_by_exclusions_configuration(detections: List[Detection], scan_type: str) -> List[Detection]:
+def _exclude_detections_by_exclusions_configuration(detections: list[Detection], scan_type: str) -> list[Detection]:
     exclusions = configuration_manager.get_exclusions_by_scan_type(scan_type)
     return [detection for detection in detections if not _should_exclude_detection(detection, exclusions)]
 
 
-def _should_exclude_detection(detection: Detection, exclusions: Dict) -> bool:
+def _should_exclude_detection(detection: Detection, exclusions: dict) -> bool:
     # FIXME(MarshalX): what the difference between by_value and by_sha?
     exclusions_by_value = exclusions.get(consts.EXCLUSIONS_BY_VALUE_SECTION_NAME, [])
     if _is_detection_sha_configured_in_exclusions(detection, exclusions_by_value):
@@ -862,7 +859,7 @@ def _should_exclude_detection(detection: Detection, exclusions: Dict) -> bool:
     return False
 
 
-def _is_detection_sha_configured_in_exclusions(detection: Detection, exclusions: List[str]) -> bool:
+def _is_detection_sha_configured_in_exclusions(detection: Detection, exclusions: list[str]) -> bool:
     detection_sha = detection.detection_details.get('sha512')
     return detection_sha in exclusions
 
@@ -886,7 +883,7 @@ def _get_cve_identifier(detection: Detection) -> Optional[str]:
 
 
 def _get_document_by_file_name(
-    documents: List[Document], file_name: str, unique_id: Optional[str] = None
+    documents: list[Document], file_name: str, unique_id: Optional[str] = None
 ) -> Optional[Document]:
     for document in documents:
         if _normalize_file_path(document.path) == _normalize_file_path(file_name) and document.unique_id == unique_id:
@@ -992,10 +989,11 @@ def _try_get_aggregation_report_url_if_needed(
         logger.debug('Failed to get aggregation report url: %s', str(e))
 
 
-def _map_detections_per_file_and_commit_id(scan_type: str, raw_detections: List[dict]) -> List[DetectionsPerFile]:
-    """Converts list of detections (async flow) to list of DetectionsPerFile objects (sync flow).
+def _map_detections_per_file_and_commit_id(scan_type: str, raw_detections: list[dict]) -> list[DetectionsPerFile]:
+    """Convert a list of detections (async flow) to list of DetectionsPerFile objects (sync flow).
 
     Args:
+        scan_type: Type of the scan.
         raw_detections: List of detections as is returned from the server.
 
     Note:
@@ -1004,6 +1002,7 @@ def _map_detections_per_file_and_commit_id(scan_type: str, raw_detections: List[
 
     Note:
         Aggregation is performed by file name and commit ID (if available)
+
     """
     detections_per_files = {}
     for raw_detection in raw_detections:
@@ -1045,7 +1044,7 @@ def _get_secret_file_name_from_detection(raw_detection: dict) -> str:
     return os.path.join(file_path, file_name)
 
 
-def _does_reach_to_max_commits_to_scan_limit(commit_ids: List[str], max_commits_count: Optional[int]) -> bool:
+def _does_reach_to_max_commits_to_scan_limit(commit_ids: list[str], max_commits_count: Optional[int]) -> bool:
     if max_commits_count is None:
         return False
 

cycode/cli/apps/scan/path/path_command.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Annotated, List
+from typing import Annotated
 
 import typer
 
@@ -11,7 +11,7 @@
 def path_command(
     ctx: typer.Context,
     paths: Annotated[
-        List[Path], typer.Argument(exists=True, resolve_path=True, help='Paths to scan', show_default=False)
+        list[Path], typer.Argument(exists=True, resolve_path=True, help='Paths to scan', show_default=False)
     ],
 ) -> None:
     add_breadcrumb('path')

cycode/cli/apps/scan/pre_commit/pre_commit_command.py

@@ -1,5 +1,5 @@
 import os
-from typing import Annotated, List, Optional
+from typing import Annotated, Optional
 
 import typer
 
@@ -21,7 +21,7 @@
 
 def pre_commit_command(
     ctx: typer.Context,
-    _: Annotated[Optional[List[str]], typer.Argument(help='Ignored arguments', hidden=True)] = None,
+    _: Annotated[Optional[list[str]], typer.Argument(help='Ignored arguments', hidden=True)] = None,
 ) -> None:
     add_breadcrumb('pre_commit')
 

cycode/cli/apps/scan/pre_receive/pre_receive_command.py

@@ -1,5 +1,5 @@
 import os
-from typing import Annotated, List, Optional
+from typing import Annotated, Optional
 
 import click
 import typer
@@ -25,7 +25,7 @@
 
 def pre_receive_command(
     ctx: typer.Context,
-    _: Annotated[Optional[List[str]], typer.Argument(help='Ignored arguments', hidden=True)] = None,
+    _: Annotated[Optional[list[str]], typer.Argument(help='Ignored arguments', hidden=True)] = None,
 ) -> None:
     try:
         add_breadcrumb('pre_receive')