From 16cfb3395e1edb94425b2d9fa40e4aeb5dfbb973 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sat, 30 May 2026 16:48:12 -0300 Subject: [PATCH 1/3] feat(cli): add pipefy auth login --device (#154) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement OAuth 2.0 Device Authorization Grant (RFC 8628) for headless sign-in flows where the loopback callback dance isn't workable: CI runners, SSH-only boxes, dev containers without a system browser. * `pipefy_auth.device` runs the device-code flow end-to-end. The CLI prints the user code and verification URL (plus the `_complete` form when the IdP advertises it), then polls the token endpoint per RFC 8628 §3.5: clamp non-positive intervals to 5s, sleep only the remaining TTL before each request, and honor `slow_down` / `authorization_pending`. * PKCE is bound to the device grant. Realms that mandate PKCE on the OIDC client (observed: piporacle staging) reject the device-auth request without a `code_challenge`; sending it on the authorization request and the matching `code_verifier` on the token poll is safe on IdPs that don't enforce it. * `pipefy auth login --device` is the CLI entry point. `--device` combined with `--no-browser` or an explicit `--callback-timeout` hard-errors via `typer.BadParameter` — the device flow has no browser-callback dance to suppress and the deadline is governed by the IdP's `expires_in`. The `_DEFAULT_CALLBACK_TIMEOUT_S` constant exists so the explicit-default check compares by value. * `discovery` parses and SSRF-validates `device_authorization_endpoint` from the OIDC document; the docstring covers the device path alongside the browser dance and token exchange. * `docs/cli/auth.md` documents the device grant under "Headless / SSH", including the IdP requirement (`device_authorization_endpoint` in OIDC discovery). Built on the foundations from the prior two PRs: device.py uses the shared `TokenResponse` / `OAuthErrorResponse` wrappers from ``pipefy_auth.responses`` and the permissive `pipefy_infra.coerce` helpers throughout. Test surface: 27 tests covering the device polling state machine, TTL-aware sleep, RFC-mandated interval clamp, slow-down handling, PKCE binding, and the conflicting-flag CLI errors. --- docs/cli/auth.md | 17 +- packages/auth/src/pipefy_auth/__init__.py | 3 + packages/auth/src/pipefy_auth/device.py | 266 ++++++++ packages/auth/src/pipefy_auth/discovery.py | 19 +- packages/cli/src/pipefy_cli/commands/auth.py | 61 +- .../cli/tests/fixtures/cli_help_golden.txt | 3 + packages/cli/tests/test_auth_device.py | 583 ++++++++++++++++++ 7 files changed, 929 insertions(+), 23 deletions(-) create mode 100644 packages/auth/src/pipefy_auth/device.py create mode 100644 packages/cli/tests/test_auth_device.py diff --git a/docs/cli/auth.md b/docs/cli/auth.md index d5380c7c..9d24b1b5 100644 --- a/docs/cli/auth.md +++ b/docs/cli/auth.md @@ -109,7 +109,7 @@ The service-account OAuth token URL (Tier 3) and the OIDC issuer URL (Tier 4) ar | Command | Status | Notes | |---------|--------|-------| -| `pipefy auth login` | Available | Browser-based PKCE login; persists the session in the OS keychain. | +| `pipefy auth login` | Available | Browser-based PKCE login (default) or device grant with `--device`; persists the session in the OS keychain. | | `pipefy auth status` | Available | Print which auth source is active, identity, and session expiry. | | `pipefy auth logout` | Available | Revoke the refresh token at the IdP and clear the stored session. | @@ -117,8 +117,9 @@ The service-account OAuth token URL (Tier 3) and the OIDC issuer URL (Tier 4) ar | Flag | Default | Effect | |------|---------|--------| -| `--no-browser` | _off_ | Print the authorization URL to stdout instead of trying to launch a browser. | -| `--callback-timeout ` | `180.0` | Seconds to wait for the browser callback (minimum 5). | +| `--device` | _off_ | Use the OAuth 2.0 device authorization grant (RFC 8628): prints a user code and verification URL instead of binding a loopback callback. | +| `--no-browser` | _off_ | Print the authorization URL to stdout instead of trying to launch a browser (PKCE flow only). | +| `--callback-timeout ` | `180.0` | Seconds to wait for the browser callback (minimum 5; PKCE flow only). | #### `pipefy auth status` flags @@ -184,16 +185,16 @@ Production: `https://signin.pipefy.com/realms/pipefy` (the `PIPEFY_AUTH_URL` def ## Headless / SSH -The PKCE flow needs a loopback HTTP server on `127.0.0.1:` and a browser that can reach it. SSH sessions and headless boxes break both assumptions. +The default PKCE flow needs a loopback HTTP server on `127.0.0.1:` and a browser that can reach it. SSH sessions and headless boxes break both assumptions. -Options today: +**Device login:** run `pipefy auth login --device` when your IdP advertises `device_authorization_endpoint` in OIDC discovery. The CLI prints a **user code** and **verification URL** (and a one-shot link when the IdP provides `verification_uri_complete`). Open the URL on any machine with a browser, enter the code, and the CLI polls until tokens arrive — no loopback listener is required. -1. **Run `pipefy auth login` on your laptop**, then copy `.env` plus the relevant secrets over. The keychain entry itself doesn't transfer between machines. +Other options: + +1. **Run `pipefy auth login` on your laptop** (PKCE), then copy `.env` plus the relevant secrets over. The keychain entry itself doesn't transfer between machines. 2. **Use a service account** (`PIPEFY_SERVICE_ACCOUNT_*`) on the headless box — this is the canonical answer for CI and servers. 3. **Static bearer** via `PIPEFY_TOKEN` for short-lived debugging. -Forthcoming: an OAuth 2.0 Device Authorization Grant (`pipefy auth login --device`) that swaps the loopback callback for a code you paste into a browser elsewhere. Tracked in issue #138. - --- ## Troubleshooting diff --git a/packages/auth/src/pipefy_auth/__init__.py b/packages/auth/src/pipefy_auth/__init__.py index addf035f..afc2c086 100644 --- a/packages/auth/src/pipefy_auth/__init__.py +++ b/packages/auth/src/pipefy_auth/__init__.py @@ -16,6 +16,7 @@ RefreshableBearerAuth, StaticBearerAuth, ) +from pipefy_auth.device import DeviceAuthorization, run_device_login from pipefy_auth.discovery import ( DiscoveryPolicy, ProviderMetadata, @@ -60,6 +61,7 @@ "AuthSettings", "CallableBearerAuth", "DEFAULT_AUTH_CLIENT_ID", + "DeviceAuthorization", "DiscoveryPolicy", "LoginError", "LoginResult", @@ -91,6 +93,7 @@ "refresh_access_token", "resolve_pipefy_auth", "revoke_session", + "run_device_login", "run_login", "store_session", "tier_for", diff --git a/packages/auth/src/pipefy_auth/device.py b/packages/auth/src/pipefy_auth/device.py new file mode 100644 index 00000000..db5aff20 --- /dev/null +++ b/packages/auth/src/pipefy_auth/device.py @@ -0,0 +1,266 @@ +"""Orchestrate the OAuth 2.0 Device Authorization Grant login (RFC 8628).""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Callable + +import httpx +from pipefy_infra.coerce import optional_str + +from pipefy_auth import _http +from pipefy_auth.discovery import ( + DiscoveryPolicy, + ProviderMetadata, + fetch_provider_metadata, +) +from pipefy_auth.flow import LoginError, LoginResult +from pipefy_auth.pkce import challenge_from_verifier, generate_verifier +from pipefy_auth.responses import OAuthErrorResponse, TokenResponse + +_DEFAULT_SCOPES = ("openid", "profile", "email", "offline_access") +_DEVICE_AUTH_TIMEOUT_S = 30.0 +_DEFAULT_POLL_INTERVAL_S = 5.0 +_SLOW_DOWN_INCREMENT_S = 5.0 +_DEVICE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code" + +_EXPIRED_TOKEN_MESSAGE = ( + "Sign-in didn't complete in time. Re-run `pipefy auth login --device` to try again." +) +_ACCESS_DENIED_MESSAGE = "Sign-in cancelled by the user." + + +@dataclass(frozen=True) +class DeviceAuthorization: + """The device-authorization response the user must act on (RFC 8628 §3.2).""" + + device_code: str + user_code: str + verification_uri: str + verification_uri_complete: str | None + expires_in: float + interval: float + + @classmethod + def from_payload(cls, payload: dict[str, object]) -> "DeviceAuthorization": + """Parse a device-authorization JSON body. Raises ``ValueError`` on malformed input.""" + missing = [ + key + for key in ("device_code", "user_code", "verification_uri") + if not payload.get(key) + ] + if "expires_in" not in payload or payload["expires_in"] is None: + missing.append("expires_in") + if missing: + raise ValueError( + "Device authorization response is missing required fields: " + + ", ".join(missing) + ) + + expires_in_raw = payload["expires_in"] + try: + expires_in = float(expires_in_raw) # type: ignore[arg-type] + except (TypeError, ValueError) as exc: + raise ValueError( + f"Device authorization response has invalid expires_in: {expires_in_raw!r}" + ) from exc + + interval_raw = payload.get("interval", _DEFAULT_POLL_INTERVAL_S) + try: + interval = float(interval_raw) # type: ignore[arg-type] + except (TypeError, ValueError) as exc: + raise ValueError( + f"Device authorization response has invalid interval: {interval_raw!r}" + ) from exc + if interval <= 0: + interval = _DEFAULT_POLL_INTERVAL_S + + return cls( + device_code=str(payload["device_code"]), + user_code=str(payload["user_code"]), + verification_uri=str(payload["verification_uri"]), + verification_uri_complete=optional_str( + payload.get("verification_uri_complete") + ), + expires_in=expires_in, + interval=interval, + ) + + +def request_device_authorization( + *, + metadata: ProviderMetadata, + client_id: str, + code_challenge: str, + scopes: tuple[str, ...] = _DEFAULT_SCOPES, + client: httpx.Client, +) -> DeviceAuthorization: + """POST to the issuer's device authorization endpoint and parse the response. + + PKCE (``code_challenge`` + ``code_challenge_method=S256``) is sent on every + request: RFC 8628 doesn't require it, but Keycloak applies a client's + "PKCE Code Challenge Method" setting uniformly across grant types, so a + realm that mandates PKCE on the authorization-code path also rejects + PKCE-less device-auth requests with ``invalid_request: Missing parameter: + code_challenge_method``. IdPs that don't enforce PKCE accept the extra + parameters and validate the verifier without complaint. + """ + if metadata.device_authorization_endpoint is None: + raise LoginError( + "OIDC discovery did not advertise a device_authorization_endpoint. " + "This issuer may not support device login." + ) + + try: + response = client.post( + metadata.device_authorization_endpoint, + data={ + "client_id": client_id, + "scope": " ".join(scopes), + "code_challenge": code_challenge, + "code_challenge_method": "S256", + }, + ) + except httpx.HTTPError as exc: + raise LoginError(f"Device authorization request failed: {exc}") from exc + + if response.status_code != 200: + raise LoginError( + OAuthErrorResponse.from_response(response).render( + fallback=f"Device authorization endpoint returned HTTP {response.status_code}", + prefix="Device authorization failed", + ) + ) + + try: + payload = response.json() + except ValueError as exc: + raise LoginError( + f"Device authorization endpoint returned non-JSON response: {exc}" + ) from exc + if not isinstance(payload, dict): + raise LoginError( + "Device authorization endpoint returned a non-object JSON payload." + ) + + try: + return DeviceAuthorization.from_payload(payload) + except ValueError as exc: + raise LoginError(str(exc)) from exc + + +def poll_device_token( + *, + metadata: ProviderMetadata, + client_id: str, + code_verifier: str, + device_auth: DeviceAuthorization, + client: httpx.Client, + sleep: Callable[[float], None] = time.sleep, + monotonic: Callable[[], float] = time.monotonic, +) -> TokenResponse: + """Poll the token endpoint until authorization completes or the device code expires.""" + deadline = monotonic() + device_auth.expires_in + interval = device_auth.interval + + while True: + if monotonic() >= deadline: + raise LoginError(_EXPIRED_TOKEN_MESSAGE) + + try: + response = client.post( + metadata.token_endpoint, + data={ + "grant_type": _DEVICE_GRANT_TYPE, + "device_code": device_auth.device_code, + "client_id": client_id, + "code_verifier": code_verifier, + }, + ) + except httpx.HTTPError as exc: + raise LoginError(f"Device token poll request failed: {exc}") from exc + + if response.status_code == 200: + 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.") + try: + return TokenResponse.from_payload(payload) + except ValueError as exc: + raise LoginError(str(exc)) from exc + + err = OAuthErrorResponse.from_response(response) + if err.error == "expired_token": + raise LoginError(_EXPIRED_TOKEN_MESSAGE) + if err.error == "access_denied": + raise LoginError(_ACCESS_DENIED_MESSAGE) + if err.error == "slow_down": + interval += _SLOW_DOWN_INCREMENT_S + elif err.error != "authorization_pending": + raise LoginError( + err.render( + fallback=f"Token endpoint returned HTTP {response.status_code}", + prefix="Device token poll failed", + ) + ) + + wait_s = min(interval, deadline - monotonic()) + if wait_s > 0: + sleep(wait_s) + + +def run_device_login( + *, + issuer_url: str, + client_id: str, + scopes: tuple[str, ...] = _DEFAULT_SCOPES, + on_device_info: Callable[[DeviceAuthorization], None] | None = None, + http_client: httpx.Client | None = None, + discovery_policy: DiscoveryPolicy = DiscoveryPolicy(), + sleep: Callable[[float], None] = time.sleep, + monotonic: Callable[[], float] = time.monotonic, +) -> LoginResult: + """Run the full device authorization grant. Returns tokens; does **not** persist them.""" + with _http.http_client(http_client, timeout=_DEVICE_AUTH_TIMEOUT_S) as http: + try: + metadata = fetch_provider_metadata( + issuer_url, policy=discovery_policy, client=http + ) + except ValueError as exc: + raise LoginError(str(exc)) from exc + + verifier = generate_verifier() + device_auth = request_device_authorization( + metadata=metadata, + client_id=client_id, + code_challenge=challenge_from_verifier(verifier), + scopes=scopes, + client=http, + ) + if on_device_info is not None: + on_device_info(device_auth) + + token = poll_device_token( + metadata=metadata, + client_id=client_id, + code_verifier=verifier, + device_auth=device_auth, + client=http, + sleep=sleep, + monotonic=monotonic, + ) + return LoginResult(issuer=metadata.issuer, token=token) + + +__all__ = [ + "DeviceAuthorization", + "poll_device_token", + "request_device_authorization", + "run_device_login", +] diff --git a/packages/auth/src/pipefy_auth/discovery.py b/packages/auth/src/pipefy_auth/discovery.py index 73c5e80a..04ddb7cf 100644 --- a/packages/auth/src/pipefy_auth/discovery.py +++ b/packages/auth/src/pipefy_auth/discovery.py @@ -27,6 +27,7 @@ class ProviderMetadata: authorization_endpoint: str token_endpoint: str end_session_endpoint: str | None = None + device_authorization_endpoint: str | None = None @dataclass(frozen=True) @@ -59,10 +60,12 @@ def fetch_provider_metadata( """Fetch and parse the issuer's OIDC discovery document. Validates that ``metadata.issuer`` matches ``issuer_url`` (per OIDC - Discovery 1.0 §4.3) and that the returned ``authorization_endpoint`` and - ``token_endpoint`` aren't pointing at internal hosts or non-HTTPS URLs — - a tampered or misconfigured discovery doc must not be allowed to redirect - the browser dance or token exchange to an attacker-controlled target. + Discovery 1.0 §4.3) and that the returned ``authorization_endpoint``, + ``token_endpoint``, and (when present) ``device_authorization_endpoint`` + aren't pointing at internal hosts or non-HTTPS URLs — a tampered or + misconfigured discovery doc must not be allowed to redirect the browser + dance, device authorization, or token exchange to an attacker-controlled + target. Raises: ValueError: When the discovery document is unreachable, malformed, @@ -111,12 +114,19 @@ def fetch_provider_metadata( token_endpoint = str(data["token_endpoint"]) end_session_endpoint = optional_str(data.get("end_session_endpoint")) + raw_device = data.get("device_authorization_endpoint") + device_authorization_endpoint = str(raw_device) if raw_device else None + endpoints: list[tuple[str, str]] = [ ("authorization_endpoint", authorization_endpoint), ("token_endpoint", token_endpoint), ] if end_session_endpoint is not None: endpoints.append(("end_session_endpoint", end_session_endpoint)) + if device_authorization_endpoint is not None: + endpoints.append( + ("device_authorization_endpoint", device_authorization_endpoint) + ) for field, value in endpoints: try: security.validate_https_url( @@ -130,6 +140,7 @@ def fetch_provider_metadata( authorization_endpoint=authorization_endpoint, token_endpoint=token_endpoint, end_session_endpoint=end_session_endpoint, + device_authorization_endpoint=device_authorization_endpoint, ) diff --git a/packages/cli/src/pipefy_cli/commands/auth.py b/packages/cli/src/pipefy_cli/commands/auth.py index 146bd275..5c19b146 100644 --- a/packages/cli/src/pipefy_cli/commands/auth.py +++ b/packages/cli/src/pipefy_cli/commands/auth.py @@ -16,6 +16,7 @@ TransportServerError, ) from pipefy_auth import ( + DeviceAuthorization, DiscoveryPolicy, LoginError, OidcClient, @@ -28,6 +29,7 @@ keychain_backend_name, load_session, revoke_session, + run_device_login, run_login, store_session, ) @@ -90,6 +92,7 @@ def signed_in(self) -> bool: _SERVICE_ACCOUNT_ENV_KEYS = tuple(_LEGACY_ENV_KEYS_TO_NEW.values()) _LEGACY_SERVICE_ACCOUNT_ENV_KEYS = tuple(_LEGACY_ENV_KEYS_TO_NEW.keys()) +_DEFAULT_CALLBACK_TIMEOUT_S = 180.0 def _session_masking_env_vars() -> list[str]: @@ -125,13 +128,18 @@ def _warn_if_masked() -> None: @auth_app.command("login") def auth_login( ctx: typer.Context, + device: bool = typer.Option( # noqa: B008 (Typer Option pattern) + False, + "--device", + help="Use the OAuth 2.0 device authorization grant (headless-friendly).", + ), no_browser: bool = typer.Option( # noqa: B008 (Typer Option pattern) False, "--no-browser", help="Print the authorization URL instead of opening a browser.", ), callback_timeout: float = typer.Option( # noqa: B008 - 180.0, + _DEFAULT_CALLBACK_TIMEOUT_S, "--callback-timeout", help="Seconds to wait for the browser callback before giving up.", min=5.0, @@ -141,6 +149,18 @@ def auth_login( # Lazy to keep keyring's ~30-80ms backend-discovery cost off every CLI startup. from keyring.errors import KeyringError + if device: + if no_browser: + raise typer.BadParameter( + "--no-browser is incompatible with --device " + "(the device flow has no browser-callback dance to suppress)." + ) + if callback_timeout != _DEFAULT_CALLBACK_TIMEOUT_S: + raise typer.BadParameter( + "--callback-timeout is incompatible with --device " + "(the device deadline is governed by the IdP's RFC 8628 expires_in)." + ) + settings, auth = settings_and_auth_from_ctx(ctx) if auth.oidc_client is None: typer.echo( @@ -170,16 +190,35 @@ def _open(url: str) -> bool: return False try: - result = run_login( - issuer_url=issuer_url, - client_id=client_id, - callback_timeout_s=callback_timeout, - open_browser=_open, - on_url=_print_url, - discovery_policy=DiscoveryPolicy( - allow_insecure_urls=settings.allow_insecure_urls - ), - ) + if device: + + def _print_device_info(dev: DeviceAuthorization) -> None: + typer.echo("") + typer.echo(f"User code: {dev.user_code}") + typer.echo(f"Visit: {dev.verification_uri}") + if dev.verification_uri_complete: + typer.echo(f"Or open: {dev.verification_uri_complete}") + typer.echo("") + + result = run_device_login( + issuer_url=issuer_url, + client_id=client_id, + on_device_info=_print_device_info, + discovery_policy=DiscoveryPolicy( + allow_insecure_urls=settings.allow_insecure_urls + ), + ) + else: + result = run_login( + issuer_url=issuer_url, + client_id=client_id, + callback_timeout_s=callback_timeout, + open_browser=_open, + on_url=_print_url, + discovery_policy=DiscoveryPolicy( + allow_insecure_urls=settings.allow_insecure_urls + ), + ) except LoginError as exc: typer.echo(f"Login failed: {exc}", err=True) raise typer.Exit(1) from exc diff --git a/packages/cli/tests/fixtures/cli_help_golden.txt b/packages/cli/tests/fixtures/cli_help_golden.txt index 4a7a8513..f16b3d3e 100644 --- a/packages/cli/tests/fixtures/cli_help_golden.txt +++ b/packages/cli/tests/fixtures/cli_help_golden.txt @@ -488,6 +488,9 @@ Usage: pipefy auth login [OPTIONS] Sign in to Pipefy via your browser and store the session in the OS keychain. ╭─ Options ────────────────────────────────────────────────────────────────────╮ +│ --device Use the OAuth 2.0 device │ +│ authorization grant │ +│ (headless-friendly). │ │ --no-browser Print the authorization URL │ │ instead of opening a │ │ browser. │ diff --git a/packages/cli/tests/test_auth_device.py b/packages/cli/tests/test_auth_device.py new file mode 100644 index 00000000..b1711641 --- /dev/null +++ b/packages/cli/tests/test_auth_device.py @@ -0,0 +1,583 @@ +"""Unit tests for OAuth device authorization grant (``pipefy auth login --device``).""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +import httpx +import pytest +from conftest import InMemoryKeyring +from pipefy_auth import device, discovery, storage +from pipefy_auth.flow import LoginError, LoginResult +from pipefy_auth.responses import TokenResponse +from typer.testing import CliRunner + +from pipefy_cli.main import app as cli_app + + +@pytest.fixture +def cli_runner() -> CliRunner: + return CliRunner(mix_stderr=False) + + +_ISSUER = "https://x.test/realms/foo" +_DEVICE_PATH = "/protocol/openid-connect/auth/device" +_TOKEN_PATH = "/protocol/openid-connect/token" + + +def _discovery_json(*, include_device: bool = True) -> dict[str, Any]: + body: dict[str, Any] = { + "issuer": _ISSUER, + "authorization_endpoint": f"{_ISSUER}/protocol/openid-connect/auth", + "token_endpoint": f"{_ISSUER}{_TOKEN_PATH}", + } + if include_device: + body["device_authorization_endpoint"] = f"{_ISSUER}{_DEVICE_PATH}" + return body + + +def _token_success() -> dict[str, Any]: + return { + "access_token": "AAA", + "refresh_token": "RRR", + "token_type": "Bearer", + "expires_in": 300, + } + + +def _clock_sleep_pair() -> tuple[ + list[float], Callable[[float], None], Callable[[], float] +]: + clock: list[float] = [0.0] + + def monotonic() -> float: + return clock[0] + + def sleep(dt: float) -> None: + clock[0] += dt + + return clock, sleep, monotonic + + +def _device_meta() -> discovery.ProviderMetadata: + return discovery.ProviderMetadata( + issuer=_ISSUER, + authorization_endpoint=f"{_ISSUER}/auth", + token_endpoint=f"{_ISSUER}{_TOKEN_PATH}", + device_authorization_endpoint=f"{_ISSUER}{_DEVICE_PATH}", + ) + + +def _device_payload(**overrides: Any) -> dict[str, Any]: + payload: dict[str, Any] = { + "device_code": "dc", + "user_code": "ABCD", + "verification_uri": "https://x.test/verify", + "expires_in": 60, + } + payload.update(overrides) + return payload + + +class TestDeviceAuthorizationFromPayload: + def test_minimal_payload_parses(self) -> None: + dev = device.DeviceAuthorization.from_payload(_device_payload()) + assert dev.device_code == "dc" + assert dev.user_code == "ABCD" + assert dev.expires_in == 60.0 + assert dev.interval == 5.0 # default when absent + assert dev.verification_uri_complete is None + + @pytest.mark.parametrize( + "missing_field", + ("device_code", "user_code", "verification_uri", "expires_in"), + ) + def test_missing_required_field_raises_value_error( + self, missing_field: str + ) -> None: + payload = _device_payload() + del payload[missing_field] + with pytest.raises(ValueError, match=missing_field): + device.DeviceAuthorization.from_payload(payload) + + def test_invalid_expires_in_raises_value_error(self) -> None: + with pytest.raises(ValueError, match="invalid expires_in"): + device.DeviceAuthorization.from_payload( + _device_payload(expires_in="not-a-number") + ) + + def test_invalid_interval_raises_value_error(self) -> None: + with pytest.raises(ValueError, match="invalid interval"): + device.DeviceAuthorization.from_payload( + _device_payload(interval="not-a-number") + ) + + def test_zero_interval_defaults_to_five(self) -> None: + dev = device.DeviceAuthorization.from_payload(_device_payload(interval=0)) + assert dev.interval == 5.0 + + def test_empty_verification_uri_complete_becomes_none(self) -> None: + dev = device.DeviceAuthorization.from_payload( + _device_payload(verification_uri_complete="") + ) + assert dev.verification_uri_complete is None + + +class TestRequestDeviceAuthorization: + def test_missing_endpoint_raises_login_error(self) -> None: + meta = discovery.ProviderMetadata( + issuer=_ISSUER, + authorization_endpoint=f"{_ISSUER}/auth", + token_endpoint=f"{_ISSUER}{_TOKEN_PATH}", + device_authorization_endpoint=None, + ) + client = httpx.Client( + transport=httpx.MockTransport(lambda _r: httpx.Response(404)) + ) + with pytest.raises(LoginError, match="device_authorization_endpoint"): + device.request_device_authorization( + metadata=meta, + client_id="pipefy-cli", + code_challenge="test-challenge", + client=client, + ) + + def test_non_200_renders_oauth_error(self) -> None: + def handler(_r: httpx.Request) -> httpx.Response: + return httpx.Response( + 400, + json={"error": "invalid_client", "error_description": "bad client"}, + ) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + with pytest.raises(LoginError, match="invalid_client.*bad client"): + device.request_device_authorization( + metadata=_device_meta(), + client_id="pipefy-cli", + code_challenge="test-challenge", + client=client, + ) + + def test_non_json_body_raises_login_error(self) -> None: + def handler(_r: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="oops") + + client = httpx.Client(transport=httpx.MockTransport(handler)) + with pytest.raises(LoginError, match="non-JSON"): + device.request_device_authorization( + metadata=_device_meta(), + client_id="pipefy-cli", + code_challenge="test-challenge", + client=client, + ) + + def test_parse_failure_wraps_value_error_as_login_error(self) -> None: + def handler(_r: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=_device_payload(expires_in="abc")) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + with pytest.raises(LoginError, match="invalid expires_in"): + device.request_device_authorization( + metadata=_device_meta(), + client_id="pipefy-cli", + code_challenge="test-challenge", + client=client, + ) + + +class TestDiscoveryDeviceEndpoint: + def test_parses_and_validates_device_endpoint(self) -> None: + payload = _discovery_json() + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/openid-configuration") + return httpx.Response(200, json=payload) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + meta = discovery.fetch_provider_metadata(_ISSUER, client=client) + assert ( + meta.device_authorization_endpoint + == payload["device_authorization_endpoint"] + ) + + def test_missing_device_endpoint_is_none(self) -> None: + payload = _discovery_json(include_device=False) + client = httpx.Client( + transport=httpx.MockTransport(lambda _r: httpx.Response(200, json=payload)) + ) + meta = discovery.fetch_provider_metadata(_ISSUER, client=client) + assert meta.device_authorization_endpoint is None + + def test_invalid_device_endpoint_ssrf(self) -> None: + payload = _discovery_json() + payload["device_authorization_endpoint"] = "https://127.0.0.1/device" + client = httpx.Client( + transport=httpx.MockTransport(lambda _r: httpx.Response(200, json=payload)) + ) + with pytest.raises(ValueError, match="device_authorization_endpoint"): + discovery.fetch_provider_metadata(_ISSUER, client=client) + + +class TestRunDeviceLogin: + def _client_for( + self, handler: Callable[[httpx.Request], httpx.Response] + ) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(handler)) + + def test_sends_pkce_on_device_auth_and_token_poll(self) -> None: + from urllib.parse import parse_qs + + from pipefy_auth.pkce import challenge_from_verifier + + captured: dict[str, dict[str, list[str]]] = {} + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + captured["device"] = parse_qs(request.content.decode("utf-8")) + return httpx.Response(200, json=_device_payload(interval=0)) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + captured["token"] = parse_qs(request.content.decode("utf-8")) + return httpx.Response(200, json=_token_success()) + return httpx.Response(404) + + _, sleep, monotonic = _clock_sleep_pair() + device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + ) + + device_form = captured["device"] + assert device_form["code_challenge_method"] == ["S256"] + challenge = device_form["code_challenge"][0] + verifier = captured["token"]["code_verifier"][0] + # Challenge must derive from the verifier sent on the poll — without + # this binding a PKCE-enforcing IdP rejects the token exchange even if + # the device-auth call accepted the params. + assert challenge_from_verifier(verifier) == challenge + + def test_happy_path(self) -> None: + polls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + return httpx.Response( + 200, + json={ + "device_code": "devc", + "user_code": "ABCD-EFGH", + "verification_uri": "https://x.test/verify", + "expires_in": 120, + "interval": 0, + }, + ) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + polls["n"] += 1 + return httpx.Response(200, json=_token_success()) + return httpx.Response(404) + + clock, sleep, monotonic = _clock_sleep_pair() + infos: list[device.DeviceAuthorization] = [] + + result = device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + on_device_info=infos.append, + ) + assert isinstance(result, LoginResult) + assert result.issuer == _ISSUER + assert result.token.access_token == "AAA" + assert polls["n"] == 1 + # First poll fires immediately (RFC 8628 §3.5: the wait applies + # *between* requests after authorization_pending, not before the first). + assert clock[0] == 0.0 + assert len(infos) == 1 + assert infos[0].user_code == "ABCD-EFGH" + + def test_authorization_pending_then_success(self) -> None: + polls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + return httpx.Response( + 200, + json={ + "device_code": "devc", + "user_code": "ZZZZ-ZZZZ", + "verification_uri": "https://x.test/verify", + "expires_in": 120, + "interval": 0, + }, + ) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + n = polls["n"] + polls["n"] += 1 + if n == 0: + return httpx.Response(400, json={"error": "authorization_pending"}) + return httpx.Response(200, json=_token_success()) + return httpx.Response(404) + + _, sleep, monotonic = _clock_sleep_pair() + result = device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + ) + assert result.token.refresh_token == "RRR" + assert polls["n"] == 2 + + def test_slow_down_then_success(self) -> None: + polls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + return httpx.Response( + 200, + json={ + "device_code": "devc", + "user_code": "AAAA-BBBB", + "verification_uri": "https://x.test/verify", + "expires_in": 120, + "interval": 0, + }, + ) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + n = polls["n"] + polls["n"] += 1 + if n == 0: + return httpx.Response(400, json={"error": "slow_down"}) + return httpx.Response(200, json=_token_success()) + return httpx.Response(404) + + _, sleep, monotonic = _clock_sleep_pair() + result = device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + ) + assert result.token.access_token == "AAA" + assert polls["n"] == 2 + + def test_expired_token_raises(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + return httpx.Response( + 200, + json={ + "device_code": "devc", + "user_code": "XXXX-YYYY", + "verification_uri": "https://x.test/verify", + "expires_in": 120, + "interval": 0, + }, + ) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + return httpx.Response(400, json={"error": "expired_token"}) + return httpx.Response(404) + + _, sleep, monotonic = _clock_sleep_pair() + with pytest.raises(LoginError, match="didn't complete in time"): + device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + ) + + def test_client_deadline_expires_without_server_expired_token(self) -> None: + polls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + return httpx.Response( + 200, + json={ + "device_code": "devc", + "user_code": "TIME-OUT", + "verification_uri": "https://x.test/verify", + "expires_in": 10, + "interval": 5, + }, + ) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + polls["n"] += 1 + return httpx.Response(400, json={"error": "authorization_pending"}) + return httpx.Response(404) + + _, sleep, monotonic = _clock_sleep_pair() + with pytest.raises(LoginError, match="didn't complete in time"): + device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + ) + assert polls["n"] == 2 + + def test_access_denied_raises(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if request.method == "GET" and path.endswith("/openid-configuration"): + return httpx.Response(200, json=_discovery_json()) + if request.method == "POST" and path.endswith(_DEVICE_PATH): + return httpx.Response( + 200, + json={ + "device_code": "devc", + "user_code": "QQQQ-WWWW", + "verification_uri": "https://x.test/verify", + "expires_in": 120, + "interval": 0, + }, + ) + if request.method == "POST" and path.endswith(_TOKEN_PATH): + return httpx.Response(400, json={"error": "access_denied"}) + return httpx.Response(404) + + _, sleep, monotonic = _clock_sleep_pair() + with pytest.raises(LoginError, match="cancelled"): + device.run_device_login( + issuer_url=_ISSUER, + client_id="pipefy-cli", + http_client=self._client_for(handler), + sleep=sleep, + monotonic=monotonic, + ) + + +class TestAuthLoginDeviceCommand: + def test_device_login_happy_path( + self, + cli_runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, + fake_keyring: InMemoryKeyring, + clean_pipefy_env: None, + saved_cwd: object, + ) -> None: + monkeypatch.setenv("PIPEFY_AUTH_URL", _ISSUER) + monkeypatch.setenv("PIPEFY_AUTH_CLIENT_ID", "pipefy-cli") + + from pipefy_auth.device import DeviceAuthorization + + from pipefy_cli.commands import auth as auth_module + + def _fake_run_device_login(**kwargs: object) -> LoginResult: + cb = kwargs.get("on_device_info") + if cb is not None: + cb( + DeviceAuthorization( + device_code="dc", + user_code="USER-CODE", + verification_uri="https://x.test/v", + verification_uri_complete="https://x.test/vc", + expires_in=60.0, + interval=1.0, + ) + ) + return LoginResult( + issuer=_ISSUER, + token=TokenResponse( + access_token="AAA", + refresh_token="RRR", + expires_in=300, + ), + ) + + monkeypatch.setattr(auth_module, "run_device_login", _fake_run_device_login) + + result = cli_runner.invoke(cli_app, ["auth", "login", "--device"]) + assert result.exit_code == 0, result.stderr + assert "USER-CODE" in result.stdout + assert "https://x.test/vc" in result.stdout + assert "Signed in to Pipefy" in result.stdout + + loaded = storage.load_session(issuer=_ISSUER, client_id="pipefy-cli") + assert loaded is not None + assert loaded.token.refresh_token == "RRR" + + def test_device_with_no_browser_errors( + self, + cli_runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, + fake_keyring: InMemoryKeyring, + clean_pipefy_env: None, + saved_cwd: object, + ) -> None: + monkeypatch.setenv("PIPEFY_AUTH_URL", _ISSUER) + result = cli_runner.invoke( + cli_app, ["auth", "login", "--device", "--no-browser"] + ) + assert result.exit_code != 0 + assert "--no-browser is incompatible" in result.stderr + + def test_device_with_explicit_callback_timeout_errors( + self, + cli_runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, + fake_keyring: InMemoryKeyring, + clean_pipefy_env: None, + saved_cwd: object, + ) -> None: + monkeypatch.setenv("PIPEFY_AUTH_URL", _ISSUER) + result = cli_runner.invoke( + cli_app, ["auth", "login", "--device", "--callback-timeout", "60"] + ) + assert result.exit_code != 0 + assert "--callback-timeout is incompatible" in result.stderr + + def test_device_with_default_callback_timeout_ok( + self, + cli_runner: CliRunner, + monkeypatch: pytest.MonkeyPatch, + fake_keyring: InMemoryKeyring, + clean_pipefy_env: None, + saved_cwd: object, + ) -> None: + # Guard against false positives in the parameter-source check: invoking + # --device WITHOUT --callback-timeout must not error out on the conflict + # detector. + monkeypatch.setenv("PIPEFY_AUTH_URL", _ISSUER) + monkeypatch.setenv("PIPEFY_AUTH_CLIENT_ID", "pipefy-cli") + + from pipefy_cli.commands import auth as auth_module + + def _fake_run_device_login(**_kwargs: object) -> LoginResult: + return LoginResult( + issuer=_ISSUER, + token=TokenResponse(access_token="A", refresh_token="R"), + ) + + monkeypatch.setattr(auth_module, "run_device_login", _fake_run_device_login) + + result = cli_runner.invoke(cli_app, ["auth", "login", "--device"]) + assert result.exit_code == 0, result.stderr From d46f20e6d67f870449fdf2ee8451f80b4217770a Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 1 Jun 2026 15:27:16 -0300 Subject: [PATCH 2/3] test(cli): strip Rich ANSI codes from --device flag-conflict assertions Linux CI runs under FORCE_COLOR=1, where Rich/Typer splits ``--no-browser`` / ``--callback-timeout`` with bold ANSI escapes between hyphens, breaking the literal substring match. Mirror the same _ANSI_ESCAPE_RE pattern already used in test_table_record_commands.py. --- packages/cli/tests/test_auth_device.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/packages/cli/tests/test_auth_device.py b/packages/cli/tests/test_auth_device.py index b1711641..37ff0187 100644 --- a/packages/cli/tests/test_auth_device.py +++ b/packages/cli/tests/test_auth_device.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from collections.abc import Callable from typing import Any @@ -15,6 +16,12 @@ from pipefy_cli.main import app as cli_app +# Rich/Typer renders option names like ``--no-browser`` in bold by default, +# splitting the dashes with ``\x1b[1m`` ANSI codes on Linux CI runners under +# ``FORCE_COLOR=1``. Strip them before the substring assert so the test passes +# on both macOS and Linux without depending on env vars. +_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") + @pytest.fixture def cli_runner() -> CliRunner: @@ -538,7 +545,7 @@ def test_device_with_no_browser_errors( cli_app, ["auth", "login", "--device", "--no-browser"] ) assert result.exit_code != 0 - assert "--no-browser is incompatible" in result.stderr + assert "--no-browser is incompatible" in _ANSI_ESCAPE_RE.sub("", result.stderr) def test_device_with_explicit_callback_timeout_errors( self, @@ -553,7 +560,9 @@ def test_device_with_explicit_callback_timeout_errors( cli_app, ["auth", "login", "--device", "--callback-timeout", "60"] ) assert result.exit_code != 0 - assert "--callback-timeout is incompatible" in result.stderr + assert "--callback-timeout is incompatible" in _ANSI_ESCAPE_RE.sub( + "", result.stderr + ) def test_device_with_default_callback_timeout_ok( self, From a9e33cb602a2c30d422ee6bc60aa26edb81b9981 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 1 Jun 2026 20:50:11 -0300 Subject: [PATCH 3/3] refactor(auth): migrate DeviceAuthorization to Pydantic v2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move DeviceAuthorization from a frozen dataclass in device.py to a Pydantic BaseModel in responses.py, matching the TokenResponse / OAuthErrorResponse pattern established in #258. - StrictInt on expires_in and interval (rejects bool, str, float) - _OptionalStr (BeforeValidator strip-or-none) on verification_uri_complete - field_validator clamps non-positive interval to the RFC 8628 §3.5 default - device.py routes both from_payload failures through _format_validation_error for consistent LoginError shape with flow.py --- packages/auth/src/pipefy_auth/device.py | 78 ++++------------------ packages/auth/src/pipefy_auth/responses.py | 43 ++++++++++-- packages/cli/tests/test_auth_device.py | 16 ++--- 3 files changed, 58 insertions(+), 79 deletions(-) diff --git a/packages/auth/src/pipefy_auth/device.py b/packages/auth/src/pipefy_auth/device.py index db5aff20..d0e20979 100644 --- a/packages/auth/src/pipefy_auth/device.py +++ b/packages/auth/src/pipefy_auth/device.py @@ -3,11 +3,10 @@ from __future__ import annotations import time -from dataclasses import dataclass from typing import Callable import httpx -from pipefy_infra.coerce import optional_str +from pydantic import ValidationError from pipefy_auth import _http from pipefy_auth.discovery import ( @@ -17,12 +16,16 @@ ) from pipefy_auth.flow import LoginError, LoginResult from pipefy_auth.pkce import challenge_from_verifier, generate_verifier -from pipefy_auth.responses import OAuthErrorResponse, TokenResponse +from pipefy_auth.responses import ( + DeviceAuthorization, + OAuthErrorResponse, + TokenResponse, + _format_validation_error, +) _DEFAULT_SCOPES = ("openid", "profile", "email", "offline_access") _DEVICE_AUTH_TIMEOUT_S = 30.0 -_DEFAULT_POLL_INTERVAL_S = 5.0 -_SLOW_DOWN_INCREMENT_S = 5.0 +_SLOW_DOWN_INCREMENT_S = 5 _DEVICE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:device_code" _EXPIRED_TOKEN_MESSAGE = ( @@ -31,63 +34,6 @@ _ACCESS_DENIED_MESSAGE = "Sign-in cancelled by the user." -@dataclass(frozen=True) -class DeviceAuthorization: - """The device-authorization response the user must act on (RFC 8628 §3.2).""" - - device_code: str - user_code: str - verification_uri: str - verification_uri_complete: str | None - expires_in: float - interval: float - - @classmethod - def from_payload(cls, payload: dict[str, object]) -> "DeviceAuthorization": - """Parse a device-authorization JSON body. Raises ``ValueError`` on malformed input.""" - missing = [ - key - for key in ("device_code", "user_code", "verification_uri") - if not payload.get(key) - ] - if "expires_in" not in payload or payload["expires_in"] is None: - missing.append("expires_in") - if missing: - raise ValueError( - "Device authorization response is missing required fields: " - + ", ".join(missing) - ) - - expires_in_raw = payload["expires_in"] - try: - expires_in = float(expires_in_raw) # type: ignore[arg-type] - except (TypeError, ValueError) as exc: - raise ValueError( - f"Device authorization response has invalid expires_in: {expires_in_raw!r}" - ) from exc - - interval_raw = payload.get("interval", _DEFAULT_POLL_INTERVAL_S) - try: - interval = float(interval_raw) # type: ignore[arg-type] - except (TypeError, ValueError) as exc: - raise ValueError( - f"Device authorization response has invalid interval: {interval_raw!r}" - ) from exc - if interval <= 0: - interval = _DEFAULT_POLL_INTERVAL_S - - return cls( - device_code=str(payload["device_code"]), - user_code=str(payload["user_code"]), - verification_uri=str(payload["verification_uri"]), - verification_uri_complete=optional_str( - payload.get("verification_uri_complete") - ), - expires_in=expires_in, - interval=interval, - ) - - def request_device_authorization( *, metadata: ProviderMetadata, @@ -146,8 +92,8 @@ def request_device_authorization( try: return DeviceAuthorization.from_payload(payload) - except ValueError as exc: - raise LoginError(str(exc)) from exc + except ValidationError as exc: + raise LoginError(_format_validation_error(exc)) from exc def poll_device_token( @@ -192,8 +138,8 @@ def poll_device_token( raise LoginError("Token endpoint returned a non-object JSON payload.") try: return TokenResponse.from_payload(payload) - except ValueError as exc: - raise LoginError(str(exc)) from exc + except ValidationError as exc: + raise LoginError(_format_validation_error(exc)) from exc err = OAuthErrorResponse.from_response(response) if err.error == "expired_token": diff --git a/packages/auth/src/pipefy_auth/responses.py b/packages/auth/src/pipefy_auth/responses.py index 521322af..1bc9a88e 100644 --- a/packages/auth/src/pipefy_auth/responses.py +++ b/packages/auth/src/pipefy_auth/responses.py @@ -1,9 +1,11 @@ """Typed OAuth response wrappers (RFC 6749 §5.1 success, §5.2 error; -RFC 8628 §3.5 error). +RFC 8628 §3.2 device authorization, §3.5 polling error). -``TokenResponse`` parses the token-endpoint success body. ``OAuthErrorResponse`` -parses the error envelope; the call site supplies the rendering prefix and -fallback so endpoint context stays out of the wire-shape model. +``TokenResponse`` parses the token-endpoint success body. +``DeviceAuthorization`` parses the device-authorization endpoint success body. +``OAuthErrorResponse`` parses the error envelope; the call site supplies the +rendering prefix and fallback so endpoint context stays out of the wire-shape +model. """ from __future__ import annotations @@ -19,8 +21,11 @@ StrictInt, StrictStr, ValidationError, + field_validator, ) +_DEFAULT_POLL_INTERVAL_S = 5 + def _strip_or_none(value: object) -> str | None: """Strip strings to ``None`` on empty; drop non-strings to ``None``. @@ -63,6 +68,34 @@ def from_payload(cls, payload: dict[str, object]) -> "TokenResponse": return cls.model_validate(payload) +class DeviceAuthorization(BaseModel): + """Parsed OAuth 2.0 device-authorization success response (RFC 8628 §3.2). + + Strict typing matches :class:`TokenResponse`. ``interval`` defaults to 5s + per RFC 8628 §3.5 when absent, and any non-positive server-sent value is + clamped to the same default (a zero or negative poll cadence would either + busy-loop or never poll). + """ + + model_config = ConfigDict(extra="ignore", frozen=True) + + device_code: Annotated[StrictStr, Field(min_length=1)] + user_code: Annotated[StrictStr, Field(min_length=1)] + verification_uri: Annotated[StrictStr, Field(min_length=1)] + verification_uri_complete: _OptionalStr = None + expires_in: StrictInt + interval: StrictInt = _DEFAULT_POLL_INTERVAL_S + + @field_validator("interval", mode="after") + @classmethod + def _clamp_nonpositive_interval(cls, value: int) -> int: + return value if value > 0 else _DEFAULT_POLL_INTERVAL_S + + @classmethod + def from_payload(cls, payload: dict[str, object]) -> "DeviceAuthorization": + return cls.model_validate(payload) + + class OAuthErrorResponse(BaseModel): """Parsed OAuth error envelope (RFC 6749 §5.2 / RFC 8628 §3.5). @@ -123,4 +156,4 @@ def _format_validation_error(exc: ValidationError) -> str: return "; ".join(clauses) or "Response failed validation." -__all__ = ["OAuthErrorResponse", "TokenResponse"] +__all__ = ["DeviceAuthorization", "OAuthErrorResponse", "TokenResponse"] diff --git a/packages/cli/tests/test_auth_device.py b/packages/cli/tests/test_auth_device.py index 37ff0187..4914061f 100644 --- a/packages/cli/tests/test_auth_device.py +++ b/packages/cli/tests/test_auth_device.py @@ -92,8 +92,8 @@ def test_minimal_payload_parses(self) -> None: dev = device.DeviceAuthorization.from_payload(_device_payload()) assert dev.device_code == "dc" assert dev.user_code == "ABCD" - assert dev.expires_in == 60.0 - assert dev.interval == 5.0 # default when absent + assert dev.expires_in == 60 + assert dev.interval == 5 # default when absent assert dev.verification_uri_complete is None @pytest.mark.parametrize( @@ -109,20 +109,20 @@ def test_missing_required_field_raises_value_error( device.DeviceAuthorization.from_payload(payload) def test_invalid_expires_in_raises_value_error(self) -> None: - with pytest.raises(ValueError, match="invalid expires_in"): + with pytest.raises(ValueError, match="expires_in"): device.DeviceAuthorization.from_payload( _device_payload(expires_in="not-a-number") ) def test_invalid_interval_raises_value_error(self) -> None: - with pytest.raises(ValueError, match="invalid interval"): + with pytest.raises(ValueError, match="interval"): device.DeviceAuthorization.from_payload( _device_payload(interval="not-a-number") ) def test_zero_interval_defaults_to_five(self) -> None: dev = device.DeviceAuthorization.from_payload(_device_payload(interval=0)) - assert dev.interval == 5.0 + assert dev.interval == 5 def test_empty_verification_uri_complete_becomes_none(self) -> None: dev = device.DeviceAuthorization.from_payload( @@ -184,7 +184,7 @@ def handler(_r: httpx.Request) -> httpx.Response: return httpx.Response(200, json=_device_payload(expires_in="abc")) client = httpx.Client(transport=httpx.MockTransport(handler)) - with pytest.raises(LoginError, match="invalid expires_in"): + with pytest.raises(LoginError, match="expires_in"): device.request_device_authorization( metadata=_device_meta(), client_id="pipefy-cli", @@ -507,8 +507,8 @@ def _fake_run_device_login(**kwargs: object) -> LoginResult: user_code="USER-CODE", verification_uri="https://x.test/v", verification_uri_complete="https://x.test/vc", - expires_in=60.0, - interval=1.0, + expires_in=60, + interval=1, ) ) return LoginResult(