feat: Add Zarr v2 metadata access and basic auth support for OIDC-protected assets

eodag/api/product/_product.py

@@ -37,12 +37,16 @@
 import geojson
 import orjson
 import requests
+from boto3 import Session
+from boto3.resources.base import ServiceResource
 from pystac import Item
-from requests import RequestException
+from requests import PreparedRequest, RequestException
 from requests.auth import AuthBase
+from requests.structures import CaseInsensitiveDict
 from shapely import geometry
 from shapely.errors import ShapelyError
 
+from eodag.plugins.authentication.aws_auth import AwsAuth
 from eodag.types.queryables import CommonStacMetadata
 from eodag.types.stac_metadata import create_stac_metadata_model
 
@@ -83,7 +87,12 @@
     _import_stac_item_from_known_provider,
     _import_stac_item_from_unknown_provider,
 )
-from eodag.utils.exceptions import DownloadError, MisconfiguredError, ValidationError
+from eodag.utils.exceptions import (
+    AddressNotFound,
+    DownloadError,
+    MisconfiguredError,
+    ValidationError,
+)
 from eodag.utils.repr import dict_to_html_table
 
 if TYPE_CHECKING:
@@ -660,6 +669,65 @@ def stream_download(
             **kwargs,
         )
 
+    def get_storage_options(
+        self,
+        asset_key: Optional[str] = None,
+    ) -> dict[str, Any]:
+        """
+        Get fsspec storage_options keyword arguments
+        """
+        auth = self.downloader_auth.authenticate() if self.downloader_auth else None
+        if self.downloader is None:
+            return {}
+
+        # default url and headers
+        try:
+            url = self.assets[asset_key]["href"] if asset_key else self.location
+        except KeyError as e:
+            raise AddressNotFound(f"{asset_key} not found in {self} assets") from e
+        headers = {**USER_AGENT}
+
+        if isinstance(auth, ServiceResource) and isinstance(
+            self.downloader_auth, AwsAuth
+        ):
+            auth_kwargs: dict[str, Any] = dict()
+            # AwsAuth
+            if s3_endpoint := getattr(self.downloader_auth.config, "s3_endpoint", None):
+                auth_kwargs["client_kwargs"] = {"endpoint_url": s3_endpoint}
+            if creds := cast(
+                Session, self.downloader_auth.s3_session
+            ).get_credentials():
+                auth_kwargs["key"] = creds.access_key
+                auth_kwargs["secret"] = creds.secret_key
+                if creds.token:
+                    auth_kwargs["token"] = creds.token
+                if requester_pays := getattr(
+                    self.downloader_auth.config, "requester_pays", False
+                ):
+                    auth_kwargs["requester_pays"] = requester_pays
+            else:
+                auth_kwargs["anon"] = True
+            return {"path": url, **auth_kwargs}
+
+        if isinstance(auth, AuthBase):
+            # update url and headers with auth
+            req = PreparedRequest()
+            req.url = url
+            req.headers = CaseInsensitiveDict(headers)
+            if auth:
+                auth(req)
+            return {"path": req.url, "headers": dict(req.headers)}
+
+        return {"path": url}
+
+    def request_asset(
+        self,
+        url: str,
+    ) -> requests.Response:
+        """Perform a GET request to the given URL using product's authentication headers."""
+        headers = self.get_storage_options().get("headers", {})
+        return requests.get(url, headers=headers, stream=True)
+
     def _init_progress_bar(
         self,
         progress_callback: Optional[ProgressCallback],

eodag/plugins/authentication/openid_connect.py

@@ -18,10 +18,12 @@
 from __future__ import annotations
 
 import datetime as dt
+import base64
 import logging
 import re
 import string
 from random import SystemRandom
+from threading import Lock
 from typing import TYPE_CHECKING, Any, Optional
 from urllib.parse import parse_qs, urlparse
 
@@ -76,6 +78,7 @@ class OIDCRefreshTokenBase(Authentication):
 
     def __init__(self, provider: str, config: PluginConfig) -> None:
         super(OIDCRefreshTokenBase, self).__init__(provider, config)
+        self._auth_lock = Lock()
 
         self.access_token = ""
         self.access_token_expiration = dt.datetime.min.replace(tzinfo=dt.timezone.utc)
@@ -252,8 +255,9 @@ class OIDCAuthorizationCodeFlowAuth(OIDCRefreshTokenBase):
         * :attr:`~eodag.config.PluginConfig.token_key` (``str``): The key pointing
           to the token in the json response to the POST request to the token server
         * :attr:`~eodag.config.PluginConfig.token_provision` (``str``) (**mandatory**): One of
-          ``qs`` or ``header``. This is how the token obtained will be used to authenticate the
-          user on protected requests. If ``qs`` is chosen, then ``token_qs_key`` is mandatory
+          ``qs``, ``header`` or ``basic``. This is how the token obtained will be used to authenticate the
+          user on protected requests. If ``qs`` is chosen, then ``token_qs_key`` is mandatory. If ``basic`` is chosen,
+          the token is used as password with username "anonymous".
         * :attr:`~eodag.config.PluginConfig.login_form_xpath` (``str``) (**mandatory**): The
           xpath to the HTML form element representing the user login form
         * :attr:`~eodag.config.PluginConfig.authentication_uri_source` (``str``) (**mandatory**): Where
@@ -301,9 +305,13 @@ def __init__(self, provider: str, config: PluginConfig) -> None:
     def validate_config_credentials(self) -> None:
         """Validate configured credentials"""
         super(OIDCAuthorizationCodeFlowAuth, self).validate_config_credentials()
-        if getattr(self.config, "token_provision", None) not in ("qs", "header"):
+        if getattr(self.config, "token_provision", None) not in (
+            "qs",
+            "header",
+            "basic",
+        ):
             raise MisconfiguredError(
-                'Provider config parameter "token_provision" must be one of "qs" or "header"'
+                'Provider config parameter "token_provision" must be one of "qs", "header", or "basic"'
             )
         if self.config.token_provision == "qs" and not getattr(
             self.config, "token_qs_key", ""
@@ -315,18 +323,20 @@ def validate_config_credentials(self) -> None:
 
     def authenticate(self) -> CodeAuthorizedAuth:
         """Authenticate"""
-        self._get_access_token()
+        with self._auth_lock:
+            self._get_access_token()
 
         return CodeAuthorizedAuth(
             self.access_token,
             self.config.token_provision,
             key=getattr(self.config, "token_qs_key", None),
+            refresh_token=self.refresh_token,
         )
 
     def _request_new_token(self) -> dict[str, str]:
         """Fetch the access token with a new authentication"""
         logger.debug("Fetching access token from %s", self.token_endpoint)
-        state = self.compute_state()
+        state = OIDCAuthorizationCodeFlowAuth.compute_state()
         authentication_response = self.authenticate_user(state)
         exchange_url = authentication_response.url
 
@@ -583,10 +593,17 @@ def compute_state() -> str:
 class CodeAuthorizedAuth(AuthBase):
     """CodeAuthorizedAuth custom authentication class to be used with requests module"""
 
-    def __init__(self, token: str, where: str, key: Optional[str] = None) -> None:
+    def __init__(
+        self,
+        token: str,
+        where: str,
+        key: Optional[str] = None,
+        refresh_token: Optional[str] = None,
+    ) -> None:
         self.token = token
         self.where = where
         self.key = key
+        self.refresh_token = refresh_token
 
     def __call__(self, request: PreparedRequest) -> PreparedRequest:
         """Perform the actual authentication"""
@@ -601,6 +618,13 @@ def __call__(self, request: PreparedRequest) -> PreparedRequest:
 
         elif self.where == "header":
             request.headers["Authorization"] = "Bearer {}".format(self.token)
+
+        if self.where == "basic" and self.refresh_token is not None:
+            auth_str = base64.b64encode(
+                f"anonymous:{self.refresh_token}".encode()
+            ).decode()
+            request.headers["Authorization"] = f"Basic {auth_str}"
+
         logger.debug(
             re.sub(
                 r"'Bearer [^']+'",

eodag/resources/collections.yml

@@ -6622,10 +6622,6 @@ DT_CLIMATE_G2_BASELINE_CONT_ICON_R1:
     the pages [Climate DT Phase2 CLTE Parameters](https://confluence.ecmwf.int/display/DDCZ/DestinE+ClimateDT+Phase+2+clte+Parameters) and
     [Climate DT Phase2 CLMN Parameters](https://confluence.ecmwf.int/display/DDCZ/DestinE+ClimateDT+Phase+2+clmn+Parameters)".
     This Generation-2 Collection gives access to 'Control Simulation' data based on the 'ICON' model.
-  instruments: []
-  constellation: Digital Twin
-  platform: DT
-  processing:level:
   keywords: ["DT", "DE", "LUMI", "Destination-Earth", "Digital-Twin", "Climate Change", "baseline", "ICON", "cont", "Phase 2"]
   eodag:sensor_type: ATMOSPHERIC
   license: CC-BY-4.0
@@ -7322,6 +7318,33 @@ DT_CLIMATE_G2_STORY_NUDGING_TPLUS2_0K_IFS_FESOM_R5:
   title: Climate Change Adaptation Digital Twin (Climate Adaptation DT) - Storyline Simulation Future Climate - IFS-FESOM - Generation-2 - Realization 5
   extent: {"spatial": {"bbox": [[-180.0, -90.0, 180.0, 90.0]]}, "temporal": {"interval": [["2017-01-01T00:00:00Z", "2025-12-31T23:59:59Z"]]}}
 
+# Digital Twin data cubes
+DT_CLIMATE_ADAPTATION_IFS_NEMO:
+  description: |
+    A selection of datasets derived from the DestinE Climate Change Adaptation Digital Twin according to the output of the IFS-NEMO model coupled with the Shared Socioeconomic Pathway 3-7.0 (experiment: SSP3-7.0) as defined in the Scenario Model Intercomparison Project (activity: ScenarioMIP)
+  instruments: []
+  constellation: Digital Twin
+  platform: DT
+  processing:level:
+  keywords: ["DT", "Digital-Twin", "Climate Change", "Atmosphere", "Sea Ice", "SCENARIOMIP", "IFS-NEMO","Socioeconomic", "SSP3-7.0"]
+  eodag:sensor_type: ATMOSPHERIC
+  license: CC-BY-4.0
+  title: DestinE Climate Adaptation DT, activity ScenarioMIP, experiment SSP3-7.0, model IFS-NEMO
+  extent: {"spatial": {"bbox": [[-180.0, -90.0, 180.0, 90.0]]}, "temporal": {"interval": [["2020-07-01T00:00:00Z", "2040-01-01T00:00:00Z"]]}}
+
+DT_CLIMATE_ADAPTATION_ICON:
+  description: |
+    A selection of datasets derived from the DestinE Climate Change Adaptation Digital Twin according to the output of the ICON model coupled with the Shared Socioeconomic Pathway 3-7.0 (experiment: SSP3-7.0) as defined in the Scenario Model Intercomparison Project (activity: ScenarioMIP)
+  instruments: []
+  constellation: Digital Twin
+  platform: DT
+  processing:level:
+  keywords: ["DT", "Digital-Twin", "Climate Change", "Atmosphere", "Sea Ice", "SCENARIOMIP", "ICON","Socioeconomic", "SSP3-7.0"]
+  eodag:sensor_type: ATMOSPHERIC
+  license: CC-BY-4.0
+  title: DestinE Climate Adaptation DT, activity ScenarioMIP, experiment SSP3-7.0, model ICON
+  extent: {"spatial": {"bbox": [[-180.0, -90.0, 180.0, 90.0]]}, "temporal": {"interval": [["2021-07-01T00:00:00Z", "2030-12-31T00:00:00Z"]]}}
+
 # MARK: METOP --------------------------------------------------------------------------
 METOP_AMSU_L1:
   description: |

eodag/resources/providers/desp_cache.yml

@@ -0,0 +1,44 @@
+desp_cache:
+  name: desp_cache
+  priority: 0
+  roles:
+    - host
+  description: Destination Earth Digital Twin for Climate Change Adaptation through Earth Observation (DESP) - Cache
+  url: https://data.destination-earth.eu
+  search: 
+    type: StacSearch
+    api_endpoint: https://hda.data.destination-earth.eu/stac/v2/search
+    pagination:
+      max_items_per_page: 1000
+      next_page_token_key: token
+    sort:
+      sort_by_tpl: '{{"sortby": [ {{"field": "{sort_param}", "direction": "{sort_order}" }} ] }}'
+      sort_order_mapping:
+        ascending: asc
+        descending: desc
+      sort_param_mapping:
+        id: id
+        start_datetime: properties.datetime
+        end_datetime: properties.end_datetime
+  download: 
+    type: HTTPDownload
+    ssl_verify: true
+    no_auth_download: True
+    auth_error_code: 401
+  auth: 
+    type: OIDCAuthorizationCodeFlowAuth
+    oidc_config_url: https://auth.destine.eu/realms/desp/.well-known/openid-configuration
+    redirect_uri: https://cacheb.dcms.destine.eu/
+    client_id: edh-public
+    user_consent_needed: false
+    token_exchange_post_data_method: data
+    login_form_xpath: //form[@id='kc-form-login']
+    authentication_uri_source: login-form
+    token_provision: basic
+  products:
+    GENERIC_COLLECTION:
+      _collection: "{collection}"
+    DT_CLIMATE_ADAPTATION_IFS_NEMO:
+      _collection: EO.ECMWF.DAT.DT_CLIMATE_ADAPTATION_IFS-NEMO
+    DT_CLIMATE_ADAPTATION_ICON:
+      _collection: EO.ECMWF.DAT.DT_CLIMATE_ADAPTATION_ICON

eodag/resources/user_conf_template.yml

@@ -140,6 +140,15 @@ dlr_eoc_geoservice:
         credentials:
             username:
             password:
+desp_cache:
+    priority: # Lower value means lower priority (Default: 0)
+    search:
+    download:
+        output_dir:
+    auth:
+        credentials:
+            username:
+            password:
 earth_search:
     priority: # Lower value means lower priority (Default: 0)
     search: # Search parameters configuration

eodag/utils/exceptions.py

@@ -168,3 +168,8 @@ def raise_if_quota_exceeded(error: Exception, provider: str) -> None:
             raise QuotaExceededError(
                 f"Too many requests on provider {provider}, please check your quota."
             )
+
+
+class DatasetCreationError(EodagError):
+    """An error indicating that :class:`xarray.Dataset` or :class:`eodag_cube.types.XarrayDict` could not be created"""
+

tests/context.py

@@ -58,6 +58,8 @@
 from eodag.plugins.authentication.aws_auth import AwsAuth
 from eodag.plugins.authentication.header import HeaderAuth
 from eodag.plugins.authentication.openid_connect import CodeAuthorizedAuth
+from eodag.plugins.authentication.header import HTTPHeaderAuth
+from eodag.plugins.authentication.qsauth import HttpQueryStringAuth
 from eodag.plugins.base import PluginTopic
 from eodag.plugins.crunch.filter_date import FilterDate
 from eodag.plugins.crunch.filter_latest_tpl_name import FilterLatestByName