quay/endpoints/callbacks.py

import logging

from flask import request, redirect, url_for, Blueprint
from flask.ext.login import current_user

from endpoints.common import render_page_template, common_login, route_show_if
from app import app, analytics, get_app_url
from data import model
from util.names import parse_repository_name
from util.validation import generate_valid_usernames
from util.http import abort
from auth.permissions import AdministerRepositoryPermission
from auth.auth import require_session_login
from peewee import IntegrityError

import features

logger = logging.getLogger(__name__)

client = app.config['HTTPCLIENT']


callback = Blueprint('callback', __name__)

def render_ologin_error(service_name,
                        error_message='Could not load user data. The token may have expired.'):
  return render_page_template('ologinerror.html', service_name=service_name,
                              error_message=error_message,
                              service_url=get_app_url(),
                              user_creation=features.USER_CREATION)

def exchange_code_for_token(code, service_name='GITHUB', for_login=True, form_encode=False,
                            redirect_suffix=''):
  code = request.args.get('code')
  id_config = service_name + '_LOGIN_CLIENT_ID' if for_login else service_name + '_CLIENT_ID'
  secret_config = service_name + '_LOGIN_CLIENT_SECRET' if for_login else service_name + '_CLIENT_SECRET'

  payload = {
    'client_id': app.config[id_config],
    'client_secret': app.config[secret_config],
    'code': code,
    'grant_type': 'authorization_code',
    'redirect_uri': '%s://%s/oauth2/%s/callback%s' % (app.config['PREFERRED_URL_SCHEME'],
                                                      app.config['SERVER_HOSTNAME'],
                                                      service_name.lower(),
                                                      redirect_suffix)
  }

  headers = {
    'Accept': 'application/json'
  }

  if form_encode:
    get_access_token = client.post(app.config[service_name + '_TOKEN_URL'],
                                   data=payload, headers=headers)
  else:
    get_access_token = client.post(app.config[service_name + '_TOKEN_URL'],
                                   params=payload, headers=headers)

  json_data = get_access_token.json()
  if not json_data:
    return ''

  token = json_data.get('access_token', '')
  return token


def get_github_user(token):
  token_param = {
    'access_token': token,
  }
  get_user = client.get(app.config['GITHUB_USER_URL'], params=token_param)

  return get_user.json()


def get_google_user(token):
  token_param = {
    'access_token': token,
    'alt': 'json',
  }

  get_user = client.get(app.config['GOOGLE_USER_URL'], params=token_param)
  return get_user.json()

def conduct_oauth_login(service_name, user_id, username, email, metadata={}):
  to_login = model.verify_federated_login(service_name.lower(), user_id)
  if not to_login:
    # See if we can create a new user.
    if not features.USER_CREATION:
      error_message = 'User creation is disabled. Please contact your administrator'
      return render_ologin_error(service_name, error_message)

    # Try to create the user
    try:
      valid = next(generate_valid_usernames(username))
      to_login = model.create_federated_user(valid, email, service_name.lower(),
                                             user_id, set_password_notification=True,
                                             metadata=metadata)

      # Success, tell analytics
      analytics.track(to_login.username, 'register', {'service': service_name.lower()})

      state = request.args.get('state', None)
      if state:
        logger.debug('Aliasing with state: %s' % state)
        analytics.alias(to_login.username, state)

    except model.InvalidEmailAddressException as ieex:
      message = "The e-mail address %s is already associated " % (email, )
      message = message + "with an existing %s account." % (app.config['REGISTRY_TITLE_SHORT'], )
      message = message + "\nPlease log in with your username and password and "
      message = message + "associate your %s account to use it in the future." % (service_name, )

      return render_ologin_error(service_name, message)

    except model.DataModelException as ex:
      return render_ologin_error(service_name, ex.message)

  if common_login(to_login):
    return redirect(url_for('web.index'))

  return render_ologin_error(service_name)

def get_google_username(user_data):
  username = user_data['email']
  at = username.find('@')
  if at > 0:
    username = username[0:at]

  return username


@callback.route('/google/callback', methods=['GET'])
@route_show_if(features.GOOGLE_LOGIN)
def google_oauth_callback():
  error = request.args.get('error', None)
  if error:
    return render_ologin_error('Google', error)

  token = exchange_code_for_token(request.args.get('code'), service_name='GOOGLE', form_encode=True)
  user_data = get_google_user(token)
  if not user_data or not user_data.get('id', None) or not user_data.get('email', None):
    return render_ologin_error('Google')

  username = get_google_username(user_data)
  metadata = {
    'service_username': user_data['email']
  }

  return conduct_oauth_login('Google', user_data['id'], username, user_data['email'],
                             metadata=metadata)


@callback.route('/github/callback', methods=['GET'])
@route_show_if(features.GITHUB_LOGIN)
def github_oauth_callback():
  error = request.args.get('error', None)
  if error:
    return render_ologin_error('GitHub', error)

  token = exchange_code_for_token(request.args.get('code'), service_name='GITHUB')
  user_data = get_github_user(token)
  if not user_data or not 'login' in user_data:
    return render_ologin_error('GitHub')

  username = user_data['login']
  github_id = user_data['id']

  v3_media_type = {
    'Accept': 'application/vnd.github.v3'
  }

  token_param = {
    'access_token': token,
  }
  get_email = client.get(app.config['GITHUB_USER_EMAILS'], params=token_param,
                         headers=v3_media_type)

  # We will accept any email, but we prefer the primary
  found_email = None
  for user_email in get_email.json():
    found_email = user_email['email']
    if user_email['primary']:
      break

  metadata = {
    'service_username': username
  }

  return conduct_oauth_login('github', github_id, username, found_email, metadata=metadata)


@callback.route('/google/callback/attach', methods=['GET'])
@route_show_if(features.GOOGLE_LOGIN)
@require_session_login
def google_oauth_attach():
  token = exchange_code_for_token(request.args.get('code'), service_name='GOOGLE',
                                  redirect_suffix='/attach', form_encode=True)

  user_data = get_google_user(token)
  if not user_data or not user_data.get('id', None):
    return render_ologin_error('Google')

  google_id = user_data['id']
  user_obj = current_user.db_user()

  username = get_google_username(user_data)
  metadata = {
    'service_username': user_data['email']
  }

  try:
    model.attach_federated_login(user_obj, 'google', google_id, metadata=metadata)
  except IntegrityError:
    err = 'Google account %s is already attached to a %s account' % (
      username, app.config['REGISTRY_TITLE_SHORT'])
    return render_ologin_error('Google', err)

  return redirect(url_for('web.user'))


@callback.route('/github/callback/attach', methods=['GET'])
@route_show_if(features.GITHUB_LOGIN)
@require_session_login
def github_oauth_attach():
  token = exchange_code_for_token(request.args.get('code'), service_name='GITHUB')
  user_data = get_github_user(token)
  if not user_data:
    return render_ologin_error('GitHub')

  github_id = user_data['id']
  user_obj = current_user.db_user()

  username = user_data['login']
  metadata = {
    'service_username': username
  }

  try:
    model.attach_federated_login(user_obj, 'github', github_id, metadata=metadata)
  except IntegrityError:
    err = 'Github account %s is already attached to a %s account' % (
      username, app.config['REGISTRY_TITLE_SHORT'])

    return render_ologin_error('GitHub', err)

  return redirect(url_for('web.user'))


@callback.route('/github/callback/trigger/<path:repository>', methods=['GET'])
@route_show_if(features.GITHUB_BUILD)
@require_session_login
@parse_repository_name
def attach_github_build_trigger(namespace, repository):
  permission = AdministerRepositoryPermission(namespace, repository)
  if permission.can():
    token = exchange_code_for_token(request.args.get('code'), service_name='GITHUB',
                                    for_login=False)
    repo = model.get_repository(namespace, repository)
    if not repo:
      msg = 'Invalid repository: %s/%s' % (namespace, repository)
      abort(404, message=msg)

    trigger = model.create_build_trigger(repo, 'github', token, current_user.db_user())
    admin_path = '%s/%s/%s' % (namespace, repository, 'admin')
    full_url = '%s%s%s' % (url_for('web.repository', path=admin_path), '?tab=trigger&new_trigger=',
                           trigger.uuid)
    logger.debug('Redirecting to full url: %s' % full_url)
    return redirect(full_url)

  abort(403)