auth: Remove AuthomaticBase class.

- Code will be moved to the EE version.
This commit is contained in:
Johannes Bornhold 2016-05-24 15:41:09 +02:00
parent 26d1f8ea5c
commit ce2aa59dea

View file

@ -477,121 +477,6 @@ class RhodeCodeExternalAuthPlugin(RhodeCodeAuthPluginBase):
return auth
class AuthomaticBase(RhodeCodeExternalAuthPlugin):
# TODO: Think about how to create and store this secret string.
# We need the secret for the authomatic library. It needs to be the same
# across requests.
def _get_authomatic_secret(self, length=40):
secret = self.get_setting_by_name('secret')
if secret is None or secret == 'None' or secret == '':
from Crypto import Random, Hash
secret_bytes = Random.new().read(length)
secret_hash = Hash.SHA256.new()
secret_hash.update(secret_bytes)
secret = secret_hash.hexdigest()
self.create_or_update_setting('secret', secret)
Session.commit()
secret = self.get_setting_by_name('secret')
return secret
def get_authomatic(self):
scope = []
if self.name == 'bitbucket':
provider_class = oauth1.Bitbucket
scope = ['account', 'email', 'repository', 'issue', 'issue:write']
elif self.name == 'github':
provider_class = oauth2.GitHub
scope = ['repo', 'public_repo', 'user:email']
elif self.name == 'google':
provider_class = oauth2.Google
scope = ['profile', 'email']
elif self.name == 'twitter':
provider_class = oauth1.Twitter
authomatic_conf = {
self.name: {
'class_': provider_class,
'consumer_key': self.get_setting_by_name('consumer_key'),
'consumer_secret': self.get_setting_by_name('consumer_secret'),
'scope': scope,
'access_headers': {'User-Agent': 'TestAppAgent'},
}
}
secret = self._get_authomatic_secret()
return Authomatic(config=authomatic_conf,
secret=secret)
def get_provider_result(self, request):
"""
Provides `authomatic.core.LoginResult` for provider and request
:param provider_name:
:param request:
:param config:
:return:
"""
response = Response()
adapter = WebObAdapter(request, response)
authomatic_inst = self.get_authomatic()
return authomatic_inst.login(adapter, self.name), response
def handle_social_data(self, session, user_id, social_data):
"""
Updates user tokens in database whenever necessary
:param request:
:param user:
:param social_data:
:return:
"""
if not self.is_active():
h.flash(_('This provider is currently disabled'),
category='warning')
return False
social_data = social_data
update_identity = False
existing_row = ExternalIdentity.by_external_id_and_provider(
social_data['user']['id'],
social_data['credentials.provider']
)
if existing_row:
Session().delete(existing_row)
update_identity = True
if not existing_row or update_identity:
if not update_identity:
h.flash(_('Your external identity is now '
'connected with your account'), category='success')
if not social_data['user']['id']:
h.flash(_('No external user id found? Perhaps permissions'
'for authentication are set incorrectly'),
category='error')
return False
ex_identity = ExternalIdentity()
ex_identity.external_id = social_data['user']['id']
ex_identity.external_username = social_data['user']['user_name']
ex_identity.provider_name = social_data['credentials.provider']
ex_identity.access_token = social_data['credentials.token']
ex_identity.token_secret = social_data['credentials.token_secret']
ex_identity.alt_token = social_data['credentials.refresh_token']
ex_identity.local_user_id = user_id
Session().add(ex_identity)
session.pop('rhodecode.social_auth', None)
return ex_identity
def callback_url(self):
try:
return url('social_auth', provider_name=self.name, qualified=True)
except TypeError:
pass
return ''
def loadplugin(plugin_id):
"""
Loads and returns an instantiated authentication plugin.