diff --git a/python/src/basecamp/oauth/_transport.py b/python/src/basecamp/oauth/_transport.py new file mode 100644 index 000000000..66a6d5367 --- /dev/null +++ b/python/src/basecamp/oauth/_transport.py @@ -0,0 +1,162 @@ +"""Bounded HTTP transport shared by the OAuth discovery and device-flow fetches. + +Sync httpx has NO total-request timeout: its timeout is per-read and RESETS on +every received chunk, so a peer dripping header or body bytes just under that +interval can hold a request open indefinitely (verified), and closing a sync +client from a watchdog thread does not interrupt a blocked read either. The +core here runs an async client under ``asyncio.wait_for`` on a dedicated +worker thread, so the deadline CANCELS the request (and closes the socket) — +the caller is bounded regardless of chunk cadence AND the work is actually +terminated, no leaked connection. +""" + +from __future__ import annotations + +import asyncio +import math +import threading +from collections.abc import Callable + +import httpx + +from basecamp.oauth.errors import OAuthError + +#: Extra time (seconds) to let a timed-out request's async cancellation/cleanup +#: unwind before the caller abandons the (daemon) worker and returns a timeout. +#: Kept SMALL so the caller's worst-case block stays ~timeout: cleanup after a +#: wait_for cancellation is just closing a socket, and joining with no grace at +#: all would race a request that completes right at the deadline. The daemon +#: worker never blocks interpreter exit if even this stalls. +_WORKER_JOIN_GRACE = 1.0 + +#: Upper bound (seconds) on a bounded request timeout. A per-request timeout +#: beyond this is nonsensical, and a huge finite value would overflow the +#: wall-clock wait primitive (asyncio.wait_for / thread join); callers clamp +#: to their operation default above it (see ``_normalize_timeout``). +_MAX_REQUEST_TIMEOUT = 3600.0 + + +def request_bounded( + method: str, + url: str, + *, + headers: dict[str, str], + params: dict[str, str] | None = None, + timeout: float, + max_body_bytes: int, + read_body: Callable[[int], bool] = lambda _status: True, + context: str = "OAuth", +) -> tuple[int, bytes]: + """SSRF-hardened request: suppress redirects, bound the WHOLE round-trip by + ``timeout``, and read the body under a genuine streaming cap that aborts once + ``max_body_bytes`` is exceeded (never a post-hoc check on an already-buffered + body). ``params``, when given, is sent as a form body (POST). + + ``timeout`` must arrive ALREADY normalized — finite, positive, and no greater + than :data:`_MAX_REQUEST_TIMEOUT` (callers run it through + ``discovery._normalize_timeout``, which this module cannot import without a + cycle). An unnormalized value would disable the ``wait_for`` deadline + (``inf`` never fires) or overflow the wait primitive. + + ``read_body(status)`` decides — from the response status, known once headers + arrive — whether the body is drained. A caller returns ``False`` for statuses + whose body it does not use, so a slow/never-ending body cannot time out + mid-read and be misclassified as a retryable transport failure instead of + the api_error the status already is. + + Transport failures propagate as :class:`httpx.HTTPError` (incl. + :class:`httpx.TimeoutException`) so callers classify them; an oversized body + raises :class:`OAuthError` (``api_error``). ``context`` labels both messages. + """ + + if params is not None and method != "POST": + # A form body on a non-POST would emit e.g. a GET-with-body — commonly + # rejected server-side and hard to debug; fail fast on the misuse. + raise ValueError("request_bounded: params (a form body) is only valid with POST") + + # Defensive enforcement of the caller-normalization contract: an inf/nan/ + # non-positive/oversized timeout would disable or overflow the wait_for + # deadline and the thread join — the very bound this module exists to + # provide. Callers run through _normalize_timeout; a violation here is a + # programming error, so fail fast. Range checks run BEFORE math.isfinite: + # isfinite converts an int to float, so an astronomically large int + # (10**400) would raise OverflowError out of the guard instead of this + # ValueError; int/float comparisons never convert, and NaN compares False + # on both, falling through to the isfinite reject. + if ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or timeout <= 0 + or timeout > _MAX_REQUEST_TIMEOUT + or not math.isfinite(timeout) + ): + raise ValueError( + "request_bounded: timeout must be a finite positive number of seconds " + f"no greater than {_MAX_REQUEST_TIMEOUT}" + ) + + async def _do() -> tuple[int, bytes]: + async with ( + httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client, + client.stream(method, url, data=params, headers=headers) as response, + ): + if not read_body(response.status_code): + return response.status_code, b"" + chunks: list[bytes] = [] + total = 0 + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > max_body_bytes: + # An oversized body is api_error, not a timeout — abort the + # stream so it is never fully buffered. + raise OAuthError("api_error", f"{context} response exceeds size cap") + chunks.append(chunk) + return response.status_code, b"".join(chunks) + + # httpx's timeout is per-read (it resets on every received chunk) and httpx has + # NO total-request timeout, so a peer slow-dripping header or body bytes just + # under that interval could otherwise hold the request open indefinitely + # (verified); closing a sync client from a watchdog does not interrupt a blocked + # read either. asyncio.wait_for CANCELS the request (and closes the socket) at + # the deadline — the caller is bounded AND the socket work is actually + # terminated. Known residual: a getaddrinfo blocked in the OS resolver runs on + # an executor thread that cancellation cannot interrupt, so a slow-DNS attempt + # leaves a daemon worker alive until the RESOLVER's own timeout (seconds to + # ~30s, OS-bounded) — the caller still returns at the deadline, the stragglers + # are bounded in lifetime by the resolver and in count by the attempt rate, + # and they never block interpreter exit. Sync Python offers no stronger + # DNS bound short of process isolation. + # + # Run it in a DEDICATED thread with its own event loop rather than calling + # asyncio.run() here: this sync helper may be invoked from code that already has + # a running loop (Jupyter/FastAPI/async CLI), where asyncio.run() raises + # RuntimeError before any request is made. wait_for bounds the thread's work at + # ~timeout, so the bounded join below normally returns almost immediately; the + # is_alive backstop after it covers only a pathological async-cleanup hang. + result: list[tuple[int, bytes]] = [] + error: list[Exception] = [] + + def _runner() -> None: + try: + result.append(asyncio.run(asyncio.wait_for(_do(), timeout))) + except Exception as exc: # captured and re-raised on the caller thread + error.append(exc) + + worker = threading.Thread(target=_runner, daemon=True) + worker.start() + # asyncio.wait_for cancels the request at `timeout`, so the worker normally + # finishes well within it. Join with a small grace for the cancellation/cleanup + # to unwind; if even that stalls (a pathological async cleanup hang), return a + # timeout rather than block the caller — the daemon worker never blocks + # interpreter exit. This is bounded AND non-leaking in every non-pathological case. + worker.join(timeout + _WORKER_JOIN_GRACE) + if worker.is_alive(): + raise httpx.ReadTimeout(f"{context} request exceeded the timeout deadline") + if error: + exc = error[0] + # On Python >= 3.11 (this package's floor) asyncio.TimeoutError IS the + # builtin TimeoutError, so this catches wait_for's deadline expiry. + if isinstance(exc, TimeoutError): + raise httpx.ReadTimeout(f"{context} request exceeded the timeout deadline") from exc + raise exc + return result[0] diff --git a/python/src/basecamp/oauth/device.py b/python/src/basecamp/oauth/device.py index f77add816..a2d53b31b 100644 --- a/python/src/basecamp/oauth/device.py +++ b/python/src/basecamp/oauth/device.py @@ -13,10 +13,8 @@ from __future__ import annotations -import asyncio import json import math -import threading import time from collections.abc import Callable from dataclasses import dataclass @@ -25,6 +23,7 @@ import httpx from basecamp._security import is_localhost, require_https, truncate +from basecamp.oauth._transport import _MAX_REQUEST_TIMEOUT, request_bounded from basecamp.oauth.config import OAuthConfig from basecamp.oauth.device_authorization import DeviceAuthorization from basecamp.oauth.discovery import _normalize_body_cap, _normalize_timeout @@ -58,23 +57,9 @@ _DEVICE_TIMEOUT = 30.0 -#: Upper bound (seconds) on a device request timeout. A per-request timeout beyond -#: this is nonsensical, and a huge finite value would overflow the wall-clock wait -#: primitive (asyncio.wait_for / thread join); clamp to the default above it. -_MAX_DEVICE_REQUEST_TIMEOUT = 3600.0 - #: Granularity (seconds) for polling ``should_cancel`` while waiting between polls. _CANCEL_POLL_INTERVAL = 0.1 -#: Extra time (seconds) to let a timed-out request's async cancellation/cleanup -#: unwind before the caller abandons the (daemon) worker and returns a timeout. -#: Kept SMALL so the caller's worst-case block stays ~timeout: near expiry the -#: request budget is clamped to the remaining code lifetime, and this grace is -#: the only overshoot past the polling deadline — cleanup after a wait_for -#: cancellation is just closing a socket, and joining with no grace at all -#: would race a request that completes right at the deadline. -_WORKER_JOIN_GRACE = 1.0 - # Cap on a device-flow response body (1 MiB) — these responses are tiny; a # larger one is a fault, so abort rather than buffer it. Mirrors discovery. MAX_DEVICE_BODY_BYTES = 1 * 1024 * 1024 @@ -94,8 +79,9 @@ def _post_form_bounded( ) -> tuple[int, bytes]: """SSRF-hardened form POST: suppress redirects, bound the timeout, and read the body under a genuine streaming cap that aborts once ``max_body_bytes`` is - exceeded (never a post-hoc check on an already-buffered body). Mirrors - :func:`basecamp.oauth.discovery._fetch_discovery_document`. + exceeded (never a post-hoc check on an already-buffered body). Delegates to + :func:`basecamp.oauth._transport.request_bounded`, which also bounds the + WHOLE round-trip (httpx's per-read timeout alone cannot). ``read_body(status)`` decides — from the response status, known once headers arrive — whether the body is drained. A caller returns ``False`` for statuses @@ -107,66 +93,19 @@ def _post_form_bounded( :class:`httpx.TimeoutException`) so callers classify them; an oversized body raises :class:`OAuthError` (``api_error``). """ - timeout = _normalize_timeout(timeout, _DEVICE_TIMEOUT, maximum=_MAX_DEVICE_REQUEST_TIMEOUT) - - async def _do() -> tuple[int, bytes]: - async with ( - httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client, - client.stream("POST", url, data=params, headers=_FORM_HEADERS) as response, - ): - if not read_body(response.status_code): - return response.status_code, b"" - chunks: list[bytes] = [] - total = 0 - async for chunk in response.aiter_bytes(): - total += len(chunk) - if total > max_body_bytes: - # An oversized body is api_error, not a timeout — abort the - # stream so it is never fully buffered. - raise OAuthError("api_error", "Device flow response exceeds size cap") - chunks.append(chunk) - return response.status_code, b"".join(chunks) - - # httpx's timeout is per-read (it resets on every received chunk) and httpx has - # NO total-request timeout, so a peer slow-dripping header or body bytes just - # under that interval could otherwise hold the POST open indefinitely (verified); - # closing a sync client from a watchdog does not interrupt a blocked read either. - # asyncio.wait_for CANCELS the request (and closes the socket) at the deadline — - # the caller is bounded AND the work is actually terminated, no leaked worker. - # - # Run it in a DEDICATED thread with its own event loop rather than calling - # asyncio.run() here: this sync helper may be invoked from code that already has - # a running loop (Jupyter/FastAPI/async CLI), where asyncio.run() raises - # RuntimeError before any request is made. wait_for bounds the thread's work at - # ~timeout, so the bounded join below normally returns almost immediately; the - # is_alive backstop after it covers only a pathological async-cleanup hang. - result: list[tuple[int, bytes]] = [] - error: list[Exception] = [] - - def _runner() -> None: - try: - result.append(asyncio.run(asyncio.wait_for(_do(), timeout))) - except Exception as exc: # captured and re-raised on the caller thread - error.append(exc) - - worker = threading.Thread(target=_runner, daemon=True) - worker.start() - # asyncio.wait_for cancels the request at `timeout`, so the worker normally - # finishes well within it. Join with a small grace for the cancellation/cleanup - # to unwind; if even that stalls (a pathological async cleanup hang), return a - # timeout rather than block the caller — the daemon worker never blocks - # interpreter exit. This is bounded AND non-leaking in every non-pathological case. - worker.join(timeout + _WORKER_JOIN_GRACE) - if worker.is_alive(): - raise httpx.ReadTimeout("Device flow request exceeded the timeout deadline") - if error: - exc = error[0] - # On Python >= 3.11 (this package's floor) asyncio.TimeoutError IS the - # builtin TimeoutError, so this catches wait_for's deadline expiry. - if isinstance(exc, TimeoutError): - raise httpx.ReadTimeout("Device flow request exceeded the timeout deadline") from exc - raise exc - return result[0] + # Normalize BEFORE the transport core, which requires a finite, positive, + # in-range timeout (see request_bounded). + timeout = _normalize_timeout(timeout, _DEVICE_TIMEOUT, maximum=_MAX_REQUEST_TIMEOUT) + return request_bounded( + "POST", + url, + headers=_FORM_HEADERS, + params=params, + timeout=timeout, + max_body_bytes=max_body_bytes, + read_body=read_body, + context="Device flow", + ) def request_device_authorization( @@ -409,7 +348,7 @@ def poll_device_token( # remaining-lifetime clamp takes min() against it, and an invalid runtime # value (None → TypeError from min, a negative, an oversized 1e100) must # resolve to the device budget BEFORE any arithmetic touches it. - timeout = _normalize_timeout(timeout, _DEVICE_TIMEOUT, maximum=_MAX_DEVICE_REQUEST_TIMEOUT) + timeout = _normalize_timeout(timeout, _DEVICE_TIMEOUT, maximum=_MAX_REQUEST_TIMEOUT) # The server-driven interval (initial value + sustained slow_down bumps) is # tracked SEPARATELY from the transient timeout backoff: each wait is diff --git a/python/src/basecamp/oauth/discovery.py b/python/src/basecamp/oauth/discovery.py index 68efad1f2..6bb4681c2 100644 --- a/python/src/basecamp/oauth/discovery.py +++ b/python/src/basecamp/oauth/discovery.py @@ -16,13 +16,13 @@ import json import math -import time from typing import Any import httpx -from basecamp._security import require_origin_root, truncate +from basecamp._security import require_origin_root from basecamp.errors import BasecampError +from basecamp.oauth._transport import _MAX_REQUEST_TIMEOUT, request_bounded from basecamp.oauth.config import ( DiscoveryResult, FallbackReason, @@ -89,10 +89,11 @@ def _normalize_timeout(timeout: object, default: float = _DISCOVERY_TIMEOUT, max ``timeout`` is *typed* ``float``, but a caller can pass ``None``, a non-number, a non-positive value, or ``float("inf")``/``nan`` at runtime. An infinite or - non-positive timeout would disable BOTH httpx's bound and the wall-clock - deadline below (``time.monotonic() > inf`` never trips), letting a slow-drip - endpoint hold the SSRF-hardened fetch open indefinitely. Fall back to - ``default`` so the bound can never be turned off — the same discipline as + non-positive timeout would disable BOTH httpx's per-read bound and the + total-request ``asyncio.wait_for`` deadline (an ``inf`` deadline never + fires), letting a slow-drip endpoint hold the SSRF-hardened fetch open + indefinitely. Fall back to ``default`` so the bound can never be turned off + — the same discipline as :func:`_normalize_body_cap`. ``bool`` is excluded (a subclass of ``int``, but a nonsensical timeout). @@ -137,41 +138,33 @@ def _fetch_discovery_document(url: str, timeout: float, max_body_bytes: int) -> """SSRF-hardened GET of a discovery document. The origin must already be validated (via :func:`require_origin_root`); this - suppresses redirects, bounds the timeout, reads the body under a genuine - streaming cap that aborts once ``max_body_bytes`` is exceeded, and maps any - non-2xx status to ``api_error`` (not ``network``). + suppresses redirects, bounds the WHOLE round-trip (via + :func:`basecamp.oauth._transport.request_bounded`), reads the body under a + genuine streaming cap that aborts once ``max_body_bytes`` is exceeded, and + maps any non-2xx status to ``api_error`` (not ``network``). """ max_body_bytes = _normalize_body_cap(max_body_bytes) - # Normalize BEFORE httpx and before the deadline: a non-finite/non-positive - # timeout must not disable either bound (see _normalize_timeout). - timeout = _normalize_timeout(timeout) + # Normalize BEFORE the transport core, which requires a finite, positive, + # in-range timeout: a non-finite/non-positive value must not disable the + # total-request bound (see _normalize_timeout / request_bounded). + timeout = _normalize_timeout(timeout, maximum=_MAX_REQUEST_TIMEOUT) try: - with httpx.stream( + # STATUS DOMINATES THE BODY (SPEC.md: non-2xx on either hop → api_error, + # never network): skip draining a non-2xx body so a stalled/dripped error + # body cannot convert the required api_error into a retryable network + # timeout. The body text was only optional diagnostics — the other SDKs + # read it best-effort at most (TS swallows read failures, Go ignores the + # read error, Kotlin never reads it); category, retryability, and + # http_status are the observable contract. + status, body = request_bounded( "GET", url, headers={"Accept": "application/json"}, timeout=timeout, - follow_redirects=False, - ) as response: - # httpx's timeout is per-operation and RESETS after each received - # chunk, so a peer dripping one byte at a time never trips the read - # timeout and can hold the caller arbitrarily long. Bound the WHOLE - # response with a wall-clock deadline so a slow-drip stream is aborted - # as a retryable network timeout regardless of chunk cadence. - deadline = time.monotonic() + timeout - chunks: list[bytes] = [] - total = 0 - for chunk in response.iter_bytes(): - if time.monotonic() > deadline: - raise OAuthError("network", "OAuth discovery timed out", retryable=True) - total += len(chunk) - if total > max_body_bytes: - # Abort the stream — leaving the ``with`` closes the - # connection, so the oversized body is never fully buffered. - raise OAuthError("api_error", "OAuth discovery response exceeds size cap") - chunks.append(chunk) - status = response.status_code - body = b"".join(chunks) + max_body_bytes=max_body_bytes, + read_body=lambda status_code: 200 <= status_code < 300, + context="OAuth discovery", + ) except httpx.TimeoutException as exc: raise OAuthError("network", "OAuth discovery timed out", retryable=True) from exc except httpx.HTTPError as exc: @@ -182,7 +175,7 @@ def _fetch_discovery_document(url: str, timeout: float, max_body_bytes: int) -> if not 200 <= status < 300: raise OAuthError( "api_error", - f"OAuth discovery failed with status {status}: {truncate(body.decode(errors='replace'))}", + f"OAuth discovery failed with status {status}", http_status=status, ) diff --git a/python/tests/oauth/test_transport.py b/python/tests/oauth/test_transport.py new file mode 100644 index 000000000..57f08bf07 --- /dev/null +++ b/python/tests/oauth/test_transport.py @@ -0,0 +1,192 @@ +"""Real-socket transport-bounding tests for the shared bounded request core. + +respx serves a complete response instantly, so it can exercise neither a +header-then-stall nor a byte-drip; these tests run a real localhost TCP server +(discovery's origin validation exempts http on localhost) and drive the +discovery fetch through :func:`basecamp.oauth._transport.request_bounded`. +""" + +from __future__ import annotations + +import json +import socket +import threading +import time + +import pytest + +from basecamp.oauth import OAuthError, discover_protected_resource +from basecamp.oauth._transport import _WORKER_JOIN_GRACE + + +def _serve_on_localhost() -> tuple[socket.socket, int]: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.bind(("127.0.0.1", 0)) + srv.listen(1) + return srv, srv.getsockname()[1] + + +def _settled_thread_count(baseline: int, deadline_s: float = 5.0) -> int: + # The transport worker is a daemon thread that asyncio.wait_for bounds at + # ~timeout; give its cancellation/cleanup a moment to unwind before counting. + deadline = time.monotonic() + deadline_s + while threading.active_count() > baseline and time.monotonic() < deadline: + time.sleep(0.05) + return threading.active_count() + + +def test_discovery_header_stall_is_bounded_and_leaks_no_worker() -> None: + # The peer sends complete headers then stalls forever without a body byte. + # The fetch must surface a retryable network timeout within ~timeout (plus + # the worker-join grace), and the transport's worker thread must be gone — + # not parked forever on the dead connection. + srv, port = _serve_on_localhost() + stop = threading.Event() + + def stall() -> None: + conn, _ = srv.accept() + conn.recv(4096) + try: + conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nContent-Type: application/json\r\n\r\n") + stop.wait() + except OSError: + pass + finally: + conn.close() + + baseline = threading.active_count() + server = threading.Thread(target=stall, daemon=True) + server.start() + timeout = 0.5 + try: + start = time.monotonic() + with pytest.raises(OAuthError) as exc_info: + discover_protected_resource(f"http://127.0.0.1:{port}", timeout=timeout) + elapsed = time.monotonic() - start + assert exc_info.value.code == "network" + assert exc_info.value.retryable + assert "timed out" in str(exc_info.value) + assert elapsed < timeout + _WORKER_JOIN_GRACE + 1.0, f"fetch not bounded by the timeout: took {elapsed:.2f}s" + finally: + stop.set() + server.join(2) + srv.close() + assert _settled_thread_count(baseline) == baseline, "leaked transport worker thread" + + +def test_params_with_non_post_fails_fast() -> None: + # A form body on a GET would emit a GET-with-body; misuse fails fast instead. + from basecamp.oauth._transport import request_bounded + + with pytest.raises(ValueError, match="only valid with POST"): + request_bounded( + "GET", + "https://issuer.example/x", + headers={}, + params={"a": "b"}, + timeout=1.0, + max_body_bytes=1024, + ) + + +def test_discovery_non_2xx_with_stalled_body_is_immediate_api_error() -> None: + # SPEC.md: non-2xx on either discovery hop → api_error, never network — + # status dominates even when the error body stalls forever, so the fetch + # classifies at header time instead of timing the body out. + srv, port = _serve_on_localhost() + stop = threading.Event() + + def stall() -> None: + conn, _ = srv.accept() + conn.recv(4096) + try: + conn.sendall(b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 1000\r\n\r\n") + stop.wait() + except OSError: + pass + finally: + conn.close() + + baseline = threading.active_count() + server = threading.Thread(target=stall, daemon=True) + server.start() + timeout = 0.5 + try: + start = time.monotonic() + with pytest.raises(OAuthError) as exc_info: + discover_protected_resource(f"http://127.0.0.1:{port}", timeout=timeout) + elapsed = time.monotonic() - start + assert exc_info.value.code == "api_error" + assert exc_info.value.http_status == 500 + assert elapsed < timeout, f"status must classify at header time, not after a body timeout: {elapsed:.2f}s" + finally: + stop.set() + server.join(2) + srv.close() + assert _settled_thread_count(baseline) == baseline, "leaked transport worker thread" + + +def test_discovery_slow_drip_is_bounded_by_the_total_timeout() -> None: + # httpx's read timeout resets on every received chunk, so a peer dripping a + # VALID discovery document byte-by-byte (each read under the timeout) would + # otherwise hold the fetch open far past it — httpx has no total timeout. + # asyncio.wait_for must cancel the whole round-trip at ~timeout regardless + # of chunk cadence, surfacing a retryable network timeout. + srv, port = _serve_on_localhost() + origin = f"http://127.0.0.1:{port}" + body = json.dumps({"resource": origin}).encode() + payload = ( + f"HTTP/1.1 200 OK\r\nContent-Length: {len(body)}\r\nContent-Type: application/json\r\n\r\n" + ).encode() + body + + def drip() -> None: + conn, _ = srv.accept() + conn.recv(4096) + try: + # ~100+ bytes dripped at 0.2s each ≈ 20s+ total; the 0.5s timeout must win. + for byte in payload: + conn.sendall(bytes([byte])) + time.sleep(0.2) + except OSError: + pass + finally: + conn.close() + + baseline = threading.active_count() + server = threading.Thread(target=drip, daemon=True) + server.start() + timeout = 0.5 + try: + start = time.monotonic() + with pytest.raises(OAuthError) as exc_info: + discover_protected_resource(origin, timeout=timeout) + elapsed = time.monotonic() - start + assert exc_info.value.code == "network" + assert exc_info.value.retryable + assert "timed out" in str(exc_info.value) + assert elapsed < timeout + _WORKER_JOIN_GRACE + 1.0, f"fetch not bounded by the timeout: took {elapsed:.2f}s" + finally: + srv.close() + server.join(5) + assert _settled_thread_count(baseline) == baseline, "leaked transport worker thread" + + +@pytest.mark.parametrize( + "timeout", + [None, "5", True, 0, -1, float("inf"), float("nan"), 3601.0, 10**400], +) +def test_invalid_timeout_fails_fast(timeout) -> None: + # request_bounded's whole purpose is a TOTAL request bound; an unnormalized + # timeout (inf/nan/non-positive/oversized/huge-int) would disable or + # overflow asyncio.wait_for and the thread join. Callers normalize, but the + # contract is enforced here too — a violation is a programming error. + from basecamp.oauth._transport import request_bounded + + with pytest.raises(ValueError, match="timeout must be a finite positive"): + request_bounded( + "GET", + "https://issuer.example/x", + headers={}, + timeout=timeout, + max_body_bytes=1024, + ) diff --git a/ruby/lib/basecamp/oauth/device_flow.rb b/ruby/lib/basecamp/oauth/device_flow.rb index d8daf3e55..56f6daa12 100644 --- a/ruby/lib/basecamp/oauth/device_flow.rb +++ b/ruby/lib/basecamp/oauth/device_flow.rb @@ -91,17 +91,15 @@ def request_device_authorization( # default (read) — Ruby treats "" as truthy, so guard on emptiness too. params["scope"] = scope unless scope.nil? || scope.empty? - # Normalize ONCE at operation entry and thread the SAME value to both the - # client construction and the request, so a non-finite/non-positive input - # cannot leave the socket timeout unbounded on the default-built client. - # The body cap gets the same discipline: an invalid value (nil, Infinity, - # negative) would disable the streaming memory bound entirely. + # Normalize ONCE at operation entry and thread the SAME value to every + # request, so a non-finite/non-positive input cannot leave the socket + # timeout unbounded. The body cap gets the same discipline: an invalid + # value (nil, Infinity, negative) would disable the streaming bound. timeout = Fetcher.normalize_timeout(timeout, default: DEVICE_REQUEST_TIMEOUT) max_body_bytes = Fetcher.normalize_body_cap(max_body_bytes) - client = http_client || build_client(timeout) status, body = begin post_form( - client, device_authorization_endpoint, params, + http_client, device_authorization_endpoint, params, timeout: timeout, max_body_bytes: max_body_bytes, # A non-2xx device-auth response is a hard failure whose body is unused. skip_status: ->(s) { !(200..299).cover?(s) } @@ -171,12 +169,11 @@ def poll_device_token( backoff_seconds = interval_seconds deadline = clock.call + expires_in - # Normalize ONCE, outside the polling loop, and reuse for the client and - # every per-poll request (see request_device_authorization). The body cap - # gets the same discipline — an invalid value would disable the bound. + # Normalize ONCE, outside the polling loop, and reuse for every per-poll + # request (see request_device_authorization). The body cap gets the same + # discipline — an invalid value would disable the bound. timeout = Fetcher.normalize_timeout(timeout, default: DEVICE_REQUEST_TIMEOUT) max_body_bytes = Fetcher.normalize_body_cap(max_body_bytes) - client = http_client || build_client(timeout) params = { "grant_type" => DEVICE_CODE_GRANT_TYPE, "device_code" => device_code, @@ -207,7 +204,7 @@ def poll_device_token( # Bound the request by the REMAINING code lifetime as well as the # per-request timeout: near expiry, a stalled token POST must not # hold the flow past the monotonic deadline for the full budget. - post_device_token(client, token_endpoint, params, + post_device_token(http_client, token_endpoint, params, timeout: [ timeout, post_remaining ].min, max_body_bytes: max_body_bytes) rescue Faraday::TimeoutError # A connection timeout is transient: back off exponentially and @@ -342,10 +339,6 @@ def valid_device_seconds?(value) value.is_a?(Numeric) && value.real? && value.finite? && value.positive? && value <= MAX_DEVICE_SECONDS end - def build_client(timeout) - Fetcher.build_client(timeout) - end - # Waits +seconds+ while observing cancellation DURING the wait. A plain # +sleep+ is not interruptible, so a cancellation set mid-wait would not be # noticed until the whole (possibly grown +slow_down+) interval elapses. @@ -368,21 +361,31 @@ def wait_cancellable(seconds, cancelled, sleeper) end end - # POSTs a form body and reads the response under the same bounded/ - # streaming cap as discovery (SPEC.md §9): the +on_data+ proc aborts the - # read the moment the accumulated size exceeds the cap, so an oversized - # response is never fully buffered. Returns +[status, body]+. The - # per-request timeout is always set here — even on an injected client — - # so a stalled socket can't hang the poll. A real adapter streams to - # +on_data+ (leaving +response.body+ empty); a test double that ignores - # the block falls back to the buffered body, still size-capped. + # POSTs a form body and returns +[status, body]+, reading under the same + # bounded/streaming cap as discovery (SPEC.md §9). + # + # With a nil +client+ the POST runs on the headers-first + # {Fetcher.stream_http} primitive: +skip_status+ classifies by status at + # HEADER time (a skipped body is never read, even one that stalls + # forever), and a watchdog bounds the whole request — including a + # stalled or byte-dripped header phase — at the timeout. An INJECTED + # Faraday connection keeps the Faraday path below. def post_form(client, url, params, timeout:, max_body_bytes:, skip_status: nil) - # +timeout+ is already normalized by the caller (request/poll entry). The - # wall-clock deadline bounds the WHOLE read: req.options.timeout below - # bounds only each socket read and resets on every on_data chunk, so a - # slow-drip peer could otherwise hang a device request past the timeout / - # code expiry while staying under the cap. +skip_status+ stops the read - # for a status whose body the caller doesn't use (non-2xx / 3xx). + if client.nil? + return Fetcher.stream_http( + :post, url, + headers: { "Content-Type" => "application/x-www-form-urlencoded", "Accept" => "application/json" }, + form: params, timeout: timeout, max_body_bytes: max_body_bytes, skip_status: skip_status + ) + end + + # Injected-client (Faraday) path. +timeout+ is already normalized by the + # caller (request/poll entry). The wall-clock deadline bounds the WHOLE + # read: req.options.timeout below bounds only each socket read and resets + # on every on_data chunk, so a slow-drip peer could otherwise hang a + # device request past the timeout / code expiry while staying under the + # cap. +skip_status+ stops the read for a status whose body the caller + # doesn't use (non-2xx / 3xx). deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout chunks, on_data = Fetcher.bounded_reader(max_body_bytes, deadline: deadline, skip_status: skip_status) response = client.post(url) do |req| @@ -403,14 +406,13 @@ def post_form(client, url, params, timeout:, max_body_bytes:, skip_status: nil) # every client shape and supported Faraday version, never buffered into # a size-cap error the caller must then untangle. # - # Known residual: Faraday exposes no headers-time callback, so a response - # whose headers arrive but whose body then stalls PAST the read timeout - # never returns from +client.post+ — it surfaces as a bounded transport - # timeout (request path: :transport; poll path: backoff, capped by code - # expiry) rather than this status-first classification. That degradation - # is bounded and redirect-safe (3xx Locations are never followed); exact - # headers-first semantics for it belong to the transport primitive shared - # with discovery, tracked as a pre-go-live hardening follow-up. + # Known residual — INJECTED clients only (the default path above is + # exact): Faraday exposes no headers-time callback, so a response whose + # headers arrive but whose body then stalls PAST the read timeout never + # returns from +client.post+ — it surfaces as a bounded transport timeout + # (request path: :transport; poll path: backoff, capped by code expiry) + # rather than this status-first classification. Bounded and + # redirect-safe (3xx Locations are never followed). return [ response.status, "" ] if skip_status && skip_status.call(response.status) body = diff --git a/ruby/lib/basecamp/oauth/discovery.rb b/ruby/lib/basecamp/oauth/discovery.rb index 08e6259f4..4c4b6798f 100644 --- a/ruby/lib/basecamp/oauth/discovery.rb +++ b/ruby/lib/basecamp/oauth/discovery.rb @@ -27,7 +27,10 @@ def initialize(http_client: nil, timeout: 10, max_body_bytes: Fetcher::DEFAULT_M # wall-clock deadline: a non-finite/non-positive timeout must not disable # either bound (see Fetcher.normalize_timeout). @timeout = Fetcher.normalize_timeout(timeout) - @http_client = http_client || Fetcher.build_client(@timeout) + # nil selects the headers-first {Fetcher.stream_http} transport (total + # wall-clock bound incl. the header phase); an injected connection keeps + # the Faraday path, verified redirect-free above. + @http_client = http_client # Normalize the public cap to a finite non-negative Integer: a nil, float, # or Float::INFINITY would otherwise disable the streaming memory bound # (an infinite/undefined cap never trips +total > max_body_bytes+), diff --git a/ruby/lib/basecamp/oauth/fetcher.rb b/ruby/lib/basecamp/oauth/fetcher.rb index 8386416e9..fa11105aa 100644 --- a/ruby/lib/basecamp/oauth/fetcher.rb +++ b/ruby/lib/basecamp/oauth/fetcher.rb @@ -2,6 +2,9 @@ require "faraday" require "json" +require "net/http" +require "openssl" +require "uri" module Basecamp module Oauth @@ -154,29 +157,16 @@ def self.bounded_reader(max_body_bytes, deadline: nil, skip_status: nil) [ chunks, reader ] end - # Builds the default SSRF-hardened Faraday connection. No redirect - # middleware is registered, so redirects are not followed. - # - # @param timeout [Integer] request + connect timeout in seconds - # @return [Faraday::Connection] - def self.build_client(timeout) - Faraday.new do |conn| - conn.options.timeout = timeout - conn.options.open_timeout = timeout - conn.adapter Faraday.default_adapter - end - end - # Rejects an INJECTED connection whose middleware stack we cannot verify to # be redirect-free. Redirect suppression is a load-bearing SSRF control (RFC # 9728 §7.7): a caller-supplied client that follows redirects would silently # chase an attacker-controlled +Location+. A class-NAME heuristic (matching # +/redirect/+) is bypassable by a follower whose class name does not contain # "redirect", so we enforce a POLICY instead of guessing by name: an injected - # connection may carry ONLY adapter handlers. The default {build_client} - # connection (adapter only) and a test's mock adapter qualify; ANY request/ - # response middleware — which could follow redirects under any name, or - # otherwise rewrite the request — is refused rather than trusted. + # connection may carry ONLY adapter handlers (an adapter-only connection or + # a test's mock adapter qualifies); ANY request/response middleware — which + # could follow redirects under any name, or otherwise rewrite the request — + # is refused rather than trusted. # # @param client [Faraday::Connection] # @raise [OauthError] +validation+ when non-adapter middleware is present @@ -205,14 +195,188 @@ def self.ensure_redirects_suppressed!(client) ) end + # Headers-first bounded HTTP over Net::HTTP — the default transport for + # every SDK-built OAuth fetch (both discovery hops and the device flow). + # Injected Faraday connections keep the Faraday path; this primitive exists + # because Faraday cannot provide these two guarantees: + # + # 1. **Status at header time.** Faraday's +on_data+ is a body callback, so a + # response whose body never arrives can only be classified after a + # timeout. Net::HTTP's block form yields the response once HEADERS are + # in: +skip_status+ classifies by status BEFORE any body read, and + # raising out of the block makes +Net::HTTP.start+'s ensure close the + # socket with the body undrained — exact status-first classification + # (SPEC.md §16) for every response shape, including a stalled body. + # 2. **A total wall-clock bound.** A per-read timeout resets on every byte, + # so a peer dripping header or body bytes defeats it. A WATCHDOG thread + # closes the connection at a monotonic deadline, which interrupts even a + # blocked or dripped HEADER read (closing the socket from another thread + # raises IOError in the blocked reader — verified on a live socket). + # +max_retries = 0+ is load-bearing: Net::HTTP's idempotent-retry would + # otherwise silently REOPEN the connection the watchdog just closed. + # Scope: the watchdog governs from session-start on. The CONNECTION + # phases (TCP connect, proxy CONNECT, TLS handshake) are not + # watchdog-interruptible — Net::HTTP marks the session started only + # after they complete — but each is natively bounded by open_timeout, + # and a post-connect deadline re-check refuses the request when setup + # consumed the budget. Total wall clock is a small multiple of the + # timeout, never unbounded. + # + # The body streams under the same cap + deadline as the Faraday path, and + # redirects are structurally never followed (+Net::HTTP#request+ has no + # follow logic). Transport failures surface as Faraday errors + # (+TimeoutError+ for timeouts, +ConnectionFailed+ for connection and + # protocol-parse failures) so both transport paths classify through the + # same caller rescues. Bounded-read violations keep raising the shared + # {BodyTooLarge} / {ReadDeadlineExceeded} markers — deliberately NOT + # Faraday errors, so each caller maps them to its own operation-specific + # error message, exactly as on the Faraday path. + # + # @param method [Symbol] +:get+ or +:post+ + # @param url [String] fully-qualified URL (already origin-validated) + # @param headers [Hash] request headers + # @param form [Hash, nil] form params; www-form-encoded into the POST body + # @param timeout [Numeric] total request bound in seconds (already + # normalized by the caller). The deadline is anchored BEFORE connect and + # open_timeout carries the same value, so the total wall time is + # ~timeout regardless of which phase stalls (the watchdog closes the + # session the moment it exists if the deadline fired mid-connect) + # @param max_body_bytes [Integer] bounded read cap in bytes + # @param skip_status [Proc, nil] statuses whose body is never read + # @return [Array(Integer, String)] status and (possibly empty) body + def self.stream_http(method, url, headers: {}, form: nil, timeout:, max_body_bytes: DEFAULT_MAX_BODY_BYTES, skip_status: nil) + uri = begin + URI.parse(url) + rescue URI::InvalidURIError + nil + end + # Fail closed on an unparsable or hostless URL ("https:foo" parses with a + # nil hostname): require_https checks only the scheme, and a nil host + # would otherwise surface as a raw ArgumentError from inside Net::HTTP — + # outside the transport's Faraday-error contract. + if uri.nil? || uri.hostname.nil? || uri.hostname.empty? + raise OauthError.new("validation", "OAuth endpoint URL has no host: #{url.inspect}") + end + + # URI#hostname strips IPv6 brackets ("[::1]" -> "::1"), which is the form + # Net::HTTP.new expects. ENV proxy handling matches faraday-net_http. + http = Net::HTTP.new(uri.hostname, uri.port) + http.use_ssl = uri.scheme == "https" + http.open_timeout = timeout + http.read_timeout = timeout + http.write_timeout = timeout + http.max_retries = 0 + + request = + case method + when :post then Net::HTTP::Post.new(uri) + when :get then Net::HTTP::Get.new(uri) + else raise ArgumentError, "stream_http supports :get and :post, got #{method.inspect}" + end + headers.each { |name, value| request[name] = value } + if form + request.body = URI.encode_www_form(form) + # A form body implies the form content type; explicit headers win. + request["Content-Type"] ||= "application/x-www-form-urlencoded" + end + + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + deadline_fired = false + watchdog = Thread.new do + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + sleep(remaining) if remaining.positive? + deadline_fired = true + # Retry until the session exists to close: the deadline can fire while + # the session is still CONNECTING (finish then raises IOError), and a + # one-shot close would leave the subsequent header read unbounded. The + # ensure below kills this thread the moment the request completes, so + # the loop cannot outlive the call. + begin + http.finish + rescue IOError + sleep(0.05) + retry + end + end + + status = nil + chunks = [] + total = 0 + http.start do |session| + # The connection phases (TCP connect, proxy CONNECT, TLS handshake) + # are each natively bounded by open_timeout — the watchdog cannot + # interrupt them because Net::HTTP only marks the session started + # once they complete (finish raises IOError until then). Re-check the + # deadline the moment the session is up: if setup consumed the whole + # budget, fail now rather than granting the request a fresh header + # wait. Worst-case wall clock is a small multiple of the timeout + # (per-phase native bounds + watchdog from headers on), never + # unbounded. + raise ReadDeadlineExceeded if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + + session.request(request) do |response| + status = response.code.to_i + # Status-first: a skipped status's body is NEVER read — the raise + # unwinds through start, whose ensure closes the socket undrained. + raise SkipBody.new(status) if skip_status&.call(status) + + response.read_body do |chunk| + raise ReadDeadlineExceeded if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + + total += chunk.bytesize + raise BodyTooLarge if total > max_body_bytes + + chunks << chunk + end + end + end + [ status, chunks.join.force_encoding(Encoding::UTF_8) ] + rescue SkipBody => e + [ e.status, "" ] + rescue Timeout::Error, Errno::ETIMEDOUT => e + # Timeout::Error covers Net::OpenTimeout/ReadTimeout/WriteTimeout alike + # (a peer that accepts the connection but stops READING trips the write + # timeout). Errno::ETIMEDOUT is a SystemCallError, but it is a TIMEOUT: + # both must map with the timeouts — the exact pair faraday-net_http + # rescues — or the device poll would terminate instead of applying its + # transient backoff. + raise Faraday::TimeoutError, "OAuth request timed out: #{e.message}" + rescue IOError => e + # The watchdog's close raises IOError in the blocked reader; only map it + # to a timeout when the deadline actually fired — any other IOError (a + # peer closing mid-headers, for example) is a connection failure. + raise Faraday::TimeoutError, "OAuth request exceeded the timeout deadline" if deadline_fired + + raise Faraday::ConnectionFailed, e.message + rescue OpenSSL::SSL::SSLError => e + # TLS failures (an unverifiable peer certificate above all) map to + # Faraday::SSLError exactly as faraday-net_http maps them, so the + # default and injected paths classify certificate rejection alike. + raise Faraday::SSLError, e.message + rescue Net::HTTPBadResponse, Net::HTTPHeaderSyntaxError, Net::ProtocolError, + SystemCallError, SocketError => e + # The parse errors are direct StandardError subclasses (not IOError), so + # a malformed status line / header must be mapped here explicitly or it + # would leak raw from the public discovery/device APIs. + raise Faraday::ConnectionFailed, e.message + ensure + watchdog&.kill + watchdog&.join + end + # Fetches +url+ and returns the parsed JSON object (a Hash). # - # The request timeout is applied per-request (not only on the connection) - # so a bounded read is enforced even when the caller INJECTS its own - # connection: an injected client's adapter default would otherwise leave the - # requested +timeout+ unenforced. This mirrors the device flow's +post_form+. + # With a nil +http_client+ the fetch runs on the headers-first + # {stream_http} primitive (total wall-clock bound incl. the header phase). + # An INJECTED connection keeps the Faraday path: the request timeout is + # applied per-request (not only on the connection) so a bounded read is + # enforced even under the injected adapter's defaults, and the wall-clock + # deadline bounds the whole body read — but Faraday exposes no + # headers-time callback, so a body that stalls past the read timeout on + # the injected path surfaces as a bounded transport timeout. # - # @param http_client [Faraday::Connection] the SSRF-hardened connection + # @param http_client [Faraday::Connection, nil] injected connection, or + # nil for the default headers-first transport # @param url [String] fully-qualified well-known URL to fetch # @param timeout [Integer] per-request timeout in seconds # @param max_body_bytes [Integer] bounded read cap in bytes @@ -220,30 +384,32 @@ def self.ensure_redirects_suppressed!(client) # @raise [OauthError] +api_error+ on non-2xx, oversized body, non-object # JSON, or parse failure; +network+ on transport failure def self.fetch_json(http_client, url, timeout:, max_body_bytes: DEFAULT_MAX_BODY_BYTES) - # Wall-clock deadline over the WHOLE read: req.options.timeout below bounds - # only each socket read and resets on every chunk, so a slow-drip peer could - # otherwise hang the fetch indefinitely while staying under max_body_bytes. - deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout - chunks, on_data = bounded_reader(max_body_bytes, deadline: deadline) - - response = http_client.get(url) do |req| - req.headers["Accept"] = "application/json" - # Bounded streaming read: abort the moment the cap is exceeded so an - # oversized body is never fully buffered. - req.options.on_data = on_data - # Apply the request timeout on every request — even an injected client — - # so a stalled socket can't hang discovery under the adapter default. - req.options.timeout = timeout - req.options.open_timeout = timeout - end - - body = chunks.join.force_encoding(Encoding::UTF_8) + status, body = + if http_client.nil? + stream_http( + :get, url, + headers: { "Accept" => "application/json" }, + timeout: timeout, max_body_bytes: max_body_bytes, + # STATUS DOMINATES THE BODY (SPEC.md: non-2xx on either hop → + # api_error, never network): skip draining a non-2xx body so a + # stalled/dripped error body cannot convert the required api_error + # into a network timeout. The body text was only optional + # diagnostics — the other SDKs read it best-effort at most. + skip_status: ->(response_status) { !(200..299).cover?(response_status) } + ) + else + faraday_fetch(http_client, url, timeout: timeout, max_body_bytes: max_body_bytes) + end - unless (200..299).cover?(response.status) + unless (200..299).cover?(status) + # Status-only, on BOTH paths: the error category, retryability, and + # http_status are the observable contract; embedding response body + # text would put attacker-influenced content in exception messages + # and diverge from the (body-less) default transport and Python. raise OauthError.new( "api_error", - "OAuth discovery failed with status #{response.status}: #{Basecamp::Security.truncate(body)}", - http_status: response.status + "OAuth discovery failed with status #{status}", + http_status: status ) end @@ -260,6 +426,40 @@ def self.fetch_json(http_client, url, timeout:, max_body_bytes: DEFAULT_MAX_BODY rescue JSON::ParserError => e raise OauthError.new("api_error", "Failed to parse OAuth discovery response: #{e.message}") end + + # The Faraday transport for an INJECTED connection. The request timeout is + # applied per-request (not only on the connection) so a bounded read is + # enforced even under the injected adapter's defaults, and the wall-clock + # deadline bounds the whole body read. + # + # Known residual (injected connections only): unlike the default + # {stream_http} path — which skips unusable bodies by status at header + # time — Faraday exposes no headers-time seam, so a non-2xx body is + # drained here under the same cap + deadline before the status error + # surfaces. Bounded, never followed; callers who inject a connection + # keep their transport's semantics by design. + # + # @return [Array(Integer, String)] status and body + def self.faraday_fetch(http_client, url, timeout:, max_body_bytes:) + # Wall-clock deadline over the WHOLE read: req.options.timeout below bounds + # only each socket read and resets on every chunk, so a slow-drip peer could + # otherwise hang the fetch indefinitely while staying under max_body_bytes. + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + chunks, on_data = bounded_reader(max_body_bytes, deadline: deadline) + + response = http_client.get(url) do |req| + req.headers["Accept"] = "application/json" + # Bounded streaming read: abort the moment the cap is exceeded so an + # oversized body is never fully buffered. + req.options.on_data = on_data + # Apply the request timeout on every request — even an injected client — + # so a stalled socket can't hang discovery under the adapter default. + req.options.timeout = timeout + req.options.open_timeout = timeout + end + + [ response.status, chunks.join.force_encoding(Encoding::UTF_8) ] + end end end end diff --git a/ruby/lib/basecamp/oauth/resource.rb b/ruby/lib/basecamp/oauth/resource.rb index cc6c203c2..06c8cfdf9 100644 --- a/ruby/lib/basecamp/oauth/resource.rb +++ b/ruby/lib/basecamp/oauth/resource.rb @@ -14,7 +14,10 @@ def initialize(http_client: nil, timeout: 10, max_body_bytes: Fetcher::DEFAULT_M # wall-clock deadline: a non-finite/non-positive timeout must not disable # either bound (see Fetcher.normalize_timeout). @timeout = Fetcher.normalize_timeout(timeout) - @http_client = http_client || Fetcher.build_client(@timeout) + # nil selects the headers-first {Fetcher.stream_http} transport (total + # wall-clock bound incl. the header phase); an injected connection keeps + # the Faraday path, verified redirect-free above. + @http_client = http_client # Normalize the public cap to a finite non-negative Integer: a nil, float, # or Float::INFINITY would otherwise disable the streaming memory bound # (an infinite/undefined cap never trips), reintroducing an SSRF/OOM risk. diff --git a/ruby/test/basecamp/oauth_transport_test.rb b/ruby/test/basecamp/oauth_transport_test.rb new file mode 100644 index 000000000..7968cd6ff --- /dev/null +++ b/ruby/test/basecamp/oauth_transport_test.rb @@ -0,0 +1,361 @@ +# frozen_string_literal: true + +require "test_helper" +require "openssl" +require "socket" + +# Acceptance tests for the headers-first default transport +# ({Basecamp::Oauth::Fetcher.stream_http}) against REAL sockets — the response +# shapes these prove (a stalled or byte-dripped header phase, a body that never +# arrives after headers) cannot be produced by WebMock stubs or Faraday test +# adapters, and are exactly the shapes the primitive exists to bound. +class OAuthTransportTest < Minitest::Test + TIMEOUT = 0.6 + + def setup + # Fully unpatch Net::HTTP (not merely allow_localhost): WebMock's patched + # Net::HTTP buffers even allowed real requests, which destroys the + # header-time semantics these tests exist to prove. + WebMock.disable! + @servers = [] + @conns = [] + @server_threads = [] + end + + def teardown + # Stop the server threads FIRST (they append to @conns), then close every + # accepted socket and listener — the stall handlers sleep for tens of + # seconds, so without the kill+join each test would leak a live thread and + # its socket well past the test's end. + @server_threads.each(&:kill).each(&:join) + @conns.each { |conn| conn.close rescue nil } + @servers.each { |server| server.close rescue nil } + WebMock.enable! + WebMock.disable_net_connect! + end + + # Starts a real TCP server; the handler receives each accepted socket after + # the request headers have been consumed. Returns [endpoint, accepts] where + # +accepts+ counts connections — the zero-retry assertions read it. Accepted + # sockets and the accept thread are tracked for teardown. + def start_server(&handler) + server = TCPServer.new("127.0.0.1", 0) + @servers << server + accepts = [] + @server_threads << Thread.new do + loop do + conn = server.accept + accepts << conn + @conns << conn + while (line = conn.gets) && line != "\r\n"; end + handler.call(conn) + rescue IOError, SystemCallError + break # server closed in teardown + end + end + [ "http://127.0.0.1:#{server.addr[1]}", accepts ] + end + + def elapsed + start = Process.clock_gettime(Process::CLOCK_MONOTONIC) + yield + Process.clock_gettime(Process::CLOCK_MONOTONIC) - start + end + + # --- status-first: a skipped status classifies at HEADER time ------------- + + def test_device_auth_non_2xx_with_stalled_body_is_immediate_api_error + # 500 headers, then NOT ONE body byte: the old body-callback transport could + # only time this out as :transport; headers-first classifies it instantly. + endpoint, = start_server do |conn| + conn.write("HTTP/1.1 500 Internal Server Error\r\nContent-Length: 1000\r\n\r\n") + sleep 30 + end + + error = nil + seconds = elapsed do + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::DeviceFlow.request_device_authorization( + device_authorization_endpoint: "#{endpoint}/device", + client_id: "basecamp-cli", timeout: TIMEOUT + ) + end + end + + assert_equal "api_error", error.type + assert_equal 500, error.http_status + assert_operator seconds, :<, TIMEOUT, "status must classify at header time, not after a body timeout" + end + + def test_token_poll_302_with_stalled_body_is_immediate_api_error_with_zero_retries + # The SPEC §16 contract this transport closes: a token 3xx whose body stalls + # must surface the redirect api_error immediately — one request, no + # transport-backoff retries toward code expiry. + endpoint, accepts = start_server do |conn| + conn.write("HTTP/1.1 302 Found\r\nLocation: https://attacker.example/\r\nContent-Length: 1000\r\n\r\n") + sleep 30 + end + + error = nil + seconds = elapsed do + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::DeviceFlow.poll_device_token( + token_endpoint: "#{endpoint}/token", client_id: "basecamp-cli", + device_code: "d", interval: 5, expires_in: 900, + timeout: TIMEOUT, sleeper: ->(_seconds) { } + ) + end + end + + assert_equal "api_error", error.type + assert_equal 302, error.http_status + assert_match(/redirect/i, error.message) + assert_equal 1, accepts.length, "a header-classified redirect must never be retried" + assert_operator seconds, :<, TIMEOUT + end + + def test_skipped_response_closes_the_connection_undrained + # Releasing the connection matters as much as classifying it: the server + # must observe the socket close (EPIPE/RST) instead of feeding a body to a + # client that will never read it. + server_saw_close = Queue.new + endpoint, = start_server do |conn| + conn.write("HTTP/1.1 302 Found\r\nLocation: https://x/\r\nContent-Length: 10000000\r\n\r\n") + begin + 1_000.times { conn.write("x" * 10_000); sleep 0.005 } + server_saw_close << false + rescue IOError, SystemCallError + server_saw_close << true + end + end + + status, body = Basecamp::Oauth::Fetcher.stream_http( + :post, "#{endpoint}/token", form: { "a" => "b" }, + timeout: TIMEOUT, skip_status: ->(s) { (300..399).cover?(s) } + ) + + assert_equal 302, status + assert_equal "", body + assert_equal true, server_saw_close.pop, "the abandoned body's socket must be torn down" + end + + # --- total wall-clock bound, including the header phase ------------------- + + def test_header_phase_stall_is_bounded_transport_error + endpoint, = start_server { |_conn| sleep 30 } # headers never arrive + + error = nil + seconds = elapsed do + error = assert_raises(Basecamp::Oauth::DeviceFlowError) do + Basecamp::Oauth::DeviceFlow.request_device_authorization( + device_authorization_endpoint: "#{endpoint}/device", + client_id: "basecamp-cli", timeout: TIMEOUT + ) + end + end + + assert_equal :transport, error.reason + assert_operator seconds, :<, TIMEOUT * 3, "a header stall must be bounded by the watchdog" + end + + def test_header_phase_drip_is_bounded_transport_error + # One header byte per 0.1s: every read succeeds inside the per-read timeout, + # so only the watchdog's monotonic deadline can bound this — the case that + # is structurally impossible to bound through Faraday's on_data. + endpoint, = start_server do |conn| + "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n".each_char do |char| + begin + conn.write(char) + rescue IOError, SystemCallError + break + end + sleep 0.1 + end + sleep 30 + end + + error = nil + seconds = elapsed do + error = assert_raises(Basecamp::Oauth::DeviceFlowError) do + Basecamp::Oauth::DeviceFlow.request_device_authorization( + device_authorization_endpoint: "#{endpoint}/device", + client_id: "basecamp-cli", timeout: TIMEOUT + ) + end + end + + assert_equal :transport, error.reason + assert_operator seconds, :<, TIMEOUT * 3, "a dripped header phase must be bounded by the watchdog" + end + + def test_discovery_non_2xx_with_stalled_body_is_immediate_api_error + # SPEC.md: non-2xx on either discovery hop → api_error, never network — + # status dominates even when the error body stalls forever. + endpoint, = start_server do |conn| + conn.write("HTTP/1.1 500 Internal Server Error\r\nContent-Length: 1000\r\n\r\n") + sleep 30 + end + + error = nil + seconds = elapsed do + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::Fetcher.fetch_json(nil, "#{endpoint}/doc", timeout: TIMEOUT) + end + end + + assert_equal "api_error", error.type + assert_equal 500, error.http_status + assert_operator seconds, :<, TIMEOUT, "status must classify at header time" + end + + def test_body_slow_drip_is_bounded_for_discovery + # A wanted (2xx) body dripped forever: the read-loop deadline bounds it and + # discovery surfaces its retryable network timeout. + endpoint, = start_server do |conn| + conn.write("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n") + loop do + conn.write("x") + sleep 0.1 + rescue IOError, SystemCallError + break + end + end + + error = nil + seconds = elapsed do + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::Fetcher.fetch_json(nil, "#{endpoint}/.well-known/oauth-authorization-server", timeout: TIMEOUT) + end + end + + assert_equal "network", error.type + assert error.retryable + assert_operator seconds, :<, TIMEOUT * 3 + end + + def test_oversized_body_aborts_streaming_read + endpoint, = start_server do |conn| + conn.write("HTTP/1.1 200 OK\r\nContent-Length: 300000\r\n\r\n") + begin + 30.times { conn.write("x" * 10_000) } + rescue IOError, SystemCallError + nil + end + end + + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::Fetcher.fetch_json( + nil, "#{endpoint}/doc", timeout: TIMEOUT, max_body_bytes: 8 * 1024 + ) + end + + assert_equal "api_error", error.type + assert_match(/size cap/i, error.message) + end + + def test_self_signed_tls_certificate_is_rejected_and_mapped + # The default transport moved from Faraday to direct Net::HTTP — prove peer + # verification survived the move: a self-signed certificate must fail the + # handshake and map to the same Faraday::SSLError → network classification + # faraday-net_http produced, never a raw OpenSSL exception (and never a + # completed request). + key = OpenSSL::PKey::RSA.new(2048) + name = OpenSSL::X509::Name.parse("/CN=127.0.0.1") + cert = OpenSSL::X509::Certificate.new + cert.version = 2 + cert.serial = 1 + cert.subject = name + cert.issuer = name + cert.public_key = key.public_key + cert.not_before = Time.now - 60 + cert.not_after = Time.now + 3600 + cert.sign(key, OpenSSL::Digest.new("SHA256")) + + ssl_context = OpenSSL::SSL::SSLContext.new + ssl_context.cert = cert + ssl_context.key = key + tcp = TCPServer.new("127.0.0.1", 0) + @servers << tcp + ssl_server = OpenSSL::SSL::SSLServer.new(tcp, ssl_context) + handshakes_completed = 0 + @server_threads << Thread.new do + loop do + @conns << ssl_server.accept + handshakes_completed += 1 + rescue OpenSSL::SSL::SSLError, IOError, SystemCallError + # The client rejecting the cert aborts the handshake server-side — + # keep accepting until teardown closes the listener. + break if tcp.closed? + end + end + + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::Fetcher.fetch_json(nil, "https://127.0.0.1:#{tcp.addr[1]}/doc", timeout: TIMEOUT) + end + + assert_equal "network", error.type + assert error.retryable + assert_match(/certificate|SSL/i, error.message) + assert_equal 0, handshakes_completed, "the client must abort the handshake, not complete it" + end + + def test_unsupported_method_fails_fast + # A typo'd verb must not silently become a GET. + assert_raises(ArgumentError) do + Basecamp::Oauth::Fetcher.stream_http(:put, "https://issuer.example/x", timeout: TIMEOUT) + end + end + + def test_hostless_url_fails_closed_as_validation_error + # "https:foo" passes the scheme-only HTTPS guard but parses with a nil + # hostname; without the explicit check it surfaced as a raw ArgumentError + # from inside Net::HTTP — outside the transport's error contract. + [ "https:foo", "https://", "http:" ].each do |url| + error = assert_raises(Basecamp::Oauth::OauthError, url) do + Basecamp::Oauth::Fetcher.stream_http(:post, url, form: { "a" => "b" }, timeout: TIMEOUT) + end + assert_equal "validation", error.type, url + assert_match(/no host/i, error.message) + end + end + + def test_malformed_http_response_maps_to_transport_error + # A non-HTTP peer (garbage status line) raises Net::HTTPBadResponse — a bare + # StandardError subclass that must be mapped, or it leaks raw from the + # public discovery/device APIs instead of the documented network error. + endpoint, = start_server do |conn| + conn.write("NOT-HTTP GARBAGE\r\n\r\n") + conn.close + end + + error = assert_raises(Basecamp::Oauth::OauthError) do + Basecamp::Oauth::Fetcher.fetch_json(nil, "#{endpoint}/doc", timeout: TIMEOUT) + end + + assert_equal "network", error.type + assert error.retryable + end + + def test_watchdog_threads_do_not_leak + endpoint, = start_server do |conn| + conn.write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}") + end + + # Warm-up request BEFORE the baseline: the process's first Net::HTTP + # connect can lazily spawn Ruby's persistent Timeout worker thread (one + # per process, never reaped). Whether that already happened depends on + # randomized suite order, so counting it after the baseline made this + # assertion flake by exactly +1 on the orderings where this test ran + # first. The warm-up folds any lazily-created runtime threads into the + # baseline; the assertion then counts only threads OUR primitive leaks. + Basecamp::Oauth::Fetcher.stream_http(:get, "#{endpoint}/doc", timeout: TIMEOUT) + + baseline = Thread.list.length + 5.times do + Basecamp::Oauth::Fetcher.stream_http(:get, "#{endpoint}/doc", timeout: TIMEOUT) + end + # The watchdog is killed and JOINED in the primitive's ensure, so no request + # leaves a thread behind. + assert_operator Thread.list.length, :<=, baseline + end +end