fix: stream large unsigned PUTs once (deferred SigV4)

s3proxy/Dockerfile

@@ -36,4 +36,5 @@ ENV S3PROXY_IP=0.0.0.0 \
 
 EXPOSE 4433
 
-CMD ["uvicorn", "s3proxy.main:app", "--host", "0.0.0.0", "--port", "4433", "--loop", "uvloop"]
+# Use the CLI entrypoint so TLS settings from Settings apply.
+CMD ["python", "-m", "s3proxy.main"]

s3proxy/client/types.py

@@ -13,6 +13,33 @@ class S3Credentials:
     service: str = "s3"
 
 
+@dataclass(slots=True)
+class ParsedHeaderAuth:
+    """Parsed SigV4 Authorization header, before signature verification."""
+
+    credentials: S3Credentials | None
+    signed_headers: list[str]
+    signature: str
+    date_stamp: str
+    region: str
+    service: str
+    amz_date: str
+    error: str | None = None
+
+    @classmethod
+    def fail(cls, error: str, *, credentials: S3Credentials | None = None) -> ParsedHeaderAuth:
+        return cls(
+            credentials=credentials,
+            signed_headers=[],
+            signature="",
+            date_stamp="",
+            region="",
+            service="",
+            amz_date="",
+            error=error,
+        )
+
+
 @dataclass(slots=True)
 class ParsedRequest:
     """Parsed S3 request information."""

s3proxy/client/verifier.py

@@ -13,7 +13,7 @@
 import structlog
 from structlog.stdlib import BoundLogger
 
-from .types import ParsedRequest, S3Credentials
+from .types import ParsedHeaderAuth, ParsedRequest, S3Credentials
 
 logger: BoundLogger = structlog.get_logger(__name__)
 
@@ -95,10 +95,8 @@ def verify(self, request: ParsedRequest, path: str) -> tuple[bool, S3Credentials
 
         return False, None, "No AWS signature found"
 
-    def _verify_header_signature(
-        self, request: ParsedRequest, path: str, auth_header: str
-    ) -> tuple[bool, S3Credentials | None, str]:
-        """Verify Authorization header signature."""
+    def _parse_header_auth(self, request: ParsedRequest, auth_header: str) -> ParsedHeaderAuth:
+        """Parse Authorization header fields without verifying the signature."""
         try:
             parts = auth_header.replace("AWS4-HMAC-SHA256 ", "").split(",")
             auth_parts = {}
@@ -112,41 +110,126 @@ def _verify_header_signature(
 
             credentials, date_stamp, region, service, error = self._parse_v4_credential(credential)
             if error:
-                return False, None, error
+                return ParsedHeaderAuth.fail(error)
 
             amz_date = request.headers.get("x-amz-date", "")
             if not amz_date:
-                return False, credentials, "Missing x-amz-date header"
-
-            try:
-                request_time = datetime.strptime(amz_date, "%Y%m%dT%H%M%SZ").replace(tzinfo=UTC)
-                if abs(datetime.now(UTC) - request_time) > CLOCK_SKEW_TOLERANCE:
-                    return False, credentials, "Request time too skewed"
-            except ValueError:
-                return False, credentials, "Invalid x-amz-date format"
+                return ParsedHeaderAuth.fail(
+                    "Missing x-amz-date header",
+                    credentials=credentials,
+                )
 
-            canonical_request = self._build_canonical_request(
-                request, path, signed_headers.split(";")
-            )
-            calculated_sig = self._compute_v4_signature(
-                canonical_request, amz_date, date_stamp, region, service, credentials.secret_key
+            return ParsedHeaderAuth(
+                credentials=credentials,
+                signed_headers=signed_headers.split(";"),
+                signature=signature,
+                date_stamp=date_stamp,
+                region=region,
+                service=service,
+                amz_date=amz_date,
             )
+        except (KeyError, ValueError, IndexError) as e:
+            return ParsedHeaderAuth.fail(f"Invalid Authorization header: {e}")
 
-            if hmac.compare_digest(calculated_sig, signature):
-                return True, credentials, ""
+    def _check_request_time(self, amz_date: str) -> str | None:
+        try:
+            request_time = datetime.strptime(amz_date, "%Y%m%dT%H%M%SZ").replace(tzinfo=UTC)
+            if abs(datetime.now(UTC) - request_time) > CLOCK_SKEW_TOLERANCE:
+                return "Request time too skewed"
+        except ValueError:
+            return "Invalid x-amz-date format"
+        return None
+
+    def prepare_header_auth(
+        self, request: ParsedRequest, auth_header: str
+    ) -> tuple[S3Credentials | None, str]:
+        """Parse credentials and clock skew. Signature verified separately."""
+        parsed = self._parse_header_auth(request, auth_header)
+        if parsed.error:
+            return None, parsed.error
+        assert parsed.credentials is not None
+        time_error = self._check_request_time(parsed.amz_date)
+        if time_error:
+            return parsed.credentials, time_error
+        return parsed.credentials, ""
+
+    def verify_with_payload_hash(
+        self,
+        request: ParsedRequest,
+        path: str,
+        auth_header: str,
+        payload_hash: str,
+    ) -> tuple[bool, S3Credentials | None, str]:
+        """Verify SigV4 using an explicit payload hash (stream-computed)."""
+        auth = self._parse_header_auth(request, auth_header)
+        if auth.error:
+            return False, auth.credentials, auth.error
+        assert auth.credentials is not None
+
+        time_error = self._check_request_time(auth.amz_date)
+        if time_error:
+            return False, auth.credentials, time_error
+
+        canonical_request = self._build_canonical_request(
+            request, path, auth.signed_headers, payload_hash=payload_hash
+        )
+        calculated_sig = self._compute_v4_signature(
+            canonical_request,
+            auth.amz_date,
+            auth.date_stamp,
+            auth.region,
+            auth.service,
+            auth.credentials.secret_key,
+        )
 
-            logger.debug(
-                "Signature verification failed",
-                method=request.method,
-                path=path,
-                signed_headers=signed_headers,
-                expected_sig=signature[:16] + "...",
-                calculated_sig=calculated_sig[:16] + "...",
-            )
-            return False, credentials, "Signature mismatch"
+        if hmac.compare_digest(calculated_sig, auth.signature):
+            return True, auth.credentials, ""
 
-        except (KeyError, ValueError, IndexError) as e:
-            return False, None, f"Invalid Authorization header: {e}"
+        logger.debug(
+            "Signature verification failed",
+            method=request.method,
+            path=path,
+            signed_headers=";".join(auth.signed_headers),
+            expected_sig=auth.signature[:16] + "...",
+            calculated_sig=calculated_sig[:16] + "...",
+        )
+        return False, auth.credentials, "Signature mismatch"
+
+    def _verify_header_signature(
+        self, request: ParsedRequest, path: str, auth_header: str
+    ) -> tuple[bool, S3Credentials | None, str]:
+        """Verify Authorization header signature."""
+        auth = self._parse_header_auth(request, auth_header)
+        if auth.error:
+            return False, auth.credentials, auth.error
+        assert auth.credentials is not None
+
+        time_error = self._check_request_time(auth.amz_date)
+        if time_error:
+            return False, auth.credentials, time_error
+
+        canonical_request = self._build_canonical_request(request, path, auth.signed_headers)
+        calculated_sig = self._compute_v4_signature(
+            canonical_request,
+            auth.amz_date,
+            auth.date_stamp,
+            auth.region,
+            auth.service,
+            auth.credentials.secret_key,
+        )
+
+        if hmac.compare_digest(calculated_sig, auth.signature):
+            return True, auth.credentials, ""
+
+        logger.debug(
+            "Signature verification failed",
+            method=request.method,
+            path=path,
+            signed_headers=";".join(auth.signed_headers),
+            expected_sig=auth.signature[:16] + "...",
+            calculated_sig=calculated_sig[:16] + "...",
+        )
+        return False, auth.credentials, "Signature mismatch"
 
     def _verify_presigned_v4(
         self, request: ParsedRequest, path: str
@@ -268,7 +351,12 @@ def _verify_presigned_v2(
             return False, None, f"Invalid V2 presigned URL: {e}"
 
     def _build_canonical_request(
-        self, request: ParsedRequest, path: str, signed_headers: list[str]
+        self,
+        request: ParsedRequest,
+        path: str,
+        signed_headers: list[str],
+        *,
+        payload_hash: str | None = None,
     ) -> str:
         """Build canonical request for signature verification."""
         method = request.method.upper()
@@ -285,9 +373,10 @@ def _build_canonical_request(
 
         signed_headers_str = ";".join(sorted(signed_headers))
 
-        payload_hash = request.headers.get(
-            "x-amz-content-sha256", hashlib.sha256(request.body).hexdigest()
-        )
+        if payload_hash is None:
+            payload_hash = request.headers.get(
+                "x-amz-content-sha256", hashlib.sha256(request.body).hexdigest()
+            )
 
         return "\n".join(
             [

s3proxy/disconnect.py

@@ -0,0 +1,13 @@
+"""Client disconnect errors for streaming uploads."""
+
+from __future__ import annotations
+
+from .errors import S3Error
+
+
+class ClientDisconnectError(S3Error):
+    """Client closed the connection mid-upload."""
+
+    @classmethod
+    def raised(cls) -> ClientDisconnectError:
+        return cls(400, "BadRequest", "Client disconnected")

s3proxy/handlers/multipart/upload_part.py

@@ -17,6 +17,7 @@
 from ... import crypto
 from ...client import S3Client, S3Credentials
 from ...errors import S3Error, raise_for_client_error, raise_for_exception
+from ...signature import deferred_signature_required, verify_deferred_payload_hash
 from ...state import (
     InternalPartMetadata,
     MultipartUploadState,
@@ -188,7 +189,11 @@ async def handle_upload_part(self, request: Request, creds: S3Credentials) -> Re
                     )
 
                 # Late signature verification for large signed uploads
-                if is_large_signed and content_sha and result["computed_sha256"] != content_sha:
+                if deferred_signature_required(request):
+                    verify_deferred_payload_hash(
+                        request, request.app.state.verifier, result["computed_sha256"]
+                    )
+                elif is_large_signed and content_sha and result["computed_sha256"] != content_sha:
                     logger.warning(
                         "UPLOAD_PART_SHA256_MISMATCH",
                         bucket=bucket,

s3proxy/handlers/objects/put.py

@@ -2,15 +2,19 @@
 
 import base64
 import hashlib
+from collections.abc import AsyncIterator
 from typing import Any
 
 import structlog
 from fastapi import Request, Response
+from starlette.requests import ClientDisconnect
 from structlog.stdlib import BoundLogger
 
 from ... import crypto
 from ...client import S3Client, S3Credentials
+from ...disconnect import ClientDisconnectError
 from ...errors import S3Error
+from ...signature import verify_deferred_payload_hash
 from ...state import (
     MultipartMetadata,
     PartMetadata,
@@ -22,6 +26,27 @@
 
 logger: BoundLogger = structlog.get_logger(__name__)
 
+_STREAM_CHUNK = 64 * 1024
+
+
+async def _iter_request_body(request: Request, decode_chunked: bool) -> AsyncIterator[bytes]:
+    """Single-pass body iterator: preloaded small body, else live stream.
+
+    Avoids reading request.stream() after request.body() pinned the payload in
+    Starlette's cache (double buffer on large uploads).
+    """
+    if decode_chunked:
+        async for chunk in decode_aws_chunked_stream(request):
+            yield chunk
+        return
+    preloaded = getattr(request.state, "s3proxy_preloaded_body", None)
+    if isinstance(preloaded, (bytes, bytearray)) and len(preloaded) > 0:
+        for offset in range(0, len(preloaded), _STREAM_CHUNK):
+            yield preloaded[offset : offset + _STREAM_CHUNK]
+        return
+    async for chunk in request.stream():
+        yield chunk
+
 
 class PutObjectMixin(BaseHandler):
     async def handle_put_object(self, request: Request, creds: S3Credentials) -> Response:
@@ -126,7 +151,11 @@ async def _put_buffered(
             content_length_mb=round(content_length / 1024 / 1024, 2),
         )
 
-        body = await request.body()
+        preloaded = getattr(request.state, "s3proxy_preloaded_body", None)
+        if isinstance(preloaded, (bytes, bytearray)) and len(preloaded) > 0:
+            body = bytes(preloaded)
+        else:
+            body = await request.body()
         if needs_chunked_decode:
             body = decode_aws_chunked(body)
 
@@ -191,7 +220,8 @@ async def _put_streaming(
         total_plaintext_size = 0
         part_num = 0
         md5_hash = hashlib.md5(usedforsecurity=False)
-        sha256_hash = hashlib.sha256() if expected_sha256 else None
+        deferred_sig = getattr(request.state, "s3proxy_deferred_sig", False) is True
+        sha256_hash = hashlib.sha256() if (expected_sha256 or deferred_sig) else None
         buffer = bytearray()
 
         async def upload_part(data: bytes) -> None:
@@ -230,10 +260,7 @@ async def upload_part(data: bytes) -> None:
                 verify_sha256=expected_sha256 is not None,
             )
 
-            if decode_chunked:
-                stream_source = decode_aws_chunked_stream(request)
-            else:
-                stream_source = request.stream()
+            stream_source = _iter_request_body(request, decode_chunked)
 
             async for chunk in stream_source:
                 buffer.extend(chunk)
@@ -254,8 +281,16 @@ async def upload_part(data: bytes) -> None:
                 buffer.clear()
                 await upload_part(part_data)
 
-            # Verify SHA256 if provided
-            if expected_sha256 is not None:
+            # Verify SHA256 if provided, or deferred SigV4 after streaming hash
+            if deferred_sig:
+                try:
+                    verify_deferred_payload_hash(
+                        request, request.app.state.verifier, sha256_hash.hexdigest()
+                    )
+                except S3Error:
+                    await client.abort_multipart_upload(bucket, key, upload_id)
+                    raise
+            elif expected_sha256 is not None:
                 computed_sha256 = sha256_hash.hexdigest()
                 if computed_sha256 != expected_sha256:
                     logger.error(
@@ -298,6 +333,9 @@ async def upload_part(data: bytes) -> None:
             )
             return Response(headers={"ETag": f'"{etag}"'})
 
+        except ClientDisconnectError, ClientDisconnect:
+            await self._safe_abort(client, bucket, key, upload_id)
+            raise ClientDisconnectError.raised() from None
         except S3Error:
             raise
         except Exception as e: