Automatic token refresh when using client_credentials

[NomAnor]

I'm experimenting with the Httpx Client using client_credentials grant. The automatic token refresh does not work as I expected.

client_id and client_secret and token_endpoint are given when the client is created, so all necessry information is available to fetch the token.

When making a request I get a MissingTokenError exception because I didn't supply a token: https://github.com/lepture/authlib/blob/ee4337cf7c825349dd23870822a3cc7df123097f/authlib/integrations/httpx_client/oauth2_client.py#L198-L202

I had expected that the token would be automatically fetched if none is available and it can automatically be fetched. Thats what the call to ensure_active_token() does.

I could cheat the system by specifying a dummy token when creating the client (-1 because of #530):

token={'expires_at': -1, 'access_token': ''}

Is it deliberate that in this situation, when the token can be fetched without user intervention, that an initial token must be supplied (via token keyword or a explicit call to fetch_token())?

[caspervk]

+1 on this issue. I have solved it locally the following way, but would love to throw away my code. I think an in-library solution could be implemented much more elegantly.

from typing import Any
from typing import Optional

from authlib.integrations.httpx_client import (
    AsyncOAuth2Client as AsyncHTTPXOAuth2Client,
)
from authlib.integrations.httpx_client import OAuth2Client as HTTPXOAuth2Client
from authlib.oauth2 import OAuth2Client
from httpx import USE_CLIENT_DEFAULT
from httpx._types import AuthTypes
from pydantic import AnyHttpUrl


class BaseAuthenticatedClient(OAuth2Client):
    def __init__(self, token_endpoint: AnyHttpUrl, *args: Any, **kwargs: Any):
        """Base used to implement authenticated HTTPX clients. Does not work on its own."""
        self.token_endpoint = token_endpoint
        super().__init__(*args, token_endpoint=token_endpoint, **kwargs)

    def should_fetch_token(
        self,
        url: str,
        withhold_token: bool = False,
        auth: Optional[AuthTypes] = USE_CLIENT_DEFAULT,  # type: ignore[assignment]
    ) -> bool:
        """
        Determine if we should fetch a token. Authlib automatically _refreshes_ tokens,
        but it does not fetch the initial one. Therefore, we should fetch a token the
        first time a request is sent; i.e. when self.token is None.

        Args:
            url: The URL of the request we are in the context of. Used to avoid
                 recursion, since fetching a token also uses our caller self.request().
            withhold_token: Forwarded from `self.request(..., withhold_token=False)`. If
             this is set, Authlib does not pass a token in the request, in which case
             there is no need to fetch one either.
            auth: Forwarded from `self.request(..., auth=USE_CLIENT_DEFAULT)`. If this
             is set, Authlib does not pass a token in the request, in which case there
             is no need to fetch one either.

        Returns: True if a token should be fetched. False otherwise.
        """
        return (
            not withhold_token
            and auth is USE_CLIENT_DEFAULT
            and self.token is None
            and url != self.token_endpoint
        )


class AuthenticatedHTTPXClient(BaseAuthenticatedClient, HTTPXOAuth2Client):
    """Synchronous HTTPX Client that automatically authenticates requests."""

    def request(
        self,
        method: str,
        url: str,
        withhold_token: bool = False,
        auth: AuthTypes = USE_CLIENT_DEFAULT,  # type: ignore[assignment]
        **kwargs: Any,
    ) -> Any:
        """
        Decorate Authlib's OAuth2Client.request() to automatically fetch a token the
        first time a request is made.
        """
        if self.should_fetch_token(url, withhold_token, auth):
            self.fetch_token()
        return super().request(method, url, withhold_token, auth, **kwargs)


class AuthenticatedAsyncHTTPXClient(BaseAuthenticatedClient, AsyncHTTPXOAuth2Client):
    """Asynchronous HTTPX Client that automatically authenticates requests."""

    async def request(
        self,
        method: str,
        url: str,
        withhold_token: bool = False,
        auth: AuthTypes = USE_CLIENT_DEFAULT,  # type: ignore[assignment]
        **kwargs: Any,
    ) -> Any:
        """
        Decorate Authlib's AsyncOAuth2Client.request() to automatically fetch a token
        the first time a request is made.
        """
        if self.should_fetch_token(url, withhold_token, auth):
            await self.fetch_token()
        return await super().request(method, url, withhold_token, auth, **kwargs)

Code is from https://github.com/magenta-aps/ra-clients/blob/ebada519d634bde9021ad14d7eee312372366351/raclients/auth.py, modified here for clarity.

If you pull token_endpoint from metadata you avoid overwriting __init__, I'm doing so here for more obvious type hints. The docstrings explain the logic pretty well, I think. Note the url != self.token_endpoint hack, which I'm not particularly fond of.

[alex-linx]

Also ran into this issue. Passing token={'expires_at': -1, 'access_token': ''} does not seem to do the job

[simensol]

I think this also applies to OAuth2Session when using client_credentials. Even passing token={'expires_at': -1, 'access_token': ''} doesn't work:

client = OAuth2Session(
  client_id=CLIENT_ID,
  client_secret=CLIENT_SECRET,
  token={'expires_at': -1, 'access_token': ''},
  token_endpoint="https://auth.example.com/oauth/token",
)

client.get(url=...)

raises InvalidTokenError(), although the documentation states that

If your OAuth2Session class was created with the token_endpoint parameter, Authlib will automatically refresh the token when it has expired:

I suspect that the automatic token update requires that the token has a refresh_token. If so, the documentation should make this clearer.

This is maybe also related to https://github.com/lepture/authlib/issues/532.