183 lines
4.8 KiB
Python
183 lines
4.8 KiB
Python
import logging
|
|
import requests
|
|
|
|
from flask import (abort, send_file, redirect, request, url_for,
|
|
render_template, make_response)
|
|
from flask.ext.login import login_user, UserMixin, login_required, logout_user
|
|
from flask.ext.principal import identity_changed, Identity, AnonymousIdentity
|
|
|
|
from data import model
|
|
from app import app, login_manager, mixpanel
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class _LoginWrappedDBUser(UserMixin):
|
|
def __init__(self, db_user):
|
|
self.db_user = db_user
|
|
|
|
def is_active(self):
|
|
return self.db_user.verified
|
|
|
|
def get_id(self):
|
|
return unicode(self.db_user.username)
|
|
|
|
|
|
@login_manager.user_loader
|
|
def load_user(username):
|
|
logger.debug('Loading user: %s' % username)
|
|
db_user = model.get_user(username)
|
|
if db_user:
|
|
return _LoginWrappedDBUser(db_user)
|
|
else:
|
|
return None
|
|
|
|
|
|
@app.route('/', methods=['GET'], defaults={'path': ''})
|
|
@app.route('/<path:path>') # Catch all
|
|
def index(path):
|
|
return send_file('templates/index.html')
|
|
|
|
|
|
@app.route('/status', methods=['GET'])
|
|
def status():
|
|
return make_response('Healthy')
|
|
|
|
|
|
@app.route('/tos', methods=['GET'])
|
|
def tos():
|
|
return send_file('templates/tos.html')
|
|
|
|
|
|
@app.route('/privacy', methods=['GET'])
|
|
def privacy():
|
|
return send_file('templates/privacy.html')
|
|
|
|
|
|
def common_login(db_user):
|
|
if login_user(_LoginWrappedDBUser(db_user)):
|
|
logger.debug('Successfully signed in as: %s' % db_user.username)
|
|
identity_changed.send(app,
|
|
identity=Identity(db_user.username, 'username'))
|
|
return True
|
|
else:
|
|
logger.debug('User could not be logged in, inactive?.')
|
|
return False
|
|
|
|
|
|
@app.route('/signin', methods=['GET'])
|
|
def render_signin_page():
|
|
return render_template('signin.html',
|
|
github_client_id=app.config['GITHUB_CLIENT_ID'])
|
|
|
|
|
|
@app.route('/signin', methods=['POST'])
|
|
def signin():
|
|
username = request.form['username']
|
|
password = request.form['password']
|
|
|
|
#TODO Allow email login
|
|
verified = model.verify_user(username, password)
|
|
if verified:
|
|
if common_login(verified):
|
|
return redirect(request.args.get('next') or url_for('index'))
|
|
else:
|
|
return render_template('signin.html',
|
|
needs_email_verification=True,
|
|
github_client_id=app.config['GITHUB_CLIENT_ID'])
|
|
|
|
else:
|
|
return render_template('signin.html',
|
|
username=username,
|
|
invalid_credentials=True,
|
|
github_client_id=app.config['GITHUB_CLIENT_ID'])
|
|
|
|
|
|
@app.route('/oauth2/github/callback', methods=['GET'])
|
|
def github_oauth_callback():
|
|
code = request.args.get('code')
|
|
payload = {
|
|
'client_id': app.config['GITHUB_CLIENT_ID'],
|
|
'client_secret': app.config['GITHUB_CLIENT_SECRET'],
|
|
'code': code,
|
|
}
|
|
headers = {
|
|
'Accept': 'application/json'
|
|
}
|
|
|
|
get_access_token = requests.post(app.config['GITHUB_TOKEN_URL'],
|
|
params=payload, headers=headers)
|
|
|
|
token = get_access_token.json()['access_token']
|
|
|
|
token_param = {
|
|
'access_token': token,
|
|
}
|
|
get_user = requests.get(app.config['GITHUB_USER_URL'], params=token_param)
|
|
|
|
user_data = get_user.json()
|
|
username = user_data['login']
|
|
github_id = user_data['id']
|
|
|
|
v3_media_type = {
|
|
'Accept': 'application/vnd.github.v3'
|
|
}
|
|
get_email = requests.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
|
|
|
|
to_login = model.verify_federated_login('github', github_id)
|
|
if not to_login:
|
|
# try to create the user
|
|
|
|
try:
|
|
to_login = model.create_federated_user(username, found_email, 'github',
|
|
github_id)
|
|
except model.DataModelException, ex:
|
|
return render_template('githuberror.html', error_message=ex.message)
|
|
|
|
if common_login(to_login):
|
|
# Success
|
|
mixpanel.track(to_login.username, 'register', {'service': 'github'})
|
|
|
|
state = request.args.get('state', None)
|
|
if state:
|
|
logger.debug('Aliasing with state: %s' % state)
|
|
mixpanel.alias(to_login.username, state)
|
|
|
|
return redirect(url_for('index'))
|
|
|
|
# TODO something bad happened, we need to tell the user somehow
|
|
return render_template('githuberror.html')
|
|
|
|
|
|
@app.route('/confirm', methods=['GET'])
|
|
def confirm_email():
|
|
code = request.values['code']
|
|
user = model.confirm_user_email(code)
|
|
|
|
common_login(user)
|
|
|
|
return redirect(url_for('index'))
|
|
|
|
|
|
@app.route('/reset', methods=['GET'])
|
|
def password_reset():
|
|
pass
|
|
|
|
|
|
@app.route("/signout")
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
|
|
identity_changed.send(app, identity=AnonymousIdentity())
|
|
|
|
return redirect(url_for('index'))
|