deepblue/GoogleDriveManagement
1
0
Fork 0
mirror of https://github.com/GAM-team/GAM.git synced 2026-06-28 09:51:36 +00:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

Centralize OAuth2.0 Credential logic (#1126)

Browse Source 
* Centralize OAuth2.0 Credential logic

Adds a Credentials class that centralizes and handles most existing
logic related to OAuth2.0 credentials, including generation, storage,
file locking, and attribute retrieval. This is a step towards
minimizing the duplicated code that handles credentials in various
methods. The goal is to eventually get to a point where there are 2
credential entry points: `auth.get_admin_credentials()` and
`auth.get_credentials_for_user(user)`. Then, we can slowly move toward
using impersonated credentials for all operations and scrap the need
for user consented credentials all together.

* Skip test_delete_removes_lock_file when testing on Windows
This commit is contained in:
ejochman
2020-03-25 03:31:47 -07:00
committed by GitHub
parent 1a32f2a6f8
commit 4444974a9e
5 changed files with 1326 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/auth/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
"""Authentication/Credentials general purpose and convenience methods."""
from . import oauth
from var import _FN_OAUTH2_TXT
from var import GC_OAUTH2_TXT
from var import GC_Values
# TODO: Move logic that determines file name into this module. We should be able
# to discover the file location without accessing a private member or waiting
# for a global initialization.
DEFAULT_OAUTH_STORAGE_FILE = _FN_OAUTH2_TXT
def get_admin_credentials_filename():
"""Gets the name of the file that stores the admin account credentials."""
# If the environment globals are loaded, use the set global value. It may have
# some custom name in it. Otherwise, just use the default name.
if GC_Values[GC_OAUTH2_TXT]:
return GC_Values[GC_OAUTH2_TXT]
else:
return DEFAULT_OAUTH_STORAGE_FILE
def get_admin_credentials():
"""Gets oauth.Credentials that are authenticated as the domain's admin user."""
credential_file = get_admin_credentials_filename()
return oauth.Credentials.from_credentials_file(credential_file)

564
src/auth/oauth.py Normal file
View File

@@ -0,0 +1,564 @@
"""OAuth2.0 user credentials."""
import datetime
import json
import os
import re
import threading
from urllib.parse import urlencode
from filelock import FileLock
import google_auth_oauthlib.flow
import google.oauth2.credentials
import google.oauth2.id_token
import fileutils
import transport
from var import GAM_INFO
from var import GM_Globals
from var import GM_WINDOWS
MESSAGE_CONSOLE_AUTHORIZATION_PROMPT = ('\nGo to the following link in your '
'browser:\n\n\t{url}\n')
MESSAGE_CONSOLE_AUTHORIZATION_CODE = 'Enter verification code: '
MESSAGE_LOCAL_SERVER_AUTHORIZATION_PROMPT = ('\nYour browser has been opened to'
' visit:\n\n\t{url}\n\nIf your '
'browser is on a different machine'
' then press CTRL+C and create a '
'file called nobrowser.txt in the '
'same folder as GAM.\n')
MESSAGE_LOCAL_SERVER_SUCCESS = ('The authentication flow has completed. You may'
' close this browser window and return to GAM.')
class CredentialsError(Exception):
"""Base error class."""
pass
class InvalidCredentialsFileError(CredentialsError):
"""Error raised when a file cannot be opened into a credentials object."""
pass
class EmptyCredentialsFileError(InvalidCredentialsFileError):
"""Error raised when a credentials file contains no content."""
pass
class InvalidClientSecretsFileFormatError(CredentialsError):
"""Error raised when a client secrets file format is invalid."""
pass
class InvalidClientSecretsFileError(CredentialsError):
"""Error raised when client secrets file cannot be read."""
pass
class Credentials(google.oauth2.credentials.Credentials):
"""Google OAuth2.0 Credentials with GAM-specific properties and methods."""
DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
def __init__(self,
token,
refresh_token=None,
id_token=None,
token_uri=None,
client_id=None,
client_secret=None,
scopes=None,
quota_project_id=None,
expiry=None,
id_token_data=None,
filename=None):
"""A thread-safe OAuth2.0 credentials object.
Credentials adds additional utility properties and methods to a
standard OAuth2.0 credentials object. When used to store credentials on
disk, it implements a file lock to avoid collision during writes.
Args:
token: Optional String, The OAuth 2.0 access token. Can be None if refresh
information is provided.
refresh_token: String, The OAuth 2.0 refresh token. If specified,
credentials can be refreshed.
id_token: String, The Open ID Connect ID Token.
token_uri: String, The OAuth 2.0 authorization server's token endpoint
URI. Must be specified for refresh, can be left as None if the token can
not be refreshed.
client_id: String, The OAuth 2.0 client ID. Must be specified for refresh,
can be left as None if the token can not be refreshed.
client_secret: String, The OAuth 2.0 client secret. Must be specified for
refresh, can be left as None if the token can not be refreshed.
scopes: Sequence[str], The scopes used to obtain authorization.
This parameter is used by :meth:`has_scopes`. OAuth 2.0 credentials can
not request additional scopes after authorization. The scopes must be
derivable from the refresh token if refresh information is provided
(e.g. The refresh token scopes are a superset of this or contain a
wild card scope like
'https://www.googleapis.com/auth/any-api').
quota_project_id: String, The project ID used for quota and billing. This
project may be different from the project used to create the
credentials.
expiry: datetime.datetime, The time at which the provided token will
expire.
id_token_data: Oauth2.0 ID Token data which was previously fetched for
this access token against the google.oauth2.id_token library.
filename: String, Path to a file that will be used to store the
credentials. If provided, a lock file of the same name and a ".lock"
extension will be created for concurrency controls. Note: New
credentials are not saved to disk until write() or refresh() are
called.
Raises:
TypeError: If id_token_data is not the required dict type.
"""
super(Credentials, self).__init__(
token=token,
refresh_token=refresh_token,
id_token=id_token,
token_uri=token_uri,
client_id=client_id,
client_secret=client_secret,
scopes=scopes,
quota_project_id=quota_project_id)
# Load data not restored by the super class
self.expiry = expiry
if id_token_data and not isinstance(id_token_data, dict):
raise TypeError(f'Expected type id_token_data dict but received '
f'{type(id_token_data)}')
self._id_token_data = id_token_data.copy() if id_token_data else None
# If a filename is provided, use a lock file to control concurrent access
# to the resource. If no filename is provided, use a thread lock that has
# the same interface as FileLock in order to simplify the implementation.
if filename:
# Convert relative paths into absolute
self._filename = os.path.abspath(filename)
lock_file = os.path.abspath(f'{self._filename}.lock')
self._lock = FileLock(lock_file)
else:
self._filename = None
self._lock = _FileLikeThreadLock()
# Use a property to prevent external mutation of the filename.
@property
def filename(self):
return self._filename
@classmethod
def from_authorized_user_info(cls, info, filename=None):
"""Generates Credentials from JSON containing authorized user info.
Args:
info: Dict, authorized user info in Google format.
filename: String, the filename used to store these credentials on disk. If
no filename is provided, the credentials will not be saved to disk.
Raises:
ValueError: If missing fields are detected in the info.
"""
keys_needed = set(('refresh_token', 'client_id', 'client_secret'))
missing = keys_needed.difference(info.keys())
if missing:
raise ValueError(
'Authorized user info was not in the expected format, missing '
f'fields {", ".join(missing)}.')
expiry = info.get('token_expiry')
if expiry:
# Convert the raw expiry to datetime
expiry = datetime.datetime.strptime(expiry, Credentials.DATETIME_FORMAT)
id_token_data = info.get('decoded_id_token')
# Provide backwards compatibility with field names when loading from JSON.
# Some field names may be different, depending on when/how the credentials
# were pickled.
return cls(
token=info.get('token', info.get('auth_token', '')),
refresh_token=info['refresh_token'],
id_token=info.get('id_token_jwt', info.get('id_token')),
token_uri=info.get('token_uri'),
client_id=info['client_id'],
client_secret=info['client_secret'],
scopes=info.get('scopes'),
quota_project_id=info.get('quota_project_id'),
expiry=expiry,
id_token_data=id_token_data,
filename=filename)
@classmethod
def from_google_oauth2_credentials(cls, credentials, filename=None):
"""Generates Credentials from a google.oauth2.Credentials object."""
info = json.loads(credentials.to_json())
# Add properties which are not exported with the native to_json() output.
info['id_token'] = credentials.id_token
if credentials.expiry:
info['token_expiry'] = credentials.expiry.strftime(
Credentials.DATETIME_FORMAT)
info['quota_project_id'] = credentials.quota_project_id
return cls.from_authorized_user_info(info, filename=filename)
@classmethod
def from_credentials_file(cls, filename):
"""Generates Credentials from a stored Credentials file.
The same file will be used to save the credentials when the access token is
refreshed.
Args:
filename: String, the name of a file containing JSON credentials to load.
The same filename will be used to save credentials back to disk.
Returns:
The credentials loaded from disk.
Raises:
InvalidCredentialsFileError: When the credentials file cannot be opened.
EmptyCredentialsFileError: When the provided file contains no credentials.
"""
file_content = fileutils.read_file(
filename, continue_on_error=True, display_errors=False)
if file_content is None:
raise InvalidCredentialsFileError(f'File {filename} could not be opened')
info = json.loads(file_content)
if not info:
raise EmptyCredentialsFileError(
f'File {filename} contains no credential data')
try:
# We read the existing data from the passed in file, but we also want to
# save future data/tokens in the same place.
return cls.from_authorized_user_info(info, filename=filename)
except ValueError as e:
raise InvalidCredentialsFileError(str(e))
@classmethod
def from_client_secrets(cls,
client_id,
client_secret,
scopes,
access_type='offline',
login_hint=None,
filename=None,
use_console_flow=False):
"""Runs an OAuth Flow from client secrets to generate credentials.
Args:
client_id: String, The OAuth2.0 Client ID.
client_secret: String, The OAuth2.0 Client Secret.
scopes: Sequence[str], A list of scopes to include in the credentials.
access_type: String, 'offline' or 'online'. Indicates whether your
application can refresh access tokens when the user is not present at
the browser. Valid parameter values are online, which is the default
value, and offline. Set the value to offline if your application needs
to refresh access tokens when the user is not present at the browser.
This is the method of refreshing access tokens described later in this
document. This value instructs the Google authorization server to return
a refresh token and an access token the first time that your application
exchanges an authorization code for tokens.
login_hint: String, The email address that will be displayed on the Google
login page as a hint for the user to login to the correct account.
filename: String, the path to a file to use to save the credentials.
use_console_flow: Boolean, True if the authentication flow should be run
strictly from a console; False to launch a browser for authentication.
Returns:
Credentials
"""
client_config = {
'installed': {
'client_id': client_id,
'client_secret': client_secret,
'redirect_uris': ['http://localhost', 'urn:ietf:wg:oauth:2.0:oob'],
'auth_uri': 'https://accounts.google.com/o/oauth2/v2/auth',
'token_uri': 'https://oauth2.googleapis.com/token',
}
}
flow = _ShortURLFlow.from_client_config(
client_config, scopes, autogenerate_code_verifier=True)
flow_kwargs = {'access_type': access_type}
if login_hint:
flow_kwargs['login_hint'] = login_hint
# TODO: Move code for browser detection somewhere in this file so that the
# messaging about `nobrowser.txt` is co-located with the logic that uses it.
if use_console_flow:
flow.run_console(
authorization_prompt_message=MESSAGE_CONSOLE_AUTHORIZATION_PROMPT,
authorization_code_message=MESSAGE_CONSOLE_AUTHORIZATION_CODE,
**flow_kwargs)
else:
flow.run_local_server(
authorization_prompt_message=MESSAGE_LOCAL_SERVER_AUTHORIZATION_PROMPT,
success_message=MESSAGE_LOCAL_SERVER_SUCCESS,
**flow_kwargs)
return cls.from_google_oauth2_credentials(
flow.credentials, filename=filename)
@classmethod
def from_client_secrets_file(cls,
client_secrets_file,
scopes,
access_type='offline',
login_hint=None,
credentials_file=None,
use_console_flow=False):
"""Runs an OAuth Flow from secrets stored on disk to generate credentials.
Args:
client_secrets_file: String, path to a file containing a client ID and
secret.
scopes: Sequence[str], A list of scopes to include in the credentials.
access_type: String, 'offline' or 'online'. Indicates whether your
application can refresh access tokens when the user is not present at
the browser. Valid parameter values are online, which is the default
value, and offline. Set the value to offline if your application needs
to refresh access tokens when the user is not present at the browser.
This is the method of refreshing access tokens described later in this
document. This value instructs the Google authorization server to return
a refresh token and an access token the first time that your application
exchanges an authorization code for tokens.
login_hint: String, The email address that will be displayed on the Google
login page as a hint for the user to login to the correct account.
credentials_file: String, the path to a file to use to save the
credentials.
use_console_flow: Boolean, True if the authentication flow should be run
strictly from a console; False to launch a browser for authentication.
Raises:
InvalidClientSecretsFileError: If the client secrets file cannot be
opened.
InvalidClientSecretsFileFormatError: If the client secrets file does not
contain the required data or the data is malformed.
Returns:
Credentials
"""
cs_data = fileutils.read_file(
client_secrets_file, continue_on_error=True, display_errors=False)
if not cs_data:
raise InvalidClientSecretsFileError(
f'File {client_secrets_file} could not be opened')
try:
cs_json = json.loads(cs_data)
client_id = cs_json['installed']['client_id']
# Chop off .apps.googleusercontent.com suffix as it's not needed
# and we need to keep things short for the Auth URL.
client_id = re.sub(r'\.apps\.googleusercontent\.com$', '', client_id)
client_secret = cs_json['installed']['client_secret']
except (ValueError, IndexError, KeyError):
raise InvalidClientSecretsFileFormatError(
f'Could not extract Client ID or Client Secret from file {client_secrets_file}'
)
return cls.from_client_secrets(
client_id,
client_secret,
scopes,
access_type=access_type,
login_hint=login_hint,
filename=credentials_file,
use_console_flow=use_console_flow)
def _fetch_id_token_data(self):
"""Fetches verification details from Google for the OAuth2.0 token.
See more: https://developers.google.com/identity/sign-in/web/backend-auth
Raises:
CredentialsError: If no id_token is present.
"""
if not self.id_token:
raise CredentialsError('Failed to fetch token data. No id_token present.')
request = transport.create_request()
self._id_token_data = google.oauth2.id_token.verify_oauth2_token(
self.id_token, request)
def get_token_value(self, field):
"""Retrieves data from the OAuth ID token.
See more: https://developers.google.com/identity/sign-in/web/backend-auth
Args:
field: The name of the key/field to fetch
Returns:
The value associated with the given key or 'Unknown' if the key data can
not be found in the access token data.
"""
if not self._id_token_data:
self._fetch_id_token_data()
# Maintain legacy GAM behavior here to return "Unknown" if the field is
# otherwise unpopulated.
return self._id_token_data.get(field, 'Unknown')
def to_json(self, strip=None):
"""Creates a JSON representation of a Credentials.
Args:
strip: Sequence[str], Optional list of members to exclude from the
generated JSON.
Returns:
str: A JSON representation of this instance, suitable to pass to
from_json().
"""
expiry = self.expiry.strftime(
Credentials.DATETIME_FORMAT) if self.expiry else None
prep = {
'token': self.token,
'refresh_token': self.refresh_token,
'token_uri': self.token_uri,
'client_id': self.client_id,
'client_secret': self.client_secret,
'id_token': self.id_token,
# Google auth doesn't currently give us scopes back on refresh.
# 'scopes': sorted(self.scopes),
'token_expiry': expiry,
'decoded_id_token': self._id_token_data,
}
# Remove empty entries
prep = {k: v for k, v in prep.items() if v is not None}
# Remove entries that explicitly need to be removed
if strip is not None:
prep = {k: v for k, v in prep.items() if k not in strip}
return json.dumps(prep, indent=2, sort_keys=True)
def refresh(self, request=None):
"""Refreshes the credential's access token.
Args:
request: google.auth.transport.Request, The object used to make HTTP
requests. If not provided, a default request will be used.
Raises:
google.auth.exceptions.RefreshError: If the credentials could not be
refreshed.
"""
with self._lock:
if request is None:
request = transport.create_request()
self._locked_refresh(request)
# Save the new tokens back to disk, if these credentials are disk-backed.
if self._filename:
self._locked_write()
def _locked_refresh(self, request):
"""Refreshes the credential's access token while the file lock is held."""
assert self._lock.is_locked
super(Credentials, self).refresh(request)
def write(self):
"""Writes credentials to disk."""
with self._lock:
self._locked_write()
def _locked_write(self):
"""Writes credentials to disk while the file lock is held."""
assert self._lock.is_locked
if not self.filename:
# If no filename was provided to the constructor, these credentials cannot
# be saved to disk.
raise CredentialsError(
'The credentials have no associated filename and cannot be saved '
'to disk.')
fileutils.write_file(self._filename, self.to_json())
def delete(self):
"""Deletes all files on disk related to these credentials."""
with self._lock:
# Only attempt to remove the file if the lock we're using is a FileLock.
if isinstance(self._lock, FileLock):
os.remove(self._filename)
if self._lock.lock_file and not GM_Globals[GM_WINDOWS]:
os.remove(self._lock.lock_file)
_REVOKE_TOKEN_BASE_URI = 'https://accounts.google.com/o/oauth2/revoke'
def revoke(self, http=None):
"""Revokes this credential's access token with the server.
Args:
http: httplib2.Http compatible object for use as a transport. If no http
is provided, a default will be used.
"""
with self._lock:
if http is None:
http = transport.create_http()
params = urlencode({'token': self.refresh_token})
revoke_uri = f'{Credentials._REVOKE_TOKEN_BASE_URI}?{params}'
http.request(revoke_uri, 'GET')
class _ShortURLFlow(google_auth_oauthlib.flow.InstalledAppFlow):
"""InstalledAppFlow which utilizes a URL shortener for authorization URLs."""
URL_SHORTENER_ENDPOINT = 'https://gam-shortn.appspot.com/create'
def authorization_url(self, http=None, **kwargs):
"""Gets a shortened authorization URL."""
long_url, state = super(_ShortURLFlow, self).authorization_url(**kwargs)
if not http:
http = transport.create_http(timeout=10)
headers = {'Content-Type': 'application/json', 'User-Agent': GAM_INFO}
try:
payload = json.dumps({'long_url': long_url})
resp, content = http.request(
_ShortURLFlow.URL_SHORTENER_ENDPOINT,
'POST',
payload,
headers=headers)
except:
return long_url, state
if resp.status != 200:
return long_url, state
try:
if isinstance(content, bytes):
content = content.decode()
return json.loads(content).get('short_url', long_url), state
except:
return long_url, state
class _FileLikeThreadLock(object):
"""A threading.lock which has the same interface as filelock.Filelock."""
def __init__(self):
"""A shell object that holds a threading.Lock.
Since we cannot inherit from built-in classes such as threading.Lock, we
just use a shell object and maintain a lock inside of it.
"""
self._lock = threading.Lock()
def __enter__(self, *args, **kwargs):
return self._lock.__enter__(*args, **kwargs)
def __exit__(self, *args, **kwargs):
return self._lock.__exit__(*args, **kwargs)
def acquire(self, **kwargs):
return self._lock.acquire(**kwargs)
def release(self):
return self._lock.release()
@property
def is_locked(self):
return self._lock.locked()
@property
def lock_file(self):
return None

669
src/auth/oauth_test.py Normal file
View File

@@ -0,0 +1,669 @@
"""Tests for oauth."""
import datetime
import json
import os
import platform
import unittest
from unittest.mock import MagicMock
from unittest.mock import patch
import google.oauth2.credentials
from auth import oauth
class CredentialsTest(unittest.TestCase):
def setUp(self):
self.fake_token = 'fake_token'
self.fake_refresh_token = 'fake_refresh_token'
self.fake_id_token = 'fake_id_token'
self.fake_token_uri = 'https://fake.token.uri'
self.fake_client_id = 'fake_client_id'
self.fake_client_secret = 'fake_client_secret'
self.fake_scopes = [
'fake_api.readonly',
'fake_other_api.write',
]
self.fake_quota_project_id = 'fake_quota_project_id'
self.fake_token_expiry = datetime.datetime(2020, 1, 1, 10)
self.fake_filename = 'fake_filename'
self.fake_token_data = {
'field': 'value',
'another-field': 'another-value',
}
self.info_with_only_required_fields = {
'refresh_token': self.fake_refresh_token,
'client_id': self.fake_client_id,
'client_secret': self.fake_client_secret,
}
super(CredentialsTest, self).setUp()
def tearDown(self):
# Remove any credential files that may have been created.
if os.path.exists(self.fake_filename):
os.remove(self.fake_filename)
if os.path.exists('%s.lock' % self.fake_filename):
os.remove('%s.lock' % self.fake_filename)
super(CredentialsTest, self).tearDown()
def test_from_authorized_user_info_only_required_info(self):
creds = oauth.Credentials.from_authorized_user_info(
self.info_with_only_required_fields)
self.assertEqual(self.fake_refresh_token, creds.refresh_token)
self.assertEqual(self.fake_client_id, creds.client_id)
self.assertEqual(self.fake_client_secret, creds.client_secret)
self.assertIsNone(creds.id_token)
self.assertIsNone(creds.expiry)
self.assertIsNone(creds.filename)
def test_from_authorized_user_info_all_info_provided(self):
info = {
'token':
self.fake_token,
'refresh_token':
self.fake_refresh_token,
'id_token':
self.fake_id_token,
'token_uri':
self.fake_token_uri,
'client_id':
self.fake_client_id,
'client_secret':
self.fake_client_secret,
'token_expiry':
self.fake_token_expiry.strftime(oauth.Credentials.DATETIME_FORMAT),
'id_token_data':
self.fake_token_data,
}
creds = oauth.Credentials.from_authorized_user_info(info)
self.assertEqual(self.fake_refresh_token, creds.refresh_token)
self.assertEqual(self.fake_client_id, creds.client_id)
self.assertEqual(self.fake_client_secret, creds.client_secret)
self.assertEqual(self.fake_id_token, creds.id_token)
self.assertEqual(self.fake_token_uri, creds.token_uri)
self.assertEqual(self.fake_token_expiry, creds.expiry)
self.assertIsNone(creds.filename)
def test_from_authorized_user_info_missing_required_info(self):
info_with_missing_fields = {'token': self.fake_token}
with self.assertRaises(ValueError):
oauth.Credentials.from_authorized_user_info(info_with_missing_fields)
def test_from_authorized_user_info_no_expiry_in_info(self):
info_with_no_token_expiry = self.info_with_only_required_fields.copy()
self.assertIsNone(info_with_no_token_expiry.get('expiry'))
creds = oauth.Credentials.from_authorized_user_info(
info_with_no_token_expiry)
self.assertIsNone(creds.expiry)
def test_init_saves_filename(self):
creds = oauth.Credentials(
token=self.fake_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
self.assertEqual(os.path.abspath(self.fake_filename), creds.filename)
@patch.object(oauth.google.oauth2.id_token, 'verify_oauth2_token')
def test_init_loads_decoded_id_token_data(self, mock_verify_token):
creds = oauth.Credentials(
token=self.fake_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token,
id_token_data=self.fake_token_data)
self.assertEqual(
self.fake_token_data.get('field'), creds.get_token_value('field'))
# Verify the fetching method was not called, since the token
# data was supposed to be loaded from the passed in info.
self.assertEqual(mock_verify_token.call_count, 0)
def test_credentials_uses_file_lock_when_filename_provided(self):
creds = oauth.Credentials(
token=self.fake_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
self.assertIsInstance(creds._lock, oauth.FileLock)
self.assertEqual(creds._lock.lock_file, '%s.lock' % creds.filename)
def test_credentials_uses_thread_lock_when_filename_not_provided(self):
creds = oauth.Credentials(
token=self.fake_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=None)
self.assertIsInstance(creds._lock, oauth._FileLikeThreadLock)
self.assertIsNone(creds.filename)
def test_from_oauth2credentials(self):
google_creds = google.oauth2.credentials.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token)
creds = oauth.Credentials.from_google_oauth2_credentials(
google_creds, filename=self.fake_filename)
self.assertEqual(google_creds.token, creds.token)
self.assertEqual(google_creds.refresh_token, creds.refresh_token)
self.assertEqual(google_creds.client_id, creds.client_id)
self.assertEqual(google_creds.client_secret, creds.client_secret)
self.assertEqual(google_creds.id_token, creds.id_token)
self.assertEqual(google_creds.expiry, creds.expiry)
self.assertEqual(google_creds.quota_project_id, creds.quota_project_id)
self.assertEqual(os.path.abspath(self.fake_filename), creds.filename)
def test_from_credentials_file_corrupt_or_missing_file_raises_error(self):
self.assertFalse(os.path.exists(self.fake_filename))
with self.assertRaises(oauth.InvalidCredentialsFileError) as e:
oauth.Credentials.from_credentials_file(self.fake_filename)
self.assertIn('could not be opened', str(e.exception))
@patch.object(oauth.fileutils, 'read_file')
def test_from_credentials_file_no_serialized_data_in_file_raises_error(
self, mock_read_file):
mock_read_file.return_value = json.dumps({})
with self.assertRaises(oauth.EmptyCredentialsFileError):
oauth.Credentials.from_credentials_file(self.fake_filename)
@patch.object(oauth.fileutils, 'read_file')
def test_from_credentials_file_missing_required_info_raises_error(
self, mock_read_file):
mock_read_file.return_value = json.dumps({
# This data is missing the required refresh_token key/value pair
'client_id': self.fake_client_id,
'client_secret': self.fake_client_secret,
})
with self.assertRaises(oauth.InvalidCredentialsFileError):
oauth.Credentials.from_credentials_file(self.fake_filename)
@patch.object(oauth._ShortURLFlow, 'from_client_config')
def test_from_client_secrets_console_flow(self, mock_flow):
flow_creds = google.oauth2.credentials.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token)
mock_flow.return_value.credentials = flow_creds
creds = oauth.Credentials.from_client_secrets(
self.fake_client_id,
self.fake_client_secret,
self.fake_scopes,
use_console_flow=True)
self.assertTrue(mock_flow.return_value.run_console.called)
self.assertFalse(mock_flow.return_value.run_local_server.called)
self.assertEqual(flow_creds.token, creds.token)
self.assertEqual(flow_creds.refresh_token, creds.refresh_token)
self.assertEqual(flow_creds.client_id, creds.client_id)
self.assertEqual(flow_creds.client_secret, creds.client_secret)
self.assertEqual(flow_creds.id_token, creds.id_token)
@patch.object(oauth._ShortURLFlow, 'from_client_config')
def test_from_client_secrets_local_server_flow(self, mock_flow):
flow_creds = google.oauth2.credentials.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token)
mock_flow.return_value.credentials = flow_creds
creds = oauth.Credentials.from_client_secrets(
self.fake_client_id,
self.fake_client_secret,
self.fake_scopes,
use_console_flow=False)
self.assertFalse(mock_flow.return_value.run_console.called)
self.assertTrue(mock_flow.return_value.run_local_server.called)
self.assertEqual(flow_creds.token, creds.token)
self.assertEqual(flow_creds.refresh_token, creds.refresh_token)
self.assertEqual(flow_creds.client_id, creds.client_id)
self.assertEqual(flow_creds.client_secret, creds.client_secret)
self.assertEqual(flow_creds.id_token, creds.id_token)
@patch.object(oauth._ShortURLFlow, 'from_client_config')
def test_from_client_secrets_uses_login_hint(self, mock_flow):
flow_creds = google.oauth2.credentials.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token)
mock_flow.return_value.credentials = flow_creds
oauth.Credentials.from_client_secrets(
self.fake_client_id,
self.fake_client_secret,
self.fake_scopes,
login_hint='someone@domain.com')
run_flow_args = mock_flow.return_value.run_local_server.call_args[1]
self.assertEqual('someone@domain.com', run_flow_args.get('login_hint'))
def test_from_client_secrets_uses_shortened_url_flow(self):
with patch.object(oauth._ShortURLFlow, 'from_client_config') as mock_flow:
flow_creds = google.oauth2.credentials.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token)
mock_flow.return_value.credentials = flow_creds
oauth.Credentials.from_client_secrets(self.fake_client_id,
self.fake_client_secret,
self.fake_scopes)
self.assertTrue(mock_flow.called)
@patch.object(oauth._ShortURLFlow, 'from_client_config')
def test_from_client_secrets_passes_credentials_filename(self, mock_flow):
flow_creds = google.oauth2.credentials.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token=self.fake_id_token)
mock_flow.return_value.credentials = flow_creds
creds = oauth.Credentials.from_client_secrets(
self.fake_client_id,
self.fake_client_secret,
self.fake_scopes,
filename=self.fake_filename)
self.assertEqual(os.path.abspath(self.fake_filename), creds.filename)
def test_from_client_secrets_file_corrupt_or_missing_file_raises_error(self):
self.assertFalse(os.path.exists(self.fake_filename))
with self.assertRaises(oauth.InvalidClientSecretsFileError):
oauth.Credentials.from_client_secrets_file(self.fake_filename,
self.fake_scopes)
@patch.object(oauth.fileutils, 'read_file')
def test_from_client_secrets_file_missing_required_json_raises_error(
self, mock_read_file):
mock_read_file.return_value = json.dumps({})
with self.assertRaises(oauth.InvalidClientSecretsFileFormatError) as e:
oauth.Credentials.from_client_secrets_file(self.fake_filename,
self.fake_scopes)
self.assertIn('Could not extract Client ID or Client Secret',
str(e.exception))
@patch.object(oauth.Credentials, 'from_client_secrets')
@patch.object(oauth.fileutils, 'read_file')
def test_from_client_secrets_file_strips_domain_from_client_id(
self, mock_read_file, mock_creds_from_client_secrets):
mock_read_file.return_value = json.dumps({
'installed': {
'client_id': self.fake_client_id + '.apps.googleusercontent.com',
'client_secret': self.fake_client_secret,
}
})
oauth.Credentials.from_client_secrets_file(self.fake_filename,
self.fake_scopes)
self.assertEqual(self.fake_client_id,
mock_creds_from_client_secrets.call_args[0][0])
def test_get_token_value_known_token_field(self):
token_data = {'known-field': 'known-value'}
creds = oauth.Credentials(
token=self.fake_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token_data=token_data)
self.assertEqual('known-value', creds.get_token_value('known-field'))
def test_get_token_value_unknown_field_returns_unknown(self):
creds = oauth.Credentials(
token=self.fake_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
id_token_data=self.fake_token_data)
self.assertEqual('Unknown', creds.get_token_value('unknown-field'))
def test_to_json_contains_all_required_fields(self):
creds = oauth.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
id_token=self.fake_id_token,
id_token_data=self.fake_token_data,
token_uri=self.fake_token_uri,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
scopes=self.fake_scopes,
quota_project_id=self.fake_quota_project_id,
expiry=self.fake_token_expiry)
json_string = creds.to_json()
json_data = json.loads(json_string)
keys = json_data.keys()
self.assertIn('token', keys)
self.assertEqual(self.fake_token, json_data['token'])
self.assertIn('refresh_token', keys)
self.assertEqual(self.fake_refresh_token, json_data['refresh_token'])
self.assertIn('id_token', keys)
self.assertEqual(self.fake_id_token, json_data['id_token'])
self.assertIn('token_uri', keys)
self.assertEqual(self.fake_token_uri, json_data['token_uri'])
self.assertIn('client_id', keys)
self.assertEqual(self.fake_client_id, json_data['client_id'])
self.assertIn('client_secret', keys)
self.assertEqual(self.fake_client_secret, json_data['client_secret'])
self.assertNotIn('scopes', keys) # Scopes are not currently saved
self.assertIn('token_expiry', keys)
self.assertEqual(
self.fake_token_expiry.strftime(oauth.Credentials.DATETIME_FORMAT),
json_data['token_expiry'])
self.assertIn('decoded_id_token', keys)
self.assertEqual(self.fake_token_data, json_data['decoded_id_token'])
def test_credentials_to_json_and_back(self):
original_creds = oauth.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
id_token=self.fake_id_token,
id_token_data=self.fake_token_data,
token_uri=self.fake_token_uri,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
scopes=self.fake_scopes,
quota_project_id=self.fake_quota_project_id,
expiry=self.fake_token_expiry)
pickled_creds = original_creds.to_json()
serialized_json = json.loads(pickled_creds)
unpickled_creds = oauth.Credentials.from_authorized_user_info(
serialized_json)
self.assertEqual(original_creds.token, unpickled_creds.token)
self.assertEqual(original_creds.refresh_token,
unpickled_creds.refresh_token)
self.assertEqual(original_creds.id_token, unpickled_creds.id_token)
self.assertEqual(original_creds.token_uri, unpickled_creds.token_uri)
self.assertEqual(original_creds.client_id, unpickled_creds.client_id)
self.assertEqual(original_creds.client_secret,
unpickled_creds.client_secret)
self.assertEqual(original_creds.expiry, unpickled_creds.expiry)
@patch.object(oauth.google.oauth2.credentials.Credentials, 'refresh')
def test_refresh_calls_super_refresh(self, mock_super_refresh):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret)
request = MagicMock()
creds.refresh(request)
self.assertTrue(mock_super_refresh.called)
self.assertEqual(request, mock_super_refresh.call_args[0][0])
def test_refresh_locks_resource_during_refresh(self):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret)
lock = creds._lock
def check_lock_is_locked(*unused_args, **unused_kwargs):
self.assertTrue(lock.is_locked)
# We need to mock the superclass refresh so it doesn't actually try to
# refresh our fake token.
# At the same time, we'll make sure the lock is held during the refresh.
with patch.object(oauth.google.oauth2.credentials.Credentials,
'refresh') as mock_refresh:
mock_refresh.side_effect = check_lock_is_locked
creds.refresh(request=MagicMock())
# Make sure our side effect was actually performed.
self.assertTrue(mock_refresh.called)
# The lock should be released after refresh
self.assertFalse(lock.is_locked)
@patch.object(oauth.google.oauth2.credentials.Credentials, 'refresh')
@patch.object(oauth.fileutils, 'write_file')
def test_refresh_writes_new_credentials_to_disk_after_refresh(
self, mock_write_file, mock_super_refresh):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
def update_access_token(unused_request):
creds.token = 'refreshed_access_token'
mock_super_refresh.side_effect = update_access_token
self.assertIsNone(creds.token)
creds.refresh(request=MagicMock())
self.assertEqual('refreshed_access_token', creds.token,
'Access token was not refreshed')
text_written_to_file = mock_write_file.call_args[0][1]
self.assertIsNotNone(text_written_to_file, 'Nothing was written to file')
saved_json = json.loads(text_written_to_file)
self.assertEqual('refreshed_access_token', saved_json['token'],
'Refreshed access token was not saved to disk')
def test_write_writes_credentials_to_disk(self):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
self.assertFalse(os.path.exists(self.fake_filename))
creds.write()
self.assertTrue(os.path.exists(self.fake_filename))
def test_write_raises_error_when_no_credentials_file_is_set(self):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret)
self.assertIsNone(creds.filename)
with self.assertRaises(oauth.CredentialsError):
creds.write()
@patch.object(oauth.google.oauth2.credentials.Credentials, 'refresh')
@patch.object(oauth.fileutils, 'write_file')
def test_write_locks_resource_during_write(self, mock_write_file,
unused_mock_super_refresh):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
lock = creds._lock
def check_lock_is_locked(*unused_args, **unused_kwargs):
self.assertTrue(creds._lock.is_locked)
mock_write_file.side_effect = check_lock_is_locked
self.assertFalse(lock.is_locked)
creds.refresh(request=MagicMock())
self.assertFalse(lock.is_locked)
self.assertTrue(mock_write_file.called)
def test_delete_removes_credentials_file(self):
self.assertFalse(os.path.exists(self.fake_filename))
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
creds.write()
self.assertTrue(os.path.exists(self.fake_filename))
creds.delete()
self.assertFalse(os.path.exists(self.fake_filename))
@unittest.skipIf(
platform.system() == 'Windows',
reason=('On Windows, Filelock deletes the lock file each time the lock '
'is released. Delete does not remove it.'))
def test_delete_removes_lock_file(self):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret,
filename=self.fake_filename)
lock_file = '%s.lock' % creds.filename
creds.write()
self.assertTrue(os.path.exists(lock_file))
creds.delete()
self.assertFalse(os.path.exists(lock_file))
def test_delete_is_noop_when_not_using_filelock(self):
creds = oauth.Credentials(
token=None,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret)
self.assertIsNone(creds.filename)
creds.delete() # This should not raise an exception.
def test_revoke_requests_credential_revoke(self):
creds = oauth.Credentials(
token=self.fake_token,
refresh_token=self.fake_refresh_token,
client_id=self.fake_client_id,
client_secret=self.fake_client_secret)
mock_http = MagicMock()
creds.revoke(http=mock_http)
uri = mock_http.request.call_args[0][0]
self.assertRegex(uri, '^%s' % oauth.Credentials._REVOKE_TOKEN_BASE_URI)
params = uri[uri.index('?'):]
self.assertIn('token=%s' % creds.refresh_token, params)
self.assertEqual('GET', mock_http.request.call_args[0][1])
class ShortUrlFlowTest(unittest.TestCase):
def setUp(self):
self.fake_client_id = 'fake_client_id'
self.fake_client_secret = 'fake_client_secret'
self.fake_scopes = [
'fake_api.readonly',
'fake_other_api.write',
]
self.fake_client_config = {
'installed': {
'client_id': self.fake_client_id,
'client_secret': self.fake_client_secret,
'redirect_uris': ['http://localhost', 'urn:ietf:wg:oauth:2.0:oob'],
'auth_uri': 'https://accounts.google.com/o/oauth2/v2/auth',
'token_uri': 'https://oauth2.googleapis.com/token',
}
}
self.long_url = 'http://example.com/some/long/url'
self.short_url = 'http://ex.co/short'
super(ShortUrlFlowTest, self).setUp()
@patch.object(oauth.google_auth_oauthlib.flow.InstalledAppFlow,
'authorization_url')
def test_shorturlflow_returns_shortened_url(self, mock_super_auth_url):
url_flow = oauth._ShortURLFlow.from_client_config(
self.fake_client_config, scopes=self.fake_scopes)
mock_super_auth_url.return_value = (self.long_url, 'fake_state')
mock_http = MagicMock()
mock_response = MagicMock()
mock_response.status = 200
content = json.dumps({'short_url': self.short_url})
mock_http.request.return_value = (mock_response, content)
url, state = url_flow.authorization_url(http=mock_http)
self.assertEqual(self.short_url, url)
self.assertEqual('fake_state', state)
# Verify request() was called with the expected arguments.
self.assertEqual(oauth._ShortURLFlow.URL_SHORTENER_ENDPOINT,
mock_http.request.call_args[0][0])
self.assertEqual('POST', mock_http.request.call_args[0][1])
self.assertIn(self.long_url, mock_http.request.call_args[0][2])
@patch.object(oauth.google_auth_oauthlib.flow.InstalledAppFlow,
'authorization_url')
def test_shorturlflow_falls_back_to_long_url_on_request_error(
self, mock_super_auth_url):
url_flow = oauth._ShortURLFlow.from_client_config(
self.fake_client_config, scopes=self.fake_scopes)
mock_super_auth_url.return_value = (self.long_url, 'fake_state')
mock_http = MagicMock()
mock_http.request.side_effect = Exception()
url, state = url_flow.authorization_url(http=mock_http)
self.assertEqual(self.long_url, url)
self.assertEqual('fake_state', state)
@patch.object(oauth.google_auth_oauthlib.flow.InstalledAppFlow,
'authorization_url')
def test_shorturlflow_falls_back_to_long_url_on_non_200_response_status(
self, mock_super_auth_url):
url_flow = oauth._ShortURLFlow.from_client_config(
self.fake_client_config, scopes=self.fake_scopes)
mock_super_auth_url.return_value = (self.long_url, 'fake_state')
mock_http = MagicMock()
mock_response = MagicMock()
mock_response.status = 404 # Use a status that is not 200
content = json.dumps({'short_url': self.short_url})
mock_http.request.return_value = (mock_response, content)
url, state = url_flow.authorization_url(http=mock_http)
self.assertEqual(self.long_url, url)
self.assertEqual('fake_state', state)
@patch.object(oauth.google_auth_oauthlib.flow.InstalledAppFlow,
'authorization_url')
def test_shorturlflow_falls_back_to_long_url_on_bad_json_response(
self, mock_super_auth_url):
url_flow = oauth._ShortURLFlow.from_client_config(
self.fake_client_config, scopes=self.fake_scopes)
mock_super_auth_url.return_value = (self.long_url, 'fake_state')
mock_http = MagicMock()
mock_response = MagicMock()
mock_response.status = 200
content = None
mock_http.request.return_value = (mock_response, content)
url, state = url_flow.authorization_url(http=mock_http)
self.assertEqual(self.long_url, url)
self.assertEqual('fake_state', state)
@patch.object(oauth.google_auth_oauthlib.flow.InstalledAppFlow,
'authorization_url')
def test_shorturlflow_falls_back_to_long_url_on_empty_short_url_field(
self, mock_super_auth_url):
url_flow = oauth._ShortURLFlow.from_client_config(
self.fake_client_config, scopes=self.fake_scopes)
mock_super_auth_url.return_value = (self.long_url, 'fake_state')
mock_http = MagicMock()
mock_response = MagicMock()
mock_response.status = 200
content = json.dumps({}) # This json content contains no "short-url" key
mock_http.request.return_value = (mock_response, content)
url, state = url_flow.authorization_url(http=mock_http)
self.assertEqual(self.long_url, url)
self.assertEqual('fake_state', state)
if __name__ == '__main__':
unittest.main()

246
src/gam.py
View File

@@ -51,16 +51,14 @@ import http.client as http_client
from multiprocessing import Pool as mp_pool
from multiprocessing import freeze_support as mp_freeze_support
from multiprocessing import set_start_method as mp_set_start_method
from urllib.parse import quote, urlencode, urlparse
from urllib.parse import urlencode, urlparse
import dateutil.parser
import googleapiclient
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http
import google.oauth2.id_token
import google.oauth2.service_account
import google_auth_oauthlib.flow
import httplib2
from cryptography import x509
@@ -69,8 +67,7 @@ from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from filelock import FileLock
import auth.oauth
import controlflow
import display
import fileutils
@@ -703,49 +700,24 @@ def readDiscoveryFile(api_version):
controlflow.invalid_json_exit(disc_file)
def getOauth2TxtStorageCredentials():
oauth_string = fileutils.read_file(GC_Values[GC_OAUTH2_TXT], continue_on_error=True, display_errors=False)
if not oauth_string:
try:
return auth.get_admin_credentials()
except auth.oauth.InvalidCredentialsFileError:
# Maintain legacy behavior of this method that returns None if no
# credential file is present.
return None
oauth_data = json.loads(oauth_string)
creds = google.oauth2.credentials.Credentials.from_authorized_user_file(GC_Values[GC_OAUTH2_TXT])
creds.token = oauth_data.get('token', oauth_data.get('auth_token', ''))
creds._id_token = oauth_data.get('id_token_jwt', oauth_data.get('id_token', None))
token_expiry = oauth_data.get('token_expiry', '1970-01-01T00:00:01Z')
creds.expiry = datetime.datetime.strptime(token_expiry, '%Y-%m-%dT%H:%M:%SZ')
GC_Values[GC_DECODED_ID_TOKEN] = oauth_data.get('decoded_id_token', '')
return creds
def getValidOauth2TxtCredentials(force_refresh=False):
"""Gets OAuth2 credentials which are guaranteed to be fresh and valid.
Locks during read and possible write so that only one process will
attempt refresh/write when running in parallel. """
lock_file = f'{GC_Values[GC_OAUTH2_TXT]}.lock'
lock = FileLock(lock_file)
with lock:
credentials = getOauth2TxtStorageCredentials()
if (credentials and credentials.expired) or force_refresh:
retries = 3
for n in range(1, retries+1):
try:
credentials.refresh(transport.create_request())
writeCredentials(credentials)
break
except google.auth.exceptions.RefreshError as e:
try:
if e.args[0] in REFRESH_PERM_ERRORS:
# remove OAuth file so we kick off auth next time
os.remove(GC_Values[GC_OAUTH2_TXT])
except SyntaxError:
pass
controlflow.system_error_exit(18, str(e))
except (google.auth.exceptions.TransportError, httplib2.ServerNotFoundError, RuntimeError) as e:
if n != retries:
controlflow.wait_on_failure(n, retries, str(e))
continue
controlflow.system_error_exit(4, str(e))
elif credentials is None or not credentials.valid:
doRequestOAuth()
credentials = getOauth2TxtStorageCredentials()
"""Gets OAuth2 credentials which are guaranteed to be fresh and valid."""
try:
credentials = auth.get_admin_credentials()
except auth.oauth.InvalidCredentialsFileError:
doRequestOAuth() # Make a new request which should store new creds.
return getValidOauth2TxtCredentials(force_refresh=force_refresh)
if credentials.expired or force_refresh:
request = transport.create_request()
credentials.refresh(request)
return credentials
def getService(api, http):
@@ -6198,57 +6170,6 @@ def getUserAttributes(i, cd, updateCmd):
body['hashFunction'] = 'crypt'
return body
def shorten_url(long_url):
simplehttp = transport.create_http(timeout=10)
url_shortnr = 'https://gam-shortn.appspot.com/create'
headers = {'Content-Type': 'application/json',
'User-Agent': GAM_INFO}
try:
resp, content = simplehttp.request(url_shortnr, 'POST',
f'{{"long_url": "{long_url}"}}', headers=headers)
except Exception:
return long_url
if resp.status != 200:
return long_url
try:
return json.loads(content).get('short_url', long_url)
except Exception:
print(content)
return long_url
class ShortURLFlow(google_auth_oauthlib.flow.InstalledAppFlow):
def authorization_url(self, **kwargs):
long_url, state = super(ShortURLFlow, self).authorization_url(**kwargs)
short_url = shorten_url(long_url)
return short_url, state
def _run_oauth_flow(client_id, client_secret, scopes, access_type, login_hint=None):
client_config = {
'installed': {
'client_id': client_id,
'client_secret': client_secret,
'redirect_uris': ['http://localhost', 'urn:ietf:wg:oauth:2.0:oob'],
'auth_uri': 'https://accounts.google.com/o/oauth2/v2/auth',
'token_uri': 'https://oauth2.googleapis.com/token',
}
}
flow = ShortURLFlow.from_client_config(client_config, scopes, autogenerate_code_verifier=True)
kwargs = {'access_type': access_type}
if login_hint:
kwargs['login_hint'] = login_hint
if not GC_Values[GC_OAUTH_BROWSER]:
flow.run_console(
authorization_prompt_message=MESSAGE_CONSOLE_AUTHORIZATION_PROMPT,
authorization_code_message=MESSAGE_CONSOLE_AUTHORIZATION_CODE,
**kwargs)
else:
flow.run_local_server(
authorization_prompt_message=MESSAGE_LOCAL_SERVER_AUTHORIZATION_PROMPT,
success_message=MESSAGE_LOCAL_SERVER_SUCCESS,
**kwargs)
return flow.credentials
def getCRMService(login_hint):
scopes = ['https://www.googleapis.com/auth/cloud-platform']
client_id = '297408095146-fug707qsjv4ikron0hugpevbrjhkmsk7.apps.googleusercontent.com'
@@ -7906,11 +7827,9 @@ def doCreateResoldCustomer():
print(f'Created customer {result["customerDomain"]} with id {result["customerId"]}')
def _getValueFromOAuth(field, credentials=None):
if not GC_Values[GC_DECODED_ID_TOKEN]:
credentials = credentials if credentials is not None else getValidOauth2TxtCredentials()
request = transport.create_request()
GC_Values[GC_DECODED_ID_TOKEN] = google.oauth2.id_token.verify_oauth2_token(credentials.id_token, request)
return GC_Values[GC_DECODED_ID_TOKEN].get(field, 'Unknown')
if not credentials:
credentials = auth.get_admin_credentials()
return credentials.get_token_value(field)
def doGetMemberInfo():
cd = buildGAPIObject('directory')
@@ -10151,89 +10070,56 @@ def OAuthInfo():
print(f'{key}: {value}')
def doDeleteOAuth():
lock_file = f'{GC_Values[GC_OAUTH2_TXT]}.lock'
lock = FileLock(lock_file, timeout=10)
with lock:
credentials = getOauth2TxtStorageCredentials()
if credentials is None:
return
simplehttp = transport.create_http()
params = {'token': credentials.refresh_token}
revoke_uri = f'https://accounts.google.com/o/oauth2/revoke?{urlencode(params)}'
sys.stderr.write('This OAuth token will self-destruct in 3...')
sys.stderr.flush()
time.sleep(1)
sys.stderr.write('2...')
sys.stderr.flush()
time.sleep(1)
sys.stderr.write('1...')
sys.stderr.flush()
time.sleep(1)
sys.stderr.write('boom!\n')
sys.stderr.flush()
simplehttp.request(revoke_uri, 'GET')
os.remove(GC_Values[GC_OAUTH2_TXT])
if not GM_Globals[GM_WINDOWS]:
try:
os.remove(lock_file)
except IOError:
pass
def writeCredentials(creds):
creds_data = {
'token': creds.token,
'refresh_token': creds.refresh_token,
'token_uri': creds.token_uri,
'client_id': creds.client_id,
'client_secret': creds.client_secret,
'id_token': creds.id_token,
'token_expiry': creds.expiry.strftime('%Y-%m-%dT%H:%M:%SZ'),
#'scopes': sorted(creds.scopes), # Google auth doesn't currently give us scopes back on refresh
}
expected_iss = ['https://accounts.google.com', 'accounts.google.com']
if _getValueFromOAuth('iss', creds) not in expected_iss:
controlflow.system_error_exit(13, f'Wrong OAuth 2.0 credentials issuer. Got {_getValueFromOAuth("iss", creds)} expected one of {", ".join(expected_iss)}')
request = transport.create_request()
creds_data['decoded_id_token'] = google.oauth2.id_token.verify_oauth2_token(creds.id_token, request)
data = json.dumps(creds_data, indent=2, sort_keys=True)
fileutils.write_file(GC_Values[GC_OAUTH2_TXT], data)
credentials = getOauth2TxtStorageCredentials()
if credentials is None:
return
sys.stderr.write('This OAuth token will self-destruct in 3...')
sys.stderr.flush()
time.sleep(1)
sys.stderr.write('2...')
sys.stderr.flush()
time.sleep(1)
sys.stderr.write('1...')
sys.stderr.flush()
time.sleep(1)
sys.stderr.write('boom!\n')
sys.stderr.flush()
credentials.revoke()
credentials.delete()
def doRequestOAuth(login_hint=None):
credentials = getOauth2TxtStorageCredentials()
if credentials is None or not credentials.valid:
scopes = getScopesFromUser()
if scopes is None:
controlflow.system_error_exit(0, '')
client_id, client_secret = getOAuthClientIDAndSecret()
login_hint = _getValidateLoginHint(login_hint)
# Needs to be set so oauthlib doesn't puke when Google changes our scopes
os.environ['OAUTHLIB_RELAX_TOKEN_SCOPE'] = 'true'
creds = _run_oauth_flow(client_id, client_secret, scopes, 'offline', login_hint)
writeCredentials(creds)
else:
print(f'It looks like you\'ve already authorized GAM. Refusing to overwrite existing file:\n\n{GC_Values[GC_OAUTH2_TXT]}')
missing_client_secrets_message = ('To use GAM you need to create an API '
'project. Please run:\n\ngam create project')
client_secrets_file = GC_Values[GC_CLIENT_SECRETS_JSON]
invalid_client_secrets_format_message = ('The format of your client secrets '
'file:\n\n%s\n\nis incorrect. '
'Please recreate the file.' %
client_secrets_file)
stored_creds = getOauth2TxtStorageCredentials()
if stored_creds and stored_creds.valid:
print('It looks like you\'ve already authorized GAM. Refusing to overwrite existing file:\n\n%s' % stored_creds.filename)
return
def getOAuthClientIDAndSecret():
"""Retrieves the OAuth client ID and client secret from JSON."""
MISSING_CLIENT_SECRETS_MESSAGE = '''To use GAM you need to create an API project. Please run:
gam create project
'''
filename = GC_Values[GC_CLIENT_SECRETS_JSON]
cs_data = fileutils.read_file(filename, continue_on_error=True, display_errors=True)
if not cs_data:
controlflow.system_error_exit(14, MISSING_CLIENT_SECRETS_MESSAGE)
scopes = getScopesFromUser()
if scopes is None:
# There were no scopes selected. Exit cleanly.
controlflow.system_error_exit(0, '')
login_hint = _getValidateLoginHint(login_hint)
# Needs to be set so oauthlib doesn't puke when Google changes our scopes
os.environ['OAUTHLIB_RELAX_TOKEN_SCOPE'] = 'true'
try:
cs_json = json.loads(cs_data)
client_id = cs_json['installed']['client_id']
# chop off .apps.googleusercontent.com suffix as it's not needed
# and we need to keep things short for the Auth URL.
client_id = re.sub(r'\.apps\.googleusercontent\.com$', '', client_id)
client_secret = cs_json['installed']['client_secret']
except (ValueError, IndexError, KeyError):
controlflow.system_error_exit(3, f'the format of your client secrets file:\n\n{filename}\n\n'
'is incorrect. Please recreate the file.')
return (client_id, client_secret)
creds = auth.oauth.Credentials.from_client_secrets_file(
client_secrets_file=client_secrets_file,
scopes=scopes,
access_type='offline',
login_hint=login_hint,
credentials_file=GC_Values[GC_OAUTH2_TXT],
use_console_flow=not GC_Values[GC_OAUTH_BROWSER])
creds.write()
except auth.oauth.InvalidClientSecretsFileError:
controlflow.system_error_exit(14, missing_client_secrets_message)
except auth.oauth.InvalidClientSecretsFileFormatError:
controlflow.system_error_exit(3, invalid_client_secrets_format_message)
OAUTH2_SCOPES = [
{'name': 'Classroom API - counts as 5 scopes',

4
src/var.py
View File

@@ -936,15 +936,11 @@ CLEAR_NONE_ARGUMENT = ['clear', 'none',]
#
MESSAGE_API_ACCESS_CONFIG = 'API access is configured in your Control Panel under: Security-Show more-Advanced settings-Manage API client access'
MESSAGE_API_ACCESS_DENIED = 'API access Denied.\n\nPlease make sure the Client ID: {0} is authorized for the API Scope(s): {1}'
MESSAGE_CONSOLE_AUTHORIZATION_PROMPT = '\nGo to the following link in your browser:\n\n\t{url}\n'
MESSAGE_CONSOLE_AUTHORIZATION_CODE = 'Enter verification code: '
MESSAGE_GAM_EXITING_FOR_UPDATE = 'GAM is now exiting so that you can overwrite this old version with the latest release'
MESSAGE_GAM_OUT_OF_MEMORY = 'GAM has run out of memory. If this is a large G Suite instance, you should use a 64-bit version of GAM on Windows or a 64-bit version of Python on other systems.'
MESSAGE_HEADER_NOT_FOUND_IN_CSV_HEADERS = 'Header "{0}" not found in CSV headers of "{1}".'
MESSAGE_HIT_CONTROL_C_TO_UPDATE = '\n\nHit CTRL+C to visit the GAM website and download the latest release or wait 15 seconds continue with this boring old version. GAM won\'t bother you with this announcement for 1 week or you can create a file named noupdatecheck.txt in the same location as gam.py or gam.exe and GAM won\'t ever check for updates.'
MESSAGE_INVALID_JSON = 'The file {0} has an invalid format.'
MESSAGE_LOCAL_SERVER_AUTHORIZATION_PROMPT = '\nYour browser has been opened to visit:\n\n\t{url}\n\nIf your browser is on a different machine then press CTRL+C and create a file called nobrowser.txt in the same folder as GAM.\n'
MESSAGE_LOCAL_SERVER_SUCCESS = 'The authentication flow has completed. You may close this browser window and return to GAM.'
MESSAGE_NO_DISCOVERY_INFORMATION = 'No online discovery doc and {0} does not exist locally'
MESSAGE_NO_TRANSFER_LACK_OF_DISK_SPACE = 'Cowardly refusing to perform migration due to lack of target drive space. Source size: {0}mb Target Free: {1}mb'
MESSAGE_RESULTS_TOO_LARGE_FOR_GOOGLE_SPREADSHEET = 'Results are too large for Google Spreadsheets. Uploading as a regular CSV file.'