Drop authlib and fetch the access token with httpx

CHANGELOG.md

@@ -8,6 +8,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 Releases prior to `1.0.0` were published before this changelog was added and
 are not documented here.
 
+## [Unreleased]
+
+### Removed
+
+- `authlib` is no longer a dependency. The SDK only used it to fetch its access
+  token, so it now makes that client-credentials request with `httpx`. This
+  drops authlib (and `joserfc`) from your dependency tree, including
+  CVE-2026-96760, which the SDK's code path never reached.
+- `AsyncFragmentClient.oauth2_client` and `SyncFragmentClient.oauth2_client` are
+  gone along with authlib.
+
+### Changed
+
+- A failed token request now raises `fragment.exceptions.TokenRequestException`
+  (with `.error` and `.description`) instead of authlib's `OAuthError`. Server
+  errors from the token endpoint still raise `httpx.HTTPStatusError`.
+
 ## [1.4.0]
 
 ### Added

fragment/client/async_client.py

@@ -1,14 +1,12 @@
-# Ignore untyped authlib
-# mypy: disable-error-code="import-untyped"
 import time
 from typing import Any, Dict, Optional
 
 import httpx
 from ariadne_codegen.client_generators.dependencies.async_base_client import (
     AsyncBaseClient,
 )
-from authlib.integrations.httpx_client import AsyncOAuth2Client
 
+from fragment.client.oauth import fetch_token_async
 from fragment.exceptions import MissingArgumentException, MissingTokenException
 
 
@@ -37,16 +35,18 @@ def __init__(
         self.auth_url = auth_url
         self.expiration_time: Optional[float] = None
         self.token: Optional[Dict[str, Any]] = None
-        self.oauth2_client = AsyncOAuth2Client(
-            client_id, client_secret, scope=auth_scope
-        )
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.auth_scope = auth_scope
 
     async def refresh_token(self) -> None:
         now = time.time()
         if self.expiration_time is None or self.expiration_time <= now:
             # Held in a local because `self.token` is declared `dict | None`,
             # so reading the attribute back is not narrowed by the assignment.
-            token = await self.oauth2_client.fetch_token(self.auth_url)
+            token = await fetch_token_async(
+                self.auth_url, self.client_id, self.client_secret, self.auth_scope
+            )
             self.token = token
             self.expiration_time = now + token["expires_in"]
 

fragment/client/oauth.py

@@ -0,0 +1,75 @@
+"""OAuth2 client-credentials token request against the Fragment auth endpoint.
+
+The SDK needs exactly one OAuth2 call: exchange a client ID and secret for an
+access token. That is a single form POST, so it is made with httpx directly
+rather than through an OAuth library, whose much larger surface (JOSE, OIDC,
+server-side grants) customers would otherwise inherit, along with its
+advisories.
+
+The request matches what authlib's httpx `OAuth2Client.fetch_token` sent:
+HTTP Basic client authentication (`client_secret_basic`) and a form body of
+`grant_type=client_credentials` and `scope`.
+"""
+
+from typing import Any, Dict
+
+import httpx
+
+from fragment.exceptions import TokenRequestException
+
+_HEADERS = {
+    "Accept": "application/json",
+    "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8",
+}
+
+
+def _token_request_kwargs(
+    client_id: str, client_secret: str, scope: str
+) -> Dict[str, Any]:
+    return dict(
+        data={"grant_type": "client_credentials", "scope": scope},
+        auth=httpx.BasicAuth(client_id, client_secret),
+        headers=_HEADERS,
+    )
+
+
+def _parse_token_response(response: httpx.Response) -> Dict[str, Any]:
+    if response.status_code >= 500:
+        response.raise_for_status()
+    try:
+        token = response.json()
+    except ValueError as e:
+        raise TokenRequestException(
+            f"token endpoint returned a non-JSON response "
+            f"(HTTP {response.status_code})"
+        ) from e
+    if not isinstance(token, dict):
+        raise TokenRequestException("token endpoint returned a non-object response")
+    if "error" in token:
+        raise TokenRequestException(token["error"], token.get("error_description"))
+    if "access_token" not in token or "expires_in" not in token:
+        raise TokenRequestException(
+            f"token endpoint response is missing access_token or expires_in "
+            f"(HTTP {response.status_code})"
+        )
+    return token
+
+
+def fetch_token(
+    auth_url: str, client_id: str, client_secret: str, scope: str
+) -> Dict[str, Any]:
+    with httpx.Client() as client:
+        response = client.post(
+            auth_url, **_token_request_kwargs(client_id, client_secret, scope)
+        )
+    return _parse_token_response(response)
+
+
+async def fetch_token_async(
+    auth_url: str, client_id: str, client_secret: str, scope: str
+) -> Dict[str, Any]:
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            auth_url, **_token_request_kwargs(client_id, client_secret, scope)
+        )
+    return _parse_token_response(response)

fragment/client/sync_client.py

@@ -1,12 +1,10 @@
-# Ignore untyped authlib
-# mypy: disable-error-code="import-untyped"
 import time
 from typing import Any, Dict, Optional
 
 import httpx
 from ariadne_codegen.client_generators.dependencies.base_client import BaseClient
-from authlib.integrations.httpx_client import OAuth2Client
 
+from fragment.client.oauth import fetch_token
 from fragment.exceptions import MissingArgumentException, MissingTokenException
 
 
@@ -35,14 +33,18 @@ def __init__(
         self.auth_url = auth_url
         self.expiration_time: Optional[float] = None
         self.token: Optional[Dict[str, Any]] = None
-        self.oauth2_client = OAuth2Client(client_id, client_secret, scope=auth_scope)
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.auth_scope = auth_scope
 
     def refresh_token(self) -> None:
         now = time.time()
         if self.expiration_time is None or self.expiration_time <= now:
             # Held in a local because `self.token` is declared `dict | None`,
             # so reading the attribute back is not narrowed by the assignment.
-            token = self.oauth2_client.fetch_token(self.auth_url)
+            token = fetch_token(
+                self.auth_url, self.client_id, self.client_secret, self.auth_scope
+            )
             self.token = token
             self.expiration_time = now + token["expires_in"]
 

fragment/exceptions.py

@@ -1,3 +1,6 @@
+from typing import Optional
+
+
 class MissingTokenException(ValueError):
     """Token not found."""
 
@@ -10,3 +13,12 @@ class MissingArgumentException(ValueError):
 
     def __init__(self, argument: str) -> None:
         super().__init__(f"{argument} must be provided")
+
+
+class TokenRequestException(Exception):
+    """The auth endpoint did not return an access token."""
+
+    def __init__(self, error: str, description: Optional[str] = None) -> None:
+        self.error = error
+        self.description = description
+        super().__init__(f"{error}: {description}" if description else error)

fragment/sdk/async_client.py

@@ -1,16 +1,14 @@
 # Generated by fragment (with the help of ariadne-codegen)
 
-# Ignore untyped authlib
-# mypy: disable-error-code="import-untyped"
 import time
 from typing import Any, Dict, Optional
 
 import httpx
 from ariadne_codegen.client_generators.dependencies.async_base_client import (
     AsyncBaseClient,
 )
-from authlib.integrations.httpx_client import AsyncOAuth2Client
 
+from fragment.client.oauth import fetch_token_async
 from fragment.exceptions import MissingArgumentException, MissingTokenException
 
 
@@ -39,16 +37,18 @@ def __init__(
         self.auth_url = auth_url
         self.expiration_time: Optional[float] = None
         self.token: Optional[Dict[str, Any]] = None
-        self.oauth2_client = AsyncOAuth2Client(
-            client_id, client_secret, scope=auth_scope
-        )
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.auth_scope = auth_scope
 
     async def refresh_token(self) -> None:
         now = time.time()
         if self.expiration_time is None or self.expiration_time <= now:
             # Held in a local because `self.token` is declared `dict | None`,
             # so reading the attribute back is not narrowed by the assignment.
-            token = await self.oauth2_client.fetch_token(self.auth_url)
+            token = await fetch_token_async(
+                self.auth_url, self.client_id, self.client_secret, self.auth_scope
+            )
             self.token = token
             self.expiration_time = now + token["expires_in"]
 

fragment/sync_sdk/async_client.py

@@ -1,16 +1,14 @@
 # Generated by fragment (with the help of ariadne-codegen)
 
-# Ignore untyped authlib
-# mypy: disable-error-code="import-untyped"
 import time
 from typing import Any, Dict, Optional
 
 import httpx
 from ariadne_codegen.client_generators.dependencies.async_base_client import (
     AsyncBaseClient,
 )
-from authlib.integrations.httpx_client import AsyncOAuth2Client
 
+from fragment.client.oauth import fetch_token_async
 from fragment.exceptions import MissingArgumentException, MissingTokenException
 
 
@@ -39,16 +37,18 @@ def __init__(
         self.auth_url = auth_url
         self.expiration_time: Optional[float] = None
         self.token: Optional[Dict[str, Any]] = None
-        self.oauth2_client = AsyncOAuth2Client(
-            client_id, client_secret, scope=auth_scope
-        )
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.auth_scope = auth_scope
 
     async def refresh_token(self) -> None:
         now = time.time()
         if self.expiration_time is None or self.expiration_time <= now:
             # Held in a local because `self.token` is declared `dict | None`,
             # so reading the attribute back is not narrowed by the assignment.
-            token = await self.oauth2_client.fetch_token(self.auth_url)
+            token = await fetch_token_async(
+                self.auth_url, self.client_id, self.client_secret, self.auth_scope
+            )
             self.token = token
             self.expiration_time = now + token["expires_in"]
 

fragment/sync_sdk/sync_client.py

@@ -1,14 +1,12 @@
 # Generated by fragment (with the help of ariadne-codegen)
 
-# Ignore untyped authlib
-# mypy: disable-error-code="import-untyped"
 import time
 from typing import Any, Dict, Optional
 
 import httpx
 from ariadne_codegen.client_generators.dependencies.base_client import BaseClient
-from authlib.integrations.httpx_client import OAuth2Client
 
+from fragment.client.oauth import fetch_token
 from fragment.exceptions import MissingArgumentException, MissingTokenException
 
 
@@ -37,14 +35,18 @@ def __init__(
         self.auth_url = auth_url
         self.expiration_time: Optional[float] = None
         self.token: Optional[Dict[str, Any]] = None
-        self.oauth2_client = OAuth2Client(client_id, client_secret, scope=auth_scope)
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.auth_scope = auth_scope
 
     def refresh_token(self) -> None:
         now = time.time()
         if self.expiration_time is None or self.expiration_time <= now:
             # Held in a local because `self.token` is declared `dict | None`,
             # so reading the attribute back is not narrowed by the assignment.
-            token = self.oauth2_client.fetch_token(self.auth_url)
+            token = fetch_token(
+                self.auth_url, self.client_id, self.client_secret, self.auth_scope
+            )
             self.token = token
             self.expiration_time = now + token["expires_in"]