diff --git a/s3proxy/client/s3.py b/s3proxy/client/s3.py index 3b82b7c..6bc7f02 100644 --- a/s3proxy/client/s3.py +++ b/s3proxy/client/s3.py @@ -2,6 +2,8 @@ from __future__ import annotations +import hashlib +import re import time from typing import TYPE_CHECKING, Any @@ -10,12 +12,79 @@ from botocore.config import Config from structlog.stdlib import BoundLogger +from ..errors import BackendIntegrityError + if TYPE_CHECKING: from ..config import Settings from .types import S3Credentials logger: BoundLogger = structlog.get_logger(__name__) +_MD5_HEX_RE = re.compile(r"[0-9a-f]{32}") + +# Warn only once per process when the backend's ETags are not body MD5s +# (SSE-KMS buckets) and the integrity check therefore cannot run. +_etag_not_md5_logged = False + + +def _body_ciphertext_md5(body: Any) -> str | None: + """MD5 hexdigest of an uploaded body, or None if it cannot be determined.""" + if isinstance(body, (bytes, bytearray, memoryview)): + return hashlib.md5(body, usedforsecurity=False).hexdigest() + digest = getattr(body, "ciphertext_md5_hexdigest", None) + if digest is not None: + return digest() + return None + + +def verify_backend_etag( + operation: str, + bucket: str, + key: str, + response_etag: str | None, + expected_md5: str | None, + part_number: int | None = None, +) -> None: + """Compare the backend's ETag against the MD5 of the ciphertext we sent. + + Backend payload signing is disabled (UNSIGNED-PAYLOAD) because hashing the + body for SigV4 forced a second full copy of every internal part; this check + replaces it as the transport-integrity guarantee on the proxy->backend hop. + ETag == body MD5 holds for RGW/MinIO/AWS without SSE-KMS; non-MD5 ETags are + skipped (nothing to compare against). + """ + global _etag_not_md5_logged + if expected_md5 is None: + return + etag = (response_etag or "").strip('"') + if not _MD5_HEX_RE.fullmatch(etag): + if not _etag_not_md5_logged: + _etag_not_md5_logged = True + logger.warning( + "BACKEND_ETAG_NOT_MD5", + operation=operation, + bucket=bucket, + key=key, + backend_etag=etag, + ) + return + if etag != expected_md5: + logger.error( + "BACKEND_ETAG_MISMATCH", + operation=operation, + bucket=bucket, + key=key, + part_number=part_number, + expected_md5=expected_md5, + backend_etag=etag, + ) + part = f" part {part_number}" if part_number is not None else "" + raise BackendIntegrityError( + f"{operation} {bucket}/{key}{part}: " + f"backend ETag {etag} != ciphertext MD5 {expected_md5}" + ) + + # Shared session to avoid repeated JSON service model loading # See: https://github.com/boto/boto3/issues/1670 _shared_session: aioboto3.Session | None = None @@ -51,13 +120,19 @@ def __init__(self, settings: Settings, credentials: S3Credentials): """Initialize S3 client with credentials.""" self.settings = settings self.credentials = credentials + # payload_signing_enabled=False: hashing the body for SigV4 forces + # botocore to hold a second full copy of every uploaded part; integrity + # is covered by verify_backend_etag instead. when_required stops + # botocore from adding flexible checksums, which would re-read (or + # aws-chunk) the body the same way. self._config = Config( signature_version="s3v4", - s3={"addressing_style": "path"}, + s3={"addressing_style": "path", "payload_signing_enabled": False}, retries={"max_attempts": 3, "mode": "adaptive"}, max_pool_connections=100, connect_timeout=10, read_timeout=300, + request_checksum_calculation="when_required", ) self._cached_client = None self._client_context = None @@ -129,7 +204,11 @@ async def put_object( CacheControl=cache_control, Expires=expires, ) - return await self._cached_client.put_object(**kwargs) + result = await self._cached_client.put_object(**kwargs) + verify_backend_etag( + "put_object", bucket, key, result.get("ETag"), _body_ciphertext_md5(body) + ) + return result async def head_object( self, @@ -183,7 +262,7 @@ async def upload_part( key: str, upload_id: str, part_number: int, - body: bytes, + body: bytes | bytearray | Any, ) -> dict[str, Any]: """Upload a part.""" start = time.monotonic() @@ -194,6 +273,16 @@ async def upload_part( PartNumber=part_number, Body=body, ) + # The body is fully sent once the response is in, so a streaming body's + # running MD5 is complete here. + verify_backend_etag( + "upload_part", + bucket, + key, + result.get("ETag"), + _body_ciphertext_md5(body), + part_number=part_number, + ) duration = time.monotonic() - start size = len(body) size_mb = size / 1024 / 1024 diff --git a/s3proxy/errors.py b/s3proxy/errors.py index 11c3e04..59dd87e 100644 --- a/s3proxy/errors.py +++ b/s3proxy/errors.py @@ -31,6 +31,15 @@ } +class BackendIntegrityError(Exception): + """Backend-stored ciphertext ETag does not match the MD5 of what we sent. + + With backend payload signing disabled (UNSIGNED-PAYLOAD), this check is the + transport-integrity guarantee on the proxy->backend hop. Mapped to a 500 by + raise_for_exception so clients retry the part. + """ + + class S3Error(HTTPException): """S3-compatible error with proper error codes. diff --git a/tests/unit/test_backend_etag_verification.py b/tests/unit/test_backend_etag_verification.py new file mode 100644 index 0000000..b46e6f6 --- /dev/null +++ b/tests/unit/test_backend_etag_verification.py @@ -0,0 +1,126 @@ +"""Backend transport integrity: UNSIGNED-PAYLOAD + ciphertext ETag verification. + +Payload signing is disabled on the proxy->backend hop (it doubled per-request +upload memory), so the MD5-vs-ETag check in S3Client is the only transport +integrity guarantee there. These tests pin both halves: the client config that +disables signing/checksums, and the verification that replaces them. +""" + +import hashlib + +import pytest + +import s3proxy.client.s3 as client_s3 +from s3proxy.client import S3Client +from s3proxy.client.s3 import verify_backend_etag +from s3proxy.errors import BackendIntegrityError, S3Error, raise_for_exception + + +@pytest.fixture +def s3_client(settings, credentials): + return S3Client(settings, credentials) + + +class _StubBackend: + """Backend stub that returns a fixed or honest (MD5) ETag.""" + + def __init__(self, etag: str | None = None): + self.etag = etag + self.calls: list[dict] = [] + + async def _respond(self, kwargs) -> dict: + self.calls.append(kwargs) + if self.etag is not None: + return {"ETag": self.etag} + body = kwargs["Body"] + if not isinstance(body, (bytes, bytearray)): + body = b"".join([chunk async for chunk in body]) + return {"ETag": f'"{hashlib.md5(body, usedforsecurity=False).hexdigest()}"'} + + async def upload_part(self, **kwargs): + return await self._respond(kwargs) + + async def put_object(self, **kwargs): + return await self._respond(kwargs) + + +class TestBackendClientConfig: + def test_payload_signing_disabled(self, s3_client): + assert s3_client._config.s3["payload_signing_enabled"] is False + + def test_flexible_checksums_only_when_required(self, s3_client): + assert s3_client._config.request_checksum_calculation == "when_required" + + +class TestUploadPartVerification: + async def test_matching_etag_passes(self, s3_client): + s3_client._cached_client = _StubBackend() + result = await s3_client.upload_part("b", "k", "uid", 1, b"ciphertext-bytes") + assert result["ETag"].strip('"') == hashlib.md5(b"ciphertext-bytes").hexdigest() + + async def test_mismatched_etag_raises(self, s3_client): + s3_client._cached_client = _StubBackend(etag='"' + "0" * 32 + '"') + with pytest.raises(BackendIntegrityError, match="part 3"): + await s3_client.upload_part("b", "k", "uid", 3, b"ciphertext-bytes") + + async def test_non_md5_etag_skips_verification(self, s3_client, monkeypatch): + monkeypatch.setattr(client_s3, "_etag_not_md5_logged", False) + s3_client._cached_client = _StubBackend(etag='"abc123-2"') + result = await s3_client.upload_part("b", "k", "uid", 1, b"whatever") + assert result["ETag"] == '"abc123-2"' + + async def test_streaming_body_verified_via_running_md5(self, s3_client): + payload = b"frame-one" * 1000 + + class _StreamBody: + def __len__(self): + return len(payload) + + def __aiter__(self): + async def gen(): + yield payload + + return gen() + + def ciphertext_md5_hexdigest(self): + return hashlib.md5(payload, usedforsecurity=False).hexdigest() + + s3_client._cached_client = _StubBackend() + result = await s3_client.upload_part("b", "k", "uid", 1, _StreamBody()) + assert result["ETag"].strip('"') == hashlib.md5(payload).hexdigest() + + class _CorruptStreamBody(_StreamBody): + def ciphertext_md5_hexdigest(self): + return "f" * 32 + + with pytest.raises(BackendIntegrityError): + await s3_client.upload_part("b", "k", "uid", 2, _CorruptStreamBody()) + + +class TestPutObjectVerification: + async def test_matching_etag_passes(self, s3_client): + s3_client._cached_client = _StubBackend() + await s3_client.put_object("b", "k", b"object-bytes") + + async def test_mismatched_etag_raises(self, s3_client): + s3_client._cached_client = _StubBackend(etag='"' + "0" * 32 + '"') + with pytest.raises(BackendIntegrityError): + await s3_client.put_object("b", "k", b"object-bytes") + + +class TestVerifyBackendEtag: + def test_none_expected_md5_is_noop(self): + verify_backend_etag("upload_part", "b", "k", '"' + "0" * 32 + '"', None) + + def test_missing_etag_skips(self): + verify_backend_etag("upload_part", "b", "k", None, "f" * 32) + + def test_uppercase_or_quoted_etag_normalized(self): + md5 = hashlib.md5(b"x", usedforsecurity=False).hexdigest() + verify_backend_etag("upload_part", "b", "k", f'"{md5}"', md5) + + def test_maps_to_retryable_internal_error(self): + with pytest.raises(S3Error) as exc_info: + raise_for_exception(BackendIntegrityError("etag mismatch")) + assert exc_info.value.status_code == 500 + assert exc_info.value.code == "InternalError"