diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..f55030b --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,44 @@ +name: Tests + +on: + push: + branches: + - '**' + pull_request: + branches: + - main + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: "pip" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" + + - name: Lint with ruff + run: | + ruff check src/ tests/ + ruff format --check src/ tests/ + + - name: Type check with ty + run: ty check src/ + + - name: Run tests with coverage + run: pytest --cov=authsome --cov-report=xml --cov-report=term -p no:xdist + + - name: Upload coverage reports to Codecov + uses: codecov/codecov-action@v4 + with: + token: ${{ secrets.CODECOV_TOKEN }} + file: ./coverage.xml + fail_ci_if_error: false diff --git a/README.md b/README.md index 92aaba5..3ad75fd 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,8 @@ [![Python 3.13+](https://img.shields.io/pypi/pyversions/authsome.svg)](https://pypi.org/project/authsome/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) [![PyPI downloads](https://img.shields.io/pypi/dm/authsome.svg)](https://pypi.org/project/authsome/) +[![Tests](https://github.com/manojbajaj95/authsome/actions/workflows/test.yml/badge.svg)](https://github.com/manojbajaj95/authsome/actions/workflows/test.yml) +[![codecov](https://codecov.io/gh/manojbajaj95/authsome/branch/main/graph/badge.svg)](https://codecov.io/gh/manojbajaj95/authsome) ```text __ __ diff --git a/pyproject.toml b/pyproject.toml index 1365995..8eedb5c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ dev = [ "pytest>=7.0", "pytest-cov>=4.0", "ruff>=0.9", + "ty", ] [project.scripts] diff --git a/src/authsome/client.py b/src/authsome/client.py index d4093bb..6268fce 100644 --- a/src/authsome/client.py +++ b/src/authsome/client.py @@ -28,7 +28,6 @@ from authsome.crypto.base import CryptoBackend from authsome.crypto.keyring_crypto import KeyringCryptoBackend from authsome.crypto.local_file_crypto import LocalFileCryptoBackend from authsome.errors import ( - ProviderNotFoundError, AuthsomeError, ConnectionNotFoundError, CredentialMissingError, @@ -342,9 +341,11 @@ class AuthClient: # Fetch client credentials client_record = self.get_provider_client_credentials(provider, profile_name) - + flow_client_id = client_record.client_id if client_record else None - flow_client_secret = self.crypto.decrypt(client_record.client_secret) if client_record and client_record.client_secret else None + flow_client_secret = ( + self.crypto.decrypt(client_record.client_secret) if client_record and client_record.client_secret else None + ) flow_api_key = None # Secure Bridge Prompts for missing interactive inputs @@ -358,19 +359,27 @@ class AuthClient: } ) missing_fields.append({"name": "client_id", "label": "Client ID", "type": "text"}) - missing_fields.append({"name": "client_secret", "label": "Client Secret (Optional)", "type": "password", "required": False}) + missing_fields.append( + { + "name": "client_secret", + "label": "Client Secret (Optional)", + "type": "password", + "required": False, + } + ) elif flow_type == FlowType.API_KEY and not flow_api_key: missing_fields.append({"name": "api_key", "label": "API Key", "type": "password"}) if missing_fields: from authsome.flows.bridge import secure_input_bridge + title = f"{definition.display_name} Credentials" inputs = secure_input_bridge(title, missing_fields) - + if flow_type in (FlowType.PKCE, FlowType.DEVICE_CODE): flow_client_id = inputs.get("client_id") secret_input = inputs.get("client_secret") - + if client_record is None: client_record = ProviderClientRecord( profile=profile_name, @@ -403,14 +412,15 @@ class AuthClient: provider=provider, ) client_record.client_id = record.metadata.pop("_dcr_client_id") - + dcr_secret_dict = record.metadata.pop("_dcr_client_secret", None) if dcr_secret_dict: from authsome.crypto.base import EncryptedField + client_record.client_secret = EncryptedField(**dcr_secret_dict) else: client_record.client_secret = None - + self._save_provider_client_credentials(client_record) # Persist the connection record @@ -565,10 +575,10 @@ class AuthClient: """ profile_name = profile or self.active_profile store = self._get_store(profile_name) - + # Ensure provider exists self.get_provider(provider) - + # 1. Log out of all connections for this provider to remotely revoke them meta_key = build_store_key( profile=profile_name, @@ -581,10 +591,10 @@ class AuthClient: # Make a copy of the list as logout will modify it for conn_name in list(metadata.connection_names): self.logout(provider, connection=conn_name, profile=profile_name) - + # 2. Delete ProviderMetadataRecord store.delete(meta_key) - + # 3. Delete ProviderClientRecord client_key = build_store_key( profile=profile_name, @@ -592,7 +602,7 @@ class AuthClient: record_type="client", ) store.delete(client_key) - + logger.info("Revoked all credentials for provider=%s in profile=%s", provider, profile_name) def remove( @@ -605,10 +615,10 @@ class AuthClient: Removes all local credential state and deletes the provider JSON file if it exists locally. """ profile_name = profile or self.active_profile - + # Revoke all credentials (cleans up connections and client secrets) self.revoke(provider, profile=profile_name) - + # Delete the provider definition JSON if it's a local/custom provider local_path = self._home / "providers" / f"{provider}.json" if local_path.exists(): @@ -687,7 +697,7 @@ class AuthClient: if "=" in line: key, value = line.split("=", 1) env[key] = value - + def _dquote(s: str) -> str: """Double-quote a token so $VAR references are expanded by the shell.""" s = s.replace("\\", "\\\\") @@ -964,7 +974,9 @@ class AuthClient: if record.refresh_token: try: refreshed = self._refresh_token(record, provider) - return self.crypto.decrypt(refreshed.access_token) # type: ignore[arg-type] + if refreshed.access_token is None: + raise RefreshFailedError("Refreshed record missing access token", provider=provider) + return self.crypto.decrypt(refreshed.access_token) except RefreshFailedError: # Check if the token hasn't actually expired yet if now < record.expires_at: @@ -1008,7 +1020,7 @@ class AuthClient: client_record = self.get_provider_client_credentials(provider_name, record.profile) client_id = None client_secret = None - + if client_record: client_id = client_record.client_id if client_record.client_secret: diff --git a/src/authsome/flows/api_key.py b/src/authsome/flows/api_key.py index 5380673..8325174 100644 --- a/src/authsome/flows/api_key.py +++ b/src/authsome/flows/api_key.py @@ -6,7 +6,6 @@ Spec §13.5: API Key Environment Import — read from environment variable, stor from __future__ import annotations - import logging from authsome.crypto.base import CryptoBackend @@ -82,5 +81,3 @@ class ApiKeyFlow(AuthFlow): account=AccountInfo(), metadata={}, ) - - diff --git a/src/authsome/flows/bridge.py b/src/authsome/flows/bridge.py index e4b0f4d..4b570e8 100644 --- a/src/authsome/flows/bridge.py +++ b/src/authsome/flows/bridge.py @@ -5,17 +5,18 @@ web browser instead of the terminal. This prevents secrets from being exposed in environments where agents or scripts might intercept standard I/O. """ -from html import escape import http.server import logging import socket import threading import urllib.parse import webbrowser +from html import escape from typing import Any logger = logging.getLogger(__name__) + def _find_free_port() -> int: """Find a free TCP port on localhost.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: @@ -25,11 +26,11 @@ def _find_free_port() -> int: class _BridgeHandler(http.server.BaseHTTPRequestHandler): """HTTP handler that renders a form and captures input.""" - + title: str = "Secure Input" fields: list[dict[str, Any]] = [] result: dict[str, str] | None = None - + def do_GET(self) -> None: """Serve the HTML form.""" html = [ @@ -38,18 +39,20 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): "", "", f"

{self.title}

", - "
" + "", ] - + for field in self.fields: label = field["label"] if field.get("type") == "static": @@ -60,7 +63,7 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): "
" f"" "
" ) continue @@ -69,10 +72,10 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): required = "required" if field.get("required", True) else "" html.append(f"") html.append(f"") - + html.append("") html.append("
") - + self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() @@ -83,10 +86,10 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): content_length = int(self.headers.get("Content-Length", 0)) post_data = self.rfile.read(content_length).decode("utf-8") parsed = urllib.parse.parse_qs(post_data) - + # Flatten the parse_qs output (which returns lists) _BridgeHandler.result = {k: v[0] for k, v in parsed.items() if v} - + # Send success page self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") @@ -98,10 +101,11 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): "" ) self.wfile.write(success_html.encode("utf-8")) - + # Notify that we are done def kill_server(): self.server.shutdown() + threading.Thread(target=kill_server, daemon=True).start() def log_message(self, format: str, *args: Any) -> None: @@ -111,35 +115,35 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): def secure_input_bridge(title: str, fields: list[dict[str, Any]]) -> dict[str, str]: """ Start a local server and open the browser to collect sensitive inputs securely. - + Args: title: The heading to display on the form. fields: A list of dicts, each representing an input field. Expected keys: 'name', 'label', 'type' (default: text), 'required' (default: True). - + Returns: A dictionary mapping field names to user input values. """ port = _find_free_port() - + # Reset state _BridgeHandler.title = title _BridgeHandler.fields = fields _BridgeHandler.result = None - + server = http.server.HTTPServer(("127.0.0.1", port), _BridgeHandler) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() - + url = f"http://127.0.0.1:{port}" print(f"\nRequires secure input for: {title}") print(f"Opening browser for secure input...\nIf the browser doesn't open, visit:\n{url}\n") webbrowser.open(url) - + # Wait for the server to shutdown (triggered by do_POST) - server_thread.join(timeout=300) # 5 minute timeout - + server_thread.join(timeout=300) # 5 minute timeout + if _BridgeHandler.result is None: raise RuntimeError("Secure input bridge timed out or was cancelled.") - + return _BridgeHandler.result diff --git a/src/authsome/flows/dcr_pkce.py b/src/authsome/flows/dcr_pkce.py index cf79921..ecc679c 100644 --- a/src/authsome/flows/dcr_pkce.py +++ b/src/authsome/flows/dcr_pkce.py @@ -109,6 +109,8 @@ class DcrPkceFlow(AuthFlow): DCR registration endpoint is discovered via .well-known if not provided. """ + callback_port: int = 7999 + def authenticate( self, provider: ProviderDefinition, @@ -139,7 +141,7 @@ class DcrPkceFlow(AuthFlow): code_verifier, code_challenge = _generate_pkce() # Start local callback server - port = 7999 + port = self.callback_port redirect_uri = f"http://127.0.0.1:{port}/callback" # Reset handler state @@ -151,30 +153,32 @@ class DcrPkceFlow(AuthFlow): server_thread = threading.Thread(target=server.handle_request, daemon=True) server_thread.start() - # Build authorization URL - state = secrets.token_urlsafe(32) - auth_params: dict[str, str] = { - "response_type": "code", - "client_id": client_id, - "redirect_uri": redirect_uri, - "state": state, - "code_challenge": code_challenge, - "code_challenge_method": "S256", - } - if effective_scopes: - auth_params["scope"] = " ".join(effective_scopes) + try: + # Build authorization URL + state = secrets.token_urlsafe(32) + auth_params: dict[str, str] = { + "response_type": "code", + "client_id": client_id, + "redirect_uri": redirect_uri, + "state": state, + "code_challenge": code_challenge, + "code_challenge_method": "S256", + } + if effective_scopes: + auth_params["scope"] = " ".join(effective_scopes) - auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}" + auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}" - logger.info("Opening browser for authorization...") - logger.debug("Authorization URL: %s", auth_url) - print(f"\nOpening browser for {provider.display_name} authorization...") - print(f"If the browser doesn't open, visit:\n{auth_url}\n") - webbrowser.open(auth_url) + logger.info("Opening browser for authorization...") + logger.debug("Authorization URL: %s", auth_url) + print(f"\nOpening browser for {provider.display_name} authorization...") + print(f"If the browser doesn't open, visit:\n{auth_url}\n") + webbrowser.open(auth_url) - # Wait for callback - server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS) - server.server_close() + # Wait for callback + server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS) + finally: + server.server_close() if _CallbackHandler.error: raise AuthenticationFailedError( diff --git a/src/authsome/flows/pkce.py b/src/authsome/flows/pkce.py index 826ef33..efd0b08 100644 --- a/src/authsome/flows/pkce.py +++ b/src/authsome/flows/pkce.py @@ -104,6 +104,8 @@ class PkceFlow(AuthFlow): `client` section (either literal or env:-prefixed). """ + callback_port: int = 7999 + def authenticate( self, provider: ProviderDefinition, @@ -135,7 +137,7 @@ class PkceFlow(AuthFlow): code_verifier, code_challenge = _generate_pkce() # Start local callback server - port = 7999 + port = self.callback_port redirect_uri = f"http://127.0.0.1:{port}/callback" # Reset handler state @@ -147,29 +149,31 @@ class PkceFlow(AuthFlow): server_thread = threading.Thread(target=server.handle_request, daemon=True) server_thread.start() - # Build authorization URL - state = secrets.token_urlsafe(32) - auth_params: dict[str, str] = { - "response_type": "code", - "client_id": client_id, - "redirect_uri": redirect_uri, - "state": state, - "code_challenge": code_challenge, - "code_challenge_method": "S256", - } - if effective_scopes: - auth_params["scope"] = " ".join(effective_scopes) + try: + # Build authorization URL + state = secrets.token_urlsafe(32) + auth_params: dict[str, str] = { + "response_type": "code", + "client_id": client_id, + "redirect_uri": redirect_uri, + "state": state, + "code_challenge": code_challenge, + "code_challenge_method": "S256", + } + if effective_scopes: + auth_params["scope"] = " ".join(effective_scopes) - auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}" + auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}" - logger.info("Opening browser for authorization...") - print(f"\nOpening browser for {provider.display_name} authorization...") - print(f"If the browser doesn't open, visit:\n{auth_url}\n") - webbrowser.open(auth_url) + logger.info("Opening browser for authorization...") + print(f"\nOpening browser for {provider.display_name} authorization...") + print(f"If the browser doesn't open, visit:\n{auth_url}\n") + webbrowser.open(auth_url) - # Wait for callback - server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS) - server.server_close() + # Wait for callback + server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS) + finally: + server.server_close() if _CallbackHandler.error: raise AuthenticationFailedError( diff --git a/src/authsome/models/connection.py b/src/authsome/models/connection.py index 5cbb134..e812b0a 100644 --- a/src/authsome/models/connection.py +++ b/src/authsome/models/connection.py @@ -117,4 +117,3 @@ class ProviderClientRecord(BaseModel): metadata: dict[str, Any] = Field(default_factory=dict) model_config = {"extra": "allow"} - diff --git a/tests/test_bridge.py b/tests/test_bridge.py new file mode 100644 index 0000000..adfd6ba --- /dev/null +++ b/tests/test_bridge.py @@ -0,0 +1,65 @@ +"""Tests for the secure browser bridge.""" + +import urllib.parse +import urllib.request +from unittest.mock import patch + +import pytest + +from authsome.flows.bridge import _BridgeHandler, _find_free_port, secure_input_bridge + + +def test_find_free_port(): + port = _find_free_port() + assert isinstance(port, int) + assert port > 0 + + +def test_secure_input_bridge_success(): + fields = [ + {"name": "api_key", "label": "API Key", "type": "password"}, + {"name": "username", "label": "Username", "required": False}, + ] + + def mock_open(url): + # 1. Test GET + req = urllib.request.Request(url) + with urllib.request.urlopen(req) as response: + assert response.status == 200 + html = response.read().decode("utf-8") + assert "Test Auth" in html + assert "API Key" in html + assert "Username" in html + + # 2. Test POST + data = urllib.parse.urlencode({"api_key": "secret123", "username": "testuser"}).encode("utf-8") + req_post = urllib.request.Request(url, data=data, method="POST") + with urllib.request.urlopen(req_post) as response: + assert response.status == 200 + html = response.read().decode("utf-8") + assert "Success!" in html + + with patch("authsome.flows.bridge.webbrowser.open", side_effect=mock_open): + res = secure_input_bridge("Test Auth", fields) + + assert res == {"api_key": "secret123", "username": "testuser"} + + +def test_secure_input_bridge_timeout(): + fields = [{"name": "key", "label": "Key"}] + + def mock_open(url): + # do nothing, simulate user ignoring the browser + pass + + with patch("authsome.flows.bridge.webbrowser.open", side_effect=mock_open): + # To avoid waiting 300s, we mock Thread.join + with patch("threading.Thread.join"): + with pytest.raises(RuntimeError, match="timed out or was cancelled"): + secure_input_bridge("Test Auth", fields) + + +def test_log_message(): + # Test _BridgeHandler.log_message does not crash + handler = _BridgeHandler.__new__(_BridgeHandler) + handler.log_message("test %s", "arg") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..940dbb1 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,404 @@ +"""Tests for the Authsome CLI.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from authsome.cli import cli +from authsome.errors import AuthsomeError, ProviderNotFoundError, StoreUnavailableError + + +@pytest.fixture +def runner(): + return CliRunner() + + +@pytest.fixture +def mock_client(): + with patch("authsome.cli.AuthClient") as mock: + client = MagicMock() + mock.return_value = client + yield client + + +def test_init_command(runner, mock_client): + mock_client.home = "/mock/home" + result = runner.invoke(cli, ["init"]) + assert result.exit_code == 0 + assert "Initialized authsome at /mock/home" in result.output + mock_client.init.assert_called_once() + + +def test_init_json(runner, mock_client): + mock_client.home = "/mock/home" + result = runner.invoke(cli, ["init", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data["status"] == "initialized" + + +def test_list_command(runner, mock_client): + mock_client.list_connections.return_value = [ + { + "name": "openai", + "connections": [ + { + "connection_name": "default", + "status": "connected", + "auth_type": "api_key", + "scopes": ["read"], + "expires_at": "2030", + } + ], + } + ] + + mock_provider = MagicMock() + mock_provider.name = "openai" + mock_provider.display_name = "OpenAI" + mock_provider.auth_type.value = "api_key" + mock_client.list_providers_by_source.return_value = {"bundled": [mock_provider], "custom": []} + + result = runner.invoke(cli, ["list"]) + assert result.exit_code == 0 + assert "Bundled Providers:" in result.output + assert "OpenAI" in result.output + assert "connected" in result.output + + +def test_list_command_no_connections_and_missing_scopes(runner, mock_client): + mock_client.list_connections.return_value = [ + { + "name": "openai", + "connections": [{"connection_name": "default", "status": "connected", "auth_type": "api_key"}], + } + ] + + mock_provider = MagicMock() + mock_provider.name = "openai" + mock_provider.display_name = "OpenAI" + mock_provider.auth_type.value = "api_key" + + mock_provider_empty = MagicMock() + mock_provider_empty.name = "github" + mock_provider_empty.display_name = "GitHub" + mock_provider_empty.auth_type.value = "oauth2" + + mock_client.list_providers_by_source.return_value = {"bundled": [mock_provider, mock_provider_empty], "custom": []} + + result = runner.invoke(cli, ["list"]) + assert result.exit_code == 0 + assert "(no connections)" in result.output + + +def test_list_json(runner, mock_client): + mock_client.list_connections.return_value = [] + mock_client.list_providers_by_source.return_value = {"bundled": [], "custom": []} + result = runner.invoke(cli, ["list", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert "bundled" in data + + +def test_login_command(runner, mock_client): + mock_record = MagicMock() + mock_record.status.value = "connected" + mock_client.login.return_value = mock_record + + result = runner.invoke(cli, ["login", "openai"]) + assert result.exit_code == 0 + assert "Successfully logged in to openai" in result.output + + +def test_login_json(runner, mock_client): + mock_record = MagicMock() + mock_record.status.value = "connected" + mock_client.login.return_value = mock_record + + result = runner.invoke(cli, ["login", "openai", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data["status"] == "success" + + +def test_login_error(runner, mock_client): + mock_client.login.side_effect = AuthsomeError("Already exists") + result = runner.invoke(cli, ["login", "openai"]) + assert result.exit_code == 1 + assert "Error: Already exists" in result.output + + +def test_error_mapping(runner, mock_client): + from authsome.errors import AuthenticationFailedError, CredentialMissingError, RefreshFailedError + + mock_client.login.side_effect = ProviderNotFoundError("test") + result = runner.invoke(cli, ["login", "test"]) + assert result.exit_code == 3 + + mock_client.login.side_effect = StoreUnavailableError("locked") + result = runner.invoke(cli, ["login", "test"]) + assert result.exit_code == 7 + + mock_client.login.side_effect = AuthenticationFailedError("fail") + result = runner.invoke(cli, ["login", "test"]) + assert result.exit_code == 4 + + mock_client.login.side_effect = CredentialMissingError("missing", provider="test") + result = runner.invoke(cli, ["login", "test"]) + assert result.exit_code == 5 + + mock_client.login.side_effect = RefreshFailedError("refresh fail", provider="test") + result = runner.invoke(cli, ["login", "test"]) + assert result.exit_code == 6 + + mock_client.login.side_effect = Exception("unknown") + result = runner.invoke(cli, ["login", "test"]) + assert result.exit_code == 1 + + +def test_logout(runner, mock_client): + result = runner.invoke(cli, ["logout", "openai"]) + assert result.exit_code == 0 + mock_client.logout.assert_called_with("openai", "default") + + +def test_logout_json(runner, mock_client): + result = runner.invoke(cli, ["logout", "openai", "--json"]) + assert result.exit_code == 0 + + +def test_revoke(runner, mock_client): + result = runner.invoke(cli, ["revoke", "openai"]) + assert result.exit_code == 0 + mock_client.revoke.assert_called_with("openai") + + +def test_revoke_json(runner, mock_client): + result = runner.invoke(cli, ["revoke", "openai", "--json"]) + assert result.exit_code == 0 + + +def test_remove(runner, mock_client): + result = runner.invoke(cli, ["remove", "openai"]) + assert result.exit_code == 0 + mock_client.remove.assert_called_with("openai") + + +def test_remove_json(runner, mock_client): + result = runner.invoke(cli, ["remove", "openai", "--json"]) + assert result.exit_code == 0 + + +def test_get(runner, mock_client): + mock_record = MagicMock() + mock_record.model_dump.return_value = {"status": "connected", "access_token": "secret"} + mock_client.crypto.decrypt.return_value = "decrypted_secret" + mock_client.get_connection.return_value = mock_record + + result = runner.invoke(cli, ["get", "openai"]) + assert result.exit_code == 0 + assert "***REDACTED***" in result.output + + +def test_get_show_secret(runner, mock_client): + mock_record = MagicMock() + mock_record.model_dump.return_value = {"status": "connected", "access_token": "secret"} + mock_record.access_token = "encrypted_secret" + mock_client.crypto.decrypt.return_value = "decrypted_secret" + mock_client.get_connection.return_value = mock_record + + result = runner.invoke(cli, ["get", "openai", "--show-secret"]) + assert result.exit_code == 0 + assert "decrypted_secret" in result.output + + +def test_get_field(runner, mock_client): + mock_record = MagicMock() + mock_record.model_dump.return_value = {"status": "connected"} + mock_client.get_connection.return_value = mock_record + + result = runner.invoke(cli, ["get", "openai", "--field", "status"]) + assert result.exit_code == 0 + assert "connected" in result.output + + +def test_get_missing_field(runner, mock_client): + mock_record = MagicMock() + mock_record.model_dump.return_value = {"status": "connected"} + mock_client.get_connection.return_value = mock_record + + result = runner.invoke(cli, ["get", "openai", "--field", "missing"]) + assert result.exit_code == 1 + + +def test_get_json(runner, mock_client): + mock_record = MagicMock() + mock_record.model_dump.return_value = {"status": "connected"} + mock_client.get_connection.return_value = mock_record + result = runner.invoke(cli, ["get", "openai", "--json"]) + assert result.exit_code == 0 + + +def test_get_field_json(runner, mock_client): + mock_record = MagicMock() + mock_record.model_dump.return_value = {"status": "connected"} + mock_client.get_connection.return_value = mock_record + result = runner.invoke(cli, ["get", "openai", "--json", "--field", "status"]) + assert result.exit_code == 0 + + +def test_inspect(runner, mock_client): + mock_def = MagicMock() + mock_def.model_dump.return_value = {"name": "openai"} + mock_client.get_provider.return_value = mock_def + + result = runner.invoke(cli, ["inspect", "openai"]) + assert result.exit_code == 0 + assert '"name": "openai"' in result.output + + +def test_inspect_json(runner, mock_client): + mock_def = MagicMock() + mock_def.model_dump.return_value = {"name": "openai"} + mock_client.get_provider.return_value = mock_def + + result = runner.invoke(cli, ["inspect", "openai", "--json"]) + assert result.exit_code == 0 + + +def test_export(runner, mock_client): + mock_client.export.return_value = "export VAR=1" + result = runner.invoke(cli, ["export", "openai", "--format", "shell"]) + assert result.exit_code == 0 + assert "export VAR=1" in result.output + + # test empty output + mock_client.export.return_value = "" + result2 = runner.invoke(cli, ["export", "openai", "--format", "shell"]) + assert result2.exit_code == 0 + assert result2.output == "" + + +def test_run(runner, mock_client): + mock_run_result = MagicMock() + mock_run_result.returncode = 0 + mock_client.run.return_value = mock_run_result + + result = runner.invoke(cli, ["run", "--provider", "openai", "echo", "hello"]) + assert result.exit_code == 0 + mock_client.run.assert_called_with(["echo", "hello"], providers=["openai"]) + + +def test_register_file_not_found(runner): + result = runner.invoke(cli, ["register", "nonexistent.json"]) + assert result.exit_code == 1 + assert "File not found" in result.output + + +def test_register_success(runner, mock_client, tmp_path): + f = tmp_path / "test.json" + f.write_text( + json.dumps( + { + "name": "test", + "display_name": "Test", + "auth_type": "api_key", + "flow": "api_key", + "api_key": {"header_name": "Auth"}, + } + ) + ) + + result = runner.invoke(cli, ["register", str(f)]) + assert result.exit_code == 0 + mock_client.register_provider.assert_called_once() + + +def test_register_success_json(runner, mock_client, tmp_path): + f = tmp_path / "test.json" + f.write_text( + json.dumps( + { + "name": "test", + "display_name": "Test", + "auth_type": "api_key", + "flow": "api_key", + "api_key": {"header_name": "Auth"}, + } + ) + ) + + result = runner.invoke(cli, ["register", str(f), "--json"]) + assert result.exit_code == 0 + + +def test_register_bad_json(runner, mock_client, tmp_path): + f = tmp_path / "bad.json" + f.write_text("{bad") + result = runner.invoke(cli, ["register", str(f)]) + assert result.exit_code == 1 + + +def test_whoami(runner, mock_client): + mock_client.home = "/mock/home" + mock_client.config.encryption.mode = "keyring" + result = runner.invoke(cli, ["whoami"]) + assert result.exit_code == 0 + assert "/mock/home" in result.output + + +def test_whoami_no_encryption_config(runner, mock_client): + mock_client.home = "/mock/home" + mock_client.config.encryption = None + result = runner.invoke(cli, ["whoami", "--json"]) + assert result.exit_code == 0 + + +def test_doctor(runner, mock_client): + mock_client.doctor.return_value = { + "home_exists": True, + "encryption": False, + "issues": ["error"], + "providers_count": 0, + } + result = runner.invoke(cli, ["doctor"]) + assert result.exit_code == 1 + assert "FAIL" in result.output + + +def test_doctor_json(runner, mock_client): + mock_client.doctor.return_value = {"home_exists": True, "encryption": True, "issues": []} + result = runner.invoke(cli, ["doctor", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data["home_exists"] is True + + +def test_doctor_all_ok(runner, mock_client): + mock_client.doctor.return_value = {"home_exists": True, "encryption": True, "issues": [], "providers_count": 1} + result = runner.invoke(cli, ["doctor"]) + assert result.exit_code == 0 + assert "OK" in result.output + + +def test_common_options_error_handling_json(runner, mock_client): + mock_client.login.side_effect = AuthsomeError("Error") + result = runner.invoke(cli, ["login", "openai", "--json"]) + assert result.exit_code == 1 + data = json.loads(result.output) + assert data["error"] == "AuthsomeError" + + +def test_echo_no_color(runner, mock_client): + mock_client.logout.return_value = None + result = runner.invoke(cli, ["logout", "openai", "--no-color"]) + assert result.exit_code == 0 + + +def test_echo_quiet(runner, mock_client): + mock_client.logout.return_value = None + result = runner.invoke(cli, ["logout", "openai", "--quiet"]) + assert result.exit_code == 0 + assert "Logged out" not in result.output diff --git a/tests/test_client.py b/tests/test_client.py index 72e6590..e9db009 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,19 +1,39 @@ """Tests for the AuthClient core.""" import json +from datetime import timedelta from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest +import requests from authsome.client import AuthClient from authsome.errors import ( + AuthsomeError, ConnectionNotFoundError, + CredentialMissingError, ProfileNotFoundError, ProviderNotFoundError, + RefreshFailedError, + TokenExpiredError, + UnsupportedFlowError, +) +from authsome.models.connection import ( + ConnectionRecord, + ProviderClientRecord, ) from authsome.models.enums import AuthType, ConnectionStatus, ExportFormat, FlowType -from authsome.models.provider import ApiKeyConfig, ProviderDefinition +from authsome.models.provider import ApiKeyConfig, OAuthConfig, ProviderDefinition +from authsome.utils import utc_now + + +@pytest.fixture +def client(tmp_path: Path) -> AuthClient: + home = tmp_path / ".authsome" + with AuthClient(home=home) as c: + c.init() + yield c class TestAuthClientInit: @@ -66,17 +86,32 @@ class TestAuthClientInit: client.init() # Should not fail assert (home / "version").exists() + def test_authclient_default_home(self, monkeypatch): + monkeypatch.delenv("AUTHSOME_HOME", raising=False) + with patch("authsome.client.Path.home", return_value=Path("/mock/home")): + with AuthClient() as c: + assert str(c.home) == "/mock/home/.authsome" + + def test_authclient_keyring_crypto(self, tmp_path, monkeypatch): + home = tmp_path / ".authsome" + home.mkdir() + (home / "config.json").write_text('{"encryption": {"mode": "keyring"}}') + with patch("authsome.client.KeyringCryptoBackend") as mock_backend: + with AuthClient(home=home) as c: + _ = c.crypto + mock_backend.assert_called_once() + + def test_load_config_bad_json(self, tmp_path): + home = tmp_path / ".authsome" + home.mkdir() + (home / "config.json").write_text("{bad") + with AuthClient(home=home) as c: + assert c.config.default_profile == "default" + class TestAuthClientProviders: """Provider operations tests.""" - @pytest.fixture - def client(self, tmp_path: Path) -> AuthClient: - home = tmp_path / ".authsome" - c = AuthClient(home=home) - c.init() - return c - def test_list_providers_includes_bundled(self, client: AuthClient) -> None: providers = client.list_providers() names = [p.name for p in providers] @@ -97,24 +132,31 @@ class TestAuthClientProviders: name="custom", display_name="Custom Provider", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, api_key=ApiKeyConfig(env_var="CUSTOM_KEY"), ) client.register_provider(custom) loaded = client.get_provider("custom") assert loaded.display_name == "Custom Provider" + def test_list_providers_by_source(self, client: AuthClient): + provider = ProviderDefinition( + name="testlocal", + display_name="Test Local", + auth_type=AuthType.API_KEY, + flow=FlowType.API_KEY, + api_key={"header_name": "Authorization"}, + ) + client.register_provider(provider) + sources = client.list_providers_by_source() + assert "bundled" in sources + assert "custom" in sources + assert len(sources["custom"]) > 0 + class TestAuthClientProfiles: """Profile management tests.""" - @pytest.fixture - def client(self, tmp_path: Path) -> AuthClient: - home = tmp_path / ".authsome" - c = AuthClient(home=home) - c.init() - return c - def test_default_profile_created(self, client: AuthClient) -> None: profiles = client.list_profiles() names = [p.name for p in profiles] @@ -135,123 +177,647 @@ class TestAuthClientProfiles: with pytest.raises(ProfileNotFoundError): client.set_default_profile("nonexistent") + def test_list_profiles_issues(self, client: AuthClient): + profiles_dir = client.home / "profiles" + bad_dir = profiles_dir / "bad" + bad_dir.mkdir() + (bad_dir / "metadata.json").write_text("{bad json") -class TestAuthClientApiKeyLogin: - """API key login integration tests.""" + profiles = client.list_profiles() + names = [p.name for p in profiles] + assert "default" in names + assert "bad" not in names - @pytest.fixture - def client(self, tmp_path: Path) -> AuthClient: - home = tmp_path / ".authsome" - c = AuthClient(home=home) - c.init() - return c + def test_list_profiles_no_dir(self, client: AuthClient): + import shutil + + profiles_dir = client.home / "profiles" + shutil.rmtree(profiles_dir) + assert client.list_profiles() == [] + + +class TestAuthClientLogin: + """Authentication flow integration tests.""" def test_api_key_login_and_get(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-123"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-test-123"}): record = client.login("openai") assert record.status == ConnectionStatus.CONNECTED assert record.auth_type == AuthType.API_KEY - # Get connection conn = client.get_connection("openai") assert conn.status == ConnectionStatus.CONNECTED + def test_login_connection_exists(self, client: AuthClient): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-1"}): + client.login("openai", "default") + + with pytest.raises(AuthsomeError, match="already exists"): + client.login("openai", "default", force=False) + + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-2"}): + client.login("openai", "default", force=True) + + def test_login_unsupported_flow(self, client: AuthClient): + def mock_get_provider(name): + mock_def = MagicMock() + mock_def.flow = MagicMock(value="invalid_flow") + return mock_def + + with patch.object(client, "get_provider", mock_get_provider): + with pytest.raises(UnsupportedFlowError): + client.login("test") + + def test_login_oauth_bridge_prompt(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://auth", token_url="http://token"), + ) + client.register_provider(provider) + + mock_record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("access"), + ) + + with patch("authsome.client._FLOW_HANDLERS") as handlers: + mock_handler = MagicMock() + mock_handler.authenticate.return_value = mock_record + handlers.get.return_value = lambda: mock_handler + + with patch( + "authsome.flows.bridge.secure_input_bridge", + return_value={"client_id": "cid", "client_secret": "csec"}, + ) as mock_bridge: + client.login("testoauth") + mock_bridge.assert_called_once() + + creds = client.get_provider_client_credentials("testoauth", "default") + assert creds.client_id == "cid" + assert client.crypto.decrypt(creds.client_secret) == "csec" + + def test_login_dcr_metadata_extraction(self, client: AuthClient): + provider = ProviderDefinition( + name="testdcr", + display_name="Test DCR", + auth_type=AuthType.OAUTH2, + flow=FlowType.DCR_PKCE, + oauth=OAuthConfig(authorization_url="http://auth", token_url="http://token"), + ) + client.register_provider(provider) + + encrypted_secret = client.crypto.encrypt("csec") + mock_record = ConnectionRecord( + schema_version=1, + provider="testdcr", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("access"), + metadata={"_dcr_client_id": "cid", "_dcr_client_secret": encrypted_secret.model_dump()}, + ) + + with patch("authsome.client._FLOW_HANDLERS") as handlers: + mock_handler = MagicMock() + mock_handler.authenticate.return_value = mock_record + handlers.get.return_value = lambda: mock_handler + + client.login("testdcr") + + creds = client.get_provider_client_credentials("testdcr", "default") + assert creds.client_id == "cid" + assert client.crypto.decrypt(creds.client_secret) == "csec" + + def test_login_dcr_metadata_extraction_no_secret(self, client: AuthClient): + provider = ProviderDefinition( + name="testdcr2", + display_name="Test DCR 2", + auth_type=AuthType.OAUTH2, + flow=FlowType.DCR_PKCE, + oauth=OAuthConfig(authorization_url="http://auth", token_url="http://token"), + ) + client.register_provider(provider) + + mock_record = ConnectionRecord( + schema_version=1, + provider="testdcr2", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("access"), + metadata={"_dcr_client_id": "cid"}, + ) + + with patch("authsome.client._FLOW_HANDLERS") as handlers: + mock_handler = MagicMock() + mock_handler.authenticate.return_value = mock_record + handlers.get.return_value = lambda: mock_handler + + client.login("testdcr2") + + creds = client.get_provider_client_credentials("testdcr2", "default") + assert creds.client_id == "cid" + assert creds.client_secret is None + + +class TestAuthClientCredentials: + """Credential retrieval tests.""" + def test_api_key_get_access_token(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-456"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-test-456"}): client.login("openai") token = client.get_access_token("openai") assert token == "sk-test-456" def test_api_key_get_auth_headers(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-789"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-test-789"}): client.login("openai") headers = client.get_auth_headers("openai") assert "Authorization" in headers assert headers["Authorization"] == "Bearer sk-test-789" - def test_api_key_multiple_connections(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="key-1"): - client.login("openai", connection_name="personal") + def test_get_access_token_oauth(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) - with patch("authsome.flows.api_key.getpass.getpass", return_value="key-2"): - client.login("openai", connection_name="work") + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("token123"), + ) + client._save_connection(record) - assert client.get_access_token("openai", connection="personal") == "key-1" - assert client.get_access_token("openai", connection="work") == "key-2" + assert client.get_access_token("testoauth") == "token123" - def test_get_nonexistent_connection(self, client: AuthClient) -> None: + def test_get_access_token_unsupported(self, client: AuthClient): + mock_record = MagicMock() + mock_record.auth_type = "UNKNOWN" + + with patch.object(client, "get_connection", return_value=mock_record): + with pytest.raises(CredentialMissingError, match="Unsupported auth type"): + client.get_access_token("test") + + def test_get_auth_headers_unsupported(self, client: AuthClient): + mock_record = MagicMock() + mock_record.auth_type = "UNKNOWN" + + with patch.object(client, "get_provider"): + with patch.object(client, "get_connection", return_value=mock_record): + with pytest.raises(CredentialMissingError, match="Cannot build headers"): + client.get_auth_headers("test") + + def test_get_auth_headers_api_key_custom(self, client: AuthClient): + provider = ProviderDefinition( + name="testapi", + display_name="Test API", + auth_type=AuthType.API_KEY, + flow=FlowType.API_KEY, + api_key={"header_name": "X-API-KEY", "header_prefix": ""}, + ) + client.register_provider(provider) + record = ConnectionRecord( + schema_version=1, + provider="testapi", + profile="default", + connection_name="default", + auth_type=AuthType.API_KEY, + status=ConnectionStatus.CONNECTED, + api_key=client.crypto.encrypt("key123"), + ) + client._save_connection(record) + + headers = client.get_auth_headers("testapi") + assert headers["X-API-KEY"] == "key123" + + def test_get_auth_headers_oauth(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("oauth123"), + ) + client._save_connection(record) + + headers = client.get_auth_headers("testoauth") + assert headers["Authorization"] == "Bearer oauth123" + + def test_get_auth_headers_api_key_no_config(self, client: AuthClient): + provider = ProviderDefinition( + name="testapi", + display_name="Test API", + auth_type=AuthType.API_KEY, + flow=FlowType.API_KEY, + ) + record = ConnectionRecord( + schema_version=1, + provider="testapi", + profile="default", + connection_name="default", + auth_type=AuthType.API_KEY, + status=ConnectionStatus.CONNECTED, + api_key=client.crypto.encrypt("key123"), + ) + client._save_connection(record) + + with patch.object(client, "get_provider", return_value=provider): + headers = client.get_auth_headers("testapi") + assert headers["Authorization"] == "Bearer key123" + + def test_get_api_key_missing(self, client: AuthClient): + provider = ProviderDefinition( + name="test", + display_name="Test", + auth_type=AuthType.API_KEY, + flow=FlowType.API_KEY, + api_key={"header_name": "Authorization"}, + ) + client.register_provider(provider) + record = ConnectionRecord( + schema_version=1, + provider="test", + profile="default", + connection_name="default", + auth_type=AuthType.API_KEY, + status=ConnectionStatus.CONNECTED, + ) + client._save_connection(record) + with pytest.raises(CredentialMissingError, match="No API key stored"): + client._get_api_key(record) + + +class TestAuthClientTokenRefresh: + """OAuth token refresh logic tests.""" + + def test_oauth_token_refresh(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + + now = utc_now() + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("old_acc"), + refresh_token=client.crypto.encrypt("ref"), + expires_at=now - timedelta(seconds=10), # expired + ) + client._save_connection(record) + + with pytest.raises(RefreshFailedError, match="No client_id"): + client.get_access_token("testoauth") + + client._save_provider_client_credentials( + ProviderClientRecord( + profile="default", + provider="testoauth", + client_id="cid", + client_secret=client.crypto.encrypt("sec"), + ) + ) + + mock_token_resp = MagicMock() + mock_token_resp.json.return_value = {"access_token": "new_acc", "refresh_token": "new_ref", "expires_in": 3600} + + with patch("authsome.client.http_client.post", return_value=mock_token_resp): + assert client.get_access_token("testoauth") == "new_acc" + + record.expires_at = now - timedelta(seconds=10) + client._save_connection(record) + with patch("authsome.client.http_client.post", side_effect=requests.RequestException("boom")): + with pytest.raises(RefreshFailedError): + client.get_access_token("testoauth") + + def test_oauth_token_no_refresh_expired(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + + now = utc_now() + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("old_acc"), + expires_at=now - timedelta(seconds=10), + ) + client._save_connection(record) + + with pytest.raises(TokenExpiredError): + client.get_access_token("testoauth") + + def test_oauth_token_refresh_failed_but_still_valid(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + + now = utc_now() + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("valid_acc"), + refresh_token=client.crypto.encrypt("ref"), + expires_at=now + timedelta(seconds=100), + ) + client._save_connection(record) + + client._save_provider_client_credentials( + ProviderClientRecord(profile="default", provider="testoauth", client_id="cid") + ) + + with patch("authsome.client.http_client.post", side_effect=requests.RequestException("boom")): + assert client.get_access_token("testoauth") == "valid_acc" + + def test_oauth_token_missing_access_token(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + ) + client._save_connection(record) + with pytest.raises(CredentialMissingError, match="No access token stored"): + client.get_access_token("testoauth") + + def test_get_access_token_valid_not_near_expiry(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + now = utc_now() + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("valid_token"), + expires_at=now + timedelta(seconds=1000), + ) + client._save_connection(record) + assert client.get_access_token("testoauth") == "valid_token" + + def test_get_access_token_valid_no_refresh(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) + now = utc_now() + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("valid_token_no_ref"), + expires_at=now + timedelta(seconds=100), + ) + client._save_connection(record) + assert client.get_access_token("testoauth") == "valid_token_no_ref" + + def test_refresh_token_no_oauth_config(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", display_name="Test OAuth", auth_type=AuthType.OAUTH2, flow=FlowType.PKCE, oauth=None + ) + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + ) + with patch.object(client, "get_provider", return_value=provider): + with pytest.raises(RefreshFailedError, match="No OAuth config"): + client._refresh_token(record, "testoauth") + + def test_refresh_token_no_refresh_token(self, client: AuthClient): + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + refresh_token=None, + ) + with patch.object(client, "get_provider", return_value=provider): + with pytest.raises(RefreshFailedError, match="No refresh token available"): + client._refresh_token(record, "testoauth") + + +class TestAuthClientLifecycle: + """Connection lifecycle tests (logout, remove, revoke).""" + + def test_logout_errors_and_revocation(self, client: AuthClient): + client.logout("openai", "nonexistent") + + provider = ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t", revocation_url="http://revoke"), + ) + client.register_provider(provider) + + record = ConnectionRecord( + schema_version=1, + provider="testoauth", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("token123"), + ) + client._save_connection(record) + client._update_provider_metadata("default", "testoauth", "default") + + with patch("authsome.client.http_client.post") as mock_post: + client.logout("testoauth") + mock_post.assert_called_once_with("http://revoke", data={"token": "token123"}, timeout=15) + + client._save_connection(record) + with patch("authsome.client.http_client.post", side_effect=requests.RequestException("boom")): + client.logout("testoauth") + + def test_remove_connection(self, client: AuthClient) -> None: + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "key"}): + client.login("openai") + + client.remove("openai") with pytest.raises(ConnectionNotFoundError): - client.get_connection("openai", connection="nonexistent") + client.get_connection("openai") + + def test_remove_nonexistent(self, client: AuthClient) -> None: + client.remove("openai") + + def test_revoke_connection(self, client: AuthClient) -> None: + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "key"}): + client.login("openai") + + client.revoke("openai") + with pytest.raises(ConnectionNotFoundError): + client.get_connection("openai") + + def test_remove_bundled_provider(self, client: AuthClient): + client.remove("openai") + assert not (client.home / "providers" / "openai.json").exists() class TestAuthClientExport: """Export operations tests.""" - @pytest.fixture - def client(self, tmp_path: Path) -> AuthClient: - home = tmp_path / ".authsome" - c = AuthClient(home=home) - c.init() - return c - def test_export_env_format(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-export"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-export"}): client.login("openai") output = client.export("openai", format=ExportFormat.ENV) assert "OPENAI_API_KEY=sk-export" in output def test_export_shell_format(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-shell"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-shell"}): client.login("openai") output = client.export("openai", format=ExportFormat.SHELL) assert "export OPENAI_API_KEY=sk-shell" in output def test_export_json_format(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-json"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-json"}): client.login("openai") output = client.export("openai", format=ExportFormat.JSON) data = json.loads(output) assert data["OPENAI_API_KEY"] == "sk-json" + def test_export_oauth_and_unknown(self, client: AuthClient): + provider = ProviderDefinition( + name="testexport", + display_name="Test Export", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"), + ) + client.register_provider(provider) -class TestAuthClientRemoveRevoke: - """Remove and revoke operations tests.""" + record = ConnectionRecord( + schema_version=1, + provider="testexport", + profile="default", + connection_name="default", + auth_type=AuthType.OAUTH2, + status=ConnectionStatus.CONNECTED, + access_token=client.crypto.encrypt("acc"), + refresh_token=client.crypto.encrypt("ref"), + ) + client._save_connection(record) - @pytest.fixture - def client(self, tmp_path: Path) -> AuthClient: - home = tmp_path / ".authsome" - c = AuthClient(home=home) - c.init() - return c + env_out = client.export("testexport", format=ExportFormat.ENV) + assert "TESTEXPORT_ACCESS_TOKEN=acc" in env_out + assert "TESTEXPORT_REFRESH_TOKEN=ref" in env_out + assert client.export("testexport", format=MagicMock()) == "" - def test_remove_connection(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="key"): + +class TestAuthClientRun: + """Run command tests.""" + + def test_run_command(self, client: AuthClient): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-run"}): client.login("openai") - client.remove("openai") - - with pytest.raises(ConnectionNotFoundError): - client.get_connection("openai") - - def test_remove_nonexistent(self, client: AuthClient) -> None: - with pytest.raises(ConnectionNotFoundError): - client.remove("openai") - - def test_revoke_connection(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="key"): - client.login("openai") - - client.revoke("openai") - - conn = client.get_connection("openai") - assert conn.status == ConnectionStatus.REVOKED - assert conn.api_key is None + with patch("authsome.client.subprocess.run") as mock_run: + client.run(["echo", "hello", "world"], providers=["openai"]) + mock_run.assert_called_once() + kwargs = mock_run.call_args[1] + assert kwargs["env"]["OPENAI_API_KEY"] == "sk-run" + assert kwargs["shell"] is True + assert "echo" in mock_run.call_args[0][0] class TestAuthClientDoctor: @@ -264,35 +830,71 @@ class TestAuthClientDoctor: results = client.doctor() assert results["home_exists"] is True - assert results["version_file"] is True - assert results["config_file"] is True - assert results["encryption"] is True - assert results["store"] is True - assert results["providers_count"] > 0 - assert results["profiles_count"] > 0 assert results["issues"] == [] + def test_doctor_issues(self, client: AuthClient, monkeypatch): + with patch.object(client.crypto, "encrypt", side_effect=Exception("crypto boom")): + res = client.doctor() + assert not res["encryption"] + assert "Encryption: crypto boom" in res["issues"] -class TestAuthClientListConnections: - """List connections tests.""" + store = client._get_store("default") + with patch.object(store, "set", side_effect=Exception("store boom")): + res = client.doctor() + assert not res["store"] + assert "Store: store boom" in res["issues"] - @pytest.fixture - def client(self, tmp_path: Path) -> AuthClient: - home = tmp_path / ".authsome" - c = AuthClient(home=home) - c.init() - return c + with patch.object(client, "list_providers", side_effect=Exception("prov boom")): + res = client.doctor() + assert "Providers: prov boom" in res["issues"] + + with patch.object(client, "list_profiles", side_effect=Exception("prof boom")): + res = client.doctor() + assert "Profiles: prof boom" in res["issues"] + + +class TestAuthClientConnections: + """Connection management tests.""" def test_list_connections_empty(self, client: AuthClient) -> None: connections = client.list_connections() assert connections == [] def test_list_connections_after_login(self, client: AuthClient) -> None: - with patch("authsome.flows.api_key.getpass.getpass", return_value="key"): + with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "key"}): client.login("openai") connections = client.list_connections() assert len(connections) == 1 assert connections[0]["name"] == "openai" - assert len(connections[0]["connections"]) == 1 - assert connections[0]["connections"][0]["status"] == "connected" + + def test_list_connections_edge_cases(self, client: AuthClient): + store = client._get_store("default") + store.set("profile:default:openai:junk:default", '{"junk": true}') + store.set("profile:default:openai:connection", '{"junk": true}') + store.set("profile:default:openai:connection:test", "") + + connections = client.list_connections() + assert connections == [] + + def test_get_nonexistent_connection(self, client: AuthClient) -> None: + with pytest.raises(ConnectionNotFoundError): + client.get_connection("openai", connection="nonexistent") + + def test_get_store_missing_profile(self, client: AuthClient): + with pytest.raises(ProfileNotFoundError): + client._get_store("missing") + + def test_metadata_cleanup(self, client: AuthClient): + client._update_provider_metadata("default", "openai", "conn1") + client._update_provider_metadata("default", "openai", "conn2") + + client._remove_from_provider_metadata("default", "openai", "conn1") + store = client._get_store("default") + meta = store.get("profile:default:openai:metadata") + assert "conn1" not in meta + assert "conn2" in meta + + client._remove_from_provider_metadata("default", "openai", "conn2") + meta = json.loads(store.get("profile:default:openai:metadata")) + assert meta["last_used_connection"] is None diff --git a/tests/test_crypto.py b/tests/test_crypto.py index b35878c..0b71490 100644 --- a/tests/test_crypto.py +++ b/tests/test_crypto.py @@ -72,6 +72,58 @@ class TestLocalFileCryptoBackend: decrypted = crypto.decrypt(encrypted) assert decrypted == original + def test_json_load_error(self, tmp_path: Path) -> None: + from authsome.errors import EncryptionUnavailableError + + key_file = tmp_path / "master.key" + key_file.write_text("invalid json") + with pytest.raises(EncryptionUnavailableError, match="Failed to read local key file"): + LocalFileCryptoBackend(tmp_path) + + def test_chmod_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import os + + def mock_chmod(path, mode): + raise OSError("Mock error") + + monkeypatch.setattr(os, "chmod", mock_chmod) + # Should not raise + _ = LocalFileCryptoBackend(tmp_path) + + def test_encrypt_not_initialized(self, crypto: LocalFileCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + crypto._aesgcm = None + with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"): + crypto.encrypt("test") + + def test_decrypt_not_initialized(self, crypto: LocalFileCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + crypto._aesgcm = None + with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"): + crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="a", ciphertext="b", tag="c")) + + def test_decrypt_unsupported_alg(self, crypto: LocalFileCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Unsupported algorithm"): + crypto.decrypt(EncryptedField(enc=1, alg="UNSUPPORTED", kid="local", nonce="a", ciphertext="b", tag="c")) + + def test_decrypt_base64_decode_error(self, crypto: LocalFileCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Failed to decode envelope"): + crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="!@#", ciphertext="b", tag="c")) + + def test_decrypt_aesgcm_error(self, crypto: LocalFileCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Decryption failed"): + crypto.decrypt( + EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="abcd", ciphertext="abcd", tag="abcd") + ) + class TestKeyringCryptoBackend: """OS Keyring crypto backend tests. @@ -100,6 +152,91 @@ class TestKeyringCryptoBackend: assert field.alg == "AES-256-GCM" assert field.kid == "local" + def test_import_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + import sys + + monkeypatch.setitem(sys.modules, "keyring", None) + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="The 'keyring' package is required"): + KeyringCryptoBackend() + + def test_get_password_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + import keyring + + def mock_get(*args, **kwargs): + raise Exception("Mock error") + + monkeypatch.setattr(keyring, "get_password", mock_get) + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Failed to access OS keyring"): + KeyringCryptoBackend() + + def test_generate_new_keyring_key(self, monkeypatch: pytest.MonkeyPatch) -> None: + import keyring + + def mock_get(*args, **kwargs): + return None + + def mock_set(*args, **kwargs): + pass + + monkeypatch.setattr(keyring, "get_password", mock_get) + monkeypatch.setattr(keyring, "set_password", mock_set) + backend = KeyringCryptoBackend() + assert backend._master_key is not None + + def test_set_password_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + import keyring + + def mock_get(*args, **kwargs): + return None + + def mock_set(*args, **kwargs): + raise Exception("Mock error") + + monkeypatch.setattr(keyring, "get_password", mock_get) + monkeypatch.setattr(keyring, "set_password", mock_set) + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Failed to store master key"): + KeyringCryptoBackend() + + def test_encrypt_not_initialized(self, crypto: KeyringCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + crypto._aesgcm = None + with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"): + crypto.encrypt("test") + + def test_decrypt_not_initialized(self, crypto: KeyringCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + crypto._aesgcm = None + with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"): + crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="a", ciphertext="b", tag="c")) + + def test_decrypt_unsupported_alg(self, crypto: KeyringCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Unsupported algorithm"): + crypto.decrypt(EncryptedField(enc=1, alg="UNSUPPORTED", kid="local", nonce="a", ciphertext="b", tag="c")) + + def test_decrypt_base64_decode_error(self, crypto: KeyringCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Failed to decode envelope"): + crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="!@#", ciphertext="b", tag="c")) + + def test_decrypt_aesgcm_error(self, crypto: KeyringCryptoBackend) -> None: + from authsome.errors import EncryptionUnavailableError + + with pytest.raises(EncryptionUnavailableError, match="Decryption failed"): + crypto.decrypt( + EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="abcd", ciphertext="abcd", tag="abcd") + ) + class TestCrossBackendCompatibility: """Verify that both backends produce compatible envelope formats.""" diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 0000000..ce859fe --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,94 @@ +"""Tests for errors.py.""" + +from authsome.errors import ( + AuthenticationFailedError, + AuthsomeError, + ConnectionNotFoundError, + CredentialMissingError, + DiscoveryError, + EncryptionUnavailableError, + InvalidProviderSchemaError, + ProfileNotFoundError, + ProviderNotFoundError, + RefreshFailedError, + StoreUnavailableError, + TokenExpiredError, + UnsupportedAuthTypeError, + UnsupportedFlowError, +) + + +def test_authsome_error_formatting(): + # Test operation present + err = AuthsomeError("message", operation="test_op") + assert str(err) == "(test_op) message" + + # Test provider present + err2 = AuthsomeError("message", provider="github") + assert str(err2) == "[github] message" + + +def test_unsupported_auth_type_error(): + err = UnsupportedAuthTypeError("magic", provider="github") + assert "Unsupported auth type: magic" in str(err) + assert "[github]" in str(err) + + +def test_unsupported_flow_error(): + err = UnsupportedFlowError("unknown", provider="github") + assert "Unsupported flow: unknown" in str(err) + + +def test_credential_missing_error(): + err = CredentialMissingError(provider="github") + assert "Credential not found" in str(err) + + +def test_token_expired_error(): + err = TokenExpiredError(provider="github") + assert "Access token expired" in str(err) + + +def test_refresh_failed_error(): + err = RefreshFailedError(provider="github") + assert "Token refresh failed: Unknown error" in str(err) + + +def test_encryption_unavailable_error(): + err = EncryptionUnavailableError() + assert "Encryption backend unavailable" in str(err) + + +def test_store_unavailable_error(): + err = StoreUnavailableError() + assert "Credential store unavailable" in str(err) + + +def test_discovery_error(): + err = DiscoveryError("timeout", provider="github") + assert "Discovery failed: timeout" in str(err) + + +def test_provider_not_found_error(): + err = ProviderNotFoundError("missing") + assert "Provider 'missing' not found" in str(err) + + +def test_invalid_provider_schema_error(): + err = InvalidProviderSchemaError("bad json", provider="github") + assert "Invalid provider schema: bad json" in str(err) + + +def test_profile_not_found_error(): + err = ProfileNotFoundError("work") + assert "Profile 'work' not found" in str(err) + + +def test_connection_not_found_error(): + err = ConnectionNotFoundError(provider="github", connection="work", profile="default") + assert "Connection 'work' not found for provider 'github' in profile 'default'" in str(err) + + +def test_authentication_failed_error(): + err = AuthenticationFailedError("invalid credentials", provider="github") + assert "Authentication failed: invalid credentials" in str(err) diff --git a/tests/test_flows.py b/tests/test_flows.py index 5258788..7f29be6 100644 --- a/tests/test_flows.py +++ b/tests/test_flows.py @@ -1,13 +1,12 @@ """Tests for authentication flows.""" from pathlib import Path -from unittest.mock import patch import pytest from authsome.crypto.local_file_crypto import LocalFileCryptoBackend -from authsome.errors import AuthenticationFailedError, CredentialMissingError -from authsome.flows.api_key import ApiKeyEnvFlow, ApiKeyPromptFlow +from authsome.errors import AuthenticationFailedError +from authsome.flows.api_key import ApiKeyFlow from authsome.models.enums import AuthType, ConnectionStatus, FlowType from authsome.models.provider import ApiKeyConfig, ProviderDefinition @@ -17,33 +16,32 @@ def _make_api_key_provider() -> ProviderDefinition: name="testapi", display_name="Test API", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, api_key=ApiKeyConfig( header_name="Authorization", header_prefix="Bearer", - env_var="TEST_API_KEY", ), ) -class TestApiKeyPromptFlow: - """API key prompt flow tests.""" +class TestApiKeyFlow: + """API key flow tests.""" @pytest.fixture def crypto(self, tmp_path: Path) -> LocalFileCryptoBackend: return LocalFileCryptoBackend(tmp_path) def test_successful_login(self, crypto: LocalFileCryptoBackend) -> None: - flow = ApiKeyPromptFlow() + flow = ApiKeyFlow() provider = _make_api_key_provider() - with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-key-123"): - record = flow.authenticate( - provider=provider, - crypto=crypto, - profile="default", - connection_name="default", - ) + record = flow.authenticate( + provider=provider, + crypto=crypto, + profile="default", + connection_name="default", + api_key="sk-test-key-123", + ) assert record.provider == "testapi" assert record.profile == "default" @@ -56,38 +54,38 @@ class TestApiKeyPromptFlow: assert decrypted == "sk-test-key-123" def test_empty_key_rejected(self, crypto: LocalFileCryptoBackend) -> None: - flow = ApiKeyPromptFlow() + flow = ApiKeyFlow() provider = _make_api_key_provider() - with patch("authsome.flows.api_key.getpass.getpass", return_value=""): - with pytest.raises(AuthenticationFailedError, match="cannot be empty"): - flow.authenticate( - provider=provider, - crypto=crypto, - profile="default", - connection_name="default", - ) + with pytest.raises(AuthenticationFailedError, match="API key was not provided"): + flow.authenticate( + provider=provider, + crypto=crypto, + profile="default", + connection_name="default", + api_key="", + ) def test_whitespace_only_rejected(self, crypto: LocalFileCryptoBackend) -> None: - flow = ApiKeyPromptFlow() + flow = ApiKeyFlow() provider = _make_api_key_provider() - with patch("authsome.flows.api_key.getpass.getpass", return_value=" "): - with pytest.raises(AuthenticationFailedError, match="cannot be empty"): - flow.authenticate( - provider=provider, - crypto=crypto, - profile="default", - connection_name="default", - ) + with pytest.raises(AuthenticationFailedError, match="cannot be empty"): + flow.authenticate( + provider=provider, + crypto=crypto, + profile="default", + connection_name="default", + api_key=" ", + ) def test_missing_api_key_config(self, crypto: LocalFileCryptoBackend) -> None: - flow = ApiKeyPromptFlow() + flow = ApiKeyFlow() provider = ProviderDefinition( name="noconfig", display_name="No Config", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, ) with pytest.raises(AuthenticationFailedError, match="missing 'api_key'"): flow.authenticate( @@ -95,58 +93,18 @@ class TestApiKeyPromptFlow: crypto=crypto, profile="default", connection_name="default", + api_key="sk-test-key-123", ) - -class TestApiKeyEnvFlow: - """API key env import flow tests.""" - - @pytest.fixture - def crypto(self, tmp_path: Path) -> LocalFileCryptoBackend: - return LocalFileCryptoBackend(tmp_path) - - def test_successful_env_import(self, crypto: LocalFileCryptoBackend, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("TEST_API_KEY", "env-key-value") - flow = ApiKeyEnvFlow() + def test_missing_api_key_parameter(self, crypto: LocalFileCryptoBackend) -> None: + flow = ApiKeyFlow() provider = _make_api_key_provider() - record = flow.authenticate( - provider=provider, - crypto=crypto, - profile="default", - connection_name="default", - ) - - assert record.status == ConnectionStatus.CONNECTED - assert record.api_key is not None - assert crypto.decrypt(record.api_key) == "env-key-value" - - def test_missing_env_var(self, crypto: LocalFileCryptoBackend, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("TEST_API_KEY", raising=False) - flow = ApiKeyEnvFlow() - provider = _make_api_key_provider() - - with pytest.raises(CredentialMissingError, match="not set or empty"): - flow.authenticate( - provider=provider, - crypto=crypto, - profile="default", - connection_name="default", - ) - - def test_no_env_var_defined(self, crypto: LocalFileCryptoBackend) -> None: - flow = ApiKeyEnvFlow() - provider = ProviderDefinition( - name="noenv", - display_name="No Env", - auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_ENV, - api_key=ApiKeyConfig(env_var=None), - ) - with pytest.raises(AuthenticationFailedError, match="does not define an env_var"): + with pytest.raises(AuthenticationFailedError, match="API key was not provided"): flow.authenticate( provider=provider, crypto=crypto, profile="default", connection_name="default", + api_key=None, ) diff --git a/tests/test_flows_dcr_pkce.py b/tests/test_flows_dcr_pkce.py new file mode 100644 index 0000000..a01c6d4 --- /dev/null +++ b/tests/test_flows_dcr_pkce.py @@ -0,0 +1,351 @@ +"""Tests for the DCR PKCE OAuth flow.""" + +import json +import urllib.parse +import urllib.request +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from authsome.crypto.local_file_crypto import LocalFileCryptoBackend +from authsome.errors import AuthenticationFailedError, DiscoveryError +from authsome.flows.dcr_pkce import DcrPkceFlow, _CallbackHandler, _find_free_port, _generate_pkce +from authsome.models.connection import ConnectionStatus +from authsome.models.enums import AuthType, FlowType +from authsome.models.provider import OAuthConfig, ProviderDefinition + + +def _make_provider() -> ProviderDefinition: + return ProviderDefinition( + name="testdcr", + display_name="Test DCR", + auth_type=AuthType.OAUTH2, + flow=FlowType.DCR_PKCE, + oauth=OAuthConfig( + authorization_url="https://auth.example.com/auth", + token_url="https://auth.example.com/token", + ), + ) + + +def test_find_free_port(): + port = _find_free_port() + assert isinstance(port, int) + assert port > 0 + + +def test_generate_pkce(): + verifier, challenge = _generate_pkce() + assert len(verifier) >= 43 + assert len(challenge) >= 43 + + +def test_callback_handler_log(): + handler = _CallbackHandler.__new__(_CallbackHandler) + handler.log_message("test %s", "msg") + + +def test_missing_oauth(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + provider.oauth = None + flow = DcrPkceFlow() + + with pytest.raises(AuthenticationFailedError, match="missing 'oauth' configuration"): + flow.authenticate(provider, crypto, "default", "default") + + +def test_discover_registration_endpoint_missing_oauth(): + flow = DcrPkceFlow() + provider = _make_provider() + provider.oauth = None + with pytest.raises(DiscoveryError, match="No OAuth config"): + flow._discover_registration_endpoint(provider) + + +def test_discover_registration_endpoint_success(): + flow = DcrPkceFlow() + provider = _make_provider() + + mock_resp1 = MagicMock() + mock_resp1.status_code = 200 + mock_resp1.json.return_value = {} + + mock_resp2 = MagicMock() + mock_resp2.status_code = 200 + mock_resp2.json.return_value = {"registration_endpoint": "https://auth.example.com/register"} + + with patch("authsome.flows.dcr_pkce.http_client.get", side_effect=[mock_resp1, mock_resp2]): + endpoint = flow._discover_registration_endpoint(provider) + assert endpoint == "https://auth.example.com/register" + + +def test_discover_registration_endpoint_failure(): + flow = DcrPkceFlow() + provider = _make_provider() + + mock_resp = MagicMock() + mock_resp.status_code = 404 + + with patch("authsome.flows.dcr_pkce.http_client.get", side_effect=[requests.RequestException("boom"), mock_resp]): + with pytest.raises(DiscoveryError, match="Could not discover registration_endpoint"): + flow._discover_registration_endpoint(provider) + + +def test_register_client_missing_oauth(): + flow = DcrPkceFlow() + provider = _make_provider() + provider.oauth = None + with pytest.raises(AuthenticationFailedError, match="No OAuth config"): + flow._register_client(provider, []) + + +def test_register_client_success(): + flow = DcrPkceFlow() + provider = _make_provider() + provider.oauth.registration_endpoint = "https://auth.example.com/register" + + mock_resp = MagicMock() + mock_resp.json.return_value = {"client_id": "new_cid", "client_secret": "new_sec"} + + with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp): + cid, sec = flow._register_client(provider, ["scope1"]) + assert cid == "new_cid" + assert sec == "new_sec" + + +def test_register_client_http_error(): + flow = DcrPkceFlow() + provider = _make_provider() + provider.oauth.registration_endpoint = "https://auth.example.com/register" + + with patch("authsome.flows.dcr_pkce.http_client.post", side_effect=requests.RequestException("boom")): + with pytest.raises(AuthenticationFailedError, match="Registration failed"): + flow._register_client(provider, []) + + +def test_register_client_invalid_json(): + flow = DcrPkceFlow() + provider = _make_provider() + provider.oauth.registration_endpoint = "https://auth.example.com/register" + + mock_resp = MagicMock() + mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0) + + with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="not valid JSON"): + flow._register_client(provider, []) + + +def test_register_client_missing_client_id(): + flow = DcrPkceFlow() + provider = _make_provider() + provider.oauth.registration_endpoint = "https://auth.example.com/register" + + mock_resp = MagicMock() + mock_resp.json.return_value = {"client_secret": "new_sec"} + + with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="missing client_id"): + flow._register_client(provider, []) + + +def test_dcr_pkce_flow_success(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + # Extract state from URL + parsed = urllib.parse.urlparse(url) + params = urllib.parse.parse_qs(parsed.query) + state = params["state"][0] + + # Send callback + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}" + req = urllib.request.Request(callback_url) + with urllib.request.urlopen(req) as resp: + assert resp.status == 200 + + # Mock DCR + dcr_resp = MagicMock() + dcr_resp.json.return_value = {"client_id": "cid", "client_secret": "sec"} + + # Mock Token Exchange + token_resp = MagicMock() + token_resp.json.return_value = { + "access_token": "mock_access", + "refresh_token": "mock_refresh", + "expires_in": 3600, + } + + with patch("authsome.flows.dcr_pkce.http_client.post", side_effect=[dcr_resp, token_resp]): + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.dcr_pkce.DcrPkceFlow._discover_registration_endpoint", return_value="url"): + record = flow.authenticate(provider, crypto, "default", "default", scopes=["test"]) + + assert record.status == ConnectionStatus.CONNECTED + assert record.metadata["_dcr_client_id"] == "cid" + assert record.metadata["_dcr_client_secret"] is not None + + +def test_dcr_pkce_flow_reuse_client(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + params = urllib.parse.parse_qs(parsed.query) + assert "scope" not in params # No scopes passed + state = params["state"][0] + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}" + urllib.request.urlopen(urllib.request.Request(callback_url)) + + token_resp = MagicMock() + token_resp.json.return_value = {"access_token": "mock_access"} + + with patch("authsome.flows.dcr_pkce.http_client.post", return_value=token_resp): + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + # Pass client_id to skip DCR + record = flow.authenticate(provider, crypto, "default", "default", client_id="existing_cid") + + assert record.status == ConnectionStatus.CONNECTED + assert record.metadata["_dcr_client_id"] == "existing_cid" + assert record.metadata["_dcr_client_secret"] is None + assert record.expires_at is None + + +def test_dcr_pkce_flow_callback_error(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + callback_url = f"http://127.0.0.1:{port}/callback?error=access_denied&error_description=User%20denied" + try: + urllib.request.urlopen(urllib.request.Request(callback_url)) + except urllib.error.HTTPError as e: + assert e.code == 400 + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with pytest.raises(AuthenticationFailedError, match="access_denied"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_dcr_pkce_flow_callback_invalid(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + callback_url = f"http://127.0.0.1:{port}/callback?other=123" + try: + urllib.request.urlopen(urllib.request.Request(callback_url)) + except urllib.error.HTTPError as e: + assert e.code == 400 + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with pytest.raises(AuthenticationFailedError, match="no code received"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_dcr_pkce_flow_state_mismatch(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state=wrong" + urllib.request.urlopen(urllib.request.Request(callback_url)) + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with pytest.raises(AuthenticationFailedError, match="state mismatch"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_dcr_pkce_flow_timeout(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + + def mock_open(url): + pass + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.dcr_pkce._CALLBACK_TIMEOUT_SECONDS", 0.01): + with pytest.raises(AuthenticationFailedError, match="timed out"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_exchange_code_http_error(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + state = urllib.parse.parse_qs(parsed.query)["state"][0] + urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}")) + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.dcr_pkce.http_client.post", side_effect=requests.RequestException("boom")): + with pytest.raises(AuthenticationFailedError, match="Token exchange failed"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_exchange_code_invalid_json(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + params = urllib.parse.parse_qs(parsed.query) + state = params["state"][0] + urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}")) + + mock_resp = MagicMock() + mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0) + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="not valid JSON"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_exchange_code_missing_access_token(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DcrPkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + params = urllib.parse.parse_qs(parsed.query) + state = params["state"][0] + urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}")) + + mock_resp = MagicMock() + mock_resp.json.return_value = {"error": "invalid_grant", "error_description": "bad code"} + + with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="bad code"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") diff --git a/tests/test_flows_device_code.py b/tests/test_flows_device_code.py new file mode 100644 index 0000000..9d803de --- /dev/null +++ b/tests/test_flows_device_code.py @@ -0,0 +1,274 @@ +"""Tests for the Device Code OAuth flow.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from authsome.crypto.local_file_crypto import LocalFileCryptoBackend +from authsome.errors import AuthenticationFailedError +from authsome.flows.device_code import DeviceCodeFlow +from authsome.models.connection import ConnectionStatus +from authsome.models.enums import AuthType, FlowType +from authsome.models.provider import OAuthConfig, ProviderDefinition + + +def _make_provider() -> ProviderDefinition: + return ProviderDefinition( + name="testdevice", + display_name="Test Device", + auth_type=AuthType.OAUTH2, + flow=FlowType.DEVICE_CODE, + oauth=OAuthConfig( + authorization_url="https://auth.example.com/auth", + device_authorization_url="https://auth.example.com/device", + token_url="https://auth.example.com/token", + ), + ) + + +def test_missing_oauth(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + provider.oauth = None + flow = DeviceCodeFlow() + + with pytest.raises(AuthenticationFailedError, match="missing 'oauth' configuration"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_missing_device_url(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + provider.oauth.device_authorization_url = None + flow = DeviceCodeFlow() + + with pytest.raises(AuthenticationFailedError, match="not have a device_authorization_url"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_missing_client_id(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + with pytest.raises(AuthenticationFailedError, match="requires a client_id"): + flow.authenticate(provider, crypto, "default", "default") + + +def test_request_device_code_http_error(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + with patch("authsome.flows.device_code.requests.post", side_effect=requests.RequestException("boom")): + with pytest.raises(AuthenticationFailedError, match="Device authorization request failed"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_request_device_code_invalid_json(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + mock_resp = MagicMock() + mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0) + + with patch("authsome.flows.device_code.requests.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="not valid JSON"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_request_device_code_missing_fields(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + mock_resp = MagicMock() + # Missing user_code and verification_uri + mock_resp.json.return_value = {"device_code": "dc123"} + + with patch("authsome.flows.device_code.requests.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="missing required fields"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_poll_for_token_timeout(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + device_resp = MagicMock() + device_resp.json.return_value = { + "device_code": "dc", + "user_code": "uc", + "verification_uri": "http://uri", + "expires_in": 1, + "interval": 1, + } + + with patch("authsome.flows.device_code.requests.post", return_value=device_resp): + with patch("authsome.flows.device_code.time.sleep"): + with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 2]): + # loop runs once and then deadline is past + with pytest.raises(AuthenticationFailedError, match="Device authorization timed out"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_poll_for_token_success_and_errors(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + # We will simulate multiple polling attempts + device_resp = MagicMock() + device_resp.json.return_value = { + "device_code": "dc", + "user_code": "uc", + "verification_uri": "http://uri", + "verification_uri_complete": "http://uri?user_code=uc", + "expires_in": 300, + "interval": 1, + } + + token_json_err = MagicMock() + token_json_err.json.side_effect = json.JSONDecodeError("msg", "doc", 0) + + token_pending = MagicMock() + token_pending.status_code = 400 + token_pending.json.return_value = {"error": "authorization_pending"} + + token_slow_down = MagicMock() + token_slow_down.status_code = 400 + token_slow_down.json.return_value = {"error": "slow_down"} + + token_success = MagicMock() + token_success.status_code = 200 + token_success.json.return_value = {"access_token": "acc", "refresh_token": "ref", "expires_in": 3600} + + mock_post_responses = [ + device_resp, # Phase 1 request + requests.RequestException("boom"), # Poll 1: request exception -> loops + token_json_err, # Poll 2: json decode error -> loops + token_pending, # Poll 3: pending -> loops + token_slow_down, # Poll 4: slow down -> interval increases + token_success, # Poll 5: success + ] + + with patch("authsome.flows.device_code.requests.post", side_effect=mock_post_responses): + with patch("authsome.flows.device_code.time.sleep") as mock_sleep: + with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0, 0, 0, 0, 0]): + record = flow.authenticate( + provider, crypto, "default", "default", scopes=["test_scope"], client_id="cid", client_secret="sec" + ) + + assert record.status == ConnectionStatus.CONNECTED + assert crypto.decrypt(record.access_token) == "acc" + assert "test_scope" in record.scopes + # Sleep is called 5 times + assert mock_sleep.call_count == 5 + # The last sleep should be interval (1) + 5 = 6 due to slow_down + assert mock_sleep.call_args_list[-1][0][0] == 6 + + +def test_poll_for_token_access_denied(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + device_resp = MagicMock() + device_resp.json.return_value = { + "device_code": "dc", + "user_code": "uc", + "verification_uri": "http://uri", + "expires_in": 300, + "interval": 1, + } + + token_denied = MagicMock() + token_denied.status_code = 400 + token_denied.json.return_value = {"error": "access_denied"} + + with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_denied]): + with patch("authsome.flows.device_code.time.sleep"): + with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]): + with pytest.raises(AuthenticationFailedError, match="User denied"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_poll_for_token_expired_token(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + device_resp = MagicMock() + device_resp.json.return_value = { + "device_code": "dc", + "user_code": "uc", + "verification_uri": "http://uri", + "expires_in": 300, + "interval": 1, + } + + token_expired = MagicMock() + token_expired.status_code = 400 + token_expired.json.return_value = {"error": "expired_token"} + + with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_expired]): + with patch("authsome.flows.device_code.time.sleep"): + with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]): + with pytest.raises(AuthenticationFailedError, match="Device code has expired"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_poll_for_token_unknown_error(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + device_resp = MagicMock() + device_resp.json.return_value = { + "device_code": "dc", + "user_code": "uc", + "verification_uri": "http://uri", + "expires_in": 300, + "interval": 1, + } + + token_err = MagicMock() + token_err.status_code = 400 + token_err.json.return_value = {"error": "unknown_error", "error_description": "weird"} + + with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_err]): + with patch("authsome.flows.device_code.time.sleep"): + with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]): + with pytest.raises(AuthenticationFailedError, match="weird"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_poll_for_token_success_no_expires_in(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = DeviceCodeFlow() + + device_resp = MagicMock() + device_resp.json.return_value = { + "device_code": "dc", + "user_code": "uc", + "verification_uri": "http://uri", + "expires_in": 300, + "interval": 1, + } + + token_success = MagicMock() + token_success.status_code = 200 + token_success.json.return_value = {"access_token": "acc", "refresh_token": "ref"} + + with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_success]): + with patch("authsome.flows.device_code.time.sleep"): + with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]): + record = flow.authenticate(provider, crypto, "default", "default", client_id="cid", client_secret="sec") + + assert record.status == ConnectionStatus.CONNECTED + assert record.expires_at is None diff --git a/tests/test_flows_pkce.py b/tests/test_flows_pkce.py new file mode 100644 index 0000000..09428c9 --- /dev/null +++ b/tests/test_flows_pkce.py @@ -0,0 +1,269 @@ +"""Tests for the PKCE OAuth flow.""" + +import json +import urllib.parse +import urllib.request +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from authsome.crypto.local_file_crypto import LocalFileCryptoBackend +from authsome.errors import AuthenticationFailedError +from authsome.flows.pkce import PkceFlow, _CallbackHandler, _find_free_port, _generate_pkce +from authsome.models.connection import ConnectionStatus +from authsome.models.enums import AuthType, FlowType +from authsome.models.provider import OAuthConfig, ProviderDefinition + + +def _make_provider() -> ProviderDefinition: + return ProviderDefinition( + name="testoauth", + display_name="Test OAuth", + auth_type=AuthType.OAUTH2, + flow=FlowType.PKCE, + oauth=OAuthConfig( + authorization_url="https://auth.example.com/auth", + token_url="https://auth.example.com/token", + ), + ) + + +def test_find_free_port(): + port = _find_free_port() + assert isinstance(port, int) + assert port > 0 + + +def test_generate_pkce(): + verifier, challenge = _generate_pkce() + assert len(verifier) >= 43 + assert len(challenge) >= 43 + + +def test_missing_oauth(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + provider.oauth = None + flow = PkceFlow() + + with pytest.raises(AuthenticationFailedError, match="missing 'oauth' configuration"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_missing_client_id(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + + with pytest.raises(AuthenticationFailedError, match="requires a client_id"): + flow.authenticate(provider, crypto, "default", "default") + + +def test_callback_handler_log(): + handler = _CallbackHandler.__new__(_CallbackHandler) + handler.log_message("test %s", "msg") + + +def test_pkce_flow_success(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + # Extract state from URL + parsed = urllib.parse.urlparse(url) + params = urllib.parse.parse_qs(parsed.query) + state = params["state"][0] + + # Send callback + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}" + req = urllib.request.Request(callback_url) + with urllib.request.urlopen(req) as resp: + assert resp.status == 200 + + mock_resp = MagicMock() + mock_resp.json.return_value = { + "access_token": "mock_access", + "refresh_token": "mock_refresh", + "expires_in": 3600, + } + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp) as mock_post: + record = flow.authenticate( + provider, crypto, "default", "default", scopes=["read", "write"], client_id="cid", client_secret="sec" + ) + + assert record.status == ConnectionStatus.CONNECTED + assert "read" in record.scopes + assert crypto.decrypt(record.access_token) == "mock_access" + assert crypto.decrypt(record.refresh_token) == "mock_refresh" + assert record.expires_at is not None + mock_post.assert_called_once() + + +def test_pkce_flow_callback_error(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + callback_url = f"http://127.0.0.1:{port}/callback?error=access_denied&error_description=User%20denied" + req = urllib.request.Request(callback_url) + try: + urllib.request.urlopen(req) + except urllib.error.HTTPError as e: + assert e.code == 400 + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with pytest.raises(AuthenticationFailedError, match="access_denied"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_flow_callback_invalid(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + # Missing code + callback_url = f"http://127.0.0.1:{port}/callback?other=123" + req = urllib.request.Request(callback_url) + try: + urllib.request.urlopen(req) + except urllib.error.HTTPError as e: + assert e.code == 400 + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with pytest.raises(AuthenticationFailedError, match="no code received"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_flow_state_mismatch(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + # Wrong state + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state=wrong_state" + req = urllib.request.Request(callback_url) + with urllib.request.urlopen(req) as _: + pass + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with pytest.raises(AuthenticationFailedError, match="state mismatch"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_exchange_http_error(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + state = urllib.parse.parse_qs(parsed.query)["state"][0] + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}" + urllib.request.urlopen(urllib.request.Request(callback_url)) + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.pkce.http_client.post", side_effect=requests.RequestException("boom")): + with pytest.raises(AuthenticationFailedError, match="Token exchange failed: boom"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_exchange_invalid_json(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + state = urllib.parse.parse_qs(parsed.query)["state"][0] + urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}")) + + mock_resp = MagicMock() + mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0) + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="Token response was not valid JSON"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_exchange_missing_access_token(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + state = urllib.parse.parse_qs(parsed.query)["state"][0] + urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}")) + + mock_resp = MagicMock() + mock_resp.json.return_value = {"error": "invalid_grant", "error_description": "bad code"} + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp): + with pytest.raises(AuthenticationFailedError, match="bad code"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_flow_no_scopes(tmp_path): + # Test missing scopes to hit line 161 + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + # Ensure no scopes are set + provider.oauth.scopes = None + flow = PkceFlow() + flow.callback_port = _find_free_port() + port = flow.callback_port + + def mock_open(url): + parsed = urllib.parse.urlparse(url) + params = urllib.parse.parse_qs(parsed.query) + assert "scope" not in params + state = params["state"][0] + callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}" + req = urllib.request.Request(callback_url) + urllib.request.urlopen(req) + + mock_resp = MagicMock() + mock_resp.json.return_value = {"access_token": "mock"} + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") + + +def test_pkce_flow_timeout(tmp_path): + crypto = LocalFileCryptoBackend(tmp_path) + provider = _make_provider() + flow = PkceFlow() + flow.callback_port = _find_free_port() + + def mock_open(url): + # Do nothing to simulate timeout + pass + + with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open): + with patch("authsome.flows.pkce._CALLBACK_TIMEOUT_SECONDS", 0.01): + with pytest.raises(AuthenticationFailedError, match="timed out"): + flow.authenticate(provider, crypto, "default", "default", client_id="cid") diff --git a/tests/test_models.py b/tests/test_models.py index 3ad5170..3cf96c8 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,7 +1,5 @@ """Tests for authsome data models.""" -import pytest - from authsome.models.config import GlobalConfig from authsome.models.connection import ( ConnectionRecord, @@ -11,7 +9,7 @@ from authsome.models.connection import ( ) from authsome.models.enums import AuthType, ConnectionStatus, ExportFormat, FlowType from authsome.models.profile import ProfileMetadata -from authsome.models.provider import ApiKeyConfig, ClientConfig, OAuthConfig, ProviderDefinition +from authsome.models.provider import ApiKeyConfig, OAuthConfig, ProviderDefinition class TestEnums: @@ -23,8 +21,7 @@ class TestEnums: def test_flow_type_values(self) -> None: assert FlowType.DCR_PKCE.value == "dcr_pkce" - assert FlowType.API_KEY_PROMPT.value == "api_key_prompt" - assert FlowType.API_KEY_ENV.value == "api_key_env" + assert FlowType.API_KEY.value == "api_key" def test_connection_status_values(self) -> None: assert ConnectionStatus.CONNECTED.value == "connected" @@ -104,40 +101,27 @@ class TestProviderDefinition: name="openai", display_name="OpenAI", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, api_key=ApiKeyConfig( header_name="Authorization", header_prefix="Bearer", - env_var="OPENAI_API_KEY", ), ) assert provider.auth_type == AuthType.API_KEY assert provider.api_key is not None - assert provider.api_key.env_var == "OPENAI_API_KEY" def test_json_roundtrip(self) -> None: provider = ProviderDefinition( name="test", display_name="Test", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, api_key=ApiKeyConfig(), ) json_str = provider.model_dump_json() restored = ProviderDefinition.model_validate_json(json_str) assert restored.name == "test" - def test_client_env_resolution(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("MY_CLIENT_ID", "resolved-id") - client = ClientConfig(client_id="env:MY_CLIENT_ID", client_secret="literal-secret") - assert client.resolve_client_id() == "resolved-id" - assert client.resolve_client_secret() == "literal-secret" - - def test_client_env_missing(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("NONEXISTENT_VAR", raising=False) - client = ClientConfig(client_id="env:NONEXISTENT_VAR") - assert client.resolve_client_id() is None - class TestEncryptedField: """Encrypted field envelope model tests.""" diff --git a/tests/test_registry.py b/tests/test_registry.py index 199a273..c8e760b 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -19,7 +19,7 @@ def _make_api_key_provider(name: str = "testprov") -> ProviderDefinition: name=name, display_name=f"Test {name}", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, api_key=ApiKeyConfig(env_var=f"{name.upper()}_KEY"), ) @@ -49,6 +49,8 @@ class TestProviderRegistry: return ProviderRegistry(home) def test_list_providers_empty(self, registry: ProviderRegistry) -> None: + # Also remove providers dir to hit line 222 + registry._providers_dir.rmdir() providers = registry.list_providers() # Should have bundled providers assert isinstance(providers, list) @@ -60,6 +62,16 @@ class TestProviderRegistry: assert "openai" in names assert "github" in names + def test_list_providers_by_source(self, registry: ProviderRegistry) -> None: + provider = _make_api_key_provider("customprov") + registry.register_provider(provider) + + sources = registry.list_providers_by_source() + assert "bundled" in sources + assert "custom" in sources + assert any(p.name == "customprov" for p in sources["custom"]) + assert any(p.name == "openai" for p in sources["bundled"]) + def test_get_bundled_provider(self, registry: ProviderRegistry) -> None: provider = registry.get_provider("openai") assert provider.name == "openai" @@ -101,7 +113,7 @@ class TestProviderRegistry: name="openai", display_name="Custom OpenAI", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, api_key=ApiKeyConfig( header_name="X-Custom", header_prefix="Key", @@ -147,7 +159,7 @@ class TestProviderRegistry: name="noapikey", display_name="No API Key", auth_type=AuthType.API_KEY, - flow=FlowType.API_KEY_PROMPT, + flow=FlowType.API_KEY, # Missing api_key section ) with pytest.raises(InvalidProviderSchemaError, match="requires an 'api_key'"): @@ -166,3 +178,78 @@ class TestProviderRegistry: ) with pytest.raises(InvalidProviderSchemaError, match="Invalid URL"): registry.register_provider(provider) + + def test_register_oauth_provider(self, registry: ProviderRegistry) -> None: + # This covers 204->exit and 195->exit + provider = _make_oauth_provider("goodoauth") + registry.register_provider(provider) + loaded = registry.get_provider("goodoauth") + assert loaded.auth_type == AuthType.OAUTH2 + + def test_validate_oauth_missing_optional_url(self, registry: ProviderRegistry) -> None: + # Test 197->195 where a URL field is explicitly empty string + provider = _make_oauth_provider("opturl") + provider.oauth.token_url = "" # type: ignore + # It won't fail validation on the empty URL, but it might fail on register or be fine + registry.register_provider(provider) + + def test_list_providers_with_local(self, registry: ProviderRegistry) -> None: + # Test line 71: list_providers loading local providers + registry.register_provider(_make_api_key_provider("localprov")) + providers = registry.list_providers() + assert any(p.name == "localprov" for p in providers) + + def test_unrecognized_auth_type(self, registry: ProviderRegistry) -> None: + provider = _make_api_key_provider() + # Bypass Pydantic validation to simulate invalid auth_type from storage + object.__setattr__(provider, "auth_type", "INVALID_TYPE") + with pytest.raises(InvalidProviderSchemaError, match="Unrecognized auth_type"): + registry._validate_provider(provider) + + def test_load_provider_file_error(self, registry: ProviderRegistry) -> None: + # Test 215-216 + registry._providers_dir.mkdir(parents=True, exist_ok=True) + bad_file = registry._providers_dir / "bad.json" + bad_file.write_text("invalid json") + + with pytest.raises(InvalidProviderSchemaError, match="Failed to parse provider file"): + registry._load_provider_file(bad_file) + + def test_load_local_providers_error_skipping(self, registry: ProviderRegistry) -> None: + # Test 225-229 + registry._providers_dir.mkdir(parents=True, exist_ok=True) + bad_file = registry._providers_dir / "bad.json" + bad_file.write_text("invalid json") + + providers = registry._load_local_providers() + assert "bad" not in providers + + def test_load_bundled_providers_errors(self, registry: ProviderRegistry, monkeypatch: pytest.MonkeyPatch) -> None: + # Test 243-246 + import importlib.resources + + # Test ModuleNotFoundError + def mock_files_error(*args, **kwargs): + raise ModuleNotFoundError() + + monkeypatch.setattr(importlib.resources, "files", mock_files_error) + assert registry._load_bundled_providers() == {} + + # Test JSONDecodeError inside bundled providers + monkeypatch.undo() + + class MockResource: + name = "bad.json" + + def read_text(self, *args, **kwargs): + return "bad json" + + class MockPkg: + def iterdir(self): + return [MockResource()] + + def mock_files_success(*args, **kwargs): + return MockPkg() + + monkeypatch.setattr(importlib.resources, "files", mock_files_success) + assert registry._load_bundled_providers() == {} diff --git a/tests/test_store.py b/tests/test_store.py index a8a86a0..d418217 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -83,3 +83,55 @@ class TestSQLiteStore: key = "profile:default:my-provider:connection:test_conn-1" store.set(key, '{"ok": true}') assert store.get(key) == '{"ok": true}' + + def test_connect_error(self, tmp_path: Path) -> None: + profile_dir = tmp_path / "profiles" / "bad" + profile_dir.mkdir(parents=True) + # Create a directory where the db file should be to trigger an error + (profile_dir / "store.db").mkdir() + from authsome.errors import StoreUnavailableError + + with pytest.raises(StoreUnavailableError): + SQLiteStore(profile_dir) + + def test_ensure_connection_error(self, store: SQLiteStore) -> None: + from authsome.errors import StoreUnavailableError + + store.close() + with pytest.raises(StoreUnavailableError, match="Store connection is closed"): + store.get("key1") + + def test_lock_acquire_release_errors(self, store: SQLiteStore, monkeypatch: pytest.MonkeyPatch) -> None: + # Mock fcntl to raise OSError + import fcntl + + def mock_flock(fd, operation): + raise OSError("Mock error") + + monkeypatch.setattr(fcntl, "flock", mock_flock) + + # lock acquire should catch OSError and just log warning + store._acquire_lock() + assert store._lock_fd is not None + + # lock release should catch OSError + store._release_lock() + assert store._lock_fd is None + + def test_double_acquire(self, store: SQLiteStore) -> None: + store._acquire_lock() + fd1 = store._lock_fd + store._acquire_lock() # should early return + assert store._lock_fd is fd1 + store._release_lock() + + def test_close_sqlite_error(self, store: SQLiteStore) -> None: + import sqlite3 + + class MockConn: + def close(self): + raise sqlite3.Error("Mock error") + + store._conn = MockConn() + store.close() # should suppress error + assert store._conn is None diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..5fbe6ef --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,71 @@ +"""Tests for utils.py.""" + +from datetime import UTC, datetime + +import pytest + +from authsome.utils import build_store_key, is_filesystem_safe, parse_rfc3339, to_rfc3339, utc_now + + +def test_utc_now(): + now = utc_now() + assert now.tzinfo == UTC + + +def test_to_rfc3339(): + dt = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC) + assert to_rfc3339(dt) == "2023-01-01T12:00:00Z" + + # Test dt without tzinfo + dt_naive = datetime(2023, 1, 1, 12, 0, 0) + assert to_rfc3339(dt_naive) == "2023-01-01T12:00:00Z" + + +def test_parse_rfc3339(): + s = "2023-01-01T12:00:00Z" + dt = parse_rfc3339(s) + assert dt.tzinfo == UTC + + s_offset = "2023-01-01T12:00:00+00:00" + dt_offset = parse_rfc3339(s_offset) + assert dt_offset.tzinfo.utcoffset(dt_offset).total_seconds() == 0 + + +def test_is_filesystem_safe(): + assert is_filesystem_safe("valid-name_1.2") is True + # Test empty name + assert is_filesystem_safe("") is False + assert is_filesystem_safe(None) is False + # Test path traversal + assert is_filesystem_safe("bad/name") is False + assert is_filesystem_safe("bad..name") is False + assert is_filesystem_safe("bad\\name") is False + assert is_filesystem_safe(".hidden") is False + + +def test_build_store_key(): + # Test definition key + assert build_store_key(record_type="definition", provider="github") == "provider:github:definition" + # Test metadata key + assert ( + build_store_key(profile="default", provider="github", record_type="metadata") + == "profile:default:github:metadata" + ) + # Test state key + assert build_store_key(profile="default", provider="github", record_type="state") == "profile:default:github:state" + # Test connection key + assert ( + build_store_key(profile="default", provider="github", record_type="connection", connection="personal") + == "profile:default:github:connection:personal" + ) + # Test client key + assert ( + build_store_key(profile="default", provider="github", record_type="client") == "profile:default:github:client" + ) + # Test value error + with pytest.raises(ValueError): + build_store_key(profile="default", provider="github", record_type="unknown") + + # Test missing provider with profile + with pytest.raises(ValueError): + build_store_key(profile="default", record_type="metadata") diff --git a/uv.lock b/uv.lock index 5a88297..598a34a 100644 --- a/uv.lock +++ b/uv.lock @@ -13,7 +13,7 @@ wheels = [ [[package]] name = "authsome" -version = "0.1.7" +version = "0.1.9" source = { editable = "." } dependencies = [ { name = "click" },