Add rsconnect token to print a stored OAuth access token

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.

docs/commands/token.md

@@ -0,0 +1,3 @@
+::: mkdocs-click
+    :module: rsconnect.main
+    :command: token

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
 

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:

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 <token>'.\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