diff --git a/.vscode/launch.json b/.vscode/launch.json index c3d91a7c..18caeb25 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -11,6 +11,16 @@ ], "console": "integratedTerminal" }, + { + "name": "Safety Auth Login --headless", + "type": "debugpy", + "request": "launch", + "module": "safety", + "args": [ + "auth","login","--headless" + ], + "console": "integratedTerminal" + }, { "name": "Safety Auth Logout", "type": "debugpy", diff --git a/safety/auth/server.py b/safety/auth/server.py index 9003df9a..9214dafc 100644 --- a/safety/auth/server.py +++ b/safety/auth/server.py @@ -10,6 +10,7 @@ import click from safety.auth.cli_utils import load_auth_session +from safety.auth.utils import is_jupyter_notebook from safety.console import main_console as console from safety.auth.constants import AUTH_SERVER_URL, CLI_AUTH_SUCCESS, CLI_LOGOUT_SUCCESS, HOST @@ -191,7 +192,6 @@ class ThreadedHTTPServer(http.server.HTTPServer): def __init__(self, server_address: Tuple, RequestHandlerClass: Any) -> None: """ Initialize the ThreadedHTTPServer. - Args: server_address (Tuple): The server address as a tuple (host, port). RequestHandlerClass (Any): The request handler class. @@ -219,9 +219,7 @@ def handle_timeout(self) -> None: headless = kwargs.get("headless", False) initial_state = kwargs.get("initial_state", None) ctx = kwargs.get("ctx", None) - - message = "Copy and paste this url into your browser:" - + message = "Copy and paste this URL into your browser:\n⚠️ Ensure there are no extra spaces, especially at line breaks, as they may break the link." if not headless: # Start a threaded HTTP server to handle the callback @@ -231,13 +229,15 @@ def handle_timeout(self) -> None: server.ctx = ctx server_thread = threading.Thread(target=server.handle_request) server_thread.start() - message = f"If the browser does not automatically open in 5 seconds, " \ - "copy and paste this url into your browser:" + message = f"If the browser does not automatically open in 5 seconds, copy and paste this url into your browser:" target = uri if headless else f"{uri}&port={PORT}" - console.print(f"{message} [link={target}]{target}[/link]") - console.print() + if is_jupyter_notebook(): + console.print(f"{message} {target}") + else: + console.print(f"{message} [link={target}]{target}[/link]") + if headless: # Handle the headless mode where user manually provides the response exchange_data = None @@ -247,18 +247,17 @@ def handle_timeout(self) -> None: exchange_data = json.loads(auth_code_text) state = exchange_data["state"] code = exchange_data["code"] - except Exception as e: + except Exception: code = state = None return auth_process(code=code, - state=state, - initial_state=initial_state, - code_verifier=ctx.obj.auth.code_verifier, - client=ctx.obj.auth.client) + state=state, + initial_state=initial_state, + code_verifier=ctx.obj.auth.code_verifier, + client=ctx.obj.auth.client) else: # Wait for the browser authentication in non-headless mode wait_msg = "waiting for browser authentication" - with console.status(wait_msg, spinner="bouncingBar"): time.sleep(2) click.launch(target) @@ -266,10 +265,7 @@ def handle_timeout(self) -> None: except OSError as e: if e.errno == socket.errno.EADDRINUSE: - reason = f"The port {HOST}:{PORT} is currently being used by another" \ - "application or process. Please choose a different port or " \ - "terminate the conflicting application/process to free up " \ - "the port." + reason = f"The port {HOST}:{PORT} is currently being used by another application or process. Please choose a different port or terminate the conflicting application/process to free up the port." else: reason = "An error occurred while performing this operation." diff --git a/safety/auth/utils.py b/safety/auth/utils.py index 922d949a..663c3a08 100644 --- a/safety/auth/utils.py +++ b/safety/auth/utils.py @@ -1,3 +1,4 @@ +from functools import lru_cache import json import logging from typing import Any, Optional, Dict, Callable, Tuple @@ -425,3 +426,21 @@ def send(self, request: requests.PreparedRequest, **kwargs: Any) -> requests.Res """ request.headers.pop("Authorization", None) return super().send(request, **kwargs) + + +@lru_cache(maxsize=1) +def is_jupyter_notebook() -> bool: + """ + Check if the code is being executed inside a Jupyter Notebook. + + Returns: + bool: True if running in a Jupyter notebook, False otherwise. + """ + try: + from IPython import get_ipython + shell = get_ipython() + # Check for the presence of 'IPKernelApp', indicating a Jupyter notebook environment. + return 'IPKernelApp' in shell.config + except (ImportError, NameError, AttributeError): + # IPython is not available or has no `config` attribute; likely not a notebook. + return False