From 3d2913aac6d0f82c3b3bb8a91861205f0a8706e9 Mon Sep 17 00:00:00 2001 From: Tushar Pagar <240662211+tushardev-365@users.noreply.github.com> Date: Sat, 15 Aug 2026 05:17:43 +0000 Subject: [PATCH] fix(storage): close the upload file handle when the request fails When a caller passes a path rather than an open file, _upload_or_update and upload_to_signed_url open the handle themselves and hand it to _request inside the files dict. The close lived after the except HTTPStatusError block, so it only ran when the request succeeded: any non-2xx raised StorageApiError straight past it and the handle was left open, with no way for the caller to reach it. It leaks whenever the exception is retained (logging with exc_info, collecting failures in a batch upload loop, an error reporter), because the traceback keeps the frame holding the reader alive. Python also emits ResourceWarning: unclosed file on that path. Move the close into a finally so it runs on both paths. The condition is unchanged, so only handles storage3 opened itself are closed and a caller-supplied stream is still left alone. --- src/storage/src/storage3/_async/file_api.py | 57 +++++++++++---------- src/storage/src/storage3/_sync/file_api.py | 57 +++++++++++---------- src/storage/tests/_async/test_client.py | 32 ++++++++++++ src/storage/tests/_sync/test_client.py | 34 +++++++++++- 4 files changed, 127 insertions(+), 53 deletions(-) diff --git a/src/storage/src/storage3/_async/file_api.py b/src/storage/src/storage3/_async/file_api.py index 474486f1..193aeb71 100644 --- a/src/storage/src/storage3/_async/file_api.py +++ b/src/storage/src/storage3/_async/file_api.py @@ -65,33 +65,38 @@ async def _request( **kwargs: Any, ) -> Response: try: - url_path = self._base_url.joinpath(*path).with_query(query_params) - headers = headers or dict() - headers.update(self._headers) - response = await self._client.request( - method, - str(url_path), - headers=headers, - json=json, - files=files, - **kwargs, - ) - response.raise_for_status() - except HTTPStatusError as exc: try: - resp = exc.response.json() - raise StorageApiError( - resp["message"], resp["error"], resp["statusCode"] - ) from exc - except KeyError as err: - message = f"Unable to parse error message: {resp.text}" - raise StorageApiError(message, "InternalError", 400) from err - - # close the resource before returning the response - if files and "file" in files and isinstance(files["file"][1], BufferedReader): - files["file"][1].close() - - return response + url_path = self._base_url.joinpath(*path).with_query(query_params) + headers = headers or dict() + headers.update(self._headers) + response = await self._client.request( + method, + str(url_path), + headers=headers, + json=json, + files=files, + **kwargs, + ) + response.raise_for_status() + except HTTPStatusError as exc: + try: + resp = exc.response.json() + raise StorageApiError( + resp["message"], resp["error"], resp["statusCode"] + ) from exc + except KeyError as err: + message = f"Unable to parse error message: {resp.text}" + raise StorageApiError(message, "InternalError", 400) from err + + return response + finally: + # close the resource on the error path too, not just on success + if ( + files + and "file" in files + and isinstance(files["file"][1], BufferedReader) + ): + files["file"][1].close() async def create_signed_upload_url( self, diff --git a/src/storage/src/storage3/_sync/file_api.py b/src/storage/src/storage3/_sync/file_api.py index 1bc22a09..f3b8eea7 100644 --- a/src/storage/src/storage3/_sync/file_api.py +++ b/src/storage/src/storage3/_sync/file_api.py @@ -65,33 +65,38 @@ def _request( **kwargs: Any, ) -> Response: try: - url_path = self._base_url.joinpath(*path).with_query(query_params) - headers = headers or dict() - headers.update(self._headers) - response = self._client.request( - method, - str(url_path), - headers=headers, - json=json, - files=files, - **kwargs, - ) - response.raise_for_status() - except HTTPStatusError as exc: try: - resp = exc.response.json() - raise StorageApiError( - resp["message"], resp["error"], resp["statusCode"] - ) from exc - except KeyError as err: - message = f"Unable to parse error message: {resp.text}" - raise StorageApiError(message, "InternalError", 400) from err - - # close the resource before returning the response - if files and "file" in files and isinstance(files["file"][1], BufferedReader): - files["file"][1].close() - - return response + url_path = self._base_url.joinpath(*path).with_query(query_params) + headers = headers or dict() + headers.update(self._headers) + response = self._client.request( + method, + str(url_path), + headers=headers, + json=json, + files=files, + **kwargs, + ) + response.raise_for_status() + except HTTPStatusError as exc: + try: + resp = exc.response.json() + raise StorageApiError( + resp["message"], resp["error"], resp["statusCode"] + ) from exc + except KeyError as err: + message = f"Unable to parse error message: {resp.text}" + raise StorageApiError(message, "InternalError", 400) from err + + return response + finally: + # close the resource on the error path too, not just on success + if ( + files + and "file" in files + and isinstance(files["file"][1], BufferedReader) + ): + files["file"][1].close() def create_signed_upload_url( self, diff --git a/src/storage/tests/_async/test_client.py b/src/storage/tests/_async/test_client.py index 8cc9c3f2..100fe4b8 100644 --- a/src/storage/tests/_async/test_client.py +++ b/src/storage/tests/_async/test_client.py @@ -8,10 +8,12 @@ import pytest from httpx import AsyncClient as HttpxClient +from httpx import Headers from httpx import HTTPStatusError, Response from storage3 import AsyncStorageClient from storage3.exceptions import StorageApiError from storage3.utils import StorageException +from yarl import URL from .. import AsyncBucketProxy from ..utils import AsyncFinalizerFactory @@ -827,3 +829,33 @@ async def test_client_list_v2_paginated( assert all(f.name.startswith(file.bucket_path) for f in result.objects) pages += 1 assert pages == 4 + + +async def test_upload_closes_file_handle_on_error(tmp_path: Path) -> None: + """A failed upload must not leak the handle storage3 opened itself.""" + source = tmp_path / "leak.txt" + source.write_bytes(b"payload") + + error_response = Mock(spec=Response) + error_response.json.return_value = { + "message": "Duplicate", + "error": "Duplicate", + "statusCode": 409, + } + + captured: dict = {} + + async def fail(*args: object, **kwargs: object) -> Response: + captured["files"] = kwargs.get("files") + raise HTTPStatusError("409", request=Mock(), response=error_response) + + client = AsyncMock() + client.headers = Headers() + client.request = fail + + proxy = AsyncBucketProxy("bucket", URL("http://localhost"), Headers(), client) + + with pytest.raises(StorageApiError): + await proxy.upload("leak.txt", str(source)) + + assert captured["files"]["file"][1].closed diff --git a/src/storage/tests/_sync/test_client.py b/src/storage/tests/_sync/test_client.py index 3a50149f..4e17f3b9 100644 --- a/src/storage/tests/_sync/test_client.py +++ b/src/storage/tests/_sync/test_client.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Generator +from collections.abc import Generator, Generator from dataclasses import dataclass from typing import TYPE_CHECKING from unittest.mock import Mock, patch @@ -8,10 +8,12 @@ import pytest from httpx import Client as HttpxClient +from httpx import Headers from httpx import HTTPStatusError, Response from storage3 import SyncStorageClient from storage3.exceptions import StorageApiError from storage3.utils import StorageException +from yarl import URL from .. import SyncBucketProxy from ..utils import SyncFinalizerFactory @@ -825,3 +827,33 @@ def test_client_list_v2_paginated( assert all(f.name.startswith(file.bucket_path) for f in result.objects) pages += 1 assert pages == 4 + + +def test_upload_closes_file_handle_on_error(tmp_path: Path) -> None: + """A failed upload must not leak the handle storage3 opened itself.""" + source = tmp_path / "leak.txt" + source.write_bytes(b"payload") + + error_response = Mock(spec=Response) + error_response.json.return_value = { + "message": "Duplicate", + "error": "Duplicate", + "statusCode": 409, + } + + captured: dict = {} + + def fail(*args: object, **kwargs: object) -> Response: + captured["files"] = kwargs.get("files") + raise HTTPStatusError("409", request=Mock(), response=error_response) + + client = Mock() + client.headers = Headers() + client.request = fail + + proxy = SyncBucketProxy("bucket", URL("http://localhost"), Headers(), client) + + with pytest.raises(StorageApiError): + proxy.upload("leak.txt", str(source)) + + assert captured["files"]["file"][1].closed