feat: added safety firewall

.devcontainer/devcontainer.json

@@ -13,7 +13,8 @@
 
   "mounts": [
     "source=/var/run/docker.sock,target=/var/run/docker.sock,type=bind",
-    "source=${localEnv:HOME}${localEnv:USERPROFILE}/.ssh,target=/home/developer/.ssh,type=bind,consistency=cached"
+    "source=${localEnv:HOME}${localEnv:USERPROFILE}/.ssh,target=/home/developer/.ssh,type=bind,consistency=cached",
+    "source=${localEnv:HOME}/.safety,target=/home/developer/.safety,type=bind,consistency=cached"
   ],
 
   "remoteEnv": {

.vscode/launch.json

@@ -10,7 +10,8 @@
             // This uses the default environment which is a virtual environment
             // created by Hatch
             "python": "${workspaceFolder}/.hatch/bin/python",
-            "console": "integratedTerminal"
+            "console": "integratedTerminal",
+            "justMyCode": false,
         }
     ],
     "inputs": [
@@ -73,6 +74,13 @@
                 "--debug scan",
                 "--disable-optional-telemetry scan",
                 "scan --output json --output-file json",
+
+                // Firewall commands
+                "init --help",
+                "init local_prj", // Directory has to be created manually
+                "pip list",
+                "pip install insecure-package",
+                "pip install fastapi",
 
                 // Check commands
                 "check",

docs/.ipynb_checkpoints/Safety-CLI-Quickstart-checkpoint.ipynb

@@ -1,5 +1,12 @@
 {
- "cells": [],
+ "cells": [
+  {
+   "metadata": {},
+   "cell_type": "raw",
+   "source": "",
+   "id": "e4a30302820cf149"
+  }
+ ],
  "metadata": {},
  "nbformat": 4,
  "nbformat_minor": 5

pyproject.toml

@@ -49,6 +49,7 @@ dependencies = [
   "setuptools>=65.5.1",
   "typer>=0.12.1",
   "typing-extensions>=4.7.1",
+  "python-levenshtein>=0.25.1",
 ]
 license = "MIT"
 license-files = ["LICENSES/*"]

safety/alerts/__init__.py

@@ -17,6 +17,10 @@ def get_safety_cli_legacy_group():
     from safety.cli_util import SafetyCLILegacyGroup
     return SafetyCLILegacyGroup
 
+def get_context_settings():
+    from safety.cli_util import CommandType
+    return {"command_type": CommandType.UTILITY}
+
 @dataclass
 class Alert:
     """
@@ -33,7 +37,8 @@ class Alert:
     policy: Any = None
     requirements_files: Any = None
 
-@click.group(cls=get_safety_cli_legacy_group(), help=CLI_ALERT_COMMAND_HELP, deprecated=True, utility_command=True)
+@click.group(cls=get_safety_cli_legacy_group(), help=CLI_ALERT_COMMAND_HELP,
+             deprecated=True, context_settings=get_context_settings())
 @click.option('--check-report', help='JSON output of Safety Check to work with.', type=click.File('r'), default=sys.stdin, required=True)
 @click.option("--key", envvar="SAFETY_API_KEY",
               help="API Key for safetycli.com's vulnerability database. Can be set as SAFETY_API_KEY "

safety/auth/cli.py

@@ -30,7 +30,7 @@
 
 LOG = logging.getLogger(__name__)
 
-auth_app = Typer(rich_markup_mode="rich")
+auth_app = Typer(rich_markup_mode="rich", name="auth")
 
 
 

safety/cli_util.py

@@ -1,4 +1,5 @@
 from collections import defaultdict
+from enum import Enum
 import logging
 import subprocess
 import sys
@@ -23,6 +24,11 @@
 
 LOG = logging.getLogger(__name__)
 
+class CommandType(Enum):
+    MAIN = "main"
+    UTILITY = "utility" 
+    BETA = "beta"
+
 def custom_print_options_panel(name: str, params: List[Any], ctx: Any, console: Console) -> None:
     """
     Print a panel with options.
@@ -284,6 +290,7 @@ def pretty_format_help(obj: Union[click.Command, click.Group],
 
 def print_main_command_panels(*,
     name: str,
+    commands_type: CommandType,
     commands: List[click.Command],
     markup_mode: MarkupMode,
     console) -> None:
@@ -330,6 +337,17 @@ def print_main_command_panels(*,
     if console.size and console.size[0] > 80:
         console_width = console.size[0]
 
+    from rich.console import Group
+
+    description = None
+
+    if commands_type is CommandType.BETA:
+        description = Group(
+            Text(""),
+            Text("These commands are experimental and part of our commitment to delivering innovative features. As we refine functionality, they may be significantly altered or, in rare cases, removed without prior notice. We welcome your feedback and encourage cautious use."),
+            Text("")
+        )
+
     commands_table.add_column(style="bold cyan", no_wrap=True, width=column_width, max_width=column_width)
     commands_table.add_column(width=console_width - column_width)
 
@@ -338,7 +356,7 @@ def print_main_command_panels(*,
     for command in commands:
         helptext = command.short_help or command.help or ""
         command_name = command.name or ""
-        command_name_text = Text(command_name)
+        command_name_text = Text(command_name, style="") if commands_type is CommandType.BETA else Text(command_name)
         rows.append(
             [
                 command_name_text,
@@ -351,9 +369,10 @@ def print_main_command_panels(*,
     for row in rows:
         commands_table.add_row(*row)
     if commands_table.row_count:
+        renderables = [description, commands_table] if description is not None else [Text(""), commands_table]
+
         console.print(
-            Panel(
-                commands_table,
+            Panel(Group(*renderables),
                 border_style=STYLE_COMMANDS_PANEL_BORDER,
                 title=name,
                 title_align=ALIGN_COMMANDS_PANEL,
@@ -404,31 +423,32 @@ def format_main_help(obj: Union[click.Command, click.Group],
         )
 
         if isinstance(obj, click.MultiCommand):
-            UTILITY_COMMANDS_PANEL_TITLE = "Commands cont."
 
-            panel_to_commands: DefaultDict[str, List[click.Command]] = defaultdict(list)
+            UTILITY_COMMANDS_PANEL_TITLE = "Utility commands"
+            BETA_COMMANDS_PANEL_TITLE = "Beta Commands :rocket:"    
+
+            COMMANDS_PANEL_TITLE_CONSTANTS = {
+                CommandType.MAIN: COMMANDS_PANEL_TITLE,
+                CommandType.UTILITY: UTILITY_COMMANDS_PANEL_TITLE,
+                CommandType.BETA: BETA_COMMANDS_PANEL_TITLE
+            }
+
+            panel_to_commands: Dict[CommandType, List[click.Command]] = {}
+
+            # Keep order of panels
+            for command_type in COMMANDS_PANEL_TITLE_CONSTANTS.keys():
+                panel_to_commands[command_type] = []
+
             for command_name in obj.list_commands(ctx):
                 command = obj.get_command(ctx, command_name)
                 if command and not command.hidden:
-                    panel_name = (
-                        UTILITY_COMMANDS_PANEL_TITLE if command.utility_command else COMMANDS_PANEL_TITLE
-                    )
-                    panel_to_commands[panel_name].append(command)
+                    command_type = command.context_settings.get("command_type", CommandType.MAIN)
+                    panel_to_commands[command_type].append(command)
 
-            # Print each command group panel
-            default_commands = panel_to_commands.get(COMMANDS_PANEL_TITLE, [])
-            print_main_command_panels(
-                name=COMMANDS_PANEL_TITLE,
-                commands=default_commands,
-                markup_mode=markup_mode,
-                console=console,
-            )
-            for panel_name, commands in panel_to_commands.items():
-                if panel_name == COMMANDS_PANEL_TITLE:
-                    # Already printed above
-                    continue
+            for command_type, commands in panel_to_commands.items():
                 print_main_command_panels(
-                    name=panel_name,
+                    name=COMMANDS_PANEL_TITLE_CONSTANTS[command_type],
+                    commands_type=command_type,
                     commands=commands,
                     markup_mode=markup_mode,
                     console=console,
@@ -566,22 +586,8 @@ def process_auth_status_not_ready(console, auth: Auth, ctx: typer.Context) -> No
         console.print(MSG_NON_AUTHENTICATED)
         sys.exit(1)
 
-class UtilityCommandMixin:
-    """
-    Mixin to add utility command functionality.
-    """
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        """
-        Initialize the UtilityCommandMixin.
 
-        Args:
-            *args: Variable length argument list.
-            **kwargs: Arbitrary keyword arguments.
-        """
-        self.utility_command = kwargs.pop('utility_command', False)
-        super().__init__(*args, **kwargs)
-
-class SafetyCLISubGroup(UtilityCommandMixin, TyperGroup):
+class SafetyCLISubGroup(TyperGroup):
     """
     Custom TyperGroup with additional functionality for Safety CLI.
     """
@@ -629,7 +635,7 @@ def command(
         """
         super().command(*args, **kwargs)
 
-class SafetyCLICommand(UtilityCommandMixin, TyperCommand):
+class SafetyCLICommand(TyperCommand):
     """
     Custom TyperCommand with additional functionality for Safety CLI.
     """
@@ -660,22 +666,7 @@ def format_usage(self, ctx: click.Context, formatter: click.HelpFormatter) -> No
         formatter.write_usage(command_path, " ".join(pieces))
 
 
-class SafetyCLIUtilityCommand(TyperCommand):
-    """
-    Custom TyperCommand designated as a utility command.
-    """
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        """
-        Initialize the SafetyCLIUtilityCommand.
-
-        Args:
-            *args: Variable length argument list.
-            **kwargs: Arbitrary keyword arguments.
-        """
-        self.utility_command = True
-        super().__init__(*args, **kwargs)
-
-class SafetyCLILegacyGroup(UtilityCommandMixin, click.Group):
+class SafetyCLILegacyGroup(click.Group):
     """
     Custom Click Group to handle legacy command-line arguments.
     """
@@ -749,7 +740,7 @@ def format_help(self, ctx: click.Context, formatter: click.HelpFormatter) -> Non
         else:
             pretty_format_help(self, ctx, markup_mode="rich")
 
-class SafetyCLILegacyCommand(UtilityCommandMixin, click.Command):
+class SafetyCLILegacyCommand(click.Command):
     """
     Custom Click Command to handle legacy command-line arguments.
     """

safety/constants.py

@@ -55,6 +55,8 @@ def get_user_dir() -> Path:
 CACHE_FILE_DIR = USER_CONFIG_DIR / f"{JSON_SCHEMA_VERSION.replace('.', '')}"
 DB_CACHE_FILE = CACHE_FILE_DIR / "cache.json"
 
+PIP_LOCK = USER_CONFIG_DIR / "pip.lock"
+
 CONFIG_FILE_NAME = "config.ini"
 CONFIG_FILE_SYSTEM = SYSTEM_CONFIG_DIR / CONFIG_FILE_NAME if SYSTEM_CONFIG_DIR else None
 CONFIG_FILE_USER = USER_CONFIG_DIR / CONFIG_FILE_NAME

safety/init/command.py

@@ -0,0 +1,97 @@
+from pathlib import Path
+
+from rich.prompt import Prompt
+from ..cli_util import CommandType, SafetyCLICommand, SafetyCLISubGroup
+import typer
+import os
+
+from safety.scan.decorators import initialize_scan
+from safety.init.constants import PROJECT_INIT_CMD_NAME, PROJECT_INIT_HELP, PROJECT_INIT_DIRECTORY_HELP
+from safety.init.main import create_project
+from safety.console import main_console as console
+from ..scan.command import scan
+from ..scan.models import ScanOutput
+from ..tool.main import configure_system, configure_local_directory, has_local_tool_files, configure_alias
+
+try:
+    from typing import Annotated
+except ImportError:
+    from typing_extensions import Annotated
+
+init_app = typer.Typer(rich_markup_mode= "rich", cls=SafetyCLISubGroup)
+
+@init_app.command(
+        cls=SafetyCLICommand,
+        help=PROJECT_INIT_HELP,
+        name=PROJECT_INIT_CMD_NAME,
+        options_metavar="[OPTIONS]",
+        context_settings={
+            "allow_extra_args": True,
+            "ignore_unknown_options": True,
+            "command_type": CommandType.BETA
+        },
+)
+def init(ctx: typer.Context,
+         directory: Annotated[
+             Path,
+             typer.Argument(
+                 exists=True,
+                 file_okay=False,
+                 dir_okay=True,
+                 writable=False,
+                 readable=True,
+                 resolve_path=True,
+                 show_default=False,
+                 help=PROJECT_INIT_DIRECTORY_HELP
+             ),
+         ] = Path(".")):
+
+    do_init(ctx, directory, False)
+
+
+def do_init(ctx: typer.Context, directory: Path, prompt_user: bool = True):
+    project_dir = directory if os.path.isabs(directory) else os.path.join(os.getcwd(), directory)
+    initialize_scan(ctx, console)
+    create_project(ctx, console, Path(project_dir))
+
+    answer = 'y' if not prompt_user else None
+    if prompt_user:
+        console.print(
+            "Safety prevents vulnerable or malicious packages from being installed on your computer. We do this by wrapping your package manager.")
+        prompt = "Do you want to enable proactive malicious package prevention?"
+        answer = Prompt.ask(prompt=prompt, choices=["y", "n"],
+                            default="y", show_default=True, console=console).lower()
+
+    if answer == 'y':
+        configure_system()
+
+    if prompt_user:
+        prompt = "Do you want to alias pip to Safety?"
+        answer = Prompt.ask(prompt=prompt, choices=["y", "n"],
+                            default="y", show_default=True, console=console).lower()
+
+    if answer == 'y':
+        configure_alias()
+
+    if has_local_tool_files(project_dir):
+        if prompt_user:
+            prompt = "Do you want to enable proactive malicious package prevention for any project in working directory?"
+            answer = Prompt.ask(prompt=prompt, choices=["y", "n"],
+                                default="y", show_default=True, console=console).lower()
+
+        if answer == 'y':
+            configure_local_directory(project_dir)
+
+        if prompt_user:
+            prompt = "It looks like your current directory contains a requirements.txt file. Would you like Safety to scan it?"
+            answer = Prompt.ask(prompt=prompt, choices=["y", "n"],
+                                default="y", show_default=True, console=console).lower()
+
+        if answer == 'y':
+            ctx.command.name = "scan"
+            ctx.params = {
+                "target": directory,
+                "output": ScanOutput.SCREEN,
+                "policy_file_path": None
+            }
+            scan(ctx=ctx, target=directory, output=ScanOutput.SCREEN, policy_file_path=None)

safety/init/constants.py

@@ -0,0 +1,6 @@
+# Project options
+PROJECT_INIT_CMD_NAME = "init"
+PROJECT_INIT_HELP = "[BETA] Creates new Safety CLI project in the current working directory."\
+"\nExample: safety project init"
+PROJECT_INIT_DIRECTORY_HELP = "[BETA] Defines a directory for creating a new project. (default: current directory)\n\n" \
+    "[bold]Example: safety project init /path/to/project[/bold]"