Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-06-25 08:26
📄 __init__.py # Copyright 2014 Google Inc. All Rights Reserved. # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the # "Software"), to deal in the Software without restriction, including # without limitation the rights to use, copy, modify, merge, publish, dis- # tribute, sublicense, and/or sell copies of the Software, and to permit # persons to whom the Software is furnished to do so, subject to the fol- # lowing conditions: # # The above copyright notice and this permission notice shall be included # in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS # IN THE SOFTWARE. """Plugin for Boto auth plugin framework, used with GCP OAuth2.0 credentials.""" from __future__ import absolute_import # Import this module so that users can simply import the top-level module. Once # imported, boto will pick up our subclasses of AuthHandler. from gcs_oauth2_boto_plugin import oauth2_plugin from gcs_oauth2_boto_plugin.oauth2_helper import SetFallbackClientIdAndSecret from gcs_oauth2_boto_plugin.oauth2_helper import SetLock	1.51 KB	2025-06-24 09:04
📄 oauth2_client.py # Copyright 2014 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """An OAuth2 client library. This library provides a client implementation of the OAuth2 protocol (see https://developers.google.com/storage/docs/authentication.html#oauth). ** Experimental API This module is experimental and is subject to modification or removal without notice. """ # This implementation is a wrapper around the oauth2client implementation # that implements caching of access tokens independent of refresh # tokens (in the python API client oauth2client, there is a single class that # encapsulates both refresh and access tokens). from __future__ import absolute_import import datetime import errno from hashlib import sha1 import json import logging import os import socket import tempfile import threading # pylint: disable=g-import-not-at-top if os.environ.get('USER_AGENT'): import boto boto.UserAgent += os.environ.get('USER_AGENT') # pylint: disable=g-bad-import-order import boto import httplib2 import oauth2client.client import oauth2client.service_account from google_reauth import reauth_creds import retry_decorator.retry_decorator from google.auth import _helpers from google.auth.crypt import base as crypt_base from google_auth_httplib2 import Request as GoogleAuthRequest from google_auth_httplib2 import AuthorizedHttp from google.oauth2 import service_account import six from six import BytesIO from six.moves import urllib LOG = logging.getLogger('oauth2_client') # Lock used for checking/exchanging refresh token, so multithreaded # operation doesn't attempt concurrent refreshes. token_exchange_lock = threading.Lock() CLOUD_PLATFORM_SCOPE = 'https://www.googleapis.com/auth/cloud-platform' FULL_CONTROL_SCOPE = 'https://www.googleapis.com/auth/devstorage.full_control' REAUTH_SCOPE = 'https://www.googleapis.com/auth/accounts.reauth' DEFAULT_SCOPE = FULL_CONTROL_SCOPE GOOGLE_OAUTH2_DEFAULT_FILE_PASSWORD = 'notasecret' # Note that these scopes don't necessarily correspond to the refresh token # being used. This list is is used for obtaining the RAPT in the reauth # flow, to determine which challenges should be used. We define this at module # level to allow us to override it for other applications if needed. RAPT_SCOPES = [CLOUD_PLATFORM_SCOPE, REAUTH_SCOPE] METADATA_SERVER = 'http://metadata.google.internal' META_TOKEN_URI = (METADATA_SERVER + '/computeMetadata/v1/instance/' 'service-accounts/default/token') META_HEADERS = { 'Metadata-Flavor': 'Google' } # pylint: disable=invalid-name _ServiceAccountCredentials = ( oauth2client.service_account.ServiceAccountCredentials) # pylint: enable=invalid-name # Note: this is copied from gsutil's gslib.cred_types. It should be kept in # sync. Also note that this library does not use HMAC, but it's preserved from # gsutil's copy to maintain compatibility. class CredTypes(object): HMAC = 'HMAC' OAUTH2_SERVICE_ACCOUNT = 'OAuth 2.0 Service Account' OAUTH2_USER_ACCOUNT = 'Oauth 2.0 User Account' GCE = 'GCE' class Error(Exception): """Base exception for the OAuth2 module.""" pass class AuthorizationCodeExchangeError(Error): """Error trying to exchange an authorization code into a refresh token.""" pass class TokenCache(object): """Interface for OAuth2 token caches.""" def PutToken(self, key, value): raise NotImplementedError def GetToken(self, key): raise NotImplementedError class NoopTokenCache(TokenCache): """A stub implementation of TokenCache that does nothing.""" def PutToken(self, key, value): pass def GetToken(self, key): return None class InMemoryTokenCache(TokenCache): """An in-memory token cache. The cache is implemented by a python dict, and inherits the thread-safety properties of dict. """ def __init__(self): super(InMemoryTokenCache, self).__init__() self.cache = dict() def PutToken(self, key, value): LOG.debug('InMemoryTokenCache.PutToken: key=%s', key) self.cache[key] = value def GetToken(self, key): value = self.cache.get(key, None) LOG.debug('InMemoryTokenCache.GetToken: key=%s%s present', key, ' not' if value is None else '') return value class FileSystemTokenCache(TokenCache): """An implementation of a token cache that persists tokens on disk. Each token object in the cache is stored in serialized form in a separate file. The cache file's name can be configured via a path pattern that is parameterized by the key under which a value is cached and optionally the current processes uid as obtained by os.getuid(). Since file names are generally publicly visible in the system, it is important that the cache key does not leak information about the token's value. If client code computes cache keys from token values, a cryptographically strong one-way function must be used. """ def __init__(self, path_pattern=None): """Creates a FileSystemTokenCache. Args: path_pattern: Optional string argument to specify the path pattern for cache files. The argument should be a path with format placeholders '%(key)s' and optionally '%(uid)s'. If the argument is omitted, the default pattern <tmpdir>/oauth2client-tokencache.%(uid)s.%(key)s is used, where <tmpdir> is replaced with the system temp dir as obtained from tempfile.gettempdir(). """ super(FileSystemTokenCache, self).__init__() self.path_pattern = path_pattern if not path_pattern: self.path_pattern = os.path.join( tempfile.gettempdir(), 'oauth2_client-tokencache.%(uid)s.%(key)s') def CacheFileName(self, key): uid = '_' try: # os.getuid() doesn't seem to work in Windows uid = str(os.getuid()) except: # pylint: disable=bare-except pass return self.path_pattern % {'key': key, 'uid': uid} def PutToken(self, key, value): """Serializes the value to the key's filename. To ensure that written tokens aren't leaked to a different users, we a) unlink an existing cache file, if any (to ensure we don't fall victim to symlink attacks and the like), b) create a new file with O_CREAT \| O_EXCL (to ensure nobody is trying to race us) If either of these steps fail, we simply give up (but log a warning). Not caching access tokens is not catastrophic, and failure to create a file can happen for either of the following reasons: - someone is attacking us as above, in which case we want to default to safe operation (not write the token); - another legitimate process is racing us; in this case one of the two will win and write the access token, which is fine; - we don't have permission to remove the old file or write to the specified directory, in which case we can't recover Args: key: the hash key to store. value: the access_token value to serialize. """ cache_file = self.CacheFileName(key) LOG.debug('FileSystemTokenCache.PutToken: key=%s, cache_file=%s', key, cache_file) try: os.unlink(cache_file) except: # pylint: disable=bare-except # Ignore failure to unlink the file; if the file exists and can't be # unlinked, the subsequent open with O_CREAT \| O_EXCL will fail. pass flags = os.O_RDWR \| os.O_CREAT \| os.O_EXCL # Accommodate Windows; stolen from python2.6/tempfile.py. if hasattr(os, 'O_NOINHERIT'): flags \|= os.O_NOINHERIT if hasattr(os, 'O_BINARY'): flags \|= os.O_BINARY try: fd = os.open(cache_file, flags, 0o600) except (OSError, IOError) as e: LOG.warning('FileSystemTokenCache.PutToken: ' 'Failed to create cache file %s: %s', cache_file, e) return f = os.fdopen(fd, 'w+b') serialized = value.Serialize() if isinstance(serialized, six.text_type): serialized = serialized.encode('utf-8') f.write(six.ensure_binary(serialized)) f.close() def GetToken(self, key): """Returns a deserialized access token from the key's filename.""" value = None cache_file = self.CacheFileName(key) try: f = open(cache_file) value = AccessToken.UnSerialize(f.read()) f.close() except (IOError, OSError) as e: if e.errno != errno.ENOENT: LOG.warning('FileSystemTokenCache.GetToken: ' 'Failed to read cache file %s: %s', cache_file, e) except Exception as e: # pylint: disable=broad-except LOG.warning('FileSystemTokenCache.GetToken: ' 'Failed to read cache file %s (possibly corrupted): %s', cache_file, e) LOG.debug('FileSystemTokenCache.GetToken: key=%s%s present (cache_file=%s)', key, ' not' if value is None else '', cache_file) return value class PKCS12Signer(crypt_base.Signer, crypt_base.FromServiceAccountMixin): """Signer for a p12 service account key.""" def __init__(self, key): self._key = key # Defined in crypt_base.Signer interface. # It is not used by p12 service account keys. @property def key_id(self): return None def sign(self, message): message = _helpers.to_bytes(message) from google.auth.crypt import _cryptography_rsa # pylint: disable=g-import-not-at-top return self._key.sign( message, _cryptography_rsa._PADDING, # pylint: disable=protected-access _cryptography_rsa._SHA256) # pylint: disable=protected-access @classmethod def from_string(cls, key_strings, key_id=None): del key_id #Unused key_string, password = (_helpers.to_bytes(k) for k in key_strings) # Cryptography package is not bundled with gsutil, Keeping it at top would throw error. from cryptography.hazmat.primitives.serialization import pkcs12 # pylint: disable=g-import-not-at-top try: key, _, _ = pkcs12.load_key_and_certificates(key_string, password) return cls(key) except: raise GsInvalidRefreshTokenError('Unable to load the keyfile, Invalid password or PKCS12 data.') class P12Credentials(service_account.Credentials): """google-auth service account credentials for p12 keys. p12 keys are not supported by the google-auth service account credentials. gsutil uses oauth2client to support p12 key users. Since oauth2client was deprecated and bundling it is security concern, we decided to support p12 in gsutil codebase. We prefer not adding it to the google-auth library because p12 is not supported from the beginning by google-auth. GCP strongly suggests users to use the JSON format. gsutil has to support it to not break users. """ _REQUIRED_FIELDS = ('service_account_email', 'token_uri', 'scopes') def authorize(self, http): return AuthorizedHttp(self, http=http) @property def access_token(self): return self.token @property def token_expiry(self): return self.expiry @classmethod def from_service_account_pkcs12_keystring(cls, key_string, password=None, kwargs): password = password or GOOGLE_OAUTH2_DEFAULT_FILE_PASSWORD signer = PKCS12Signer.from_string((key_string, password)) missing_fields = [f for f in cls._REQUIRED_FIELDS if f not in kwargs] if missing_fields: raise MissingFieldsError('Missing fields: {}.'.format( ', '.join(missing_fields))) creds = cls(signer, kwargs) return creds def CreateP12ServiceAccountCredentials(key_string, password=None, kwargs): """Creates a service account credentials from a p12 key and handles import errors.""" try: return P12Credentials.from_service_account_pkcs12_keystring( key_string, password, *kwargs) except ImportError: raise MissingDependencyError( ('pyca/cryptography is not available. Either install it, or please consider using the .json keyfile')) class OAuth2Client(object): """Common logic for OAuth2 clients.""" def __init__(self, cache_key_base, access_token_cache=None, datetime_strategy=datetime.datetime, auth_uri=None, token_uri=None, disable_ssl_certificate_validation=False, proxy_host=None, proxy_port=None, proxy_user=None, proxy_pass=None, ca_certs_file=None): # datetime_strategy is used to invoke utcnow() on; it is injected into the # constructor for unit testing purposes. self.auth_uri = auth_uri self.token_uri = token_uri self.cache_key_base = cache_key_base self.datetime_strategy = datetime_strategy self.access_token_cache = access_token_cache or InMemoryTokenCache() self.disable_ssl_certificate_validation = disable_ssl_certificate_validation self.ca_certs_file = ca_certs_file if proxy_host and proxy_port: self._proxy_info = httplib2.ProxyInfo(httplib2.socks.PROXY_TYPE_HTTP, proxy_host, proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass, proxy_rdns=True) else: self._proxy_info = None def CreateHttpRequest(self): return httplib2.Http( ca_certs=self.ca_certs_file, disable_ssl_certificate_validation=( self.disable_ssl_certificate_validation), proxy_info=self._proxy_info) def GetAccessToken(self): """Obtains an access token for this client. This client's access token cache is first checked for an existing, not-yet-expired access token. If none is found, the client obtains a fresh access token from the OAuth2 provider's token endpoint. Returns: The cached or freshly obtained AccessToken. Raises: oauth2client.client.AccessTokenRefreshError if an error occurs. """ # Ensure only one thread at a time attempts to get (and possibly refresh) # the access token. This doesn't prevent concurrent refresh attempts across # multiple gsutil instances, but at least protects against multiple threads # simultaneously attempting to refresh when gsutil -m is used. token_exchange_lock.acquire() try: cache_key = self.CacheKey() LOG.debug('GetAccessToken: checking cache for key %s', cache_key) access_token = self.access_token_cache.GetToken(cache_key) LOG.debug('GetAccessToken: token from cache: %s', access_token) if access_token is None or access_token.ShouldRefresh(): rapt = None if access_token is None else access_token.rapt_token LOG.debug('GetAccessToken: fetching fresh access token...') access_token = self.FetchAccessToken(rapt_token=rapt) LOG.debug('GetAccessToken: fresh access token: %s', access_token) self.access_token_cache.PutToken(cache_key, access_token) return access_token finally: token_exchange_lock.release() def CacheKey(self): """Computes a cache key. The cache key is computed as the SHA1 hash of the refresh token for user accounts, or the hash of the gs_service_client_id for service accounts, which satisfies the FileSystemTokenCache requirement that cache keys do not leak information about token values. Returns: A hash key. """ h = sha1() # Unicode objects (i.e. strings in Python 3) must be encoded before hashing if isinstance(self.cache_key_base, six.text_type): val = self.cache_key_base.encode('utf-8') else: val = self.cache_key_base h.update(val) return h.hexdigest() def GetAuthorizationHeader(self): """Gets the access token HTTP authorization header value. Returns: The value of an Authorization HTTP header that authenticates requests with an OAuth2 access token. """ return 'Bearer %s' % self.GetAccessToken().token class _BaseOAuth2ServiceAccountClient(OAuth2Client): """Base class for OAuth2ServiceAccountClients. Args: client_id: The OAuth2 client ID of this client. access_token_cache: An optional instance of a TokenCache. If omitted or None, an InMemoryTokenCache is used. auth_uri: The URI for OAuth2 authorization. token_uri: The URI used to refresh access tokens. datetime_strategy: datetime module strategy to use. disable_ssl_certificate_validation: True if certifications should not be validated. proxy_host: An optional string specifying the host name of an HTTP proxy to be used. proxy_port: An optional int specifying the port number of an HTTP proxy to be used. proxy_user: An optional string specifying the user name for interacting with the HTTP proxy. proxy_pass: An optional string specifying the password for interacting with the HTTP proxy. ca_certs_file: The cacerts.txt file to use. """ def __init__(self, client_id, access_token_cache=None, auth_uri=None, token_uri=None, datetime_strategy=datetime.datetime, disable_ssl_certificate_validation=False, proxy_host=None, proxy_port=None, proxy_user=None, proxy_pass=None, ca_certs_file=None): super(_BaseOAuth2ServiceAccountClient, self).__init__( cache_key_base=client_id, auth_uri=auth_uri, token_uri=token_uri, access_token_cache=access_token_cache, datetime_strategy=datetime_strategy, disable_ssl_certificate_validation=disable_ssl_certificate_validation, proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass, ca_certs_file=ca_certs_file) self._client_id = client_id def FetchAccessToken(self, rapt_token=None): credentials = self.GetCredentials() request = self.CreateHttpRequest() if isinstance(credentials, P12Credentials): request = GoogleAuthRequest(request) credentials.refresh(request) return AccessToken(credentials.access_token, credentials.token_expiry, datetime_strategy=self.datetime_strategy, rapt_token=rapt_token) class OAuth2ServiceAccountClient(_BaseOAuth2ServiceAccountClient): """An OAuth2 service account client using .p12 or .pem keys.""" def __init__(self, client_id, private_key, password, access_token_cache=None, auth_uri=None, token_uri=None, datetime_strategy=datetime.datetime, disable_ssl_certificate_validation=False, proxy_host=None, proxy_port=None, proxy_user=None, proxy_pass=None, ca_certs_file=None): # Avoid long repeated kwargs list. # pylint: disable=g-doc-args """Creates an OAuth2ServiceAccountClient. Args: client_id: The OAuth2 client ID of this client. private_key: The private key associated with this service account. password: The private key password used for the crypto signer. Keyword arguments match the _BaseOAuth2ServiceAccountClient class. """ # pylint: enable=g-doc-args super(OAuth2ServiceAccountClient, self).__init__( client_id, auth_uri=auth_uri, token_uri=token_uri, access_token_cache=access_token_cache, datetime_strategy=datetime_strategy, disable_ssl_certificate_validation=disable_ssl_certificate_validation, proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass, ca_certs_file=ca_certs_file) self._private_key = private_key self._password = password def GetCredentials(self): return CreateP12ServiceAccountCredentials(self._private_key, self._password, scopes=[DEFAULT_SCOPE], service_account_email=self._client_id, token_uri=self.token_uri) class OAuth2JsonServiceAccountClient(_BaseOAuth2ServiceAccountClient): """An OAuth2 service account client using .json keys.""" def __init__(self, json_key_dict, access_token_cache=None, auth_uri=None, token_uri=None, datetime_strategy=datetime.datetime, disable_ssl_certificate_validation=False, proxy_host=None, proxy_port=None, proxy_user=None, proxy_pass=None, ca_certs_file=None): # Avoid long repeated kwargs list. # pylint: disable=g-doc-args """Creates an OAuth2JsonServiceAccountClient. Args: json_key_dict: dictionary from the json private key file. Includes: client_id: The OAuth2 client ID of this client. client_email: The email associated with this client. private_key_id: The private key id associated with this service account. private_key_pkcs8_text: The pkcs8 text containing the private key data. Keyword arguments match the _BaseOAuth2ServiceAccountClient class. """ # pylint: enable=g-doc-args super(OAuth2JsonServiceAccountClient, self).__init__( json_key_dict['client_id'], auth_uri=auth_uri, token_uri=token_uri, access_token_cache=access_token_cache, datetime_strategy=datetime_strategy, disable_ssl_certificate_validation=disable_ssl_certificate_validation, proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass, ca_certs_file=ca_certs_file) self._json_key_dict = json_key_dict self._service_account_email = json_key_dict['client_email'] self._private_key_id = json_key_dict['private_key_id'] self._private_key_pkcs8_text = json_key_dict['private_key'] def GetCredentials(self): return _ServiceAccountCredentials.from_json_keyfile_dict( self._json_key_dict, scopes=[DEFAULT_SCOPE], token_uri=self.token_uri) class GsAccessTokenRefreshError(Exception): """Transient error when requesting access token.""" def __init__(self, e): super(GsAccessTokenRefreshError, self).__init__(e) class GsInvalidRefreshTokenError(Exception): def __init__(self, e): super(GsInvalidRefreshTokenError, self).__init__(e) class MissingDependencyError(Exception): def __init__(self, e): super(MissingDependencyError, self).__init__(e) class MissingFieldsError(Exception): def __init__(self, e): super(MissingFieldsError, self).__init__(e) class OAuth2UserAccountClient(OAuth2Client): """An OAuth2 client.""" def __init__(self, token_uri, client_id, client_secret, refresh_token, auth_uri=None, access_token_cache=None, datetime_strategy=datetime.datetime, disable_ssl_certificate_validation=False, proxy_host=None, proxy_port=None, proxy_user=None, proxy_pass=None, ca_certs_file=None): """Creates an OAuth2UserAccountClient. Args: token_uri: The URI used to refresh access tokens. client_id: The OAuth2 client ID of this client. client_secret: The OAuth2 client secret of this client. refresh_token: The token used to refresh the access token. auth_uri: The URI for OAuth2 authorization. access_token_cache: An optional instance of a TokenCache. If omitted or None, an InMemoryTokenCache is used. datetime_strategy: datetime module strategy to use. disable_ssl_certificate_validation: True if certifications should not be validated. proxy_host: An optional string specifying the host name of an HTTP proxy to be used. proxy_port: An optional int specifying the port number of an HTTP proxy to be used. proxy_user: An optional string specifying the user name for interacting with the HTTP proxy. proxy_pass: An optional string specifying the password for interacting with the HTTP proxy. ca_certs_file: The cacerts.txt file to use. """ super(OAuth2UserAccountClient, self).__init__( cache_key_base=refresh_token, auth_uri=auth_uri, token_uri=token_uri, access_token_cache=access_token_cache, datetime_strategy=datetime_strategy, disable_ssl_certificate_validation=disable_ssl_certificate_validation, proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass, ca_certs_file=ca_certs_file) self.token_uri = token_uri self.client_id = client_id self.client_secret = client_secret self.refresh_token = refresh_token def GetCredentials(self): """Fetches a credentials objects from the provider's token endpoint.""" access_token = self.GetAccessToken() credentials = reauth_creds.Oauth2WithReauthCredentials( access_token.token, self.client_id, self.client_secret, self.refresh_token, access_token.expiry, self.token_uri, None) # user_agent return credentials @retry_decorator.retry( GsAccessTokenRefreshError, tries=boto.config.get('OAuth2', 'oauth2_refresh_retries', 6), timeout_secs=1) def FetchAccessToken(self, rapt_token=None): """Fetches an access token from the provider's token endpoint. Fetches an access token from this client's OAuth2 provider's token endpoint. Args: rapt_token: (str) The RAPT to be passed when refreshing the access token. Returns: The fetched AccessToken. """ try: http = self.CreateHttpRequest() credentials = reauth_creds.Oauth2WithReauthCredentials( None, # access_token self.client_id, self.client_secret, self.refresh_token, None, # token_expiry self.token_uri, None, # user_agent scopes=RAPT_SCOPES, rapt_token=rapt_token) credentials.refresh(http) return AccessToken( credentials.access_token, credentials.token_expiry, datetime_strategy=self.datetime_strategy, rapt_token=credentials.rapt_token) except oauth2client.client.AccessTokenRefreshError as e: if 'Invalid response 403' in e.message: # This is the most we can do at the moment to accurately detect rate # limiting errors since they come back as 403s with no further # information. raise GsAccessTokenRefreshError(e) elif 'invalid_grant' in e.message: LOG.info(""" Attempted to retrieve an access token from an invalid refresh token. Two common cases in which you will see this error are: 1. Your refresh token was revoked. 2. Your refresh token was typed incorrectly. """) raise GsInvalidRefreshTokenError(e) else: raise class OAuth2GCEClient(OAuth2Client): """OAuth2 client for GCE instance.""" def __init__(self): super(OAuth2GCEClient, self).__init__( cache_key_base='', # Only InMemoryTokenCache can be used with empty cache_key_base. access_token_cache=InMemoryTokenCache()) @retry_decorator.retry(GsAccessTokenRefreshError, tries=6, timeout_secs=1) def FetchAccessToken(self, rapt_token=None): """Fetches an access token from the provider's token endpoint. Fetches an access token from the GCE metadata server. Args: rapt_token: (str) Ignored for this class. Service accounts don't use reauth credentials. Returns: The fetched AccessToken. """ del rapt_token # Unused for service account credentials. # Note that rapt_token is used only for user credentials, and thus should # always be None in the context of GCE (service account) credentials. response = None try: http = httplib2.Http() response, content = http.request(META_TOKEN_URI, method='GET', body=None, headers=META_HEADERS) content = six.ensure_text(content) except Exception as e: raise GsAccessTokenRefreshError(e) if response.status == 200: d = json.loads(content) return AccessToken( d['access_token'], datetime.datetime.now() + datetime.timedelta(seconds=d.get('expires_in', 0)), datetime_strategy=self.datetime_strategy, rapt_token=None) def _IsGCE(): """Returns True if running on a GCE instance, otherwise False.""" try: http = httplib2.Http() response, _ = http.request(METADATA_SERVER) return response.status == 200 except (httplib2.ServerNotFoundError, socket.error): # We might see something like "No route to host" propagated as a socket # error. We might also catch transient socket errors, but at that point # we're going to fail anyway, just with a different error message. With # this approach, we'll avoid having to enumerate all possible non-transient # socket errors. return False except Exception as e: # pylint: disable=broad-except LOG.warning("Failed to determine whether we're running on GCE, so we'll" "assume that we aren't: %s", e) return False return False def CreateOAuth2GCEClient(): return OAuth2GCEClient() if _IsGCE() else None class AccessToken(object): """Encapsulates an OAuth2 access token.""" def __init__(self, token, expiry, datetime_strategy=datetime.datetime, rapt_token=None): self.token = token self.expiry = expiry self.datetime_strategy = datetime_strategy # RAPT isn't technically part of the access token, but we can use the # RAPT from an expired access token when refreshing to get a new access # token. This allows users to only answer reauth challenges when the RAPT # expires (usually ~once a day), as opposed to every time the access token # expires (usually after an hour). self.rapt_token = rapt_token @staticmethod def UnSerialize(query): """Creates an AccessToken object from its serialized form.""" def GetValue(d, key): return (d.get(key, [None]))[0] kv = urllib.parse.parse_qs(query) if 'token' not in kv or not kv['token']: return None expiry = None expiry_tuple = GetValue(kv, 'expiry') if expiry_tuple: try: expiry = datetime.datetime( [int(n) for n in expiry_tuple.split(',')]) except: # pylint: disable=bare-except return None rapt_token = GetValue(kv, 'rapt_token') return AccessToken(GetValue(kv, 'token'), expiry, rapt_token=rapt_token) def Serialize(self): """Serializes this object as URI-encoded key-value pairs.""" # There's got to be a better way to serialize a datetime. Unfortunately, # there is no reliable way to convert into a unix epoch. kv = {'token': self.token} if self.expiry: t = self.expiry tupl = (t.year, t.month, t.day, t.hour, t.minute, t.second, t.microsecond) kv['expiry'] = ','.join([str(i) for i in tupl]) if self.rapt_token: kv['rapt_token'] = self.rapt_token return urllib.parse.urlencode(kv) def ShouldRefresh(self, time_delta=300): """Whether the access token needs to be refreshed. Args: time_delta: refresh access token when it expires within time_delta secs. Returns: True if the token is expired or about to expire, False if the token should be expected to work. Note that the token may still be rejected, e.g. if it has been revoked server-side. """ if self.expiry is None: return False return (self.datetime_strategy.utcnow() + datetime.timedelta(seconds=time_delta) > self.expiry) def __eq__(self, other): return self.token == other.token and self.expiry == other.expiry def __ne__(self, other): return not self.__eq__(other) def __str__(self): return 'AccessToken(token=%s, expiry=%sZ)' % (self.token, self.expiry)	31.59 KB	2025-06-24 09:04
📄 oauth2_helper.py # Copyright 2014 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Helper routines to facilitate use of oauth2_client.""" from __future__ import absolute_import import io import json import os import sys import time import webbrowser from gcs_oauth2_boto_plugin import oauth2_client import oauth2client.client from six.moves import input # pylint: disable=redefined-builtin UTF8 = 'utf-8' CLIENT_ID = None CLIENT_SECRET = None GOOGLE_OAUTH2_PROVIDER_AUTHORIZATION_URI = ( 'https://accounts.google.com/o/oauth2/auth') GOOGLE_OAUTH2_PROVIDER_TOKEN_URI = ( 'https://oauth2.googleapis.com/token') GOOGLE_OAUTH2_DEFAULT_FILE_PASSWORD = 'notasecret' OOB_REDIRECT_URI = 'urn:ietf:wg:oauth:2.0:oob' def OAuth2ClientFromBotoConfig( config, cred_type=oauth2_client.CredTypes.OAUTH2_USER_ACCOUNT): """Create a client type based on credentials supplied in boto config.""" token_cache = None token_cache_type = config.get('OAuth2', 'token_cache', 'file_system') if token_cache_type == 'file_system': if config.has_option('OAuth2', 'token_cache_path_pattern'): token_cache = oauth2_client.FileSystemTokenCache( path_pattern=config.get('OAuth2', 'token_cache_path_pattern')) else: token_cache = oauth2_client.FileSystemTokenCache() elif token_cache_type == 'in_memory': token_cache = oauth2_client.InMemoryTokenCache() else: raise Exception( 'Invalid value for config option OAuth2/token_cache: %s' % token_cache_type) proxy_host = None proxy_port = None proxy_user = None proxy_pass = None if (config.has_option('Boto', 'proxy') and config.has_option('Boto', 'proxy_port')): proxy_host = config.get('Boto', 'proxy') proxy_port = int(config.get('Boto', 'proxy_port')) proxy_user = config.get('Boto', 'proxy_user', None) proxy_pass = config.get('Boto', 'proxy_pass', None) provider_authorization_uri = config.get( 'OAuth2', 'provider_authorization_uri', GOOGLE_OAUTH2_PROVIDER_AUTHORIZATION_URI) provider_token_uri = config.get( 'OAuth2', 'provider_token_uri', GOOGLE_OAUTH2_PROVIDER_TOKEN_URI) if cred_type == oauth2_client.CredTypes.OAUTH2_SERVICE_ACCOUNT: service_client_id = config.get('Credentials', 'gs_service_client_id', '') private_key_filename = config.get('Credentials', 'gs_service_key_file', '') with io.open(private_key_filename, 'rb') as private_key_file: private_key = private_key_file.read() keyfile_is_utf8 = False try: private_key = private_key.decode(UTF8) # P12 keys won't be encoded as UTF8 bytes. keyfile_is_utf8 = True except UnicodeDecodeError: pass if keyfile_is_utf8: try: json_key_dict = json.loads(private_key) except ValueError: raise Exception('Could not parse JSON keyfile "%s" as valid JSON' % private_key_filename) for json_entry in ('client_id', 'client_email', 'private_key_id', 'private_key'): if json_entry not in json_key_dict: raise Exception('The JSON private key file at %s ' 'did not contain the required entry: %s' % (private_key_filename, json_entry)) return oauth2_client.OAuth2JsonServiceAccountClient( json_key_dict, access_token_cache=token_cache, auth_uri=provider_authorization_uri, token_uri=provider_token_uri, disable_ssl_certificate_validation=not(config.getbool( 'Boto', 'https_validate_certificates', True)), proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass) else: key_file_pass = config.get('Credentials', 'gs_service_key_file_password', GOOGLE_OAUTH2_DEFAULT_FILE_PASSWORD) return oauth2_client.OAuth2ServiceAccountClient( service_client_id, private_key, key_file_pass, access_token_cache=token_cache, auth_uri=provider_authorization_uri, token_uri=provider_token_uri, disable_ssl_certificate_validation=not(config.getbool( 'Boto', 'https_validate_certificates', True)), proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass) elif cred_type == oauth2_client.CredTypes.OAUTH2_USER_ACCOUNT: client_id = config.get('OAuth2', 'client_id', os.environ.get('OAUTH2_CLIENT_ID', CLIENT_ID)) if not client_id: raise Exception( 'client_id for your application obtained from ' 'https://console.developers.google.com must be set in a boto config ' 'or with OAUTH2_CLIENT_ID environment variable or with ' 'gcs_oauth2_boto_plugin.SetFallbackClientIdAndSecret function.') client_secret = config.get('OAuth2', 'client_secret', os.environ.get('OAUTH2_CLIENT_SECRET', CLIENT_SECRET)) ca_certs_file = config.get_value('Boto', 'ca_certificates_file') if ca_certs_file == 'system': ca_certs_file = None if not client_secret: raise Exception( 'client_secret for your application obtained from ' 'https://console.developers.google.com must be set in a boto config ' 'or with OAUTH2_CLIENT_SECRET environment variable or with ' 'gcs_oauth2_boto_plugin.SetFallbackClientIdAndSecret function.') return oauth2_client.OAuth2UserAccountClient( provider_token_uri, client_id, client_secret, config.get('Credentials', 'gs_oauth2_refresh_token'), auth_uri=provider_authorization_uri, access_token_cache=token_cache, disable_ssl_certificate_validation=not(config.getbool( 'Boto', 'https_validate_certificates', True)), proxy_host=proxy_host, proxy_port=proxy_port, proxy_user=proxy_user, proxy_pass=proxy_pass, ca_certs_file=ca_certs_file) else: raise Exception('You have attempted to create an OAuth2 client without ' 'setting up OAuth2 credentials.') def OAuth2ApprovalFlow(client, scopes, launch_browser=False): """Run the OAuth2 flow to fetch a refresh token. Returns the refresh token.""" flow = oauth2client.client.OAuth2WebServerFlow( client.client_id, client.client_secret, scopes, auth_uri=client.auth_uri, token_uri=client.token_uri, redirect_uri=OOB_REDIRECT_URI) approval_url = flow.step1_get_authorize_url() if launch_browser: sys.stdout.write( 'Attempting to launch a browser with the OAuth2 approval dialog at ' 'URL: %s\n\n' '[Note: due to a Python bug, you may see a spurious error message ' '"object is not\ncallable [...] in [...] Popen.__del__" which can be ' 'ignored.]\n\n' % approval_url) else: sys.stdout.write( 'Please navigate your browser to the following URL:\n%s\n' % approval_url) sys.stdout.write( 'In your browser you should see a page that requests you to authorize ' 'access to Google Cloud Platform APIs and Services on your behalf. ' 'After you approve, an authorization code will be displayed.\n\n') if (launch_browser and not webbrowser.open(approval_url, new=1, autoraise=True)): sys.stdout.write( 'Launching browser appears to have failed; please navigate a browser ' 'to the following URL:\n%s\n' % approval_url) # Short delay; webbrowser.open on linux insists on printing out a message # which we don't want to run into the prompt for the auth code. time.sleep(2) code = input('Enter the authorization code: ') credentials = flow.step2_exchange(code, http=client.CreateHttpRequest()) return credentials.refresh_token def SetFallbackClientIdAndSecret(client_id, client_secret): global CLIENT_ID global CLIENT_SECRET CLIENT_ID = client_id CLIENT_SECRET = client_secret def SetLock(lock): oauth2_client.token_exchange_lock = lock	8.34 KB	2025-06-24 09:04
📄 oauth2_plugin.py # Copyright 2014 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Boto auth plugin for OAuth2.0 for Google Cloud Storage.""" from __future__ import absolute_import import boto.auth_handler from gcs_oauth2_boto_plugin import oauth2_client from gcs_oauth2_boto_plugin import oauth2_helper IS_SERVICE_ACCOUNT = False class OAuth2Auth(boto.auth_handler.AuthHandler): """AuthHandler for working with OAuth2 user account credentials.""" capability = ['google-oauth2', 's3'] def __init__(self, path, config, provider): self.oauth2_client = None if provider.name == 'google': if config.has_option('Credentials', 'gs_oauth2_refresh_token'): self.oauth2_client = oauth2_helper.OAuth2ClientFromBotoConfig(config) elif config.has_option('GoogleCompute', 'service_account'): self.oauth2_client = oauth2_client.CreateOAuth2GCEClient() if not self.oauth2_client: raise boto.auth_handler.NotReadyToAuthenticate() def add_auth(self, http_request): http_request.headers['Authorization'] = ( self.oauth2_client.GetAuthorizationHeader()) class OAuth2ServiceAccountAuth(boto.auth_handler.AuthHandler): """AuthHandler for working with OAuth2 service account credentials.""" capability = ['google-oauth2', 's3'] def __init__(self, path, config, provider): if (provider.name == 'google' and config.has_option('Credentials', 'gs_service_key_file')): self.oauth2_client = oauth2_helper.OAuth2ClientFromBotoConfig( config, cred_type=oauth2_client.CredTypes.OAUTH2_SERVICE_ACCOUNT) # If we make it to this point, then we will later attempt to authenticate # as a service account based on how the boto auth plugins work. This is # global so that command.py can access this value once it's set. # TODO: replace this approach with a way to get the current plugin # from boto so that we don't have to have global variables. global IS_SERVICE_ACCOUNT IS_SERVICE_ACCOUNT = True else: raise boto.auth_handler.NotReadyToAuthenticate() def add_auth(self, http_request): http_request.headers['Authorization'] = ( self.oauth2_client.GetAuthorizationHeader())	2.67 KB	2025-06-24 09:04
📄 test_oauth2_client.py # Copyright 2014 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unit tests for oauth2_client and related classes.""" from __future__ import absolute_import import datetime import logging import os import stat import sys import unittest from freezegun import freeze_time from gcs_oauth2_boto_plugin import oauth2_client import httplib2 try: from unittest import mock except ImportError: import mock LOG = logging.getLogger('test_oauth2_client') ACCESS_TOKEN = 'abc123' RAPT_TOKEN = 'rapt123' TOKEN_URI = 'https://provider.example.com/oauth/provider?mode=token' AUTH_URI = 'https://provider.example.com/oauth/provider?mode=authorize' DEFAULT_CA_CERTS_FILE = os.path.abspath( os.path.join('gslib', 'data', 'cacerts.txt')) IS_WINDOWS = 'win32' in str(sys.platform).lower() class MockDateTime(object): def __init__(self): self.mock_now = None def utcnow(self): # pylint: disable=invalid-name return self.mock_now class MockOAuth2ServiceAccountClient(oauth2_client.OAuth2ServiceAccountClient): """Mock service account client for testing OAuth2 with service accounts.""" def __init__(self, client_id, private_key, password, auth_uri, token_uri, datetime_strategy): super(MockOAuth2ServiceAccountClient, self).__init__( client_id, private_key, password, auth_uri=auth_uri, token_uri=token_uri, datetime_strategy=datetime_strategy, ca_certs_file=DEFAULT_CA_CERTS_FILE) self.Reset() def Reset(self): self.fetched_token = False def FetchAccessToken(self, rapt_token=None): self.fetched_token = True return oauth2_client.AccessToken( ACCESS_TOKEN, GetExpiry(self.datetime_strategy, 3600), datetime_strategy=self.datetime_strategy, rapt_token=None) class MockOAuth2UserAccountClient(oauth2_client.OAuth2UserAccountClient): """Mock user account client for testing OAuth2 with user accounts.""" def __init__(self, token_uri, client_id, client_secret, refresh_token, auth_uri, datetime_strategy): super(MockOAuth2UserAccountClient, self).__init__( token_uri, client_id, client_secret, refresh_token, auth_uri=auth_uri, datetime_strategy=datetime_strategy, ca_certs_file=DEFAULT_CA_CERTS_FILE) self.Reset() def Reset(self): self.fetched_token = False def FetchAccessToken(self, rapt_token=None): self.fetched_token = True return oauth2_client.AccessToken( ACCESS_TOKEN, GetExpiry(self.datetime_strategy, 3600), datetime_strategy=self.datetime_strategy, rapt_token=RAPT_TOKEN if rapt_token is None else rapt_token) def GetExpiry(datetime_strategy, length_in_seconds): token_expiry = (datetime_strategy.utcnow() + datetime.timedelta(seconds=length_in_seconds)) return token_expiry def CreateMockUserAccountClient(mock_datetime): return MockOAuth2UserAccountClient( TOKEN_URI, 'clid', 'clsecret', 'ref_token_abc123', AUTH_URI, mock_datetime) def CreateMockServiceAccountClient(mock_datetime): return MockOAuth2ServiceAccountClient( 'clid', 'private_key', 'password', AUTH_URI, TOKEN_URI, mock_datetime) class OAuth2AccountClientTest(unittest.TestCase): """Unit tests for OAuth2UserAccountClient and OAuth2ServiceAccountClient.""" def setUp(self): self.tempdirs = [] self.mock_datetime = MockDateTime() self.start_time = datetime.datetime(2011, 3, 1, 11, 25, 13, 300826) self.mock_datetime.mock_now = self.start_time def testGetAccessTokenUserAccount(self): self.client = CreateMockUserAccountClient(self.mock_datetime) self._RunGetAccessTokenTest(expected_rapt=RAPT_TOKEN) def testGetAccessTokenServiceAccount(self): self.client = CreateMockServiceAccountClient(self.mock_datetime) self._RunGetAccessTokenTest() def _RunGetAccessTokenTest(self, expected_rapt=None): """Tests access token gets with self.client.""" access_token_1 = 'abc123' self.assertFalse(self.client.fetched_token) token_1 = self.client.GetAccessToken() # There's no access token in the cache; verify that we fetched a fresh # token. self.assertTrue(self.client.fetched_token) self.assertEqual(access_token_1, token_1.token) self.assertEqual(self.start_time + datetime.timedelta(minutes=60), token_1.expiry) self.assertEqual(token_1.rapt_token, expected_rapt) # Advance time by less than expiry time, and fetch another token. self.client.Reset() self.mock_datetime.mock_now = ( self.start_time + datetime.timedelta(minutes=55)) token_2 = self.client.GetAccessToken() # Since the access token wasn't expired, we get the cache token, and there # was no refresh request. self.assertEqual(token_1, token_2) self.assertEqual(access_token_1, token_2.token) self.assertFalse(self.client.fetched_token) # Advance time past expiry time, and fetch another token. self.client.Reset() self.mock_datetime.mock_now = ( self.start_time + datetime.timedelta(minutes=55, seconds=1)) self.client.datetime_strategy = self.mock_datetime token_3 = self.client.GetAccessToken() # This should have resulted in a refresh request and a fresh access token. self.assertTrue(self.client.fetched_token) self.assertEqual( self.mock_datetime.mock_now + datetime.timedelta(minutes=60), token_3.expiry) self.assertEqual(token_3.rapt_token, expected_rapt) class TestCreateP12ServiceAccountCredentials(unittest.TestCase): @mock.patch.object(oauth2_client.PKCS12Signer, 'from_string') def test_from_service_account_pkcs12_keystring(self, mock_signer): mock_signer.return_value = 'MOCK_SIGNER' with mock.patch.object(oauth2_client.P12Credentials, '__init__') as mock_p12creds: mock_p12creds.return_value = None oauth2_client.P12Credentials.from_service_account_pkcs12_keystring( 'MY_KEY', service_account_email='test@google.com', token_uri= 'TOKEN_URI', scopes= ['MYSCOPEEEEE'] ) mock_p12creds.assert_called_once_with('MOCK_SIGNER', service_account_email='test@google.com', token_uri= 'TOKEN_URI', scopes= ['MYSCOPEEEEE']) @mock.patch.object(oauth2_client.PKCS12Signer, 'from_string') def test_from_service_account_pkcs12_keystring_raises_missing_field(self, mock_signer): mock_signer.return_value = 'MOCK_SIGNER' with self.assertRaises(oauth2_client.MissingFieldsError) as exc: oauth2_client.P12Credentials.from_service_account_pkcs12_keystring( 'MY_KEY', token_uri= 'TOKEN_URI', scopes= ['MYSCOPEEEEE'] ) class AccessTokenTest(unittest.TestCase): """Unit tests for access token functions.""" def testShouldRefresh(self): """Tests that token.ShouldRefresh returns the correct value.""" mock_datetime = MockDateTime() start = datetime.datetime(2011, 3, 1, 11, 25, 13, 300826) expiry = start + datetime.timedelta(minutes=60) token = oauth2_client.AccessToken( 'foo', expiry, datetime_strategy=mock_datetime) mock_datetime.mock_now = start self.assertFalse(token.ShouldRefresh()) mock_datetime.mock_now = start + datetime.timedelta(minutes=54) self.assertFalse(token.ShouldRefresh()) mock_datetime.mock_now = start + datetime.timedelta(minutes=55) self.assertFalse(token.ShouldRefresh()) mock_datetime.mock_now = start + datetime.timedelta( minutes=55, seconds=1) self.assertTrue(token.ShouldRefresh()) mock_datetime.mock_now = start + datetime.timedelta( minutes=61) self.assertTrue(token.ShouldRefresh()) mock_datetime.mock_now = start + datetime.timedelta(minutes=58) self.assertFalse(token.ShouldRefresh(time_delta=120)) mock_datetime.mock_now = start + datetime.timedelta( minutes=58, seconds=1) self.assertTrue(token.ShouldRefresh(time_delta=120)) def testShouldRefreshNoExpiry(self): """Tests token.ShouldRefresh with no expiry time.""" mock_datetime = MockDateTime() start = datetime.datetime(2011, 3, 1, 11, 25, 13, 300826) token = oauth2_client.AccessToken( 'foo', None, datetime_strategy=mock_datetime) mock_datetime.mock_now = start self.assertFalse(token.ShouldRefresh()) mock_datetime.mock_now = start + datetime.timedelta( minutes=472) self.assertFalse(token.ShouldRefresh()) def testSerialization(self): """Tests token serialization.""" expiry = datetime.datetime(2011, 3, 1, 11, 25, 13, 300826) token = oauth2_client.AccessToken('foo', expiry, rapt_token=RAPT_TOKEN) serialized_token = token.Serialize() LOG.debug('testSerialization: serialized_token=%s', serialized_token) token2 = oauth2_client.AccessToken.UnSerialize(serialized_token) self.assertEqual(token, token2) class FileSystemTokenCacheTest(unittest.TestCase): """Unit tests for FileSystemTokenCache.""" def setUp(self): self.cache = oauth2_client.FileSystemTokenCache() self.start_time = datetime.datetime(2011, 3, 1, 10, 25, 13, 300826) self.token_1 = oauth2_client.AccessToken( 'token1', self.start_time, rapt_token=RAPT_TOKEN) self.token_2 = oauth2_client.AccessToken( 'token2', self.start_time + datetime.timedelta(seconds=492), rapt_token=RAPT_TOKEN) self.key = 'token1key' def tearDown(self): try: os.unlink(self.cache.CacheFileName(self.key)) except: # pylint: disable=bare-except pass def testPut(self): self.cache.PutToken(self.key, self.token_1) # Assert that the cache file exists and has correct permissions. if not IS_WINDOWS: self.assertEqual( 0o600, stat.S_IMODE(os.stat(self.cache.CacheFileName(self.key)).st_mode)) def testPutGet(self): """Tests putting and getting various tokens.""" # No cache file present. self.assertEqual(None, self.cache.GetToken(self.key)) # Put a token self.cache.PutToken(self.key, self.token_1) cached_token = self.cache.GetToken(self.key) self.assertEqual(self.token_1, cached_token) # Put a different token self.cache.PutToken(self.key, self.token_2) cached_token = self.cache.GetToken(self.key) self.assertEqual(self.token_2, cached_token) def testGetBadFile(self): f = open(self.cache.CacheFileName(self.key), 'w') f.write('blah') f.close() self.assertEqual(None, self.cache.GetToken(self.key)) def testCacheFileName(self): """Tests configuring the cache with a specific file name.""" cache = oauth2_client.FileSystemTokenCache( path_pattern='/var/run/ccache/token.%(uid)s.%(key)s') if IS_WINDOWS: uid = '_' else: uid = os.getuid() self.assertEqual('/var/run/ccache/token.%s.abc123' % uid, cache.CacheFileName('abc123')) cache = oauth2_client.FileSystemTokenCache( path_pattern='/var/run/ccache/token.%(key)s') self.assertEqual('/var/run/ccache/token.abc123', cache.CacheFileName('abc123')) class RefreshTokenTest(unittest.TestCase): """Unit tests for refresh tokens.""" def setUp(self): self.mock_datetime = MockDateTime() self.start_time = datetime.datetime(2011, 3, 1, 10, 25, 13, 300826) self.mock_datetime.mock_now = self.start_time self.client = CreateMockUserAccountClient(self.mock_datetime) def testUniqeId(self): cred_id = self.client.CacheKey() self.assertEqual('0720afed6871f12761fbea3271f451e6ba184bf5', cred_id) def testGetAuthorizationHeader(self): self.assertEqual('Bearer %s' % ACCESS_TOKEN, self.client.GetAuthorizationHeader()) class FakeResponse(object): def __init__(self, status): self._status = status @property def status(self): return self._status class OAuth2GCEClientTest(unittest.TestCase): """Unit tests for OAuth2GCEClient.""" def setUp(self): patcher = mock.patch( 'gcs_oauth2_boto_plugin.oauth2_client.httplib2.Http', autospec=True) self.addCleanup(patcher.stop) self.mocked_http_class = patcher.start() self.mock_http = self.mocked_http_class.return_value @freeze_time('2014-03-26 01:01:01') def testFetchAccessToken(self): token = 'my_token' self.mock_http.request.return_value = ( FakeResponse(200), '{"access_token":"%(TOKEN)s", "expires_in": %(EXPIRES_IN)d}' % { 'TOKEN': token, 'EXPIRES_IN': 42 }) client = oauth2_client.OAuth2GCEClient() self.assertEqual( str(client.FetchAccessToken()), 'AccessToken(token=%s, expiry=2014-03-26 01:01:43Z)' % token) self.mock_http.request.assert_called_with( oauth2_client.META_TOKEN_URI, method='GET', body=None, headers=oauth2_client.META_HEADERS) def testIsGCENotFound(self): self.mock_http.request.return_value = (FakeResponse(404), '') self.assertFalse(oauth2_client._IsGCE()) self.mock_http.request.assert_called_once_with( oauth2_client.METADATA_SERVER) def testIsGCEServerNotFound(self): self.mock_http.request.side_effect = httplib2.ServerNotFoundError() self.assertFalse(oauth2_client._IsGCE()) self.mock_http.request.assert_called_once_with( oauth2_client.METADATA_SERVER) def testIsGCETrue(self): self.mock_http.request.return_value = (FakeResponse(200), '') self.assertTrue(oauth2_client._IsGCE()) self.mock_http.request.assert_called_once_with( oauth2_client.METADATA_SERVER) if __name__ == '__main__': unittest.main()	13.8 KB	2025-06-24 09:04

Terminal