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.token: Optional[Mapping[str, str]] = None
    self._set_earthdata_system(PROD)

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):
                    logger.error(
                        f"Authentication with Earthdata Login failed with:\n{auth_resp.text[0:1000]}"
                    )
                    logger.error(
                        f"Consider accepting the EULAs available at {self._eula_url} and applications at {self._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
            logger.info(
                f"Credentials for the cloud provider {daac} are not available"
            )
            return {}
    else:
        logger.info("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.token:
        # 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, system=None)

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
system	Optional[System]	the EDL endpoint to log in to Earthdata, defaults to PROD	None

Returns:

Type	Description
Any	An instance of Auth.

Source code in earthaccess/auth.py

def login(
    self,
    strategy: str = "netrc",
    persist: bool = False,
    system: Optional[System] = None,
) -> 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.
        system: the EDL endpoint to log in to Earthdata, defaults to PROD

    Returns:
        An instance of Auth.
    """
    if system is not None:
        self._set_earthdata_system(system)

    if self.authenticated and (system == self.system):
        logger.debug("We are already authenticated with NASA EDL")
        return self

    if strategy == "interactive":
        self._interactive(persist)
    elif strategy == "netrc":
        self._netrc()
    elif 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

@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.
    """
    return self.authenticated