lenticular_cloud2/lenticular_cloud/views/auth.py

181 lines
6.2 KiB
Python
Raw Normal View History

2020-05-09 18:00:07 +00:00
from urllib.parse import urlencode, parse_qs
import flask
from flask import Blueprint, redirect
from flask import current_app, session
from flask.templating import render_template
from flask_babel import gettext
2020-05-09 18:00:07 +00:00
from flask import request, url_for, jsonify
2020-05-30 21:33:59 +00:00
from flask_login import login_required, login_user, logout_user, current_user
2020-05-09 18:00:07 +00:00
import logging
from urllib.parse import urlparse
from base64 import b64decode, b64encode
2020-05-21 11:20:27 +00:00
import http
2020-05-27 19:16:14 +00:00
import crypt
import ory_hydra_client
from datetime import datetime
2020-06-21 09:52:37 +00:00
import logging
2020-05-09 18:00:07 +00:00
2020-05-27 19:16:14 +00:00
from ..model import db, User, SecurityUser, UserSignUp
from ..form.auth import ConsentForm, LoginForm, RegistrationForm
2020-05-09 18:00:07 +00:00
from ..auth_providers import AUTH_PROVIDER_LIST
2020-06-21 09:52:37 +00:00
logger = logging.getLogger(__name__)
2020-05-21 11:20:27 +00:00
auth_views = Blueprint('auth', __name__, url_prefix='/auth')
2020-05-09 18:00:07 +00:00
2020-05-21 11:20:27 +00:00
@auth_views.route('/consent', methods=['GET', 'POST'])
def consent():
"""Always grant consent."""
# DUMMPY ONLY
2020-05-09 18:00:07 +00:00
form = ConsentForm()
remember_for = 60*60*24*30 # remember for 7 days
try:
consent_request = current_app.hydra_api.get_consent_request(
request.args['consent_challenge'])
2020-06-21 09:52:37 +00:00
except ory_hydra_client.exceptions.ApiValueError:
2020-06-21 11:58:57 +00:00
logger.info('ory exception - could not fetch user data ApiValueError')
return redirect(url_for('frontend.index'))
except ory_hydra_client.exceptions.ApiException:
logger.exception('ory exception - could not fetch user data')
return redirect(url_for('frontend.index'))
2020-05-30 17:00:08 +00:00
2020-05-21 11:20:27 +00:00
requested_scope = consent_request.requested_scope
requested_audiences = consent_request.requested_access_token_audience
if form.validate_on_submit() or consent_request.skip:
2020-05-30 21:43:55 +00:00
user = User.query.get(consent_request.subject)
2020-05-30 21:33:59 +00:00
token_data = {
'name': str(user.username),
2020-05-30 21:43:55 +00:00
'preferred_username': str(user.username),
'email': str(user.email),
2020-05-30 21:33:59 +00:00
'email_verified': True,
'groups': [group.name for group in user.groups]
2020-05-30 21:33:59 +00:00
}
id_token_data = {}
if 'openid' in requested_scope:
id_token_data = token_data
resp = current_app.hydra_api.accept_consent_request(
consent_request.challenge, body={
'grant_scope': requested_scope,
'grant_access_token_audience': requested_audiences,
'remember': form.data['remember'],
'remember_for': remember_for,
2020-05-30 17:00:08 +00:00
'session': {
2020-05-30 21:33:59 +00:00
'access_token': token_data,
'id_token': id_token_data
}
})
return redirect(resp.redirect_to)
return render_template(
'auth/consent.html.j2',
form=form,
client=consent_request.client,
requested_scope=requested_scope,
requested_audiences=requested_audiences)
2020-05-09 18:00:07 +00:00
@auth_views.route('/login', methods=['GET', 'POST'])
def login():
2020-05-21 11:20:27 +00:00
login_challenge = request.args.get('login_challenge')
try:
login_request = current_app.hydra_api.get_login_request(login_challenge)
2022-02-06 22:57:01 +00:00
except ory_hydra_client.exceptions.ApiException as e:
logger.exception("could not fetch login request")
return redirect(url_for('frontend.index'))
2020-05-21 11:20:27 +00:00
if login_request.skip:
resp = current_app.hydra_api.accept_login_request(
login_challenge,
body={'subject': login_request.subject})
return redirect(resp.redirect_to)
2020-05-09 18:00:07 +00:00
form = LoginForm()
if form.validate_on_submit():
user = User.query_().by_username(form.data['name'])
2020-05-13 18:08:28 +00:00
if user:
session['username'] = str(user.username)
else:
session['user'] = None
2020-05-09 18:00:07 +00:00
session['auth_providers'] = []
return redirect(
url_for('auth.login_auth', login_challenge=login_challenge))
return render_template('auth/login.html.j2', form=form)
2020-05-09 18:00:07 +00:00
@auth_views.route('/login/auth', methods=['GET', 'POST'])
def login_auth():
2020-05-21 11:20:27 +00:00
login_challenge = request.args.get('login_challenge')
try:
login_request = current_app.hydra_api.get_login_request(login_challenge)
except ory_hydra_client.exceptions.ApiValueError:
return redirect(url_for('frontend.index'))
2020-05-09 18:00:07 +00:00
if 'username' not in session:
return redirect(url_for('auth.login'))
auth_forms = {}
user = User.query_().by_username(session['username'])
2020-05-09 18:00:07 +00:00
for auth_provider in AUTH_PROVIDER_LIST:
form = auth_provider.get_form()
if auth_provider.get_name() not in session['auth_providers'] and\
auth_provider.check_auth(user, form):
session['auth_providers'].append(auth_provider.get_name())
if auth_provider.get_name() not in session['auth_providers']:
auth_forms[auth_provider.get_name()]=form
2020-05-09 18:00:07 +00:00
if len(session['auth_providers']) >= 2:
2020-05-21 11:20:27 +00:00
remember_me = True
# if db_user is None:
# db_user = User(username=session['username'])
# db.session.add(db_user)
# db.session.commit()
2020-05-21 11:20:27 +00:00
subject = user.id
user.last_login = datetime.now()
db.session.commit()
2020-05-21 11:20:27 +00:00
resp = current_app.hydra_api.accept_login_request(
login_challenge, body={
'subject': subject,
2020-05-30 17:00:08 +00:00
'remember': remember_me,
})
2020-05-21 11:20:27 +00:00
return redirect(resp.redirect_to)
return render_template('auth/login_auth.html.j2', forms=auth_forms)
2020-05-09 18:00:07 +00:00
@auth_views.route("/logout")
def logout():
2020-05-21 11:20:27 +00:00
logout_challenge = request.args.get('logout_challenge')
logout_request = current_app.hydra_api.get_logout_request(logout_challenge)
resp = current_app.hydra_api.accept_logout_request(logout_challenge)
# TODO confirm
2020-05-21 11:20:27 +00:00
return redirect(resp.redirect_to)
2020-05-09 18:00:07 +00:00
2020-05-27 19:16:14 +00:00
@auth_views.route("/sign_up", methods=["GET"])
2020-05-27 19:16:14 +00:00
def sign_up():
form = RegistrationForm()
return render_template('auth/sign_up.html.j2', form=form)
@auth_views.route("/sign_up", methods=["POST"])
def sign_up_submit():
2020-05-27 19:16:14 +00:00
form = RegistrationForm()
if form.validate_on_submit():
user = UserSignUp()
user.username = form.data['username']
user.password = crypt.crypt(form.data['password'])
user.alternative_email = form.data['alternative_email']
db.session.add(user)
db.session.commit()
return jsonify({})
return jsonify({
'status': 'error',
'errors': form.errors
})