diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 3093d384..ba110352 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- `rsconnect token` prints the OAuth access token stored by `rsconnect login` for a + Posit Connect server, so other tools, such as the Connect SDKs, can make API + requests with your login. `--refresh` gets a new token after the server rejects + the current one, and `--format json` prints machine-readable output. It exits + with status 4 when you need to log in again, and it never prints API keys. +- OAuth token refreshes are serialized across processes, so concurrent commands + don't invalidate each other's rotated refresh tokens. A command whose token was + already refreshed by another process uses the new token without refreshing again. + - When client credentials can't publish to a Posit Connect Cloud account because the account requires single sign-on (SSO), the error now says so, instead of suggesting the publisher role. diff --git a/docs/commands/token.md b/docs/commands/token.md new file mode 100644 index 00000000..54414e5b --- /dev/null +++ b/docs/commands/token.md @@ -0,0 +1,3 @@ +::: mkdocs-click + :module: rsconnect.main + :command: token diff --git a/mkdocs.yml b/mkdocs.yml index ddfc2981..d9ce7f4c 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -56,6 +56,7 @@ nav: - remove: commands/remove.md - set-default: commands/set-default.md - system: commands/system.md + - token: commands/token.md - version: commands/version.md - write-manifest: commands/write-manifest.md diff --git a/rsconnect/api.py b/rsconnect/api.py index 2185b5ca..36315c01 100644 --- a/rsconnect/api.py +++ b/rsconnect/api.py @@ -553,77 +553,29 @@ def _can_refresh_token(self) -> bool: return isinstance(self._server, RSConnectServer) and bool(self._server.oauth_client_id) def _attempt_token_refresh(self) -> bool: - from .oauth import ( - InvalidClientError, - discover_oauth_metadata, - keyring_delete_tokens, - keyring_get_tokens, - keyring_store_token, - refresh_access_token, - register_client, - ) - from .metadata import ServerStore + from .oauth import InvalidClientError, NotLoggedInError, refresh_stored_tokens server = cast(RSConnectServer, self._server) - - # The keyring is where a login stores its tokens; the entry's own fields are - # the fallback for a machine without one. - _, refresh_token = keyring_get_tokens(server.url) - if not refresh_token: - entry = ServerStore().saved_entry(server.server_name, server.url) - if entry: - refresh_token = entry.get("oauth_refresh_token") # type: ignore[assignment] - if not refresh_token: - return False - try: - metadata = discover_oauth_metadata(server.url, server.insecure, server.ca_data) - token_response = refresh_access_token( - metadata, server.oauth_client_id or "", refresh_token, server.insecure, server.ca_data + new_access = refresh_stored_tokens( + server.url, + server.oauth_client_id or "", + server.server_name, + server.insecure, + server.ca_data, + rejected_access_token=server.oauth_access_token, ) + except NotLoggedInError: + return False except InvalidClientError: - # Client was deleted server-side; clear stale tokens and re-register - keyring_delete_tokens(server.url) - store = ServerStore() - entry = store.saved_entry(server.server_name, server.url) - if entry: - entry_name = str(entry.get("name", server.server_name or server.url)) - store.update_oauth_tokens(entry_name, None, None, None) - try: - metadata = discover_oauth_metadata(server.url, server.insecure, server.ca_data) - new_client_id = register_client(metadata, server.url, server.insecure, server.ca_data) - server.oauth_client_id = new_client_id - if entry: - entry["oauth_client_id"] = new_client_id # type: ignore[typeddict-unknown-key] - store._set(entry_name, entry) # type: ignore[possibly-undefined] - logger.warning("OAuth client was re-registered; please run `rsconnect login` again.") - except Exception as exc: - logger.warning( - "OAuth client re-registration failed (%s). Please run `rsconnect login` again." % type(exc).__name__ - ) + logger.warning("The OAuth client was deleted on the server; please run `rsconnect login` again.") return False except Exception as exc: logger.warning("OAuth token refresh failed (%s)." % type(exc).__name__) return False - new_access = token_response["access_token"] - new_refresh = token_response.get("refresh_token", refresh_token) - expires_in = token_response.get("expires_in") - import time - - new_expiry = time.time() + expires_in if expires_in else None - self.authorization(f"Bearer {new_access}") server.oauth_access_token = new_access - - stored = keyring_store_token(server.url, new_access, new_refresh) - if not stored: - store = ServerStore() - entry = store.saved_entry(server.server_name, server.url) - if entry: - entry_name = str(entry.get("name", server.server_name or server.url)) - store.update_oauth_tokens(entry_name, new_access, new_refresh, new_expiry) - return True def _tweak_response(self, response: HTTPResponse) -> JsonData | HTTPResponse: diff --git a/rsconnect/main.py b/rsconnect/main.py index 4c974fee..40b4f373 100644 --- a/rsconnect/main.py +++ b/rsconnect/main.py @@ -1822,6 +1822,101 @@ def logout( click.echo('Logged out from "%s".' % (name or server)) +# `token` exits with this when the user has to log in again: the server has no +# stored OAuth credentials, or it rejected them. Same code as `gh` uses. +TOKEN_EXIT_LOGIN_REQUIRED = 4 + + +@cli.command( + short_help="Print an OAuth access token for a Posit Connect server.", + help=( + "Print the OAuth access token stored by 'rsconnect login' for a Posit Connect server, so other tools " + "can make API requests as you with 'Authorization: Bearer '.\n\n" + "The token is printed as is, without contacting the server. When the server rejects it with a 401, " + "run the command again with --refresh to get a new one, then retry the request. Concurrent refreshes " + "are safe: if another process has already refreshed, its token is printed.\n\n" + "Only tokens from 'rsconnect login' are printed, never API keys. The command never prompts.\n\n" + "Output: the bare token, or with --format json an object with the keys access_token, token_type " + '("Bearer"), and server. Nothing is printed on stdout on failure.\n\n' + "Exit status: 0 on success; 4 when you need to log in again (no stored OAuth credentials, or the " + "server rejected them); 1 for other errors, such as an unreachable server." + ), +) +@click.option("--name", "-n", help="The nickname of the Posit Connect server.") +@click.option("--server", "-s", envvar="CONNECT_SERVER", help="The URL of the Posit Connect server.") +@click.option("--refresh", is_flag=True, help="Get a new access token, after the server rejected the current one.") +@click.option( + "--format", + "output_format", + type=click.Choice(["text", "json"]), + default="text", + show_default=True, + help="Print the bare token, or a JSON object.", +) +@click.pass_context +def token( + ctx: click.Context, + name: Optional[str], + server: Optional[str], + refresh: bool, + output_format: str, +): + from .oauth import ( + InvalidClientError, + InvalidGrantError, + NotLoggedInError, + refresh_stored_tokens, + stored_tokens, + ) + + # Name the login command the way the user invoked this one, which differs + # when another CLI mounts these commands (e.g. `posit connect token`). + login_command = "%s login" % ctx.command_path.rsplit(" ", 1)[0] + + def fail(message: str, exit_code: int = 1) -> Never: + click.secho("Error: %s" % message, fg="bright_red", err=True) + sys.exit(exit_code) + + def login_required(message: str) -> Never: + fail("%s Run '%s' to log in." % (message, login_command), TOKEN_EXIT_LOGIN_REQUIRED) + + try: + server_data = server_store.resolve(name, server) + except RSConnectException as exc: + fail(exc.message) + if not server_data.from_store or not server_data.url: + if not server: + fail("Specify a server with -s/--server or -n/--name.") + login_required("No saved credentials for %s." % server) + url = server_data.url + client_id = server_data.oauth_client_id + if not client_id: + login_required('Server "%s" was not added with OAuth login.' % server_data.name) + + try: + access_token, _ = stored_tokens(url, server_data.name) + if refresh or not access_token: + access_token = refresh_stored_tokens( + url, + client_id, + server_data.name, + bool(server_data.insecure), + server_data.ca_data, + rejected_access_token=access_token, + ) + except NotLoggedInError: + login_required("No OAuth credentials are stored for %s." % url) + except (InvalidGrantError, InvalidClientError): + login_required("%s rejected the stored OAuth credentials." % url) + except RSConnectException as exc: + fail(exc.message) + + if output_format == "json": + click.echo(json.dumps({"access_token": access_token, "token_type": "Bearer", "server": url})) + else: + click.echo(access_token) + + def _get_names_to_check(file_or_directory: str) -> list[str]: """ A function to determine a set files to look for in getting information about a diff --git a/rsconnect/oauth.py b/rsconnect/oauth.py index aef9ac70..3ddc5d7d 100644 --- a/rsconnect/oauth.py +++ b/rsconnect/oauth.py @@ -7,14 +7,18 @@ from __future__ import annotations import base64 +import errno import hashlib +import os import queue import secrets +import sys import threading import time import webbrowser +from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, HTTPServer as _HTTPServer -from typing import Any, Dict, Iterable, Mapping, Optional, Tuple, cast +from typing import Any, Dict, Generator, Iterable, Mapping, Optional, Tuple, cast from urllib.parse import parse_qs, urlencode, urlparse, urlunparse import click @@ -74,6 +78,13 @@ def __init__(self, description: Optional[str] = None) -> None: super().__init__("The OAuth grant is invalid, expired, or has been revoked.") +class NotLoggedInError(RSConnectException): + """Raised when no OAuth tokens are stored for a Posit Connect server.""" + + def __init__(self, server_url: str) -> None: + super().__init__("No OAuth credentials are stored for %s." % server_url) + + def _check_oauth_error_response(response: HTTPResponse) -> None: """Check an HTTPResponse for OAuth error codes and raise appropriately.""" if response.json_data and isinstance(response.json_data, dict): @@ -882,3 +893,173 @@ def keyring_get_tokens(key: str) -> Tuple[Optional[str], Optional[str]]: def keyring_delete_tokens(key: str) -> None: """Delete OAuth tokens from the system keyring.""" keyring_delete_values(key, (ACCESS_TOKEN_FIELD, REFRESH_TOKEN_FIELD)) + + +def stored_tokens(server_url: str, server_name: Optional[str] = None) -> Tuple[Optional[str], Optional[str]]: + """The (access_token, refresh_token) that `rsconnect login` stored for a Posit Connect server. + + The keyring is where a login stores its tokens; the server's saved entry holds + them instead on a machine without one. + """ + access_token, refresh_token = keyring_get_tokens(server_url) + if access_token is None and refresh_token is None: + from .metadata import ServerStore + + entry = ServerStore().saved_entry(server_name, server_url) + if entry: + access_token = entry.get("oauth_access_token") + refresh_token = entry.get("oauth_refresh_token") + return access_token, refresh_token + + +def refresh_stored_tokens( + server_url: str, + client_id: str, + server_name: Optional[str] = None, + insecure: bool = False, + ca_data: Optional[str | bytes] = None, + *, + rejected_access_token: Optional[str], +) -> str: + """Exchange the stored refresh token for a new access token, store both, and return the access token. + + Pass the access token the server rejected as ``rejected_access_token``. Refresh + tokens rotate, so two processes refreshing at once would invalidate each + other's tokens. The exchange therefore happens under a lock shared across + processes, and if the stored access token no longer matches the rejected one, + another process has already refreshed and its token is returned as is. + + Raises NotLoggedInError when no refresh token is stored, and InvalidGrantError + when the server rejects it. Raises InvalidClientError when the OAuth client + no longer exists on the server, after clearing the stored tokens and + registering a new client, so the next `rsconnect login` works. + """ + from .metadata import ServerStore + + with _refresh_lock(): + access_token, refresh_token = stored_tokens(server_url, server_name) + if access_token and access_token != rejected_access_token: + return access_token + if not refresh_token: + raise NotLoggedInError(server_url) + + try: + metadata = discover_oauth_metadata(server_url, insecure, ca_data) + token_response = refresh_access_token(metadata, client_id, refresh_token, insecure, ca_data) + except InvalidClientError: + _replace_deleted_client(server_url, server_name, insecure, ca_data) + raise + + new_access = str(token_response["access_token"]) + new_refresh = token_response.get("refresh_token", refresh_token) + expires_in = token_response.get("expires_in") + new_expiry = time.time() + expires_in if expires_in else None + + if not keyring_store_token(server_url, new_access, new_refresh): + store = ServerStore() + entry = store.saved_entry(server_name, server_url) + if entry: + entry_name = str(entry.get("name", server_name or server_url)) + store.update_oauth_tokens(entry_name, new_access, new_refresh, new_expiry) + return new_access + + +def _replace_deleted_client( + server_url: str, server_name: Optional[str], insecure: bool, ca_data: Optional[str | bytes] +) -> None: + """Clear the tokens of an OAuth client the server deleted, and register a new client in its place.""" + from .metadata import ServerStore + + keyring_delete_tokens(server_url) + store = ServerStore() + entry = store.saved_entry(server_name, server_url) + if not entry: + return + entry_name = str(entry.get("name", server_name or server_url)) + store.update_oauth_tokens(entry_name, None, None, None) + try: + metadata = discover_oauth_metadata(server_url, insecure, ca_data) + new_client_id = register_client(metadata, server_url, insecure, ca_data) + except Exception as exc: + logger.debug("OAuth client re-registration failed (%s)." % type(exc).__name__) + return + entry = store.saved_entry(server_name, server_url) or entry + entry["oauth_client_id"] = new_client_id # type: ignore[typeddict-unknown-key] + store._set(entry_name, entry) # pyright: ignore[reportPrivateUsage] + + +# Long enough for another process to finish a refresh, which is two HTTP requests. +_REFRESH_LOCK_TIMEOUT = 60.0 +_REFRESH_LOCK_POLL_INTERVAL = 0.05 + + +def _refresh_lock_path() -> str: + from .metadata import config_dirname + + return os.path.join(config_dirname(), "oauth-refresh.lock") + + +@contextmanager +def _refresh_lock(timeout: float = _REFRESH_LOCK_TIMEOUT) -> Generator[None, None, None]: + """Hold the lock that serializes OAuth token refreshes across processes. + + One lock covers every server: refreshes are rare and quick. When the lock + file cannot be created (a read-only config directory, say), the refresh goes + ahead unlocked, as it would have before the lock existed. + """ + path = _refresh_lock_path() + try: + os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True) + descriptor = os.open(path, os.O_CREAT | os.O_RDWR, 0o600) + except OSError as exc: + logger.debug("Could not open the OAuth refresh lock (%s); refreshing without it." % type(exc).__name__) + yield + return + + try: + deadline = time.monotonic() + timeout + while not _try_lock(descriptor): + if time.monotonic() >= deadline: + raise RSConnectException("Timed out waiting for another process to refresh the OAuth tokens.") + time.sleep(_REFRESH_LOCK_POLL_INTERVAL) + try: + yield + finally: + _unlock(descriptor) + finally: + os.close(descriptor) + + +def _try_lock(descriptor: int) -> bool: + """Take an exclusive lock on an open file without blocking. Returns whether it was taken.""" + if sys.platform == "win32": + import msvcrt + + os.lseek(descriptor, 0, os.SEEK_SET) + try: + msvcrt.locking(descriptor, msvcrt.LK_NBLCK, 1) + except OSError: + return False + return True + + import fcntl + + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + if exc.errno in (errno.EAGAIN, errno.EACCES, errno.EWOULDBLOCK): + return False + raise + return True + + +def _unlock(descriptor: int) -> None: + if sys.platform == "win32": + import msvcrt + + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(descriptor, fcntl.LOCK_UN) diff --git a/tests/conftest.py b/tests/conftest.py index 6822ad3e..b85d8e6f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ import sys from typing import Iterator +from unittest import mock import pytest @@ -37,3 +38,11 @@ def no_system_keyring() -> Iterator[None]: deletes a still-current working directory on Windows, which fails. """ yield from _no_system_keyring() + + +@pytest.fixture(autouse=True) +def oauth_refresh_lock_in_tmp(tmp_path_factory: pytest.TempPathFactory) -> Iterator[None]: + """Keep the OAuth refresh lock file out of the user's real config directory.""" + path = str(tmp_path_factory.getbasetemp() / "oauth-refresh.lock") + with mock.patch("rsconnect.oauth._refresh_lock_path", return_value=path): + yield diff --git a/tests/test_oauth_token.py b/tests/test_oauth_token.py new file mode 100644 index 00000000..efc5b382 --- /dev/null +++ b/tests/test_oauth_token.py @@ -0,0 +1,349 @@ +"""Tests for the shared OAuth token refresh and the `rsconnect token` command.""" + +from __future__ import annotations + +import inspect +import json +import threading +from typing import Any, Iterator +from unittest.mock import MagicMock, patch + +import click +import pytest +from click.testing import CliRunner + +from rsconnect.exception import RSConnectException +from rsconnect.metadata import ServerStore +from rsconnect.oauth import ( + InvalidClientError, + InvalidGrantError, + NotLoggedInError, + _refresh_lock, + refresh_stored_tokens, + stored_tokens, +) + +FAKE_URL = "https://connect.example.com" +FAKE_METADATA: dict[str, Any] = {"token_endpoint": f"{FAKE_URL}/oauth/v1/token"} + + +@pytest.fixture +def store(tmp_path: Any) -> Iterator[ServerStore]: + """A server store in a temporary directory, used by both the command and the refresh.""" + server_store = ServerStore(str(tmp_path)) + with patch("rsconnect.main.server_store", server_store): + with patch("rsconnect.metadata.ServerStore", return_value=server_store): + yield server_store + + +@pytest.fixture +def keyring() -> Iterator[dict[str, tuple[Any, Any]]]: + """An in-memory stand-in for the keyring, keyed by server URL.""" + tokens: dict[str, tuple[Any, Any]] = {} + + def store_token(key: str, access_token: Any, refresh_token: Any) -> bool: + tokens[key] = (access_token, refresh_token) + return True + + with patch("rsconnect.oauth.keyring_get_tokens", side_effect=lambda key: tokens.get(key, (None, None))): + with patch("rsconnect.oauth.keyring_store_token", side_effect=store_token): + with patch("rsconnect.oauth.keyring_delete_tokens", side_effect=lambda key: tokens.pop(key, None)): + yield tokens + + +@pytest.fixture +def token_endpoint() -> Iterator[MagicMock]: + """Patch OAuth discovery, and return the mock for the refresh token request.""" + with patch("rsconnect.oauth.discover_oauth_metadata", return_value=FAKE_METADATA): + with patch("rsconnect.oauth.refresh_access_token") as refresh: + refresh.return_value = {"access_token": "new-at", "refresh_token": "new-rt", "expires_in": 3600} + yield refresh + + +def _runner() -> CliRunner: + options = {"mix_stderr": False} if "mix_stderr" in inspect.signature(CliRunner).parameters else {} + return CliRunner(**options) + + +def _invoke(args: list[str]) -> Any: + from rsconnect.main import cli + + return _runner().invoke(cli, ["token", *args], env={"CONNECT_SERVER": None}, prog_name="rsconnect") + + +class TestStoredTokens: + def test_reads_the_keyring(self, store: ServerStore, keyring: dict[str, Any]): + keyring[FAKE_URL] = ("at", "rt") + assert stored_tokens(FAKE_URL) == ("at", "rt") + + def test_falls_back_to_the_saved_entry(self, store: ServerStore, keyring: dict[str, Any]): + store.set("myserver", FAKE_URL, oauth_client_id="c", oauth_access_token="at", oauth_refresh_token="rt") + assert stored_tokens(FAKE_URL, "myserver") == ("at", "rt") + + +class TestRefreshStoredTokens: + def test_refreshes_and_stores_rotated_tokens( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("old-at", "old-rt") + + result = refresh_stored_tokens(FAKE_URL, "client-1", rejected_access_token="old-at") + + assert result == "new-at" + assert keyring[FAKE_URL] == ("new-at", "new-rt") + assert token_endpoint.call_args.args[1:3] == ("client-1", "old-rt") + + def test_keeps_the_refresh_token_when_the_server_does_not_rotate_it( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("old-at", "old-rt") + token_endpoint.return_value = {"access_token": "new-at"} + + refresh_stored_tokens(FAKE_URL, "client-1", rejected_access_token="old-at") + + assert keyring[FAKE_URL] == ("new-at", "old-rt") + + def test_returns_a_token_another_process_already_refreshed( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("newer-at", "newer-rt") + + result = refresh_stored_tokens(FAKE_URL, "client-1", rejected_access_token="old-at") + + assert result == "newer-at" + token_endpoint.assert_not_called() + + def test_stores_in_the_saved_entry_without_a_keyring(self, store: ServerStore, token_endpoint: MagicMock): + store.set("myserver", FAKE_URL, oauth_client_id="c", oauth_access_token="old-at", oauth_refresh_token="old-rt") + with patch("rsconnect.oauth.keyring_get_tokens", return_value=(None, None)): + with patch("rsconnect.oauth.keyring_store_token", return_value=False): + refresh_stored_tokens(FAKE_URL, "c", "myserver", rejected_access_token="old-at") + + entry = store.get_by_name("myserver") + assert entry is not None + assert entry.get("oauth_access_token") == "new-at" + assert entry.get("oauth_refresh_token") == "new-rt" + assert isinstance(entry.get("oauth_token_expiry"), float) + + def test_not_logged_in_without_a_refresh_token( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + with pytest.raises(NotLoggedInError): + refresh_stored_tokens(FAKE_URL, "client-1", rejected_access_token=None) + token_endpoint.assert_not_called() + + def test_rejected_refresh_token_keeps_the_stored_tokens( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("old-at", "old-rt") + token_endpoint.side_effect = InvalidGrantError() + + with pytest.raises(InvalidGrantError): + refresh_stored_tokens(FAKE_URL, "client-1", rejected_access_token="old-at") + assert keyring[FAKE_URL] == ("old-at", "old-rt") + + def test_deleted_client_clears_tokens_and_registers_a_new_client( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + store.set("myserver", FAKE_URL, oauth_client_id="client-1") + keyring[FAKE_URL] = ("old-at", "old-rt") + token_endpoint.side_effect = InvalidClientError() + + with patch("rsconnect.oauth.register_client", return_value="client-2"): + with pytest.raises(InvalidClientError): + refresh_stored_tokens(FAKE_URL, "client-1", "myserver", rejected_access_token="old-at") + + assert FAKE_URL not in keyring + entry = store.get_by_name("myserver") + assert entry is not None + assert entry.get("oauth_client_id") == "client-2" + + +class TestRefreshLock: + def test_a_second_holder_waits_for_the_first(self): + with _refresh_lock(): + with pytest.raises(RSConnectException, match="Timed out"): + with _refresh_lock(timeout=0.1): + pass + # Released on exit, so it can be taken again. + with _refresh_lock(timeout=0.1): + pass + + def test_concurrent_refreshes_exchange_the_refresh_token_once( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("old-at", "old-rt") + results: list[str] = [] + + def refresh() -> None: + results.append(refresh_stored_tokens(FAKE_URL, "client-1", rejected_access_token="old-at")) + + threads = [threading.Thread(target=refresh) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert results == ["new-at"] * 4 + assert token_endpoint.call_count == 1 + + def test_refreshes_unlocked_when_the_lock_file_cannot_be_created(self, tmp_path: Any): + blocker = tmp_path / "not-a-directory" + blocker.write_text("") + with patch("rsconnect.oauth._refresh_lock_path", return_value=str(blocker / "oauth-refresh.lock")): + with _refresh_lock(): + pass + + +class TestClientRefresh: + """The refresh a client attempts when a request gets a 401.""" + + def _client(self, access_token: str) -> Any: + from rsconnect.api import RSConnectClient, RSConnectServer + + server = RSConnectServer(FAKE_URL, None, oauth_access_token=access_token, oauth_client_id="client-1") + return RSConnectClient(server) + + def test_refreshes_and_applies_the_new_token( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("old-at", "old-rt") + client = self._client("old-at") + + assert client._attempt_token_refresh() is True + assert client._headers["Authorization"] == "Bearer new-at" + assert keyring[FAKE_URL] == ("new-at", "new-rt") + + def test_uses_a_token_another_process_already_refreshed( + self, store: ServerStore, keyring: dict[str, Any], token_endpoint: MagicMock + ): + keyring[FAKE_URL] = ("newer-at", "newer-rt") + client = self._client("old-at") + + assert client._attempt_token_refresh() is True + assert client._headers["Authorization"] == "Bearer newer-at" + token_endpoint.assert_not_called() + + +class TestTokenCommand: + @pytest.fixture(autouse=True) + def logged_in(self, store: ServerStore, keyring: dict[str, Any]) -> None: + store.set("myserver", FAKE_URL, oauth_client_id="client-1") + keyring[FAKE_URL] = ("stored-at", "stored-rt") + + def test_prints_the_stored_token_without_contacting_the_server(self, token_endpoint: MagicMock): + result = _invoke(["-n", "myserver"]) + + assert result.exit_code == 0, result.stderr + assert result.stdout == "stored-at\n" + token_endpoint.assert_not_called() + + def test_resolves_the_server_by_url(self): + result = _invoke(["-s", FAKE_URL]) + + assert result.exit_code == 0, result.stderr + assert result.stdout == "stored-at\n" + + def test_uses_the_only_saved_server_by_default(self): + result = _invoke([]) + + assert result.exit_code == 0, result.stderr + assert result.stdout == "stored-at\n" + + def test_json(self): + result = _invoke(["-n", "myserver", "--format", "json"]) + + assert result.exit_code == 0, result.stderr + assert json.loads(result.stdout) == {"access_token": "stored-at", "token_type": "Bearer", "server": FAKE_URL} + + def test_refresh(self, keyring: dict[str, Any], token_endpoint: MagicMock): + result = _invoke(["-n", "myserver", "--refresh"]) + + assert result.exit_code == 0, result.stderr + assert result.stdout == "new-at\n" + assert token_endpoint.call_args.args[2] == "stored-rt" + assert keyring[FAKE_URL] == ("new-at", "new-rt") + + def test_refreshes_when_only_a_refresh_token_is_stored(self, keyring: dict[str, Any], token_endpoint: MagicMock): + keyring[FAKE_URL] = (None, "stored-rt") + + result = _invoke(["-n", "myserver"]) + + assert result.exit_code == 0, result.stderr + assert result.stdout == "new-at\n" + + def test_logged_out(self, keyring: dict[str, Any]): + keyring.clear() + + result = _invoke(["-n", "myserver"]) + + assert result.exit_code == 4 + assert result.stdout == "" + assert "Run 'rsconnect login' to log in." in result.stderr + + def test_revoked(self, token_endpoint: MagicMock): + token_endpoint.side_effect = InvalidGrantError() + + result = _invoke(["-n", "myserver", "--refresh"]) + + assert result.exit_code == 4 + assert result.stdout == "" + assert "rejected the stored OAuth credentials" in result.stderr + + def test_other_refresh_failures_do_not_ask_for_login(self, token_endpoint: MagicMock): + token_endpoint.side_effect = RSConnectException("OAuth request failed: HTTP 503.") + + result = _invoke(["-n", "myserver", "--refresh"]) + + assert result.exit_code == 1 + assert result.stdout == "" + assert "HTTP 503" in result.stderr + + def test_never_prints_an_api_key(self, store: ServerStore): + store.set("keyserver", "https://other.example.com", api_key="secret-api-key") + + result = _invoke(["-n", "keyserver"]) + + assert result.exit_code == 4 + assert "secret-api-key" not in result.stdout + result.stderr + assert "not added with OAuth login" in result.stderr + + def test_unknown_url(self): + result = _invoke(["-s", "https://unknown.example.com"]) + + assert result.exit_code == 4 + assert "No saved credentials" in result.stderr + + def test_unknown_nickname(self): + result = _invoke(["-n", "nope"]) + + assert result.exit_code == 1 + assert "does not exist" in result.stderr + + def test_no_server(self, store: ServerStore): + store.set("second", "https://other.example.com", oauth_client_id="c") + + result = _invoke([]) + + assert result.exit_code == 1 + assert "Specify a server" in result.stderr + + def test_login_hint_names_the_mounting_cli(self, keyring: dict[str, Any]): + from rsconnect.main import cli + + keyring.clear() + + @click.group() + def posit() -> None: + pass + + @posit.group() + def connect() -> None: + pass + + connect.add_command(cli.commands["token"]) + + result = _runner().invoke(posit, ["connect", "token", "-n", "myserver"], env={"CONNECT_SERVER": None}) + + assert result.exit_code == 4 + assert "Run 'posit connect login' to log in." in result.stderr