diff --git a/safety/cli.py b/safety/cli.py index 14ec46ae..750bae31 100644 --- a/safety/cli.py +++ b/safety/cli.py @@ -1224,7 +1224,45 @@ def print_check_updates_header(console): class Output(str, Enum): SCREEN = "screen" JSON = "json" +# new extracted functions to help refactoring (making the code more readable) +def _handle_auth_error_output(console) -> None: + """Maneja la salida en consola cuando el usuario no está autenticado.""" + if console.quiet: + console.quiet = False + response = { + "status": 401, + "message": "Authenticated failed, please authenticate Safety and try again", + "data": {}, + } + console.print_json(json.dumps(response)) + else: + console.print() + console.print("[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`") + +def _evaluate_and_print_versions(console, current_version: str, latest_available_version: str) -> None: + """Compara las versiones y renderiza el mensaje adecuado para el usuario.""" + try: + parsed_latest = packaging_version.parse(latest_available_version) + parsed_current = packaging_version.parse(current_version) + if parsed_latest > parsed_current: + console.print(f"Update available: Safety version {latest_available_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.") + console.print() + console.print(f"Pip: To install the updated version of Safety directly via pip, run: pip install safety=={latest_available_version}") + elif parsed_latest < parsed_current: + 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}") + else: + console.print("You are already using the latest stable version of Safety.") + + 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}") + console.print("Please report this issue or try again later.") +# end of new functions por refactoring @cli_app.command( cls=SafetyCLICommand, @@ -1258,8 +1296,6 @@ def check_updates( print_check_updates_header(console) - wait_msg = "Authenticating and checking for Safety CLI updates" - VERSION = get_version() PYTHON_VERSION = platform.python_version() OS_TYPE = platform.system() @@ -1268,7 +1304,7 @@ def check_updates( data = None console.print() - with console.status(wait_msg, spinner=DEFAULT_SPINNER): + with console.status("Authenticating and checking for Safety CLI updates", spinner=DEFAULT_SPINNER): try: data = ctx.obj.auth.platform.check_updates( version=1, @@ -1283,25 +1319,9 @@ def check_updates( except Exception as e: LOG.exception(f"Failed to check updates, reason: {e}") raise e - +#here if not authenticated: - if console.quiet: - console.quiet = False - response = { - "status": 401, - "message": "Authenticated failed, please authenticate Safety and try again", - "data": {}, - } - console.print_json(json.dumps(response)) - else: - console.print() - console.print( - "[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`" - ) + _handle_auth_error_output(console) sys.exit(1) if not data: @@ -1313,15 +1333,13 @@ def check_updates( organization = data.get("organization", "-") account = data.get("user_email", "-") - current_version = ( - f"Current version: {VERSION} (Python {PYTHON_VERSION} on {OS_TYPE})" - ) latest_available_version = data.get("safety_updates", {}).get("stable_version", "-") - +#here + current_version_str= f"Current version: {VERSION} (Phyton {PYTHON_VERSION} on {OS_TYPE})" details = [ f"Organization: {organization}", f"Account: {account}", - current_version, + current_version_str, f"Latest stable available version: {latest_available_version}", ] @@ -1329,42 +1347,10 @@ def check_updates( console.print(Padding(msg, (0, 0, 0, 1)), emoji=True) console.print() - - if latest_available_version: - try: - # Compare the current version and the latest available version using packaging.version - if packaging_version.parse( - latest_available_version - ) > packaging_version.parse(VERSION): - console.print( - f"Update available: Safety version {latest_available_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." - ) - console.print() - console.print( - f"Pip: To install the updated version of Safety directly via pip, run: pip install safety=={latest_available_version}" - ) - elif packaging_version.parse( - latest_available_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}" - ) - else: - console.print( - "You are already using the latest stable version of Safety." - ) - 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}" - ) - console.print("Please report this issue or try again later.") - + #here + if latest_available_version and latest_available_version != "-": + _evaluate_and_print_versions(console, VERSION, latest_available_version) + if console.quiet: console.quiet = False response = {"status": 200, "message": "", "data": data} diff --git a/safety/cli_util.py b/safety/cli_util.py index 9ac8cbfb..97e75c96 100644 --- a/safety/cli_util.py +++ b/safety/cli_util.py @@ -193,6 +193,103 @@ def inner(ctx, *args, **kwargs): return inner +# functions for refactoring pretty_help_format +def _categorize_params(obj, ctx): + """Clasifica los parámetros en argumentos y opciones según sus paneles.""" + from typer.rich_utils import ARGUMENTS_PANEL_TITLE, OPTIONS_PANEL_TITLE + import click + from collections import defaultdict + + panel_to_arguments = defaultdict(list) + panel_to_options = defaultdict(list) + + for param in obj.get_params(ctx): + if getattr(param, "hidden", False): + continue + + if isinstance(param, click.Argument): + panel_name = getattr(param, "rich_help_panel", None) or ARGUMENTS_PANEL_TITLE + panel_to_arguments[panel_name].append(param) + elif isinstance(param, click.Option): + panel_name = getattr(param, "rich_help_panel", None) or OPTIONS_PANEL_TITLE + panel_to_options[panel_name].append(param) + + return panel_to_arguments, panel_to_options + +def _categorize_commands(obj, ctx): + """Agrupa los subcomandos disponibles por nombre de panel.""" + from typer.rich_utils import COMMANDS_PANEL_TITLE + import click + from collections import defaultdict + + panel_to_commands = defaultdict(list) + if isinstance(obj, click.MultiCommand): + for command_name in obj.list_commands(ctx): + command = obj.get_command(ctx, command_name) + if command and not command.hidden: + panel_name = getattr(command, "rich_help_panel", None) or COMMANDS_PANEL_TITLE + panel_to_commands[panel_name].append(command) + + return panel_to_commands + +def _print_commands_panels(panel_to_commands, console): + """Imprime los paneles de comandos, dando prioridad al panel por defecto.""" + from typer.rich_utils import COMMANDS_PANEL_TITLE + if not panel_to_commands: + return + + if COMMANDS_PANEL_TITLE in panel_to_commands: + custom_print_commands_panel( + name=COMMANDS_PANEL_TITLE, + commands=panel_to_commands[COMMANDS_PANEL_TITLE], + console=console, + ) + + for panel_name, commands in panel_to_commands.items(): + if panel_name != COMMANDS_PANEL_TITLE: + custom_print_commands_panel(name=panel_name, commands=commands, console=console) + +def _print_params_panels(panel_dict, default_title, ctx, console): + """Imprime genéricamente paneles de parámetros (Opciones o Argumentos).""" + if not panel_dict: + return + + if default_title in panel_dict: + custom_print_options_panel( + name=default_title, + params=panel_dict[default_title], + ctx=ctx, + console=console + ) + + for panel_name, params in panel_dict.items(): + if panel_name != default_title: + custom_print_options_panel(name=panel_name, params=params, ctx=ctx, console=console) + +def _print_global_options(ctx, console): + """Extrae e imprime las opciones globales heredadas del comando padre.""" + import click + if ctx.parent: + params = [param for param in ctx.parent.command.params if isinstance(param, click.Option)] + if params: + custom_print_options_panel( + name="Global-Options", + params=params, + ctx=ctx.parent, + console=console, + ) + +def _print_epilog(obj, console): + """Formatea e imprime el epílogo del comando.""" + from rich.align import Align + from rich.padding import Padding + + if obj.epilog: + lines = obj.epilog.split("\n\n") + epilogue = "\n".join([x.replace("\n", " ").strip() for x in lines]) + epilogue_text = custom_make_rich_text(text=epilogue) + console.print(Padding(Align(epilogue_text, pad=False), 1)) + def pretty_format_help( obj: Union[click.Command, click.Group], ctx: click.Context, markup_mode: MarkupMode @@ -224,119 +321,25 @@ def pretty_format_help( # Print command / group help if we have some if obj.help: console.print() - - # Print with some padding - console.print( + console.print( # Print with some padding Padding(Align(custom_get_help_text(obj=obj), pad=False), (0, 1, 0, 1)) ) - # Print usage console.print( Padding(highlighter(obj.get_usage(ctx)), 1), style=STYLE_USAGE_COMMAND ) - - if isinstance(obj, click.MultiCommand): - panel_to_commands: DefaultDict[str, List[click.Command]] = defaultdict(list) - for command_name in obj.list_commands(ctx): - command = obj.get_command(ctx, command_name) - if command and not command.hidden: - panel_name = ( - getattr(command, "rich_help_panel", None) - or COMMANDS_PANEL_TITLE - ) - panel_to_commands[panel_name].append(command) - - # Print each command group panel - default_commands = panel_to_commands.get(COMMANDS_PANEL_TITLE, []) - custom_print_commands_panel( - name=COMMANDS_PANEL_TITLE, - commands=default_commands, - console=console, - ) - for panel_name, commands in panel_to_commands.items(): - if panel_name == COMMANDS_PANEL_TITLE: - # Already printed above - continue - custom_print_commands_panel( - name=panel_name, - commands=commands, - console=console, - ) - - panel_to_arguments: DefaultDict[str, List[click.Argument]] = defaultdict(list) - panel_to_options: DefaultDict[str, List[click.Option]] = defaultdict(list) - for param in obj.get_params(ctx): - # Skip if option is hidden - if getattr(param, "hidden", False): - continue - if isinstance(param, click.Argument): - panel_name = ( - getattr(param, "rich_help_panel", None) or ARGUMENTS_PANEL_TITLE - ) - panel_to_arguments[panel_name].append(param) - elif isinstance(param, click.Option): - panel_name = ( - getattr(param, "rich_help_panel", None) or OPTIONS_PANEL_TITLE - ) - panel_to_options[panel_name].append(param) - - default_options = panel_to_options.get(OPTIONS_PANEL_TITLE, []) - custom_print_options_panel( - name=OPTIONS_PANEL_TITLE, - params=default_options, - ctx=ctx, - console=console, - ) - for panel_name, options in panel_to_options.items(): - if panel_name == OPTIONS_PANEL_TITLE: - # Already printed above - continue - custom_print_options_panel( - name=panel_name, - params=options, - ctx=ctx, - console=console, - ) - - default_arguments = panel_to_arguments.get(ARGUMENTS_PANEL_TITLE, []) - custom_print_options_panel( - name=ARGUMENTS_PANEL_TITLE, - params=default_arguments, - ctx=ctx, - console=console, - ) - for panel_name, arguments in panel_to_arguments.items(): - if panel_name == ARGUMENTS_PANEL_TITLE: - # Already printed above - continue - custom_print_options_panel( - name=panel_name, - params=arguments, - ctx=ctx, - console=console, - ) - - if ctx.parent: - params = [] - for param in ctx.parent.command.params: - if isinstance(param, click.Option): - params.append(param) - - custom_print_options_panel( - name="Global-Options", - params=params, - ctx=ctx.parent, - console=console, - ) - - # Epilogue if we have it - if obj.epilog: - # Remove single linebreaks, replace double with single - lines = obj.epilog.split("\n\n") - epilogue = "\n".join([x.replace("\n", " ").strip() for x in lines]) - epilogue_text = custom_make_rich_text(text=epilogue) - console.print(Padding(Align(epilogue_text, pad=False), 1)) - + # Command panel + panel_to_commands = _categorize_commands(obj, ctx) + _print_commands_panels(panel_to_commands, console) + # Args and options panels + panel_to_arguments, panel_to_options = _categorize_params(obj, ctx) + _print_params_panels(panel_to_options, OPTIONS_PANEL_TITLE, ctx, console) + _print_params_panels(panel_to_arguments, ARGUMENTS_PANEL_TITLE, ctx, console) + # Global options + _print_global_options(ctx, console) + # Epilogue + _print_epilog(obj, console) + def print_main_command_panels( *, diff --git a/safety/safety.py b/safety/safety.py index a86c1679..3dcca67d 100644 --- a/safety/safety.py +++ b/safety/safety.py @@ -682,8 +682,57 @@ def is_vulnerable( prereleases=True, ) ) +# Refactor the check function to extract smaller functions and improve readability and maintainability. +def _ensure_full_db(db_full, auth, db_mirror, cached, telemetry): + """Lazy loads the full database if it does not exist yet.""" + if not db_full: + return fetch_database( + auth=auth, + full=True, + db=db_mirror, + cached=cached, + telemetry=telemetry, + ) + return db_full + +def _extract_pyup_vuln_id(data: Dict[str, Any]) -> str: + """Isolates complex logic about vulnerabilty ID extraction.""" + try: + return str( + next( + filter( + lambda i: i.get("type", None) == "pyup", + data.get("ids", []), + ) + ).get("id", "") + ) + except StopIteration: + return "" + +def _process_single_vulnerability( + data, vuln_id, specifier, db_full, name, pkg, req, ignore_vulns, ignore_severity_rules, include_ignored, is_env_scan +): + """Builds and aplies rules to determinate if a vulnerability should be reported, ignored or discarded.""" + cve = get_cve_from(data, db_full) + + ignore_vuln_if_needed(pkg, vuln_id, cve, ignore_vulns, ignore_severity_rules, req) + + vulnerability = get_vulnerability_from( + vuln_id, cve, data, specifier, db_full, name, pkg, ignore_vulns, req, + ) + + should_add_vuln = not (vulnerability.is_transitive and is_env_scan) + + # safe manage in case ignore_vulne is None + safe_ignore_vulns = ignore_vulns or {} + is_not_ignored = include_ignored or vulnerability.vulnerability_id not in safe_ignore_vulns + if is_not_ignored and should_add_vuln: + return vulnerability + + return None +#refactor of the check function per se @sync_safety_context def check( *, @@ -702,23 +751,6 @@ def check( ) -> tuple: """ Performs a vulnerability check on the provided packages. - - Args: - auth (Auth): The authentication object. - packages (List[Package]): The list of packages to check. - db_mirror (Union[Optional[str], bool]): The database mirror. - cached (int): The cache validity in seconds. - ignore_vulns (Optional[Dict[str, Any]]): The ignored vulnerabilities. - ignore_severity_rules (Optional[Dict[str, Any]]): The severity rules for ignoring vulnerabilities. - proxy (Optional[Dict[str, Any]]): The proxy settings. - include_ignored (bool): Whether to include ignored vulnerabilities. - is_env_scan (bool): Whether it is an environment scan. - telemetry (bool): Whether to include telemetry data. - params (Optional[Dict[str, Any]]): Additional parameters. - project (Optional[str]): The project name. - - Returns: - tuple: A tuple containing the list of vulnerabilities and the full database. """ SafetyContext().command = "check" db = fetch_database(auth=auth, db=db_mirror, cached=cached, telemetry=telemetry) @@ -736,84 +768,54 @@ def check( for req in requirements: vuln_per_req = {} name = canonicalize_name(req.name) - pkg = found_pkgs.get(name, None) + if not pkg: + continue + if not pkg.version: - if not db_full: - db_full = fetch_database( - auth=auth, - full=True, - db=db_mirror, - cached=cached, - telemetry=telemetry, - ) + db_full = _ensure_full_db(db_full, auth, db_mirror, cached, telemetry) pkg.refresh_from(db_full) - if name in vulnerable_packages: - # we have a candidate here, build the spec set - for specifier in db["vulnerable_packages"][name]: - spec_set = SpecifierSet(specifiers=specifier) - - if is_vulnerable(spec_set, req, pkg): - if not db_full: - db_full = fetch_database( - auth=auth, - full=True, - db=db_mirror, - cached=cached, - telemetry=telemetry, - ) - if not pkg.latest_version: - pkg.refresh_from(db_full) - - for data in get_vulnerabilities( - pkg=name, spec=specifier, db=db_full - ): - try: - vuln_id: str = str( - next( - filter( - lambda i: i.get("type", None) == "pyup", - data.get("ids", []), - ) - ).get("id", "") - ) - except StopIteration: - vuln_id: str = "" - - if vuln_id in vuln_per_req: - vuln_per_req[vuln_id].vulnerable_spec.add(specifier) - continue - - cve = get_cve_from(data, db_full) - - ignore_vuln_if_needed( - pkg, vuln_id, cve, ignore_vulns, ignore_severity_rules, req - ) + if name not in vulnerable_packages: + continue - vulnerability = get_vulnerability_from( - vuln_id, - cve, - data, - specifier, - db_full, - name, - pkg, - ignore_vulns, - req, - ) + # we have a candidate here, build the spec set + for specifier in db["vulnerable_packages"][name]: + spec_set = SpecifierSet(specifiers=specifier) - should_add_vuln = not ( - vulnerability.is_transitive and is_env_scan - ) + if not is_vulnerable(spec_set, req, pkg): + continue + + db_full = _ensure_full_db(db_full, auth, db_mirror, cached, telemetry) + + if not pkg.latest_version: + pkg.refresh_from(db_full) + + for data in get_vulnerabilities(pkg=name, spec=specifier, db=db_full): + vuln_id = _extract_pyup_vuln_id(data) + + if vuln_id in vuln_per_req: + vuln_per_req[vuln_id].vulnerable_spec.add(specifier) + continue + + vulnerability = _process_single_vulnerability( + data=data, + vuln_id=vuln_id, + specifier=specifier, + db_full=db_full, + name=name, + pkg=pkg, + req=req, + ignore_vulns=ignore_vulns, + ignore_severity_rules=ignore_severity_rules, + include_ignored=include_ignored, + is_env_scan=is_env_scan + ) - if ( - include_ignored - or vulnerability.vulnerability_id not in ignore_vulns - ) and should_add_vuln: - vuln_per_req[vulnerability.vulnerability_id] = vulnerability - vulnerabilities.append(vulnerability) + if vulnerability: + vuln_per_req[vulnerability.vulnerability_id] = vulnerability + vulnerabilities.append(vulnerability) return vulnerabilities, db_full