Allow user to control behavior when login fails; warn by default

earthaccess/api.py

@@ -4,9 +4,19 @@
 import requests
 import s3fs
 from fsspec import AbstractFileSystem
-from typing_extensions import Any, Dict, List, Mapping, Optional, Union, deprecated
+from typing_extensions import (
+    Any,
+    Dict,
+    List,
+    Literal,
+    Mapping,
+    Optional,
+    Union,
+    deprecated,
+)
 
 import earthaccess
+from earthaccess.exceptions import LoginStrategyUnavailable
 from earthaccess.services import DataServices
 
 from .auth import Auth
@@ -161,7 +171,12 @@ def search_services(count: int = -1, **kwargs: Any) -> List[Any]:
     return query.get(hits if count < 1 else min(count, hits))
 
 
-def login(strategy: str = "all", persist: bool = False, system: System = PROD) -> Auth:
+def login(
+    strategy: str = "all",
+    persist: bool = False,
+    system: System = PROD,
+    on_failure: Literal["warn", "error"] = "warn",
+) -> Auth:
     """Authenticate with Earthdata login (https://urs.earthdata.nasa.gov/).
 
     Parameters:
@@ -174,6 +189,7 @@ def login(strategy: str = "all", persist: bool = False, system: System = PROD) -
             * **"environment"**: retrieve username and password from `$EARTHDATA_USERNAME` and `$EARTHDATA_PASSWORD`.
         persist: will persist credentials in a .netrc file
         system: the Earthdata system to access, defaults to PROD
+        on_failure: Whether to print a warning or raise an error when login fails.
 
     Returns:
         An instance of Auth.
@@ -186,16 +202,25 @@ def login(strategy: str = "all", persist: bool = False, system: System = PROD) -
         for strategy in ["environment", "netrc", "interactive"]:
             try:
                 earthaccess.__auth__.login(
-                    strategy=strategy, persist=persist, system=system
+                    strategy=strategy,
+                    persist=persist,
+                    system=system,
+                    on_failure=on_failure,
                 )
-            except Exception:
-                pass
+            except LoginStrategyUnavailable as err:
+                logger.debug(err)
+                continue
 
             if earthaccess.__auth__.authenticated:
                 earthaccess.__store__ = Store(earthaccess.__auth__)
                 break
     else:
-        earthaccess.__auth__.login(strategy=strategy, persist=persist, system=system)
+        earthaccess.__auth__.login(
+            strategy=strategy,
+            persist=persist,
+            system=system,
+            on_failure=on_failure,
+        )
         if earthaccess.__auth__.authenticated:
             earthaccess.__store__ = Store(earthaccess.__auth__)
 

earthaccess/auth.py

@@ -6,15 +6,16 @@
 import shutil
 from netrc import NetrcParseError
 from pathlib import Path
-from typing import Any, Dict, List, Mapping, Optional
+from typing import Any, Dict, List, Literal, Mapping, Optional
 from urllib.parse import urlparse
 
 import requests  # type: ignore
 from tinynetrc import Netrc
 from typing_extensions import deprecated
 
-from .daac import DAACS
-from .system import PROD, System
+from earthaccess.daac import DAACS
+from earthaccess.exceptions import LoginAttemptFailure, LoginStrategyUnavailable
+from earthaccess.system import PROD, System
 
 try:
     user_agent = f"earthaccess v{importlib.metadata.version('earthaccess')}"
@@ -99,6 +100,8 @@ def login(
         strategy: str = "netrc",
         persist: bool = False,
         system: Optional[System] = None,
+        *,
+        on_failure: Literal["warn", "error"],
     ) -> Any:
         """Authenticate with Earthdata login.
 
@@ -112,6 +115,7 @@ def login(
                     Retrieve a username and password from $EARTHDATA_USERNAME and $EARTHDATA_PASSWORD.
             persist: Will persist credentials in a `.netrc` file.
             system: the EDL endpoint to log in to Earthdata, defaults to PROD
+            on_failure: Whether to print a warning or raise an error when login fails.
 
         Returns:
             An instance of Auth.
@@ -124,11 +128,11 @@ def login(
             return self
 
         if strategy == "interactive":
-            self._interactive(persist)
+            self._interactive(persist, on_failure=on_failure)
         elif strategy == "netrc":
-            self._netrc()
+            self._netrc(on_failure=on_failure)
         elif strategy == "environment":
-            self._environment()
+            self._environment(on_failure=on_failure)
 
         return self
 
@@ -234,63 +238,96 @@ class Session instance with Auth and bearer token headers
             )
         return session
 
-    def _interactive(self, persist_credentials: bool = False) -> bool:
+    def _interactive(
+        self,
+        persist_credentials: bool = False,
+        *,
+        on_failure: Literal["warn", "error"] = "warn",
+    ) -> bool:
         username = input("Enter your Earthdata Login username: ")
         password = getpass.getpass(prompt="Enter your Earthdata password: ")
-        authenticated = self._get_credentials(username, password)
+        authenticated = self._get_credentials(username, password, on_failure=on_failure)
         if authenticated:
             logger.debug("Using user provided credentials for EDL")
             if persist_credentials:
                 self._persist_user_credentials(username, password)
         return authenticated
 
-    def _netrc(self) -> bool:
+    def _netrc(
+        self,
+        *,
+        on_failure: Literal["warn", "error"],
+    ) -> bool:
         netrc_loc = netrc_path()
 
         try:
             my_netrc = Netrc(str(netrc_loc))
         except FileNotFoundError as err:
-            raise FileNotFoundError(f"No .netrc found at {netrc_loc}") from err
+            raise LoginStrategyUnavailable(f"No .netrc found at {netrc_loc}") from err
         except NetrcParseError as err:
-            raise NetrcParseError(f"Unable to parse .netrc file {netrc_loc}") from err
+            raise LoginStrategyUnavailable(
+                f"Unable to parse .netrc file {netrc_loc}"
+            ) from err
 
-        if (creds := my_netrc[self.system.edl_hostname]) is None:
+        creds = my_netrc[self.system.edl_hostname]
+        if creds is None:
             return False
 
         username = creds["login"]
         password = creds["password"]
-        authenticated = self._get_credentials(username, password)
+
+        if username is None:
+            raise LoginStrategyUnavailable(
+                f"Username not found in .netrc file {netrc_loc}"
+            )
+        if password is None:
+            raise LoginStrategyUnavailable(
+                f"Password not found in .netrc file {netrc_loc}"
+            )
+
+        authenticated = self._get_credentials(username, password, on_failure=on_failure)
 
         if authenticated:
             logger.debug("Using .netrc file for EDL")
 
         return authenticated
 
-    def _environment(self) -> bool:
+    def _environment(
+        self,
+        *,
+        on_failure: Literal["warn", "error"],
+    ) -> bool:
         username = os.getenv("EARTHDATA_USERNAME")
         password = os.getenv("EARTHDATA_PASSWORD")
-        authenticated = self._get_credentials(username, password)
-        if authenticated:
-            logger.debug("Using environment variables for EDL")
-        else:
-            logger.debug(
+
+        if not username or not password:
+            raise LoginStrategyUnavailable(
                 "EARTHDATA_USERNAME and EARTHDATA_PASSWORD are not set in the current environment, try "
                 "setting them or use a different strategy (netrc, interactive)"
             )
-        return authenticated
+
+        logger.debug("Using environment variables for EDL")
+        return self._get_credentials(username, password, on_failure=on_failure)
 
     def _get_credentials(
-        self, username: Optional[str], password: Optional[str]
+        self,
+        username: Optional[str],
+        password: Optional[str],
+        *,
+        on_failure: Literal["warn", "error"],
     ) -> bool:
         if username is not None and password is not None:
             token_resp = self._find_or_create_token(username, password)
 
             if not (token_resp.ok):  # type: ignore
-                logger.info(
-                    f"Authentication with Earthdata Login failed with:\n{token_resp.text}"
-                )
-                return False
-            logger.debug("You're now authenticated with NASA Earthdata Login")
+                msg = f"Authentication with Earthdata Login failed with:\n{token_resp.text}"
+                if on_failure == "warn":
+                    logger.warning(msg)
+                    return False
+                elif on_failure == "error":
+                    raise LoginAttemptFailure(msg)
+
+            logger.info("You're now authenticated with NASA Earthdata Login")
             self.username = username
             self.password = password
 

earthaccess/exceptions.py

@@ -0,0 +1,13 @@
+class LoginStrategyUnavailable(Exception):
+    """The selected login strategy was skipped.
+
+    This should be raised when a login strategy can't be attempted, for example because
+    "environment" was selected but the envvars are not populated.
+
+    DO NOT raise this exception when a login strategy is attempted and failed, for
+    example because credentials were rejected.
+    """
+
+
+class LoginAttemptFailure(Exception):
+    """The login attempt failed."""