{message}
" + ).format(title=html.escape(title), message=html.escape(message)) + encoded = body.encode("utf-8") + + self.send_response(200) + self._send_cors_headers() + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def do_POST(self): # pylint: disable=invalid-name + """Handle background JSON handoffs from the web login page.""" + parsed = urlparse(self.path) + if parsed.path != CALLBACK_PATH: + self.send_error(404) + return + + try: + content_length = int(self.headers.get("Content-Length", "0")) + except ValueError: + content_length = 0 + + raw_body = self.rfile.read(content_length) if content_length > 0 else b"{}" + try: + payload = json.loads(raw_body.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError): + self.send_response(400) + self._send_cors_headers() + self.send_header("Content-Type", "application/json; charset=utf-8") + self.end_headers() + self.wfile.write(b'{"error":"Invalid JSON payload."}') + return + + params = { + key: value + for key, value in payload.items() + if isinstance(key, str) and isinstance(value, str) + } + self._store_callback_params(params) + + encoded = b'{"status":"ok"}' + self.send_response(200) + self._send_cors_headers() + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def log_message(self, format, *args): # pylint: disable=redefined-builtin + return + + +def wait_for_login( + environment, + profile: str, + port: Optional[int] = None, + timeout: int = 120, + use_local_ui: bool = False, + announce_login: Optional[Callable[[Dict[str, str]], None]] = None, +): # pylint: disable=too-many-arguments,too-many-locals + """Run the browser-based login flow and persist the resulting API key.""" + host = "127.0.0.1" + callback_port = port if port is not None else _find_available_port(host) + callback_url = f"http://{host}:{callback_port}{CALLBACK_PATH}" + state = secrets.token_urlsafe(24) + server = _LoginCallbackServer((host, callback_port)) + server.timeout = 0.2 + + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + login_url = build_login_url( + environment, callback_url, state, profile, use_local_ui=use_local_ui + ) + + try: + try: + opened = webbrowser.open(login_url) + except webbrowser.Error: # pragma: no cover - platform/browser dependent + opened = False + + if announce_login is not None: + announce_login( + { + "login_url": login_url, + "callback_url": callback_url, + "browser_opened": "true" if opened else "false", + "environment": environment.name, + "profile": profile, + } + ) + + if not server.callback_event.wait(timeout): + raise LoginError( + f"Timed out waiting for login callback after {timeout} seconds. " + f"Retry with the same environment and open the printed URL manually if needed." + ) + + params = server.callback_params + if params.get("state") != state: + raise LoginError("Login callback state mismatch.") + if "error" in params: + raise LoginError(params["error"]) + + apikey = params.get("apikey") + if not apikey: + raise LoginError("Login callback did not include an API key.") + + storage_environment = None if environment.name == Env.prod.name else environment.name + store_apikey(apikey, profile=profile, environment_name=storage_environment) + user_config.UserConfig = user_config.BasicUserConfig() + return { + "status": "success", + "login_url": login_url, + "callback_url": callback_url, + "environment": environment.name, + "profile": profile, + "browser_opened": opened, + "email": params.get("email"), + } + finally: + server.shutdown() + server.server_close() + thread.join(timeout=1) diff --git a/flow360/user_config.py b/flow360/user_config.py index 57031e7e0..134a06f51 100644 --- a/flow360/user_config.py +++ b/flow360/user_config.py @@ -3,6 +3,7 @@ """ import os +from typing import Optional import toml @@ -12,6 +13,86 @@ config_file = os.path.join(flow360_dir, "config.toml") DEFAULT_PROFILE = "default" +CONFIG_DIR_MODE = 0o700 +CONFIG_FILE_MODE = 0o600 + + +def _ensure_permissions(path: str, mode: int): + """Best-effort permission hardening for local config paths.""" + try: + os.chmod(path, mode) + except PermissionError: + pass + + +def ensure_config_dir(): + """Ensure the Flow360 config directory exists.""" + os.makedirs(os.path.dirname(config_file), exist_ok=True) + _ensure_permissions(os.path.dirname(config_file), CONFIG_DIR_MODE) + + +def read_user_config(): + """Read the user config file if present.""" + if os.path.exists(config_file): + _ensure_permissions(config_file, CONFIG_FILE_MODE) + with open(config_file, encoding="utf-8") as file_handler: + return toml.loads(file_handler.read()) + return {} + + +def write_user_config(config): + """Write the user config file.""" + ensure_config_dir() + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + file_descriptor = os.open(config_file, flags, CONFIG_FILE_MODE) + with os.fdopen(file_descriptor, "w", encoding="utf-8") as file_handler: + file_handler.write(toml.dumps(config)) + _ensure_permissions(config_file, CONFIG_FILE_MODE) + + +def store_apikey( + apikey: str, profile: str = DEFAULT_PROFILE, environment_name: Optional[str] = None +): + """Store an API key using the same config layout consumed by UserConfig.""" + config = read_user_config() + + if environment_name in (None, "", prod.name): + entry = {profile: {"apikey": apikey}} + else: + entry = {profile: {environment_name: {"apikey": apikey}}} + + # Avoid importing CLI modules at import time because the wider package has lazy-import paths. + from flow360.cli import dict_utils # pylint: disable=import-outside-toplevel + + dict_utils.merge_overwrite(config, entry) + write_user_config(config) + return config + + +def delete_apikey(profile: str = DEFAULT_PROFILE, environment_name: Optional[str] = None): + """Delete a stored API key for the selected profile/environment if present.""" + config = read_user_config() + profile_config = config.get(profile) + + if not isinstance(profile_config, dict): + return False, config + + removed = False + if environment_name in (None, "", prod.name): + removed = profile_config.pop("apikey", None) is not None + else: + env_config = profile_config.get(environment_name) + if isinstance(env_config, dict): + removed = env_config.pop("apikey", None) is not None + if not env_config: + profile_config.pop(environment_name, None) + + if not profile_config: + config.pop(profile, None) + + if removed: + write_user_config(config) + return removed, config class BasicUserConfig: @@ -58,10 +139,7 @@ def set_profile(self, profile: str = DEFAULT_PROFILE): log.info(f"Using profile={profile} for apikey") def _read_config(self): - self.config = {} - if os.path.exists(config_file): - with open(config_file, encoding="utf-8") as file_handler: - self.config = toml.loads(file_handler.read()) + self.config = read_user_config() def apikey(self, env): """get apikey diff --git a/tests/v1/_test_cli.py b/tests/v1/_test_cli.py index ccd663548..aaf3b791f 100644 --- a/tests/v1/_test_cli.py +++ b/tests/v1/_test_cli.py @@ -1,33 +1,77 @@ -import os.path -from os.path import expanduser - import toml from click.testing import CliRunner -home = expanduser("~") +import flow360.cli.app as app +import flow360.user_config as user_config +from flow360.cli import flow360 + + +def _patch_config_file(monkeypatch, tmp_path): + config_path = tmp_path / "config.toml" + monkeypatch.setattr(user_config, "config_file", str(config_path)) + monkeypatch.setattr(app, "config_file", str(config_path)) + return config_path class TestClass: - def test_no_configure(self): + def test_no_configure(self, monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + runner = CliRunner() + + result = runner.invoke(flow360, ["configure", "--apikey", "apikey"]) + + assert result.exit_code == 0 + assert result.output == "done.\n" + config = toml.loads(config_path.read_text()) + assert config.get("default", {}).get("apikey", "") == "apikey" + + def test_configure(self, monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + config_path.write_text(toml.dumps({"default": {"apikey": "apikey"}})) + runner = CliRunner() + + result = runner.invoke(flow360, ["configure", "--apikey", "apikey"]) + + assert result.exit_code == 0 + assert result.output == "done.\n" + config = toml.loads(config_path.read_text()) + assert config.get("default", {}).get("apikey", "") == "apikey" + + def test_logout_removes_default_apikey(self, monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + config_path.write_text(toml.dumps({"default": {"apikey": "apikey"}})) + runner = CliRunner() + + result = runner.invoke(flow360, ["logout"]) + + assert result.exit_code == 0 + assert ( + result.output == "Removed stored API key for profile 'default' in environment 'prod'.\n" + ) + config = toml.loads(config_path.read_text()) + assert "apikey" not in config.get("default", {}) + + def test_logout_removes_dev_apikey(self, monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + config_path.write_text(toml.dumps({"default": {"dev": {"apikey": "dev-key"}}})) runner = CliRunner() - if os.path.exists(f"{home}/.flow360/config.toml"): - os.remove(f"{home}/.flow360/config.toml") - from flow360.cli import flow360 - result = runner.invoke(flow360, ["configure"], input="apikey") + result = runner.invoke(flow360, ["logout", "--dev"]) + assert result.exit_code == 0 - assert result.output == "API Key: apikey\ndone.\n" - with open(f"{home}/.flow360/config.toml") as f: - config = toml.loads(f.read()) - assert config.get("default", {}).get("apikey", "") == "apikey" + assert ( + result.output == "Removed stored API key for profile 'default' in environment 'dev'.\n" + ) + config = toml.loads(config_path.read_text()) + assert "dev" not in config.get("default", {}) - def test_configure(self): + def test_logout_reports_missing_apikey(self, monkeypatch, tmp_path): + _patch_config_file(monkeypatch, tmp_path) runner = CliRunner() - from flow360.cli import flow360 - result = runner.invoke(flow360, ["configure"], input="apikey") + result = runner.invoke(flow360, ["logout", "--dev"]) + assert result.exit_code == 0 - assert result.output == "API Key[apikey]: apikey\ndone.\n" - with open(f"{home}/.flow360/config.toml") as f: - config = toml.loads(f.read()) - assert config.get("default", {}).get("apikey", "") == "apikey" + assert ( + result.output == "No stored API key found for profile 'default' in environment 'dev'.\n" + ) diff --git a/tests/v1/test_cli_login.py b/tests/v1/test_cli_login.py new file mode 100644 index 000000000..2e479fe64 --- /dev/null +++ b/tests/v1/test_cli_login.py @@ -0,0 +1,184 @@ +from threading import Thread +from time import sleep +from urllib.parse import parse_qs, urlparse + +import requests +import toml +from click.testing import CliRunner + +import flow360.cli.app as app +import flow360.cli.auth as auth +import flow360.user_config as user_config +from flow360.cli import flow360 +from flow360.environment import Env + + +def _post_callback_with_retry( + callback_url: str, payload: dict, attempts: int = 50, delay: float = 0.1 +): + last_error = None + for _ in range(attempts): + try: + response = requests.post(callback_url, json=payload, timeout=5) + response.raise_for_status() + return + except requests.RequestException as error: + last_error = error + sleep(delay) + + if last_error is not None: + raise last_error + + +def _patch_config_file(monkeypatch, tmp_path): + config_path = tmp_path / "config.toml" + monkeypatch.setattr(user_config, "config_file", str(config_path)) + monkeypatch.setattr(app, "config_file", str(config_path)) + return config_path + + +def test_configure_stores_dev_apikey(monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + runner = CliRunner() + + result = runner.invoke(flow360, ["configure", "--apikey", "dev-key", "--dev"]) + + assert result.exit_code == 0 + config = toml.loads(config_path.read_text()) + assert config["default"]["dev"]["apikey"] == "dev-key" + + +def test_login_uses_dev_web_url_with_manual_fallback(monkeypatch, tmp_path): + _patch_config_file(monkeypatch, tmp_path) + monkeypatch.setattr(auth, "_find_available_port", lambda host: 8765) + monkeypatch.setattr(auth.secrets, "token_urlsafe", lambda _: "state123") + monkeypatch.setattr(auth.webbrowser, "open", lambda _: False) + runner = CliRunner() + + def complete_manual_login(): + _post_callback_with_retry( + "http://127.0.0.1:8765/callback", + {"state": "state123", "apikey": "dev-manual-key", "email": "dev@example.com"}, + ) + + Thread(target=complete_manual_login, daemon=True).start() + + result = runner.invoke(flow360, ["login", "--dev"]) + + assert result.exit_code == 0 + assert "Starting local login server on http://127.0.0.1:8765/callback." in result.output + assert "flow360.dev-simulation.cloud/account/cli-login" in result.output + assert ( + "Could not open your browser automatically. Navigate to this URL to authenticate:" + in result.output + ) + assert "Successfully logged in as dev@example.com" in result.output + + +def test_login_local_uses_local_dev_frontend(monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + monkeypatch.setattr(auth, "_find_available_port", lambda host: 8765) + monkeypatch.setattr(auth.secrets, "token_urlsafe", lambda _: "state123") + monkeypatch.setattr(auth.webbrowser, "open", lambda _: False) + runner = CliRunner() + + def complete_local_login(): + _post_callback_with_retry( + "http://127.0.0.1:8765/callback", + {"state": "state123", "apikey": "local-dev-key", "email": "local@example.com"}, + ) + + Thread(target=complete_local_login, daemon=True).start() + + result = runner.invoke(flow360, ["login", "--local"]) + + assert result.exit_code == 0 + assert "Starting local login server on http://127.0.0.1:8765/callback." in result.output + assert "local.dev-simulation.cloud:3000/account/cli-login" in result.output + assert "Successfully logged in as local@example.com" in result.output + config = toml.loads(config_path.read_text()) + assert config["default"]["dev"]["apikey"] == "local-dev-key" + + +def test_login_prints_fallback_message_when_browser_opens(monkeypatch, tmp_path): + _patch_config_file(monkeypatch, tmp_path) + monkeypatch.setattr(auth.secrets, "token_urlsafe", lambda _: "state123") + + def fake_open(login_url): + parsed = urlparse(login_url) + params = parse_qs(parsed.query) + callback_url = params["callback_url"][0] + Thread( + target=lambda: requests.post( + callback_url, + json={ + "state": "state123", + "apikey": "dev-browser-key", + "email": "browser@example.com", + }, + timeout=5, + ), + daemon=True, + ).start() + return True + + monkeypatch.setattr(auth.webbrowser, "open", fake_open) + runner = CliRunner() + + result = runner.invoke(flow360, ["login", "--dev"]) + + assert result.exit_code == 0 + assert "Starting local login server on " in result.output + assert "flow360.dev-simulation.cloud/account/cli-login" in result.output + assert "If your browser did not open, navigate to this URL to authenticate:" in result.output + assert "Successfully logged in as browser@example.com" in result.output + + +def test_build_login_url_omits_prod_env(): + login_url = auth.build_login_url( + environment=Env.prod, + callback_url="http://127.0.0.1:8765/callback", + state="state123", + profile="default", + ) + + parsed = urlparse(login_url) + params = parse_qs(parsed.query) + + assert parsed.netloc == "flow360.simulation.cloud" + assert params["profile"] == ["default"] + assert params["state"] == ["state123"] + assert params["callback_url"] == ["http://127.0.0.1:8765/callback"] + assert "env" not in params + + +def test_wait_for_login_stores_dev_apikey(monkeypatch, tmp_path): + config_path = _patch_config_file(monkeypatch, tmp_path) + monkeypatch.setattr(auth.secrets, "token_urlsafe", lambda _: "state123") + + def fake_open(login_url): + parsed = urlparse(login_url) + params = parse_qs(parsed.query) + callback_url = params["callback_url"][0] + Thread( + target=lambda: requests.post( + callback_url, + json={ + "state": "state123", + "apikey": "dev-browser-key", + "email": "browser@example.com", + }, + timeout=5, + ), + daemon=True, + ).start() + return True + + monkeypatch.setattr(auth.webbrowser, "open", fake_open) + + result = auth.wait_for_login(environment=Env.dev, profile="default", timeout=5) + + assert result["status"] == "success" + assert result["email"] == "browser@example.com" + config = toml.loads(config_path.read_text()) + assert config["default"]["dev"]["apikey"] == "dev-browser-key"