feat: add security events for firewall users

Author: yeisonvargasf

pyproject.toml

@@ -43,8 +43,10 @@ dependencies = [
   "psutil~=6.1.0",
   "pydantic>=2.6.0,<2.10.0",
   "requests", # TODO: To be replaced by httpx or aiohttp
+  "httpx",
+  "tenacity",
   "ruamel.yaml>=0.17.21",
-  "safety-schemas==0.0.11",
+  "safety-schemas==0.0.12",
   # TODO: To be removed after migrate away from pkg_resources
   "setuptools>=65.5.1",
   "typer>=0.12.1",
@@ -235,7 +237,7 @@ style = [
 ]
 fmt = [
   "ruff format {args:.}",
-  "ruff --fix {args:.}",
+  "ruff check --fix {args:.}",
   "style",
 ]
 all = [

safety/auth/cli.py

@@ -7,6 +7,7 @@
 from safety.console import main_console as console
 from safety.constants import MSG_FINISH_REGISTRATION_TPL, MSG_VERIFICATION_HINT
 from safety.meta import get_version
+from safety.decorators import notify
 
 try:
     from typing import Annotated
@@ -27,6 +28,7 @@
     get_token,
 )
 from safety.auth.server import process_browser_callback
+from safety.util import initialize_event_bus
 from safety.scan.constants import (
     CLI_AUTH_COMMAND_HELP,
     CLI_AUTH_HEADLESS_HELP,
@@ -37,6 +39,7 @@
 )
 
 from ..cli_util import SafetyCLISubGroup, get_command_for, pass_safety_cli_obj
+from safety.error_handlers import handle_cmd_exception
 from .constants import (
     MSG_FAIL_LOGIN_AUTHED,
     MSG_FAIL_REGISTER_AUTHED,
@@ -50,19 +53,20 @@
 auth_app = Typer(rich_markup_mode="rich", name="auth")
 
 
-
 CMD_LOGIN_NAME = "login"
 CMD_REGISTER_NAME = "register"
 CMD_STATUS_NAME = "status"
 CMD_LOGOUT_NAME = "logout"
 DEFAULT_CMD = CMD_LOGIN_NAME
 
-@auth_app.callback(invoke_without_command=True,
-                   cls=SafetyCLISubGroup,
-                   help=CLI_AUTH_COMMAND_HELP,
-                   epilog=DEFAULT_EPILOG,
-                   context_settings={"allow_extra_args": True,
-                                     "ignore_unknown_options": True})
+
+@auth_app.callback(
+    invoke_without_command=True,
+    cls=SafetyCLISubGroup,
+    help=CLI_AUTH_COMMAND_HELP,
+    epilog=DEFAULT_EPILOG,
+    context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
+)
 @pass_safety_cli_obj
 def auth(ctx: typer.Context) -> None:
     """
@@ -71,12 +75,11 @@ def auth(ctx: typer.Context) -> None:
     Args:
         ctx (typer.Context): The Typer context object.
     """
-    LOG.info('auth started')
+    LOG.info("auth started")
 
     # If no subcommand is invoked, forward to the default command
     if not ctx.invoked_subcommand:
-        default_command = get_command_for(name=DEFAULT_CMD,
-                                          typer_instance=auth_app)
+        default_command = get_command_for(name=DEFAULT_CMD, typer_instance=auth_app)
         return ctx.forward(default_command)
 
 
@@ -99,6 +102,7 @@ def fail_if_authenticated(ctx: typer.Context, with_msg: str) -> None:
         console.print(with_msg.format(email=email))
         sys.exit(0)
 
+
 def render_email_note(auth: Auth) -> str:
     """
     Renders a note indicating whether email verification is required.
@@ -111,8 +115,8 @@ def render_email_note(auth: Auth) -> str:
     """
     return "" if auth.email_verified else "[red](email verification required)[/red]"
 
-def render_successful_login(auth: Auth,
-                            organization: Optional[str] = None) -> None:
+
+def render_successful_login(auth: Auth, organization: Optional[str] = None) -> None:
     """
     Renders a message indicating a successful login.
 
@@ -132,34 +136,34 @@ def render_successful_login(auth: Auth,
         details = [f"[green][bold]Account:[/bold] {email}[/green] {email_note}"]
 
     if organization:
-        details.insert(0,
-                        "[green][bold]Organization:[/bold] " \
-                        f"{organization}[green]")
+        details.insert(0, f"[green][bold]Organization:[/bold] {organization}[green]")
 
     for msg in details:
         console.print(Padding(msg, (0, 0, 0, 1)), emoji=True)
 
 
 @auth_app.command(name=CMD_LOGIN_NAME, help=CLI_AUTH_LOGIN_HELP)
+@handle_cmd_exception
+@notify
 def login(
     ctx: typer.Context,
     headless: Annotated[
         Optional[bool],
         typer.Option(
             "--headless",
             help=CLI_AUTH_HEADLESS_HELP,
-        )
-    ] = None
+        ),
+    ] = None,
 ) -> None:
-    headless = headless is True 
+    headless = headless is True
     """
     Authenticate Safety CLI with your safetycli.com account using your default browser.
 
     Args:
         ctx (typer.Context): The Typer context object.
         headless (bool): Whether to run in headless mode.
     """
-    LOG.info('login started')
+    LOG.info("login started")
 
     # Check if the user is already authenticated
     fail_if_authenticated(ctx, with_msg=MSG_FAIL_LOGIN_AUTHED)
@@ -168,26 +172,33 @@ def login(
 
     info = None
 
-    brief_msg: str = "Redirecting your browser to log in; once authenticated, " \
-            "return here to start using Safety"
+    brief_msg: str = (
+        "Redirecting your browser to log in; once authenticated, "
+        "return here to start using Safety"
+    )
 
     if ctx.obj.auth.org:
-        console.print(f"Logging into [bold]{ctx.obj.auth.org.name}[/bold] " \
-                      "organization.")
+        console.print(
+            f"Logging into [bold]{ctx.obj.auth.org.name}[/bold] organization."
+        )
 
     if headless:
         brief_msg = "Running in headless mode. Please copy and open the following URL in a browser"
 
     # Get authorization data and generate the authorization URL
-    uri, initial_state = get_authorization_data(client=ctx.obj.auth.client,
-                                                code_verifier=ctx.obj.auth.code_verifier,
-                                                organization=ctx.obj.auth.org, headless=headless)
+    uri, initial_state = get_authorization_data(
+        client=ctx.obj.auth.client,
+        code_verifier=ctx.obj.auth.code_verifier,
+        organization=ctx.obj.auth.org,
+        headless=headless,
+    )
     click.secho(brief_msg)
     click.echo()
 
     # Process the browser callback to complete the authentication
-    info = process_browser_callback(uri, initial_state=initial_state, ctx=ctx, headless=headless)
-
+    info = process_browser_callback(
+        uri, initial_state=initial_state, ctx=ctx, headless=headless
+    )
 
     if info:
         if info.get("email", None):
@@ -198,50 +209,60 @@ def login(
             if headless:
                 console.print()
 
-            render_successful_login(ctx.obj.auth, organization=organization)
-
             initialize(ctx, refresh=True)
+            initialize_event_bus(ctx=ctx)
+            render_successful_login(ctx.obj.auth, organization=organization)
 
             console.print()
             if ctx.obj.auth.org or ctx.obj.auth.email_verified:
                 console.print(
-                    "[tip]Tip[/tip]: now try [bold]`safety scan`[/bold] in your project’s root " \
-                    "folder to run a project scan or [bold]`safety -–help`[/bold] to learn more.")
+                    "[tip]Tip[/tip]: now try [bold]`safety scan`[/bold] in your project’s root "
+                    "folder to run a project scan or [bold]`safety -–help`[/bold] to learn more."
+                )
             else:
-                console.print(MSG_FINISH_REGISTRATION_TPL.format(email=ctx.obj.auth.email))
+                console.print(
+                    MSG_FINISH_REGISTRATION_TPL.format(email=ctx.obj.auth.email)
+                )
                 console.print()
                 console.print(MSG_VERIFICATION_HINT)
         else:
             click.secho("Safety is now authenticated but your email is missing.")
     else:
         msg = ":stop_sign: [red]"
         if ctx.obj.auth.org:
-            msg += f"Error logging into {ctx.obj.auth.org.name} organization " \
+            msg += (
+                f"Error logging into {ctx.obj.auth.org.name} organization "
                 f"with auth ID: {ctx.obj.auth.org.id}."
+            )
         else:
             msg += "Error logging into Safety."
 
-        msg += " Please try again, or use [bold]`safety auth -–help`[/bold] " \
+        msg += (
+            " Please try again, or use [bold]`safety auth -–help`[/bold] "
             "for more information[/red]"
+        )
 
         console.print(msg, emoji=True)
 
+
 @auth_app.command(name=CMD_LOGOUT_NAME, help=CLI_AUTH_LOGOUT_HELP)
+@handle_cmd_exception
+@notify
 def logout(ctx: typer.Context) -> None:
     """
     Log out of your current session.
 
     Args:
         ctx (typer.Context): The Typer context object.
     """
-    LOG.info('logout started')
+    LOG.info("logout started")
 
-    id_token = get_token('id_token')
+    id_token = get_token("id_token")
 
     msg = MSG_NON_AUTHENTICATED
 
     if id_token:
-         # Clean the session if an ID token is found
+        # Clean the session if an ID token is found
         if clean_session(ctx.obj.auth.client):
             msg = MSG_LOGOUT_DONE
         else:
@@ -251,13 +272,23 @@ def logout(ctx: typer.Context) -> None:
 
 
 @auth_app.command(name=CMD_STATUS_NAME, help=CLI_AUTH_STATUS_HELP)
-@click.option("--ensure-auth/--no-ensure-auth", default=False,
-              help="This will keep running the command until an" \
-                "authentication is made.")
-@click.option("--login-timeout", "-w", type=int, default=600,
-              help="Max time allowed to wait for an authentication.")
-def status(ctx: typer.Context, ensure_auth: bool = False,
-           login_timeout: int = 600)  -> None:
+@click.option(
+    "--ensure-auth/--no-ensure-auth",
+    default=False,
+    help="This will keep running the command until anauthentication is made.",
+)
+@click.option(
+    "--login-timeout",
+    "-w",
+    type=int,
+    default=600,
+    help="Max time allowed to wait for an authentication.",
+)
+@handle_cmd_exception
+@notify
+def status(
+    ctx: typer.Context, ensure_auth: bool = False, login_timeout: int = 600
+) -> None:
     """
     Display Safety CLI's current authentication status.
 
@@ -266,7 +297,7 @@ def status(ctx: typer.Context, ensure_auth: bool = False,
         ensure_auth (bool): Whether to keep running until authentication is made.
         login_timeout (int): Max time allowed to wait for authentication.
     """
-    LOG.info('status started')
+    LOG.info("status started")
     current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
     safety_version = get_version()
     console.print(f"[{current_time}]: Safety {safety_version}")
@@ -283,20 +314,28 @@ def status(ctx: typer.Context, ensure_auth: bool = False,
         verified = is_email_verified(info)
         email_status = " [red](email not verified)[/red]" if not verified else ""
 
-        console.print(f'[green]Authenticated as {info["email"]}[/green]{email_status}')
+        console.print(f"[green]Authenticated as {info['email']}[/green]{email_status}")
     elif ensure_auth:
-        console.print('Safety is not authenticated. Launching default browser to log in')
+        console.print(
+            "Safety is not authenticated. Launching default browser to log in"
+        )
         console.print()
-        uri, initial_state = get_authorization_data(client=ctx.obj.auth.client,
-                                                    code_verifier=ctx.obj.auth.code_verifier,
-                                                    organization=ctx.obj.auth.org, ensure_auth=ensure_auth)
+        uri, initial_state = get_authorization_data(
+            client=ctx.obj.auth.client,
+            code_verifier=ctx.obj.auth.code_verifier,
+            organization=ctx.obj.auth.org,
+            ensure_auth=ensure_auth,
+        )
 
         # Process the browser callback to complete the authentication
-        info = process_browser_callback(uri, initial_state=initial_state,
-                                          timeout=login_timeout, ctx=ctx)
+        info = process_browser_callback(
+            uri, initial_state=initial_state, timeout=login_timeout, ctx=ctx
+        )
 
         if not info:
-            console.print(f'[red]Timeout error ({login_timeout} seconds): not successfully authenticated without the timeout period.[/red]')
+            console.print(
+                f"[red]Timeout error ({login_timeout} seconds): not successfully authenticated without the timeout period.[/red]"
+            )
             sys.exit(1)
 
         organization = None
@@ -311,33 +350,37 @@ def status(ctx: typer.Context, ensure_auth: bool = False,
 
 
 @auth_app.command(name=CMD_REGISTER_NAME)
+@handle_cmd_exception
+@notify
 def register(ctx: typer.Context) -> None:
     """
     Create a new user account for the safetycli.com service.
 
     Args:
         ctx (typer.Context): The Typer context object.
     """
-    LOG.info('register started')
+    LOG.info("register started")
 
     # Check if the user is already authenticated
     fail_if_authenticated(ctx, with_msg=MSG_FAIL_REGISTER_AUTHED)
 
     # Get authorization data and generate the registration URL
-    uri, initial_state = get_authorization_data(client=ctx.obj.auth.client,
-                                                code_verifier=ctx.obj.auth.code_verifier,
-                                                sign_up=True)
-
-    console.print("Redirecting your browser to register for a free account. Once registered, return here to start using Safety.")
+    uri, initial_state = get_authorization_data(
+        client=ctx.obj.auth.client,
+        code_verifier=ctx.obj.auth.code_verifier,
+        sign_up=True,
+    )
+
+    console.print(
+        "Redirecting your browser to register for a free account. Once registered, return here to start using Safety."
+    )
     console.print()
 
     # Process the browser callback to complete the registration
-    info = process_browser_callback(uri,
-                                    initial_state=initial_state, ctx=ctx)
+    info = process_browser_callback(uri, initial_state=initial_state, ctx=ctx)
 
     if info:
-        console.print(f'[green]Successfully registered {info.get("email")}[/green]')
+        console.print(f"[green]Successfully registered {info.get('email')}[/green]")
         console.print()
     else:
-        console.print('[red]Unable to register in this time, try again.[/red]')
-
+        console.print("[red]Unable to register in this time, try again.[/red]")