Disable backend payload signing; verify ciphertext MD5 against backend ETag

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

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.
 

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"