refactor(auth): consolidate OAuth response parsing

packages/auth/src/pipefy_auth/__init__.py

@@ -38,6 +38,7 @@
     resolve_pipefy_auth,
     tier_for,
 )
+from pipefy_auth.responses import OAuthErrorResponse, TokenResponse
 from pipefy_auth.revoke import (
     RevocationError,
     RevocationUnsupportedError,
@@ -62,6 +63,7 @@
     "DiscoveryPolicy",
     "LoginError",
     "LoginResult",
+    "OAuthErrorResponse",
     "OidcClient",
     "ProviderMetadata",
     "RefreshError",
@@ -75,6 +77,7 @@
     "SessionDeleteError",
     "StaticBearerAuth",
     "StoredSession",
+    "TokenResponse",
     "__version__",
     "configure_keychain_backend",
     "delete_session",

packages/auth/src/pipefy_auth/discovery.py

@@ -81,7 +81,7 @@ def fetch_provider_metadata(
     if response.status_code != 200:
         # Status-only; never echo the raw body. Discovery isn't OAuth so there's
         # no RFC 6749 ``error`` field to surface, and a `[:N]` window of the
-        # body is the same echo-channel class scrubbed in ``_format_token_error``.
+        # body is the same echo-channel class scrubbed in ``OAuthErrorResponse``.
         raise ValueError(f"OIDC discovery failed ({response.status_code}) at {url}")
 
     try:

packages/auth/src/pipefy_auth/flow.py

@@ -9,6 +9,7 @@
 from urllib.parse import urlencode
 
 import httpx
+from pydantic import ValidationError
 
 from pipefy_auth import _http
 from pipefy_auth.discovery import (
@@ -18,6 +19,11 @@
 )
 from pipefy_auth.loopback import CallbackResult, LoopbackCapture
 from pipefy_auth.pkce import challenge_from_verifier, generate_verifier
+from pipefy_auth.responses import (
+    OAuthErrorResponse,
+    TokenResponse,
+    _format_validation_error,
+)
 
 _DEFAULT_SCOPES = ("openid", "profile", "email", "offline_access")
 _TOKEN_EXCHANGE_TIMEOUT_S = 30.0
@@ -29,10 +35,10 @@ class LoginError(RuntimeError):
 
 @dataclass(frozen=True)
 class LoginResult:
-    """The token response, plus the resolved issuer (post-discovery)."""
+    """The OAuth token, plus the resolved issuer (post-discovery)."""
 
     issuer: str
-    token_response: dict[str, object]
+    token: TokenResponse
 
 
 def build_authorization_url(
@@ -65,7 +71,7 @@ def exchange_code(
     redirect_uri: str,
     code_verifier: str,
     client: httpx.Client,
-) -> dict[str, object]:
+) -> TokenResponse:
     """Exchange an authorization code for tokens at the issuer's token endpoint."""
     try:
         response = client.post(
@@ -82,42 +88,28 @@ def exchange_code(
         raise LoginError(f"Token exchange request failed: {exc}") from exc
 
     if response.status_code != 200:
-        raise LoginError(_format_token_error(response))
+        # ``error_description`` is free-form per RFC 6749 §5.2 — its content
+        # reflects the IdP's framing, and surfacing it at all is part of trusting
+        # the IdP. A length cap wouldn't change that (an attacker can truncate
+        # to fit any cap, and a tight cap kills legitimate error context).
+        raise LoginError(
+            OAuthErrorResponse.from_response(response).render(
+                fallback=f"Token endpoint returned HTTP {response.status_code}",
+                prefix="Token exchange failed",
+            )
+        )
     try:
         payload = response.json()
     except ValueError as exc:
         raise LoginError(f"Token endpoint returned non-JSON response: {exc}") from exc
     if not isinstance(payload, dict):
         raise LoginError("Token endpoint returned a non-object JSON payload.")
-    return payload
-
-
-def _format_token_error(response: httpx.Response) -> str:
-    """Render a token-exchange non-200 as a user-safe message.
-
-    Only OAuth-standard ``error`` / ``error_description`` fields are surfaced;
-    raw bodies are never echoed (RFC 6749 doesn't forbid IdPs from including
-    submitted params like ``code_verifier`` in error responses, and a `[:N]`
-    snippet would be a guaranteed echo channel under a hostile IdP).
-    """
-    generic = f"Token endpoint returned HTTP {response.status_code}"
     try:
-        payload = response.json()
-    except ValueError:
-        return generic
-    if not isinstance(payload, dict):
-        return generic
-    error = payload.get("error")
-    if not error:
-        return generic
-    # ``error_description`` is free-form per RFC 6749 §5.2 — its content reflects
-    # the IdP's framing, and surfacing it at all is part of trusting the IdP. A
-    # length cap wouldn't change that (an attacker can truncate to fit any cap,
-    # and a tight cap kills legitimate error context).
-    description = payload.get("error_description")
-    if description:
-        return f"Token exchange failed: {error}: {description}"
-    return f"Token exchange failed: {error}"
+        return TokenResponse.from_payload(payload)
+    except ValidationError as exc:
+        raise LoginError(_format_validation_error(exc)) from exc
+    except ValueError as exc:
+        raise LoginError(str(exc)) from exc
 
 
 def run_login(
@@ -188,15 +180,15 @@ def run_login(
             _ensure_callback_ok(callback, expected_state=state)
             code = cast(str, callback.code)
 
-            token_response = exchange_code(
+            token = exchange_code(
                 metadata=metadata,
                 client_id=client_id,
                 code=code,
                 redirect_uri=capture.redirect_uri,
                 code_verifier=verifier,
                 client=http,
             )
-    return LoginResult(issuer=metadata.issuer, token_response=token_response)
+    return LoginResult(issuer=metadata.issuer, token=token)
 
 
 def _ensure_callback_ok(callback: CallbackResult, *, expected_state: str) -> None:

packages/auth/src/pipefy_auth/refresh.py

@@ -5,10 +5,16 @@
 import time
 
 import httpx
+from pydantic import ValidationError
 
 from pipefy_auth import _http
 from pipefy_auth.discovery import fetch_provider_metadata
 from pipefy_auth.locks import RefreshLockTimeout, file_lock, refresh_lock_path
+from pipefy_auth.responses import (
+    OAuthErrorResponse,
+    TokenResponse,
+    _format_validation_error,
+)
 from pipefy_auth.storage import StoredSession, load_session, store_session
 
 _LEEWAY_S = 60
@@ -79,33 +85,20 @@ def _refresh_error_from(response: httpx.Response) -> RefreshError:
     """Render a non-200 refresh response as a structured ``RefreshError``.
 
     Only OAuth-standard ``error`` / ``error_description`` fields are surfaced;
-    raw bodies are never echoed (same threat model as the token-exchange scrub
-    in ``flow._format_token_error`` — a hostile IdP could echo submitted params
-    like ``refresh_token`` in error responses, and a ``[:N]`` window would be a
-    guaranteed leak channel).
+    raw bodies are never echoed (a hostile IdP could echo submitted params
+    like ``refresh_token`` in error responses, and a ``[:N]`` window would be
+    a guaranteed leak channel).
     """
-    generic = f"Refresh failed (HTTP {response.status_code})"
-    try:
-        payload = response.json()
-    except ValueError:
-        return RefreshError(generic)
-    if not isinstance(payload, dict):
-        return RefreshError(generic)
-    error_value = payload.get("error")
-    error_code = error_value if isinstance(error_value, str) else None
-    description_value = payload.get("error_description")
-    description = description_value if isinstance(description_value, str) else None
-    if not error_code:
-        return RefreshError(generic)
-    if description:
-        return RefreshError(
-            f"{generic}: {error_code}: {description}", error_code=error_code
-        )
-    return RefreshError(f"{generic}: {error_code}", error_code=error_code)
+    fallback = f"Refresh failed (HTTP {response.status_code})"
+    err = OAuthErrorResponse.from_response(response)
+    return RefreshError(
+        err.render(fallback=fallback, prefix="Refresh failed"),
+        error_code=err.error,
+    )
 
 
 def _is_stale(session: StoredSession, leeway_s: int) -> bool:
-    expires_in = session.expires_in or 0
+    expires_in = session.token.expires_in or 0
     deadline = session.obtained_at + expires_in - leeway_s
     return time.time() >= deadline
 
@@ -117,29 +110,38 @@ def _refresh_and_store(
     client_id: str,
     http_client: httpx.Client | None,
 ) -> StoredSession:
-    token_response = refresh_access_token(
+    prior = session.token
+    payload = refresh_access_token(
         issuer=issuer,
         client_id=client_id,
-        refresh_token=session.refresh_token,
+        refresh_token=prior.refresh_token,
         http_client=http_client,
     )
     # Carry forward fields the IdP may omit from a refresh response so the
     # rotated session keeps its full shape — without ``expires_in`` the next
     # freshness check treats the token as already expired and refreshes again
     # on the very next call.
-    token_response.setdefault("refresh_token", session.refresh_token)
-    if session.expires_in is not None:
-        token_response.setdefault("expires_in", session.expires_in)
-    if session.refresh_expires_in is not None:
-        token_response.setdefault("refresh_expires_in", session.refresh_expires_in)
-    if session.scope is not None:
-        token_response.setdefault("scope", session.scope)
-    if session.id_token is not None:
-        token_response.setdefault("id_token", session.id_token)
+    payload.setdefault("refresh_token", prior.refresh_token)
+    if prior.expires_in is not None:
+        payload.setdefault("expires_in", prior.expires_in)
+    if prior.refresh_expires_in is not None:
+        payload.setdefault("refresh_expires_in", prior.refresh_expires_in)
+    if prior.scope is not None:
+        payload.setdefault("scope", prior.scope)
+    if prior.id_token is not None:
+        payload.setdefault("id_token", prior.id_token)
+    try:
+        new_token = TokenResponse.from_payload(payload)
+    except ValidationError as exc:
+        raise RefreshError(
+            f"Refresh response malformed: {_format_validation_error(exc)}"
+        ) from exc
+    except ValueError as exc:
+        raise RefreshError(f"Refresh response malformed: {exc}") from exc
     return store_session(
         issuer=session.issuer,
         client_id=session.client_id,
-        token_response=token_response,
+        token=new_token,
     )
 
 

packages/auth/src/pipefy_auth/resolver.py

@@ -80,7 +80,7 @@ def _token() -> str:
             raise RuntimeError(
                 "Stored Pipefy session was removed; run `pipefy auth login` again."
             )
-        return session.access_token
+        return session.token.access_token
 
     def _force_refresh() -> str | None:
         try:
@@ -91,7 +91,7 @@ def _force_refresh() -> str | None:
             )
         except RefreshError:
             return None
-        return session.access_token if session is not None else None
+        return session.token.access_token if session is not None else None
 
     return RefreshableBearerAuth(token_provider=_token, force_refresh=_force_refresh)