mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-08-24 08:16:51 -04:00
Merge pull request #40 from alphagov/form_views_refactor
Form views refactor
This commit is contained in:
@@ -14,6 +14,8 @@ def insert_user(user):
|
||||
db.session.commit()
|
||||
|
||||
|
||||
# TODO Would be better to have a generic get and update for user
|
||||
# something that replicates the sql functionality.
|
||||
def get_user_by_id(id):
|
||||
return User.query.filter_by(id=id).first()
|
||||
|
||||
|
||||
@@ -26,6 +26,12 @@ verify_code = '^\d{5}$'
|
||||
|
||||
|
||||
class RegisterUserForm(Form):
|
||||
|
||||
def __init__(self, existing_email_addresses, existing_mobile_numbers, *args, **kwargs):
|
||||
self.existing_emails = existing_email_addresses
|
||||
self.existing_mobiles = existing_mobile_numbers
|
||||
super(RegisterUserForm, self).__init__(*args, **kwargs)
|
||||
|
||||
name = StringField('Full name',
|
||||
validators=[DataRequired(message='Name can not be empty')])
|
||||
email_address = StringField('Email address', validators=[
|
||||
@@ -42,9 +48,21 @@ class RegisterUserForm(Form):
|
||||
Length(10, 255, message='Password must be at least 10 characters'),
|
||||
Blacklist(message='That password is blacklisted, too common')])
|
||||
|
||||
def validate_email_address(self, field):
|
||||
# Validate email address is unique.
|
||||
if field.data in self.existing_emails:
|
||||
raise ValidationError('Email address already exists')
|
||||
|
||||
def validate_mobile_number(self, field):
|
||||
# Validate mobile number is unique
|
||||
# Code to re-added later
|
||||
# if field.data in self.existing_mobiles:
|
||||
# raise ValidationError('Mobile number already exists')
|
||||
pass
|
||||
|
||||
|
||||
class TwoFactorForm(Form):
|
||||
sms_code = StringField('sms code', validators=[DataRequired(message='Please enter your code'),
|
||||
sms_code = StringField('sms code', validators=[DataRequired(message='Enter verification code'),
|
||||
Regexp(regex=verify_code, message='Code must be 5 digits')])
|
||||
|
||||
def validate_sms_code(self, a):
|
||||
@@ -76,9 +94,9 @@ class EmailNotReceivedForm(Form):
|
||||
|
||||
|
||||
class TextNotReceivedForm(Form):
|
||||
mobile_number = StringField('Mobile phone number',
|
||||
validators=[DataRequired(message='Please enter your mobile number'),
|
||||
Regexp(regex=mobile_number, message='Please enter a +44 mobile number')])
|
||||
mobile_number = StringField('Mobile phone number', validators=[
|
||||
DataRequired(message='Please enter your mobile number'),
|
||||
Regexp(regex=mobile_number, message='Please enter a +44 mobile number')])
|
||||
|
||||
|
||||
class AddServiceForm(Form):
|
||||
@@ -86,7 +104,8 @@ class AddServiceForm(Form):
|
||||
self.service_names = service_names
|
||||
super(AddServiceForm, self).__init__(*args, **kwargs)
|
||||
|
||||
service_name = StringField(validators=[DataRequired(message='Please enter your service name')])
|
||||
service_name = StringField(validators=[
|
||||
DataRequired(message='Please enter your service name')])
|
||||
|
||||
def validate_service_name(self, a):
|
||||
if self.service_name.data in self.service_names:
|
||||
@@ -95,6 +114,7 @@ class AddServiceForm(Form):
|
||||
|
||||
def validate_codes(field, code_type):
|
||||
codes = verify_codes_dao.get_codes(user_id=session['user_id'], code_type=code_type)
|
||||
# TODO need to remove this manual logging.
|
||||
print('validate_codes for user_id: {} are {}'.format(session['user_id'], codes))
|
||||
if not [code for code in codes if validate_code(field, code)]:
|
||||
raise ValidationError('Code does not match')
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from flask import render_template, redirect, jsonify, session
|
||||
from flask import (
|
||||
render_template, redirect, jsonify, session, url_for)
|
||||
|
||||
from app.main import main
|
||||
from app.main.dao import users_dao
|
||||
@@ -6,39 +7,28 @@ from app.main.forms import EmailNotReceivedForm, TextNotReceivedForm
|
||||
from app.main.views import send_sms_code, send_email_code
|
||||
|
||||
|
||||
@main.route("/email-not-received", methods=['GET'])
|
||||
def email_not_received():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
return render_template('views/email-not-received.html',
|
||||
form=EmailNotReceivedForm(email_address=user.email_address))
|
||||
|
||||
|
||||
@main.route('/email-not-received', methods=['POST'])
|
||||
@main.route('/email-not-received', methods=['GET', 'POST'])
|
||||
def check_and_resend_email_code():
|
||||
form = EmailNotReceivedForm()
|
||||
# TODO there needs to be a way to regenerate a session id
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
form = EmailNotReceivedForm(email_address=user.email_address)
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
users_dao.update_email_address(id=user.id, email_address=form.email_address.data)
|
||||
send_email_code(user_id=user.id, email=user.email_address)
|
||||
return redirect('/verify')
|
||||
return jsonify(form.errors), 400
|
||||
return redirect(url_for('.verify'))
|
||||
return render_template('views/email-not-received.html', form=form)
|
||||
|
||||
|
||||
@main.route("/text-not-received", methods=['GET'])
|
||||
def text_not_received():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
return render_template('views/text-not-received.html', form=TextNotReceivedForm(mobile_number=user.mobile_number))
|
||||
|
||||
|
||||
@main.route('/text-not-received', methods=['POST'])
|
||||
@main.route('/text-not-received', methods=['GET', 'POST'])
|
||||
def check_and_resend_text_code():
|
||||
form = TextNotReceivedForm()
|
||||
# TODO there needs to be a way to regenerate a session id
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
form = TextNotReceivedForm(mobile_number=user.mobile_number)
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
users_dao.update_mobile_number(id=user.id, mobile_number=form.mobile_number.data)
|
||||
send_sms_code(user_id=user.id, mobile_number=user.mobile_number)
|
||||
return redirect('/verify')
|
||||
return jsonify(form.errors), 400
|
||||
return redirect(url_for('.verify'))
|
||||
return render_template('views/text-not-received.html', form=form)
|
||||
|
||||
|
||||
@main.route('/verification-not-received', methods=['GET'])
|
||||
@@ -48,6 +38,7 @@ def verification_code_not_received():
|
||||
|
||||
@main.route('/send-new-code', methods=['GET'])
|
||||
def check_and_resend_verification_code():
|
||||
# TODO there needs to be a way to generate a new session id
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
send_sms_code(user.id, user.mobile_number)
|
||||
return redirect('/two-factor')
|
||||
return redirect(url_for('main.two_factor'))
|
||||
|
||||
@@ -12,14 +12,19 @@ from app.main.views import send_sms_code, send_email_code
|
||||
from app.models import User
|
||||
|
||||
|
||||
@main.route("/register", methods=['GET'])
|
||||
def render_register():
|
||||
return render_template('views/register.html', form=RegisterUserForm())
|
||||
|
||||
|
||||
@main.route('/register', methods=['POST'])
|
||||
# TODO how do we handle duplicate unverifed email addresses?
|
||||
# malicious or otherwise.
|
||||
@main.route('/register', methods=['GET', 'POST'])
|
||||
def process_register():
|
||||
form = RegisterUserForm()
|
||||
try:
|
||||
existing_emails, existing_mobiles = zip(
|
||||
*[(x.email_address, x.mobile_number) for x in
|
||||
users_dao.get_all_users()])
|
||||
except ValueError:
|
||||
# Value error is raised if the db is empty.
|
||||
existing_emails, existing_mobiles = [], []
|
||||
|
||||
form = RegisterUserForm(existing_emails, existing_mobiles)
|
||||
|
||||
if form.validate_on_submit():
|
||||
user = User(name=form.name.data,
|
||||
@@ -28,16 +33,16 @@ def process_register():
|
||||
password=form.password.data,
|
||||
created_at=datetime.now(),
|
||||
role_id=1)
|
||||
try:
|
||||
users_dao.insert_user(user)
|
||||
send_sms_code(user_id=user.id, mobile_number=form.mobile_number.data)
|
||||
send_email_code(user_id=user.id, email=form.email_address.data)
|
||||
session['expiry_date'] = str(datetime.now() + timedelta(hours=1))
|
||||
session['user_id'] = user.id
|
||||
except AdminApiClientException as e:
|
||||
return jsonify(admin_api_client_error=e.value)
|
||||
except SQLAlchemyError:
|
||||
return jsonify(database_error='encountered database error'), 400
|
||||
else:
|
||||
return jsonify(form.errors), 400
|
||||
return redirect('/verify')
|
||||
users_dao.insert_user(user)
|
||||
# TODO possibly there should be some exception handling
|
||||
# for sending sms and email codes.
|
||||
# How do we report to the user there is a problem with
|
||||
# sending codes apart from service unavailable?
|
||||
# at the moment i believe http 500 is fine.
|
||||
send_sms_code(user_id=user.id, mobile_number=form.mobile_number.data)
|
||||
send_email_code(user_id=user.id, email=form.email_address.data)
|
||||
session['expiry_date'] = str(datetime.now() + timedelta(hours=1))
|
||||
session['user_id'] = user.id
|
||||
return redirect('/verify')
|
||||
|
||||
return render_template('views/register.html', form=form)
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from flask import render_template, redirect, jsonify
|
||||
from flask import (
|
||||
render_template, redirect, jsonify, url_for)
|
||||
from flask import session
|
||||
|
||||
from app.main import main
|
||||
@@ -8,28 +9,23 @@ from app.main.forms import LoginForm
|
||||
from app.main.views import send_sms_code
|
||||
|
||||
|
||||
@main.route("/sign-in", methods=(['GET']))
|
||||
def render_sign_in():
|
||||
return render_template('views/signin.html', form=LoginForm())
|
||||
@main.route('/sign-in', methods=(['GET', 'POST']))
|
||||
def sign_in():
|
||||
try:
|
||||
form = LoginForm()
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_email(form.email_address.data)
|
||||
if user:
|
||||
if not user.is_locked() and user.is_active() and check_hash(form.password.data, user.password):
|
||||
send_sms_code(user.id, user.mobile_number)
|
||||
session['user_id'] = user.id
|
||||
return redirect(url_for('.two_factor'))
|
||||
else:
|
||||
users_dao.increment_failed_login_count(user.id)
|
||||
# Vague error message for login
|
||||
form.password.errors.append('Username or password is incorrect')
|
||||
|
||||
|
||||
@main.route('/sign-in', methods=(['POST']))
|
||||
def process_sign_in():
|
||||
form = LoginForm()
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_email(form.email_address.data)
|
||||
if user is None:
|
||||
return jsonify(authorization=False), 401
|
||||
if user.is_locked():
|
||||
return jsonify(locked_out=True), 401
|
||||
if not user.is_active():
|
||||
return jsonify(active_user=False), 401
|
||||
if check_hash(form.password.data, user.password):
|
||||
send_sms_code(user.id, user.mobile_number)
|
||||
session['user_id'] = user.id
|
||||
else:
|
||||
users_dao.increment_failed_login_count(user.id)
|
||||
return jsonify(authorization=False), 401
|
||||
else:
|
||||
return jsonify(form.errors), 400
|
||||
return redirect('/two-factor')
|
||||
return render_template('views/signin.html', form=form)
|
||||
except:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
@@ -3,6 +3,7 @@ from flask_login import login_required
|
||||
|
||||
from app.main import main
|
||||
|
||||
# TODO move this to the templates directory
|
||||
message_templates = [
|
||||
{
|
||||
'name': 'Reminder',
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import traceback
|
||||
|
||||
from flask import render_template, redirect, jsonify, session
|
||||
from flask import (
|
||||
render_template, redirect, jsonify, session, url_for)
|
||||
|
||||
from flask_login import login_user
|
||||
|
||||
from app.main import main
|
||||
@@ -8,22 +9,14 @@ from app.main.dao import users_dao, verify_codes_dao
|
||||
from app.main.forms import TwoFactorForm
|
||||
|
||||
|
||||
@main.route("/two-factor", methods=['GET'])
|
||||
def render_two_factor():
|
||||
return render_template('views/two-factor.html', form=TwoFactorForm())
|
||||
@main.route('/two-factor', methods=['GET', 'POST'])
|
||||
def two_factor():
|
||||
form = TwoFactorForm()
|
||||
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='sms')
|
||||
login_user(user)
|
||||
return redirect(url_for('.dashboard'))
|
||||
|
||||
@main.route('/two-factor', methods=['POST'])
|
||||
def process_two_factor():
|
||||
try:
|
||||
form = TwoFactorForm()
|
||||
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='sms')
|
||||
login_user(user)
|
||||
return redirect('/dashboard')
|
||||
else:
|
||||
return jsonify(form.errors), 400
|
||||
except:
|
||||
traceback.print_exc()
|
||||
return render_template('views/two-factor.html', form=form)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import traceback
|
||||
from flask import (
|
||||
render_template, redirect, jsonify, session, url_for)
|
||||
|
||||
from flask import render_template, redirect, jsonify, session
|
||||
from flask_login import login_user
|
||||
|
||||
from app.main import main
|
||||
@@ -8,23 +8,16 @@ from app.main.dao import users_dao, verify_codes_dao
|
||||
from app.main.forms import VerifyForm
|
||||
|
||||
|
||||
@main.route('/verify', methods=['GET'])
|
||||
def render_verify():
|
||||
return render_template('views/verify.html', form=VerifyForm())
|
||||
|
||||
|
||||
@main.route('/verify', methods=['POST'])
|
||||
def process_verify():
|
||||
try:
|
||||
form = VerifyForm()
|
||||
if form.validate_on_submit():
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='email')
|
||||
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='sms')
|
||||
users_dao.activate_user(user.id)
|
||||
login_user(user)
|
||||
return redirect('/add-service')
|
||||
else:
|
||||
return jsonify(form.errors), 400
|
||||
except:
|
||||
traceback.print_exc()
|
||||
@main.route('/verify', methods=['GET', 'POST'])
|
||||
def verify():
|
||||
# TODO there needs to be a way to regenerate a session id
|
||||
# or handle gracefully.
|
||||
user = users_dao.get_user_by_id(session['user_id'])
|
||||
form = VerifyForm()
|
||||
if form.validate_on_submit():
|
||||
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='email')
|
||||
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='sms')
|
||||
users_dao.activate_user(user.id)
|
||||
login_user(user)
|
||||
return redirect(url_for('.add_service'))
|
||||
return render_template('views/verify.html', form=form)
|
||||
|
||||
Reference in New Issue
Block a user