fix: aws filter fallback and chunks sorting

eodag/plugins/authentication/aws_auth.py

@@ -165,6 +165,22 @@ def authenticate(self) -> S3ServiceResource:
         self.s3_resource = self._create_s3_resource()
         return self.s3_resource
 
+    def get_bucket_objects(self, bucket_name: str) -> BucketObjectsCollection:
+        """Get the object collection for an S3 bucket.
+
+        The returned collection is configured for requester-pays buckets when
+        required by the authentication configuration.
+
+        :param bucket_name: Name of the S3 bucket
+        :returns: Collection of objects in the bucket, filtered for requester-pays if necessary
+        """
+        if not self.s3_resource:
+            self.s3_resource = self._create_s3_resource()
+        objects = self.s3_resource.Bucket(bucket_name).objects
+        if self.config.requester_pays:
+            return objects.filter(RequestPayer="requester")
+        return objects
+
     def _get_authenticated_objects(
         self, bucket_name: str, prefix: str
     ) -> BucketObjectsCollection:
@@ -174,15 +190,8 @@ def _get_authenticated_objects(
         :param prefix: Prefix used to filter objects
         :returns: The boto3 authenticated objects
         """
-        if not self.s3_resource:
-            self.s3_resource = self._create_s3_resource()
         try:
-            if self.config.requester_pays:
-                objects = self.s3_resource.Bucket(bucket_name).objects.filter(
-                    RequestPayer="requester"
-                )
-            else:
-                objects = self.s3_resource.Bucket(bucket_name).objects
+            objects = self.get_bucket_objects(bucket_name)
             list(objects.filter(Prefix=prefix).limit(1))
             if objects:
                 logger.debug(
@@ -217,7 +226,7 @@ def authenticate_objects(
         :return: authenticated objects per bucket
         """
 
-        authenticated_objects: dict[str, Any] = {}
+        authenticated_objects: dict[str, BucketObjectsCollection] = {}
         auth_error_messages: set[str] = set()
         for _, pack in enumerate(bucket_names_and_prefixes):
 

eodag/plugins/download/aws.py

@@ -59,13 +59,19 @@
     NotAvailableError,
     TimeOutError,
 )
-from eodag.utils.s3 import S3FileInfo, open_s3_zipped_object, stream_download_from_s3
+from eodag.utils.s3 import (
+    S3FileInfo,
+    list_files_in_s3_prefix,
+    open_s3_zipped_object,
+    stream_download_from_s3,
+)
 
 from .base import Download
 
 if TYPE_CHECKING:
     from mypy_boto3_s3 import S3ServiceResource
     from mypy_boto3_s3.client import S3Client
+    from mypy_boto3_s3.service_resource import BucketObjectsCollection, ObjectSummary
 
     from eodag.api.product import EOProduct
     from eodag.config import PluginConfig
@@ -256,7 +262,6 @@ def download(
                         file or with environment variables.
         :returns: The absolute path to the downloaded product in the local filesystem
         """
-
         if progress_callback is None:
             logger.info(
                 "Progress bar unavailable, please call product.download() instead of plugin.download()"
@@ -430,9 +435,7 @@ def _download_file_in_zip(
         progress_callback: ProgressCallback,
         executor: ThreadPoolExecutor,
     ):
-        """
-        Download file in zip from a prefix like `foo/bar.zip!file.txt`
-        """
+        """Download file in zip from a prefix like `foo/bar.zip!file.txt`."""
         if downloader_auth.s3_resource is None:
             logger.debug("Cannot check files in s3 zip without s3 resource")
             return bucket_names_and_prefixes
@@ -504,9 +507,9 @@ def _download_preparation(
         progress_callback: ProgressCallback,
         **kwargs: Unpack[DownloadConf],
     ) -> tuple[Optional[str], Optional[str]]:
-        """
-        Preparation for the download:
+        """Prepare the download.
 
+        It means:
         - check if file was already downloaded
         - get file path
         - create directories
@@ -533,8 +536,7 @@ def _download_preparation(
         return product_local_path, record_filename
 
     def _configure_safe_build(self, build_safe: bool, product: EOProduct):
-        """
-        Updates the product properties with fetch metadata if safe build is enabled
+        """Update the product properties with fetch metadata if safe build is enabled.
 
         :param build_safe: if safe build is enabled
         :param product: product to be updated
@@ -579,8 +581,7 @@ def _get_bucket_names_and_prefixes(
         ignore_assets: bool,
         complementary_url_keys: list[str],
     ) -> list[tuple[str, Optional[str]]]:
-        """
-        Retrieves the bucket names and path prefixes for the assets
+        """Retrieve the bucket names and path prefixes for the assets.
 
         :param product: product for which the assets shall be downloaded
         :param asset_filter: text for which the assets should be filtered
@@ -635,14 +636,13 @@ def _get_bucket_names_and_prefixes(
     def _get_unique_products(
         self,
         bucket_names_and_prefixes: list[tuple[str, Optional[str]]],
-        authenticated_objects: dict[str, Any],
+        authenticated_objects: dict[str, BucketObjectsCollection],
         asset_filter: Optional[str],
         ignore_assets: bool,
         product: EOProduct,
         raise_error: bool = True,
-    ) -> set[Any]:
-        """
-        Retrieve unique product chunks based on authenticated objects and asset filters
+    ) -> set[ObjectSummary]:
+        """Retrieve unique product chunks based on authenticated objects and asset filters.
 
         :param bucket_names_and_prefixes: list of bucket names and corresponding path prefixes
         :param authenticated_objects: available objects per bucket
@@ -652,12 +652,15 @@ def _get_unique_products(
         :param raise_error: raise error if there is nothing to download
         :return: set of product chunks that can be downloaded
         """
-        product_chunks: list[Any] = []
+        product_chunks: list[ObjectSummary] = []
         for bucket_name, prefix in bucket_names_and_prefixes:
             # unauthenticated items filtered out
             if bucket_name in authenticated_objects.keys():
                 product_chunks.extend(
-                    authenticated_objects[bucket_name].filter(Prefix=prefix)
+                    list_files_in_s3_prefix(
+                        prefix or "",
+                        authenticated_objects[bucket_name],
+                    )
                 )
 
         unique_product_chunks = set(product_chunks)
@@ -691,8 +694,7 @@ def stream_download(
         timeout: float = DEFAULT_DOWNLOAD_TIMEOUT,
         **kwargs: Unpack[DownloadConf],
     ) -> StreamResponse:
-        """
-        Stream EO product data as a FastAPI-compatible `StreamResponse`, with support for partial downloads,
+        """Stream EO product data as a FastAPI-compatible `StreamResponse`, with support for partial downloads,
         asset filtering, and on-the-fly compression.
 
         This method streams data from one or more S3 objects that belong to a given EO product.
@@ -855,7 +857,7 @@ def _get_commonpath(
     def get_product_bucket_name_and_prefix(
         self, product: EOProduct, url: Optional[str] = None
     ) -> tuple[str, Optional[str]]:
-        """Extract bucket name and prefix from product URL
+        """Extract bucket name and prefix from product URL.
 
         :param product: The EO product to download
         :param url: (optional) URL to use as product.location
@@ -880,7 +882,7 @@ def get_product_bucket_name_and_prefix(
         return bucket, prefix
 
     def check_manifest_file_list(self, product_path: str) -> None:
-        """Checks if products listed in manifest.safe exist"""
+        """Check if products listed in manifest.safe exist."""
         manifest_path_list = [
             os.path.join(d, x)
             for d, _, f in os.walk(product_path)
@@ -904,7 +906,7 @@ def check_manifest_file_list(self, product_path: str) -> None:
                 logger.warning("SAFE build: %s is missing" % safe_file.get("href"))
 
     def finalize_s2_safe_product(self, product_path: str) -> None:
-        """Add missing dirs to downloaded product"""
+        """Add missing dirs to downloaded product."""
         try:
             logger.debug("Finalize SAFE product")
             manifest_path_list = [
@@ -970,7 +972,7 @@ def get_chunk_dest_path(
         dir_prefix: Optional[str] = None,
         build_safe: bool = False,
     ) -> str:
-        """Get chunk SAFE destination path"""
+        """Get chunk SAFE destination path."""
         if not build_safe:
             if dir_prefix is None:
                 dir_prefix = chunk.key

eodag/utils/s3.py

@@ -51,6 +51,7 @@
     from zipfile import ZipInfo
 
     from mypy_boto3_s3.client import S3Client
+    from mypy_boto3_s3.service_resource import BucketObjectsCollection, ObjectSummary
 
     from eodag.api.product import EOProduct  # type: ignore
 
@@ -526,6 +527,42 @@ def stream_download_from_s3(
     )
 
 
+def list_files_in_s3_prefix(
+    prefix: str,
+    objects: BucketObjectsCollection,
+) -> list[ObjectSummary]:
+    """List files matching an S3 object key or a descendant prefix.
+
+    S3 prefix matching is lexical, so sibling keys such as ``product.txt`` are
+    excluded when listing ``product``. Directory marker objects are excluded,
+    except for a standalone exact object. Some S3-compatible services require
+    a trailing slash to list descendants, which is retried when needed.
+
+    :param prefix: S3 object key prefix to list
+    :param objects: Collection of S3 objects to filter
+    :returns: List of S3 objects matching the prefix criteria
+    """
+    descendant_prefix = f"{prefix.rstrip('/')}/"
+    matching_objects = list(objects.filter(Prefix=prefix))
+    if prefix and not matching_objects:
+        matching_objects = list(objects.filter(Prefix=descendant_prefix))
+
+    has_descendants = any(
+        item.key.startswith(descendant_prefix) and not item.key.endswith("/")
+        for item in matching_objects
+    )
+    return [
+        item
+        for item in matching_objects
+        if not item.key.endswith("/")
+        and (
+            not prefix
+            or (item.key == prefix and not (has_descendants and item.size == 0))
+            or item.key.startswith(descendant_prefix)
+        )
+    ]
+
+
 def update_assets_from_s3(
     product: EOProduct,
     auth: AwsAuth,
@@ -578,29 +615,25 @@ def update_assets_from_s3(
                     }
         else:
             # List files in prefix
-            s3_objects = s3_client.list_objects(
-                Bucket=bucket, Prefix=prefix, MaxKeys=300
-            )
-            for item_s3 in s3_objects.get("Contents", []):
+            bucket_objects = auth.get_bucket_objects(bucket)
+            for item_s3 in list_files_in_s3_prefix(prefix, bucket_objects):
 
-                url = "s3://{bucket}/{key}".format(bucket=bucket, key=item_s3["Key"])
+                url = "s3://{bucket}/{key}".format(bucket=bucket, key=item_s3.key)
                 key, roles = product.driver.guess_asset_key_and_roles(url, product)
                 if key is not None:
                     asset_data: dict[str, Any] = {
                         "title": key,
                         "roles": roles,
                         "href": url,
-                        "type": guess_file_type(item_s3["Key"]),
+                        "type": guess_file_type(item_s3.key),
                     }
-                    etag = item_s3.get("ETag")
+                    etag = item_s3.e_tag
                     # multipart-upload ETags ("<hash>-N") are not MD5 digests
                     if isinstance(etag, str) and "-" not in etag:
                         asset_data["file:checksum"] = etag.strip('"')
-                    size = item_s3.get("Size")
-                    if size is not None:
-                        asset_data["file:size"] = size
+                    asset_data["file:size"] = item_s3.size
 
-                    if last_modified := to_iso_utc_string(item_s3.get("LastModified")):
+                    if last_modified := to_iso_utc_string(item_s3.last_modified):
                         asset_data["updated"] = last_modified
 
                     assets_data[key] = asset_data

tests/context.py

@@ -117,6 +117,7 @@
 from eodag.utils.import_system import import_all_modules, patch_owslib_requests
 from eodag.utils.notebook import NotebookWidgets, check_ipython, check_notebook
 from eodag.utils.s3 import (
+    list_files_in_s3_prefix,
     list_files_in_s3_zipped_object,
     update_assets_from_s3,
     open_s3_zipped_object,

tests/units/test_auth_plugins.py

@@ -854,6 +854,29 @@ def test_plugins_auth_aws_authenticate_objects(
         auth_objects = plugin.authenticate_objects(buckets_prefixes)
         self.assertDictEqual({"a": auth_objects_a, "b": auth_objects_b}, auth_objects)
 
+    @mock.patch(
+        "eodag.plugins.authentication.aws_auth.AwsAuth._create_s3_resource",
+        autospec=True,
+    )
+    def test_plugins_auth_aws_get_bucket_objects_requester_pays(
+        self, mock_create_s3_resource
+    ):
+        """get_bucket_objects must configure requester-pays collections."""
+        plugin = self.get_auth_plugin("provider_with_auth_keys")
+        plugin.config.requester_pays = True
+        plugin.s3_resource = None
+        bucket_objects = (
+            mock_create_s3_resource.return_value.Bucket.return_value.objects
+        )
+
+        objects = plugin.get_bucket_objects("requester-pays-bucket")
+
+        mock_create_s3_resource.return_value.Bucket.assert_called_once_with(
+            "requester-pays-bucket"
+        )
+        bucket_objects.filter.assert_called_once_with(RequestPayer="requester")
+        self.assertIs(objects, bucket_objects.filter.return_value)
+
     @mock.patch(
         "eodag.plugins.authentication.aws_auth.AwsAuth._create_s3_resource",
         autospec=True,