Updates/asf netrc

sar_pipeline/preparation/downloads/orbits.py

@@ -9,6 +9,7 @@
 logger = logging.getLogger(__name__)
 
 from sar_pipeline.utils.general import log_timing
+from sar_pipeline.preparation.downloads.scenes import create_asf_netrc_file
 
 VALID_ORBIT_DATA_SOURCES = ["ASF", "CDSE"]
 
@@ -113,6 +114,9 @@ def download_orbits(
                     "ASF credentials are not set. Provide them as arguments or set EARTHDATA_LOGIN and EARTHDATA_PASSWORD as environment variables."
                 )
             cdse_user, cdse_password = None, None
+            if not os.getenv("NETRC"):
+                # create the netrc if not existing for authentication
+                create_asf_netrc_file(asf_user, asf_password)
 
         else:
             raise ValueError(f"Source must be either 'CDSE' or 'ASF', got '{source}'.")
@@ -122,15 +126,29 @@ def download_orbits(
         # The logic in eof.download.main() tries CDSE first by default. set force_asf by source
         try:
             force_asf = source == "ASF"
-            orbit_paths = eof.download.main(
-                sentinel_file=scene_safe_file,
-                save_dir=save_dir,
-                cdse_user=cdse_user,
-                cdse_password=cdse_password,
-                force_asf=force_asf,
-                asf_user=asf_user,
-                asf_password=asf_password,
-            )
+            try:
+                orbit_paths = eof.download.main(
+                    sentinel_file=scene_safe_file,
+                    save_dir=save_dir,
+                    cdse_user=cdse_user,
+                    cdse_password=cdse_password,
+                    force_asf=force_asf,
+                    asf_user=asf_user,
+                    asf_password=asf_password,
+                )
+            except:
+                if force_asf:
+                    # remove asf credentials to read from .netrc
+                    logger.info(
+                        "ASF credentials failed. Falling back to .netrc authentication"
+                    )
+                    orbit_paths = eof.download.main(
+                        sentinel_file=scene_safe_file,
+                        save_dir=save_dir,
+                        cdse_user=cdse_user,
+                        cdse_password=cdse_password,
+                        force_asf=force_asf,
+                    )
 
             if len(orbit_paths) == 1:
                 break

sar_pipeline/preparation/downloads/scenes.py

@@ -8,6 +8,7 @@
 import shapely
 import ast
 import re
+import tempfile
 from cdsetool.query import query_features, FeatureQuery
 from cdsetool.credentials import Credentials
 from cdsetool.download import download_features
@@ -84,6 +85,50 @@ def query_scene_from_asf(scene: str) -> ASFSearchResults:
     return search_results
 
 
+def create_asf_netrc_file(asf_login: str, asf_pass: str) -> Path:
+    """
+    Create a temporary `.netrc` file containing ASF/Earthdata login credentials.
+
+    This function generates a temporary `.netrc` file to authenticate with the
+    Alaska Satellite Facility (ASF) or NASA Earthdata services. The file is
+    written to a secure temporary location, assigned appropriate file permissions
+    (`chmod 600`), and its path is set as the `NETRC` environment variable so that
+    tools using `requests` or other netrc-aware libraries can locate it.
+
+    Parameters
+    ----------
+    asf_login : str
+        The username associated with the ASF/NASA Earthdata account.
+    asf_pass : str
+        The password associated with the ASF/NASA Earthdata account.
+
+    Returns
+    -------
+    Path
+        The filesystem path to the created temporary `.netrc` file.
+    """
+
+    logger.info(f"Creating temporary .netrc file with ASF/Earthdata credentials")
+    with tempfile.NamedTemporaryFile("w", delete=False) as f:
+        f.write(
+            f"""machine urs.earthdata.nasa.gov
+        login {asf_login}
+        password {asf_pass}
+        """
+        )
+        netrc_path = f.name
+
+    logger.info(f"Temporary .netrc created at {netrc_path}")
+
+    # Set file permissions to 600 (required)
+    os.chmod(netrc_path, 0o600)
+
+    # Optional: tell requests/netrc-aware tools to use this file
+    os.environ["NETRC"] = netrc_path
+
+    return netrc_path
+
+
 @log_timing
 def download_scene_from_asf(
     scene: str,
@@ -151,8 +196,22 @@ def download_scene_from_asf(
             )
             MissingCredentialsError(err_string)
 
-    session = asf_search.ASFSession()
-    session.auth_with_creds(asf_login, asf_pass)
+    # create the NETRC file if it doesn't exist
+    # this is a fallback if auth_with_creds fails
+    if not os.getenv("NETRC"):
+        create_asf_netrc_file(asf_login, asf_pass)
+
+    try:
+        logger.info("Attempting to authenticate with ASF session")
+        session = asf_search.ASFSession()
+        session.auth_with_creds(asf_login, asf_pass)
+        use_session = True
+    except:
+        logger.error(
+            "ASF session authentication failed. Attempting fo fall back to .netrc file.",
+            exc_info=True,
+        )
+        use_session = False
 
     if make_folder:
         os.makedirs(download_folder, exist_ok=True)
@@ -167,7 +226,10 @@ def download_scene_from_asf(
         logger.info(f"Skipping download, zipped scene exists at : {scene_zip_path}")
     else:
         try:
-            asf_scene_metadata.download(path=download_folder, session=session)
+            if use_session:
+                asf_scene_metadata.download(path=download_folder, session=session)
+            else:
+                asf_scene_metadata.download(path=download_folder)
         except:
             logger.error(
                 "An error occurred while running the download command",