diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a1d07d..0ea1629 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## Unreleased maintenance + +- Reject non-standard JSON number constants and numbers outside the supported finite float range; report bounded integer decoder failures as observable HTTP errors instead of leaking raw parser exceptions. +- Redact complete URL credentials before truncating diagnostics so a long password cannot expose its prefix; retain the final output limit after replacements. +- Finish single async HTTP response and transport cleanup before returning cancellation, including interrupted body reads, multipart uploads and repeated cancellation during connection release. +- Reject unsafe manifest records and Windows junctions before enumerating or hashing release sources, while continuing to exclude build artifacts from extracted-source scans. +- Close every cluster client even if one close fails; finish async client cleanup before reporting errors or cancellation and drain cancelled health/bulk siblings safely. +- Validate integer bulk bounds, worker counts, routing weights and failure thresholds before consuming work. +- Accept valid JSON `null` responses in sync/async requests instead of incorrectly reporting HTTP 200 as malformed JSON. +- Protect async sessions against delayed authentication, logout and unauthorized responses so stale requests cannot restore a signed-out session or replace/clear a newer account. +- Move async attachment snapshot, multipart file reads and atomic download writes off the event loop; drain active disk workers before returning cancellation or closing owned files. +- Never replay multipart file bodies across cluster endpoints, including PUT and keyed POST requests. +- Cancel and drain sibling operations when an async bulk item is cancelled, so no owned operation continues after bulk returns. +- Reject non-integer async concurrency values before consuming work; fractional semaphore counts could otherwise bypass the parallelism limit. +- Keep these fixes separate from the frozen v0.5.2 publication candidate. + ## 0.5.2 — 2026-10-11 - Add lazy synchronous and asynchronous cursor/offset traversal with resumable cursors, strict page validation, cycle detection, item bounds and cancellation propagation. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 96eae1e..87a0e3d 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -05d98f7ab3cc7a9d653c9a9902ca2ab5c3936c0509da776d21880e4fb6270f56 CHANGELOG.md +6b05c59b9fc679a94d250abe4f78748ef6b49b218a446834d6614019379a7069 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -25,8 +25,8 @@ ce84a06b898c79cd6199e9eac425e40a3b8f7f8a346d6bbdd019580de56d7652 LICENSE-HISTOR b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe LICENSE-METADATA.json 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16 NOTICE.md c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae OWNERSHIP.md -7d1bcaa4e570d431553ec4abd420f5072204f7feba574cee43027e1eee3ef2de PYTHON_LIBRARY.md -d14e5e605ba26ce24ec0a85c053e00577f6382be97d133c8b2b2662c43fb6375 README.md +8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c PYTHON_LIBRARY.md +e29158a6782790a9aa7c3c00566e45387e9ce5e673c9482a9b898661dff08d28 README.md c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb RELEASE.md 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a RELEASE_ASSETS.md 858ea89e51f4217a45b042979c718d17f9c2481dc1494c0ba4f49e4ff31caa96 RELEASE_CHECKLIST.md @@ -37,11 +37,12 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 26e0ab83aa1ff6bd0ae3882787a92663aa19db128e1241ffc99375fb9ca0c121 VERSION 8f004523844f4b429a9ed8e15e5d46e2957a4c9854a772520267c34be6cf7998 contract-tests/test_against_main.py 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py +5c1d12bf6337372eac54c08109583c439db3952bafb0008afd0b9c1402cfa2cd json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py -1fdcaad555395a478241288fe757bac93ecb28d5da4a5491eaf64c3af5b439a6 json_api_forge/client.py -879c685c76c87e073bec9a3ff933a9e5103ddc948a172a0b8ea3eabeb2a9081d json_api_forge/control_plane.py -4b2333de0eea36838e8641d668563075a6ff2a269c9a4894c0025a4c84c88537 json_api_forge/enterprise.py -0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py +c850df04c590083a73afc0b66cf1ef2c4eede44c5b38c4de6ab14451f836490a json_api_forge/client.py +5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c json_api_forge/control_plane.py +c0f60c048a1512d9fe820055ce3606551c951b8a9a909d0ff999cf3f8829abe2 json_api_forge/enterprise.py +477f7e0957458e597c2a78b6d4522a8cce0f8ba306f76f4da2dc5d98ac436aa9 json_api_forge/errors.py 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705 json_api_forge/integrations/__init__.py 7dde121bcbe377560c684cbfc5875f0dc099e4d393eb494ce61da9731d562d54 json_api_forge/integrations/ddm.py 100f1f5e0bf1172503ca9ea63614d5616a7f46395ac67e3dfc9b20ce10d863ff json_api_forge/integrations/younglion.py @@ -50,11 +51,18 @@ c2d0db468c7e41c65791983f512b484cb850c5f65ae725069b7aaab6cc78b3f8 json_api_forge 0d7c4adff9f102814c0b6b24ca8745d138b3cca96febea98b2cf355de5d02c4d json_api_forge/pagination.py 01ba4719c80b6fe911b091a7c05124b64eeece964e09c058ef8f9805daca546b json_api_forge/py.typed d0cf58ccb32c816f0c85cd4a29a28b976343409647e4affcd84932acf98898c2 pyproject.toml -915a198346437156ef6388e6d3ac4f859478c26d9b48abf5a89e65a30d6e96ed scripts/check_critical_coverage.py -11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a scripts/check_manifest.py +dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64 scripts/check_critical_coverage.py +fc719b85b872e730b347f4674e2aaa3a3bc8bff3dfe361b41e46837dee709062 scripts/check_manifest.py 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py +50830855c94d3e55c4bd74d6df786d95875929fbe72653fd85ff98d8c2250d3b tests/test_async_attachment_io.py +3c00da7a012803b247ea5dc5820bb8c6448e9798f60e38abb09ff801ce01c546 tests/test_async_session_races.py +668b74e65c5d3ce0769feef88b5d9aeb389f6b69c444630d1f5929f0d9399e1c tests/test_async_transport_cleanup.py +cc6694028ea0148fc4e5dc7877740c1133b21afee030390c9e039b68aa34ce79 tests/test_cluster_lifecycle.py +ff662164581e2f72607604cb9f09d5c56b88440d171aa0242716ff1d693b36b4 tests/test_manifest_security.py +61109979cdb492ac7976947d804b3e717871fa367346f3cdfa911382b25c59a3 tests/test_night_cluster.py 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py 9959ece37e0df7010bb2c87407e843b53a8d4b5ccfb9693e4fd6a4fb2e8943d0 tests/test_python_library.py -fd19f7b746b195fddbcd2997b787b68d625f3fe2b0cb9899bcb44570ff563766 tests/test_v051_security.py +40db1d05ac5607c4bae91b750fb6c431791fbfeb27c090684b46b4908eea345a tests/test_response_json.py +1a44d0909278cc2750a80f757631dd2f51d1dc6a3ea206f5ac85f5210f9de7a5 tests/test_v051_security.py fa33f2147dd8e704684e752e5f4d0caec848f9c8ffb176a2dc202d8cf1513e4d tests/test_v052_pagination.py diff --git a/PYTHON_LIBRARY.md b/PYTHON_LIBRARY.md index 6b93e26..a873f97 100644 --- a/PYTHON_LIBRARY.md +++ b/PYTHON_LIBRARY.md @@ -43,6 +43,8 @@ Iteration fetches on demand, buffers at most one bounded response and stops at ` Collection iterators and cluster bulk helpers are bounded. The default bulk input ceiling is 10,000 items; raise it only after evaluating memory and downstream capacity. +Multipart file uploads are never retried or failed over automatically, even with an idempotency key. Retry explicitly with a fresh body after checking the upload outcome. Cancelling async bulk work, including cancellation raised by an individual operation, cancels and waits for its sibling operations before returning. Completed server-side effects cannot be rolled back by cancellation. + ## Control-plane security - Remote HTTP is rejected; `allow_insecure_http=True` is restricted to loopback development endpoints. diff --git a/README.md b/README.md index 1f587d2..4d2ce19 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,8 @@ notes = asyncio.run(main()) Bounded retries are opt-in. The retry/failover policy treats GET, HEAD, OPTIONS, PUT and DELETE as idempotent HTTP methods. Other methods, including POST/PATCH, need an explicit idempotency key before replay is permitted. The server endpoint must actually support idempotency; adding a header alone does not make arbitrary side effects replay-safe. +JSON responses use Python integers and finite floating-point numbers. Non-standard `NaN`/`Infinity` constants, exponent values outside the supported finite float range (for example, `1e999`), and integers rejected by the interpreter's configured digit limit produce `ForgeHTTPError` with the original response status and request ID. Finite numbers and JSON `null` retain their usual Python values. + ## Editor control-plane client ```python @@ -82,10 +84,16 @@ The control-plane clients cover founder setup, invitation enrollment, profiles, Security defaults include HTTPS-only remote URLs, no ambient proxy inheritance, no redirects or retained cookies, strict same-origin paths, exact token formats, bounded response bodies, non-symlink attachment snapshots and atomic downloads. WebRTC call tickets are placed in a URL fragment for the local call client, never in an HTTP query. v0.5.2 uses the revision 2 HTTP signaling call client. +Async attachment snapshots, multipart file reads and atomic download writes run in worker threads so slow storage does not freeze other async tasks. Cancellation waits for an active disk operation to finish before closing its file or returning. If cancellation arrives during the atomic save, the destination may already contain the complete download when cancellation is reported; no background writer remains after that report. + +The async control-plane client discards an authentication response with `ForgeSessionError` if the session changed while it was pending, including sign-out, client close or a newer sign-in. An old request's unauthorized/logout response cannot clear a newer session. + ## Cluster and optional integrations `ForgeCluster` and `AsyncForgeCluster` provide deterministic routing, circuit breakers, bounded bulk work and idempotency-gated failover across multiple Forge endpoints. +Cluster shutdown attempts every client close and reports the first error after cleanup finishes. Async shutdown also finishes cleanup before propagating caller cancellation. Bulk and health operations cancel and drain their owned siblings before returning a cancelled result; worker/item counts and integer routing policies are checked before work starts. + YoungLion/DDM adapters remain optional: ```bash diff --git a/json_api_forge/_io.py b/json_api_forge/_io.py new file mode 100644 index 0000000..516f805 --- /dev/null +++ b/json_api_forge/_io.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from functools import partial +from typing import Any, TypeVar + +_T = TypeVar("_T") + + +async def run_io(function: Callable[..., _T], *args: Any, **kwargs: Any) -> _T: + """Run blocking IO without abandoning owned files when the caller cancels.""" + worker = asyncio.create_task(asyncio.to_thread(partial(function, *args, **kwargs))) + return await wait_owned(worker) + + +async def wait_owned(worker: asyncio.Future[_T]) -> _T: + """Finish owned cleanup before returning caller cancellation.""" + try: + return await asyncio.shield(worker) + except asyncio.CancelledError: + # An owned operation must finish before its context can return or close + # its resources, even if cancellation repeats during cleanup. + while not worker.done(): + try: + await asyncio.shield(worker) + except asyncio.CancelledError: + continue + except Exception: + break + if not worker.cancelled(): + worker.exception() + raise diff --git a/json_api_forge/client.py b/json_api_forge/client.py index d430a50..1bec902 100644 --- a/json_api_forge/client.py +++ b/json_api_forge/client.py @@ -9,11 +9,13 @@ import unicodedata import uuid from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence +from contextlib import asynccontextmanager from typing import Any, Literal from urllib.parse import quote, unquote, urlsplit import httpx +from ._io import run_io, wait_owned from .errors import ForgeHTTPError, ForgeResponseTooLarge, ForgeTransportError, safe_error_text from .models import ForgeResponse, JsonObject, RequestAttempt from .options import RetryPolicy @@ -149,12 +151,27 @@ def _headers( return values +def _reject_json_constant(value: str) -> Any: + raise ValueError(f"Invalid JSON number constant: {value}") + + +def _finite_json_float(value: str) -> float: + result = float(value) + if not math.isfinite(result): + raise ValueError("JSON number exceeds the supported finite float range") + return result + + def _decode_response(response: httpx.Response, content: bytes, *, expect_json: bool) -> ForgeResponse[Any]: detail: Any + valid_json = True try: - decoded = json.loads(content) if content else None - except (UnicodeDecodeError, json.JSONDecodeError, RecursionError): + decoded = json.loads(content, parse_constant=_reject_json_constant, parse_float=_finite_json_float) if content else None + except (ValueError, RecursionError): + # ValueError also covers the interpreter's bounded integer conversion, + # UnicodeDecodeError and JSONDecodeError; keep these HTTP failures observable. decoded = None + valid_json = False if response.is_redirect: raise ForgeHTTPError( response.status_code, @@ -175,7 +192,7 @@ def _decode_response(response: httpx.Response, content: bytes, *, expect_json: b response.headers.get("X-Request-ID"), response.headers.get("Retry-After"), ) - if expect_json and decoded is None and content: + if expect_json and not valid_json and content: raise ForgeHTTPError(response.status_code, "Server returned non-JSON content", response.headers.get("X-Request-ID")) data = decoded if expect_json else content return ForgeResponse( @@ -234,6 +251,55 @@ async def _bounded_content_async(response: httpx.Response, max_response_bytes: i return bytes(chunks) +class _AsyncFileStream(httpx.AsyncByteStream): + """Keep httpx multipart's synchronous file reads off the event loop.""" + + def __init__(self, stream: httpx.SyncByteStream): + self._chunks = iter(stream) + + async def __aiter__(self) -> AsyncIterator[bytes]: + while True: + chunk = await run_io(next, self._chunks, None) + if chunk is None: + break + yield chunk + + +class _OwnedResponseStream(httpx.AsyncByteStream): + """Finish releasing the HTTP connection even when cancellation repeats.""" + + def __init__(self, stream: httpx.AsyncByteStream): + self._stream = stream + + async def __aiter__(self) -> AsyncIterator[bytes]: + async for chunk in self._stream: + yield chunk + + async def aclose(self) -> None: + # httpx marks a Response closed before awaiting the underlying stream. + # Shield here, including its automatic close at the end of body reads; + # a later Response.aclose() cannot resume an interrupted first close. + await wait_owned(asyncio.create_task(self._stream.aclose())) + + +@asynccontextmanager +async def _stream_async(client: httpx.AsyncClient, method: str, path: str, **kwargs: Any) -> AsyncIterator[httpx.Response]: + if kwargs.get("files") is None: + async with client.stream(method, path, **kwargs) as response: + response.stream = _OwnedResponseStream(response.stream) + yield response + return + # Request assembly also seeks/fstats files to compute Content-Length. + request = await run_io(client.build_request, method, path, **kwargs) + request.stream = _AsyncFileStream(request.stream) + response = await client.send(request, stream=True) + response.stream = _OwnedResponseStream(response.stream) + try: + yield response + finally: + await response.aclose() + + class ForgeClient: """Synchronous, bounded client for one JSON API Forge server.""" @@ -509,7 +575,8 @@ async def request( try: self._client.cookies.clear() try: - async with self._client.stream( + async with _stream_async( + self._client, normalized_method, normalized_path, json=json_body, @@ -642,7 +709,7 @@ async def batch_create_items(self, project: str, resource: str, payloads: Sequen return await self.request("POST", f"api/{_segment(project)}/v1/{_route(resource)}/_batch", json_body=batch_payload(payloads)) async def aclose(self) -> None: - await self._client.aclose() + await wait_owned(asyncio.create_task(self._client.aclose())) async def __aenter__(self) -> AsyncForgeClient: return self diff --git a/json_api_forge/control_plane.py b/json_api_forge/control_plane.py index 3fae545..860d68f 100644 --- a/json_api_forge/control_plane.py +++ b/json_api_forge/control_plane.py @@ -4,14 +4,15 @@ import re import stat import tempfile -from collections.abc import Iterator, Mapping, Sequence -from contextlib import contextmanager +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence +from contextlib import asynccontextmanager, contextmanager from pathlib import Path from typing import Any, BinaryIO from urllib.parse import unquote, urlencode, urljoin, urlsplit, urlunsplit import httpx +from ._io import run_io from .client import AsyncForgeClient, ForgeClient, _base_url, _controls, _route, _segment from .errors import ForgeHTTPError, ForgeSessionError from .models import ForgeResponse, JsonObject @@ -184,6 +185,16 @@ def _open_attachment(source: str | os.PathLike[str], max_bytes: int) -> Iterator snapshot.close() +@asynccontextmanager +async def _open_attachment_async(source: str | os.PathLike[str], max_bytes: int) -> AsyncIterator[tuple[str, BinaryIO]]: + attachment = _open_attachment(source, max_bytes) + try: + yield await run_io(attachment.__enter__) + finally: + # run_io drains an interrupted snapshot/read before its file closes. + await run_io(attachment.__exit__, None, None, None) + + class _ControlPlaneBase: def _initialize_control_plane(self, base_url: str, *, allow_insecure_http: bool, max_attachment_bytes: int) -> None: if not 1024 <= int(max_attachment_bytes) <= 512 * 1024 * 1024: @@ -191,6 +202,7 @@ def _initialize_control_plane(self, base_url: str, *, allow_insecure_http: bool, self._base_url = _base_url(base_url, allow_insecure_http=allow_insecure_http) self.max_attachment_bytes = int(max_attachment_bytes) self._session = bytearray() + self._session_revision = 0 @property def has_session(self) -> bool: @@ -202,6 +214,7 @@ def set_session_token(self, token: str) -> None: self._session.extend(encoded) def clear_session(self) -> None: + self._session_revision += 1 for index in range(len(self._session)): self._session[index] = 0 self._session.clear() @@ -222,7 +235,9 @@ def _headers(self, *, authenticated: bool, extra: Mapping[str, str] | None = Non values[name] = value return values - def _adopt_authentication(self, response: ForgeResponse[Any]) -> ForgeResponse[JsonObject]: + def _adopt_authentication(self, response: ForgeResponse[Any], *, revision: int | None = None) -> ForgeResponse[JsonObject]: + if revision is not None and revision != self._session_revision: + raise ForgeSessionError("Session changed while authentication was pending; discard the stale response") verified = _object_response(response) payload = dict(verified.data) token = payload.pop("access_token", None) @@ -553,6 +568,7 @@ async def _request( expect_json: bool = True, max_response_bytes: int | None = None, ) -> ForgeResponse[Any]: + revision = self._session_revision try: return await self._transport.request( method, @@ -565,7 +581,7 @@ async def _request( max_response_bytes=max_response_bytes, ) except ForgeHTTPError as exc: - if authenticated and exc.status_code == 401: + if authenticated and exc.status_code == 401 and revision == self._session_revision: self.clear_session() raise @@ -580,6 +596,7 @@ async def setup_founder( display_name: str, ) -> ForgeResponse[JsonObject]: self.clear_session() + revision = self._session_revision token = _credential(setup_token, name="setup token", minimum=32) response = await self._request( "POST", @@ -588,17 +605,18 @@ async def setup_founder( authenticated=False, extra_headers={"X-Forge-Setup-Token": token}, ) - return self._adopt_authentication(response) + return self._adopt_authentication(response, revision=revision) async def login(self, username: str, password: str) -> ForgeResponse[JsonObject]: self.clear_session() + revision = self._session_revision response = await self._request( "POST", _control_path("auth", "login"), json_body=_profile_payload(username, password), authenticated=False, ) - return self._adopt_authentication(response) + return self._adopt_authentication(response, revision=revision) async def register( self, @@ -608,16 +626,19 @@ async def register( display_name: str, ) -> ForgeResponse[JsonObject]: self.clear_session() + revision = self._session_revision payload = _profile_payload(username, password, display_name) payload["invitation"] = _credential(invitation, name="invitation", pattern=_INVITATION_TOKEN) response = await self._request("POST", _control_path("auth", "register"), json_body=payload, authenticated=False) - return self._adopt_authentication(response) + return self._adopt_authentication(response, revision=revision) async def logout(self) -> ForgeResponse[Any]: + revision = self._session_revision try: return await self._request("POST", _control_path("auth", "logout"), expect_json=False) finally: - self.clear_session() + if revision == self._session_revision: + self.clear_session() async def capabilities(self) -> ForgeResponse[JsonObject]: return _object_response(await self._request("GET", _control_path("capabilities"))) @@ -752,7 +773,7 @@ async def attachments(self, area_id: str, *, limit: int = 100) -> ForgeResponse[ return _object_response(await self._request("GET", _control_path("areas", area_id, "attachments"), params={"limit": limit})) async def upload_attachment(self, area_id: str, source: str | os.PathLike[str]) -> ForgeResponse[JsonObject]: - with _open_attachment(source, self.max_attachment_bytes) as (name, handle): + async with _open_attachment_async(source, self.max_attachment_bytes) as (name, handle): response = await self._request( "POST", _control_path("areas", area_id, "attachments"), @@ -773,7 +794,7 @@ async def download_attachment( ) if not isinstance(response.data, bytes): raise ForgeHTTPError(response.status_code, "Attachment response was not binary", response.request_id) - return _copy_response(response, _atomic_save(target, response.data)) + return _copy_response(response, await run_io(_atomic_save, target, response.data)) async def start_call(self, area_id: str, *, mode: str = "video") -> ForgeResponse[JsonObject]: return _object_response(await self._request("POST", _control_path("calls"), json_body={"area_id": area_id, "mode": mode})) diff --git a/json_api_forge/enterprise.py b/json_api_forge/enterprise.py index 40a9b61..ed30d6d 100644 --- a/json_api_forge/enterprise.py +++ b/json_api_forge/enterprise.py @@ -12,6 +12,7 @@ from itertools import islice from typing import Any, Generic, TypeVar +from ._io import wait_owned from .client import AsyncForgeClient, ForgeClient, _route, _segment from .errors import ForgeClusterUnavailable, ForgeHTTPError, ForgeTransportError from .models import ForgeResponse @@ -39,7 +40,7 @@ class ForgeEndpoint: def __post_init__(self) -> None: if not self.name or len(self.name) > 64 or any(character in self.name for character in "\r\n\0"): raise ValueError("endpoint name is invalid") - if not 1 <= self.weight <= 100: + if type(self.weight) is not int or not 1 <= self.weight <= 100: raise ValueError("endpoint weight must be between 1 and 100") @@ -49,7 +50,7 @@ class CircuitBreakerPolicy: recovery_seconds: float = 15.0 def __post_init__(self) -> None: - if not 1 <= self.failure_threshold <= 100: + if type(self.failure_threshold) is not int or not 1 <= self.failure_threshold <= 100: raise ValueError("failure_threshold must be between 1 and 100") if not 0 <= self.recovery_seconds <= 3600: raise ValueError("recovery_seconds must be between 0 and 3600") @@ -77,11 +78,14 @@ def _retryable(error: Exception) -> bool: def _failover_permitted(method: str, options: Mapping[str, Any]) -> bool: + if options.get("files") is not None: + # Multipart bodies may contain consumed streams and cannot be safely replayed. + return False return method.upper() in {"GET", "HEAD", "OPTIONS", "PUT", "DELETE"} or bool(options.get("idempotency_key")) def _bounded_values(values: Iterable[T], maximum: int) -> list[T]: - if not 1 <= maximum <= 100_000: + if type(maximum) is not int or not 1 <= maximum <= 100_000: raise ValueError("max_items must be between 1 and 100000") result = list(islice(values, maximum + 1)) if len(result) > maximum: @@ -89,6 +93,19 @@ def _bounded_values(values: Iterable[T], maximum: int) -> list[T]: return result +async def _gather_owned(tasks: Sequence[asyncio.Task[T]]) -> list[T]: + try: + return list(await asyncio.gather(*tasks)) + except BaseException: + # A cancelled item makes gather return before its siblings finish. + # Cancel once, then shield/drain cleanup even if the caller cancels again. + for task in tasks: + if not task.done(): + task.cancel() + await wait_owned(asyncio.gather(*tasks, return_exceptions=True)) + raise + + class ForgeCluster: """Failover/routing facade for horizontally deployed Forge services.""" @@ -211,9 +228,9 @@ def bulk_create( idempotency_key: Callable[[int, Mapping[str, Any]], str | None] | None = None, routing_key: Callable[[int, Mapping[str, Any]], str | None] | None = None, ) -> list[BulkResult[ForgeResponse[Any]]]: - values = [dict(payload) for payload in _bounded_values(payloads, max_items)] - if not 1 <= max_workers <= 64: + if type(max_workers) is not int or not 1 <= max_workers <= 64: raise ValueError("max_workers must be between 1 and 64") + values = [dict(payload) for payload in _bounded_values(payloads, max_items)] results: list[BulkResult[ForgeResponse[Any]] | None] = [None] * len(values) def submit(index: int) -> ForgeResponse[Any]: @@ -246,8 +263,15 @@ def health_all(self) -> dict[str, ForgeResponse[Any] | Exception]: return result def close(self) -> None: + first_error: Exception | None = None for client in self._clients.values(): - client.close() + try: + client.close() + except Exception as exc: + if first_error is None: + first_error = exc + if first_error is not None: + raise first_error def __enter__(self) -> ForgeCluster: return self @@ -368,7 +392,7 @@ async def bulk( concurrency: int = 16, max_items: int = 10_000, ) -> list[BulkResult[T]]: - if not 1 <= concurrency <= 256: + if type(concurrency) is not int or not 1 <= concurrency <= 256: raise ValueError("concurrency must be between 1 and 256") bounded_operations = _bounded_values(operations, max_items) semaphore = asyncio.Semaphore(concurrency) @@ -380,7 +404,8 @@ async def run(index: int, operation: Callable[[], Awaitable[T]]) -> BulkResult[T except Exception as exc: return BulkResult(index=index, error=exc) - return list(await asyncio.gather(*(run(index, operation) for index, operation in enumerate(bounded_operations)))) + tasks = [asyncio.create_task(run(index, operation)) for index, operation in enumerate(bounded_operations)] + return await _gather_owned(tasks) async def health_all(self) -> dict[str, ForgeResponse[Any] | Exception]: async def health(endpoint: ForgeEndpoint): @@ -389,10 +414,13 @@ async def health(endpoint: ForgeEndpoint): except Exception as exc: return endpoint.name, exc - return dict(await asyncio.gather(*(health(endpoint) for endpoint in self.endpoints))) + return dict(await _gather_owned([asyncio.create_task(health(endpoint)) for endpoint in self.endpoints])) async def aclose(self) -> None: - await asyncio.gather(*(client.aclose() for client in self._clients.values())) + results = await wait_owned(asyncio.gather(*(client.aclose() for client in self._clients.values()), return_exceptions=True)) + for result in results: + if isinstance(result, BaseException): + raise result async def __aenter__(self) -> AsyncForgeCluster: return self diff --git a/json_api_forge/errors.py b/json_api_forge/errors.py index 013bef4..ebbc996 100644 --- a/json_api_forge/errors.py +++ b/json_api_forge/errors.py @@ -8,7 +8,9 @@ def safe_error_text(value: Any) -> str: """Bound printable diagnostics; structured HTTP detail remains available separately.""" - result = str(value)[:4096] + # Truncating first can remove the final @ from URL credentials and leave + # their private prefix visible. Redact the complete field, then bound it. + result = str(value) result = re.sub(r"(?i)(https?://)[^/\s@]+@", r"\1[redacted]@", result) result = re.sub(r"(?i)\bBearer\s+[^\s,;\"']+", "Bearer [redacted]", result) result = re.sub( @@ -16,7 +18,7 @@ def safe_error_text(value: Any) -> str: r"\1[redacted]", result, ) - return "".join(" " if unicodedata.category(char) in {"Cc", "Cf"} else char for char in result) + return "".join(" " if unicodedata.category(char) in {"Cc", "Cf"} else char for char in result[:4096]) class ForgeError(RuntimeError): diff --git a/scripts/check_critical_coverage.py b/scripts/check_critical_coverage.py index 1eea9bb..5fb899d 100755 --- a/scripts/check_critical_coverage.py +++ b/scripts/check_critical_coverage.py @@ -8,6 +8,7 @@ from pathlib import Path MODULE_FLOORS: dict[str, float] = { + "json_api_forge/_io.py": 95.0, "json_api_forge/client.py": 80.0, "json_api_forge/pagination.py": 90.0, "json_api_forge/control_plane.py": 90.0, diff --git a/scripts/check_manifest.py b/scripts/check_manifest.py index 0d3800e..aa68213 100755 --- a/scripts/check_manifest.py +++ b/scripts/check_manifest.py @@ -3,6 +3,9 @@ import argparse import hashlib +import os +import re +import stat import subprocess from pathlib import Path @@ -10,6 +13,50 @@ MANIFEST = ROOT / "MANIFEST.sha256" +def _linked(metadata: os.stat_result) -> bool: + return stat.S_ISLNK(metadata.st_mode) or bool( + getattr(metadata, "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0) + ) + + +def _safe_path(relative: str) -> Path | None: + """Inspect every ancestor before looking inside it, including NTFS junctions.""" + current = ROOT + for part in ["", *relative.split("/")]: + if part: + current = current / part + try: + metadata = current.lstat() + except FileNotFoundError: + return None + if _linked(metadata): + raise ValueError(f"Release paths may not contain links or junctions: {relative}") + return current + + +def _fallback_files() -> set[str]: + """Enumerate extracted source without descending into excluded/link directories.""" + _safe_path("") + pending = [ROOT] + result = set() + while pending: + directory = pending.pop() + with os.scandir(directory) as entries: + for entry in entries: + path = directory / entry.name + relative = path.relative_to(ROOT).as_posix() + if not _release_path(relative): + continue + metadata = entry.stat(follow_symlinks=False) + if _linked(metadata): + raise ValueError(f"Release paths may not contain links or junctions: {relative}") + if stat.S_ISDIR(metadata.st_mode): + pending.append(path) + elif stat.S_ISREG(metadata.st_mode): + result.add(relative) + return result + + def digest(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: @@ -20,7 +67,7 @@ def digest(path: Path) -> str: def _release_path(relative: str) -> bool: parts = relative.split("/") - if not relative or any(part in {"", ".", ".."} for part in parts) or "\\" in relative: + if not relative or any(part in {"", ".", ".."} for part in parts) or "\\" in relative or ":" in relative: raise ValueError(f"Unsafe release path: {relative!r}") if any(ord(char) < 32 or ord(char) == 127 for char in relative): raise ValueError(f"Control character in release path: {relative!r}") @@ -68,15 +115,13 @@ def release_files() -> set[str]: ).decode() candidates = {value for value in raw.split(chr(0)) if value} except (OSError, subprocess.SubprocessError, UnicodeDecodeError, ValueError): - candidates = {path.relative_to(ROOT).as_posix() for path in ROOT.rglob("*")} + candidates = _fallback_files() result = set() for relative in candidates: if not _release_path(relative): continue - path = ROOT / relative - if path.is_symlink(): - raise ValueError(f"Release files may not be symbolic links: {relative}") - if path.is_file(): + path = _safe_path(relative) + if path is not None and path.is_file(): result.add(relative) return result @@ -104,14 +149,20 @@ def main(argv: list[str] | None = None) -> int: d, rel = line.split(" ", 1) except ValueError as exc: raise SystemExit(f"Malformed manifest line {n}") from exc + try: + allowed = _release_path(rel) + except ValueError as exc: + raise SystemExit(f"Unsafe manifest line {n}: {exc}") from exc + if not allowed or re.fullmatch(r"[0-9a-f]{64}", d) is None: + raise SystemExit(f"Invalid release path or digest on manifest line {n}") if rel in expected: raise SystemExit(f"Duplicate manifest entry: {rel}") expected[rel] = d actual = release_files() errors = [] for rel, d in expected.items(): - p = ROOT / rel - if not p.is_file(): + p = _safe_path(rel) + if p is None or not p.is_file(): errors.append(f"missing: {rel}") elif digest(p) != d: errors.append(f"hash mismatch: {rel}") diff --git a/tests/test_async_attachment_io.py b/tests/test_async_attachment_io.py new file mode 100644 index 0000000..9280e3c --- /dev/null +++ b/tests/test_async_attachment_io.py @@ -0,0 +1,240 @@ +from __future__ import annotations + +import asyncio +import io +import threading +from contextlib import contextmanager +from pathlib import Path + +import httpx +import pytest + +from json_api_forge import AsyncEditorControlPlaneClient, AsyncForgeClient, control_plane +from json_api_forge._io import run_io + +SESSION = "jfe_session_" + "A" * 48 + + +@pytest.mark.asyncio +async def test_attachment_snapshot_save_and_close_run_outside_event_loop(tmp_path: Path, monkeypatch) -> None: + loop_thread = threading.get_ident() + source = tmp_path / "source.txt" + source.write_bytes(b"upload data") + original_open = control_plane._open_attachment + original_save = control_plane._atomic_save + seen = [] + + @contextmanager + def checked_open(*args): + assert threading.get_ident() != loop_thread + with original_open(*args) as upload: + seen.append("snapshot") + yield upload + assert threading.get_ident() != loop_thread + seen.append("closed") + + def checked_save(*args): + assert threading.get_ident() != loop_thread + seen.append("save") + return original_save(*args) + + monkeypatch.setattr(control_plane, "_open_attachment", checked_open) + monkeypatch.setattr(control_plane, "_atomic_save", checked_save) + + async def handler(request): + if request.method == "POST": + assert b"upload data" in await request.aread() + return httpx.Response(201, json={"id": "one"}) + return httpx.Response(200, content=b"download data") + + async with AsyncEditorControlPlaneClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + client.set_session_token(SESSION) + await client.upload_attachment("area", source) + saved = await client.download_attachment("one", tmp_path / "saved.txt") + assert saved.data.read_bytes() == b"download data" + assert seen == ["snapshot", "closed", "save"] + + +@pytest.mark.asyncio +async def test_async_multipart_seek_and_read_run_outside_event_loop() -> None: + loop_thread = threading.get_ident() + reads = [] + + class CheckedFile(io.BytesIO): + def seek(self, *args): + assert threading.get_ident() != loop_thread + return super().seek(*args) + + def read(self, *args): + assert threading.get_ident() != loop_thread + reads.append(True) + return super().read(*args) + + async def handler(request): + assert b"multipart bytes" in await request.aread() + return httpx.Response(200, json={"ok": True}) + + handle = CheckedFile(b"multipart bytes") + async with AsyncForgeClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + assert (await client.request("POST", "upload", files={"upload": ("file.txt", handle)})).data["ok"] + assert reads + assert not handle.closed # Generic caller-owned files remain caller-owned. + handle.close() + + +@pytest.mark.asyncio +async def test_cancelled_snapshot_drains_before_closing_owned_file(tmp_path: Path, monkeypatch) -> None: + started = threading.Event() + release = threading.Event() + closed = threading.Event() + handle = io.BytesIO(b"snapshot") + + @contextmanager + def slow_open(*_args): + started.set() + assert release.wait(5) + try: + yield "snapshot.txt", handle + finally: + handle.close() + closed.set() + + monkeypatch.setattr(control_plane, "_open_attachment", slow_open) + async with AsyncEditorControlPlaneClient("https://forge.test") as client: + client.set_session_token(SESSION) + task = asyncio.create_task(client.upload_attachment("area", tmp_path / "unused")) + try: + assert await asyncio.to_thread(started.wait, 2) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + assert not closed.is_set() + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert closed.is_set() and handle.closed + + +@pytest.mark.asyncio +async def test_cancelled_io_preserves_cancellation_if_worker_fails() -> None: + started = threading.Event() + release = threading.Event() + + def failure(): + started.set() + assert release.wait(5) + raise OSError("disk failed") + + task = asyncio.create_task(run_io(failure)) + try: + assert await asyncio.to_thread(started.wait, 2) + task.cancel() + await asyncio.sleep(0) + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_io_failure_is_reported_normally() -> None: + def failure(): + raise OSError("disk failed") + + with pytest.raises(OSError, match="disk failed"): + await run_io(failure) + + +@pytest.mark.asyncio +async def test_cancelled_multipart_drains_read_before_snapshot_closes(tmp_path: Path, monkeypatch) -> None: + reading = threading.Event() + release = threading.Event() + closed = threading.Event() + + class SlowFile(io.BytesIO): + def read(self, *args): + reading.set() + assert release.wait(5) + assert not self.closed + return super().read(*args) + + handle = SlowFile(b"slow file") + + @contextmanager + def snapshot(*_args): + try: + yield "snapshot.bin", handle + finally: + handle.close() + closed.set() + + class ReadingTransport(httpx.AsyncBaseTransport): + async def handle_async_request(self, request): + await request.aread() + return httpx.Response(201, json={"id": "one"}) + + monkeypatch.setattr(control_plane, "_open_attachment", snapshot) + async with AsyncEditorControlPlaneClient("https://forge.test", transport=ReadingTransport()) as client: + client.set_session_token(SESSION) + task = asyncio.create_task(client.upload_attachment("area", tmp_path / "unused")) + try: + assert await asyncio.to_thread(reading.wait, 2) + task.cancel() + await asyncio.sleep(0) + assert not task.done() and not closed.is_set() + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert closed.is_set() + + +@pytest.mark.asyncio +async def test_cancelled_download_finishes_atomic_save_before_returning(tmp_path: Path, monkeypatch) -> None: + saving = threading.Event() + release = threading.Event() + original_save = control_plane._atomic_save + target = tmp_path / "saved.bin" + target.write_bytes(b"old") + + def slow_save(*args): + saving.set() + assert release.wait(5) + return original_save(*args) + + monkeypatch.setattr(control_plane, "_atomic_save", slow_save) + async with AsyncEditorControlPlaneClient( + "https://forge.test", transport=httpx.MockTransport(lambda _: httpx.Response(200, content=b"new")) + ) as client: + client.set_session_token(SESSION) + task = asyncio.create_task(client.download_attachment("one", target)) + try: + assert await asyncio.to_thread(saving.wait, 2) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + assert target.read_bytes() == b"old" + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert target.read_bytes() == b"new" + assert list(tmp_path.iterdir()) == [target] + + +@pytest.mark.asyncio +async def test_failed_snapshot_is_reported_without_attempting_upload(tmp_path: Path) -> None: + requests = [] + + def handler(request): + requests.append(request) + return httpx.Response(200, json={}) + + async with AsyncEditorControlPlaneClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + client.set_session_token(SESSION) + with pytest.raises(ValueError, match="unavailable"): + await client.upload_attachment("area", tmp_path / "missing") + assert not requests diff --git a/tests/test_async_session_races.py b/tests/test_async_session_races.py new file mode 100644 index 0000000..da26acd --- /dev/null +++ b/tests/test_async_session_races.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import asyncio + +import httpx +import pytest + +from json_api_forge import AsyncEditorControlPlaneClient, ForgeHTTPError, ForgeSessionError + +OLD_SESSION = "jfe_session_" + "A" * 48 +NEW_SESSION = "jfe_session_" + "B" * 48 + + +@pytest.mark.asyncio +async def test_old_unauthorized_request_cannot_clear_new_session() -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def handler(request): + if request.headers.get("Authorization") == f"Bearer {OLD_SESSION}": + started.set() + await release.wait() + return httpx.Response(401, json={"detail": "expired"}) + assert request.headers["Authorization"] == f"Bearer {NEW_SESSION}" + return httpx.Response(200, json={"username": "new account"}) + + async with AsyncEditorControlPlaneClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + client.set_session_token(OLD_SESSION) + previous = asyncio.create_task(client.profile()) + await started.wait() + client.set_session_token(NEW_SESSION) + release.set() + with pytest.raises(ForgeHTTPError): + await previous + assert client.has_session + assert (await client.profile()).data["username"] == "new account" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("authentication", ["login", "setup", "register"]) +@pytest.mark.parametrize("interruption", ["clear", "close", "new_session"]) +async def test_late_authentication_cannot_restore_or_replace_session(authentication, interruption) -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def handler(_request): + started.set() + await release.wait() + return httpx.Response(200, json={"access_token": OLD_SESSION, "profile": {"username": "old"}}) + + async with AsyncEditorControlPlaneClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + if authentication == "login": + operation = client.login("old", "password") + elif authentication == "setup": + operation = client.setup_founder("S" * 32, "old", "password", "Old") + else: + operation = client.register("jfi_" + "S" * 48, "old", "password", "Old") + pending = asyncio.create_task(operation) + await started.wait() + if interruption == "clear": + client.clear_session() + elif interruption == "close": + await client.aclose() + else: + client.set_session_token(NEW_SESSION) + release.set() + with pytest.raises(ForgeSessionError, match="changed"): + await pending + assert client.has_session == (interruption == "new_session") + if client.has_session: + assert client._headers(authenticated=True)["Authorization"] == f"Bearer {NEW_SESSION}" + + +@pytest.mark.asyncio +async def test_old_logout_cannot_clear_replacement_session() -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def handler(_request): + started.set() + await release.wait() + return httpx.Response(204) + + async with AsyncEditorControlPlaneClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + client.set_session_token(OLD_SESSION) + pending = asyncio.create_task(client.logout()) + await started.wait() + client.set_session_token(NEW_SESSION) + release.set() + await pending + assert client._headers(authenticated=True)["Authorization"] == f"Bearer {NEW_SESSION}" + + +@pytest.mark.asyncio +async def test_latest_login_wins_when_responses_arrive_out_of_order() -> None: + started = asyncio.Event() + release = asyncio.Event() + + async def handler(request): + if b'"username":"old"' in request.content: + started.set() + await release.wait() + return httpx.Response(200, json={"access_token": OLD_SESSION}) + return httpx.Response(200, json={"access_token": NEW_SESSION}) + + async with AsyncEditorControlPlaneClient("https://forge.test", transport=httpx.MockTransport(handler)) as client: + old_login = asyncio.create_task(client.login("old", "password")) + await started.wait() + await client.login("new", "password") + release.set() + with pytest.raises(ForgeSessionError, match="changed"): + await old_login + assert client._headers(authenticated=True)["Authorization"] == f"Bearer {NEW_SESSION}" + + +@pytest.mark.asyncio +async def test_current_unauthorized_request_clears_only_current_session() -> None: + async with AsyncEditorControlPlaneClient( + "https://forge.test", transport=httpx.MockTransport(lambda _: httpx.Response(401, json={"detail": "expired"})) + ) as client: + client.set_session_token(NEW_SESSION) + with pytest.raises(ForgeHTTPError): + await client.profile() + assert not client.has_session diff --git a/tests/test_async_transport_cleanup.py b/tests/test_async_transport_cleanup.py new file mode 100644 index 0000000..4a7ef3f --- /dev/null +++ b/tests/test_async_transport_cleanup.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import asyncio +import io + +import httpx +import pytest + +from json_api_forge import AsyncForgeClient, ForgeTransportError + + +class SlowClosingStream(httpx.AsyncByteStream): + def __init__(self, outcome: str): + self.outcome = outcome + self.read_started = asyncio.Event() + self.close_started = asyncio.Event() + self.release_close = asyncio.Event() + self.closed = False + self.close_calls = 0 + + async def __aiter__(self): + self.read_started.set() + if self.outcome == "cancel_read": + await asyncio.Event().wait() + if self.outcome == "read_error": + raise httpx.ReadError("response interrupted") + yield b'{"ok":true}' + + async def aclose(self): + self.close_calls += 1 + self.close_started.set() + await self.release_close.wait() + self.closed = True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("multipart", [False, True]) +@pytest.mark.parametrize("outcome", ["success", "read_error", "cancel_read"]) +async def test_response_close_is_drained_before_repeated_cancellation_returns(multipart, outcome): + stream = SlowClosingStream(outcome) + + async def handle(request): + await request.aread() + return httpx.Response(200, stream=stream) + + async with AsyncForgeClient("https://forge.test", transport=httpx.MockTransport(handle)) as client: + options = {"files": {"upload": ("sample.txt", io.BytesIO(b"sample"))}} if multipart else {} + task = asyncio.create_task(client.request("POST", "upload", **options)) + try: + await asyncio.wait_for(stream.read_started.wait(), 5) + if outcome == "cancel_read": + task.cancel() + await asyncio.wait_for(stream.close_started.wait(), 5) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done(), "Request returned before its response connection was closed" + stream.release_close.set() + with pytest.raises(asyncio.CancelledError): + await task + assert stream.closed + assert stream.close_calls == 1 + finally: + stream.release_close.set() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("multipart", [False, True]) +async def test_response_read_failure_preserves_error_after_cleanup(multipart): + stream = SlowClosingStream("read_error") + stream.release_close.set() + transport = httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)) + async with AsyncForgeClient("https://forge.test", transport=transport) as client: + options = {"files": {"upload": ("sample.txt", io.BytesIO(b"sample"))}} if multipart else {} + with pytest.raises(ForgeTransportError, match="response interrupted"): + await client.request("POST", "upload", **options) + assert stream.closed + assert stream.close_calls == 1 + + +@pytest.mark.asyncio +async def test_single_client_close_finishes_transport_cleanup_before_returning_cancellation(): + started = asyncio.Event() + release = asyncio.Event() + completed = asyncio.Event() + + class Transport(httpx.AsyncBaseTransport): + async def handle_async_request(self, request): + return httpx.Response(200, json={}) + + async def aclose(self): + started.set() + await release.wait() + completed.set() + + client = AsyncForgeClient("https://forge.test", transport=Transport()) + task = asyncio.create_task(client.aclose()) + try: + await asyncio.wait_for(started.wait(), 5) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done(), "Client returned before its HTTP transport was closed" + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert completed.is_set() + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) diff --git a/tests/test_cluster_lifecycle.py b/tests/test_cluster_lifecycle.py new file mode 100644 index 0000000..9a72817 --- /dev/null +++ b/tests/test_cluster_lifecycle.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from json_api_forge import AsyncForgeCluster, CircuitBreakerPolicy, ForgeCluster, ForgeEndpoint + +ENDPOINTS = [ForgeEndpoint("first", "https://first.test"), ForgeEndpoint("second", "https://second.test")] + + +def test_sync_close_failure_still_closes_every_client() -> None: + closed = [] + + class Client: + def __init__(self, name): + self.name = name + + def close(self): + closed.append(self.name) + if self.name == "first": + raise OSError("first close failed") + + cluster = ForgeCluster(ENDPOINTS, client_factory=lambda endpoint: Client(endpoint.name)) + with pytest.raises(OSError, match="first close failed"): + cluster.close() + assert closed == ["first", "second"] + + +@pytest.mark.asyncio +async def test_async_close_failure_waits_for_every_client() -> None: + started = asyncio.Event() + release = asyncio.Event() + completed = asyncio.Event() + workers = [] + + class Client: + def __init__(self, name): + self.name = name + + async def aclose(self): + workers.append(asyncio.current_task()) + if self.name == "first": + raise OSError("first close failed") + started.set() + await release.wait() + completed.set() + + cluster = AsyncForgeCluster(ENDPOINTS, client_factory=lambda endpoint: Client(endpoint.name)) + task = asyncio.create_task(cluster.aclose()) + try: + await started.wait() + await asyncio.sleep(0) + await asyncio.sleep(0) + assert not task.done(), "Close returned while another client was still closing" + release.set() + with pytest.raises(OSError, match="first close failed"): + await task + assert completed.is_set() + finally: + release.set() + await asyncio.gather(task, *workers, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_cancelled_cluster_close_finishes_cleanup_before_returning() -> None: + started = asyncio.Event() + release = asyncio.Event() + completed = asyncio.Event() + + class Client: + async def aclose(self): + started.set() + await release.wait() + completed.set() + + cluster = AsyncForgeCluster(ENDPOINTS[:1], client_factory=lambda _: Client()) + task = asyncio.create_task(cluster.aclose()) + try: + await started.wait() + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert completed.is_set() + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_cancelled_health_child_cancels_and_drains_sibling_probe() -> None: + started = asyncio.Event() + stopped = asyncio.Event() + probes = [] + + class Client: + def __init__(self, name): + self.name = name + + async def health(self): + probes.append(asyncio.current_task()) + if self.name == "first": + await started.wait() + raise asyncio.CancelledError + started.set() + try: + await asyncio.Event().wait() + finally: + stopped.set() + + async def aclose(self): + pass + + async with AsyncForgeCluster(ENDPOINTS, client_factory=lambda endpoint: Client(endpoint.name)) as cluster: + try: + with pytest.raises(asyncio.CancelledError): + await cluster.health_all() + assert stopped.is_set(), "Health returned with an active orphaned probe" + assert all(task.done() for task in probes) + finally: + for task in probes: + task.cancel() + await asyncio.gather(*probes, return_exceptions=True) + + +@pytest.mark.parametrize("maximum", [True, False, 1.5, float("nan"), float("inf"), "2", 0, 100001]) +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_bulk_maximum_rejects_non_integer_before_consuming_work(maximum, asynchronous) -> None: + def work(): + raise AssertionError("Invalid maximum consumed input") + yield + + if asynchronous: + + async def run(): + async with AsyncForgeCluster(ENDPOINTS[:1]) as cluster: + with pytest.raises(ValueError, match="max_items"): + await cluster.bulk(work(), max_items=maximum) + + asyncio.run(run()) + else: + with ForgeCluster(ENDPOINTS[:1]) as cluster: + with pytest.raises(ValueError, match="max_items"): + cluster.bulk_create("project", "records", work(), max_items=maximum) + + +@pytest.mark.parametrize("workers", [True, False, 1.5, float("nan"), float("inf"), "2", 0, 65]) +def test_sync_bulk_rejects_invalid_workers_before_consuming_work(workers) -> None: + def work(): + raise AssertionError("Invalid worker count consumed input") + yield + + with ForgeCluster(ENDPOINTS[:1]) as cluster: + with pytest.raises(ValueError, match="max_workers"): + cluster.bulk_create("project", "records", work(), max_workers=workers) + + +@pytest.mark.parametrize("value", [True, False, 1.5, float("nan"), float("inf"), "2", 0, 101]) +@pytest.mark.parametrize("policy", ["weight", "failure_threshold"]) +def test_integer_cluster_policies_reject_invalid_types(value, policy) -> None: + with pytest.raises(ValueError, match=policy): + if policy == "weight": + ForgeEndpoint("first", "https://first.test", weight=value) + else: + CircuitBreakerPolicy(failure_threshold=value) + + +@pytest.mark.asyncio +async def test_repeated_caller_cancellation_does_not_interrupt_bulk_cleanup() -> None: + started = asyncio.Event() + cleanup_started = asyncio.Event() + release_cleanup = asyncio.Event() + completed = asyncio.Event() + + async def slow_mutation(): + started.set() + try: + await asyncio.Event().wait() + finally: + cleanup_started.set() + await release_cleanup.wait() + completed.set() + + async def cancelled_item(): + await started.wait() + raise asyncio.CancelledError + + async with AsyncForgeCluster(ENDPOINTS[:1]) as cluster: + pending = asyncio.create_task(cluster.bulk([cancelled_item, slow_mutation], concurrency=2)) + try: + await cleanup_started.wait() + pending.cancel() + await asyncio.sleep(0) + pending.cancel() + await asyncio.sleep(0) + assert not pending.done() + release_cleanup.set() + with pytest.raises(asyncio.CancelledError): + await pending + assert completed.is_set() + finally: + release_cleanup.set() + await asyncio.gather(pending, return_exceptions=True) diff --git a/tests/test_manifest_security.py b/tests/test_manifest_security.py new file mode 100644 index 0000000..0e98580 --- /dev/null +++ b/tests/test_manifest_security.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import importlib.util +import os +import subprocess +from pathlib import Path + +import pytest + +SPEC = importlib.util.spec_from_file_location("sdk_manifest_check", Path(__file__).parents[1] / "scripts/check_manifest.py") +manifest = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(manifest) + + +@pytest.fixture +def manifest_root(tmp_path, monkeypatch): + root = tmp_path / "release" + root.mkdir() + monkeypatch.setattr(manifest, "ROOT", root) + monkeypatch.setattr(manifest, "MANIFEST", root / "MANIFEST.sha256") + monkeypatch.setattr(manifest.subprocess, "check_output", lambda *_args, **_kwargs: root.as_posix()) + return root + + +@pytest.mark.parametrize( + "relative,checksum", + [ + ("../private.txt", "a" * 64), + ("/absolute/private.txt", "a" * 64), + ("C:/outside/private.txt", "a" * 64), + ("scripts\\tool.py", "a" * 64), + ("README.md:secret-stream", "a" * 64), + (".env", "a" * 64), + ("build/private.txt", "a" * 64), + ("README.md", "not-a-sha256"), + ("README.md", "A" * 64), + ], +) +def test_unsafe_expected_record_is_rejected_before_hashing(manifest_root, monkeypatch, relative, checksum): + manifest.MANIFEST.write_text(f"{checksum} {relative}\n", encoding="utf-8") + monkeypatch.setattr(manifest, "release_files", lambda: set()) + hashes = [] + monkeypatch.setattr(manifest, "digest", lambda path: hashes.append(path) or "a" * 64) + monkeypatch.setattr(manifest.Path, "is_file", lambda _path: True) + with pytest.raises(SystemExit) as error: + manifest.main([]) + assert hashes == [] + assert "manifest line" in str(error.value) + + +@pytest.mark.skipif(os.name != "nt", reason="actual Windows junction security regression") +@pytest.mark.parametrize("enumerator", ["git", "fallback"]) +def test_source_junction_rejected_before_descending(manifest_root, tmp_path, monkeypatch, enumerator): + outside = tmp_path / "outside" + outside.mkdir() + (outside / "private-key.txt").write_bytes(b"external secret fixture") + junction = manifest_root / "json_api_forge" + created = subprocess.run(["cmd", "/c", "mklink", "/J", str(junction), str(outside)], capture_output=True) + if created.returncode: + pytest.skip("Junction creation unavailable") + scanned = [] + original_scan = os.scandir + + def checked_scan(path): + scanned.append(Path(path)) + assert Path(path) != junction, "Fallback descended into an external junction" + return original_scan(path) + + try: + assert not junction.is_symlink(), "Fixture must reproduce the junction/symlink distinction" + if enumerator == "git": + outputs = [manifest_root.as_posix(), b"json_api_forge/private-key.txt\0"] + monkeypatch.setattr(manifest.subprocess, "check_output", lambda *_args, **_kwargs: outputs.pop(0)) + else: + monkeypatch.setattr(manifest.subprocess, "check_output", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError())) + monkeypatch.setattr(os, "scandir", checked_scan) + with pytest.raises(ValueError, match="junction"): + manifest.release_files() + assert junction not in scanned + finally: + junction.rmdir() # Removes only the verified fixture junction, never its target. + + +@pytest.mark.skipif(os.name != "nt", reason="actual Windows junction security regression") +def test_ignored_build_junction_is_not_scanned_or_rejected(manifest_root, tmp_path, monkeypatch): + outside = tmp_path / "artifacts" + outside.mkdir() + (outside / "private.txt").write_bytes(b"ignored fixture") + (manifest_root / "README.md").write_bytes(b"source") + junction = manifest_root / "build" + created = subprocess.run(["cmd", "/c", "mklink", "/J", str(junction), str(outside)], capture_output=True) + if created.returncode: + pytest.skip("Junction creation unavailable") + original_scan = os.scandir + + def checked_scan(path): + assert Path(path) != junction, "Ignored junction was descended into" + return original_scan(path) + + try: + monkeypatch.setattr(manifest.subprocess, "check_output", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError())) + monkeypatch.setattr(os, "scandir", checked_scan) + assert manifest.release_files() == {"README.md"} + finally: + junction.rmdir() + + +def test_roundtrip_and_relocated_git_checkout_use_safe_fallback(manifest_root, monkeypatch): + (manifest_root / "README.md").write_bytes(b"source") + monkeypatch.setattr(manifest.subprocess, "check_output", lambda *_args, **_kwargs: "other-checkout") + assert manifest.main(["--write"]) == 0 + assert manifest.main([]) == 0 + + +@pytest.mark.skipif(os.name != "nt", reason="actual Windows junction security regression") +def test_expected_path_under_unlisted_junction_is_never_hashed(manifest_root, tmp_path, monkeypatch): + outside = tmp_path / "outside" + outside.mkdir() + (outside / "private.txt").write_bytes(b"outside secret fixture") + junction = manifest_root / "json_api_forge" + created = subprocess.run(["cmd", "/c", "mklink", "/J", str(junction), str(outside)], capture_output=True) + if created.returncode: + pytest.skip("Junction creation unavailable") + try: + manifest.MANIFEST.write_text(f"{'a' * 64} json_api_forge/private.txt\n", encoding="utf-8") + monkeypatch.setattr(manifest, "release_files", lambda: set()) + hashes = [] + monkeypatch.setattr(manifest, "digest", lambda path: hashes.append(path) or "a" * 64) + with pytest.raises(ValueError, match="junction"): + manifest.main([]) + assert not hashes + finally: + junction.rmdir() diff --git a/tests/test_night_cluster.py b/tests/test_night_cluster.py new file mode 100644 index 0000000..e54c412 --- /dev/null +++ b/tests/test_night_cluster.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import asyncio +import io + +import httpx +import pytest + +from json_api_forge import AsyncForgeClient, AsyncForgeCluster, ForgeClient, ForgeCluster, ForgeEndpoint, ForgeHTTPError + + +@pytest.mark.asyncio +@pytest.mark.parametrize("concurrency", [True, False, 1.5, float("nan"), float("inf"), "2", 0, 257]) +async def test_bulk_rejects_invalid_concurrency_before_consuming_work(concurrency): + def work(): + raise AssertionError("Invalid concurrency must not consume or start work") + yield + + async with AsyncForgeCluster([ForgeEndpoint("only", "https://forge.test")]) as cluster: + with pytest.raises(ValueError, match="concurrency"): + await cluster.bulk(work(), concurrency=concurrency) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("method,idempotency_key", [("PUT", None), ("POST", "upload-once")]) +def test_cluster_never_replays_multipart_uploads(asynchronous, method, idempotency_key): + calls = [] + endpoints = [ForgeEndpoint("primary", "https://primary.test"), ForgeEndpoint("secondary", "https://secondary.test")] + + def factory(endpoint): + def handler(request): + calls.append(endpoint.name) + assert b"attachment content" in request.content + return httpx.Response(503 if endpoint.name == "primary" else 201, json={"endpoint": endpoint.name}) + + client = AsyncForgeClient if asynchronous else ForgeClient + return client(endpoint.base_url, transport=httpx.MockTransport(handler)) + + if asynchronous: + + async def run(): + async with AsyncForgeCluster(endpoints, client_factory=factory) as cluster: + with pytest.raises(ForgeHTTPError): + await cluster.request( + method, + "upload", + files={"upload": ("file.txt", io.BytesIO(b"attachment content"))}, + idempotency_key=idempotency_key, + ) + + asyncio.run(run()) + else: + with ForgeCluster(endpoints, client_factory=factory) as cluster: + with pytest.raises(ForgeHTTPError): + cluster.request( + method, + "upload", + files={"upload": ("file.txt", io.BytesIO(b"attachment content"))}, + idempotency_key=idempotency_key, + ) + assert calls == ["primary"] + + +@pytest.mark.asyncio +async def test_cancelled_bulk_item_stops_other_running_operations_before_returning(): + started = asyncio.Event() + stopped = asyncio.Event() + workers = [] + + async def slow_mutation(): + workers.append(asyncio.current_task()) + started.set() + try: + await asyncio.Event().wait() + finally: + stopped.set() + + async def cancel_item(): + await started.wait() + raise asyncio.CancelledError + + async with AsyncForgeCluster([ForgeEndpoint("only", "https://forge.test")]) as cluster: + try: + with pytest.raises(asyncio.CancelledError): + await cluster.bulk([cancel_item, slow_mutation], concurrency=2) + assert stopped.is_set(), "Bulk returned with an unowned operation still running" + assert all(task.done() for task in workers) + finally: + for task in workers: + task.cancel() + await asyncio.gather(*workers, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_bulk_caller_cancellation_does_not_start_queued_operations(): + started = asyncio.Event() + active = calls = 0 + + async def mutation(): + nonlocal active, calls + active += 1 + calls += 1 + if active == 2: + started.set() + try: + await asyncio.Event().wait() + finally: + active -= 1 + + async with AsyncForgeCluster([ForgeEndpoint("only", "https://forge.test")]) as cluster: + task = asyncio.create_task(cluster.bulk([mutation] * 20, concurrency=2)) + try: + await asyncio.wait_for(started.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert active == 0 and calls == 2 + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) diff --git a/tests/test_response_json.py b/tests/test_response_json.py new file mode 100644 index 0000000..9d54b7c --- /dev/null +++ b/tests/test_response_json.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import asyncio +import sys + +import httpx +import pytest + +from json_api_forge import AsyncForgeClient, ForgeClient, ForgeHTTPError + + +@pytest.mark.parametrize("content, expected", [(b"null", None), (b" null \n", None), (b"false", False), (b"0", 0), (b'""', "")]) +def test_sync_response_accepts_valid_json_scalar(content, expected) -> None: + with ForgeClient("https://forge.test", transport=httpx.MockTransport(lambda _: httpx.Response(200, content=content))) as client: + assert client.request("GET", "operation").data == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("content, expected", [(b"null", None), (b" null \n", None), (b"false", False), (b"0", 0), (b'""', "")]) +async def test_async_response_accepts_valid_json_scalar(content, expected) -> None: + async with AsyncForgeClient( + "https://forge.test", transport=httpx.MockTransport(lambda _: httpx.Response(200, content=content)) + ) as client: + assert (await client.request("GET", "operation")).data == expected + + +@pytest.mark.parametrize("content", [b"undefined", b"null suffix", b"\xff", b"{", b" \n"]) +def test_malformed_json_remains_rejected(content) -> None: + with ForgeClient("https://forge.test", transport=httpx.MockTransport(lambda _: httpx.Response(200, content=content))) as client: + with pytest.raises(ForgeHTTPError, match="non-JSON"): + client.request("GET", "operation") + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("content", [b"NaN", b"Infinity", b"-Infinity", b'{"items": [{"value": NaN}]}', b"1e999", b'{"value": -1e999}']) +def test_non_finite_response_numbers_are_rejected_and_observed(content, asynchronous) -> None: + events = [] + transport = httpx.MockTransport(lambda _: httpx.Response(200, content=content, headers={"X-Request-ID": "invalid-number"})) + + async def run(): + async with AsyncForgeClient("https://forge.test", transport=transport, observer=events.append) as client: + await client.request("GET", "operation") + + with pytest.raises(ForgeHTTPError, match="non-JSON") as caught: + if asynchronous: + asyncio.run(run()) + else: + with ForgeClient("https://forge.test", transport=transport, observer=events.append) as client: + client.request("GET", "operation") + assert caught.value.status_code == 200 + assert caught.value.request_id == "invalid-number" + assert len(events) == 1 and events[0].status_code == 200 + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("content, expected", [(b"1.25", 1.25), (b"1.7976931348623157e308", 1.7976931348623157e308)]) +def test_finite_response_numbers_preserve_supported_float_range(content, expected, asynchronous) -> None: + transport = httpx.MockTransport(lambda _: httpx.Response(200, content=content)) + + async def run(): + async with AsyncForgeClient("https://forge.test", transport=transport) as client: + return (await client.request("GET", "operation")).data + + if asynchronous: + assert asyncio.run(run()) == expected + else: + with ForgeClient("https://forge.test", transport=transport) as client: + assert client.request("GET", "operation").data == expected + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("status", [200, 503]) +def test_integer_decoder_policy_failures_keep_http_error_contract(status, asynchronous) -> None: + maximum = getattr(sys, "get_int_max_str_digits", lambda: 0)() + if not maximum: + pytest.skip("This Python interpreter does not limit integer string conversion") + events = [] + transport = httpx.MockTransport( + lambda _: httpx.Response(status, content=b"9" * (maximum + 1), headers={"X-Request-ID": "bounded-integer"}) + ) + + async def run(): + async with AsyncForgeClient("https://forge.test", transport=transport, observer=events.append) as client: + await client.request("GET", "operation") + + with pytest.raises(ForgeHTTPError) as caught: + if asynchronous: + asyncio.run(run()) + else: + with ForgeClient("https://forge.test", transport=transport, observer=events.append) as client: + client.request("GET", "operation") + assert caught.value.status_code == status + assert caught.value.request_id == "bounded-integer" + assert len(str(caught.value)) < 2300 + assert len(events) == 1 and events[0].status_code == status diff --git a/tests/test_v051_security.py b/tests/test_v051_security.py index de5f314..3bbf636 100644 --- a/tests/test_v051_security.py +++ b/tests/test_v051_security.py @@ -7,6 +7,7 @@ from json_api_forge import AsyncForgeClient, ForgeClient, ForgeHTTPError, RetryPolicy from json_api_forge.client import _base_url, _headers, _relative_path from json_api_forge.control_plane import _credential +from json_api_forge.errors import safe_error_text @pytest.mark.parametrize( @@ -71,6 +72,22 @@ def server(request): assert error.detail["password"] == "sensitive" +def test_long_url_credentials_are_redacted_before_diagnostic_truncation(): + secret = "credential-must-stay-private" * 200 + detail = f"Upstream https://account:{secret}@example.test/path returned an error" + rendered = safe_error_text(detail) + assert "credential-must-stay-private" not in rendered + assert "[redacted]@example.test/path" in rendered + assert len(rendered) <= 4096 + + +def test_error_text_stays_bounded_after_replacing_early_credentials(): + detail = "https://a@example.test/path " + "public detail " * 1000 + rendered = safe_error_text(detail) + assert "https://a@" not in rendered + assert len(rendered) <= 4096 + + @pytest.mark.parametrize("asynchronous", [False, True]) def test_encoded_responses_are_rejected_before_their_body_is_read(asynchronous): class UnreadableStream(httpx.SyncByteStream, httpx.AsyncByteStream):