Support --access-token parameter for CDSE (#62)

* Support CDSE access-token fed by caller

* Update doc for --access-token

* Improve DbC doc for get_access_token()

and fix post-condition on `get_netrc_credentials()`

* Remove comment.

* Fix 3 authentication use cases on CDSE

---------

Co-authored-by: Luc Hermitte <luc.hermitte@csgroup.eu>

LICENSE.txt

@@ -1,7 +1,7 @@
 MIT License
 
 Copyright (c) 2018-2020 Scott Staniewicz
-Copyright (c) 2024 Luc Hermitte, CS Group, support for double authentication on CDSE
+Copyright (c) 2024 Luc Hermitte, CS Group, refactor authentication on CDSE
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -105,6 +105,10 @@ Options:
   --force-asf                     Force the downloader to search ASF instead
                                   of ESA.
   --debug                         Set logging level to DEBUG
+  --cdse-access-token TEXT        Copernicus Data Space Ecosystem access-
+                                  token. The access token can be generated
+                                  beforehand. See https://documentation.datasp
+                                  ace.copernicus.eu/APIs/Token.html
   --cdse-user TEXT                Copernicus Data Space Ecosystem username. If
                                   not provided the program asks for it
   --cdse-password TEXT            Copernicus Data Space Ecosystem password. If

eof/_auth.py

@@ -87,7 +87,12 @@ def _file_is_0600(filename: Filename):
 
 
 def get_netrc_credentials(host: str, netrc_file: Optional[Filename] = None) -> tuple[str, str]:
-    """Get username and password from netrc file for a given host."""
+    """
+    Get username and password from netrc file for a given host.
+
+    :return: username and password found for host in netrc_file
+    :postcondition: username and password are non empty strings.
+    """
     netrc_file = netrc_file or "~/.netrc"
     netrc_file = Path(netrc_file).expanduser()
     _logger.debug(f"Using {netrc_file=!r}")
@@ -98,6 +103,8 @@ def get_netrc_credentials(host: str, netrc_file: Optional[Filename] = None) -> t
     username, _, password = auth
     if username is None or password is None:
         raise ValueError(f"No username/password found for {host} in ~/.netrc")
+    if not username or not password:
+        raise ValueError(f"Empty username/password found for {host} in ~/.netrc")
     return username, password
 
 

eof/cli.py

@@ -1,6 +1,7 @@
 """
 CLI tool for downloading Sentinel 1 EOF files
 """
+
 from __future__ import annotations
 
 import logging
@@ -66,6 +67,11 @@
     is_flag=True,
     help="Set logging level to DEBUG",
 )
+@click.option(
+    "--cdse-access-token",
+    help="Copernicus Data Space Ecosystem access-token. "
+    "The access token can be generated beforehand. See https://documentation.dataspace.copernicus.eu/APIs/Token.html",
+)
 @click.option(
     "--cdse-user",
     help="Copernicus Data Space Ecosystem username. "
@@ -120,6 +126,7 @@ def cli(
     debug: bool,
     asf_user: str = "",
     asf_password: str = "",
+    cdse_access_token: Optional[str] = None,
     cdse_user: str = "",
     cdse_password: str = "",
     cdse_2fa_token: str = "",
@@ -154,6 +161,7 @@ def cli(
         force_asf=force_asf,
         asf_user=asf_user,
         asf_password=asf_password,
+        cdse_access_token=cdse_access_token,
         cdse_user=cdse_user,
         cdse_password=cdse_password,
         cdse_2fa_token=cdse_2fa_token,

eof/dataspace_client.py

@@ -32,16 +32,25 @@ class DataspaceClient:
     T1 = timedelta(seconds=60)
 
     def __init__(
-            self,
-            username: str = "",
-            password: str = "",
-            token_2fa: str = "",
-            netrc_file: Optional[Filename] = None,
+        self,
+        access_token: Optional[str] = None,
+        username: str = "",
+        password: str = "",
+        token_2fa: str = "",
+        netrc_file: Optional[Filename] = None,
     ):
-        if not (username and password):
-            logger.debug("Get credentials form netrc")
+        self._access_token = access_token
+        if access_token:
+            logger.debug("Using provided CDSE access token")
+        else:
             try:
-                username, password = get_netrc_credentials(DATASPACE_HOST, netrc_file)
+                if not (username and password):
+                    logger.debug(f"Get credentials form netrc ({netrc_file!r})")
+                    # Shall we keep username if explicitly set?
+                    username, password = get_netrc_credentials(DATASPACE_HOST, netrc_file)
+                else:
+                    logger.debug("Using provided username and password")
+                self._access_token = get_access_token(username, password, token_2fa)
             except FileNotFoundError:
                 logger.warning("No netrc file found.")
             except ValueError as e:
@@ -50,14 +59,17 @@ def __init__(
                 logger.warning(
                     f"No CDSE credentials found in netrc file {netrc_file!r}. Please create one using {SIGNUP_URL}"
                 )
+            except Exception as e:
+                logger.warning(f"Error: {str(e)}")
 
-        self._username = username
-        self._password = password
-        self._token_2fa = token_2fa
-        self._netrc_file = netrc_file
+            # Obtain an access token the download request from the provided credentials
 
+    def __bool__(self):
+        """Tells whether the object has been correctly initialized"""
+        return bool(self._access_token)
+
+    @staticmethod
     def query_orbit(
-        self,
         t0: datetime,
         t1: datetime,
         satellite_id: str,
@@ -75,8 +87,8 @@ def query_orbit(
         # range
         return query_orbit_file_service(query)
 
+    @staticmethod
     def query_orbit_for_product(
-        self,
         product,
         orbit_type: str = "precise",
         t0_margin: timedelta = T0,
@@ -85,16 +97,16 @@ def query_orbit_for_product(
         if isinstance(product, str):
             product = S1Product(product)
 
-        return self.query_orbit_by_dt(
+        return DataspaceClient.query_orbit_by_dt(
             [product.start_time],
             [product.mission],
             orbit_type=orbit_type,
             t0_margin=t0_margin,
             t1_margin=t1_margin,
         )
 
+    @staticmethod
     def query_orbit_by_dt(
-        self,
         orbit_dts,
         missions,
         orbit_type: str = "precise",
@@ -126,7 +138,7 @@ def query_orbit_by_dt(
         for dt, mission in zip(orbit_dts, missions):
             # Only check for precise orbits if that is what we want
             if orbit_type == "precise":
-                products = self.query_orbit(
+                products = DataspaceClient.query_orbit(
                     dt - t0_margin,
                     dt + t1_margin,
                     # dt - timedelta(seconds=T_ORBIT + 60),
@@ -148,7 +160,7 @@ def query_orbit_by_dt(
                 all_results.append(result)
             else:
                 # try with RESORB
-                products = self.query_orbit(
+                products = DataspaceClient.query_orbit(
                     dt - timedelta(seconds=T_ORBIT + 60),
                     dt + timedelta(seconds=60),
                     mission,
@@ -177,17 +189,13 @@ def download_all(
         self,
         query_results: list[dict],
         output_directory: Filename,
-        netrc_file : Optional[Filename] = None,
         max_workers: int = 3,
     ):
         """Download all the specified orbit products."""
         return download_all(
             query_results,
             output_directory=output_directory,
-            username=self._username,
-            password=self._password,
-            token_2fa=self._token_2fa,
-            netrc_file=netrc_file,
+            access_token=self._access_token,
             max_workers=max_workers,
         )
 
@@ -287,18 +295,16 @@ def query_orbit_file_service(query: str) -> list[dict]:
     return query_results
 
 
-def get_access_token(username, password, token_2fa, netrc_file) -> Optional[str]:
+def get_access_token(username: Optional[str], password: Optional[str], token_2fa: Optional[str]) -> str:
     """Get an access token for the Copernicus Data Space Ecosystem (CDSE) API.
 
     Code from https://documentation.dataspace.copernicus.eu/APIs/Token.html
+
+    :raises ValueError: if either username or password is empty
+    :raises RuntimeError: if the access token cannot be created
     """
     if not (username and password):
-        logger.debug("Get credentials form netrc")
-        try:
-            username, password = get_netrc_credentials(DATASPACE_HOST, netrc_file)
-        except FileNotFoundError:
-            logger.warning("No netrc file found.")
-            return None
+        raise ValueError("Username and password values are expected!")
 
     data = {
         "client_id": "cdse-public",
@@ -313,18 +319,17 @@ def get_access_token(username, password, token_2fa, netrc_file) -> Optional[str]
         r = requests.post(AUTH_URL, data=data)
         r.raise_for_status()
     except Exception as err:
-        raise RuntimeError(f"Access token creation failed. Reason: {str(err)}")
+        raise RuntimeError(f"CDSE access token creation failed. Reason: {str(err)}")
 
     # Parse the access token from the response
     try:
         access_token = r.json()["access_token"]
+        return access_token
     except KeyError:
         raise RuntimeError(
-            'Failed to parsed expected field "access_token" from authentication response.'
+            'Failed to parse expected field "access_token" from CDSE authentication response.'
         )
 
-    return access_token
-
 
 def download_orbit_file(
     request_url, output_directory, orbit_file_name, access_token
@@ -382,17 +387,14 @@ def download_orbit_file(
             if chunk:
                 outfile.write(chunk)
 
-    logger.info(f"Orbit file downloaded to {output_orbit_file_path}")
+    logger.info(f"Orbit file downloaded to {output_orbit_file_path!r}")
     return output_orbit_file_path
 
 
 def download_all(
     query_results: list[dict],
     output_directory: Filename,
-    username: str = "",
-    password: str = "",
-    token_2fa: str = "",
-    netrc_file: Optional[Filename] = None,
+    access_token: Optional[str],
     max_workers: int = 3,
 ) -> list[Path]:
     """Download all the specified orbit products.
@@ -414,14 +416,14 @@ def download_all(
         Note that >4 connections will result in a HTTP 429 Error
 
     """
+    if not access_token:
+        raise RuntimeError("Invalid CDSE access token. Aborting.")
     downloaded_paths: list[Path] = []
     # Select an appropriate orbit file from the list returned from the query
     # orbit_file_name, orbit_file_request_id = select_orbit_file(
     #     query_results, start_time, stop_time
     # )
-    # Obtain an access token the download request from the provided credentials
 
-    access_token = get_access_token(username, password, token_2fa, netrc_file)
     output_names = []
     download_urls = []
     for query_result in query_results:

eof/download.py

@@ -21,6 +21,7 @@
 
 See parsers for Sentinel file naming description
 """
+
 from __future__ import annotations
 
 import glob
@@ -51,6 +52,7 @@ def download_eofs(
     force_asf: bool = False,
     asf_user: str = "",
     asf_password: str = "",
+    cdse_access_token: Optional[str] = None,
     cdse_user: str = "",
     cdse_password: str = "",
     cdse_2fa_token: str = "",
@@ -93,8 +95,14 @@ def download_eofs(
 
     # First, check that Scihub isn't having issues
     if not force_asf:
-        client = DataspaceClient(username=cdse_user, password=cdse_password, token_2fa=cdse_2fa_token, netrc_file=netrc_file)
-        if client._username and client._password:
+        client = DataspaceClient(
+            access_token=cdse_access_token,
+            username=cdse_user,
+            password=cdse_password,
+            token_2fa=cdse_2fa_token,
+            netrc_file=netrc_file,
+        )
+        if client:
             # try to search on scihub
             if sentinel_file:
                 query = client.query_orbit_for_product(
@@ -212,6 +220,7 @@ def main(
     force_asf: bool = False,
     asf_user: str = "",
     asf_password: str = "",
+    cdse_access_token: Optional[str] = None,
     cdse_user: str = "",
     cdse_password: str = "",
     cdse_2fa_token: str = "",
@@ -258,6 +267,7 @@ def main(
         force_asf=force_asf,
         asf_user=asf_user,
         asf_password=asf_password,
+        cdse_access_token=cdse_access_token,
         cdse_user=cdse_user,
         cdse_password=cdse_password,
         cdse_2fa_token=cdse_2fa_token,