OAuth 2.0 For Installed CLI Clients¶

Use this reference when a Python CLI calls an OAuth-protected API. Keep authentication, token lifecycle, HTTP transport, and API resources as separate ownership boundaries.

Recommended Default¶

For an interactive installed CLI, use:

Public client + Authorization Code + PKCE using S256 + loopback callback + Authlib 1.8+ + HTTPX2 + protected credential store + centralized token refresh + minimum scopes.

Follow the current OAuth 2.0 Security Best Current Practice and the OAuth 2.0 guidance for native applications. Do not use the implicit grant, Resource Owner Password Credentials grant, or an embedded client secret.

Responsibility Boundaries¶

Keep dependencies pointed inward toward authentication and transport primitives:

CLI commands
├── login/logout/status -> OAuthManager -> authorization server
└── API commands -> ApiClient/resources -> OAuthTransport
                                           └── TokenManager
                                               ├── TokenStore
                                               └── OAuth client / HTTP transport

Keep Authlib's HTTPX2 OAuth client behind the authentication boundary. API resources should request an authenticated transport or call TokenManager.get_valid_access_token(); they should not depend directly on Authlib.

Library Selection¶

Use one library per responsibility and keep each one behind an application-owned interface:

HTTPX2 is the actively developed continuation of HTTPX under Pydantic stewardship. It keeps the familiar client, request, response, authentication, and mock-transport APIs while using the httpx2 import. Authlib 1.8 moved its HTTP client integration to HTTPX2, so new OAuth clients can use one HTTP implementation:

uv add "Authlib>=1.8" "httpx2>=2" keyring

Do not call httpx2.alias_httpx() from reusable library code. It changes imports process-wide and exists as a temporary application-level migration aid. If an existing dependency still requires httpx, keep both clients behind local abstractions and remove HTTPX only after dependency and transport tests pass.

Configuration And Secret State¶

Separate public OAuth configuration from sensitive OAuth state.

Public configuration may contain:

• Client ID for the registered public client.
• Issuer, authorization endpoint, token endpoint, and optional revocation or device-authorization endpoint.
• Exact registered redirect URI rules.
• Required scopes and, when supported, resource or audience indicators.

Sensitive state includes:

• Access tokens.
• Refresh tokens.
• Token expiry and related token response fields when they reveal account or authorization state.

Represent the complete token response and store it through a narrow abstraction. A TypedDict documents the common fields without discarding provider-specific fields from the runtime dictionary:

from typing import Protocol, Required, TypedDict


class OAuthToken(TypedDict, total=False):
    access_token: Required[str]
    refresh_token: str
    token_type: str
    expires_at: float
    expires_in: int
    scope: str


class TokenStore(Protocol):
    async def load(self) -> OAuthToken | None: ...
    async def save(self, token: OAuthToken) -> None: ...
    async def delete(self) -> None: ...

For a desktop CLI, keyring remains the conservative default because it selects macOS Keychain, Windows Credential Locker, Secret Service, or KWallet as available. Its API is synchronous, so move calls off the event-loop thread. Store one JSON document per account so access-token and refresh-token rotation is replaced as one logical update:

import json
from typing import cast

import anyio
import keyring
from keyring.backend import KeyringBackend
from keyring.errors import KeyringError


class CredentialStoreError(RuntimeError):
    pass


class KeyringTokenStore:
    def __init__(
        self,
        service: str,
        account: str,
        backend: KeyringBackend | None = None,
    ) -> None:
        self._service = service
        self._account = account
        self._backend = backend or keyring.get_keyring()
        if self._backend.priority <= 0:
            raise CredentialStoreError("No protected credential store is available")

    async def load(self) -> OAuthToken | None:
        try:
            raw = await anyio.to_thread.run_sync(
                self._backend.get_password, self._service, self._account
            )
        except KeyringError as error:
            raise CredentialStoreError("Could not read OAuth credentials") from error
        if raw is None:
            return None
        try:
            value = json.loads(raw)
        except json.JSONDecodeError as error:
            raise CredentialStoreError("Stored OAuth credentials are invalid") from error
        if not isinstance(value, dict) or not isinstance(value.get("access_token"), str):
            raise CredentialStoreError("Stored OAuth credentials are invalid")
        return cast("OAuthToken", value)

    async def save(self, token: OAuthToken) -> None:
        encoded = json.dumps(token, separators=(",", ":"))
        try:
            await anyio.to_thread.run_sync(
                self._backend.set_password,
                self._service,
                self._account,
                encoded,
            )
        except KeyringError as error:
            raise CredentialStoreError("Could not save OAuth credentials") from error

    async def delete(self) -> None:
        if await self.load() is None:
            return
        try:
            await anyio.to_thread.run_sync(
                self._backend.delete_password, self._service, self._account
            )
        except KeyringError as error:
            raise CredentialStoreError("Could not delete OAuth credentials") from error

Fail closed when no recommended backend is available. Do not install keyrings.alt as an automatic fallback; it intentionally includes possibly insecure backends.

Storage Alternatives¶

• Consider rust-native-keyring when native compiled wheels, the Rust keyring ecosystem, and its richer credential-store selection fit the supported platforms. Its Python package is still 0.x, so pin it, verify wheel availability, and test lock/unlock and deletion behavior on every target OS before preferring it over keyring.
• Consider the official 1Password Python SDK when users already rely on 1Password and desktop authorization prompts, auditing, or shared vault policy are product requirements. It is asynchronous and supports desktop-app authorization, but its SDK is also still version 0; it is not a transparent local-keyring replacement.
• Use a managed service such as AWS Secrets Manager, Azure Key Vault, Google Secret Manager, or HashiCorp Vault for unattended or centrally governed deployments. Authenticate with workload identity or another deployment-owned mechanism; do not solve storage by introducing a second long-lived bootstrap secret.

For a headless environment without a credential service, require an explicit storage backend appropriate to its threat model; do not silently fall back to plaintext config. Never emit tokens through logs, telemetry, exceptions, shell output, or status commands.

Interactive Authorization¶

Use Authorization Code with PKCE for browser-based login:

1. Generate a cryptographically random state and a fresh PKCE code_verifier for every attempt.
2. Derive the S256 code challenge and construct the authorization URL with the exact requested redirect URI and minimum scopes.
3. Bind a temporary listener to 127.0.0.1 on an ephemeral port. Do not expose it on all interfaces.
4. Open the system browser and wait for one callback with a short timeout and cancellation path.
5. Reject OAuth errors, a missing code, or any callback whose state does not exactly match.
6. Exchange the code using the original verifier and redirect URI.
7. Persist the complete returned token state atomically, then stop the listener immediately.
8. Return a minimal success page and CLI message without displaying credentials.

Treat the CLI as a public client. A secret distributed inside source, a package, a binary, or an environment-independent configuration file cannot authenticate installed copies of the CLI.

The following manager shows the Authlib-owned part of a loopback flow. A separate callback receiver should bind 127.0.0.1, accept one request, enforce a timeout and maximum request size, then pass the complete callback URL to finish():

import secrets
from dataclasses import dataclass
from urllib.parse import parse_qs, urlsplit

from authlib.integrations.httpx_client import AsyncOAuth2Client


@dataclass(frozen=True, slots=True)
class OAuthConfig:
    client_id: str
    authorization_endpoint: str
    token_endpoint: str
    redirect_uri: str
    scopes: tuple[str, ...]


@dataclass(frozen=True, slots=True)
class PendingAuthorization:
    url: str
    state: str
    code_verifier: str


class OAuthManager:
    def __init__(self, config: OAuthConfig, store: TokenStore) -> None:
        self._config = config
        self._store = store
        self._client = AsyncOAuth2Client(
            client_id=config.client_id,
            redirect_uri=config.redirect_uri,
            scope=" ".join(config.scopes),
            code_challenge_method="S256",
            token_endpoint_auth_method="none",
        )

    def begin(self) -> PendingAuthorization:
        verifier = secrets.token_urlsafe(64)
        url, state = self._client.create_authorization_url(
            self._config.authorization_endpoint,
            code_verifier=verifier,
        )
        return PendingAuthorization(url=url, state=state, code_verifier=verifier)

    async def finish(self, callback_url: str, pending: PendingAuthorization) -> OAuthToken:
        callback = urlsplit(callback_url)
        expected = urlsplit(self._config.redirect_uri)
        callback_target = (callback.scheme, callback.hostname, callback.port, callback.path)
        expected_target = (expected.scheme, expected.hostname, expected.port, expected.path)
        if callback_target != expected_target:
            raise AuthenticationError("OAuth callback used an unexpected redirect URI")

        query = parse_qs(callback.query)
        if query.get("state") != [pending.state]:
            raise AuthenticationError("OAuth callback state did not match")
        if "error" in query:
            raise AuthenticationError("Authorization server rejected login")
        code = query.get("code", [None])[0]
        if code is None:
            raise AuthenticationError("OAuth callback did not contain a code")

        result = await self._client.fetch_token(
            self._config.token_endpoint,
            code=code,
            code_verifier=pending.code_verifier,
        )
        token = cast("OAuthToken", dict(result))
        await self._store.save(token)
        return token

    async def aclose(self) -> None:
        await self._client.aclose()

Do not persist PendingAuthorization: state and code_verifier are short-lived, single-attempt values. Define AuthenticationError in the application's stable error taxonomy and keep callback query values out of its message.

If a browser or loopback listener is impractical and the provider exposes it, use the standardized Device Authorization Grant. Respect the server-provided polling interval, slow_down, expiration, and cancellation behavior. Do not invent a device flow against a provider that does not advertise or document one.

Authorization Server Metadata¶

Prefer OAuth 2.0 Authorization Server Metadata or OpenID Connect discovery when the provider supports it. Validate that discovered metadata belongs to the configured issuer and require HTTPS for non-loopback endpoints.

Use explicit endpoints when discovery is unavailable or when a controlled deployment intentionally pins them. Do not mix endpoints discovered from one issuer with configuration from another.

Token Lifecycle¶

Centralize expiry and refresh decisions in TokenManager:

load token
├── missing -> authentication required
├── valid beyond refresh leeway -> return access token
└── expired or near expiry
    └── acquire refresh lock
        ├── reload token
        ├── return it if another task refreshed it
        └── refresh, atomically persist the full response, and return it

Use a small expiry leeway, commonly 60 seconds, to avoid starting a request with a token that expires in transit. For a single-process async CLI, an asyncio.Lock is sufficient for in-process refresh coordination. If multiple processes can share one token store, add storage-level coordination or optimistic versioning; an in-process lock cannot prevent cross-process races.

Preserve a refresh token when the server omits it from a refresh response, but replace it whenever rotation returns a new one. Save the newly returned token as one atomic state update so an older writer cannot restore a superseded refresh token.

Classify missing, expired-without-refresh, rejected-refresh, and revoked credentials as authentication failures with a clear path to log in again. Do not turn refresh failures into anonymous API calls.

A small refresher adapter and token manager keep Authlib details out of storage and API resources:

import asyncio
import time
from collections.abc import Callable


class AuthenticationError(RuntimeError):
    pass


class AuthlibTokenRefresher:
    def __init__(self, config: OAuthConfig) -> None:
        self._config = config

    async def refresh(self, token: OAuthToken) -> OAuthToken:
        refresh_token = token.get("refresh_token")
        if refresh_token is None:
            raise AuthenticationError("Login is required")
        async with AsyncOAuth2Client(
            client_id=self._config.client_id,
            token=dict(token),
            token_endpoint_auth_method="none",
        ) as client:
            result = await client.refresh_token(
                self._config.token_endpoint,
                refresh_token=refresh_token,
            )
        refreshed = cast("OAuthToken", dict(result))
        refreshed.setdefault("refresh_token", refresh_token)
        return refreshed


class TokenManager:
    def __init__(
        self,
        store: TokenStore,
        refresher: AuthlibTokenRefresher,
        *,
        refresh_leeway: float = 60.0,
        clock: Callable[[], float] = time.time,
    ) -> None:
        self._store = store
        self._refresher = refresher
        self._refresh_leeway = refresh_leeway
        self._clock = clock
        self._lock = asyncio.Lock()

    def _is_usable(self, token: OAuthToken) -> bool:
        expires_at = token.get("expires_at")
        return expires_at is None or expires_at > self._clock() + self._refresh_leeway

    async def get_valid_access_token(self, *, force_refresh: bool = False) -> str:
        token = await self._store.load()
        if token is None:
            raise AuthenticationError("Login is required")
        if not force_refresh and self._is_usable(token):
            return token["access_token"]

        async with self._lock:
            token = await self._store.load()
            if token is None:
                raise AuthenticationError("Login is required")
            if not force_refresh and self._is_usable(token):
                return token["access_token"]
            refreshed = await self._refresher.refresh(token)
            await self._store.save(refreshed)
            return refreshed["access_token"]

This lock covers one process. Replace or augment it with storage-level compare-and-swap or an inter-process lock when several processes share the same credential entry.

Authenticated HTTP Transport¶

The transport should:

1. Obtain a valid access token before sending a protected request.
2. Inject the authorization header without exposing the token to API resource code.
3. Apply the project's normal timeout, TLS, proxy, retry, and error-mapping policy.
4. Optionally react to one 401 by forcing one synchronized refresh and replaying the request once.
5. Raise an authentication error if the replay is still unauthorized.

Do not refresh blindly on every 401; unauthorized responses can indicate revocation, malformed credentials, the wrong audience, or another authentication failure. Never create an unbounded refresh or request loop. Replay only requests whose body can be safely regenerated, and do not treat 403 as an expiry signal.

HTTPX2 custom authentication is a compact way to apply the token manager to every API request. This example retries one bodyless, read-only request after a synchronized refresh and leaves all other 401 responses untouched:

import httpx2


class OAuthAuth(httpx2.Auth):
    requires_response_body = True

    def __init__(self, tokens: TokenManager) -> None:
        self._tokens = tokens

    def sync_auth_flow(self, request: httpx2.Request):
        raise RuntimeError("OAuthAuth requires httpx2.AsyncClient")
        yield request

    async def async_auth_flow(self, request: httpx2.Request):
        access_token = await self._tokens.get_valid_access_token()
        request.headers["Authorization"] = f"Bearer {access_token}"
        response = yield request

        if response.status_code != 401 or request.method not in {"GET", "HEAD", "OPTIONS"}:
            return
        access_token = await self._tokens.get_valid_access_token(force_refresh=True)
        request.headers["Authorization"] = f"Bearer {access_token}"
        yield request


class ApiClient:
    def __init__(self, base_url: str, tokens: TokenManager) -> None:
        self._http = httpx2.AsyncClient(
            base_url=base_url,
            auth=OAuthAuth(tokens),
            timeout=httpx2.Timeout(20.0, connect=5.0),
        )

    async def get_project(self, project_id: str) -> dict[str, object]:
        response = await self._http.get(f"/projects/{project_id}")
        response.raise_for_status()
        value = response.json()
        if not isinstance(value, dict):
            raise ValueError("Expected an object response")
        return value

    async def aclose(self) -> None:
        await self._http.aclose()

For streaming requests, uploads, or state-changing methods, omit automatic replay unless the API supplies an idempotency mechanism and the request body can be rebuilt. Map the final HTTPX2 response or exception into application errors before it reaches a CLI command.

Scopes And Token Restrictions¶

• Request only scopes needed by the CLI's supported operations.
• Keep scopes explicit in configuration and stable in tests.
• Request offline access or equivalent provider-specific scope only when refresh tokens are needed.
• Use audience or resource restrictions when supported.
• Detect when stored authorization lacks scopes required by a command and direct the user through deliberate reauthorization rather than quietly broadening every login.

CLI Surface¶

Expose a small authentication surface consistent with the existing command framework:

mycli login
mycli logout
mycli auth status

login starts interactive authorization and reports only progress and outcome. logout deletes local token state and uses the provider's revocation endpoint when supported; explain if remote revocation fails after local deletion. auth status may show account identity, granted scopes, issuer, and expiry, but never a token or authorization code.

A manual auth refresh command is optional and primarily diagnostic. Normal API commands should not require users to manage refresh timing.

Suggested Package Shape¶

Adapt names to the existing project rather than forcing this exact tree:

src/mycli/
├── cli.py
├── errors.py
├── api/
│   ├── client.py
│   ├── transport.py
│   └── resources/
└── auth/
    ├── config.py
    ├── manager.py
    ├── token.py
    └── store.py

Keep token models independent from provider client objects so storage, tests, and API code do not inherit Authlib's internal representation as a public application contract.

Branching Guidance¶

• If the provider supports loopback redirects: use Authorization Code with PKCE and an ephemeral 127.0.0.1 listener.
• If the execution environment cannot open a browser or accept a loopback callback: use Device Authorization Grant only when the provider supports it.
• If the API is called by unattended automation rather than a person: use the provider's machine-to-machine grant and credential mechanism; do not reuse an interactive user's refresh token as service identity.
• If the provider supports discovery: derive endpoints from validated issuer metadata; otherwise pin explicit HTTPS endpoints.
• If requests are concurrent in one process: serialize refresh with a lock and re-read state after acquisition.
• If token state is shared across processes: use inter-process coordination and atomic persistence.
• If an existing CLI already has HTTP and configuration abstractions: integrate at those boundaries instead of replacing the command framework or resource layer.

Tests And Verification¶

Test protocol behavior without depending on a live identity provider:

• Login creates fresh state and verifier values and uses S256.
• Callback handling accepts the expected state and rejects missing, mismatched, duplicate, timed-out, and OAuth-error callbacks.
• The listener binds only to loopback and always shuts down.
• Token exchange uses the same redirect URI and verifier as authorization.
• Valid tokens bypass refresh; near-expiry tokens refresh once.
• Concurrent requests cause one refresh and all callers observe the persisted replacement token.
• Refresh-token rotation cannot be overwritten by stale state.
• A 401 causes at most one eligible replay; a second 401 fails.
• Status and error output remain useful without revealing token values.
• Logout clears local state and handles optional remote revocation explicitly.

Use deterministic clocks, fake token stores, and httpx2.MockTransport for unit tests. Add a provider integration test only when the project has suitable isolated credentials and CI secret handling.

Completion Checks¶

1. The CLI is registered and implemented as a public client without an embedded secret.
2. Interactive login uses Authorization Code with PKCE S256, fresh state, exact redirect validation, and a bounded loopback listener.
3. Device authorization is conditional on provider support and follows server polling instructions.
4. Token storage is replaceable, protected, atomic, and absent from logs and normal output.
5. One component owns expiry, synchronized refresh, and refresh-token rotation.
6. API resources remain independent of OAuth protocol and storage details.
7. Scopes and audience are minimal and explicit.
8. Authentication retries are bounded and replay only eligible requests.
9. Login, logout, status, refresh, concurrency, callback rejection, and redaction paths have focused verification.
10. Provider-specific behavior and installed dependency versions are checked against current primary documentation.