New option to prevent ~/.netrc modifications

eof/_auth.py

@@ -6,48 +6,77 @@
 from pathlib import Path
 
 from ._types import Filename
+from .log import logger as _logger
 
 NASA_HOST = "urs.earthdata.nasa.gov"
 DATASPACE_HOST = "dataspace.copernicus.eu"
 
 
-def setup_netrc(netrc_file: Filename = "~/.netrc", host: str = NASA_HOST):
+def check_netrc(netrc_file: Filename = "~/.netrc"):
+    """Chech that the netrc file exists and has the proper permissions."""
+    if not _file_is_0600(netrc_file):
+        # User has a netrc file, but it's not set up correctly
+        _logger.warning(
+            f"Your netrc file ('{netrc_file}') does not have the "
+            f"correct permissions: 0600* (read/write for user only).",
+        )
+
+
+def setup_netrc(
+    netrc_file: Filename = "~/.netrc",
+    host: str = NASA_HOST,
+    dryrun: bool = False,
+):
     """Prompt user for NASA/Dataspace username/password, store as attribute of ~/.netrc."""
     netrc_file = Path(netrc_file).expanduser()
     try:
         n = netrc.netrc(netrc_file)
         has_correct_permission = _file_is_0600(netrc_file)
         if not has_correct_permission:
             # User has a netrc file, but it's not set up correctly
-            print(
-                "Your ~/.netrc file does not have the correct"
-                " permissions.\n*Changing permissions to 0600*"
-                " (read/write for user only).",
-            )
-            os.chmod(netrc_file, 0o600)
+            if dryrun:
+                _logger.warning(
+                    f"Your netrc file ('{netrc_file}') does not have the "
+                    f"correct permissions: 0600* (read/write for user only).",
+                )
+            else:
+                _logger.warning(
+                    "Your ~/.netrc file does not have the correct"
+                    " permissions.\n*Changing permissions to 0600*"
+                    " (read/write for user only).",
+                )
+                os.chmod(netrc_file, 0o600)
         # Check account exists, as well is having username and password
+        authenticator = n.authenticators(host)
+        if authenticator is not None:
+            username, _, password = authenticator
+
         _has_existing_entry = (
             host in n.hosts
-            and n.authenticators(host)[0]  # type: ignore
-            and n.authenticators(host)[2]  # type: ignore
+            and username  # type: ignore
+            and password  # type: ignore
         )
         if _has_existing_entry:
-            return
+            return username, password
     except FileNotFoundError:
-        # User doesn't have a netrc file, make one
-        print("No ~/.netrc file found, creating one.")
-        Path(netrc_file).write_text("")
-        n = netrc.netrc(netrc_file)
+        if not dryrun:
+            # User doesn't have a netrc file, make one
+            print("No ~/.netrc file found, creating one.")
+            Path(netrc_file).write_text("")
+            n = netrc.netrc(netrc_file)
 
     username, password = _get_username_pass(host)
-    # Add account to netrc file
-    n.hosts[host] = (username, None, password)
-    print(f"Saving credentials to {netrc_file} (machine={host}).")
-    with open(netrc_file, "w") as f:
-        f.write(str(n))
-    # Set permissions to 0600 (read/write for user only)
-    # https://www.ibm.com/docs/en/aix/7.1?topic=formats-netrc-file-format-tcpip
-    os.chmod(netrc_file, 0o600)
+    if not dryrun:
+        # Add account to netrc file
+        n.hosts[host] = (username, None, password)
+        print(f"Saving credentials to {netrc_file} (machine={host}).")
+        with open(netrc_file, "w") as f:
+            f.write(str(n))
+        # Set permissions to 0600 (read/write for user only)
+        # https://www.ibm.com/docs/en/aix/7.1?topic=formats-netrc-file-format-tcpip
+        os.chmod(netrc_file, 0o600)
+
+    return username, password
 
 
 def _file_is_0600(filename: Filename):

eof/asf_client.py

@@ -7,7 +7,6 @@
 
 import requests
 
-from ._auth import NASA_HOST, setup_netrc
 from ._select_orbit import T_ORBIT, ValidityError, last_valid_orbit
 from ._types import Filename
 from .log import logger
@@ -25,7 +24,6 @@ class ASFClient:
     eof_lists = {"precise": None, "restituted": None}
 
     def __init__(self, cache_dir: Optional[Filename] = None):
-        setup_netrc(host=NASA_HOST)
         self._cache_dir = cache_dir
 
     def get_full_eof_list(self, orbit_type="precise", max_dt=None):

eof/cli.py

@@ -8,6 +8,7 @@
 import click
 
 from eof import download, log
+from eof._auth import NASA_HOST, DATASPACE_HOST, setup_netrc
 
 
 @click.command()
@@ -63,6 +64,29 @@
     is_flag=True,
     help="Set logging level to DEBUG",
 )
+@click.option(
+    "--cdse-username",
+    help="Copernicus Data Space Ecosystem username. "
+    "If not provided the program asks for it",
+)
+@click.option(
+    "--cdse-password",
+    help="Copernicus Data Space Ecosystem password. "
+    "If not provided the program asks for it",
+)
+@click.option(
+    "--asf-username",
+    help="ASF username. If not provided the program asks for it",
+)
+@click.option(
+    "--asf-password",
+    help="ASF password. If not provided the program asks for it",
+)
+@click.option(
+    "--setup-netrc",
+    is_flag=True,
+    help="save credentials provided interactively in the ~/.netrc file if necessary",
+)
 def cli(
     search_path: str,
     save_dir: str,
@@ -72,6 +96,11 @@ def cli(
     orbit_type: str,
     force_asf: bool,
     debug: bool,
+    asf_user: str = "",
+    asf_password: str = "",
+    cdse_user: str = "",
+    cdse_password: str = "",
+    save_credantials: bool = False,
 ):
     """Download Sentinel precise orbit files.
 
@@ -82,6 +111,12 @@ def cli(
     With no arguments, searches current directory for Sentinel 1 products
     """
     log._set_logger_handler(level=logging.DEBUG if debug else logging.INFO)
+    dryrun = not save_credantials
+    if not (asf_user and asf_password):
+        setup_netrc(host=NASA_HOST, dryrun=dryrun)
+    if not (cdse_user and cdse_password):
+        setup_netrc(host=DATASPACE_HOST, dryrun=dryrun)
+
     download.main(
         search_path=search_path,
         save_dir=save_dir,
@@ -90,4 +125,8 @@ def cli(
         date=date,
         orbit_type=orbit_type,
         force_asf=force_asf,
+        asf_user=asf_user,
+        asf_password=asf_password,
+        cdse_user=cdse_user,
+        cdse_password=cdse_password,
     )

eof/dataspace_client.py

@@ -6,7 +6,7 @@
 
 import requests
 
-from ._auth import DATASPACE_HOST, get_netrc_credentials, setup_netrc
+from ._auth import DATASPACE_HOST, get_netrc_credentials
 from ._select_orbit import T_ORBIT
 from ._types import Filename
 from .log import logger
@@ -29,8 +29,9 @@ class DataspaceClient:
     T0 = timedelta(seconds=T_ORBIT + 60)
     T1 = timedelta(seconds=60)
 
-    def __init__(self):
-        setup_netrc(host=DATASPACE_HOST)
+    def __init__(self, username: str = "", password: str = ""):
+        self._username = username
+        self._password = password
 
     def query_orbit(
         self,
@@ -151,7 +152,12 @@ def query_orbit_by_dt(
 
     def download_all(self, query_results: list[dict], output_directory: Filename):
         """Download all the specified orbit products."""
-        return download_all(query_results, output_directory=output_directory)
+        return download_all(
+            query_results,
+            output_directory=output_directory,
+            username=self._username,
+            password=self._password,
+        )
 
 
 def _construct_orbit_file_query(
@@ -249,12 +255,14 @@ def query_orbit_file_service(query: str) -> list[dict]:
     return query_results
 
 
-def get_access_token():
+def get_access_token(username, password):
     """Get an access token for the Copernicus Data Space Ecosystem (CDSE) API.
 
     Code from https://documentation.dataspace.copernicus.eu/APIs/Token.html
     """
-    username, password = get_netrc_credentials(DATASPACE_HOST)
+    if not (username and password):
+        logger.debug("Get credentials form netrc")
+        username, password = get_netrc_credentials(DATASPACE_HOST)
     data = {
         "client_id": "cdse-public",
         "username": username,
@@ -338,7 +346,12 @@ def download_orbit_file(
     return output_orbit_file_path
 
 
-def download_all(query_results: list[dict], output_directory: Filename) -> list[Path]:
+def download_all(
+    query_results: list[dict],
+    output_directory: Filename,
+    username: str = "",
+    password: str = "",
+) -> list[Path]:
     """Download all the specified orbit products.
 
     Parameters
@@ -347,14 +360,19 @@ def download_all(query_results: list[dict], output_directory: Filename) -> list[
         list of results from the query
     output_directory : str | Path
         Directory to save the orbit files to.
+    username : str
+        CDSE username
+    password : str
+        CDSE password
+
     """
     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()
+    access_token = get_access_token(username, password)
     for query_result in query_results:
         query_result = query_results[0]
         orbit_file_name = query_result["Name"]

eof/download.py

@@ -46,9 +46,11 @@ def download_eofs(
     sentinel_file=None,
     save_dir=".",
     orbit_type="precise",
-    force_asf=False,
-    asf_user="",
-    asf_password="",
+    force_asf: bool = False,
+    asf_user: str = "",
+    asf_password: str = "",
+    cdse_user: str = "",
+    cdse_password: str = "",
 ):
     """Downloads and saves EOF files for specific dates
 
@@ -86,7 +88,7 @@ def download_eofs(
 
     # First, check that Scihub isn't having issues
     if not force_asf:
-        client = DataspaceClient()
+        client = DataspaceClient(username=cdse_user, password=cdse_password)
         # try to search on scihub
         if sentinel_file:
             query = client.query_orbit_for_product(sentinel_file, orbit_type=orbit_type)
@@ -101,7 +103,14 @@ def download_eofs(
 
     # For failures from scihub, try ASF
     if not dataspace_successful:
+        from ._auth import NASA_HOST, get_netrc_credentials
+
         logger.warning("Dataspace failed, trying ASF")
+
+        if not (asf_user and asf_password):
+            logger.debug("Get credentials form netrc")
+            asf_user, asf_password = get_netrc_credentials(NASA_HOST)
+
         asfclient = ASFClient()
         urls = asfclient.get_download_urls(orbit_dts, missions, orbit_type=orbit_type)
         # Download and save all links in parallel
@@ -231,9 +240,11 @@ def main(
     mission=None,
     date=None,
     orbit_type="precise",
-    force_asf=False,
-    asf_user="",
-    asf_password="",
+    force_asf: bool = False,
+    asf_user: str = "",
+    asf_password: str = "",
+    cdse_user: str = "",
+    cdse_password: str = "",
 ):
     """Function used for entry point to download eofs"""
 
@@ -275,4 +286,6 @@ def main(
         force_asf=force_asf,
         asf_user=asf_user,
         asf_password=asf_password,
+        cdse_user=cdse_user,
+        cdse_password=cdse_password,
     )