diff --git a/sdk/storage/azure-storage-blob/CHANGELOG.md b/sdk/storage/azure-storage-blob/CHANGELOG.md
index 76fc27d102a0..a499547cd109 100644
--- a/sdk/storage/azure-storage-blob/CHANGELOG.md
+++ b/sdk/storage/azure-storage-blob/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Release History
 
+## 12.30.1 (Unreleased)
+
+### Bugs Fixed
+- Fixed a bug where client-side encryption 2.0 could not detect a rearrangement of otherwise-untampered authenticated regions in blob content. This is now detected and exceptions are thrown. For data recovery purposes, this behavior can be reverted by setting the "AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS" environment variable.
+- Fixed a bug in client-side encryption where version downgrades, and other metadata tampering, was only detected at the start of a download.
+
 ## 12.30.0 (2026-06-08)
 
 ### Features Added
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_download.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_download.py
index a2f50ebc91ec..8ce2d813b432 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_download.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_download.py
@@ -27,7 +27,8 @@
     decrypt_blob,
     get_adjusted_download_range_and_offset,
     is_encryption_v2,
-    parse_encryption_data
+    parse_encryption_data,
+    _GCMRegionNonceValidator,
 )
 
 if TYPE_CHECKING:
@@ -60,7 +61,13 @@ def process_range_and_offset(
     return (start_range, end_range), (start_offset, end_offset)
 
 
-def process_content(data: Any, start_offset: int, end_offset: int, encryption: Dict[str, Any]) -> bytes:
+def process_content(
+    data: Any,
+    start_offset: int,
+    end_offset: int,
+    encryption: Dict[str, Any],
+    expected_encryption_data: Optional["_EncryptionData"],
+) -> bytes:
     if data is None:
         raise ValueError("Response cannot be None.")
 
@@ -76,6 +83,8 @@ def process_content(data: Any, start_offset: int, end_offset: int, encryption: D
                 start_offset,
                 end_offset,
                 data.response.headers,
+                expected_encryption_data,
+                encryption.get("gcm_nonce_validator"),
             )
         except Exception as error:
             raise HttpResponseError(message="Decryption failed.", response=data.response, error=error) from error
@@ -234,7 +243,9 @@ def _download_chunk(self, chunk_start: int, chunk_end: int) -> Tuple[bytes, int]
                     process_storage_error(error)
 
                 try:
-                    chunk_data = process_content(response, offset[0], offset[1], self.encryption_options)
+                    chunk_data = process_content(
+                        response, offset[0], offset[1], self.encryption_options, self.encryption_data
+                    )
                     retry_active = False
                 except (IncompleteReadError, HttpResponseError, DecodeError, ServiceResponseError) as error:
                     retry_total -= 1
@@ -380,6 +391,8 @@ def __init__(
 
         if self._encryption_options.get("key") is not None or self._encryption_options.get("resolver") is not None:
             self._get_encryption_data_request()
+            if is_encryption_v2(self._encryption_data):
+                self._encryption_options["gcm_nonce_validator"] = _GCMRegionNonceValidator()
 
         # The service only provides transactional MD5s for chunks under 4MB.
         # If validate_content is on, get only self.MAX_CHUNK_GET_SIZE for the first
@@ -520,7 +533,8 @@ def _initial_request(self):
                         response,
                         self._initial_offset[0],
                         self._initial_offset[1],
-                        self._encryption_options
+                        self._encryption_options,
+                        self._encryption_data,
                     )
                 retry_active = False
             except (IncompleteReadError, HttpResponseError, DecodeError, ServiceResponseError) as error:
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_encryption.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_encryption.py
index 2153d1da1da6..5c4832a4b672 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_encryption.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_encryption.py
@@ -8,6 +8,7 @@
 import math
 import os
 import sys
+import threading
 import warnings
 from collections import OrderedDict
 from io import BytesIO
@@ -52,6 +53,13 @@
 _ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION = (
     "The require_encryption flag is set, but encryption is not supported for this method."
 )
+_ERROR_ENCRYPTION_METADATA_MISMATCH = (
+    "The encryption metadata in the download response does not match the encryption metadata "
+    "retrieved at the start of the download. The blob's encryption metadata may have been modified "
+    "while the download was in progress."
+)
+
+_ERROR_INVALID_ENCRYPTION_METADATA = "The encryption metadata is not valid and may have been modified."
 
 
 class KeyEncryptionKey(Protocol):
@@ -216,6 +224,49 @@ def __init__(
         self.wrapped_content_key = wrapped_content_key
         self.key_wrapping_metadata = key_wrapping_metadata
 
+    def matches(self, other: "_EncryptionData") -> bool:
+        """
+        Determines whether this encryption data refers to the same encrypted content by comparing
+        every field that affects decryption. This is used to detect whether a blob's encryption
+        metadata has changed partway through a download, which could indicate the blob was
+        overwritten or tampered with.
+
+        :param _EncryptionData other: The encryption data to compare against.
+        :return: True if the decryption-relevant metadata matches, False otherwise.
+        :rtype: bool
+        """
+        if (
+            self.encryption_agent.protocol != other.encryption_agent.protocol
+            or self.encryption_agent.encryption_algorithm != other.encryption_agent.encryption_algorithm
+        ):
+            return False
+
+        if (
+            self.wrapped_content_key.key_id != other.wrapped_content_key.key_id
+            or self.wrapped_content_key.algorithm != other.wrapped_content_key.algorithm
+            or self.wrapped_content_key.encrypted_key != other.wrapped_content_key.encrypted_key
+        ):
+            return False
+
+        # Compare the content encryption IV (used for AES-CBC / V1).
+        if self.content_encryption_IV != other.content_encryption_IV:
+            return False
+
+        # Compare the encrypted region info (used for AES-GCM / V2).
+        self_region = self.encrypted_region_info
+        other_region = other.encrypted_region_info
+        if (self_region is None) != (other_region is None):
+            return False
+        if self_region is not None and other_region is not None:
+            if (
+                self_region.data_length != other_region.data_length
+                or self_region.nonce_length != other_region.nonce_length
+                or self_region.tag_length != other_region.tag_length
+            ):
+                return False
+
+        return True
+
 
 class GCMBlobEncryptionStream:
     """
@@ -665,7 +716,7 @@ def _validate_and_unwrap_cek(
         version_2_bytes = encryption_data.encryption_agent.protocol.encode().ljust(8, b"\0")
         cek_version_bytes = content_encryption_key[: len(version_2_bytes)]
         if cek_version_bytes != version_2_bytes:
-            raise ValueError("The encryption metadata is not valid and may have been modified.")
+            raise ValueError(_ERROR_INVALID_ENCRYPTION_METADATA)
 
         # Remove version from the start of the cek.
         content_encryption_key = content_encryption_key[len(version_2_bytes) :]
@@ -839,7 +890,96 @@ def generate_blob_encryption_data(
     return content_encryption_key, initialization_vector, encryption_data
 
 
-def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
+def _parse_content_range(content_range: str) -> Tuple[int, int, int]:
+    """
+    Parses a Content-Range header of the form 'bytes x-y/size' into its
+    start, end, and total size components.
+
+    :param str content_range: The Content-Range header value.
+    :return: A tuple of (start, end, total size).
+    :rtype: Tuple[int, int, int]
+    """
+    # Format: 'bytes x-y/size' -- ignore the leading 'bytes' word.
+    byte_range, size = content_range.split(" ")[1].split("/")
+    start, end = byte_range.split("-")
+    return int(start), int(end), int(size)
+
+
+def _region_nonce_encodings(nonce_length: int) -> Dict[str, Callable[[int], bytes]]:
+    """
+    Returns the supported per-region nonce encodings, keyed by the SDK that produces them.
+
+    The per-region nonce is a counter of the region's position, but each SDK encodes it
+    differently, so all supported encodings must be understood for interoperability:
+
+    * Python: zero-based counter, big-endian across the whole nonce (value in trailing bytes).
+    * Java: zero-based counter, big-endian in the leading 8 bytes, trailing bytes zeroed.
+    * .NET: one-based counter, little-endian in the trailing 8 bytes, leading bytes zeroed.
+
+    These encodings share the same value space, so they must not be accepted independently
+    per region (for example Java's nonce for region 1 is identical to .NET's nonce for
+    region 16,777,215). A single encoding is instead selected and enforced across the whole
+    download; see ``decrypt_blob``.
+
+    :param int nonce_length: The length of the nonce in bytes.
+    :return: A mapping of SDK name to a function returning that SDK's nonce for a region index.
+    :rtype: Dict[str, Callable[[int], bytes]]
+    """
+    encodings: Dict[str, Callable[[int], bytes]] = {
+        "python": lambda index: index.to_bytes(nonce_length, "big"),
+    }
+
+    counter_length = 8
+    pad = nonce_length - counter_length
+    encodings["java"] = lambda index: index.to_bytes(counter_length, "big") + b"\x00" * pad
+    encodings["dotnet"] = lambda index: b"\x00" * pad + (index + 1).to_bytes(counter_length, "little")
+
+    return encodings
+
+
+class _GCMRegionNonceValidator:
+    """
+    Enforces that every region across a whole download uses a single nonce encoding.
+
+    ``decrypt_blob`` runs once per HTTP chunk, so the candidate encodings are shared and
+    intersected across all chunks (including concurrent ones) rather than reset per call.
+    Otherwise the encoding could change at a chunk boundary and, at an encoding collision,
+    let a relocated region pass validation.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._candidates: Optional[Dict[str, Callable[[int], bytes]]] = None
+        # Set once the candidates collapse to a single encoding; read lock-free thereafter.
+        self._encoding: Optional[Callable[[int], bytes]] = None
+
+    def validate_region(self, region_index: int, nonce: bytes, nonce_length: int) -> None:
+        """
+        Narrows the shared candidate encodings to those consistent with this region.
+
+        :param int region_index: The zero-based index of the region within the blob.
+        :param bytes nonce: The nonce read from the region.
+        :param int nonce_length: The length of the nonce in bytes.
+        """
+        encoding = self._encoding
+        if encoding is not None:
+            if encoding(region_index) != nonce:
+                raise ValueError(_ERROR_INVALID_ENCRYPTION_METADATA)
+            return
+
+        with self._lock:
+            if self._candidates is None:
+                self._candidates = _region_nonce_encodings(nonce_length)
+            self._candidates = {
+                name: encode for name, encode in self._candidates.items() if encode(region_index) == nonce
+            }
+            if not self._candidates:
+                raise ValueError(_ERROR_INVALID_ENCRYPTION_METADATA)
+            if len(self._candidates) == 1:
+                self._encoding = next(iter(self._candidates.values()))
+
+
+def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements,too-many-branches
     require_encryption: bool,
     key_encryption_key: Optional[KeyEncryptionKey],
     key_resolver: Optional[Callable[[str], KeyEncryptionKey]],
@@ -847,6 +987,8 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
     start_offset: int,
     end_offset: int,
     response_headers: Dict[str, Any],
+    expected_encryption_data: Optional[_EncryptionData] = None,
+    nonce_validator: Optional["_GCMRegionNonceValidator"] = None,
 ) -> bytes:
     """
     Decrypts the given blob contents and returns only the requested range.
@@ -874,12 +1016,22 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
     :param Dict[str, Any] response_headers:
         A dictionary of response headers from the download request. Expected to include the
         'x-ms-meta-encryptiondata' header if the blob was encrypted.
+    :param Optional[_EncryptionData] expected_encryption_data:
+        The encryption data retrieved at the start of the download. If provided, the encryption
+        metadata on this response is validated against it to detect the blob's encryption metadata
+        being modified (tampered with) partway through a download.
+    :param Optional[_GCMRegionNonceValidator] nonce_validator:
+        Shared state used to enforce a single V2 nonce encoding across every chunk of a Blob download.
+        Required for V2 decryption unless the caller has set the
+        AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS environment variable.
     :return: The decrypted blob content.
     :rtype: bytes
     """
     try:
         encryption_data = _dict_to_encryption_data(loads(response_headers["x-ms-meta-encryptiondata"]))
     except Exception as exc:  # pylint: disable=broad-except
+        if expected_encryption_data is not None:
+            raise ValueError(_ERROR_ENCRYPTION_METADATA_MISMATCH) from exc
         if require_encryption:
             raise ValueError(
                 "Encryption required, but received data does not contain appropriate metadata."
@@ -888,6 +1040,10 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
 
         return content
 
+    # Validate that the encryption metadata has not changed since the start of the download.
+    if expected_encryption_data is not None and not expected_encryption_data.matches(encryption_data):
+        raise ValueError(_ERROR_ENCRYPTION_METADATA_MISMATCH)
+
     algorithm = encryption_data.encryption_agent.encryption_algorithm
     if algorithm not in (_EncryptionAlgorithm.AES_CBC_256, _EncryptionAlgorithm.AES_GCM_256):
         raise ValueError("Specified encryption algorithm is not supported.")
@@ -904,16 +1060,7 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
         iv: Optional[bytes] = None
         unpad = False
         if "content-range" in response_headers:
-            content_range = response_headers["content-range"]
-            # Format: 'bytes x-y/size'
-
-            # Ignore the word 'bytes'
-            content_range = content_range.split(" ")
-
-            content_range = content_range[1].split("-")
-            content_range = content_range[1].split("/")
-            end_range = int(content_range[0])
-            blob_size = int(content_range[1])
+            _, end_range, blob_size = _parse_content_range(response_headers["content-range"])
 
             if start_offset >= 16:
                 iv = content[:16]
@@ -957,6 +1104,25 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
         tag_length = encryption_data.encrypted_region_info.tag_length
         region_length = nonce_length + data_length + tag_length
 
+        # The per-region nonce is a counter of the region's index within the blob. The
+        # downloaded content always begins on a region boundary, so derive the first
+        # region's index from the download range (0 when the whole blob was downloaded).
+        # This lets us validate each nonce and detect reordered regions.
+        start_range = 0
+        if "content-range" in response_headers:
+            start_range, _, _ = _parse_content_range(response_headers["content-range"])
+        nonce_counter = start_range // region_length
+
+        # Bypass nonce validation via an environment variable for data-recovery scenarios
+        # where regions were reordered. Not recommended: it can allow tampered data through.
+        validate_nonce = os.environ.get(
+            "AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS", ""
+        ).strip().lower() not in ("true", "1")
+
+        # A validator is required to enforce a single nonce encoding across the whole download.
+        if validate_nonce and nonce_validator is None:
+            raise ValueError("A nonce validator is required to decrypt Encryption V2 content.")
+
         decrypted_content = bytearray()
         while offset < total_size:
             # Process one encryption region at a time
@@ -965,6 +1131,10 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
 
             # First bytes are the nonce
             nonce = encrypted_region[:nonce_length]
+            # Validate the nonce matches the expected counter for this region under a single
+            # consistent encoding. A mismatch indicates the regions were reordered or tampered with.
+            if nonce_validator is not None and validate_nonce:
+                nonce_validator.validate_region(nonce_counter, nonce, nonce_length)
             ciphertext_with_tag = encrypted_region[nonce_length:]
 
             aesgcm = AESGCM(content_encryption_key)
@@ -972,6 +1142,7 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
             decrypted_content.extend(decrypted_data)
 
             offset += process_size
+            nonce_counter += 1
 
         # Read the caller requested data from the decrypted content
         return decrypted_content[start_offset:end_offset]
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_version.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_version.py
index f2d1dea9d438..42b819748f88 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/_version.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/_version.py
@@ -4,4 +4,4 @@
 # license information.
 # --------------------------------------------------------------------------
 
-VERSION = "12.30.0"
+VERSION = "12.30.1"
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/_download_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/_download_async.py
index 30cbb0c68fbf..b69b6b28501a 100644
--- a/sdk/storage/azure-storage-blob/azure/storage/blob/aio/_download_async.py
+++ b/sdk/storage/azure-storage-blob/azure/storage/blob/aio/_download_async.py
@@ -30,7 +30,8 @@
     adjust_blob_size_for_encryption,
     decrypt_blob,
     is_encryption_v2,
-    parse_encryption_data
+    parse_encryption_data,
+    _GCMRegionNonceValidator,
 )
 
 if TYPE_CHECKING:
@@ -44,7 +45,13 @@
 T = TypeVar('T', bytes, str)
 
 
-async def process_content(data: Any, start_offset: int, end_offset: int, encryption: Dict[str, Any]) -> bytes:
+async def process_content(
+    data: Any,
+    start_offset: int,
+    end_offset: int,
+    encryption: Dict[str, Any],
+    expected_encryption_data: Optional["_EncryptionData"],
+) -> bytes:
     if data is None:
         raise ValueError("Response cannot be None.")
     if hasattr(data.response, "is_stream_consumed") and data.response.is_stream_consumed:
@@ -60,7 +67,9 @@ async def process_content(data: Any, start_offset: int, end_offset: int, encrypt
                 content,
                 start_offset,
                 end_offset,
-                data.response.headers
+                data.response.headers,
+                expected_encryption_data,
+                encryption.get("gcm_nonce_validator"),
             )
         except Exception as error:
             raise HttpResponseError(
@@ -143,7 +152,9 @@ async def _download_chunk(self, chunk_start: int, chunk_end: int) -> Tuple[bytes
                     process_storage_error(error)
 
                 try:
-                    chunk_data = await process_content(response, offset[0], offset[1], self.encryption_options)
+                    chunk_data = await process_content(
+                        response, offset[0], offset[1], self.encryption_options, self.encryption_data
+                    )
                     retry_active = False
                 except (IncompleteReadError, HttpResponseError, DecodeError, ServiceResponseError) as error:
                     retry_total -= 1
@@ -316,6 +327,8 @@ async def _get_encryption_data_request(self) -> None:
     async def _setup(self) -> None:
         if self._encryption_options.get("key") is not None or self._encryption_options.get("resolver") is not None:
             await self._get_encryption_data_request()
+            if is_encryption_v2(self._encryption_data):
+                self._encryption_options["gcm_nonce_validator"] = _GCMRegionNonceValidator()
 
         # The service only provides transactional MD5s for chunks under 4MB.
         # If validate_content is on, get only self.MAX_CHUNK_GET_SIZE for the first
@@ -430,7 +443,8 @@ async def _initial_request(self):
                         response,
                         self._initial_offset[0],
                         self._initial_offset[1],
-                        self._encryption_options
+                        self._encryption_options,
+                        self._encryption_data,
                     )
                 retry_active = False
             except (IncompleteReadError, HttpResponseError, DecodeError, ServiceResponseError) as error:
diff --git a/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2.py b/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2.py
index f09f212ad067..d455f5196516 100644
--- a/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2.py
+++ b/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2.py
@@ -22,9 +22,15 @@
 
 from azure.core import MatchConditions
 from azure.core.exceptions import HttpResponseError, ResourceExistsError
-from azure.storage.blob import BlobServiceClient, BlobType, ContentSettings
+from azure.storage.blob import BlobBlock, BlobServiceClient, BlobType, ContentSettings
 from azure.storage.blob._encryption import (
-    _dict_to_encryption_data, _GCM_NONCE_LENGTH, _GCM_TAG_LENGTH, _validate_and_unwrap_cek,
+    _dict_to_encryption_data,
+    _GCM_NONCE_LENGTH,
+    _GCM_TAG_LENGTH,
+    _GCMRegionNonceValidator,
+    _region_nonce_encodings,
+    _validate_and_unwrap_cek,
+    decrypt_blob,
 )
 
 
@@ -385,6 +391,55 @@ def test_encryption_v2_v1_downgrade(self, **kwargs):
 
         assert 'Decryption failed.' in str(e.value)
 
+    @pytest.mark.live_test_only
+    @BlobPreparer()
+    def test_encryption_v2_v1_downgrade_mid_download(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        self._setup(storage_account_name, storage_account_key)
+        kek = KeyWrapper("key1")
+        bsc = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=storage_account_key.secret,
+            max_single_get_size=4 * MiB,
+            max_chunk_get_size=4 * MiB,
+            require_encryption=True,
+            encryption_version="2.0",
+            key_encryption_key=kek,
+        )
+
+        blob = bsc.get_blob_client(self.container_name, self._get_blob_reference())
+        content = b"abcd" * 2 * MiB  # 8 MiB spans multiple encryption regions -> multiple download requests
+
+        blob.upload_blob(content, overwrite=True)
+
+        # Build tampered metadata that downgrades the blob from V2 to V1
+        metadata = blob.get_blob_properties().metadata
+        tampered = loads(metadata["encryptiondata"])
+        tampered["EncryptionAgent"]["Protocol"] = "1.0"
+        tampered["EncryptionAgent"]["EncryptionAlgorithm"] = "AES_CBC_256"
+        tampered["ContentEncryptionIV"] = base64.b64encode(os.urandom(16)).decode("utf-8")
+        tampered_header = dumps(tampered)
+
+        # Simulate the service returning downgraded (V1) encryption metadata partway through the
+        # download by tampering with the response headers of every request after the initial one.
+        from azure.storage.blob import _download as download_module
+
+        real_process_content = download_module.process_content
+        call_count = {"value": 0}
+
+        def tampering_process_content(data, start_offset, end_offset, encryption, expected_encryption_data):
+            call_count["value"] += 1
+            if call_count["value"] > 1:
+                data.response.headers["x-ms-meta-encryptiondata"] = tampered_header
+            return real_process_content(data, start_offset, end_offset, encryption, expected_encryption_data)
+
+        # Act / Assert
+        with mock.patch.object(download_module, "process_content", tampering_process_content):
+            with pytest.raises(HttpResponseError) as e:
+                blob.download_blob().readall()
+
     @BlobPreparer()
     @recorded_by_proxy
     @mock.patch('os.urandom', mock_urandom)
@@ -418,6 +473,45 @@ def test_encryption_modify_cek(self, **kwargs):
 
         assert 'Decryption failed.' in str(e.value)
 
+    @pytest.mark.live_test_only
+    @BlobPreparer()
+    def test_encryption_reordered_regions(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        self._setup(storage_account_name, storage_account_key)
+        kek = KeyWrapper("key1")
+        # Each encrypted region is the plaintext region plus a nonce and tag. Size each
+        # block to a full encrypted region so every committed block is exactly one region.
+        region_length = 4 * MiB + _GCM_NONCE_LENGTH + _GCM_TAG_LENGTH
+        bsc = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=storage_account_key.secret,
+            max_single_put_size=1024,
+            max_block_size=region_length,
+            require_encryption=True,
+            encryption_version="2.0",
+            key_encryption_key=kek,
+        )
+
+        blob = bsc.get_blob_client(self.container_name, self._get_blob_reference())
+        content = b"abcd" * 3 * MiB  # 12 MiB -- three full 4 MiB encryption regions
+        blob.upload_blob(content, overwrite=True)
+
+        # Reorder the committed blocks so the encryption regions are out of order.
+        plain_blob = self.bsc.get_blob_client(self.container_name, self._get_blob_reference())
+        metadata = plain_blob.get_blob_properties().metadata
+        committed, _ = plain_blob.get_block_list(block_list_type="committed")
+        reordered = committed[:-2] + committed[-1:] + committed[-2:-1]
+        reordered = [BlobBlock(block_id=block.id) for block in reordered]
+        plain_blob.commit_block_list(reordered, metadata=metadata)
+
+        # Act / Assert -- a region's nonce no longer matches its position
+        with pytest.raises(HttpResponseError) as e:
+            blob.download_blob().readall()
+
+        assert "Decryption failed." in str(e.value)
+
     @BlobPreparer()
     @recorded_by_proxy
     @mock.patch('os.urandom', mock_urandom)
@@ -1236,3 +1330,171 @@ def assert_user_agent(request):
 
         blob.upload_blob(content, overwrite=True, raw_request_hook=assert_user_agent)
         blob.download_blob(raw_request_hook=assert_user_agent).readall()
+
+
+class TestGCMRegionNonceValidation:
+    REGION_DATA_LENGTH = 32
+
+    @staticmethod
+    def _encryption_headers(kek, cek, protocol, library, data_length=REGION_DATA_LENGTH):
+        # Wrap the CEK the way the V2 protocol requires (version prefix padded to 8 bytes).
+        wrapped_cek = kek.wrap_key(protocol.encode().ljust(8, b"\x00") + cek)
+        encryption_data = {
+            "WrappedContentKey": {
+                "KeyId": kek.get_kid(),
+                "EncryptedKey": base64.b64encode(wrapped_cek).decode(),
+                "Algorithm": kek.get_key_wrap_algorithm(),
+            },
+            "EncryptionAgent": {"Protocol": protocol, "EncryptionAlgorithm": "AES_GCM_256"},
+            "EncryptedRegionInfo": {"DataLength": data_length, "NonceLength": _GCM_NONCE_LENGTH},
+            "KeyWrappingMetadata": {"EncryptionLibrary": library},
+        }
+        return {"x-ms-meta-encryptiondata": dumps(encryption_data)}
+
+    @staticmethod
+    def _encrypt_regions(cek, nonce_for_region, plaintext_regions):
+        aesgcm = AESGCM(cek)
+        return [
+            nonce_for_region(i) + aesgcm.encrypt(nonce_for_region(i), region, None)
+            for i, region in enumerate(plaintext_regions)
+        ]
+
+    @staticmethod
+    def _decrypt(kek, headers, content, end_offset, nonce_validator):
+        return decrypt_blob(
+            require_encryption=True,
+            key_encryption_key=kek,
+            key_resolver=None,
+            content=content,
+            start_offset=0,
+            end_offset=end_offset,
+            response_headers=headers,
+            nonce_validator=nonce_validator,
+        )
+
+    def test_decrypt_dotnet_v2_1_nonce_encoding(self):
+        # Regression test for cross-SDK interoperability. The .NET Storage SDK encodes each
+        # region's GCM nonce as a one-based counter written little-endian into the final 8
+        # nonce bytes. Python must still decrypt these .NET-produced V2.1 blobs while
+        # continuing to detect reordered regions.
+        kek = KeyWrapper("key1")
+        cek = os.urandom(32)
+        num_regions = 3
+        plaintext_regions = [bytes([i]) * self.REGION_DATA_LENGTH for i in range(num_regions)]
+
+        def dotnet_nonce(region_index):
+            # 4 zero bytes + one-based counter, little-endian, 8 bytes -- see .NET
+            # GcmAuthenticatedCryptographicTransform.GetNewNonce().
+            return b"\x00\x00\x00\x00" + (region_index + 1).to_bytes(8, "little")
+
+        encrypted_regions = self._encrypt_regions(cek, dotnet_nonce, plaintext_regions)
+        headers = self._encryption_headers(kek, cek, "2.1", "Dotnet")
+        plaintext = b"".join(plaintext_regions)
+
+        # Act / Assert -- the .NET nonce encoding is accepted and the content round-trips.
+        decrypted = self._decrypt(kek, headers, b"".join(encrypted_regions), len(plaintext), _GCMRegionNonceValidator())
+        assert decrypted == plaintext
+
+        # Reordering the .NET-produced regions must still be detected.
+        reordered = encrypted_regions[0] + encrypted_regions[2] + encrypted_regions[1]
+        with pytest.raises(ValueError):
+            self._decrypt(kek, headers, reordered, len(plaintext), _GCMRegionNonceValidator())
+
+    def test_decrypt_java_v2_nonce_encoding(self):
+        # Regression test for cross-SDK interoperability. The Java Storage SDK encodes each
+        # region's GCM nonce as a zero-based counter written big-endian into the leading 8
+        # nonce bytes (ByteBuffer.allocate(12).putLong(index)), which differs from Python's
+        # full-width big-endian counter. Python must still decrypt Java-produced V2 blobs
+        # while continuing to detect reordered regions.
+        kek = KeyWrapper("key1")
+        cek = os.urandom(32)
+        num_regions = 3
+        plaintext_regions = [bytes([i]) * self.REGION_DATA_LENGTH for i in range(num_regions)]
+
+        def java_nonce(region_index):
+            # Zero-based counter, big-endian, in the leading 8 bytes; trailing bytes zeroed.
+            return region_index.to_bytes(8, "big") + b"\x00" * (_GCM_NONCE_LENGTH - 8)
+
+        encrypted_regions = self._encrypt_regions(cek, java_nonce, plaintext_regions)
+        headers = self._encryption_headers(kek, cek, "2.0", "Java")
+        plaintext = b"".join(plaintext_regions)
+
+        # Act / Assert -- the Java nonce encoding is accepted and the content round-trips.
+        decrypted = self._decrypt(kek, headers, b"".join(encrypted_regions), len(plaintext), _GCMRegionNonceValidator())
+        assert decrypted == plaintext
+
+        # Reordering the Java-produced regions must still be detected.
+        reordered = encrypted_regions[0] + encrypted_regions[2] + encrypted_regions[1]
+        with pytest.raises(ValueError):
+            self._decrypt(kek, headers, reordered, len(plaintext), _GCMRegionNonceValidator())
+
+    def test_decrypt_rejects_mixed_nonce_encodings(self):
+        # Regression test: the supported SDK nonce encodings share a value space, so accepting
+        # them independently per region would weaken reorder detection. For example Java's
+        # nonce for region 1 is identical to .NET's nonce for region 16,777,215, so a Java
+        # region could be moved to that position and still pass a per-region union check.
+        # decrypt_blob must instead select a single encoding and enforce it consistently.
+        encodings = _region_nonce_encodings(_GCM_NONCE_LENGTH)
+        # Document the overlap that motivates single-encoding enforcement.
+        assert encodings["java"](1) == encodings["dotnet"](16_777_215)
+
+        kek = KeyWrapper("key1")
+        cek = os.urandom(32)
+        aesgcm = AESGCM(cek)
+
+        # Region 0 uses the Java/Python encoding (all zeros); region 1 uses the .NET encoding.
+        # A per-region union check would accept both; single-encoding enforcement rejects the mix.
+        region0_nonce = encodings["java"](0)
+        region1_nonce = encodings["dotnet"](1)
+        region0 = region0_nonce + aesgcm.encrypt(region0_nonce, b"\x00" * self.REGION_DATA_LENGTH, None)
+        region1 = region1_nonce + aesgcm.encrypt(region1_nonce, b"\x11" * self.REGION_DATA_LENGTH, None)
+        headers = self._encryption_headers(kek, cek, "2.0", "Mixed")
+
+        # Act / Assert -- the mixed encoding is rejected rather than silently accepted.
+        with pytest.raises(ValueError):
+            self._decrypt(kek, headers, region0 + region1, 2 * self.REGION_DATA_LENGTH, _GCMRegionNonceValidator())
+
+    def test_nonce_validator_enforces_single_encoding_across_chunks(self):
+        # decrypt_blob runs once per download chunk, so a shared validator must intersect the
+        # candidate encodings across chunks; otherwise the encoding could change at a chunk
+        # boundary and, at a collision, let a relocated region pass.
+        encodings = _region_nonce_encodings(_GCM_NONCE_LENGTH)
+        # The collision that makes per-chunk validation unsound: Java region 1 == .NET region 16,777,215.
+        assert encodings["java"](1) == encodings["dotnet"](16_777_215)
+
+        validator = _GCMRegionNonceValidator()
+        # First chunk: two Java regions resolve the encoding to Java.
+        validator.validate_region(0, encodings["java"](0), _GCM_NONCE_LENGTH)
+        validator.validate_region(1, encodings["java"](1), _GCM_NONCE_LENGTH)
+
+        # Later chunk: a region relocated to the colliding .NET index carries Java's region-1
+        # nonce. Its only consistent encoding is .NET, which conflicts with the resolved Java
+        # encoding, so the shared validator rejects it.
+        with pytest.raises(ValueError):
+            validator.validate_region(16_777_215, encodings["java"](1), _GCM_NONCE_LENGTH)
+
+    def test_env_var_bypasses_nonce_validation(self):
+        # The AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS escape hatch disables nonce
+        # validation for data-recovery scenarios. With it set, a mix of nonce encodings that
+        # would normally be rejected must decrypt, and no validator is required.
+        encodings = _region_nonce_encodings(_GCM_NONCE_LENGTH)
+        kek = KeyWrapper("key1")
+        cek = os.urandom(32)
+        aesgcm = AESGCM(cek)
+
+        # Two regions using incompatible encodings (Java for region 0, .NET for region 1).
+        region0_plaintext = b"\x00" * self.REGION_DATA_LENGTH
+        region1_plaintext = b"\x11" * self.REGION_DATA_LENGTH
+        region0_nonce = encodings["java"](0)
+        region1_nonce = encodings["dotnet"](1)
+        region0 = region0_nonce + aesgcm.encrypt(region0_nonce, region0_plaintext, None)
+        region1 = region1_nonce + aesgcm.encrypt(region1_nonce, region1_plaintext, None)
+        headers = self._encryption_headers(kek, cek, "2.0", "Mixed")
+        plaintext = region0_plaintext + region1_plaintext
+
+        # Act / Assert -- with the bypass set, decryption succeeds without a validator.
+        with mock.patch.dict(os.environ, {"AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS": "true"}):
+            decrypted = self._decrypt(
+                kek, headers, region0 + region1, 2 * self.REGION_DATA_LENGTH, nonce_validator=None
+            )
+        assert decrypted == plaintext
diff --git a/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2_async.py b/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2_async.py
index e74482f65e26..6b1097cb9f77 100644
--- a/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2_async.py
+++ b/sdk/storage/azure-storage-blob/tests/test_blob_encryption_v2_async.py
@@ -23,7 +23,7 @@
 
 from azure.core import MatchConditions
 from azure.core.exceptions import HttpResponseError, ResourceExistsError
-from azure.storage.blob import BlobType, ContentSettings
+from azure.storage.blob import BlobBlock, BlobType, ContentSettings
 from azure.storage.blob._encryption import (
     _dict_to_encryption_data,
     _GCM_NONCE_LENGTH,
@@ -392,6 +392,55 @@ async def test_encryption_v2_v1_downgrade(self, **kwargs):
 
         assert 'Decryption failed.' in str(e.value)
 
+    @pytest.mark.live_test_only
+    @BlobPreparer()
+    async def test_encryption_v2_v1_downgrade_mid_download(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        await self._setup(storage_account_name, storage_account_key)
+        kek = KeyWrapper("key1")
+        bsc = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=storage_account_key.secret,
+            max_single_get_size=4 * MiB,
+            max_chunk_get_size=4 * MiB,
+            require_encryption=True,
+            encryption_version="2.0",
+            key_encryption_key=kek,
+        )
+
+        blob = bsc.get_blob_client(self.container_name, self._get_blob_reference())
+        content = b"abcd" * 2 * MiB  # 8 MiB spans multiple encryption regions -> multiple download requests
+
+        await blob.upload_blob(content, overwrite=True)
+
+        # Build tampered metadata that downgrades the blob from V2 to V1
+        metadata = (await blob.get_blob_properties()).metadata
+        tampered = loads(metadata["encryptiondata"])
+        tampered["EncryptionAgent"]["Protocol"] = "1.0"
+        tampered["EncryptionAgent"]["EncryptionAlgorithm"] = "AES_CBC_256"
+        tampered["ContentEncryptionIV"] = base64.b64encode(os.urandom(16)).decode("utf-8")
+        tampered_header = dumps(tampered)
+
+        # Simulate the service returning downgraded (V1) encryption metadata partway through the
+        # download by tampering with the response headers of every request after the initial one.
+        from azure.storage.blob.aio import _download_async as download_module
+
+        real_process_content = download_module.process_content
+        call_count = {"value": 0}
+
+        async def tampering_process_content(data, start_offset, end_offset, encryption, expected_encryption_data):
+            call_count["value"] += 1
+            if call_count["value"] > 1:
+                data.response.headers["x-ms-meta-encryptiondata"] = tampered_header
+            return await real_process_content(data, start_offset, end_offset, encryption, expected_encryption_data)
+
+        # Act / Assert
+        with mock.patch.object(download_module, "process_content", tampering_process_content):
+            with pytest.raises(HttpResponseError) as e:
+                await (await blob.download_blob()).readall()
+
     @BlobPreparer()
     @recorded_by_proxy_async
     async def test_encryption_modify_cek(self, **kwargs):
@@ -425,6 +474,45 @@ async def test_encryption_modify_cek(self, **kwargs):
 
         assert 'Decryption failed.' in str(e.value)
 
+    @pytest.mark.live_test_only
+    @BlobPreparer()
+    async def test_encryption_reordered_regions(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        await self._setup(storage_account_name, storage_account_key)
+        kek = KeyWrapper("key1")
+        # Each encrypted region is the plaintext region plus a nonce and tag. Size each
+        # block to a full encrypted region so every committed block is exactly one region.
+        region_length = 4 * MiB + _GCM_NONCE_LENGTH + _GCM_TAG_LENGTH
+        bsc = BlobServiceClient(
+            self.account_url(storage_account_name, "blob"),
+            credential=storage_account_key.secret,
+            max_single_put_size=1024,
+            max_block_size=region_length,
+            require_encryption=True,
+            encryption_version="2.0",
+            key_encryption_key=kek,
+        )
+
+        blob = bsc.get_blob_client(self.container_name, self._get_blob_reference())
+        content = b"abcd" * 3 * MiB  # 12 MiB -- three full 4 MiB encryption regions
+        await blob.upload_blob(content, overwrite=True)
+
+        # Reorder the committed blocks so the encryption regions are out of order.
+        plain_blob = self.bsc.get_blob_client(self.container_name, self._get_blob_reference())
+        metadata = (await plain_blob.get_blob_properties()).metadata
+        committed, _ = await plain_blob.get_block_list(block_list_type="committed")
+        reordered = committed[:-2] + committed[-1:] + committed[-2:-1]
+        reordered = [BlobBlock(block_id=block.id) for block in reordered]
+        await plain_blob.commit_block_list(reordered, metadata=metadata)
+
+        # Act / Assert -- a region's nonce no longer matches its position
+        with pytest.raises(HttpResponseError) as e:
+            await (await blob.download_blob()).readall()
+
+        assert "Decryption failed." in str(e.value)
+
     @BlobPreparer()
     @recorded_by_proxy_async
     async def test_case_insensitive_metadata_key(self, **kwargs):
diff --git a/sdk/storage/azure-storage-blob/tests/test_retry.py b/sdk/storage/azure-storage-blob/tests/test_retry.py
index 68bbce9db121..8d16b77445e6 100644
--- a/sdk/storage/azure-storage-blob/tests/test_retry.py
+++ b/sdk/storage/azure-storage-blob/tests/test_retry.py
@@ -666,14 +666,14 @@ def test_retry_on_service_response_error(self, **kwargs):
         # Mock the internal response to raise ServiceResponseError on first chunk processing
         from azure.storage.blob._download import process_content as real_process_content
 
-        def mock_process_content_with_error(response, start_offset, end_offset, encryption):
+        def mock_process_content_with_error(response, start_offset, end_offset, encryption, expected_encryption_data):
             retry_counter.simple_count(retry)
             conn_error = AzureError("Connection reset by peer")
             if retry_counter.count == 1:
                 raise ServiceResponseError(conn_error, error=conn_error)
             if retry_counter.count == 2:
                 raise ServiceResponseTimeoutError(conn_error, error=conn_error)
-            return real_process_content(response, start_offset, end_offset, encryption)
+            return real_process_content(response, start_offset, end_offset, encryption, expected_encryption_data)
 
         # Act
         try:
diff --git a/sdk/storage/azure-storage-blob/tests/test_retry_async.py b/sdk/storage/azure-storage-blob/tests/test_retry_async.py
index faeced96d1ae..669681432f17 100644
--- a/sdk/storage/azure-storage-blob/tests/test_retry_async.py
+++ b/sdk/storage/azure-storage-blob/tests/test_retry_async.py
@@ -644,14 +644,16 @@ async def test_retry_on_service_response_error(self, **kwargs):
         # Mock the internal response to raise ServiceResponseError on first chunk processing
         from azure.storage.blob.aio._download_async import process_content as real_process_content
 
-        async def mock_process_content_with_error(response, start_offset, end_offset, encryption):
+        async def mock_process_content_with_error(
+            response, start_offset, end_offset, encryption, expected_encryption_data
+        ):
             retry_counter.simple_count(retry)
             conn_error = AzureError("Connection reset by peer")
             if retry_counter.count == 1:
                 raise ServiceResponseError(conn_error, error=conn_error)
             if retry_counter.count == 2:
                 raise ServiceResponseTimeoutError(conn_error, error=conn_error)
-            return await real_process_content(response, start_offset, end_offset, encryption)
+            return await real_process_content(response, start_offset, end_offset, encryption, expected_encryption_data)
 
         # Act
         try:
diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_encryption.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_encryption.py
index 5d9fcb187987..5c4832a4b672 100644
--- a/sdk/storage/azure-storage-queue/azure/storage/queue/_encryption.py
+++ b/sdk/storage/azure-storage-queue/azure/storage/queue/_encryption.py
@@ -8,6 +8,7 @@
 import math
 import os
 import sys
+import threading
 import warnings
 from collections import OrderedDict
 from io import BytesIO
@@ -41,11 +42,7 @@
 _ENCRYPTION_PROTOCOL_V1 = "1.0"
 _ENCRYPTION_PROTOCOL_V2 = "2.0"
 _ENCRYPTION_PROTOCOL_V2_1 = "2.1"
-_VALID_ENCRYPTION_PROTOCOLS = [
-    _ENCRYPTION_PROTOCOL_V1,
-    _ENCRYPTION_PROTOCOL_V2,
-    _ENCRYPTION_PROTOCOL_V2_1,
-]
+_VALID_ENCRYPTION_PROTOCOLS = [_ENCRYPTION_PROTOCOL_V1, _ENCRYPTION_PROTOCOL_V2, _ENCRYPTION_PROTOCOL_V2_1]
 _ENCRYPTION_V2_PROTOCOLS = [_ENCRYPTION_PROTOCOL_V2, _ENCRYPTION_PROTOCOL_V2_1]
 _GCM_REGION_DATA_LENGTH = 4 * 1024 * 1024
 _GCM_NONCE_LENGTH = 12
@@ -56,6 +53,13 @@
 _ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION = (
     "The require_encryption flag is set, but encryption is not supported for this method."
 )
+_ERROR_ENCRYPTION_METADATA_MISMATCH = (
+    "The encryption metadata in the download response does not match the encryption metadata "
+    "retrieved at the start of the download. The blob's encryption metadata may have been modified "
+    "while the download was in progress."
+)
+
+_ERROR_INVALID_ENCRYPTION_METADATA = "The encryption metadata is not valid and may have been modified."
 
 
 class KeyEncryptionKey(Protocol):
@@ -220,6 +224,49 @@ def __init__(
         self.wrapped_content_key = wrapped_content_key
         self.key_wrapping_metadata = key_wrapping_metadata
 
+    def matches(self, other: "_EncryptionData") -> bool:
+        """
+        Determines whether this encryption data refers to the same encrypted content by comparing
+        every field that affects decryption. This is used to detect whether a blob's encryption
+        metadata has changed partway through a download, which could indicate the blob was
+        overwritten or tampered with.
+
+        :param _EncryptionData other: The encryption data to compare against.
+        :return: True if the decryption-relevant metadata matches, False otherwise.
+        :rtype: bool
+        """
+        if (
+            self.encryption_agent.protocol != other.encryption_agent.protocol
+            or self.encryption_agent.encryption_algorithm != other.encryption_agent.encryption_algorithm
+        ):
+            return False
+
+        if (
+            self.wrapped_content_key.key_id != other.wrapped_content_key.key_id
+            or self.wrapped_content_key.algorithm != other.wrapped_content_key.algorithm
+            or self.wrapped_content_key.encrypted_key != other.wrapped_content_key.encrypted_key
+        ):
+            return False
+
+        # Compare the content encryption IV (used for AES-CBC / V1).
+        if self.content_encryption_IV != other.content_encryption_IV:
+            return False
+
+        # Compare the encrypted region info (used for AES-GCM / V2).
+        self_region = self.encrypted_region_info
+        other_region = other.encrypted_region_info
+        if (self_region is None) != (other_region is None):
+            return False
+        if self_region is not None and other_region is not None:
+            if (
+                self_region.data_length != other_region.data_length
+                or self_region.nonce_length != other_region.nonce_length
+                or self_region.tag_length != other_region.tag_length
+            ):
+                return False
+
+        return True
+
 
 class GCMBlobEncryptionStream:
     """
@@ -312,10 +359,7 @@ def is_encryption_v2(encryption_data: Optional[_EncryptionData]) -> bool:
 
 
 def modify_user_agent_for_encryption(
-    user_agent: str,
-    moniker: str,
-    encryption_version: str,
-    request_options: Dict[str, Any],
+    user_agent: str, moniker: str, encryption_version: str, request_options: Dict[str, Any]
 ) -> None:
     """
     Modifies the request options to contain a user agent string updated with encryption information.
@@ -367,10 +411,7 @@ def get_adjusted_upload_size(length: int, encryption_version: str) -> int:
 
 
 def get_adjusted_download_range_and_offset(
-    start: int,
-    end: int,
-    length: Optional[int],
-    encryption_data: Optional[_EncryptionData],
+    start: int, end: int, length: Optional[int], encryption_data: Optional[_EncryptionData]
 ) -> Tuple[Tuple[int, int], Tuple[int, int]]:
     """
     Gets the new download range and offsets into the decrypted data for
@@ -589,17 +630,11 @@ def _dict_to_encryption_data(encryption_data_dict: Dict[str, Any]) -> _Encryptio
     if "EncryptedRegionInfo" in encryption_data_dict:
         encrypted_region_info = encryption_data_dict["EncryptedRegionInfo"]
         region_info = _EncryptedRegionInfo(
-            encrypted_region_info["DataLength"],
-            encrypted_region_info["NonceLength"],
-            _GCM_TAG_LENGTH,
+            encrypted_region_info["DataLength"], encrypted_region_info["NonceLength"], _GCM_TAG_LENGTH
         )
 
     encryption_data = _EncryptionData(
-        encryption_iv,
-        region_info,
-        encryption_agent,
-        wrapped_content_key,
-        key_wrapping_metadata,
+        encryption_iv, region_info, encryption_agent, wrapped_content_key, key_wrapping_metadata
     )
 
     return encryption_data
@@ -672,8 +707,7 @@ def _validate_and_unwrap_cek(
         raise ValueError("Provided or resolved key-encryption-key does not match the id of key used to encrypt.")
     # Will throw an exception if the specified algorithm is not supported.
     content_encryption_key = key_encryption_key.unwrap_key(
-        encryption_data.wrapped_content_key.encrypted_key,
-        encryption_data.wrapped_content_key.algorithm,
+        encryption_data.wrapped_content_key.encrypted_key, encryption_data.wrapped_content_key.algorithm
     )
 
     # For V2, the version is included with the cek. We need to validate it
@@ -682,7 +716,7 @@ def _validate_and_unwrap_cek(
         version_2_bytes = encryption_data.encryption_agent.protocol.encode().ljust(8, b"\0")
         cek_version_bytes = content_encryption_key[: len(version_2_bytes)]
         if cek_version_bytes != version_2_bytes:
-            raise ValueError("The encryption metadata is not valid and may have been modified.")
+            raise ValueError(_ERROR_INVALID_ENCRYPTION_METADATA)
 
         # Remove version from the start of the cek.
         content_encryption_key = content_encryption_key[len(version_2_bytes) :]
@@ -856,7 +890,96 @@ def generate_blob_encryption_data(
     return content_encryption_key, initialization_vector, encryption_data
 
 
-def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
+def _parse_content_range(content_range: str) -> Tuple[int, int, int]:
+    """
+    Parses a Content-Range header of the form 'bytes x-y/size' into its
+    start, end, and total size components.
+
+    :param str content_range: The Content-Range header value.
+    :return: A tuple of (start, end, total size).
+    :rtype: Tuple[int, int, int]
+    """
+    # Format: 'bytes x-y/size' -- ignore the leading 'bytes' word.
+    byte_range, size = content_range.split(" ")[1].split("/")
+    start, end = byte_range.split("-")
+    return int(start), int(end), int(size)
+
+
+def _region_nonce_encodings(nonce_length: int) -> Dict[str, Callable[[int], bytes]]:
+    """
+    Returns the supported per-region nonce encodings, keyed by the SDK that produces them.
+
+    The per-region nonce is a counter of the region's position, but each SDK encodes it
+    differently, so all supported encodings must be understood for interoperability:
+
+    * Python: zero-based counter, big-endian across the whole nonce (value in trailing bytes).
+    * Java: zero-based counter, big-endian in the leading 8 bytes, trailing bytes zeroed.
+    * .NET: one-based counter, little-endian in the trailing 8 bytes, leading bytes zeroed.
+
+    These encodings share the same value space, so they must not be accepted independently
+    per region (for example Java's nonce for region 1 is identical to .NET's nonce for
+    region 16,777,215). A single encoding is instead selected and enforced across the whole
+    download; see ``decrypt_blob``.
+
+    :param int nonce_length: The length of the nonce in bytes.
+    :return: A mapping of SDK name to a function returning that SDK's nonce for a region index.
+    :rtype: Dict[str, Callable[[int], bytes]]
+    """
+    encodings: Dict[str, Callable[[int], bytes]] = {
+        "python": lambda index: index.to_bytes(nonce_length, "big"),
+    }
+
+    counter_length = 8
+    pad = nonce_length - counter_length
+    encodings["java"] = lambda index: index.to_bytes(counter_length, "big") + b"\x00" * pad
+    encodings["dotnet"] = lambda index: b"\x00" * pad + (index + 1).to_bytes(counter_length, "little")
+
+    return encodings
+
+
+class _GCMRegionNonceValidator:
+    """
+    Enforces that every region across a whole download uses a single nonce encoding.
+
+    ``decrypt_blob`` runs once per HTTP chunk, so the candidate encodings are shared and
+    intersected across all chunks (including concurrent ones) rather than reset per call.
+    Otherwise the encoding could change at a chunk boundary and, at an encoding collision,
+    let a relocated region pass validation.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._candidates: Optional[Dict[str, Callable[[int], bytes]]] = None
+        # Set once the candidates collapse to a single encoding; read lock-free thereafter.
+        self._encoding: Optional[Callable[[int], bytes]] = None
+
+    def validate_region(self, region_index: int, nonce: bytes, nonce_length: int) -> None:
+        """
+        Narrows the shared candidate encodings to those consistent with this region.
+
+        :param int region_index: The zero-based index of the region within the blob.
+        :param bytes nonce: The nonce read from the region.
+        :param int nonce_length: The length of the nonce in bytes.
+        """
+        encoding = self._encoding
+        if encoding is not None:
+            if encoding(region_index) != nonce:
+                raise ValueError(_ERROR_INVALID_ENCRYPTION_METADATA)
+            return
+
+        with self._lock:
+            if self._candidates is None:
+                self._candidates = _region_nonce_encodings(nonce_length)
+            self._candidates = {
+                name: encode for name, encode in self._candidates.items() if encode(region_index) == nonce
+            }
+            if not self._candidates:
+                raise ValueError(_ERROR_INVALID_ENCRYPTION_METADATA)
+            if len(self._candidates) == 1:
+                self._encoding = next(iter(self._candidates.values()))
+
+
+def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements,too-many-branches
     require_encryption: bool,
     key_encryption_key: Optional[KeyEncryptionKey],
     key_resolver: Optional[Callable[[str], KeyEncryptionKey]],
@@ -864,6 +987,8 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
     start_offset: int,
     end_offset: int,
     response_headers: Dict[str, Any],
+    expected_encryption_data: Optional[_EncryptionData] = None,
+    nonce_validator: Optional["_GCMRegionNonceValidator"] = None,
 ) -> bytes:
     """
     Decrypts the given blob contents and returns only the requested range.
@@ -891,12 +1016,22 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
     :param Dict[str, Any] response_headers:
         A dictionary of response headers from the download request. Expected to include the
         'x-ms-meta-encryptiondata' header if the blob was encrypted.
+    :param Optional[_EncryptionData] expected_encryption_data:
+        The encryption data retrieved at the start of the download. If provided, the encryption
+        metadata on this response is validated against it to detect the blob's encryption metadata
+        being modified (tampered with) partway through a download.
+    :param Optional[_GCMRegionNonceValidator] nonce_validator:
+        Shared state used to enforce a single V2 nonce encoding across every chunk of a Blob download.
+        Required for V2 decryption unless the caller has set the
+        AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS environment variable.
     :return: The decrypted blob content.
     :rtype: bytes
     """
     try:
         encryption_data = _dict_to_encryption_data(loads(response_headers["x-ms-meta-encryptiondata"]))
     except Exception as exc:  # pylint: disable=broad-except
+        if expected_encryption_data is not None:
+            raise ValueError(_ERROR_ENCRYPTION_METADATA_MISMATCH) from exc
         if require_encryption:
             raise ValueError(
                 "Encryption required, but received data does not contain appropriate metadata."
@@ -905,11 +1040,12 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
 
         return content
 
+    # Validate that the encryption metadata has not changed since the start of the download.
+    if expected_encryption_data is not None and not expected_encryption_data.matches(encryption_data):
+        raise ValueError(_ERROR_ENCRYPTION_METADATA_MISMATCH)
+
     algorithm = encryption_data.encryption_agent.encryption_algorithm
-    if algorithm not in (
-        _EncryptionAlgorithm.AES_CBC_256,
-        _EncryptionAlgorithm.AES_GCM_256,
-    ):
+    if algorithm not in (_EncryptionAlgorithm.AES_CBC_256, _EncryptionAlgorithm.AES_GCM_256):
         raise ValueError("Specified encryption algorithm is not supported.")
 
     version = encryption_data.encryption_agent.protocol
@@ -924,16 +1060,7 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
         iv: Optional[bytes] = None
         unpad = False
         if "content-range" in response_headers:
-            content_range = response_headers["content-range"]
-            # Format: 'bytes x-y/size'
-
-            # Ignore the word 'bytes'
-            content_range = content_range.split(" ")
-
-            content_range = content_range[1].split("-")
-            content_range = content_range[1].split("/")
-            end_range = int(content_range[0])
-            blob_size = int(content_range[1])
+            _, end_range, blob_size = _parse_content_range(response_headers["content-range"])
 
             if start_offset >= 16:
                 iv = content[:16]
@@ -977,6 +1104,25 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
         tag_length = encryption_data.encrypted_region_info.tag_length
         region_length = nonce_length + data_length + tag_length
 
+        # The per-region nonce is a counter of the region's index within the blob. The
+        # downloaded content always begins on a region boundary, so derive the first
+        # region's index from the download range (0 when the whole blob was downloaded).
+        # This lets us validate each nonce and detect reordered regions.
+        start_range = 0
+        if "content-range" in response_headers:
+            start_range, _, _ = _parse_content_range(response_headers["content-range"])
+        nonce_counter = start_range // region_length
+
+        # Bypass nonce validation via an environment variable for data-recovery scenarios
+        # where regions were reordered. Not recommended: it can allow tampered data through.
+        validate_nonce = os.environ.get(
+            "AZURE_STORAGE_CSE_V2_ALLOW_MISORDERED_AUTH_REGIONS", ""
+        ).strip().lower() not in ("true", "1")
+
+        # A validator is required to enforce a single nonce encoding across the whole download.
+        if validate_nonce and nonce_validator is None:
+            raise ValueError("A nonce validator is required to decrypt Encryption V2 content.")
+
         decrypted_content = bytearray()
         while offset < total_size:
             # Process one encryption region at a time
@@ -985,6 +1131,10 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
 
             # First bytes are the nonce
             nonce = encrypted_region[:nonce_length]
+            # Validate the nonce matches the expected counter for this region under a single
+            # consistent encoding. A mismatch indicates the regions were reordered or tampered with.
+            if nonce_validator is not None and validate_nonce:
+                nonce_validator.validate_region(nonce_counter, nonce, nonce_length)
             ciphertext_with_tag = encrypted_region[nonce_length:]
 
             aesgcm = AESGCM(content_encryption_key)
@@ -992,6 +1142,7 @@ def decrypt_blob(  # pylint: disable=too-many-locals,too-many-statements
             decrypted_content.extend(decrypted_data)
 
             offset += process_size
+            nonce_counter += 1
 
         # Read the caller requested data from the decrypted content
         return decrypted_content[start_offset:end_offset]