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

CHANGELOG.md

@@ -7,18 +7,25 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 ## [Unreleased]
 
-### Changed
+### Added
 - `search_datasets` now accepts a `has_granules` keyword argument. Use
   `has_granules=False` to search for metadata about collections with no
   associated granules. The default value set in `DataCollections` remains `True`.
   ([#939](https://github.com/nsidc/earthaccess/issues/939))
   ([**@juliacollins**](https://github.com/juliacollins))
 
+### Changed
+- **Breaking**: Throw an error by default when login credentials are rejected. Pass
+  `earthaccess.login(..., raise_on_failure=False)` to disable and print a warning
+  instead. ([#946](https://github.com/nsidc/earthaccess/pull/946))
+  ([**@mfisher87**](https://github.com/mfisher87))
+
 ## [v0.13.0] - 2025-01-28
 
 ### Changed
-- Integration tests: Test are no longer randomized! this means each fail should be reproducible, we are testing the most
-  popular datasets from all DAACs, see files under tests/integration/popular_collections.
+- Integration tests: Test are no longer randomized! this means each fail should be
+  reproducible, we are testing the most popular datasets from all DAACs, see files under
+  tests/integration/popular_collections.
   ([#215](https://github.com/nsidc/earthaccess/issues/215))
   ([**@mfisher87**](https://github.com/mfisher87))
 

docs/user_guide/backwards-compatibility.md

@@ -43,6 +43,8 @@ Our project follows the [SPEC0](https://scientific-python.org/specs/spec-0000/)
 
 ## Migration guides
 
-Under Development 🚧
+### 0.14.0
 
-This section will contain migration guides for future releases. Check back soon for updates!
+* `earthaccess.login()` will now raise an exception by default if Earthdata Login rejects credentials.
+  We recommend you adjust your code to handle this. If you want to disable
+  this, pass `earthaccess.login(..., raise_on_failure=False)`.

earthaccess/api.py

@@ -4,9 +4,18 @@
 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,
+    Mapping,
+    Optional,
+    Union,
+    deprecated,
+)
 
 import earthaccess
+from earthaccess.exceptions import LoginStrategyUnavailable
 from earthaccess.services import DataServices
 
 from .auth import Auth
@@ -161,7 +170,13 @@ 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,
+    *,
+    raise_on_failure: bool = True,
+) -> 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
+        raise_on_failure: Whether to raise an error when login fails; if `False`, only a warning will be printed
 
     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,
+                    raise_on_failure=raise_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,
+            raise_on_failure=raise_on_failure,
+        )
         if earthaccess.__auth__.authenticated:
             earthaccess.__store__ = Store(earthaccess.__auth__)
 

earthaccess/auth.py

@@ -13,8 +13,9 @@
 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,
+        *,
+        raise_on_failure: bool = True,
     ) -> 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
+            raise_on_failure: Whether to raise an error when login fails; if `False`, only a warning will be printed
 
         Returns:
             An instance of Auth.
@@ -124,11 +128,11 @@ def login(
             return self
 
         if strategy == "interactive":
-            self._interactive(persist)
+            self._interactive(persist, raise_on_failure=raise_on_failure)
         elif strategy == "netrc":
-            self._netrc()
+            self._netrc(raise_on_failure=raise_on_failure)
         elif strategy == "environment":
-            self._environment()
+            self._environment(raise_on_failure=raise_on_failure)
 
         return self
 
@@ -234,63 +238,109 @@ 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,
+        *,
+        raise_on_failure: bool,
+    ) -> 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,
+            raise_on_failure=raise_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,
+        *,
+        raise_on_failure: bool,
+    ) -> 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,
+            raise_on_failure=raise_on_failure,
+        )
 
         if authenticated:
             logger.debug("Using .netrc file for EDL")
 
         return authenticated
 
-    def _environment(self) -> bool:
+    def _environment(
+        self,
+        *,
+        raise_on_failure: bool,
+    ) -> 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,
+            raise_on_failure=raise_on_failure,
+        )
 
     def _get_credentials(
-        self, username: Optional[str], password: Optional[str]
+        self,
+        username: Optional[str],
+        password: Optional[str],
+        *,
+        raise_on_failure: bool,
     ) -> 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 raise_on_failure:
+                    logger.error(msg)
+                    raise LoginAttemptFailure(msg)
+                else:
+                    logger.warning(msg)
+                    return False
+
+            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."""

tests/integration/test_auth.py

@@ -5,6 +5,7 @@
 import pytest
 import requests
 import s3fs
+from earthaccess.exceptions import LoginStrategyUnavailable
 
 logger = logging.getLogger(__name__)
 
@@ -24,8 +25,8 @@ def test_auth_can_read_from_netrc_file(mock_netrc):
     assert auth.authenticated
 
 
-def test_auth_throws_exception_if_netrc_is_not_present(mock_missing_netrc):
-    with pytest.raises(FileNotFoundError):
+def test_auth_strategy_unavailable_netrc_is_not_present(mock_missing_netrc):
+    with pytest.raises(LoginStrategyUnavailable):
         earthaccess.login(strategy="netrc")
 
 

tests/unit/test_auth.py

@@ -6,6 +6,7 @@
 import pytest
 import responses
 from earthaccess import Auth
+from earthaccess.exceptions import LoginAttemptFailure
 
 logger = logging.getLogger(__name__)
 
@@ -95,9 +96,7 @@ def test_auth_fails_for_wrong_credentials(self, user_input, user_password):
         )
         # Test
         auth = Auth()
-        auth.login(strategy="interactive")
-        with pytest.raises(Exception) as e_info:
-            self.assertEqual(auth.authenticated, False)
-            self.assertEqual(e_info, Exception)
-            self.assertEqual(auth.password, "password")
-            self.assertEqual(auth.token, json_response)
+        with pytest.raises(LoginAttemptFailure):
+            auth.login(strategy="interactive")
+
+        self.assertEqual(auth.authenticated, False)