From 8f3b91dc198b34b81f22114d5b1240c45d8c92dc Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Sun, 11 Oct 2026 23:29:25 +0400 Subject: [PATCH 01/10] Prevent multipart failover and drain cancelled async bulk work --- CHANGELOG.md | 6 ++ MANIFEST.sha256 | 7 ++- PYTHON_LIBRARY.md | 2 + json_api_forge/enterprise.py | 14 ++++- tests/test_night_cluster.py | 108 +++++++++++++++++++++++++++++++++++ 5 files changed, 133 insertions(+), 4 deletions(-) create mode 100644 tests/test_night_cluster.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a1d07d..ca73a02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased maintenance + +- 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. +- 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..5e52a73 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 +d05e715206217ceccf849a291e7a0be52ccc0aa415e54ea72690da692598fd3b CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -25,7 +25,7 @@ ce84a06b898c79cd6199e9eac425e40a3b8f7f8a346d6bbdd019580de56d7652 LICENSE-HISTOR b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe LICENSE-METADATA.json 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16 NOTICE.md c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae OWNERSHIP.md -7d1bcaa4e570d431553ec4abd420f5072204f7feba574cee43027e1eee3ef2de PYTHON_LIBRARY.md +8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c PYTHON_LIBRARY.md d14e5e605ba26ce24ec0a85c053e00577f6382be97d133c8b2b2662c43fb6375 README.md c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb RELEASE.md 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a RELEASE_ASSETS.md @@ -40,7 +40,7 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 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 +55b637cf358799989da2c075ab4dea34bb2d3f59efc234a002b1fb11fb8109bb json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705 json_api_forge/integrations/__init__.py 7dde121bcbe377560c684cbfc5875f0dc099e4d393eb494ce61da9731d562d54 json_api_forge/integrations/ddm.py @@ -53,6 +53,7 @@ d0cf58ccb32c816f0c85cd4a29a28b976343409647e4affcd84932acf98898c2 pyproject.toml 915a198346437156ef6388e6d3ac4f859478c26d9b48abf5a89e65a30d6e96ed scripts/check_critical_coverage.py 11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a scripts/check_manifest.py 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py +b87f02e106c60f3f1d903afa8a4ac50f30377f5551b9fafde5f637dce21ba797 tests/test_night_cluster.py 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py 9959ece37e0df7010bb2c87407e843b53a8d4b5ccfb9693e4fd6a4fb2e8943d0 tests/test_python_library.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/json_api_forge/enterprise.py b/json_api_forge/enterprise.py index 40a9b61..7726977 100644 --- a/json_api_forge/enterprise.py +++ b/json_api_forge/enterprise.py @@ -77,6 +77,9 @@ 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")) @@ -380,7 +383,16 @@ 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)] + try: + return list(await asyncio.gather(*tasks)) + except BaseException: + # A cancelled item makes gather return before its siblings finish. + # Own and drain them so mutations cannot continue after bulk exits. + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise async def health_all(self) -> dict[str, ForgeResponse[Any] | Exception]: async def health(endpoint: ForgeEndpoint): diff --git a/tests/test_night_cluster.py b/tests/test_night_cluster.py new file mode 100644 index 0000000..8b46e62 --- /dev/null +++ b/tests/test_night_cluster.py @@ -0,0 +1,108 @@ +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.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) From 9888a10f0833fe50ca0d6f2732afcf33100866f4 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Sun, 11 Oct 2026 23:43:47 +0400 Subject: [PATCH 02/10] Reject fractional async bulk concurrency before scheduling work --- CHANGELOG.md | 1 + MANIFEST.sha256 | 6 +++--- json_api_forge/enterprise.py | 2 +- tests/test_night_cluster.py | 12 ++++++++++++ 4 files changed, 17 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca73a02..d93a167 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - 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 diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 5e52a73..445a9e0 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -d05e715206217ceccf849a291e7a0be52ccc0aa415e54ea72690da692598fd3b CHANGELOG.md +bedae8a0d3fce5e6c66e9ab1b3d7d9aecdd2135b4c7a24f1d92123ad43c7a32e CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -40,7 +40,7 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py 1fdcaad555395a478241288fe757bac93ecb28d5da4a5491eaf64c3af5b439a6 json_api_forge/client.py 879c685c76c87e073bec9a3ff933a9e5103ddc948a172a0b8ea3eabeb2a9081d json_api_forge/control_plane.py -55b637cf358799989da2c075ab4dea34bb2d3f59efc234a002b1fb11fb8109bb json_api_forge/enterprise.py +e352fea0b4d70c59f71b5ba4a12d2fd0e40b8289fae9c7f2e440156546a35c2e json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705 json_api_forge/integrations/__init__.py 7dde121bcbe377560c684cbfc5875f0dc099e4d393eb494ce61da9731d562d54 json_api_forge/integrations/ddm.py @@ -53,7 +53,7 @@ d0cf58ccb32c816f0c85cd4a29a28b976343409647e4affcd84932acf98898c2 pyproject.toml 915a198346437156ef6388e6d3ac4f859478c26d9b48abf5a89e65a30d6e96ed scripts/check_critical_coverage.py 11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a scripts/check_manifest.py 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py -b87f02e106c60f3f1d903afa8a4ac50f30377f5551b9fafde5f637dce21ba797 tests/test_night_cluster.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 diff --git a/json_api_forge/enterprise.py b/json_api_forge/enterprise.py index 7726977..2cf61e5 100644 --- a/json_api_forge/enterprise.py +++ b/json_api_forge/enterprise.py @@ -371,7 +371,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) diff --git a/tests/test_night_cluster.py b/tests/test_night_cluster.py index 8b46e62..e54c412 100644 --- a/tests/test_night_cluster.py +++ b/tests/test_night_cluster.py @@ -9,6 +9,18 @@ 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): From 80357d15bcb8ea01e21d67699bd4ab1aa1f372a8 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Mon, 12 Oct 2026 22:12:13 +0400 Subject: [PATCH 03/10] Keep async attachment disk IO responsive and cancellation safe --- CHANGELOG.md | 1 + MANIFEST.sha256 | 12 +- README.md | 2 + json_api_forge/_io.py | 28 ++++ json_api_forge/client.py | 35 ++++- json_api_forge/control_plane.py | 19 ++- scripts/check_critical_coverage.py | 1 + tests/test_async_attachment_io.py | 240 +++++++++++++++++++++++++++++ 8 files changed, 328 insertions(+), 10 deletions(-) create mode 100644 json_api_forge/_io.py create mode 100644 tests/test_async_attachment_io.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d93a167..dfeb9dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased maintenance +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 445a9e0..34fec44 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -bedae8a0d3fce5e6c66e9ab1b3d7d9aecdd2135b4c7a24f1d92123ad43c7a32e CHANGELOG.md +93b46e51536e60ec793568aff4b0e22c00938535b64bf14f0a585906eb36e69d CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -26,7 +26,7 @@ b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe LICENSE-METADA 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16 NOTICE.md c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae OWNERSHIP.md 8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c PYTHON_LIBRARY.md -d14e5e605ba26ce24ec0a85c053e00577f6382be97d133c8b2b2662c43fb6375 README.md +e52cb318f99479866cc8676c861bf95df649eb311a23130d5a3ec20f1f455330 README.md c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb RELEASE.md 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a RELEASE_ASSETS.md 858ea89e51f4217a45b042979c718d17f9c2481dc1494c0ba4f49e4ff31caa96 RELEASE_CHECKLIST.md @@ -37,9 +37,10 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 26e0ab83aa1ff6bd0ae3882787a92663aa19db128e1241ffc99375fb9ca0c121 VERSION 8f004523844f4b429a9ed8e15e5d46e2957a4c9854a772520267c34be6cf7998 contract-tests/test_against_main.py 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py +eeb90b1269857ce09cd81a00c3c94e61e6d0918d3edcc587556e3b36432ce539 json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py -1fdcaad555395a478241288fe757bac93ecb28d5da4a5491eaf64c3af5b439a6 json_api_forge/client.py -879c685c76c87e073bec9a3ff933a9e5103ddc948a172a0b8ea3eabeb2a9081d json_api_forge/control_plane.py +96053601fa56469be0004ef6697637fb72b29d0a78ebc239205befec3378746f json_api_forge/client.py +1acd472715c2877be11a2b19897b6b5042bc1af0d577475c6fb949681b10dcfb json_api_forge/control_plane.py e352fea0b4d70c59f71b5ba4a12d2fd0e40b8289fae9c7f2e440156546a35c2e json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705 json_api_forge/integrations/__init__.py @@ -50,9 +51,10 @@ 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 +dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64 scripts/check_critical_coverage.py 11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a scripts/check_manifest.py 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py +50830855c94d3e55c4bd74d6df786d95875929fbe72653fd85ff98d8c2250d3b tests/test_async_attachment_io.py 61109979cdb492ac7976947d804b3e717871fa367346f3cdfa911382b25c59a3 tests/test_night_cluster.py 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py diff --git a/README.md b/README.md index 1f587d2..88b123f 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,8 @@ 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. + ## Cluster and optional integrations `ForgeCluster` and `AsyncForgeCluster` provide deterministic routing, circuit breakers, bounded bulk work and idempotency-gated failover across multiple Forge endpoints. diff --git a/json_api_forge/_io.py b/json_api_forge/_io.py new file mode 100644 index 0000000..28d92cc --- /dev/null +++ b/json_api_forge/_io.py @@ -0,0 +1,28 @@ +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))) + try: + return await asyncio.shield(worker) + except asyncio.CancelledError: + # Thread cancellation cannot stop an OS operation. Drain it before an + # enclosing context closes its handle, even if cancellation repeats. + 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..3141e3f 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 from .errors import ForgeHTTPError, ForgeResponseTooLarge, ForgeTransportError, safe_error_text from .models import ForgeResponse, JsonObject, RequestAttempt from .options import RetryPolicy @@ -234,6 +236,36 @@ 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 + + +@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: + 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) + try: + yield response + finally: + await response.aclose() + + class ForgeClient: """Synchronous, bounded client for one JSON API Forge server.""" @@ -509,7 +541,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, diff --git a/json_api_forge/control_plane.py b/json_api_forge/control_plane.py index 3fae545..847b323 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: @@ -752,7 +763,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 +784,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/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/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 From 2f02d50d064db1ff90f43bc401455ebe5f39c119 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Mon, 12 Oct 2026 22:14:30 +0400 Subject: [PATCH 04/10] Discard stale async control plane session responses --- CHANGELOG.md | 1 + MANIFEST.sha256 | 7 +- README.md | 2 + json_api_forge/control_plane.py | 22 ++++-- tests/test_async_session_races.py | 124 ++++++++++++++++++++++++++++++ 5 files changed, 147 insertions(+), 9 deletions(-) create mode 100644 tests/test_async_session_races.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dfeb9dd..4e00d35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased maintenance +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 34fec44..c52164c 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -93b46e51536e60ec793568aff4b0e22c00938535b64bf14f0a585906eb36e69d CHANGELOG.md +58e00430da57dc19fb1ec203e68f60df637fdc60f93586742d6eff3ba380d216 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -26,7 +26,7 @@ b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe LICENSE-METADA 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16 NOTICE.md c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae OWNERSHIP.md 8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c PYTHON_LIBRARY.md -e52cb318f99479866cc8676c861bf95df649eb311a23130d5a3ec20f1f455330 README.md +b42862f21232440451fc0810baadf265810f2e2bb03d923955b0db9faa3d0d17 README.md c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb RELEASE.md 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a RELEASE_ASSETS.md 858ea89e51f4217a45b042979c718d17f9c2481dc1494c0ba4f49e4ff31caa96 RELEASE_CHECKLIST.md @@ -40,7 +40,7 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO eeb90b1269857ce09cd81a00c3c94e61e6d0918d3edcc587556e3b36432ce539 json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py 96053601fa56469be0004ef6697637fb72b29d0a78ebc239205befec3378746f json_api_forge/client.py -1acd472715c2877be11a2b19897b6b5042bc1af0d577475c6fb949681b10dcfb json_api_forge/control_plane.py +5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c json_api_forge/control_plane.py e352fea0b4d70c59f71b5ba4a12d2fd0e40b8289fae9c7f2e440156546a35c2e json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705 json_api_forge/integrations/__init__.py @@ -55,6 +55,7 @@ dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64 scripts/check_ 11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a scripts/check_manifest.py 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py 50830855c94d3e55c4bd74d6df786d95875929fbe72653fd85ff98d8c2250d3b tests/test_async_attachment_io.py +3c00da7a012803b247ea5dc5820bb8c6448e9798f60e38abb09ff801ce01c546 tests/test_async_session_races.py 61109979cdb492ac7976947d804b3e717871fa367346f3cdfa911382b25c59a3 tests/test_night_cluster.py 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py diff --git a/README.md b/README.md index 88b123f..0e0a0ad 100644 --- a/README.md +++ b/README.md @@ -84,6 +84,8 @@ Security defaults include HTTPS-only remote URLs, no ambient proxy inheritance, 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. diff --git a/json_api_forge/control_plane.py b/json_api_forge/control_plane.py index 847b323..860d68f 100644 --- a/json_api_forge/control_plane.py +++ b/json_api_forge/control_plane.py @@ -202,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: @@ -213,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() @@ -233,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) @@ -564,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, @@ -576,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 @@ -591,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", @@ -599,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, @@ -619,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"))) 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 From 43f6a75b19451d50ec1167cdf37f44a4270e0f84 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Mon, 12 Oct 2026 22:15:48 +0400 Subject: [PATCH 05/10] Accept valid null JSON responses without masking parse errors --- CHANGELOG.md | 1 + MANIFEST.sha256 | 5 +++-- json_api_forge/client.py | 4 +++- tests/test_response_json.py | 28 ++++++++++++++++++++++++++++ 4 files changed, 35 insertions(+), 3 deletions(-) create mode 100644 tests/test_response_json.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e00d35..83c59bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased maintenance +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index c52164c..dfe8d6a 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -58e00430da57dc19fb1ec203e68f60df637fdc60f93586742d6eff3ba380d216 CHANGELOG.md +b4ae10e8ebcbad3ea145ea8b72b7b6df9f72c5a7d510ca36f23fb258112dc6a9 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -39,7 +39,7 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py eeb90b1269857ce09cd81a00c3c94e61e6d0918d3edcc587556e3b36432ce539 json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py -96053601fa56469be0004ef6697637fb72b29d0a78ebc239205befec3378746f json_api_forge/client.py +5ddae24026be2ff2495b7e1c3921f6cf523ce389a47a7d2f75a93ebfb14deece json_api_forge/client.py 5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c json_api_forge/control_plane.py e352fea0b4d70c59f71b5ba4a12d2fd0e40b8289fae9c7f2e440156546a35c2e json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py @@ -60,5 +60,6 @@ dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64 scripts/check_ 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py 9959ece37e0df7010bb2c87407e843b53a8d4b5ccfb9693e4fd6a4fb2e8943d0 tests/test_python_library.py +056272674921c9f628b66171b79ff458c6f068ee04033521618097f5e09e96e9 tests/test_response_json.py fd19f7b746b195fddbcd2997b787b68d625f3fe2b0cb9899bcb44570ff563766 tests/test_v051_security.py fa33f2147dd8e704684e752e5f4d0caec848f9c8ffb176a2dc202d8cf1513e4d tests/test_v052_pagination.py diff --git a/json_api_forge/client.py b/json_api_forge/client.py index 3141e3f..23275a6 100644 --- a/json_api_forge/client.py +++ b/json_api_forge/client.py @@ -153,10 +153,12 @@ def _headers( 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 = None + valid_json = False if response.is_redirect: raise ForgeHTTPError( response.status_code, @@ -177,7 +179,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( diff --git a/tests/test_response_json.py b/tests/test_response_json.py new file mode 100644 index 0000000..530be2f --- /dev/null +++ b/tests/test_response_json.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +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") From 662ed6a3b300a1db76afd771d075f3e89832b523 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Mon, 12 Oct 2026 22:23:14 +0400 Subject: [PATCH 06/10] Drain cluster lifecycle tasks and reject invalid bulk policies --- CHANGELOG.md | 2 + MANIFEST.sha256 | 9 +- README.md | 2 + json_api_forge/_io.py | 9 +- json_api_forge/enterprise.py | 50 +++++--- tests/test_cluster_lifecycle.py | 208 ++++++++++++++++++++++++++++++++ 6 files changed, 257 insertions(+), 23 deletions(-) create mode 100644 tests/test_cluster_lifecycle.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 83c59bf..c85034b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased maintenance +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index dfe8d6a..084671b 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -b4ae10e8ebcbad3ea145ea8b72b7b6df9f72c5a7d510ca36f23fb258112dc6a9 CHANGELOG.md +2673bd1384c77a17ea1f375f3aad5acfa9d5c034ec4502946488cce23769baa1 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -26,7 +26,7 @@ b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe LICENSE-METADA 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16 NOTICE.md c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae OWNERSHIP.md 8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c PYTHON_LIBRARY.md -b42862f21232440451fc0810baadf265810f2e2bb03d923955b0db9faa3d0d17 README.md +1c070e3089ced72b9ce2d0cc347783918b641df9de34b0c834c5d34e78f81182 README.md c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb RELEASE.md 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a RELEASE_ASSETS.md 858ea89e51f4217a45b042979c718d17f9c2481dc1494c0ba4f49e4ff31caa96 RELEASE_CHECKLIST.md @@ -37,11 +37,11 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 26e0ab83aa1ff6bd0ae3882787a92663aa19db128e1241ffc99375fb9ca0c121 VERSION 8f004523844f4b429a9ed8e15e5d46e2957a4c9854a772520267c34be6cf7998 contract-tests/test_against_main.py 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py -eeb90b1269857ce09cd81a00c3c94e61e6d0918d3edcc587556e3b36432ce539 json_api_forge/_io.py +5c1d12bf6337372eac54c08109583c439db3952bafb0008afd0b9c1402cfa2cd json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py 5ddae24026be2ff2495b7e1c3921f6cf523ce389a47a7d2f75a93ebfb14deece json_api_forge/client.py 5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c json_api_forge/control_plane.py -e352fea0b4d70c59f71b5ba4a12d2fd0e40b8289fae9c7f2e440156546a35c2e json_api_forge/enterprise.py +c0f60c048a1512d9fe820055ce3606551c951b8a9a909d0ff999cf3f8829abe2 json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705 json_api_forge/integrations/__init__.py 7dde121bcbe377560c684cbfc5875f0dc099e4d393eb494ce61da9731d562d54 json_api_forge/integrations/ddm.py @@ -56,6 +56,7 @@ dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64 scripts/check_ 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py 50830855c94d3e55c4bd74d6df786d95875929fbe72653fd85ff98d8c2250d3b tests/test_async_attachment_io.py 3c00da7a012803b247ea5dc5820bb8c6448e9798f60e38abb09ff801ce01c546 tests/test_async_session_races.py +cc6694028ea0148fc4e5dc7877740c1133b21afee030390c9e039b68aa34ce79 tests/test_cluster_lifecycle.py 61109979cdb492ac7976947d804b3e717871fa367346f3cdfa911382b25c59a3 tests/test_night_cluster.py 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py diff --git a/README.md b/README.md index 0e0a0ad..903e50c 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,8 @@ The async control-plane client discards an authentication response with `ForgeSe `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 index 28d92cc..516f805 100644 --- a/json_api_forge/_io.py +++ b/json_api_forge/_io.py @@ -11,11 +11,16 @@ 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: - # Thread cancellation cannot stop an OS operation. Drain it before an - # enclosing context closes its handle, even if cancellation repeats. + # 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) diff --git a/json_api_forge/enterprise.py b/json_api_forge/enterprise.py index 2cf61e5..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") @@ -84,7 +85,7 @@ def _failover_permitted(method: str, options: Mapping[str, Any]) -> bool: 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: @@ -92,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.""" @@ -214,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]: @@ -249,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 @@ -384,15 +405,7 @@ async def run(index: int, operation: Callable[[], Awaitable[T]]) -> BulkResult[T return BulkResult(index=index, error=exc) tasks = [asyncio.create_task(run(index, operation)) for index, operation in enumerate(bounded_operations)] - try: - return list(await asyncio.gather(*tasks)) - except BaseException: - # A cancelled item makes gather return before its siblings finish. - # Own and drain them so mutations cannot continue after bulk exits. - for task in tasks: - task.cancel() - await asyncio.gather(*tasks, return_exceptions=True) - raise + return await _gather_owned(tasks) async def health_all(self) -> dict[str, ForgeResponse[Any] | Exception]: async def health(endpoint: ForgeEndpoint): @@ -401,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/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) From 4bb56bdce3cbabf06c045833f981cd746791f5b0 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Mon, 12 Oct 2026 22:33:10 +0400 Subject: [PATCH 07/10] Reject unsafe manifest paths and NTFS junction traversal --- CHANGELOG.md | 1 + MANIFEST.sha256 | 5 +- scripts/check_manifest.py | 67 ++++++++++++++-- tests/test_manifest_security.py | 133 ++++++++++++++++++++++++++++++++ 4 files changed, 196 insertions(+), 10 deletions(-) create mode 100644 tests/test_manifest_security.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c85034b..e174cd1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased maintenance +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 084671b..403b52e 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -2673bd1384c77a17ea1f375f3aad5acfa9d5c034ec4502946488cce23769baa1 CHANGELOG.md +47f85996f62af29fb2891714c43dd1c82b74980be91298cf57cc055e27fa7c98 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -52,11 +52,12 @@ c2d0db468c7e41c65791983f512b484cb850c5f65ae725069b7aaab6cc78b3f8 json_api_forge 01ba4719c80b6fe911b091a7c05124b64eeece964e09c058ef8f9805daca546b json_api_forge/py.typed d0cf58ccb32c816f0c85cd4a29a28b976343409647e4affcd84932acf98898c2 pyproject.toml dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64 scripts/check_critical_coverage.py -11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a scripts/check_manifest.py +fc719b85b872e730b347f4674e2aaa3a3bc8bff3dfe361b41e46837dee709062 scripts/check_manifest.py 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c tests/__init__.py 50830855c94d3e55c4bd74d6df786d95875929fbe72653fd85ff98d8c2250d3b tests/test_async_attachment_io.py 3c00da7a012803b247ea5dc5820bb8c6448e9798f60e38abb09ff801ce01c546 tests/test_async_session_races.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 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_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() From 2b23c2f59a49a6579e7add87facbb9c1dfbc68b5 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Thu, 15 Oct 2026 22:54:09 +0400 Subject: [PATCH 08/10] Drain single-client HTTP cleanup under repeated cancellation --- CHANGELOG.md | 1 + MANIFEST.sha256 | 5 +- json_api_forge/client.py | 23 +++++- tests/test_async_transport_cleanup.py | 113 ++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 4 deletions(-) create mode 100644 tests/test_async_transport_cleanup.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e174cd1..ba65389 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased maintenance +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 403b52e..7e0f8a3 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -47f85996f62af29fb2891714c43dd1c82b74980be91298cf57cc055e27fa7c98 CHANGELOG.md +b8a1e2508a42967cdcafe5a78632388e70cda2d5dabcd8e91f37a3d1c723e994 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -39,7 +39,7 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py 5c1d12bf6337372eac54c08109583c439db3952bafb0008afd0b9c1402cfa2cd json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py -5ddae24026be2ff2495b7e1c3921f6cf523ce389a47a7d2f75a93ebfb14deece json_api_forge/client.py +175897daa7446c03fb948187afa962f8af5593c802bf641d3c6c25b529363b6a json_api_forge/client.py 5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c json_api_forge/control_plane.py c0f60c048a1512d9fe820055ce3606551c951b8a9a909d0ff999cf3f8829abe2 json_api_forge/enterprise.py 0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.py @@ -56,6 +56,7 @@ fc719b85b872e730b347f4674e2aaa3a3bc8bff3dfe361b41e46837dee709062 scripts/check_ 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 diff --git a/json_api_forge/client.py b/json_api_forge/client.py index 23275a6..b8124d0 100644 --- a/json_api_forge/client.py +++ b/json_api_forge/client.py @@ -15,7 +15,7 @@ import httpx -from ._io import run_io +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 @@ -252,16 +252,35 @@ async def __aiter__(self) -> AsyncIterator[bytes]: 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: @@ -677,7 +696,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/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) From e04b22979625ca7a90c3f38c9f81e1f3fdcb1300 Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Sat, 17 Oct 2026 20:08:10 +0400 Subject: [PATCH 09/10] Handle malformed JSON numbers and redact complete URL credentials --- CHANGELOG.md | 2 ++ MANIFEST.sha256 | 10 ++++---- json_api_forge/client.py | 10 ++++++-- json_api_forge/errors.py | 6 +++-- tests/test_response_json.py | 51 +++++++++++++++++++++++++++++++++++++ tests/test_v051_security.py | 17 +++++++++++++ 6 files changed, 87 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ba65389..e3fa3b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased maintenance +- Reject non-standard JSON number constants and 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 7e0f8a3..4ad50c0 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f .gitignore 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398 .python-version 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55 AI_USAGE_POLICY.md 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md -b8a1e2508a42967cdcafe5a78632388e70cda2d5dabcd8e91f37a3d1c723e994 CHANGELOG.md +29c63f027994a62ac0c232cad97a53c230a8b70063889c2528d8bb4bf70d6656 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -39,10 +39,10 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py 5c1d12bf6337372eac54c08109583c439db3952bafb0008afd0b9c1402cfa2cd json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py -175897daa7446c03fb948187afa962f8af5593c802bf641d3c6c25b529363b6a json_api_forge/client.py +dbc3e1f40a85a8e71e5fd26e10d117cff9df46c2bed3a3c7ad02edd43ff10ea5 json_api_forge/client.py 5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c json_api_forge/control_plane.py c0f60c048a1512d9fe820055ce3606551c951b8a9a909d0ff999cf3f8829abe2 json_api_forge/enterprise.py -0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7 json_api_forge/errors.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 @@ -63,6 +63,6 @@ ff662164581e2f72607604cb9f09d5c56b88440d171aa0242716ff1d693b36b4 tests/test_man 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py 9959ece37e0df7010bb2c87407e843b53a8d4b5ccfb9693e4fd6a4fb2e8943d0 tests/test_python_library.py -056272674921c9f628b66171b79ff458c6f068ee04033521618097f5e09e96e9 tests/test_response_json.py -fd19f7b746b195fddbcd2997b787b68d625f3fe2b0cb9899bcb44570ff563766 tests/test_v051_security.py +3aeffb6b9c1b20ffdb17a513d0f3a50023929077a72ec42b3df43ba443966a8d tests/test_response_json.py +1a44d0909278cc2750a80f757631dd2f51d1dc6a3ea206f5ac85f5210f9de7a5 tests/test_v051_security.py fa33f2147dd8e704684e752e5f4d0caec848f9c8ffb176a2dc202d8cf1513e4d tests/test_v052_pagination.py diff --git a/json_api_forge/client.py b/json_api_forge/client.py index b8124d0..f8eb4b8 100644 --- a/json_api_forge/client.py +++ b/json_api_forge/client.py @@ -151,12 +151,18 @@ def _headers( return values +def _reject_json_constant(value: str) -> Any: + raise ValueError(f"Invalid JSON number constant: {value}") + + 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) 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: 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/tests/test_response_json.py b/tests/test_response_json.py index 530be2f..b9c9431 100644 --- a/tests/test_response_json.py +++ b/tests/test_response_json.py @@ -1,5 +1,8 @@ from __future__ import annotations +import asyncio +import sys + import httpx import pytest @@ -26,3 +29,51 @@ 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}]}']) +def test_non_json_number_constants_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("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): From bed95a2de7c4b029a6bf21347d4111c0b733e39a Mon Sep 17 00:00:00 2001 From: Cavanshirpro Date: Sat, 17 Oct 2026 20:10:41 +0400 Subject: [PATCH 10/10] Reject overflowing response floats while preserving finite numbers --- CHANGELOG.md | 2 +- MANIFEST.sha256 | 8 ++++---- README.md | 2 ++ json_api_forge/client.py | 9 ++++++++- tests/test_response_json.py | 20 ++++++++++++++++++-- 5 files changed, 33 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e3fa3b7..0ea1629 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## Unreleased maintenance -- Reject non-standard JSON number constants and report bounded integer decoder failures as observable HTTP errors instead of leaking raw parser exceptions. +- 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. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 4ad50c0..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 -29c63f027994a62ac0c232cad97a53c230a8b70063889c2528d8bb4bf70d6656 CHANGELOG.md +6b05c59b9fc679a94d250abe4f78748ef6b49b218a446834d6614019379a7069 CHANGELOG.md f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3 CONTRIBUTING.md @@ -26,7 +26,7 @@ b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe LICENSE-METADA 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16 NOTICE.md c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae OWNERSHIP.md 8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c PYTHON_LIBRARY.md -1c070e3089ced72b9ce2d0cc347783918b641df9de34b0c834c5d34e78f81182 README.md +e29158a6782790a9aa7c3c00566e45387e9ce5e673c9482a9b898661dff08d28 README.md c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb RELEASE.md 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a RELEASE_ASSETS.md 858ea89e51f4217a45b042979c718d17f9c2481dc1494c0ba4f49e4ff31caa96 RELEASE_CHECKLIST.md @@ -39,7 +39,7 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d THIRD_PARTY_NO 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164 json_api_forge/__init__.py 5c1d12bf6337372eac54c08109583c439db3952bafb0008afd0b9c1402cfa2cd json_api_forge/_io.py 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c json_api_forge/_version.py -dbc3e1f40a85a8e71e5fd26e10d117cff9df46c2bed3a3c7ad02edd43ff10ea5 json_api_forge/client.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 @@ -63,6 +63,6 @@ ff662164581e2f72607604cb9f09d5c56b88440d171aa0242716ff1d693b36b4 tests/test_man 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d tests/test_python_control_plane.py 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9 tests/test_python_enterprise.py 9959ece37e0df7010bb2c87407e843b53a8d4b5ccfb9693e4fd6a4fb2e8943d0 tests/test_python_library.py -3aeffb6b9c1b20ffdb17a513d0f3a50023929077a72ec42b3df43ba443966a8d tests/test_response_json.py +40db1d05ac5607c4bae91b750fb6c431791fbfeb27c090684b46b4908eea345a tests/test_response_json.py 1a44d0909278cc2750a80f757631dd2f51d1dc6a3ea206f5ac85f5210f9de7a5 tests/test_v051_security.py fa33f2147dd8e704684e752e5f4d0caec848f9c8ffb176a2dc202d8cf1513e4d tests/test_v052_pagination.py diff --git a/README.md b/README.md index 903e50c..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 diff --git a/json_api_forge/client.py b/json_api_forge/client.py index f8eb4b8..1bec902 100644 --- a/json_api_forge/client.py +++ b/json_api_forge/client.py @@ -155,11 +155,18 @@ 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, parse_constant=_reject_json_constant) if content else None + 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. diff --git a/tests/test_response_json.py b/tests/test_response_json.py index b9c9431..9d54b7c 100644 --- a/tests/test_response_json.py +++ b/tests/test_response_json.py @@ -32,8 +32,8 @@ def test_malformed_json_remains_rejected(content) -> None: @pytest.mark.parametrize("asynchronous", [False, True]) -@pytest.mark.parametrize("content", [b"NaN", b"Infinity", b"-Infinity", b'{"items": [{"value": NaN}]}']) -def test_non_json_number_constants_are_rejected_and_observed(content, asynchronous) -> None: +@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"})) @@ -52,6 +52,22 @@ async def run(): 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: