Change auth strategy to using find_or_create_token endpoint (#782)

Co-authored-by: Chuck Daniels <cjdaniels4@gmail.com>

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Unreleased
 
+- Automatically refresh EDL token and deprecate the `Auth.refresh_tokens` method with no replacement, as there is no longer a need to explicitly refresh ([#484](https://github.com/nsidc/earthaccess/issues/484)) ([**@fwfichtner**](https://github.com/fwfichtner))
 - Deprecate `earthaccess.get_s3fs_session` and `Store.get_s3fs_session`. Use
  `earthaccess.get_s3_filesystem` and `Store.get_s3_filesystem`, respectively,
  instead ([#766](https://github.com/nsidc/earthaccess/issues/766))([**@Sherwin-14**](https://github.com/Sherwin-14))

earthaccess/auth.py

@@ -6,11 +6,12 @@
 import shutil
 from netrc import NetrcParseError
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, 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
@@ -72,7 +73,7 @@ class Auth(object):
  def __init__(self) -> None:
  # Maybe all these predefined URLs should be in a constants.py file
  self.authenticated = False
- self.tokens: List = []
+ self.token: Optional[Mapping[str, str]] = None
  self._set_earthdata_system(PROD)
 
  def login(
@@ -116,70 +117,23 @@ def _set_earthdata_system(self, system: System) -> None:
  self.system = system
 
  # Maybe all these predefined URLs should be in a constants.py file
- self.EDL_GET_TOKENS_URL = f"https://{self.system.edl_hostname}/api/users/tokens"
- self.EDL_GENERATE_TOKENS_URL = (
- f"https://{self.system.edl_hostname}/api/users/token"
- )
- self.EDL_REVOKE_TOKEN = (
- f"https://{self.system.edl_hostname}/api/users/revoke_token"
+ self.EDL_FIND_OR_CREATE_TOKEN_URL = (
+ f"https://{self.system.edl_hostname}/api/users/find_or_create_token"
  )
 
  self._eula_url = (
  f"https://{self.system.edl_hostname}/users/earthaccess/unaccepted_eulas"
  )
  self._apps_url = f"https://{self.system.edl_hostname}/application_search"
 
+ @deprecated("No replacement, as tokens are now refreshed automatically.")
  def refresh_tokens(self) -> bool:
  """Refresh CMR tokens.
 
  Tokens are used to do authenticated queries on CMR for restricted and early access datasets.
  This method renews the tokens to make sure we can query the collections allowed to our EDL user.
  """
- if len(self.tokens) == 0:
- resp_tokens = self._generate_user_token(
- username=self.username, password=self.password
- )
- if resp_tokens.ok:
- self.token = resp_tokens.json()
- self.tokens = [self.token]
- logger.debug(
- f"earthaccess generated a token for CMR with expiration on: {self.token['expiration_date']}"
- )
- return True
- else:
- return False
- if len(self.tokens) == 1:
- resp_tokens = self._generate_user_token(
- username=self.username, password=self.password
- )
- if resp_tokens.ok:
- self.token = resp_tokens.json()
- self.tokens.extend(self.token)
- logger.debug(
- f"earthaccess generated a token for CMR with expiration on: {self.token['expiration_date']}"
- )
- return True
- else:
- return False
-
- if len(self.tokens) == 2:
- resp_revoked = self._revoke_user_token(self.token["access_token"])
- if resp_revoked:
- resp_tokens = self._generate_user_token(
- username=self.username, password=self.password
- )
- if resp_tokens.ok:
- self.token = resp_tokens.json()
- self.tokens[0] = self.token
- logger.debug(
- f"earthaccess generated a token for CMR with expiration on: {self.token['expiration_date']}"
- )
- return True
- else:
- logger.info(resp_tokens)
- return False
-
- return False
+ return self.authenticated
 
  def get_s3_credentials(
  self,
@@ -253,7 +207,7 @@ def get_session(self, bearer_token: bool = True) -> requests.Session:
  class Session instance with Auth and bearer token headers
  """
  session = SessionWithHeaderRedirection()
- if bearer_token and self.authenticated:
+ if bearer_token and self.token:
  # This will avoid the use of the netrc after we are logged in
  session.trust_env = False
  session.headers.update(
@@ -306,7 +260,7 @@ def _get_credentials(
  self, username: Optional[str], password: Optional[str]
  ) -> bool:
  if username is not None and password is not None:
- token_resp = self._get_user_tokens(username, password)
+ token_resp = self._find_or_create_token(username, password)
 
  if not (token_resp.ok): # type: ignore
  logger.info(
@@ -317,60 +271,26 @@ def _get_credentials(
  self.username = username
  self.password = password
 
- self.tokens = token_resp.json()
+ token = token_resp.json()
+ logger.debug(
+ f"Using token with expiration date: {token['expiration_date']}"
+ )
+ self.token = token
  self.authenticated = True
 
- if len(self.tokens) == 0:
- self.refresh_tokens()
- logger.debug(
- f"earthaccess generated a token for CMR with expiration on: {self.token['expiration_date']}"
- )
- self.token = self.tokens[0]
- elif len(self.tokens) > 0:
- self.token = self.tokens[0]
- logger.debug(
- f"Using token with expiration date: {self.token['expiration_date']}"
- )
-
  return self.authenticated
 
- def _get_user_tokens(self, username: str, password: str) -> Any:
- session = SessionWithHeaderRedirection(username, password)
- auth_resp = session.get(
- self.EDL_GET_TOKENS_URL,
- headers={
- "Accept": "application/json",
- },
- timeout=10,
- )
- return auth_resp
-
- def _generate_user_token(self, username: str, password: str) -> Any:
+ def _find_or_create_token(self, username: str, password: str) -> Any:
  session = SessionWithHeaderRedirection(username, password)
  auth_resp = session.post(
- self.EDL_GENERATE_TOKENS_URL,
+ self.EDL_FIND_OR_CREATE_TOKEN_URL,
  headers={
  "Accept": "application/json",
  },
  timeout=10,
  )
  return auth_resp
 
- def _revoke_user_token(self, token: str) -> bool:
- if self.authenticated:
- session = SessionWithHeaderRedirection(self.username, self.password)
- auth_resp = session.post(
- self.EDL_REVOKE_TOKEN,
- params={"token": token},
- headers={
- "Accept": "application/json",
- },
- timeout=10,
- )
- return auth_resp.ok
- else:
- return False
-
  def _persist_user_credentials(self, username: str, password: str) -> bool:
  # See: https://github.com/sloria/tinynetrc/issues/34
  try:

tests/unit/test_auth.py

@@ -17,13 +17,11 @@ class TestCreateAuth(unittest.TestCase):
  def test_auth_gets_proper_credentials(self, user_input, user_password):
  user_input.return_value = "user"
  user_password.return_value = "password"
- json_response = [
- {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"},
- {"access_token": "EDL-token-2", "expiration_date": "12/16/2021"},
- ]
+ json_response = {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"}
+
  responses.add(
- responses.GET,
- "https://urs.earthdata.nasa.gov/api/users/tokens",
+ responses.POST,
+ "https://urs.earthdata.nasa.gov/api/users/find_or_create_token",
  json=json_response,
  status=200,
  )
@@ -41,7 +39,7 @@ def test_auth_gets_proper_credentials(self, user_input, user_password):
  self.assertEqual(auth.authenticated, False)
  auth.login(strategy="interactive")
  self.assertEqual(auth.authenticated, True)
- self.assertTrue(auth.token in json_response)
+ self.assertEqual(auth.token, json_response)
 
  # test that we are creating a session with the proper headers
  self.assertTrue("User-Agent" in headers)
@@ -56,9 +54,9 @@ def test_auth_can_create_proper_credentials(self, user_input, user_password):
  json_response = {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"}
 
  responses.add(
- responses.GET,
- "https://urs.earthdata.nasa.gov/api/users/tokens",
- json=[],
+ responses.POST,
+ "https://urs.earthdata.nasa.gov/api/users/find_or_create_token",
+ json=json_response,
  status=200,
  )
  responses.add(
@@ -67,12 +65,7 @@ def test_auth_can_create_proper_credentials(self, user_input, user_password):
  json={},
  status=200,
  )
- responses.add(
- responses.POST,
- "https://urs.earthdata.nasa.gov/api/users/token",
- json=json_response,
- status=200,
- )
+
  # Test
  auth = Auth()
  auth.login(strategy="interactive")
@@ -89,8 +82,8 @@ def test_auth_fails_for_wrong_credentials(self, user_input, user_password):
  json_response = {"error": "wrong credentials"}
 
  responses.add(
- responses.GET,
- "https://urs.earthdata.nasa.gov/api/users/tokens",
+ responses.POST,
+ "https://urs.earthdata.nasa.gov/api/users/find_or_create_token",
  json=json_response,
  status=401,
  )
@@ -100,12 +93,6 @@ def test_auth_fails_for_wrong_credentials(self, user_input, user_password):
  json=json_response,
  status=401,
  )
- responses.add(
- responses.POST,
- "https://urs.earthdata.nasa.gov/api/users/token",
- json=json_response,
- status=401,
- )
  # Test
  auth = Auth()
  auth.login(strategy="interactive")

tests/unit/test_deprecations.py

@@ -16,13 +16,11 @@
 def auth(user_input, user_password):
  user_input.return_value = "user"
  user_password.return_value = "password"
- json_response = [
- {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"},
- {"access_token": "EDL-token-2", "expiration_date": "12/16/2021"},
- ]
+ json_response = {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"}
+
  responses.add(
- responses.GET,
- "https://urs.earthdata.nasa.gov/api/users/tokens",
+ responses.POST,
+ "https://urs.earthdata.nasa.gov/api/users/find_or_create_token",
  json=json_response,
  status=200,
  )

tests/unit/test_store.py

@@ -14,20 +14,17 @@ class TestStoreSessions(unittest.TestCase):
  def setUp(self):
  os.environ["EARTHDATA_USERNAME"] = "user"
  os.environ["EARTHDATA_PASSWORD"] = "password"
- json_response = [
- {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"},
- {"access_token": "EDL-token-2", "expiration_date": "12/16/2021"},
- ]
+ json_response = {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"}
  responses.add(
- responses.GET,
- "https://urs.earthdata.nasa.gov/api/users/tokens",
+ responses.POST,
+ "https://urs.earthdata.nasa.gov/api/users/find_or_create_token",
  json=json_response,
  status=200,
  )
  self.auth = Auth()
  self.auth.login(strategy="environment")
  self.assertEqual(self.auth.authenticated, True)
- self.assertTrue(self.auth.token in json_response)
+ self.assertEqual(self.auth.token, json_response)
 
  def tearDown(self):
  self.auth = None

tests/unit/test_uat.py

@@ -17,13 +17,10 @@ class TestUatEnvironmentArgument:
  )
  def test_uat_login_when_uat_selected(self):
  """Test the correct env is queried based on what's selected at login-time."""
- json_response = [
- {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"},
- {"access_token": "EDL-token-2", "expiration_date": "12/16/2021"},
- ]
+ json_response = {"access_token": "EDL-token-1", "expiration_date": "12/15/2021"}
  responses.add(
- responses.GET,
- "https://uat.urs.earthdata.nasa.gov/api/users/tokens",
+ responses.POST,
+ "https://uat.urs.earthdata.nasa.gov/api/users/find_or_create_token",
  json=json_response,
  status=200,
  )