diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index 3bf0f24..4a9f12d 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -63,7 +63,7 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) async with self._client(creds) as client: upload_id, part_num = self._extract_multipart_params(request) copy_source = request.headers.get("x-amz-copy-source", "") - copy_source_range = request.headers.get("x-amz-copy-source-range") + raw_copy_source_range = request.headers.get("x-amz-copy-source-range") src_bucket, src_key = self._parse_copy_source(copy_source) state = await self.multipart_manager.get_upload(bucket, key, upload_id) @@ -84,6 +84,12 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) src_wrapped_dek = src_metadata.get(self.settings.dektag_name) src_multipart_meta = await load_multipart_metadata(client, src_bucket, src_key) + total_plaintext = self._copy_plaintext_size( + head_resp, None, src_wrapped_dek, src_multipart_meta + ) + copy_source_range = self._normalize_copy_source_range( + raw_copy_source_range, total_plaintext + ) plaintext_size = self._copy_plaintext_size( head_resp, copy_source_range, src_wrapped_dek, src_multipart_meta ) @@ -198,6 +204,22 @@ def _copy_plaintext_size( start, end = self._parse_copy_source_range(copy_source_range, total) return end - start + 1 + def _normalize_copy_source_range( + self, copy_source_range: str | None, total_plaintext_size: int + ) -> 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). + """ + 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: + return None + return copy_source_range + def _can_passthrough_part_copy( self, copy_source_range: str | None, @@ -340,25 +362,7 @@ async def _gated_passthrough_copy_part( src_multipart_meta: MultipartMetadata | None, plaintext_size: int, ) -> Response: - """Passthrough with the same pipeline cap as the re-encrypt path.""" - if plaintext_size > crypto.STREAMING_THRESHOLD: - async with _copy_pipeline_semaphore: - return await self._passthrough_copy_part( - client, - bucket, - key, - upload_id, - part_num, - state, - src_bucket, - src_key, - copy_source, - head_resp, - src_metadata, - src_wrapped_dek, - src_multipart_meta, - plaintext_size, - ) + """Server-side ciphertext copy; does not use the streaming pipeline semaphore.""" return await self._passthrough_copy_part( client, bucket, diff --git a/tests/integration/passthrough_verify.py b/tests/integration/passthrough_verify.py index 8780d94..3bfbd61 100644 --- a/tests/integration/passthrough_verify.py +++ b/tests/integration/passthrough_verify.py @@ -304,14 +304,35 @@ def check_dek_adoption(ctx: RunContext, source: str, dest: str) -> None: def check_reencrypt_control(ctx: RunContext, source: str, dest: str, size: int) -> None: + # Partial range forces streaming re-encrypt; full-object range is passthrough. + partial_end = min(size - 1, 32 * MB - 1) peak, enc = poll_during( ctx, - lambda: upload_part_copy(ctx, dest, source, byte_range=f"bytes=0-{size - 1}"), + lambda: upload_part_copy(ctx, dest, source, byte_range=f"bytes=0-{partial_end}"), ) - ctx.ok("encrypts full object", enc >= size * 0.5, f"{enc / MB:.0f}MB") + ctx.ok("encrypts partial range", enc >= (partial_end + 1) * 0.5, f"{enc / MB:.0f}MB") ctx.ok("high peak memory", peak >= CHUNK_PEAK * 0.5, f"{peak / MB:.2f}MB") +def check_scylla_manifest_full_range_passthrough( + ctx: RunContext, source: str, dest: str, size: int +) -> None: + """Scylla sends bytes=0-(size-1) on manifest UploadPartCopy; must passthrough.""" + peak, enc = poll_during( + ctx, + lambda: upload_part_copy(ctx, dest, source, byte_range=f"bytes=0-{size - 1}"), + ) + ctx.ok("full-range zero bytes_encrypted", enc <= 2 * MB, f"{enc / MB:.2f}MB") + ctx.ok("full-range low peak memory", peak <= CHUNK_PEAK * 0.5, f"{peak / MB:.2f}MB") + src_ct, dest_ct = raw_ciphertext(ctx, source), raw_ciphertext(ctx, dest) + ctx.ok("full-range ciphertext identical", src_ct == dest_ct) + ctx.ok( + "UPLOAD_PART_COPY_PASSTHROUGH logged", + log_has_passthrough(ctx.log_path), + str(ctx.log_path), + ) + + def check_scylla_scale(ctx: RunContext, source: str, dest: str) -> None: t0 = time.monotonic() peak, enc = poll_during(ctx, lambda: upload_part_copy(ctx, dest, source)) diff --git a/tests/integration/test_upload_part_copy_passthrough_e2e.py b/tests/integration/test_upload_part_copy_passthrough_e2e.py index 161ef7d..e598579 100644 --- a/tests/integration/test_upload_part_copy_passthrough_e2e.py +++ b/tests/integration/test_upload_part_copy_passthrough_e2e.py @@ -17,6 +17,7 @@ check_concurrent_flood, check_dek_adoption, check_reencrypt_control, + check_scylla_manifest_full_range_passthrough, check_scylla_scale, log_has_passthrough, poll_during, @@ -78,9 +79,21 @@ def test_range_copy_reencrypts_with_high_memory(self, passthrough_env): ctx.failures.clear() check_reencrypt_control(ctx, "sst/source-96mb.bin", "sst/dest-d.bin", QUICK_SIZE) assert proc.poll() is None - _assert_check(ctx, "encrypts full object") + _assert_check(ctx, "encrypts partial range") _assert_check(ctx, "high peak memory") + def test_scylla_manifest_full_range_passthrough(self, passthrough_env): + ctx, proc, _ = passthrough_env + ctx.failures.clear() + check_scylla_manifest_full_range_passthrough( + ctx, "sst/source-1280mb.bin", "sst/dest-1280mb-fullrange.bin", SCYLLA_SSTABLE_SIZE + ) + assert proc.poll() is None + _assert_check(ctx, "full-range zero bytes_encrypted") + _assert_check(ctx, "full-range low peak memory") + _assert_check(ctx, "full-range ciphertext identical") + _assert_check(ctx, "UPLOAD_PART_COPY_PASSTHROUGH logged") + def test_passthrough_log_line(self, passthrough_env): _, proc, log_path = passthrough_env assert proc.poll() is None @@ -116,17 +129,18 @@ def test_passthrough_encrypt_delta_vs_reencrypt(self, passthrough_env): peak_pt, enc_pt = poll_during( ctx, lambda: upload_part_copy(ctx, "sst/dest-compare-pt.bin", "sst/source-96mb.bin") ) + partial_end = min(QUICK_SIZE - 1, 32 * MB - 1) peak_re, enc_re = poll_during( ctx, lambda: upload_part_copy( ctx, "sst/dest-compare-re.bin", "sst/source-96mb.bin", - byte_range=f"bytes=0-{QUICK_SIZE - 1}", + byte_range=f"bytes=0-{partial_end}", ), ) assert proc.poll() is None assert enc_pt <= 1 * MB - assert enc_re >= QUICK_SIZE * 0.5 + assert enc_re >= (partial_end + 1) * 0.5 assert peak_pt <= CHUNK_PEAK * 0.5 assert peak_re >= CHUNK_PEAK * 0.5 diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py index 005e90e..d9d4989 100644 --- a/tests/unit/test_upload_part_copy_passthrough.py +++ b/tests/unit/test_upload_part_copy_passthrough.py @@ -39,13 +39,15 @@ async def _fake_client(creds): handler._client = _fake_client -def _copy_part_request(dest_path, copy_source, upload_id, part_number=1): +def _copy_part_request(dest_path, copy_source, upload_id, part_number=1, copy_source_range=None): req = MagicMock() req.url.path = dest_path req.url.query = f"uploadId={upload_id}&partNumber={part_number}" req.headers = { "x-amz-copy-source": copy_source, } + if copy_source_range is not None: + req.headers["x-amz-copy-source-range"] = copy_source_range return req @@ -274,6 +276,166 @@ async def test_range_copy_still_reencrypts(mock_s3, settings, manager, credentia assert any(c[0] == "upload_part" for c in during) +@pytest.mark.asyncio +async def test_normalize_copy_source_range_treats_full_object_as_whole( + settings, manager, credentials +): + 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 partial == f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}" + + +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() + 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) + + src_meta = MultipartMetadata( + version=2, + part_count=1, + total_plaintext_size=len(src_plaintext), + 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 + + +@pytest.mark.asyncio +async def test_scylla_manifest_full_range_uses_passthrough_not_streaming( + mock_s3, settings, manager, credentials, monkeypatch +): + """Full-object CopySourceRange must not fall into UPLOAD_PART_COPY_STREAMING.""" + 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 = (crypto.STREAMING_THRESHOLD // crypto.MAX_BUFFER_SIZE) + 2 + src_dek, src_plaintext, ciphertext_blob, src_meta = _build_large_multipart_source( + mock_s3, kid, kek, num_internal_parts=num_parts + ) + 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) + + full_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=full_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) + assert updated.dek == src_dek + part = updated.parts[1] + assert part.plaintext_size == len(src_plaintext) + assert len(part.internal_parts) == num_parts + + +@pytest.mark.asyncio +async def test_large_passthrough_not_blocked_by_pipeline_semaphore( + mock_s3, settings, manager, credentials, monkeypatch +): + """Passthrough must not queue behind the streaming copy pipeline semaphore.""" + import asyncio + + from s3proxy.handlers.multipart.copy import reset_copy_pipeline_semaphore + + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) + reset_copy_pipeline_semaphore(1) + + 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 = (crypto.STREAMING_THRESHOLD // crypto.MAX_BUFFER_SIZE) + 2 + src_dek, src_plaintext, ciphertext_blob, src_meta = _build_large_multipart_source( + mock_s3, kid, kek, num_internal_parts=num_parts + ) + + 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) + + gate = asyncio.Event() + + async def blocked_streaming(*args, **kwargs): + gate.set() + await asyncio.Event().wait() + + handler._streaming_copy_part = blocked_streaming # type: ignore[method-assign] + + passthrough_task = asyncio.create_task( + 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=f"bytes=0-{len(src_plaintext) - 1}", + ), + credentials, + ) + ) + + for _ in range(200): + if passthrough_task.done(): + break + await asyncio.sleep(0.01) + + assert passthrough_task.done() + assert not gate.is_set() + await passthrough_task + + def _extract_upload_id(xml_body: bytes) -> str: import xml.etree.ElementTree as ET