Documentation for Auth

Bases: object

Authentication class for operations that require Earthdata login (EDL).

Source code in earthaccess/auth.py

def __init__(self) -> None:
    # Maybe all these predefined URLs should be in a constants.py file
    self.authenticated = False
    self.tokens: List = []
    self.EDL_GET_TOKENS_URL = "https://urs.earthdata.nasa.gov/api/users/tokens"
    self.EDL_GET_PROFILE = "https://urs.earthdata.nasa.gov/api/users/<USERNAME>?client_id=ntD0YGC_SM3Bjs-Tnxd7bg"
    self.EDL_GENERATE_TOKENS_URL = "https://urs.earthdata.nasa.gov/api/users/token"
    self.EDL_REVOKE_TOKEN = "https://urs.earthdata.nasa.gov/api/users/revoke_token"

get_s3_credentials(daac=None, provider=None, endpoint=None)

Gets AWS S3 credentials for a given NASA cloud provider. The easier way is to use the DAAC short name; provider is optional if we know it.

Parameters:

Name	Type	Description	Default
daac	Optional[str]	The name of a NASA DAAC, e.g. NSIDC or PODAAC.	None
provider	Optional[str]	A valid cloud provider. Each DAAC has a provider code for their cloud distributions.	None
endpoint	Optional[str]	Getting the credentials directly from the S3Credentials URL.	None

Returns:

Type	Description
Dict[str, str]	A Python dictionary with the temporary AWS S3 credentials.

Source code in earthaccess/auth.py

def get_s3_credentials(
    self,
    daac: Optional[str] = None,
    provider: Optional[str] = None,
    endpoint: Optional[str] = None,
) -> Dict[str, str]:
    """Gets AWS S3 credentials for a given NASA cloud provider.
    The easier way is to use the DAAC short name; provider is optional if we know it.

    Parameters:
        daac: The name of a NASA DAAC, e.g. NSIDC or PODAAC.
        provider: A valid cloud provider. Each DAAC has a provider code for their cloud distributions.
        endpoint: Getting the credentials directly from the S3Credentials URL.

    Returns:
        A Python dictionary with the temporary AWS S3 credentials.
    """
    if self.authenticated:
        session = SessionWithHeaderRedirection(self.username, self.password)
        if endpoint is None:
            auth_url = self._get_cloud_auth_url(
                daac_shortname=daac, provider=provider
            )
        else:
            auth_url = endpoint
        if auth_url.startswith("https://"):
            cumulus_resp = session.get(auth_url, timeout=15, allow_redirects=True)
            auth_resp = session.get(
                cumulus_resp.url, allow_redirects=True, timeout=15
            )
            if not (auth_resp.ok):  # type: ignore
                # Let's try to authenticate with Bearer tokens
                _session = self.get_session()
                cumulus_resp = _session.get(
                    auth_url, timeout=15, allow_redirects=True
                )
                auth_resp = _session.get(
                    cumulus_resp.url, allow_redirects=True, timeout=15
                )
                if not (auth_resp.ok):
                    print(
                        f"Authentication with Earthdata Login failed with:\n{auth_resp.text[0:1000]}"
                    )
                    eula_url = "https://urs.earthdata.nasa.gov/users/earthaccess/unaccepted_eulas"
                    apps_url = "https://urs.earthdata.nasa.gov/application_search"
                    print(
                        f"Consider accepting the EULAs available at {eula_url} and applications at {apps_url}"
                    )
                    return {}

                return auth_resp.json()
            return auth_resp.json()
        else:
            # This happens if the cloud provider doesn't list the S3 credentials or the DAAC
            # does not have cloud collections yet
            print(f"Credentials for the cloud provider {daac} are not available")
            return {}
    else:
        print("We need to authenticate with EDL first")
        return {}

get_session(bearer_token=True)

Returns a new request session instance.

Parameters:

Name	Type	Description	Default
bearer_token	bool	whether to include bearer token	True

Returns:

Type	Description
Session	class Session instance with Auth and bearer token headers

Source code in earthaccess/auth.py

def get_session(self, bearer_token: bool = True) -> requests.Session:
    """Returns a new request session instance.

    Parameters:
        bearer_token: whether to include bearer token

    Returns:
        class Session instance with Auth and bearer token headers
    """
    session = SessionWithHeaderRedirection()
    if bearer_token and self.authenticated:
        # This will avoid the use of the netrc after we are logged in
        session.trust_env = False
        session.headers.update(
            {"Authorization": f'Bearer {self.token["access_token"]}'}
        )
    return session

login(strategy='netrc', persist=False)

Authenticate with Earthdata login.

Parameters:

Name	Type	Description	Default
strategy	str	The authentication method. "interactive": Enter a username and password. "netrc": (default) Retrieve a username and password from ~/.netrc. "environment": Retrieve a username and password from $EARTHDATA_USERNAME and $EARTHDATA_PASSWORD.	'netrc'
persist	bool	Will persist credentials in a .netrc file.	False

Returns:

Type	Description
Any	An instance of Auth.

Source code in earthaccess/auth.py

def login(self, strategy: str = "netrc", persist: bool = False) -> Any:
    """Authenticate with Earthdata login.

    Parameters:
        strategy:
            The authentication method.

            * **"interactive"**: Enter a username and password.
            * **"netrc"**: (default) Retrieve a username and password from ~/.netrc.
            * **"environment"**:
                Retrieve a username and password from $EARTHDATA_USERNAME and $EARTHDATA_PASSWORD.
        persist: Will persist credentials in a `.netrc` file.

    Returns:
        An instance of Auth.
    """
    if self.authenticated:
        logger.debug("We are already authenticated with NASA EDL")
        return self
    if strategy == "interactive":
        self._interactive(persist)
    if strategy == "netrc":
        self._netrc()
    if strategy == "environment":
        self._environment()
    return self

refresh_tokens()

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.

Source code in earthaccess/auth.py

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:
                print(resp_tokens)
                return False

    return False