Source code for plugins.service_account

from ferris import plugins, settings, ndb
import hashlib

plugins.register('service_account')


def get_config():
    config = settings.get('oauth2_service_account')
    if not config['private_key'] or not config['client_email'] or not config['domain']:
        raise RuntimeError("OAuth2 Service Account is not configured correctly")
    return config


from oauth2client.client import SignedJwtAssertionCredentials
from oauth2client.appengine import StorageByKeyName, CredentialsNDBProperty


[docs]def build_credentials(scope, user=None): """ Builds service account credentials using the configuration stored in settings and masquerading as the provided user. """ config = get_config() if not user: user = config['default_user'] if not isinstance(scope, (list, tuple)): scope = [scope] key = generate_storage_key(config['client_email'], scope, user) storage = StorageByKeyName(ServiceAccountStorage, key, 'credentials') creds = SignedJwtAssertionCredentials( service_account_name=config['client_email'], private_key=config['private_key'], scope=scope, prn=user) creds.set_store(storage) return creds
def credentials_to_token(credentials): """ Transforms an Oauth2 credentials object into an OAuth2Token object to be used with the legacy gdata API """ import httplib2 import gdata.gauth credentials.refresh(httplib2.Http()) token = gdata.gauth.OAuth2Token( client_id=credentials.client_id, client_secret=credentials.client_secret, scope=credentials.scope, user_agent='lolidk/wtfbbq/cloudsherpas', access_token=credentials.access_token, refresh_token=credentials.refresh_token) return token class ServiceAccountStorage(ndb.Model): """ Tracks access tokens in the database. The key is based on the scopes, user, and clientid """ credentials = CredentialsNDBProperty() @classmethod def _get_kind(cls): return '_ferris_OAuth2ServiceAccountStorage' def generate_storage_key(client_id, scopes, user): s = u"%s%s%s" % (client_id, sorted(scopes), user) hash = hashlib.sha1(s.encode()) return hash.hexdigest()