From c49543b6f0a16851c83e4accd8195d3286be64ec Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Wed, 8 Jul 2026 22:20:40 +0200 Subject: [PATCH] fix: range-aware passthrough for Scylla manifest UploadPartCopy PR #121 only normalized ranges that matched metadata total_plaintext_size. Prod sends bytes=0-(~4768MB-1) while sidecar metadata totals ~6GB, so copies still routed to UPLOAD_PART_COPY_STREAMING and timed out at 300s. Allow large ranged copies from offset 0 when they align to internal ciphertext segments, filter segments for passthrough, and log UPLOAD_PART_COPY_ROUTE with passthrough_blocked_reason for future diagnosis. Co-authored-by: Cursor --- s3proxy/handlers/multipart/copy.py | 226 +++++++++++++++--- .../unit/test_upload_part_copy_passthrough.py | 199 ++++++++++++++- 2 files changed, 386 insertions(+), 39 deletions(-) diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index 4a9f12d..333084c 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -88,24 +88,44 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) head_resp, None, src_wrapped_dek, src_multipart_meta ) copy_source_range = self._normalize_copy_source_range( - raw_copy_source_range, total_plaintext + raw_copy_source_range, total_plaintext, head_resp, src_wrapped_dek ) plaintext_size = self._copy_plaintext_size( head_resp, copy_source_range, src_wrapped_dek, src_multipart_meta ) + passthrough_block = self._passthrough_block_reason( + copy_source_range, + raw_copy_source_range, + src_wrapped_dek, + src_multipart_meta, + src_metadata, + creds, + plaintext_size, + head_resp, + ) + route = ( + "passthrough" + if passthrough_block is None + else ("streaming" if plaintext_size > crypto.STREAMING_THRESHOLD else "simple") + ) + self._log_upload_part_copy_route( + bucket=bucket, + key=key, + part_num=part_num, + raw_copy_source_range=raw_copy_source_range, + normalized_copy_source_range=copy_source_range, + metadata_total_plaintext=total_plaintext, + range_plaintext_size=plaintext_size, + route=route, + passthrough_blocked_reason=passthrough_block, + ) + if plaintext_size <= crypto.STREAMING_THRESHOLD: # Small copies buffer the whole object + re-encrypt it; gate them # by the limiter too (they carry no body, so the request-level # reservation was ~nothing and a small-object flood ran unbounded). - if self._can_passthrough_part_copy( - copy_source_range, - src_wrapped_dek, - src_multipart_meta, - src_metadata, - creds, - plaintext_size, - ): + if passthrough_block is None: return await self._gated_passthrough_copy_part( client, bucket, @@ -121,6 +141,7 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) src_wrapped_dek, src_multipart_meta, plaintext_size, + copy_source_range, ) peak = crypto.copy_pipeline_peak(plaintext_size) async with concurrency.reserve_copy_memory(peak): @@ -139,14 +160,7 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) src_wrapped_dek, src_multipart_meta, ) - if self._can_passthrough_part_copy( - copy_source_range, - src_wrapped_dek, - src_multipart_meta, - src_metadata, - creds, - plaintext_size, - ): + if passthrough_block is None: return await self._gated_passthrough_copy_part( client, bucket, @@ -162,6 +176,7 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) src_wrapped_dek, src_multipart_meta, plaintext_size, + copy_source_range, ) return await self._streaming_copy_part( client, @@ -204,45 +219,172 @@ def _copy_plaintext_size( start, end = self._parse_copy_source_range(copy_source_range, total) return end - start + 1 + def _parse_raw_copy_source_range(self, copy_source_range: str) -> tuple[int, int]: + """Parse x-amz-copy-source-range without clamping end to object size.""" + range_str = copy_source_range.replace("bytes=", "") + try: + start, end = map(int, range_str.split("-")) + except (ValueError, TypeError) as err: + raise S3Error.invalid_range("Invalid copy source range format") from err + if start > end: + raise S3Error.invalid_range("Range not satisfiable") + return start, end + + def _head_plaintext_size(self, head_resp: dict, src_wrapped_dek: str | None) -> int | None: + if not src_wrapped_dek: + return None + size_str = head_resp.get("Metadata", {}).get("plaintext-size") + if not size_str: + return None + return int(size_str) + def _normalize_copy_source_range( - self, copy_source_range: str | None, total_plaintext_size: int + self, + copy_source_range: str | None, + total_plaintext_size: int, + head_resp: dict, + src_wrapped_dek: str | None, ) -> str | None: """Treat a range spanning the entire object as 'copy whole object'. - Scylla Manager often sends ``bytes=0-(size-1)`` on manifest UploadPartCopy - even when copying the full SST. That must not force the streaming re-encrypt - path (which queues behind the copy pipeline and exceeds the 300s client timeout). + Scylla Manager often sends ``bytes=0-(size-1)`` on manifest UploadPartCopy. + When that matches metadata or HEAD plaintext-size, clear the range so routing + treats it as a whole-object copy. Ranges shorter than metadata total (prod + manifest shape) are left intact and handled by range-aware passthrough. """ if not copy_source_range: return None - start, end = self._parse_copy_source_range(copy_source_range, total_plaintext_size) - if start == 0 and end == total_plaintext_size - 1: + raw_start, raw_end = self._parse_raw_copy_source_range(copy_source_range) + end = self._parse_copy_source_range(copy_source_range, total_plaintext_size)[1] + if raw_start == 0 and end == total_plaintext_size - 1: + return None + head_plaintext = self._head_plaintext_size(head_resp, src_wrapped_dek) + if raw_start == 0 and head_plaintext and raw_end >= head_plaintext - 1: return None return copy_source_range - def _can_passthrough_part_copy( + def _segments_for_plaintext_range( + self, + segments: list[_CiphertextSegment], + range_start: int, + range_end: int, + ) -> list[_CiphertextSegment] | None: + """Segments fully inside [range_start, range_end], or None if range splits a segment.""" + if range_start > range_end: + return [] + selected: list[_CiphertextSegment] = [] + pt_offset = 0 + for seg in segments: + seg_start = pt_offset + seg_end = pt_offset + seg.plaintext_size - 1 + if seg_end < range_start: + pt_offset += seg.plaintext_size + continue + if seg_start > range_end: + break + if seg_start < range_start or seg_end > range_end: + return None + selected.append(seg) + pt_offset += seg.plaintext_size + return selected + + def _passthrough_block_reason( self, copy_source_range: str | None, + raw_copy_source_range: str | None, src_wrapped_dek: str | None, src_multipart_meta: MultipartMetadata | None, src_metadata: dict, creds: S3Credentials, plaintext_size: int, - ) -> bool: - """True when a native server-side copy preserves decryptability without re-encrypt.""" - if copy_source_range: - return False + head_resp: dict, + ) -> str | None: + """None when server-side ciphertext copy is allowed; else a short reason code.""" if not src_wrapped_dek and not src_multipart_meta: - return False - # Large PutObject blobs have no sidecar part map; streaming re-encrypt frames them. + return "unencrypted_source" if not src_multipart_meta and plaintext_size > crypto.STREAMING_THRESHOLD: - return False + return "large_put_object_no_sidecar" if src_multipart_meta: src_kid = src_multipart_meta.kid else: src_kid = src_metadata.get(self.settings.kidtag_name, "") - needs_rekey = bool(src_kid) and src_kid != creds.access_key - return not needs_rekey + if src_kid and src_kid != creds.access_key: + return "needs_rekey" + + if not copy_source_range: + return None + + # Small ranged copies stay on streaming re-encrypt (tests + partial object copies). + if plaintext_size <= crypto.STREAMING_THRESHOLD: + return "small_ranged_copy" + + if not src_multipart_meta: + return "ranged_copy_no_multipart_meta" + + total = src_multipart_meta.total_plaintext_size + range_start, range_end = self._parse_copy_source_range(copy_source_range, total) + if range_start != 0: + return "ranged_copy_nonzero_start" + + segments = self._source_ciphertext_segments( + src_multipart_meta, head_resp, src_wrapped_dek, src_metadata + ) + filtered = self._segments_for_plaintext_range(segments, range_start, range_end) + if filtered is None: + return "ranged_copy_segment_misaligned" + if not filtered: + return "ranged_copy_empty_segments" + return None + + def _log_upload_part_copy_route( + self, + *, + bucket: str, + key: str, + part_num: int, + raw_copy_source_range: str | None, + normalized_copy_source_range: str | None, + metadata_total_plaintext: int, + range_plaintext_size: int, + route: str, + passthrough_blocked_reason: str | None, + ) -> None: + logger.info( + "UPLOAD_PART_COPY_ROUTE", + bucket=bucket, + key=key, + part_num=part_num, + raw_copy_source_range=raw_copy_source_range, + normalized_copy_source_range=normalized_copy_source_range, + metadata_total_plaintext_mb=f"{metadata_total_plaintext / 1024 / 1024:.2f}MB", + range_plaintext_mb=f"{range_plaintext_size / 1024 / 1024:.2f}MB", + route=route, + passthrough_blocked_reason=passthrough_blocked_reason, + ) + + def _can_passthrough_part_copy( + self, + copy_source_range: str | None, + src_wrapped_dek: str | None, + src_multipart_meta: MultipartMetadata | None, + src_metadata: dict, + creds: S3Credentials, + plaintext_size: int, + ) -> bool: + """True when a native server-side copy preserves decryptability without re-encrypt.""" + return ( + self._passthrough_block_reason( + copy_source_range, + copy_source_range, + src_wrapped_dek, + src_multipart_meta, + src_metadata, + creds, + plaintext_size, + {}, + ) + is None + ) def _resolve_source_dek( self, @@ -361,6 +503,7 @@ async def _gated_passthrough_copy_part( src_wrapped_dek: str | None, src_multipart_meta: MultipartMetadata | None, plaintext_size: int, + copy_source_range: str | None = None, ) -> Response: """Server-side ciphertext copy; does not use the streaming pipeline semaphore.""" return await self._passthrough_copy_part( @@ -378,6 +521,7 @@ async def _gated_passthrough_copy_part( src_wrapped_dek, src_multipart_meta, plaintext_size, + copy_source_range, ) async def _passthrough_copy_part( @@ -396,6 +540,7 @@ async def _passthrough_copy_part( src_wrapped_dek: str | None, src_multipart_meta: MultipartMetadata | None, plaintext_size: int, + copy_source_range: str | None = None, ) -> Response: """Server-side copy of encrypted ciphertext; destination adopts the source DEK.""" src_dek, src_kid = self._resolve_source_dek( @@ -411,6 +556,15 @@ async def _passthrough_copy_part( if not segments: raise S3Error.invalid_request("Copy source has no ciphertext segments") + if copy_source_range and src_multipart_meta: + range_start, range_end = self._parse_copy_source_range( + copy_source_range, src_multipart_meta.total_plaintext_size + ) + filtered = self._segments_for_plaintext_range(segments, range_start, range_end) + if filtered is None: + raise S3Error.invalid_request("Copy range splits an encrypted segment") + segments = filtered + copy_source_path = copy_source or f"/{src_bucket}/{quote(src_key, safe='/')}" logger.info( @@ -422,6 +576,7 @@ async def _passthrough_copy_part( src_key=src_key, plaintext_mb=f"{plaintext_size / 1024 / 1024:.2f}MB", segments=len(segments), + copy_source_range=copy_source_range, ) has_internal_layout = bool( @@ -447,7 +602,7 @@ async def _passthrough_copy_part( client, src_bucket, src_key, - None, + copy_source_range, src_wrapped_dek, src_multipart_meta, head_resp, @@ -511,7 +666,7 @@ async def _passthrough_copy_part( client, src_bucket, src_key, - None, + copy_source_range, src_wrapped_dek, src_multipart_meta, head_resp, @@ -539,6 +694,7 @@ async def _passthrough_copy_part( part_num=part_num, internal_parts=len(internal_parts), plaintext_mb=f"{total_plaintext / 1024 / 1024:.2f}MB", + copy_source_range=copy_source_range, ) return Response( content=xml_responses.upload_part_copy_result(etag, format_iso8601(datetime.now(UTC))), diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py index d9d4989..0fae68f 100644 --- a/tests/unit/test_upload_part_copy_passthrough.py +++ b/tests/unit/test_upload_part_copy_passthrough.py @@ -282,13 +282,204 @@ async def test_normalize_copy_source_range_treats_full_object_as_whole( ): handler = _copy_handler(settings, manager) total = crypto.STREAMING_THRESHOLD + crypto.MAX_BUFFER_SIZE - assert handler._normalize_copy_source_range(None, total) is None - assert handler._normalize_copy_source_range(f"bytes=0-{total - 1}", total) is None - assert handler._normalize_copy_source_range(f"bytes=0-{total + 9999}", total) is None - partial = handler._normalize_copy_source_range(f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}", total) + assert handler._normalize_copy_source_range(None, total, {}, None) is None + assert handler._normalize_copy_source_range(f"bytes=0-{total - 1}", total, {}, None) is None + assert handler._normalize_copy_source_range(f"bytes=0-{total + 9999}", total, {}, None) is None + partial = handler._normalize_copy_source_range( + f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}", total, {}, None + ) assert partial == f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}" +def test_segments_for_plaintext_range_selects_aligned_prefix(settings, manager): + from s3proxy.handlers.multipart.copy import _CiphertextSegment + + handler = _copy_handler(settings, manager) + chunk = crypto.MAX_BUFFER_SIZE + ct = chunk + 16 + segments = [_CiphertextSegment(chunk, ct, i * ct) for i in range(5)] + # First two internal parts (64MB). + selected = handler._segments_for_plaintext_range(segments, 0, chunk * 2 - 1) + assert selected is not None + assert len(selected) == 2 + # Partial segment at the end is not passthrough-safe. + assert handler._segments_for_plaintext_range(segments, 0, chunk) is None + + +def _build_scylla_prod_shape_source( + kid, kek, *, num_internal_parts: int, metadata_inflate_ratio: float +): + """Prod shape: metadata total_plaintext > Scylla manifest copy range. + + Scylla uploads ~122 client parts (~6GB metadata total) but manifest copies + ~149 internal chunks (~4.7GB). Simulate by inflating metadata total while + keeping only the first ``num_internal_parts`` segments real. + """ + src_dek = crypto.generate_dek() + chunk_size = crypto.MAX_BUFFER_SIZE + src_plaintext = b"M" * (chunk_size * num_internal_parts) + + ciphertext_blob = bytearray() + internal_parts_meta = [] + for i, start in enumerate(range(0, len(src_plaintext), chunk_size), 1): + chunk = src_plaintext[start : start + chunk_size] + ct = crypto.encrypt_frame(chunk, src_dek, "src-upload", i, 0) + internal_parts_meta.append( + InternalPartMetadata( + internal_part_number=i, + plaintext_size=len(chunk), + ciphertext_size=len(ct), + etag=hashlib.md5(ct, usedforsecurity=False).hexdigest(), + ) + ) + ciphertext_blob.extend(ct) + + inflated_total = int(len(src_plaintext) * metadata_inflate_ratio) + src_meta = MultipartMetadata( + version=2, + part_count=1, + total_plaintext_size=inflated_total, + parts=[ + PartMetadata( + part_number=1, + plaintext_size=len(src_plaintext), + ciphertext_size=len(ciphertext_blob), + etag="ignored", + md5=hashlib.md5(src_plaintext, usedforsecurity=False).hexdigest(), + internal_parts=internal_parts_meta, + ) + ], + wrapped_dek=crypto.wrap_key(src_dek, kek), + kid=kid, + ) + return src_dek, src_plaintext, bytes(ciphertext_blob), src_meta, inflated_total + + +@pytest.mark.asyncio +async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough( + mock_s3, settings, manager, credentials, monkeypatch +): + """Regression: range bytes=0-(N-1) with N < metadata total must passthrough, not stream.""" + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) + handler = _copy_handler(settings, manager) + _patch_client(handler, mock_s3) + await mock_s3.create_bucket(BUCKET) + + kid, kek = settings.keyring.key_for(credentials.access_key) + num_parts = 149 # prod manifest internal part count + src_dek, src_plaintext, ciphertext_blob, src_meta, inflated_total = ( + _build_scylla_prod_shape_source( + kid, kek, num_internal_parts=num_parts, metadata_inflate_ratio=1.27 + ) + ) + assert inflated_total > len(src_plaintext) + assert len(src_plaintext) > crypto.STREAMING_THRESHOLD + + await mock_s3.put_object(BUCKET, "sst/big-Data.db", ciphertext_blob) + await save_multipart_metadata(mock_s3, BUCKET, "sst/big-Data.db", src_meta) + + resp_create = await mock_s3.create_multipart_upload(BUCKET, "sst/big-Data.db.sm_manifest") + upload_id = resp_create["UploadId"] + await manager.create_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id, src_dek, kid) + + scylla_range = f"bytes=0-{len(src_plaintext) - 1}" + mark = len(mock_s3.call_history) + await handler.handle_upload_part_copy( + _copy_part_request( + f"/{BUCKET}/sst/big-Data.db.sm_manifest", + f"/{BUCKET}/sst/big-Data.db", + upload_id, + copy_source_range=scylla_range, + ), + credentials, + ) + during = mock_s3.call_history[mark:] + + assert any(c[0] == "upload_part_copy" for c in during) + assert not any(c[0] == "upload_part" for c in during) + + updated = await manager.get_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id) + part = updated.parts[1] + assert part.plaintext_size == len(src_plaintext) + assert len(part.internal_parts) == num_parts + + +def test_prod_shape_passthrough_eligibility_and_route(settings, manager, credentials): + """Unit check for prod mismatch: metadata total > range plaintext, still eligible.""" + handler = _copy_handler(settings, manager) + kid, kek = settings.keyring.key_for(credentials.access_key) + num_parts = 149 + _, src_plaintext, _, src_meta, inflated_total = _build_scylla_prod_shape_source( + kid, kek, num_internal_parts=num_parts, metadata_inflate_ratio=1.27 + ) + scylla_range = f"bytes=0-{len(src_plaintext) - 1}" + block = handler._passthrough_block_reason( + scylla_range, + scylla_range, + "wrapped-dek", + src_meta, + {}, + credentials, + len(src_plaintext), + {}, + ) + assert block is None + assert inflated_total > len(src_plaintext) + assert ( + handler._normalize_copy_source_range(scylla_range, inflated_total, {}, "wrapped-dek") + == scylla_range + ) + + +@pytest.mark.asyncio +async def test_small_partial_range_blocked_from_passthrough(settings, manager, credentials): + handler = _copy_handler(settings, manager) + kid, kek = settings.keyring.key_for(credentials.access_key) + chunk = crypto.MAX_BUFFER_SIZE + src_meta = MultipartMetadata( + version=2, + part_count=1, + total_plaintext_size=chunk * 3, + parts=[ + PartMetadata( + part_number=1, + plaintext_size=chunk * 3, + ciphertext_size=chunk * 3, + etag="x", + internal_parts=[ + InternalPartMetadata(1, chunk, chunk, "e1"), + InternalPartMetadata(2, chunk, chunk, "e2"), + InternalPartMetadata(3, chunk, chunk, "e3"), + ], + ) + ], + wrapped_dek=crypto.wrap_key(crypto.generate_dek(), kek), + kid=kid, + ) + partial_range = f"bytes=0-{chunk - 1}" + reason = handler._passthrough_block_reason( + partial_range, + partial_range, + "wrapped", + src_meta, + {}, + credentials, + chunk, + {}, + ) + assert reason == "small_ranged_copy" + + +@pytest.mark.asyncio +async def test_normalize_does_not_clear_scylla_shorter_than_metadata_range(settings, manager): + """Shorter-than-metadata range must stay set for range-aware passthrough.""" + handler = _copy_handler(settings, manager) + metadata_total = 6_000_000_000 + scylla_end = 4_999_741_439 # ~4767MB, prod-shaped + raw = f"bytes=0-{scylla_end}" + assert handler._normalize_copy_source_range(raw, metadata_total, {}, "dek") == raw + + def _build_large_multipart_source(mock_s3, kid, kek, *, num_internal_parts: int): """Encrypted multipart source above STREAMING_THRESHOLD (Scylla manifest shape).""" src_dek = crypto.generate_dek()