diff --git a/chipflow/auth.py b/chipflow/auth.py new file mode 100644 index 00000000..39670caf --- /dev/null +++ b/chipflow/auth.py @@ -0,0 +1,308 @@ +# SPDX-License-Identifier: BSD-2-Clause + +""" +ChipFlow authentication helper module. + +Handles authentication for ChipFlow API with multiple fallback methods: +1. Environment variable CHIPFLOW_API_KEY +2. GitHub CLI token authentication (if gh is available) +3. OAuth 2.0 Device Flow +""" + +import logging +import os +import subprocess +import sys +import time +import requests +from pathlib import Path +import json + +logger = logging.getLogger(__name__) + + +class AuthenticationError(Exception): + """Exception raised when authentication fails.""" + pass + + +def get_credentials_file(): + """Get path to credentials file.""" + config_dir = Path.home() / ".config" / "chipflow" + return config_dir / "credentials" + + +def save_api_key(api_key: str): + """Save API key to credentials file.""" + creds_file = get_credentials_file() + creds_file.parent.mkdir(parents=True, exist_ok=True) + + creds_data = {"api_key": api_key} + creds_file.write_text(json.dumps(creds_data)) + creds_file.chmod(0o600) + + logger.info(f"API key saved to {creds_file}") + + +def load_saved_api_key(): + """Load API key from credentials file if it exists.""" + creds_file = get_credentials_file() + if not creds_file.exists(): + return None + + try: + creds_data = json.loads(creds_file.read_text()) + return creds_data.get("api_key") + except (json.JSONDecodeError, KeyError): + logger.warning(f"Invalid credentials file at {creds_file}") + return None + + +def is_gh_authenticated(): + """Check if GitHub CLI is installed and authenticated.""" + try: + result = subprocess.run( + ["gh", "auth", "status"], + capture_output=True, + text=True, + timeout=5 + ) + return result.returncode == 0 + except (subprocess.TimeoutExpired, FileNotFoundError): + return False + + +def get_gh_token(): + """Get GitHub token from gh CLI.""" + try: + result = subprocess.run( + ["gh", "auth", "token"], + capture_output=True, + text=True, + check=True, + timeout=5 + ) + return result.stdout.strip() + except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): + return None + + +def authenticate_with_github_token(api_origin: str, interactive: bool = True): + """ + Authenticate using GitHub CLI token. + + Args: + api_origin: ChipFlow API origin URL + interactive: Whether to show interactive messages + + Returns: + API key on success, None on failure + """ + if interactive: + print("🔍 Checking for GitHub CLI authentication...") + + if not is_gh_authenticated(): + if interactive: + print("⚠️ GitHub CLI is not authenticated or not installed") + return None + + gh_token = get_gh_token() + if not gh_token: + if interactive: + print("⚠️ Could not get GitHub token from gh CLI") + return None + + if interactive: + print("🔑 Authenticating with GitHub token...") + + try: + response = requests.post( + f"{api_origin}/auth/github-token", + json={"github_token": gh_token}, + timeout=10 + ) + + if response.status_code == 200: + try: + api_key = response.json()["api_key"] + save_api_key(api_key) + if interactive: + print("✅ Authenticated using GitHub CLI!") + return api_key + except (KeyError, ValueError) as e: + if interactive: + print("⚠️ Invalid response from authentication server") + logger.debug(f"Invalid JSON response on success: {e}, body: {response.text[:200]}") + return None + else: + try: + error_msg = response.json().get("error_description", "Unknown error") + except ValueError: + error_msg = f"HTTP {response.status_code}" + logger.debug(f"Non-JSON error response: {response.text[:200]}") + + if interactive: + print(f"⚠️ GitHub token authentication failed: {error_msg}") + logger.debug(f"GitHub token auth failed: {response.status_code} - {error_msg}") + return None + + except requests.exceptions.RequestException as e: + if interactive: + print(f"⚠️ Network error during GitHub token authentication: {e}") + logger.debug(f"Network error during GitHub token auth: {e}") + return None + + +def authenticate_with_device_flow(api_origin: str, interactive: bool = True): + """ + Authenticate using OAuth 2.0 Device Flow. + + Args: + api_origin: ChipFlow API origin URL + interactive: Whether to show interactive messages + + Returns: + API key on success, raises AuthenticationError on failure + """ + if interactive: + print("\n🌐 Starting device flow authentication...") + + try: + # Step 1: Initiate device flow + response = requests.post(f"{api_origin}/auth/device/init", timeout=10) + response.raise_for_status() + data = response.json() + + device_code = data["device_code"] + user_code = data["user_code"] + verification_uri = data["verification_uri"] + interval = data["interval"] + expires_in = data["expires_in"] + + # Step 2: Display instructions + if interactive: + print(f"\n📋 To authenticate, please visit:\n {verification_uri}\n") + print(f" And enter this code:\n {user_code}\n") + print("⏳ Waiting for authorization...") + + # Try to open browser + try: + import webbrowser + webbrowser.open(verification_uri) + except Exception: + pass # Silently fail if browser opening doesn't work + + # Step 3: Poll for authorization + max_attempts = expires_in // interval + for attempt in range(max_attempts): + time.sleep(interval) + + try: + poll_response = requests.post( + f"{api_origin}/auth/device/poll", + json={"device_code": device_code}, + timeout=10 + ) + + if poll_response.status_code == 200: + # Success! + api_key = poll_response.json()["api_key"] + save_api_key(api_key) + if interactive: + print("\n✅ Authentication successful!") + return api_key + + elif poll_response.status_code == 202: + # Still pending + if interactive and sys.stdout.isatty(): + print(".", end="", flush=True) + continue + + else: + # Error + error = poll_response.json() + error_desc = error.get("error_description", "Unknown error") + raise AuthenticationError(f"Device flow failed: {error_desc}") + + except requests.exceptions.RequestException as e: + logger.debug(f"Poll request failed: {e}") + continue + + raise AuthenticationError("Device flow authentication timed out") + + except requests.exceptions.RequestException as e: + raise AuthenticationError(f"Network error during device flow: {e}") + + +def get_api_key(api_origin: str | None = None, interactive: bool = True, force_login: bool = False): + """ + Get API key using the following priority: + 1. CHIPFLOW_API_KEY environment variable + 2. Saved credentials file (unless force_login is True) + 3. GitHub CLI token authentication + 4. Device flow authentication + + Args: + api_origin: ChipFlow API origin URL (defaults to CHIPFLOW_API_ORIGIN env var or production) + interactive: Whether to show interactive messages and prompts + force_login: Force re-authentication even if credentials exist + + Returns: + API key string + + Raises: + AuthenticationError: If all authentication methods fail + """ + if api_origin is None: + api_origin = os.environ.get("CHIPFLOW_API_ORIGIN", "https://build.chipflow.org") + + # Method 1: Check environment variable + api_key = os.environ.get("CHIPFLOW_API_KEY") + if api_key: + logger.debug("Using API key from CHIPFLOW_API_KEY environment variable") + return api_key + + # Check for deprecated env var + api_key = os.environ.get("CHIPFLOW_API_KEY_SECRET") + if api_key: + if interactive: + print("⚠️ CHIPFLOW_API_KEY_SECRET is deprecated. Please use CHIPFLOW_API_KEY instead.") + logger.warning("Using deprecated CHIPFLOW_API_KEY_SECRET environment variable") + return api_key + + # Method 2: Check saved credentials (unless force_login) + if not force_login: + api_key = load_saved_api_key() + if api_key: + logger.debug("Using saved API key from credentials file") + return api_key + + # Method 3: Try GitHub CLI token authentication + api_key = authenticate_with_github_token(api_origin, interactive=interactive) + if api_key: + return api_key + + # Method 4: Fall back to device flow + if interactive: + print("\n💡 GitHub CLI not available. Using device flow authentication...") + + try: + return authenticate_with_device_flow(api_origin, interactive=interactive) + except AuthenticationError as e: + raise AuthenticationError( + f"All authentication methods failed. {e}\n\n" + "Please either:\n" + " 1. Set CHIPFLOW_API_KEY environment variable\n" + " 2. Install and authenticate with GitHub CLI: gh auth login\n" + " 3. Complete the device flow authorization" + ) + + +def logout(): + """Remove saved credentials.""" + creds_file = get_credentials_file() + if creds_file.exists(): + creds_file.unlink() + print(f"✅ Logged out. Credentials removed from {creds_file}") + else: + print("ℹ️ No saved credentials found") diff --git a/chipflow/auth_command.py b/chipflow/auth_command.py new file mode 100644 index 00000000..8b16dbc5 --- /dev/null +++ b/chipflow/auth_command.py @@ -0,0 +1,79 @@ +# SPDX-License-Identifier: BSD-2-Clause + +""" +ChipFlow authentication command for CLI. + +Provides `chipflow login` and `chipflow logout` commands. +""" + +import sys +from .auth import get_api_key, logout as auth_logout, AuthenticationError +from .utils import ChipFlowError + + +class AuthCommand: + """Authentication management for ChipFlow.""" + + def __init__(self, config): + """Initialize the auth command. + + Args: + config: ChipFlow configuration object + """ + self.config = config + + def build_cli_parser(self, parser): + """Build CLI argument parser for auth command.""" + subparsers = parser.add_subparsers(dest="action", required=True) + + # Login command + login_parser = subparsers.add_parser( + "login", + help="Authenticate with ChipFlow API" + ) + login_parser.add_argument( + "--force", + action="store_true", + help="Force re-authentication even if already logged in" + ) + + # Logout command + subparsers.add_parser( + "logout", + help="Remove saved credentials" + ) + + def run_cli(self, args): + """Execute the auth command based on parsed arguments.""" + if args.action == "login": + self._login(force=args.force) + elif args.action == "logout": + self._logout() + else: + raise ChipFlowError(f"Unknown auth action: {args.action}") + + def _login(self, force=False): + """Perform login/authentication.""" + import os + + api_origin = os.environ.get("CHIPFLOW_API_ORIGIN", "https://build.chipflow.org") + + print(f"🔐 Authenticating with ChipFlow API ({api_origin})...") + + try: + api_key = get_api_key( + api_origin=api_origin, + interactive=True, + force_login=force + ) + print("\n✅ Successfully authenticated!") + print(f" API key: {api_key[:20]}...") + print("\n💡 You can now use `chipflow silicon submit` to submit designs") + + except AuthenticationError as e: + print(f"\n❌ Authentication failed: {e}") + sys.exit(1) + + def _logout(self): + """Perform logout.""" + auth_logout() diff --git a/chipflow/cli.py b/chipflow/cli.py index 99b139c0..6bcc0a3f 100644 --- a/chipflow/cli.py +++ b/chipflow/cli.py @@ -15,6 +15,7 @@ _parse_config, ) from .packaging import PinCommand +from .auth_command import AuthCommand class UnexpectedError(ChipFlowError): pass @@ -34,6 +35,7 @@ def run(argv=sys.argv[1:]): commands = {} commands["pin"] = PinCommand(config) + commands["auth"] = AuthCommand(config) if config.chipflow.steps: steps = DEFAULT_STEPS |config.chipflow.steps diff --git a/chipflow/platform/silicon_step.py b/chipflow/platform/silicon_step.py index d0df7ce3..252640d7 100644 --- a/chipflow/platform/silicon_step.py +++ b/chipflow/platform/silicon_step.py @@ -23,6 +23,7 @@ from ..utils import top_components from .silicon import SiliconPlatform from ..utils import ChipFlowError +from ..auth import get_api_key, AuthenticationError logger = logging.getLogger(__name__) @@ -114,23 +115,19 @@ def submit(self, rtlil_path, args): --wait: Wait until build has completed. Use '-v' to increase level of verbosity --log-file : Log full debug output to file """ + chipflow_api_origin = os.environ.get("CHIPFLOW_API_ORIGIN", "https://build.chipflow.org") + if not args.dry_run: - # Check for CHIPFLOW_API_KEY_SECRET or CHIPFLOW_API_KEY - if not os.environ.get("CHIPFLOW_API_KEY") and not os.environ.get("CHIPFLOW_API_KEY_SECRET"): - raise ChipFlowError( - "Environment variable `CHIPFLOW_API_KEY` must be set to submit a design." - ) - # Log a deprecation warning if CHIPFLOW_API_KEY_SECRET is used - if os.environ.get("CHIPFLOW_API_KEY_SECRET"): - logger.warning( - "Environment variable `CHIPFLOW_API_KEY_SECRET` is deprecated. " - "Please migrate to using `CHIPFLOW_API_KEY` instead." - ) - self._chipflow_api_key = os.environ.get("CHIPFLOW_API_KEY") or os.environ.get("CHIPFLOW_API_KEY_SECRET") - if self._chipflow_api_key is None: - raise ChipFlowError( - "Environment variable `CHIPFLOW_API_KEY` is empty." + # Get API key using the new authentication helper + # This will try env var first, then saved credentials, then gh token, then device flow + try: + interactive = sys.stdout.isatty() + self._chipflow_api_key = get_api_key( + api_origin=chipflow_api_origin, + interactive=interactive ) + except AuthenticationError as e: + raise ChipFlowError(str(e)) if not sys.stdout.isatty(): interval = 5000 # lets not animate.. else: @@ -193,7 +190,6 @@ def network_err(e): fh.close() exit(1) - chipflow_api_origin = os.environ.get("CHIPFLOW_API_ORIGIN", "https://build.chipflow.org") build_submit_url = f"{chipflow_api_origin}/build/submit" sp.info(f"> Submitting {submission_name} for project {self.config.chipflow.project_name} to ChipFlow Cloud {chipflow_api_origin}") diff --git a/docs/chipflow-commands.rst b/docs/chipflow-commands.rst index fc7aadca..c53afe59 100644 --- a/docs/chipflow-commands.rst +++ b/docs/chipflow-commands.rst @@ -6,6 +6,40 @@ It implements several subcommands, which can be customised or added to in the `` .. _chipflow_toml: chipflow-toml-guide.rst +``chipflow auth`` +----------------- + +The ``chipflow auth`` command manages authentication with the ChipFlow API. + +``chipflow auth login`` +~~~~~~~~~~~~~~~~~~~~~~~ + +Authenticate with the ChipFlow API using one of the following methods (tried in order): + +1. **GitHub CLI token** (recommended) - If you have ``gh`` installed and authenticated, this method is instant +2. **Device flow** - Opens your browser to authorize via GitHub OAuth + +Your API key will be saved locally in ``~/.config/chipflow/credentials`` for future use. + +Options: + +- ``--force``: Force re-authentication even if already logged in + +Examples:: + + # Authenticate (will auto-detect best method) + chipflow auth login + + # Force re-authentication + chipflow auth login --force + +``chipflow auth logout`` +~~~~~~~~~~~~~~~~~~~~~~~~ + +Remove saved credentials from your system:: + + chipflow auth logout + ``chipflow pin lock`` --------------------- @@ -21,7 +55,14 @@ The ``chipflow silicon`` subcommand is used to send the design to the cloud buil - ``chipflow silicon prepare`` links the design, including all Amaranth modules and any external Verilog components, into a single RTLIL file that is ready to be sent to the cloud builder. - ``chipflow silicon submit`` sends the linked design along with the ``pins.lock`` file containing pinout information to the ChipFlow cloud service for the build. With the ``--dry-run`` argument, it can be used for a local test that the design is ready to be submitted. -Submitting the design to the cloud requires the ``CHIPFLOW_API_KEY`` environment variable to be set with a valid API key obtained from the cloud interface. +Authentication for submission: + +1. If you've run ``chipflow auth login``, your saved credentials will be used automatically +2. If ``CHIPFLOW_API_KEY`` environment variable is set, it will be used +3. Otherwise, if ``gh`` (GitHub CLI) is installed and authenticated, it will authenticate automatically +4. As a last resort, you'll be prompted to complete device flow authentication + +Most users should simply run ``chipflow auth login`` once and authentication will be automatic for all future submissions. ``chipflow sim`` ---------------- diff --git a/docs/getting-started.rst b/docs/getting-started.rst index 8514b663..a9f832ea 100644 --- a/docs/getting-started.rst +++ b/docs/getting-started.rst @@ -73,30 +73,58 @@ directory and then run: pdm lock -d pdm install -Set up the environment ----------------------- +Set up authentication +--------------------- -Generate your API key by going to https://build.chipflow.org/ and logging in with your GitHub account. +ChipFlow supports multiple authentication methods. Choose the one that works best for you: -Click on the 'User' menu, then on ‘Create/Refresh API Key’ Your new API key will appear at the -top. +Method 1: Using the CLI (Recommended) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +The easiest way to authenticate is using the ``chipflow auth login`` command: + +:: + + pdm run chipflow auth login + +This will automatically: + +1. Check if you have the GitHub CLI (``gh``) installed and authenticated +2. If yes, instantly authenticate using your GitHub token +3. If no, guide you through the device flow where you'll authorize via your browser + +Your API key will be saved locally for future use. + +Method 2: Manual API key (Alternative) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +If you prefer to manually manage your API key: + +1. Go to https://build.chipflow.org/ and log in with your GitHub account +2. Click on the 'User' menu, then on 'Create/Refresh API Key' +3. Your new API key will appear at the top .. figure:: _assets/api-key.png :alt: Image showing a newly generated API Key Image showing a newly generated API Key -.. warning: +.. warning:: Copy it now, as you will not see it again! -Next, create a file called ``.env`` at the top level in the -``chipflow-examples`` directory, containing the line below, substituting -your key from the page above: +4. Create a file called ``.env`` at the top level in the + ``chipflow-examples`` directory, containing: :: CHIPFLOW_API_KEY= +To log out and remove saved credentials: + +:: + + pdm run chipflow auth logout + Running a chip build -------------------- diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 00000000..478e5022 --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,351 @@ +# SPDX-License-Identifier: BSD-2-Clause + +import unittest +import json +import tempfile +from pathlib import Path +from unittest import mock + +from chipflow.auth import ( + get_api_key, + authenticate_with_github_token, + authenticate_with_device_flow, + is_gh_authenticated, + get_gh_token, + save_api_key, + load_saved_api_key, + logout, + AuthenticationError, +) + + +class TestAuthHelpers(unittest.TestCase): + """Test helper functions in auth module""" + + @mock.patch('chipflow.auth.subprocess.run') + def test_is_gh_authenticated_success(self, mock_run): + """Test checking if gh is authenticated - success case""" + mock_run.return_value.returncode = 0 + self.assertTrue(is_gh_authenticated()) + mock_run.assert_called_once() + + @mock.patch('chipflow.auth.subprocess.run') + def test_is_gh_authenticated_not_authenticated(self, mock_run): + """Test checking if gh is authenticated - not authenticated""" + mock_run.return_value.returncode = 1 + self.assertFalse(is_gh_authenticated()) + + @mock.patch('chipflow.auth.subprocess.run') + def test_is_gh_authenticated_not_installed(self, mock_run): + """Test checking if gh is authenticated - not installed""" + mock_run.side_effect = FileNotFoundError() + self.assertFalse(is_gh_authenticated()) + + @mock.patch('chipflow.auth.subprocess.run') + def test_get_gh_token_success(self, mock_run): + """Test getting GitHub token - success""" + mock_run.return_value.stdout = "ghp_test123\n" + token = get_gh_token() + self.assertEqual(token, "ghp_test123") + + @mock.patch('chipflow.auth.subprocess.run') + def test_get_gh_token_failure(self, mock_run): + """Test getting GitHub token - failure""" + mock_run.side_effect = FileNotFoundError() + token = get_gh_token() + self.assertIsNone(token) + + def test_save_and_load_api_key(self): + """Test saving and loading API key""" + with tempfile.TemporaryDirectory() as tmpdir: + # Mock the credentials file path + with mock.patch('chipflow.auth.get_credentials_file') as mock_creds_file: + creds_file = Path(tmpdir) / "credentials" + mock_creds_file.return_value = creds_file + + # Save API key + test_key = "cf_test_12345" + save_api_key(test_key) + + # Verify file exists and has correct permissions + self.assertTrue(creds_file.exists()) + # Note: File permissions check skipped as it's platform-specific + + # Load API key + loaded_key = load_saved_api_key() + self.assertEqual(loaded_key, test_key) + + def test_load_api_key_no_file(self): + """Test loading API key when file doesn't exist""" + with mock.patch('chipflow.auth.get_credentials_file') as mock_creds_file: + mock_creds_file.return_value = Path("/nonexistent/credentials") + key = load_saved_api_key() + self.assertIsNone(key) + + def test_logout(self): + """Test logout removes credentials file""" + with tempfile.TemporaryDirectory() as tmpdir: + with mock.patch('chipflow.auth.get_credentials_file') as mock_creds_file: + creds_file = Path(tmpdir) / "credentials" + mock_creds_file.return_value = creds_file + + # Create a credentials file + creds_file.write_text(json.dumps({"api_key": "test"})) + self.assertTrue(creds_file.exists()) + + # Logout + with mock.patch('builtins.print'): + logout() + + # File should be deleted + self.assertFalse(creds_file.exists()) + + +class TestGitHubTokenAuth(unittest.TestCase): + """Test GitHub token authentication""" + + @mock.patch('chipflow.auth.save_api_key') + @mock.patch('chipflow.auth.requests.post') + @mock.patch('chipflow.auth.get_gh_token') + @mock.patch('chipflow.auth.is_gh_authenticated') + @mock.patch('builtins.print') + def test_github_token_auth_success( + self, mock_print, mock_is_gh, mock_get_token, mock_post, mock_save + ): + """Test successful GitHub token authentication""" + mock_is_gh.return_value = True + mock_get_token.return_value = "ghp_test123" + mock_post.return_value.status_code = 200 + mock_post.return_value.json.return_value = {"api_key": "cf_test_key"} + + api_key = authenticate_with_github_token("https://test.api", interactive=True) + + self.assertEqual(api_key, "cf_test_key") + mock_save.assert_called_once_with("cf_test_key") + mock_post.assert_called_once() + + @mock.patch('chipflow.auth.is_gh_authenticated') + @mock.patch('builtins.print') + def test_github_token_auth_not_authenticated(self, mock_print, mock_is_gh): + """Test GitHub token auth when gh not authenticated""" + mock_is_gh.return_value = False + + api_key = authenticate_with_github_token("https://test.api", interactive=True) + + self.assertIsNone(api_key) + + @mock.patch('chipflow.auth.requests.post') + @mock.patch('chipflow.auth.get_gh_token') + @mock.patch('chipflow.auth.is_gh_authenticated') + @mock.patch('builtins.print') + def test_github_token_auth_invalid_token( + self, mock_print, mock_is_gh, mock_get_token, mock_post + ): + """Test GitHub token auth with invalid token""" + mock_is_gh.return_value = True + mock_get_token.return_value = "invalid_token" + mock_post.return_value.status_code = 401 + mock_post.return_value.json.return_value = { + "error": "invalid_token", + "error_description": "Invalid GitHub token" + } + + api_key = authenticate_with_github_token("https://test.api", interactive=True) + + self.assertIsNone(api_key) + + @mock.patch('chipflow.auth.requests.post') + @mock.patch('chipflow.auth.get_gh_token') + @mock.patch('chipflow.auth.is_gh_authenticated') + @mock.patch('builtins.print') + def test_github_token_auth_network_error( + self, mock_print, mock_is_gh, mock_get_token, mock_post + ): + """Test GitHub token auth with network error""" + import requests + mock_is_gh.return_value = True + mock_get_token.return_value = "ghp_test123" + mock_post.side_effect = requests.exceptions.ConnectionError("Network error") + + api_key = authenticate_with_github_token("https://test.api", interactive=True) + + self.assertIsNone(api_key) + + +class TestDeviceFlowAuth(unittest.TestCase): + """Test device flow authentication""" + + @mock.patch('chipflow.auth.save_api_key') + @mock.patch('chipflow.auth.time.sleep') + @mock.patch('chipflow.auth.requests.post') + @mock.patch('builtins.print') + @mock.patch('webbrowser.open') + def test_device_flow_success(self, mock_browser, mock_print, mock_post, mock_sleep, mock_save): + """Test successful device flow authentication""" + # Mock init response + init_response = mock.Mock() + init_response.status_code = 200 + init_response.json.return_value = { + "device_code": "device123", + "user_code": "ABCD-1234", + "verification_uri": "https://test.api/auth", + "interval": 1, + "expires_in": 60 + } + + # Mock poll response - success on first try + poll_response = mock.Mock() + poll_response.status_code = 200 + poll_response.json.return_value = {"api_key": "cf_test_key"} + + mock_post.side_effect = [init_response, poll_response] + + api_key = authenticate_with_device_flow("https://test.api", interactive=True) + + self.assertEqual(api_key, "cf_test_key") + mock_save.assert_called_once_with("cf_test_key") + + @mock.patch('chipflow.auth.time.sleep') + @mock.patch('chipflow.auth.requests.post') + @mock.patch('builtins.print') + @mock.patch('webbrowser.open') + def test_device_flow_pending_then_success(self, mock_browser, mock_print, mock_post, mock_sleep): + """Test device flow with pending state then success""" + # Mock init response + init_response = mock.Mock() + init_response.status_code = 200 + init_response.json.return_value = { + "device_code": "device123", + "user_code": "ABCD-1234", + "verification_uri": "https://test.api/auth", + "interval": 1, + "expires_in": 60 + } + + # Mock poll responses - pending, then success + pending_response = mock.Mock() + pending_response.status_code = 202 + pending_response.json.return_value = { + "error": "authorization_pending" + } + + success_response = mock.Mock() + success_response.status_code = 200 + success_response.json.return_value = {"api_key": "cf_test_key"} + + mock_post.side_effect = [init_response, pending_response, success_response] + + with mock.patch('chipflow.auth.save_api_key'): + api_key = authenticate_with_device_flow("https://test.api", interactive=True) + + self.assertEqual(api_key, "cf_test_key") + + @mock.patch('chipflow.auth.time.sleep') + @mock.patch('chipflow.auth.requests.post') + @mock.patch('builtins.print') + @mock.patch('webbrowser.open') + def test_device_flow_timeout(self, mock_browser, mock_print, mock_post, mock_sleep): + """Test device flow timeout""" + # Mock init response + init_response = mock.Mock() + init_response.status_code = 200 + init_response.json.return_value = { + "device_code": "device123", + "user_code": "ABCD-1234", + "verification_uri": "https://test.api/auth", + "interval": 1, + "expires_in": 2 # Very short timeout + } + + # Mock poll response - always pending + pending_response = mock.Mock() + pending_response.status_code = 202 + pending_response.json.return_value = { + "error": "authorization_pending" + } + + mock_post.side_effect = [init_response, pending_response, pending_response, pending_response] + + with self.assertRaises(AuthenticationError) as ctx: + authenticate_with_device_flow("https://test.api", interactive=True) + + self.assertIn("timed out", str(ctx.exception)) + + +class TestGetAPIKey(unittest.TestCase): + """Test the main get_api_key function with fallback logic""" + + def test_get_api_key_from_env_var(self): + """Test getting API key from environment variable""" + with mock.patch.dict('os.environ', {'CHIPFLOW_API_KEY': 'env_key'}): + api_key = get_api_key(interactive=False) + self.assertEqual(api_key, 'env_key') + + @mock.patch('chipflow.auth.load_saved_api_key') + def test_get_api_key_from_saved_credentials(self, mock_load): + """Test getting API key from saved credentials""" + mock_load.return_value = "saved_key" + with mock.patch.dict('os.environ', {}, clear=True): + api_key = get_api_key(interactive=False) + self.assertEqual(api_key, "saved_key") + + @mock.patch('chipflow.auth.authenticate_with_github_token') + @mock.patch('chipflow.auth.load_saved_api_key') + def test_get_api_key_gh_token_fallback(self, mock_load, mock_gh_auth): + """Test fallback to GitHub token authentication""" + mock_load.return_value = None + mock_gh_auth.return_value = "gh_key" + + with mock.patch.dict('os.environ', {}, clear=True): + api_key = get_api_key(interactive=True) + self.assertEqual(api_key, "gh_key") + + @mock.patch('chipflow.auth.authenticate_with_device_flow') + @mock.patch('chipflow.auth.authenticate_with_github_token') + @mock.patch('chipflow.auth.load_saved_api_key') + def test_get_api_key_device_flow_fallback( + self, mock_load, mock_gh_auth, mock_device_flow + ): + """Test fallback to device flow authentication""" + mock_load.return_value = None + mock_gh_auth.return_value = None + mock_device_flow.return_value = "device_key" + + with mock.patch.dict('os.environ', {}, clear=True): + with mock.patch('builtins.print'): + api_key = get_api_key(interactive=True) + self.assertEqual(api_key, "device_key") + + @mock.patch('chipflow.auth.authenticate_with_device_flow') + @mock.patch('chipflow.auth.authenticate_with_github_token') + @mock.patch('chipflow.auth.load_saved_api_key') + def test_get_api_key_all_methods_fail( + self, mock_load, mock_gh_auth, mock_device_flow + ): + """Test when all authentication methods fail""" + mock_load.return_value = None + mock_gh_auth.return_value = None + mock_device_flow.side_effect = AuthenticationError("All methods failed") + + with mock.patch.dict('os.environ', {}, clear=True): + with mock.patch('builtins.print'): + with self.assertRaises(AuthenticationError) as ctx: + get_api_key(interactive=True) + self.assertIn("All authentication methods failed", str(ctx.exception)) + + @mock.patch('chipflow.auth.load_saved_api_key') + def test_get_api_key_force_login_ignores_saved(self, mock_load): + """Test force_login parameter ignores saved credentials""" + mock_load.return_value = "saved_key" + + with mock.patch.dict('os.environ', {}, clear=True): + with mock.patch('chipflow.auth.authenticate_with_github_token') as mock_gh: + mock_gh.return_value = "new_key" + api_key = get_api_key(interactive=True, force_login=True) + self.assertEqual(api_key, "new_key") + # Should not have called load_saved_api_key due to force_login + mock_load.assert_not_called() + + +if __name__ == "__main__": + unittest.main()