diff --git a/safety/constants.py b/safety/constants.py index 4afbd57b..065ffe30 100644 --- a/safety/constants.py +++ b/safety/constants.py @@ -161,4 +161,18 @@ def get_config_setting(name: str) -> Optional[str]: EXIT_CODE_MALFORMED_DB = 69 EXIT_CODE_INVALID_PROVIDED_REPORT = 70 EXIT_CODE_INVALID_REQUIREMENT = 71 -EXIT_CODE_EMAIL_NOT_VERIFIED = 72 \ No newline at end of file +EXIT_CODE_EMAIL_NOT_VERIFIED = 72 + +EXIT_CODE_REASON_MAPPING = { + EXIT_CODE_OK: "No vulnerabilities found.", + EXIT_CODE_FAILURE: "General failure.", + EXIT_CODE_VULNERABILITIES_FOUND: "Vulnerabilities found.", + EXIT_CODE_INVALID_API_KEY: "Invalid API key.", + EXIT_CODE_TOO_MANY_REQUESTS: "Too many requests.", + EXIT_CODE_UNABLE_TO_LOAD_LOCAL_VULNERABILITY_DB: "Unable to load local vulnerability database.", + EXIT_CODE_UNABLE_TO_FETCH_VULNERABILITY_DB: "Unable to fetch vulnerability database.", + EXIT_CODE_MALFORMED_DB: "Malformed vulnerability database.", + EXIT_CODE_INVALID_PROVIDED_REPORT: "Invalid provided report.", + EXIT_CODE_INVALID_REQUIREMENT: "Invalid requirement.", + EXIT_CODE_EMAIL_NOT_VERIFIED: "Email not verified." +} \ No newline at end of file diff --git a/safety/scan/command.py b/safety/scan/command.py index 433e8742..bdadc80d 100644 --- a/safety/scan/command.py +++ b/safety/scan/command.py @@ -5,10 +5,11 @@ import logging from pathlib import Path import sys +import json from typing import Any, List, Optional, Set, Tuple from typing_extensions import Annotated -from safety.constants import EXIT_CODE_VULNERABILITIES_FOUND +from safety.constants import EXIT_CODE_VULNERABILITIES_FOUND, EXIT_CODE_REASON_MAPPING from safety.safety import process_fixes, process_fixes_scan from safety.scan.finder.handlers import ECOSYSTEM_HANDLER_MAPPING, FileHandler from safety.scan.validators import output_callback, save_as_callback @@ -48,13 +49,16 @@ class ScannableEcosystems(Enum): PYTHON = Ecosystem.PYTHON.value -def process_report(obj: Any, console: Console, report: ReportModel, output: str, - save_as: Optional[Tuple[str, Path]], **kwargs): +def process_report(obj: Any, console: Console, report: ReportModel, output: str, + save_as: Optional[Tuple[str, Path]], exit_code: int = 0, exit_reason: str = "No vulnerabilities found.", **kwargs): wait_msg = "Processing report" with console.status(wait_msg, spinner=DEFAULT_SPINNER) as status: - json_format = report.as_v30().json() - + report_data = report.as_v30().model_dump() + report_data["exit_code"] = exit_code + report_data["exit_reason"] = exit_reason + json_format = json.dumps(report_data) + export_type, export_path = None, None if save_as: @@ -74,12 +78,12 @@ def process_report(obj: Any, console: Console, report: ReportModel, output: str, spdx_version = None if export_type: spdx_version = export_type.version if export_type.version and ScanExport.is_format(export_type, ScanExport.SPDX) else None - + if not spdx_version and output: spdx_version = output.version if output.version and ScanOutput.is_format(output, 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: html_format = render_scan_html(report, obj) @@ -89,7 +93,7 @@ def process_report(obj: Any, console: Console, report: ReportModel, output: str, ScanExport.SPDX: spdx_format, ScanExport.SPDX_2_3: spdx_format, ScanExport.SPDX_2_2: spdx_format, - } + } output_format_mapping = { ScanOutput.JSON: json_format, @@ -106,7 +110,7 @@ def process_report(obj: Any, console: Console, report: ReportModel, output: str, msg = f"Saving {export_type} report at: {export_path}" status.update(msg) LOG.debug(msg) - save_report_as(report.metadata.scan_type, export_type, Path(export_path), + save_report_as(report.metadata.scan_type, export_type, Path(export_path), report_to_export) report_url = None @@ -131,7 +135,7 @@ def process_report(obj: Any, console: Console, report: ReportModel, output: str, f"[link]{project_url}[/link]") elif report.metadata.scan_type is ScanType.system_scan: lines.append(f"System scan report: [link]{report_url}[/link]") - + for line in lines: console.print(line, emoji=True) @@ -142,14 +146,14 @@ def process_report(obj: Any, console: Console, report: ReportModel, output: str, if output is ScanOutput.JSON: kwargs = {"json": report_to_output} else: - kwargs = {"data": report_to_output} + kwargs = {"data": report_to_output} console.print_json(**kwargs) else: console.print(report_to_output) console.quiet = True - + return report_url @@ -157,10 +161,10 @@ def generate_updates_arguments() -> list: """Generates a list of file types and update limits for apply fixes.""" fixes = [] limit_type = SecurityUpdates.UpdateLevel.PATCH - DEFAULT_FILE_TYPES = [FileType.REQUIREMENTS_TXT, FileType.PIPENV_LOCK, + DEFAULT_FILE_TYPES = [FileType.REQUIREMENTS_TXT, FileType.PIPENV_LOCK, FileType.POETRY_LOCK, FileType.VIRTUAL_ENVIRONMENT] fixes.extend([(default_file_type, limit_type) for default_file_type in DEFAULT_FILE_TYPES]) - + return fixes @@ -197,7 +201,7 @@ def scan(ctx: typer.Context, ] = ScanOutput.SCREEN, detailed_output: Annotated[bool, typer.Option("--detailed-output", - help=SCAN_DETAILED_OUTPUT, + help=SCAN_DETAILED_OUTPUT, show_default=False) ] = False, save_as: Annotated[Optional[Tuple[ScanExport, Path]], @@ -221,7 +225,7 @@ def scan(ctx: typer.Context, )] = None, apply_updates: Annotated[bool, typer.Option("--apply-fixes", - help=SCAN_APPLY_FIXES, + help=SCAN_APPLY_FIXES, show_default=False) ] = False ): @@ -240,9 +244,9 @@ def scan(ctx: typer.Context, ecosystems = [Ecosystem(member.value) for member in list(ScannableEcosystems)] to_include = {file_type: paths for file_type, paths in ctx.obj.config.scan.include_files.items() if file_type.ecosystem in ecosystems} - file_finder = FileFinder(target=target, ecosystems=ecosystems, + file_finder = FileFinder(target=target, ecosystems=ecosystems, max_level=ctx.obj.config.scan.max_depth, - exclude=ctx.obj.config.scan.ignore, + exclude=ctx.obj.config.scan.ignore, include_files=to_include, console=console) @@ -260,7 +264,7 @@ def scan(ctx: typer.Context, with console.status(wait_msg, spinner=DEFAULT_SPINNER): path, file_paths = file_finder.search() - print_detected_ecosystems_section(console, file_paths, + print_detected_ecosystems_section(console, file_paths, include_safety_prjs=True) target_ecosystems = ", ".join([member.value for member in ecosystems]) @@ -274,7 +278,7 @@ def scan(ctx: typer.Context, count = 0 ignored = set() - + affected_count = 0 dependency_vuln_detected = False @@ -288,7 +292,7 @@ def scan(ctx: typer.Context, display_apply_fix_suggestion = False with console.status(wait_msg, spinner=DEFAULT_SPINNER) as status: - for path, analyzed_file in process_files(paths=file_paths, + for path, analyzed_file in process_files(paths=file_paths, config=config): count += len(analyzed_file.dependency_results.dependencies) @@ -298,7 +302,7 @@ def scan(ctx: typer.Context, if detailed_output: vulns_ignored = analyzed_file.dependency_results.ignored_vulns_data \ .values() - ignored_vulns_data = itertools.chain(vulns_ignored, + ignored_vulns_data = itertools.chain(vulns_ignored, ignored_vulns_data) ignored.update(analyzed_file.dependency_results.ignored_vulns.keys()) @@ -309,7 +313,7 @@ def scan(ctx: typer.Context, 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 to_fix_spec = [] @@ -327,10 +331,10 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: for spec in affected_specifications: if file_matched_for_fix: to_fix_spec.append(spec) - + console.print() vulns_to_report = sorted( - [vuln for vuln in spec.vulnerabilities if not vuln.ignored], + [vuln for vuln in spec.vulnerabilities if not vuln.ignored], key=sort_vulns_by_score, reverse=True) @@ -346,14 +350,14 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: console.print(Padding(f"{msg}]", (0, 0, 0, 1)), emoji=True, overflow="crop") - + if detailed_output or vulns_found < 3: for vuln in vulns_to_report: - render_to_console(vuln, console, - rich_kwargs={"emoji": True, + render_to_console(vuln, console, + rich_kwargs={"emoji": True, "overflow": "crop"}, detailed_output=detailed_output) - + lines = [] # Put remediation here @@ -381,16 +385,16 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: console.print(Padding(line, (0, 0, 0, 1)), emoji=True) console.print( - Padding(f"Learn more: [link]{spec.remediation.more_info_url}[/link]", - (0, 0, 0, 1)), emoji=True) + Padding(f"Learn more: [link]{spec.remediation.more_info_url}[/link]", + (0, 0, 0, 1)), emoji=True) else: console.print() console.print(f":white_check_mark: [file_title]{path.relative_to(target)}: No issues found.[/file_title]", emoji=True) if(ctx.obj.auth.stage == Stage.development - and analyzed_file.ecosystem == Ecosystem.PYTHON - and analyzed_file.file_type == FileType.REQUIREMENTS_TXT + and analyzed_file.ecosystem == Ecosystem.PYTHON + and analyzed_file.file_type == FileType.REQUIREMENTS_TXT and any(affected_specifications) and not apply_updates): display_apply_fix_suggestion = True @@ -405,12 +409,12 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: if file_matched_for_fix: to_fix_files.append((file, to_fix_spec)) - files.append(file) + files.append(file) if display_apply_fix_suggestion: console.print() print_fixes_section(console, requirements_txt_found, detailed_output) - + console.print() print_brief(console, ctx.obj.project, count, affected_count, fixes_count) @@ -418,19 +422,21 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: is_detailed_output=detailed_output, ignored_vulns_data=ignored_vulns_data) - + version = ctx.obj.schema metadata = ctx.obj.metadata telemetry = ctx.obj.telemetry ctx.obj.project.files = files report = ReportModel(version=version, - metadata=metadata, + metadata=metadata, telemetry=telemetry, files=[], projects=[ctx.obj.project]) - - report_url = process_report(ctx.obj, console, report, **{**ctx.params}) + + exit_reason = EXIT_CODE_REASON_MAPPING.get(exit_code, "Unknown reason") + + report_url = process_report(ctx.obj, console, report, exit_code=exit_code, exit_reason=exit_reason, **ctx.params) project_url = f"{SAFETY_PLATFORM_URL}{ctx.obj.project.url_path}" if apply_updates: @@ -440,7 +446,7 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: no_output = output is not ScanOutput.SCREEN prompt = output is ScanOutput.SCREEN - + # TODO: rename that 'no_output' confusing name if not no_output: console.print() @@ -462,11 +468,11 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: if any(policy_limits): update_limits = [policy_limit.value for policy_limit in policy_limits] - - fixes = process_fixes_scan(file_to_fix, + + process_fixes_scan(file_to_fix, specs_to_fix, update_limits, output, no_output=no_output, prompt=prompt) - + if not no_output: console.print("-" * console.size.width) @@ -484,7 +490,7 @@ def sort_vulns_by_score(vuln: Vulnerability) -> int: @scan_system_app.command( cls=SafetyCLICommand, help=CLI_SYSTEM_SCAN_COMMAND_HELP, - options_metavar="[COMMAND-OPTIONS]", + options_metavar="[COMMAND-OPTIONS]", name=CMD_SYSTEM_NAME, epilog=DEFAULT_EPILOG) @handle_cmd_exception @inject_metadata @@ -521,7 +527,7 @@ def system_scan(ctx: typer.Context, typer.Option( help=SYSTEM_SCAN_OUTPUT_HELP, show_default=False) - ] = SystemScanOutput.SCREEN, + ] = SystemScanOutput.SCREEN, save_as: Annotated[Optional[Tuple[SystemScanExport, Path]], typer.Option( help=SYSTEM_SCAN_SAVE_AS_HELP, @@ -575,9 +581,9 @@ def system_scan(ctx: typer.Context, for file_type, paths in target_paths.items(): current = file_paths.get(file_type, set()) current.update(paths) - file_paths[file_type] = current + file_paths[file_type] = current - scan_project_command = get_command_for(name=CMD_PROJECT_NAME, + scan_project_command = get_command_for(name=CMD_PROJECT_NAME, typer_instance=scan_project_app) projects_dirs = set() @@ -587,12 +593,12 @@ def system_scan(ctx: typer.Context, with console.status(":mag:", spinner=DEFAULT_SPINNER) as status: # Handle projects first if FileType.SAFETY_PROJECT.value in file_paths.keys(): - projects_file_paths = file_paths[FileType.SAFETY_PROJECT.value] + projects_file_paths = file_paths[FileType.SAFETY_PROJECT.value] basic_params = ctx.params.copy() basic_params.pop("targets", None) prjs_console = Console(quiet=True) - + for project_path in projects_file_paths: projects_dirs.add(project_path.parent) project_dir = str(project_path.parent) @@ -607,7 +613,7 @@ def system_scan(ctx: typer.Context, if not project or not project.id: LOG.warn(f"{project_path} parsed but project id is not defined or valid.") continue - + if not ctx.obj.platform_enabled: msg = f"project found and skipped, navigate to `{project.project_path}` and scan this project with ‘safety scan’" console.print(f"{project.id}: {msg}") @@ -615,8 +621,8 @@ def system_scan(ctx: typer.Context, msg = f"Existing project found at {project_dir}" console.print(f"{project.id}: {msg}") - project_data[project.id] = {"path": project_dir, - "report_url": None, + project_data[project.id] = {"path": project_dir, + "report_url": None, "project_url": None, "failed_exception": None} @@ -642,7 +648,7 @@ def system_scan(ctx: typer.Context, "save_as": (None, None), "upload_request_id": upload_request_id, "local_policy": local_policy_file, "console": prjs_console} try: - # TODO: Refactor to avoid calling invoke, also, launch + # TODO: Refactor to avoid calling invoke, also, launch # this on background. console.print( Padding(f"Running safety scan for {project.id} project", @@ -660,7 +666,7 @@ def system_scan(ctx: typer.Context, (0, 0, 0, 1)), emoji=True) LOG.exception(f"Failed to run scan on project {project.id}, " \ f"Upload request ID: {upload_request_id}. Reason {e}") - + console.print() file_paths.pop(FileType.SAFETY_PROJECT.value, None) @@ -670,18 +676,18 @@ def system_scan(ctx: typer.Context, status.update(":mag: Finishing projects processing.") for k, f_paths in file_paths.items(): - file_paths[k] = {fp for fp in f_paths - if not should_exclude(excludes=projects_dirs, + file_paths[k] = {fp for fp in f_paths + if not should_exclude(excludes=projects_dirs, to_analyze=fp)} - + pkgs_count = 0 file_count = 0 venv_count = 0 for path, analyzed_file in process_files(paths=file_paths, config=config): status.update(f":mag: {path}") - files.append(FileModel(location=path, - file_type=analyzed_file.file_type, + files.append(FileModel(location=path, + file_type=analyzed_file.file_type, results=analyzed_file.dependency_results)) file_pkg_count = len(analyzed_file.dependency_results.dependencies) @@ -718,7 +724,7 @@ def system_scan(ctx: typer.Context, pkgs_count += file_pkg_count console.print(f":package: {file_pkg_count} {msg} in {path}", emoji=True) - + if affected_pkgs_count <= 0: msg = "No vulnerabilities found" else: @@ -738,7 +744,7 @@ def system_scan(ctx: typer.Context, telemetry=telemetry, files=files, projects=projects) - + console.print() total_count = sum([finder.file_count for finder in file_finders], 0) console.print(f"Searched {total_count:,} files for dependency security issues") @@ -749,16 +755,16 @@ def system_scan(ctx: typer.Context, console.print() proccessed = dict(filter( - lambda item: item[1]["report_url"] and item[1]["project_url"], + lambda item: item[1]["report_url"] and item[1]["project_url"], project_data.items())) - + if proccessed: run_word = "runs" if len(proccessed) == 1 else "run" console.print(f"Project {pluralize('scan', len(proccessed))} {run_word} on {len(proccessed)} existing {pluralize('project', len(proccessed))}:") for prj, data in proccessed.items(): console.print(f"[bold]{prj}[/bold] at {data['path']}") - for detail in [f"{prj} dashboard: {data['project_url']}"]: + for detail in [f"{prj} dashboard: {data['project_url']}"]: console.print(Padding(detail, (0, 0, 0, 1)), emoji=True, overflow="crop") - process_report(ctx.obj, console, report, **{**ctx.params}) + process_report(ctx.obj, console, report, **ctx.params)