Source code for globus_sdk.auth.client_types.base

from __future__ import print_function, unicode_literals

import collections
import logging

import six

from globus_sdk import exc
from globus_sdk.auth.token_response import OAuthTokenResponse
from globus_sdk.authorizers import NullAuthorizer
from globus_sdk.base import BaseClient, safe_stringify

logger = logging.getLogger(__name__)


[docs]class AuthClient(BaseClient): """ Client for the `Globus Auth API <https://docs.globus.org/api/auth/>`_ This class provides helper methods for most common resources in the Auth API, and the common low-level interface from :class:`BaseClient <globus_sdk.base.BaseClient>` of ``get``, ``put``, ``post``, and ``delete`` methods, which can be used to access any API resource. There are generally two types of resources, distinguished by the type of authentication which they use. Resources available to end users of Globus are authenticated with a Globus Auth Token ("Authentication: Bearer ..."), while resources available to OAuth Clients are authenticated using Basic Auth with the Client's ID and Secret. Some resources may be available with either authentication type. **Examples** Initializing an ``AuthClient`` to authenticate a user making calls to the Globus Auth service with an access token takes the form >>> from globus_sdk import AuthClient, AccessTokenAuthorizer >>> ac = AuthClient(authorizer=AccessTokenAuthorizer('<token_string>')) You can, of course, use other kinds of Authorizers (notably the ``RefreshTokenAuthorizer``). .. automethodlist:: globus_sdk.AuthClient """ error_class = exc.AuthAPIError def __init__(self, client_id=None, authorizer=None, **kwargs): self.client_id = client_id # an AuthClient may contain a GlobusOAuth2FlowManager in order to # encapsulate the functionality of various different types of flow # managers self.current_oauth2_flow_manager = None BaseClient.__init__(self, "auth", authorizer=authorizer, **kwargs)
[docs] def get_identities(self, usernames=None, ids=None, provision=False, **params): r""" GET /v2/api/identities Given ``usernames=<U>`` or (exclusive) ``ids=<I>`` as keyword arguments, looks up identity information for the set of identities provided. ``<U>`` and ``<I>`` in this case are comma-delimited strings listing multiple Identity Usernames or Identity IDs, or iterables of strings, each of which is an Identity Username or Identity ID. If Globus Auth's identity auto-provisioning behavior is desired, ``provision=True`` may be specified. Available with any authentication/client type. **Examples** >>> ac = globus_sdk.AuthClient(...) >>> # by IDs >>> r = ac.get_identities(ids="46bd0f56-e24f-11e5-a510-131bef46955c") >>> r.data {u'identities': [{u'email': None, u'id': u'46bd0f56-e24f-11e5-a510-131bef46955c', u'identity_provider': u'7daddf46-70c5-45ee-9f0f-7244fe7c8707', u'name': None, u'organization': None, u'status': u'unused', u'username': u'globus@globus.org'}]} >>> ac.get_identities( >>> ids=",".join( >>> ("46bd0f56-e24f-11e5-a510-131bef46955c", >>> "168edc3d-c6ba-478c-9cf8-541ff5ebdc1c")) ... >>> # or by usernames >>> ac.get_identities(usernames='globus@globus.org') ... >>> ac.get_identities( >>> usernames='globus@globus.org,auth@globus.org') ... You could also use iterables: >>> ac.get_identities( >>> usernames=['globus@globus.org', 'auth@globus.org']) ... >>> ac.get_identities( >>> ids=["46bd0f56-e24f-11e5-a510-131bef46955c", >>> "168edc3d-c6ba-478c-9cf8-541ff5ebdc1c"]) ... **External Documentation** See `Identities Resources \ <https://docs.globus.org/api/auth/reference/ #v2_api_identities_resources>`_ in the API documentation for details. """ def _convert_listarg(val): if isinstance(val, collections.Iterable) and not isinstance( val, six.string_types ): return ",".join(safe_stringify(x) for x in val) else: return safe_stringify(val) self.logger.info("Looking up Globus Auth Identities") # if either of these params has a truthy value, stringify it safely, # letting us consume args whose `__str__` methods produce "the right # thing" # most notably, lets `ids` take a single UUID object safely if usernames: params["usernames"] = _convert_listarg(usernames) params["provision"] = ( "false" if str(provision).lower() == "false" else "true" ) if ids: params["ids"] = _convert_listarg(ids) self.logger.debug("params={}".format(params)) if "usernames" in params and "ids" in params: self.logger.warning( ( "get_identities call with both usernames and " "identities set! Expected to result in errors" ) ) return self.get("/v2/api/identities", params=params)
[docs] def oauth2_get_authorize_url(self, additional_params=None): """ Get the authorization URL to which users should be sent. This method may only be called after ``oauth2_start_flow`` has been called on this ``AuthClient``. :param additional_params: Additional query parameters to include in the authorize URL. Primarily for internal use :type additional_params: dict, optional :rtype: ``string`` """ if not self.current_oauth2_flow_manager: self.logger.error( ("OutOfOrderOperations(" "get_authorize_url before start_flow)") ) raise exc.GlobusSDKUsageError( ( "Cannot get authorize URL until starting an OAuth2 flow. " "Call the oauth2_start_flow() method on this " "AuthClient to resolve" ) ) auth_url = self.current_oauth2_flow_manager.get_authorize_url( additional_params=additional_params ) self.logger.info("Got authorization URL: {}".format(auth_url)) return auth_url
[docs] def oauth2_exchange_code_for_tokens(self, auth_code): """ Exchange an authorization code for a token or tokens. :rtype: :class:`OAuthTokenResponse \ <globus_sdk.auth.token_response.OAuthTokenResponse>` :param auth_code: An auth code typically obtained by sending the user to the authorize URL. The code is a very short-lived credential which this method is exchanging for tokens. Tokens are the credentials used to authenticate against Globus APIs. :type auth_code: str """ self.logger.info( ( "Final Step of 3-legged OAuth2 Flows: " "Exchanging authorization code for token(s)" ) ) if not self.current_oauth2_flow_manager: self.logger.error( ("OutOfOrderOperations(" "exchange_code before start_flow)") ) raise exc.GlobusSDKUsageError( ( "Cannot exchange auth code until starting an OAuth2 flow. " "Call the oauth2_start_flow() method on this " "AuthClient to resolve" ) ) return self.current_oauth2_flow_manager.exchange_code_for_tokens(auth_code)
[docs] def oauth2_refresh_token(self, refresh_token, additional_params=None): r""" Exchange a refresh token for a :class:`OAuthTokenResponse <globus_sdk.auth.token_response.OAuthTokenResponse>`, containing an access token. Does a token call of the form .. code-block:: none refresh_token=<refresh_token> grant_type=refresh_token plus any additional parameters you may specify. :param refresh_token: A Globus Refresh Token as a string :type refresh_token: str :param additional_params: A dict of extra params to encode in the refresh call. :type additional_params: dict, optional """ self.logger.info( ("Executing token refresh; " "typically requires client credentials") ) form_data = {"refresh_token": refresh_token, "grant_type": "refresh_token"} if additional_params: form_data.update(additional_params) return self.oauth2_token(form_data)
[docs] def oauth2_validate_token(self, token, additional_params=None): """ Validate a token. It can be an Access Token or a Refresh token. This call can be used to check tokens issued to your client, confirming that they are or are not still valid. The resulting response has the form ``{"active": True}`` when the token is valid, and ``{"active": False}`` when it is not. It is not necessary to validate tokens immediately after receiving them from the service -- any tokens which you are issued will be valid at that time. This is more for the purpose of doing checks like - confirm that ``oauth2_revoke_token`` succeeded - at application boot, confirm no need to do fresh login :param token: The token which should be validated. Can be a refresh token or an access token :type token: str :param additional_params: Additional parameters to include in the validation body. Primarily for internal use :type additional_params: dict, optional **Examples** Revoke a token and confirm that it is no longer active: >>> from globus_sdk import ConfidentialAppAuthClient >>> ac = ConfidentialAppAuthClient(CLIENT_ID, CLIENT_SECRET) >>> ac.oauth2_revoke_token('<token_string>') >>> data = ac.oauth2_validate_token('<token_string>') >>> assert not data['active'] During application boot, check if the user needs to do a login, even if a token is present: >>> from globus_sdk import ConfidentialAppAuthClient >>> ac = ConfidentialAppAuthClient(CLIENT_ID, CLIENT_SECRET) >>> # this is not an SDK function, but a hypothetical function which >>> # you use to load a token out of configuration data >>> tok = load_token_from_config(...) >>> >>> if not tok or not ac.oauth2_validate_token(tok)['active']: >>> # do_new_login() is another hypothetical helper >>> tok = do_new_login() >>> # at this point, tok is expected to be a valid token """ self.logger.info("Validating token") body = {"token": token} # if this client has no way of authenticating itself but # it does have a client_id, we'll send that in the request no_authentication = self.authorizer is None or isinstance( self.authorizer, NullAuthorizer ) if no_authentication and self.client_id: self.logger.debug("Validating token with unauthenticated client") body.update({"client_id": self.client_id}) if additional_params: body.update(additional_params) return self.post("/v2/oauth2/token/validate", text_body=body)
[docs] def oauth2_revoke_token(self, token, additional_params=None): """ Revoke a token. It can be an Access Token or a Refresh token. This call should be used to revoke tokens issued to your client, rendering them inert and not further usable. Typically, this is incorporated into "logout" functionality, but it should also be used if the client detects that its tokens are in an unsafe location (e.x. found in a world-readable logfile). You can check the "active" status of the token after revocation if you want to confirm that it was revoked. :param token: The token which should be revoked :type token: str :param additional_params: Additional parameters to include in the revocation body, which can help speed the revocation process. Primarily for internal use **Examples** >>> from globus_sdk import ConfidentialAppAuthClient >>> ac = ConfidentialAppAuthClient(CLIENT_ID, CLIENT_SECRET) >>> ac.oauth2_revoke_token('<token_string>') """ self.logger.info("Revoking token") body = {"token": token} # if this client has no way of authenticating itself but # it does have a client_id, we'll send that in the request no_authentication = self.authorizer is None or isinstance( self.authorizer, NullAuthorizer ) if no_authentication and self.client_id: self.logger.debug("Revoking token with unauthenticated client") body.update({"client_id": self.client_id}) if additional_params: body.update(additional_params) return self.post("/v2/oauth2/token/revoke", text_body=body)
[docs] def oauth2_token(self, form_data, response_class=OAuthTokenResponse): """ This is the generic form of calling the OAuth2 Token endpoint. It takes ``form_data``, a dict which will be encoded in a form POST body on the request. Generally, users of the SDK should not call this method unless they are implementing OAuth2 flows. :param response_class: This is used by calls to the oauth2_token endpoint which need to specialize their responses. For example, :meth:`oauth2_get_dependent_tokens \ <globus_sdk.ConfidentialAppAuthClient.oauth2_get_dependent_tokens>` requires a specialize response class to handle the dramatically different format of the Dependent Token Grant response :type response_class: class, optional :rtype: ``response_class`` """ self.logger.info("Fetching new token from Globus Auth") # use the fact that requests implicitly encodes the `data` parameter as # a form POST return self.post( "/v2/oauth2/token", response_class=response_class, text_body=form_data )
[docs] def oauth2_userinfo(self): """ Call the Userinfo endpoint of Globus Auth. Userinfo is specified as part of the OpenID Connect (OIDC) standard, and Globus Auth's Userinfo is OIDC-compliant. The exact data returned will depend upon the set of OIDC-related scopes which were used to acquire the token being used for this call. For details, see the **External Documentation** below. **Examples** >>> ac = AuthClient(...) >>> info = ac.oauth2_userinfo() >>> print('Effective Identity "{}" has Full Name "{}" and Email "{}"' >>> .format(info["sub"], info["name"], info["email"])) **External Documentation** See `Userinfo \ <https://docs.globus.org/api/auth/reference/ #get_or_post_v2_oauth2_userinfo_resource>`_ in the API documentation for details. """ self.logger.info("Looking up OIDC-style Userinfo from Globus Auth") return self.get("/v2/oauth2/userinfo")