From b7c10da03c4a9aefd2bd560ff469154212b4961c Mon Sep 17 00:00:00 2001 From: DLS29 Date: Mon, 25 May 2026 21:44:13 +0300 Subject: [PATCH 1/3] Fixes #575: renaming and some minor changes -renaming some functions and variables in cli.py and commands.py -deleted "version" in check_updates in cli.py because version was not used and is hardcoded one --- safety/cli.py | 67 ++++++++++++++++++------------------------ safety/scan/command.py | 49 +++++++++++++++--------------- 2 files changed, 53 insertions(+), 63 deletions(-) diff --git a/safety/cli.py b/safety/cli.py index 14ec46ae..0c0c551c 100644 --- a/safety/cli.py +++ b/safety/cli.py @@ -128,7 +128,7 @@ LOG = logging.getLogger(__name__) -def preprocess_args(f): +def debug_proccesing(f): if "--debug" in sys.argv: index = sys.argv.index("--debug") if len(sys.argv) > index + 1: @@ -178,7 +178,7 @@ def configure_logger(ctx, param, debug): @click.option("--debug", is_flag=True, help=CLI_DEBUG_HELP, callback=configure_logger) @click.version_option(version=get_version()) @click.pass_context -@preprocess_args +@debug_proccesing def cli(ctx, debug, disable_optional_telemetry): """ Scan and secure Python projects against package vulnerabilities. To get started navigate to a Python project and run `safety scan`. @@ -279,7 +279,7 @@ def inner(ctx: "SafetyCustomContext", *args, **kwargs): def print_deprecation_message( - old_command: str, deprecation_date: datetime, new_command: Optional[str] = None + old_command: str, deprecation_date: datetime, suggested_command: Optional[str] = None ) -> None: """ Print a formatted deprecation message for a command. @@ -297,7 +297,7 @@ def print_deprecation_message( Parameters: - old_command (str): The name of the deprecated command. - deprecation_date (datetime): The date when the command will no longer be supported. - - new_command (str, optional): The name of the alternative command to suggest. Default is None. + - suggested_command (str, optional): The name of the alternative command to suggest. Default is None. """ click.echo("\n") click.echo(click.style(BAR_LINE, fg="yellow", bold=True)) @@ -311,11 +311,11 @@ def print_deprecation_message( ) ) - if new_command: + if suggested_command: click.echo("\n") click.echo( click.style("We highly encourage switching to the new ", fg="green") - + click.style(f"`{new_command}`", fg="green", bold=True) + + click.style(f"`{suggested_command}`", fg="green", bold=True) + click.style( " command which is easier to use, more powerful, and can be set up to mimic the deprecated command if required.", fg="green", @@ -559,7 +559,7 @@ def check( and not no_prompt ) kwargs = {"version": json_version} if output == "json" else {} - print_deprecation_message("check", date(2024, 6, 1), new_command="scan") + print_deprecation_message("check", date(2024, 6, 1), suggested_command="scan") # try: packages = safety_core.get_packages(files, stdin) @@ -701,7 +701,7 @@ def check( ) safety_core.save_report(save_html, "safety-report.html", html_report) - print_deprecation_message("check", date(2024, 6, 1), new_command="scan") + print_deprecation_message("check", date(2024, 6, 1), suggested_command="scan") if exit_code and found_vulns: LOG.info("Exiting with default code for vulnerabilities found") sys.exit(EXIT_CODE_VULNERABILITIES_FOUND) @@ -764,7 +764,7 @@ def license(ctx, db, output, cache, files): """ Find the open source licenses used by your Python dependencies. """ - print_deprecation_message("license", date(2024, 6, 1), new_command=None) + print_deprecation_message("license", date(2024, 6, 1), suggested_command=None) LOG.info("Running license command") packages = safety_core.get_packages(files, False) licenses_db = {} @@ -793,10 +793,11 @@ def license(ctx, db, output, cache, files): ) click.secho(output_report, nl=True) - print_deprecation_message("license", date(2024, 6, 1), new_command=None) + print_deprecation_message("license", date(2024, 6, 1), suggested_command=None) @cli.command( + "generate", cls=SafetyCLILegacyCommand, context_settings={CONTEXT_COMMAND_TYPE: CommandType.UTILITY}, help=CLI_GENERATE_HELP, @@ -811,7 +812,7 @@ def license(ctx, db, output, cache, files): @click.pass_context @handle_cmd_exception @notify -def generate(ctx, name, path, minimum_cvss_severity): +def generate_CLI_file(ctx, name, path, minimum_cvss_severity): """Create a boilerplate Safety CLI policy file NAME is the name of the file type to generate. Valid values are: policy_file @@ -946,6 +947,7 @@ def generate_policy_file(name, path): @cli.command( + "validate", cls=SafetyCLILegacyCommand, context_settings={CONTEXT_COMMAND_TYPE: CommandType.UTILITY}, ) @@ -959,7 +961,7 @@ def generate_policy_file(name, path): @click.pass_context @handle_cmd_exception @notify -def validate(ctx, name, version, path): +def validate_policy_file(ctx, name, version, path): """Verify that a local policy file is valid. NAME is the name of the file type to validate. Valid values are: policy_file""" if name != "policy_file": click.secho( @@ -983,7 +985,7 @@ def validate(ctx, name, version, path): ) sys.exit(EXIT_CODE_FAILURE) - def fail_validation(e): + def fail_message(e): click.secho(str(e).lstrip(), fg="red", file=sys.stderr) sys.exit(EXIT_CODE_FAILURE) @@ -1000,7 +1002,7 @@ def fail_validation(e): policy = load_policy_file(Path(path)) except Exception as e: - fail_validation(e) + fail_message(e) click.secho( f"The Safety policy ({version}) file " @@ -1241,10 +1243,6 @@ class Output(str, Enum): @notify def check_updates( ctx: "SafetyCustomContext", - version: Annotated[ - int, - typer.Option(min=1), - ] = 1, output: Annotated[ Output, typer.Option(help="The main output generated by Safety CLI.") ] = Output.SCREEN, @@ -1271,7 +1269,6 @@ def check_updates( with console.status(wait_msg, spinner=DEFAULT_SPINNER): try: data = ctx.obj.auth.platform.check_updates( - version=1, safety_version=VERSION, python_version=PYTHON_VERSION, os_type=OS_TYPE, @@ -1294,13 +1291,11 @@ def check_updates( } console.print_json(json.dumps(response)) else: - console.print() console.print( - "[red]Safety is not authenticated, please first authenticate and try again.[/red]" + "\n[red]Safety is not authenticated, please first authenticate and try again.[/red]" ) - console.print() console.print( - "To authenticate, use the `auth` command: `safety auth login` Or for more help: `safety auth —help`" + "\nTo authenticate, use the `auth` command: `safety auth login` Or for more help: `safety auth —help`" ) sys.exit(1) @@ -1316,43 +1311,39 @@ def check_updates( current_version = ( f"Current version: {VERSION} (Python {PYTHON_VERSION} on {OS_TYPE})" ) - latest_available_version = data.get("safety_updates", {}).get("stable_version", "-") + latest_stable_version = data.get("safety_updates", {}).get("stable_version", "-") details = [ f"Organization: {organization}", f"Account: {account}", current_version, - f"Latest stable available version: {latest_available_version}", + f"Latest stable available version: {latest_stable_version}", ] for msg in details: - console.print(Padding(msg, (0, 0, 0, 1)), emoji=True) + console.print(Padding(msg, (0, 0, 1, 1)), emoji=True) - console.print() - - if latest_available_version: + if latest_stable_version: try: # Compare the current version and the latest available version using packaging.version if packaging_version.parse( - latest_available_version + latest_stable_version ) > packaging_version.parse(VERSION): console.print( - f"Update available: Safety version {latest_available_version}" + f"Update available: Safety version {latest_stable_version}" ) - console.print() console.print( - f"If Safety was installed from a requirements file, update Safety to version {latest_available_version} in that requirements file." + f"\nIf Safety was installed from a requirements file, update Safety to version {latest_stable_version} in that requirements file." ) - console.print() console.print( - f"Pip: To install the updated version of Safety directly via pip, run: pip install safety=={latest_available_version}" + f"\nPip: To install the updated version of Safety directly via pip, run: pip install safety=={latest_stable_version}" ) elif packaging_version.parse( - latest_available_version + latest_stable_version ) < packaging_version.parse(VERSION): # Notify user about downgrading console.print( - f"Latest stable version is {latest_available_version}. If you want to downgrade to this version, you can run: pip install safety=={latest_available_version}" + f"Latest stable version is {latest_stable_version}. If you want to downgrade to this version, you can run: pip install safety=={latest_stable_version}" ) else: console.print( @@ -1361,7 +1352,7 @@ def check_updates( except InvalidVersion as invalid_version: LOG.exception(f"Invalid version format encountered: {invalid_version}") console.print( - f"Error: Invalid version format encountered for the latest available version: {latest_available_version}" + f"Error: Invalid version format encountered for the latest available version: {latest_stable_version}" ) console.print("Please report this issue or try again later.") diff --git a/safety/scan/command.py b/safety/scan/command.py index 3ba663a9..3cd792c1 100644 --- a/safety/scan/command.py +++ b/safety/scan/command.py @@ -169,7 +169,7 @@ def process_report( obj: "SafetyContext", console: Console, report: ReportModel, - output: str, + output_format: str, save_as: Optional[Tuple[str, Path]], detailed_output: bool = False, filter_keys: Optional[List[str]] = None, @@ -200,7 +200,7 @@ def process_report( export_type, export_path = save_as export_type = ScanExport(export_type) - output = ScanOutput(output) + output_format = ScanOutput(output_format) report_to_export = None report_to_output = None @@ -209,7 +209,7 @@ def process_report( spdx_format, html_format = None, None if ScanExport.is_format(export_type, ScanExport.SPDX) or ScanOutput.is_format( - output, ScanOutput.SPDX + output_format, ScanOutput.SPDX ): spdx_version = None if export_type: @@ -220,16 +220,16 @@ def process_report( else None ) - if not spdx_version and output: + if not spdx_version and output_format: spdx_version = ( - output.version - if output.version and ScanOutput.is_format(output, ScanOutput.SPDX) + output_format.version + if output_format.version and ScanOutput.is_format(output_format, ScanOutput.SPDX) else None ) spdx_format = render_scan_spdx(report, obj, spdx_version=spdx_version) - if export_type is ScanExport.HTML or output is ScanOutput.HTML: + if export_type is ScanExport.HTML or output_format is ScanOutput.HTML: html_format = render_scan_html(report, obj) save_as_format_mapping = { @@ -249,7 +249,7 @@ def process_report( } report_to_export = save_as_format_mapping.get(export_type, None) - report_to_output = output_format_mapping.get(output, None) + report_to_output = output_format_mapping.get(output_format, None) if report_to_export: msg = f"Saving {export_type} report at: {export_path}" @@ -276,7 +276,7 @@ def process_report( except Exception as e: raise e - if output is ScanOutput.SCREEN: + if output_format is ScanOutput.SCREEN: console.print() lines = [] @@ -300,11 +300,11 @@ def process_report( for line in lines: console.print(line, emoji=True) - if output.is_silent(): + if output_format.is_silent(): console.quiet = False - if output is ScanOutput.JSON or ScanOutput.is_format(output, ScanOutput.SPDX): - if output is ScanOutput.JSON: + if output_format is ScanOutput.JSON or ScanOutput.is_format(output_format, ScanOutput.SPDX): + if output_format is ScanOutput.JSON: if detailed_output: report_to_output = add_cve_details_to_report( report_to_output, obj.project.files @@ -492,7 +492,7 @@ def validate_save_as( def initialize_file_finder( ctx: "SafetyContext", - target: Path, + target_directory: Path, console: Optional[Console], ecosystems: List[Ecosystem], ) -> FileFinder: @@ -515,7 +515,7 @@ def initialize_file_finder( } file_finder = FileFinder( - target=target, + target=target_directory, ecosystems=ecosystems, max_level=ctx.obj.config.scan.max_depth, exclude=ctx.obj.config.scan.ignore, @@ -588,9 +588,8 @@ def print_file_info(console: Console, path: Path, target: Path) -> None: path (Path): The file path of the current file. target (Path): The base path to which the file path is relative. """ - console.print() msg = f"{ICON_PENCIL} {TAG_FILE_TITLE_START}{path.relative_to(target)}:{TAG_FILE_TITLE_END}" - console.print(msg) + console.print(f"\n{msg}") def sort_and_filter_vulnerabilities( @@ -635,8 +634,8 @@ def count_critical_vulnerabilities(vulnerabilities: List[Vulnerability]) -> int: def generate_vulnerability_message( - spec_name: str, - spec_raw: str, + dependency_name: str, + dependency_raw: str, vulns_found: int, critical_vulns_count: int, vuln_word: str, @@ -645,8 +644,8 @@ def generate_vulnerability_message( Generate a formatted message for vulnerabilities in a specific dependency. Args: - spec_name (str): Name of the dependency. - spec_raw (str): Raw specification string of the dependency. + dependency_name (str): Name of the dependency. + dependency_raw (str): Raw specification string of the dependency. vulns_found (int): Number of vulnerabilities found. critical_vulns_count (int): Number of critical vulnerabilities found. vuln_word (str): Pluralized form of the word "vulnerability." @@ -654,7 +653,7 @@ def generate_vulnerability_message( Returns: str: Formatted vulnerability message. """ - msg = f"{TAG_DEP_NAME_START}{spec_name}{TAG_DEP_NAME_END}{TAG_SPECIFIER_START}{spec_raw.replace(spec_name, '')}{TAG_SPECIFIER_END} [{vulns_found} {vuln_word} found" + msg = f"{TAG_DEP_NAME_START}{dependency_name}{TAG_DEP_NAME_END}{TAG_SPECIFIER_START}{dependency_raw.replace(dependency_name, '')}{TAG_SPECIFIER_END} [{vulns_found} {vuln_word} found" if ( vulns_found > CRITICAL_VULN_THRESHOLD @@ -774,7 +773,7 @@ def process_file_fixes( specs_to_fix: List[Any], options: Dict, policy_limits: List[SecurityUpdates.UpdateLevel], - output: ScanOutput, + outpu_format: ScanOutput, no_output: bool, prompt: bool, ) -> Any: @@ -786,7 +785,7 @@ def process_file_fixes( specs_to_fix (List[Any]): The specifications to fix in the file. options (Dict): Mapping of file types to update limits. policy_limits (List[SecurityUpdates.UpdateLevel]): Policy-defined update limits. - output (ScanOutput): The scan output format. + output_format (ScanOutput): The scan output format. no_output (bool): Whether to suppress output. prompt (bool): Whether to prompt the user for confirmation. @@ -811,7 +810,7 @@ def process_file_fixes( file_to_fix, specs_to_fix, update_limits, - output, + outpu_format, no_output=no_output, prompt=prompt, ) @@ -1096,7 +1095,7 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: obj=ctx.obj, console=console, report=report, - output=output, + output_format=output, save_as=save_as if save_as and all(save_as) else None, detailed_output=detailed_output, filter_keys=filter_keys, From 7941732b1c381e770057ae46ec992d15fdfba085 Mon Sep 17 00:00:00 2001 From: DLS29 Date: Mon, 25 May 2026 22:15:30 +0300 Subject: [PATCH 2/3] refactor: restore preprocess_args for test suite compatibility --- safety/cli.py | 4 ++-- tests/test_cli.py | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/safety/cli.py b/safety/cli.py index 0c0c551c..7be9172c 100644 --- a/safety/cli.py +++ b/safety/cli.py @@ -128,7 +128,7 @@ LOG = logging.getLogger(__name__) -def debug_proccesing(f): +def debug_prossesing(f): if "--debug" in sys.argv: index = sys.argv.index("--debug") if len(sys.argv) > index + 1: @@ -178,7 +178,7 @@ def configure_logger(ctx, param, debug): @click.option("--debug", is_flag=True, help=CLI_DEBUG_HELP, callback=configure_logger) @click.version_option(version=get_version()) @click.pass_context -@debug_proccesing +@debug_prossesing def cli(ctx, debug, disable_optional_telemetry): """ Scan and secure Python projects against package vulnerabilities. To get started navigate to a Python project and run `safety scan`. diff --git a/tests/test_cli.py b/tests/test_cli.py index 03901b49..905b52c1 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -831,9 +831,9 @@ def test_debug_flag(self, *args): def test_debug_flag_with_value_1(self, *args): sys.argv = ["safety", "--debug", "1", "scan"] - from safety.cli import preprocess_args + from safety.cli import debug_prossesing - @preprocess_args + @debug_prossesing def dummy_function(): pass @@ -855,9 +855,9 @@ def dummy_function(): def test_debug_flag_with_value_true(self, *args): sys.argv = ["safety", "--debug", "true", "scan"] - from safety.cli import preprocess_args + from safety.cli import debug_prossesing - @preprocess_args + @debug_prossesing def dummy_function(): pass From 75e236b870eb77d6af8b66bfdc545dffb0ebd918 Mon Sep 17 00:00:00 2001 From: DLS29 Date: Tue, 26 May 2026 17:25:43 +0300 Subject: [PATCH 3/3] Second patch to #575 --- safety/scan/init_scan.py | 6 +++--- safety/scan/main.py | 10 +++++----- safety/scan/validators.py | 8 ++++---- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/safety/scan/init_scan.py b/safety/scan/init_scan.py index b442778e..d36f38a0 100644 --- a/safety/scan/init_scan.py +++ b/safety/scan/init_scan.py @@ -510,7 +510,7 @@ def start_scan( Args: ctx: The Typer context object containing configuration and project information - target: The target directory to scan + targets_directory: The target directory to scan use_server_matching: Whether to use server-side vulnerability matching Returns: @@ -526,7 +526,7 @@ def start_scan( ) scan_type = ScanType(command_name) - targets = [target] + targets_directory = [target] if not scan_type: raise SafetyException("Missing scan_type.") @@ -534,7 +534,7 @@ def start_scan( metadata = MetadataModel( scan_type=scan_type, stage=stage, - scan_locations=targets, + scan_locations=targets_directory, authenticated=is_authenticated, authentication_type=auth_type, telemetry=telemetry, diff --git a/safety/scan/main.py b/safety/scan/main.py index 4faaad10..8f3b3ad1 100644 --- a/safety/scan/main.py +++ b/safety/scan/main.py @@ -141,7 +141,7 @@ def resolve_policy( def save_report_as( - scan_type: ScanType, export_type: ScanExport, at: Path, report: Any + scan_type: ScanType, export_type: ScanExport, save_path: Path, report: Any ) -> None: """ Saves the scan report to the specified location. @@ -149,17 +149,17 @@ def save_report_as( Args: scan_type (ScanType): The type of scan. export_type (ScanExport): The type of export. - at (Path): The path to save the report. + save_path (Path): The path to save the report. report (Any): The report content. """ tag = int(time.time()) - if at.is_dir(): - at = at / Path( + if save_path.is_dir(): + save_path = save_path / Path( f"{scan_type.value}-{export_type.get_default_file_name(tag=tag)}" ) - with open(at, "w+") as report_file: + with open(save_path, "w+") as report_file: report_file.write(report) diff --git a/safety/scan/validators.py b/safety/scan/validators.py index 17ed318a..5c10a304 100644 --- a/safety/scan/validators.py +++ b/safety/scan/validators.py @@ -53,20 +53,20 @@ def save_as_callback( ) -def output_callback(output: ScanOutput) -> str: +def output_callback(output_format: ScanOutput) -> str: """ Callback function to handle output parameter and validate if spdx extension is installed. Args: - output (ScanOutput): The output format. + output_format (ScanOutput): The output format. Returns: str: The validated output format. """ - if ScanOutput.is_format(output, ScanExport.SPDX): + if ScanOutput.is_format(output_format, ScanExport.SPDX): raise_if_not_spdx_extension_installed() - return output.value + return output_format.value def fail_if_not_allowed_stage(ctx: "SafetyContext"):