From e45502b2eaab7dd968cdd4edece5b179b0757ae5 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 7 Oct 2026 13:35:18 +0400 Subject: [PATCH] =?UTF-8?q?fix(files):=20=D0=B4=D0=B5=D1=82=D0=B5=D1=80?= =?UTF-8?q?=D0=BC=D0=B8=D0=BD=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=D0=BD?= =?UTF-8?q?=D0=BE=D0=B5=20=D0=BE=D1=81=D0=B2=D0=BE=D0=B1=D0=BE=D0=B6=D0=B4?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D0=B5=20S3-=D0=BF=D0=BE=D1=82=D0=BE=D0=BA?= =?UTF-8?q?=D0=B0=20=D0=BF=D1=80=D0=B8=20=D0=BE=D1=82=D0=B4=D0=B0=D1=87?= =?UTF-8?q?=D0=B5=20=D0=B8=D1=81=D1=85=D0=BE=D0=B4=D0=BD=D0=BE=D0=B3=D0=BE?= =?UTF-8?q?=20=D1=84=D0=B0=D0=B9=D0=BB=D0=B0=20(T9.1,=20T9.2,=20T9.4,=20T9?= =?UTF-8?q?.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ClosingStreamingResponse закрывает body_iterator в finally (под shield), ChainedStream закрывает исходный поток и до начала чтения; заголовок X-Content-Type-Options: nosniff; - ASCII-имя в Content-Disposition строится из file_type, а не из расширения исходного имени; - ObjectStorage.stream типизирован как AsyncGenerator, убраны два type: ignore. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/src/api/endpoints/files/read.py | 5 ++- api/src/common/streaming.py | 50 +++++++++++++++++++++++++ api/src/core/clients.py | 4 +- api/src/core/storage.py | 4 +- api/src/services/primary_files.py | 27 ++++++-------- api/tests/conftest.py | 4 +- api/tests/test_api_files_content.py | 58 ++++++++++++++++++++++++++++- 7 files changed, 127 insertions(+), 25 deletions(-) create mode 100644 api/src/common/streaming.py diff --git a/api/src/api/endpoints/files/read.py b/api/src/api/endpoints/files/read.py index a2b1ba4..236a7c3 100644 --- a/api/src/api/endpoints/files/read.py +++ b/api/src/api/endpoints/files/read.py @@ -16,6 +16,7 @@ from fastapi.responses import StreamingResponse from src.api.dependencies.auth import require from src.api.dependencies.files import PrimaryFilesServiceDep +from src.common.streaming import ClosingStreamingResponse from src.schemas.common import ErrorDetail, Page, PageParams from src.schemas.files import FileOut @@ -95,7 +96,9 @@ async def get_file_content( ) -> StreamingResponse: """Содержимое исходного файла из S3 потоком (в память целиком не читается); `inline` — для просмотра в браузере.""" stream, media_type, header = await service.content(file_id, disposition) - return StreamingResponse(stream, media_type=media_type, headers={"Content-Disposition": header}) + return ClosingStreamingResponse( + stream, media_type=media_type, headers={"Content-Disposition": header, "X-Content-Type-Options": "nosniff"} + ) @router.delete( diff --git a/api/src/common/streaming.py b/api/src/common/streaming.py new file mode 100644 index 0000000..119f222 --- /dev/null +++ b/api/src/common/streaming.py @@ -0,0 +1,50 @@ +"""Потоковая отдача файла с детерминированным освобождением источника (T9.1, Review Focus 4 плана Среза 3b). + +Starlette `StreamingResponse.stream_response` не вызывает `aclose()` у `body_iterator`: при обрыве клиента или если тело +не начали итерировать поток S3 освобождался бы только сборщиком мусора. `ClosingStreamingResponse` закрывает итератор +в `finally`, а `ChainedStream` гарантирует, что `aclose` закрывает и исходный поток, даже если обёртку не начинали +читать (у неначатого асинхронного генератора `finally` не выполняется, поэтому обёртка — объект, а не генератор). +""" + +from collections.abc import AsyncGenerator + +import anyio +from fastapi.responses import StreamingResponse + + +class ChainedStream: + """Уже прочитанная первая порция и остальной поток `rest`; `aclose` всегда закрывает `rest` (там `close` + `release` + ответа S3), в том числе при завершении, ошибке и обрыве потребителя, начатом и не начатом чтении.""" + + def __init__(self, first: bytes, rest: AsyncGenerator[bytes, None]) -> None: + self._first = first + self._rest = rest + + def __aiter__(self) -> "ChainedStream": + return self + + async def __anext__(self) -> bytes: + if self._first: + chunk, self._first = self._first, b"" + return chunk + return await anext(self._rest) + + async def aclose(self) -> None: + """Закрыть исходный поток (идемпотентно: у завершённого генератора `aclose` — no-op).""" + await self._rest.aclose() + + +class ClosingStreamingResponse(StreamingResponse): + """`StreamingResponse`, закрывающий `body_iterator` (`aclose`) после отдачи, ошибки или обрыва клиента. + + Закрытие идёт под `CancelScope(shield=True)`: при обрыве клиента задача отдачи отменяется, а освобождение + соединения S3 требует `await`.""" + + async def stream_response(self, send) -> None: # type: ignore[no-untyped-def] # noqa: ANN001 + try: + await super().stream_response(send) + finally: + aclose = getattr(self.body_iterator, "aclose", None) + if aclose is not None: + with anyio.CancelScope(shield=True): + await aclose() diff --git a/api/src/core/clients.py b/api/src/core/clients.py index 0d2860f..097d5b6 100644 --- a/api/src/core/clients.py +++ b/api/src/core/clients.py @@ -2,7 +2,7 @@ по docs/context/03-architecture.md (референс — reference-old-arm/api/src/core/clients.py). """ -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from io import BytesIO from miniopy_async import Minio @@ -47,7 +47,7 @@ class S3Storage: response.close() await response.release() - async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]: + async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncGenerator[bytes, None]: """Объект порциями; `NoSuchKey` → `ObjectNotFoundError` до первой порции; ответ отпускается и при обрыве.""" try: response = await self.client.get_object(self.bucket, obj_name) diff --git a/api/src/core/storage.py b/api/src/core/storage.py index cdfb87a..9eaf8e4 100644 --- a/api/src/core/storage.py +++ b/api/src/core/storage.py @@ -4,7 +4,7 @@ `services/directory_import.py`: тот же интерфейс нужен загрузке файлов и маппингу результатов (Срез 2c, D-12). """ -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from typing import Protocol @@ -23,7 +23,7 @@ class ObjectStorage(Protocol): """Содержимое объекта; отсутствие объекта — `ObjectNotFoundError`.""" ... - def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]: + def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncGenerator[bytes, None]: """Объект порциями по `chunk_size` байт; отсутствие объекта — `ObjectNotFoundError` до первой порции. Закрытие итератора (`aclose`) освобождает соединение с S3.""" diff --git a/api/src/services/primary_files.py b/api/src/services/primary_files.py index e5be153..7d29469 100644 --- a/api/src/services/primary_files.py +++ b/api/src/services/primary_files.py @@ -18,7 +18,7 @@ audit-trail «События файлов и документов»; design D-4) import hashlib import uuid -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from dataclasses import dataclass from typing import Literal @@ -30,7 +30,8 @@ from starlette import status as http_status from src.api.dependencies.audit import RequestAudit from src.common.clock import utc_now from src.common.exceptions import ConflictError, DomainError, DomainValidationError, NotFoundError -from src.common.filenames import CONTENT_TYPES, content_disposition, extension, safe_filename +from src.common.filenames import CONTENT_TYPES, content_disposition, safe_filename +from src.common.streaming import ChainedStream from src.contract import CONTRACT_VERSION, request_subject from src.core.config import Settings from src.core.storage import ObjectNotFoundError, ObjectStorage @@ -116,16 +117,10 @@ class RejectedFile: ) -async def _chain(first: bytes, rest: AsyncIterator[bytes]) -> AsyncIterator[bytes]: - """Первая (уже прочитанная) порция и остальной поток; при завершении, ошибке и обрыве потребителя поток S3 - закрывается (`aclose` — там `close` + `release` ответа), соединение не остаётся занятым.""" - try: - if first: - yield first - async for chunk in rest: - yield chunk - finally: - await rest.aclose() # type: ignore[attr-defined] +def _chain(first: bytes, rest: AsyncGenerator[bytes, None]) -> ChainedStream: + """Первая (уже прочитанная) порция и остальной поток; `aclose` закрывает поток S3 (там `close` + `release` + ответа) — соединение не остаётся занятым при завершении, ошибке и обрыве потребителя (в т. ч. не начатом).""" + return ChainedStream(first, rest) def upload_key(file_id: uuid.UUID, job_id: uuid.UUID, name: str) -> str: @@ -255,7 +250,7 @@ class PrimaryFilesService: async def content( self, file_id: uuid.UUID, disposition: Literal["attachment", "inline"] = "attachment" - ) -> tuple[AsyncIterator[bytes], str, str]: + ) -> tuple[ChainedStream, str, str]: """Исходный файл потоком: `(поток, Content-Type, Content-Disposition)` (spec documents-api, D-10). Файл любого статуса (в т. ч. `SUPERSEDED`) в области видимости; чужой и несуществующий — 404. Ключ и имя @@ -275,11 +270,11 @@ class PrimaryFilesService: except ObjectNotFoundError: raise FileContentNotFoundError() from None except BaseException: - await iterator.aclose() # type: ignore[attr-defined] + await iterator.aclose() raise name = file.original_name or "" - suffix = extension(name) or file.file_type.value.lower() - header = content_disposition(name or f"file.{suffix}", disposition=disposition, ascii_fallback=f"file.{suffix}") + ascii_name = f"file.{file.file_type.value.lower()}" + header = content_disposition(name or ascii_name, disposition=disposition, ascii_fallback=ascii_name) media_type = CONTENT_TYPES.get(file.file_type.value.lower(), "application/octet-stream") return _chain(first, iterator), media_type, header diff --git a/api/tests/conftest.py b/api/tests/conftest.py index b1caa39..76c72d9 100644 --- a/api/tests/conftest.py +++ b/api/tests/conftest.py @@ -13,7 +13,7 @@ S3 — `FakeStorage` (объекты в словаре). """ import uuid -from collections.abc import AsyncIterator, Awaitable, Callable +from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager from functools import partial from pathlib import Path @@ -222,7 +222,7 @@ class FakeStorage: except KeyError: raise ObjectNotFoundError(obj_name) from None - async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]: + async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncGenerator[bytes, None]: try: data = self.objects[obj_name] except KeyError: diff --git a/api/tests/test_api_files_content.py b/api/tests/test_api_files_content.py index f4e9256..06b05e6 100644 --- a/api/tests/test_api_files_content.py +++ b/api/tests/test_api_files_content.py @@ -1,12 +1,13 @@ """Исходный файл потоком (spec documents-api «Исходный файл»; design D-10).""" -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator import pytest from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from src.common.streaming import ClosingStreamingResponse from src.core.storage import ObjectNotFoundError from src.models.enums import FileStatus, FileType, RecognitionStatus from src.models.files import S3Object, StoredFile @@ -20,6 +21,7 @@ PDF = b"%PDF-1.4\n" + bytes(range(256)) * 4096 # ~1 МиБ: проверка п HTTP_OK = 200 HTTP_NOT_FOUND = 404 HTTP_FORBIDDEN = 403 +SECOND_BODY = 3 async def _file( @@ -46,6 +48,7 @@ async def test_inline_pdf(app, seeded: AsyncSession, auth_client: AuthClientFact assert response.headers["content-type"] == "application/pdf" assert response.headers["content-disposition"] == "inline; filename=\"file.pdf\"; filename*=UTF-8''aep_6.pdf" assert response.content == PDF + assert response.headers["x-content-type-options"] == "nosniff" # T9.5 async def test_cyrillic_name_and_missing(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ANN001 @@ -136,7 +139,7 @@ class _Source: self._fail_after = fail_after self.iterator = self._run() - async def _run(self) -> AsyncIterator[bytes]: + async def _run(self) -> AsyncGenerator[bytes, None]: try: for index in range(3): if self._fail_after == index: @@ -165,3 +168,54 @@ async def test_chain_closes_source_on_error_and_yields_all() -> None: complete = _Source() assert b"".join([chunk async for chunk in _chain(await anext(complete.iterator, b""), complete.iterator)]) == b"xxx" assert complete.closed + + +async def test_chain_aclose_closes_source_when_never_iterated() -> None: + """T9.1: поток не начали отдавать (первая порция уже прочитана) — `aclose` всё равно закрывает S3-поток.""" + source = _Source() + chain = _chain(await anext(source.iterator, b""), source.iterator) + await chain.aclose() + assert source.closed + + +async def test_response_closes_stream_when_client_drops_after_first_chunk() -> None: + """T9.1: `send` падает на втором теле (обрыв клиента) — `body_iterator` закрыт, S3-поток освобождён.""" + source = _Source() + chain = _chain(await anext(source.iterator, b""), source.iterator) + response = ClosingStreamingResponse(chain, media_type="application/pdf") + sent: list[str] = [] + + async def send(message: dict) -> None: # type: ignore[type-arg] + sent.append(message["type"]) + if len(sent) == SECOND_BODY: # start, первая порция, вторая порция + raise OSError("клиент отключился") + + with pytest.raises(OSError, match="клиент"): + await response.stream_response(send) + assert source.closed + + +async def test_response_closes_stream_when_send_fails_at_start() -> None: + """T9.1: тело не начали итерировать (падает уже отправка заголовков) — поток тоже закрыт.""" + source = _Source() + chain = _chain(await anext(source.iterator, b""), source.iterator) + response = ClosingStreamingResponse(chain) + + async def send(message: dict) -> None: # type: ignore[type-arg] + raise OSError("клиент отключился") + + with pytest.raises(OSError, match="клиент"): + await response.stream_response(send) + assert source.closed + + +async def test_ascii_extension_from_file_type(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None: # noqa: ANN001 + """T9.2: ASCII-имя берётся из `file_type`, а не из расширения исходного имени (нестандартное/кириллическое).""" + storage = FakeStorage() + app.state.storage = storage + file = await _file(seeded, storage, "отчёт.экс", FileType.XLSX) + client = await auth_client(KA_RF) + response = await client.get(f"/files/{file.id}/content") + disposition = response.headers["content-disposition"] + assert disposition.startswith("attachment; filename=\"file.xlsx\"; filename*=UTF-8''") + assert disposition.isascii()