darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-02-05 02:42:26 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #117 from alphagov/password_validation_for_service_changes

Browse Source 
Password validation for service changes
This commit is contained in:
Adam Shimali
2016-01-28 11:37:51 +00:00
parent 816aecd8fd e322a5f1c3
commit 5dccba56c9
67 changed files with 1326 additions and 2045 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/__init__.py
View File

@@ -16,7 +16,6 @@ import app.proxy_fix
from config import configs
from utils import logging
db = SQLAlchemy()
login_manager = LoginManager()
csrf = CsrfProtect()
@@ -31,7 +30,6 @@ def create_app(config_name, config_overrides=None):
application.config['NOTIFY_ADMIN_ENVIRONMENT'] = config_name
application.config.from_object(configs[config_name])
init_app(application, config_overrides)
db.init_app(application)
logging.init_app(application)
init_csrf(application)

11
app/main/dao/roles_dao.py
View File

@@ -1,11 +0,0 @@
from app import db
from app.models import Roles
def insert_role(role):
db.session.add(role)
db.session.commit()
def get_role_by_id(id):
return Roles.query.filter_by(id=id).first()

60
app/main/dao/users_dao.py
View File

@@ -2,8 +2,7 @@ from datetime import datetime
from sqlalchemy.orm import load_only
from app import db, login_manager
from app.models import User
from app import login_manager
from app.main.encryption import hashpw
from app import user_api_client
@@ -14,12 +13,6 @@ def load_user(user_id):
return get_user_by_id(user_id)
def insert_user(user):
user.password = hashpw(user.password)
db.session.add(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):
@@ -34,8 +27,18 @@ def get_user_by_email(email_address):
return user_api_client.get_user_by_email(email_address)
def verify_password(user, password):
return user_api_client.verify_password(user, password)
def verify_password(user_id, password):
return user_api_client.verify_password(user_id, password)
def update_user(user):
return user_api_client.update_user(user)
def increment_failed_login_count(id):
user = get_user_by_id(id)
user.failed_login_count += 1
return user_api_client.update_user(user)
def activate_user(user):
@@ -43,29 +46,20 @@ def activate_user(user):
return user_api_client.update_user(user)
def update_email_address(id, email_address):
user = get_user_by_id(id)
user.email_address = email_address
return user_api_client.update_user(user)
def is_email_unique(email_address):
if user_api_client.get_user_by_email(email_address):
return False
return True
def update_mobile_number(id, mobile_number):
user = get_user_by_id(id)
user.mobile_number = mobile_number
db.session.add(user)
db.session.commit()
def update_password(user, password):
user.password = hashpw(password)
user.password_changed_at = datetime.now()
user.state = 'active'
db.session.add(user)
db.session.commit()
def request_password_reset(email):
user = get_user_by_email(email)
def request_password_reset(user):
user.state = 'request_password_reset'
db.session.add(user)
db.session.commit()
user_api_client.update_user(user)
def send_verify_code(user_id, code_type, to=None):
return user_api_client.send_verify_code(user_id, code_type)
def check_verify_code(user_id, code, code_type):
return user_api_client.check_verify_code(user_id, code, code_type)

55
app/main/dao/verify_codes_dao.py
View File

@@ -1,55 +0,0 @@
from datetime import datetime, timedelta
from app import db
from app.main.encryption import hashpw
from app.models import VerifyCodes
def add_code(user_id, code, code_type):
code = VerifyCodes(user_id=user_id,
code=hashpw(code),
code_type=code_type,
expiry_datetime=datetime.now() + timedelta(hours=1))
db.session.add(code)
db.session.commit()
return code
def get_codes(user_id, code_type=None):
if not code_type:
return VerifyCodes.query.filter_by(user_id=user_id, code_used=False).all()
return VerifyCodes.query.filter_by(user_id=user_id, code_type=code_type, code_used=False).all()
def get_code_by_code(user_id, code, code_type):
return VerifyCodes.query.filter_by(user_id=user_id, code=hashpw(code), code_type=code_type).first()
def use_code(id):
verify_code = VerifyCodes.query.get(id)
verify_code.code_used = True
db.session.add(verify_code)
db.session.commit()
def use_code_for_user_and_type(user_id, code_type):
codes = VerifyCodes.query.filter_by(user_id=user_id, code_type=code_type, code_used=False).all()
for verify_code in codes:
verify_code.code_used = True
db.session.add(verify_code)
db.session.commit()
def get_code_by_id(id):
return VerifyCodes.query.get(id)
def add_code_with_expiry(user_id, code, code_type, expiry):
code = VerifyCodes(user_id=user_id,
code=hashpw(code),
code_type=code_type,
expiry_datetime=expiry)
db.session.add(code)
db.session.commit()

102
app/main/forms.py
View File

@@ -11,8 +11,7 @@ from wtforms import (
)
from wtforms.validators import DataRequired, Email, Length, Regexp
from app.main.validators import Blacklist, ValidateUserCodes, CsvFileValidator
from app.main.dao import verify_codes_dao
from app.main.validators import Blacklist, CsvFileValidator
from app.main.encryption import check_hash
@@ -84,16 +83,14 @@ def sms_code():
return StringField('Text message confirmation code',
validators=[DataRequired(message='Text message confirmation code can not be empty'),
Regexp(regex=verify_code,
message='Text message confirmation code must be 5 digits'),
ValidateUserCodes(code_type='sms')])
message='Text message confirmation code must be 5 digits')])
def email_code():
verify_code = '^\d{5}$'
return StringField("Email confirmation code",
validators=[DataRequired(message='Email confirmation code can not be empty'),
Regexp(regex=verify_code, message='Email confirmation code must be 5 digits'),
ValidateUserCodes(code_type='email')])
Regexp(regex=verify_code, message='Email confirmation code must be 5 digits')])
class LoginForm(Form):
@@ -108,8 +105,8 @@ class LoginForm(Form):
class RegisterUserForm(Form):
def __init__(self, existing_email_addresses, *args, **kwargs):
self.existing_emails = existing_email_addresses
def __init__(self, unique_email_func, *args, **kwargs):
self.unique_email_func = unique_email_func
super(RegisterUserForm, self).__init__(*args, **kwargs)
name = StringField('Full name',
@@ -120,36 +117,50 @@ class RegisterUserForm(Form):
def validate_email_address(self, field):
# Validate email address is unique.
if self.existing_emails(field.data):
if not self.unique_email_func(field.data):
raise ValidationError('Email address already exists')
class TwoFactorForm(Form):
def __init__(self, user_codes, *args, **kwargs):
def __init__(self, validate_code_func, *args, **kwargs):
'''
Keyword arguments:
user_codes -- List of user code objects which have the fields
(code_type, expiry_datetime, code)
validate_code_func -- Validates the code with the API.
'''
self.user_codes = user_codes
self.validate_code_func = validate_code_func
super(TwoFactorForm, self).__init__(*args, **kwargs)
sms_code = sms_code()
def validate_sms_code(self, field):
is_valid, reason = self.validate_code_func(field.data)
if not is_valid:
raise ValidationError(reason)
class VerifyForm(Form):
def __init__(self, user_codes, *args, **kwargs):
def __init__(self, validate_code_func, *args, **kwargs):
'''
Keyword arguments:
user_codes -- List of user code objects which have the fields
(code_type, expiry_datetime, code)
validate_code_func -- Validates the code with the API.
'''
self.user_codes = user_codes
self.validate_code_func = validate_code_func
super(VerifyForm, self).__init__(*args, **kwargs)
sms_code = sms_code()
email_code = email_code()
def _validate_code(self, cde, code_type):
is_valid, reason = self.validate_code_func(cde, code_type)
if not is_valid:
raise ValidationError(reason)
def validate_email_code(self, field):
self._validate_code(field.data, 'email')
def validate_sms_code(self, field):
self._validate_code(field.data, 'sms')
class EmailNotReceivedForm(Form):
email_address = email_address()
@@ -186,8 +197,17 @@ class ServiceNameForm(Form):
class ConfirmPasswordForm(Form):
def __init__(self, validate_password_func, *args, **kwargs):
self.validate_password_func = validate_password_func
super(ConfirmPasswordForm, self).__init__(*args, **kwargs)
password = PasswordField(u'Enter password')
def validate_password(self, field):
if not self.validate_password_func(field.data):
raise ValidationError('Invalid password')
class TemplateForm(Form):
name = StringField(
@@ -201,17 +221,35 @@ class TemplateForm(Form):
class ForgotPasswordForm(Form):
def __init__(self, user_email_exists_func, *args, **kwargs):
self._user_email_exists_func = user_email_exists_func
super(ForgotPasswordForm, self).__init__(*args, **kwargs)
email_address = email_address()
def validate_email_address(self, field):
if not self._user_email_exists_func(field.data):
raise ValidationError('The email is not registered on our system')
class NewPasswordForm(Form):
new_password = password()
class ChangePasswordForm(Form):
def __init__(self, validate_password_func, *args, **kwargs):
self.validate_password_func = validate_password_func
super(ChangePasswordForm, self).__init__(*args, **kwargs)
old_password = password('Current password')
new_password = password('New password')
def validate_old_password(self, field):
if not self.validate_password_func(field.data):
raise ValidationError('Invalid password')
class CsvUploadForm(Form):
file = FileField('File to upload', validators=[DataRequired(
@@ -223,20 +261,50 @@ class ChangeNameForm(Form):
class ChangeEmailForm(Form):
def __init__(self, validate_email_func, *args, **kwargs):
self.validate_email_func = validate_email_func
super(ChangeEmailForm, self).__init__(*args, **kwargs)
email_address = email_address()
def validate_email_address(self, field):
is_valid = self.validate_email_func(field.data)
if not is_valid:
raise ValidationError("The email address is already in use")
class ConfirmEmailForm(Form):
def __init__(self, validate_code_func, *args, **kwargs):
self.validate_code_func = validate_code_func
super(ConfirmEmailForm, self).__init__(*args, **kwargs)
email_code = email_code()
def validate_email_code(self, field):
is_valid, msg = self.validate_code_func(field.data)
if not is_valid:
raise ValidationError(msg)
class ChangeMobileNumberForm(Form):
mobile_number = mobile_number()
class ConfirmMobileNumberForm(Form):
def __init__(self, validate_code_func, *args, **kwargs):
self.validate_code_func = validate_code_func
super(ConfirmMobileNumberForm, self).__init__(*args, **kwargs)
sms_code = sms_code()
def validate_sms_code(self, field):
is_valid, msg = self.validate_code_func(field.data)
if not is_valid:
raise ValidationError(msg)
class CreateKeyForm(Form):
def __init__(self, existing_key_names=[], *args, **kwargs):

26
app/main/validators.py
View File

@@ -14,32 +14,6 @@ class Blacklist(object):
raise ValidationError(self.message)
class ValidateUserCodes(object):
def __init__(self,
expiry_msg='Code has expired',
invalid_msg='Code does not match',
code_type=None):
self.expiry_msg = expiry_msg
self.invalid_msg = invalid_msg
self.code_type = code_type
def __call__(self, form, field):
# TODO would be great to do this sql query but
# not couple those parts of the code.
user_codes = getattr(form, 'user_codes', [])
valid_code = False
for code in user_codes:
if check_hash(field.data, code.code) and self.code_type == code.code_type:
if code.expiry_datetime <= datetime.now():
raise ValidationError(self.expiry_msg)
else:
# Valid code
valid_code = True
break
if not valid_code:
raise ValidationError(self.invalid_msg)
class CsvFileValidator(object):
def __init__(self, message='Not a csv file'):

21
app/main/views/code_not_received.py
View File

@@ -1,20 +1,22 @@
from flask import (
render_template, redirect, session, url_for)
from flask_login import current_user
from app.main import main
from app.main.dao import users_dao
from app.main.forms import EmailNotReceivedForm, TextNotReceivedForm
from app.notify_client.sender import send_sms_code, send_email_code
@main.route('/email-not-received', methods=['GET', 'POST'])
def check_and_resend_email_code():
# TODO there needs to be a way to regenerate a session id
user = users_dao.get_user_by_email(session['user_email'])
user = users_dao.get_user_by_email(session['user_details']['email'])
form = EmailNotReceivedForm(email_address=user.email_address)
if form.validate_on_submit():
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)
users_dao.send_verify_code(user.id, 'email', to=form.email_address.data)
user.email_address = form.email_address.data
users_dao.update_user(user)
return redirect(url_for('.verify'))
return render_template('views/email-not-received.html', form=form)
@@ -22,11 +24,12 @@ def check_and_resend_email_code():
@main.route('/text-not-received', methods=['GET', 'POST'])
def check_and_resend_text_code():
# TODO there needs to be a way to regenerate a session id
user = users_dao.get_user_by_email(session['user_email'])
user = users_dao.get_user_by_email(session['user_details']['email'])
form = TextNotReceivedForm(mobile_number=user.mobile_number)
if form.validate_on_submit():
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)
users_dao.send_verify_code(user.id, 'sms', to=form.mobile_number.data)
user.mobile_number = form.mobile_number.data
users_dao.update_user(user)
return redirect(url_for('.verify'))
return render_template('views/text-not-received.html', form=form)
@@ -39,6 +42,6 @@ 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_email(session['user_email'])
send_sms_code(user.id, user.mobile_number)
user = users_dao.get_user_by_email(session['user_details']['email'])
users_dao.send_verify_code(user.id, 'sms')
return redirect(url_for('main.two_factor'))

20
app/main/views/forgot_password.py
View File

@@ -7,13 +7,15 @@ from app.notify_client.sender import send_change_password_email
@main.route('/forgot-password', methods=['GET', 'POST'])
def forgot_password():
form = ForgotPasswordForm()
def _email_exists(email):
return not users_dao.is_email_unique(email)
form = ForgotPasswordForm(_email_exists)
if form.validate_on_submit():
if users_dao.get_user_by_email(form.email_address.data):
users_dao.request_password_reset(form.email_address.data)
send_change_password_email(form.email_address.data)
return render_template('views/password-reset-sent.html')
else:
current_app.logger.info('The email address used does not exist.')
else:
return render_template('views/forgot-password.html', form=form)
user = users_dao.get_user_by_email(form.email_address.data)
users_dao.request_password_reset(user)
send_change_password_email(form.email_address.data)
return render_template('views/password-reset-sent.html')
return render_template('views/forgot-password.html', form=form)

9
app/main/views/new_password.py
View File

@@ -3,7 +3,7 @@ from flask import (render_template, url_for, redirect, flash)
from app.main import main
from app.main.dao import users_dao
from app.main.forms import NewPasswordForm
from app.notify_client.sender import check_token, send_sms_code
from app.notify_client.sender import check_token
@main.route('/new-password/<path:token>', methods=['GET', 'POST'])
@@ -21,8 +21,11 @@ def new_password(token):
form = NewPasswordForm()
if form.validate_on_submit():
users_dao.update_password(user, form.new_password.data)
send_sms_code(user.id, user.mobile_number)
users_dao.send_verify_code(user.id, 'sms')
session['user_details'] = {
'id': user.id,
'email': user.email_address,
'password': form.new_password.data}
return redirect(url_for('main.two_factor'))
else:
return render_template('views/new-password.html', token=token, form=form, user=user)

12
app/main/views/register.py
View File

@@ -18,17 +18,13 @@ from app.main.forms import RegisterUserForm
from app import user_api_client
# TODO how do we handle duplicate unverifed email addresses?
# malicious or otherwise.
from app.notify_client.sender import send_sms_code, send_email_code
@main.route('/register', methods=['GET', 'POST'])
def register():
if current_user and current_user.is_authenticated():
return redirect(url_for('main.choose_service'))
form = RegisterUserForm(users_dao.get_user_by_email)
form = RegisterUserForm(users_dao.is_email_unique)
if form.validate_on_submit():
try:
@@ -47,10 +43,10 @@ def register():
# 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=user.mobile_number)
send_email_code(user_id=user.id, email=user.email_address)
users_dao.send_verify_code(user.id, 'sms')
users_dao.send_verify_code(user.id, 'email')
session['expiry_date'] = str(datetime.now() + timedelta(hours=1))
session['user_details'] = {"email": user.email_address, "id": user.id}
return redirect('/verify')
return redirect(url_for('main.verify'))
return render_template('views/register.html', form=form)

23
app/main/views/service_settings.py
View File

@@ -1,10 +1,11 @@
from flask import (
render_template, redirect, request, url_for, abort, session)
from flask_login import login_required
from flask_login import (login_required, current_user)
from app.main import main
from app.main.dao.services_dao import (
get_service_by_id, delete_service, update_service)
from app.main.dao.users_dao import verify_password
from app.main.forms import ConfirmPasswordForm, ServiceNameForm
from client.errors import HTTPError
@@ -61,7 +62,10 @@ def service_name_change_confirm(service_id):
else:
raise e
form = ConfirmPasswordForm()
# Validate password for form
def _check_password(pwd):
return verify_password(current_user, pwd)
form = ConfirmPasswordForm(_check_password)
if form.validate_on_submit():
service['name'] = session['service_name_change']
@@ -128,9 +132,10 @@ def service_status_change_confirm(service_id):
else:
raise e
# TODO validate password, will leave until
# user management has been moved to the api.
form = ConfirmPasswordForm()
# Validate password for form
def _check_password(pwd):
return verify_password(current_user, pwd)
form = ConfirmPasswordForm(_check_password)
if form.validate_on_submit():
service['active'] = True
@@ -175,9 +180,11 @@ def service_delete_confirm(service_id):
abort(404)
else:
raise e
# TODO validate password, will leave until
# user management has been moved to the api.
form = ConfirmPasswordForm()
# Validate password for form
def _check_password(pwd):
return verify_password(current_user, pwd)
form = ConfirmPasswordForm(_check_password)
if form.validate_on_submit():
try:

30
app/main/views/sign_in.py
View File

@@ -6,37 +6,29 @@ from flask import (
abort
)
from flask.ext.login import current_user
from app.main import main
from app.main.dao import users_dao
from app.main.forms import LoginForm
from app.notify_client.sender import send_sms_code
@main.route('/sign-in', methods=(['GET', 'POST']))
def sign_in():
if current_user and current_user.is_authenticated():
return redirect(url_for('main.choose_service'))
try:
form = LoginForm()
if form.validate_on_submit():
user = _get_and_verify_user(form.email_address.data, form.password.data)
if user:
send_sms_code(user.id, user.mobile_number)
session['user_email'] = user.email_address
return redirect(url_for('.two_factor'))
else:
# Vague error message for login in case of user not known, locked, inactive or password not verified
form.password.errors.append('Username or password is incorrect')
form = LoginForm()
if form.validate_on_submit():
user = _get_and_verify_user(form.email_address.data, form.password.data)
if user:
users_dao.send_verify_code(user.id, 'sms')
session['user_details'] = {"email": user.email_address, "id": user.id}
return redirect(url_for('.two_factor'))
else:
# Vague error message for login in case of user not known, locked, inactive or password not verified
form.password.errors.append('Username or password is incorrect')
return render_template('views/signin.html', form=form)
except:
import traceback
traceback.print_exc()
abort(500)
return render_template('views/signin.html', form=form)
def _get_and_verify_user(email_address, password):

22
app/main/views/two_factor.py
View File

@@ -5,20 +5,30 @@ from flask import (
from flask_login import login_user
from app.main import main
from app.main.dao import users_dao, verify_codes_dao
from app.main.dao import users_dao
from app.main.forms import TwoFactorForm
@main.route('/two-factor', methods=['GET', 'POST'])
def two_factor():
# TODO handle user_email not in session
user = users_dao.get_user_by_email(session['user_email'])
codes = verify_codes_dao.get_codes(user.id)
form = TwoFactorForm(codes)
user_id = session['user_details']['id']
def _check_code(code):
return users_dao.check_verify_code(user_id, code, "sms")
form = TwoFactorForm(_check_code)
if form.validate_on_submit():
verify_codes_dao.use_code_for_user_and_type(user_id=user.id, code_type='sms')
login_user(user)
try:
user = users_dao.get_user_by_id(user_id)
# Check if coming from new password page
if 'password' in session['user_details']:
user.set_password(session['user_details']['password'])
users_dao.update_user(user)
login_user(user)
finally:
del session['user_details']
return redirect(url_for('.choose_service'))
return render_template('views/two-factor.html', form=form)

205
app/main/views/user_profile.py
View File

@@ -1,139 +1,200 @@
from flask import request, render_template, redirect, url_for
from flask import (
request, render_template, redirect, url_for, session)
from flask.ext.login import current_user
from flask_login import login_required
from app.main import main
from app.main.dao.users_dao import (
verify_password, update_user, check_verify_code, is_email_unique,
send_verify_code)
from app.main.forms import (
ChangePasswordForm, ChangeNameForm, ChangeEmailForm, ConfirmEmailForm,
ChangeMobileNumberForm, ConfirmMobileNumberForm, ConfirmPasswordForm
)
NEW_EMAIL = 'new-email'
NEW_MOBILE = 'new-mob'
NEW_EMAIL_PASSWORD_CONFIRMED = 'new-email-password-confirmed'
NEW_MOBILE_PASSWORD_CONFIRMED = 'new-mob-password-confirmed'
@main.route("/user-profile")
@login_required
def user_profile():
return render_template('views/user-profile.html')
@main.route("/user-profile/name", methods=['GET', 'POST'])
@login_required
def user_profile_name():
form = ChangeNameForm()
form = ChangeNameForm(new_name=current_user.name)
if request.method == 'GET':
if current_user.is_authenticated():
form.new_name.data = current_user.name
return render_template(
'views/user-profile/change.html',
thing='name',
form_field=form.new_name
)
elif request.method == 'POST':
if form.validate_on_submit():
current_user.name = form.new_name.data
update_user(current_user)
return redirect(url_for('.user_profile'))
return render_template(
'views/user-profile/change.html',
thing='name',
form_field=form.new_name
)
@main.route("/user-profile/email", methods=['GET', 'POST'])
@login_required
def user_profile_email():
form = ChangeEmailForm()
def _is_email_unique(email):
return is_email_unique(email)
form = ChangeEmailForm(_is_email_unique,
email_address=current_user.email_address)
if request.method == 'GET':
if current_user.is_authenticated():
form.email_address.data = current_user.email_address
return render_template(
'views/user-profile/change.html',
thing='email address',
form_field=form.email_address
)
elif request.method == 'POST':
if form.validate_on_submit():
session[NEW_EMAIL] = form.email_address.data
return redirect(url_for('.user_profile_email_authenticate'))
return render_template(
'views/user-profile/change.html',
thing='email address',
form_field=form.email_address
)
@main.route("/user-profile/email/authenticate", methods=['GET', 'POST'])
@login_required
def user_profile_email_authenticate():
form = ConfirmPasswordForm()
# Validate password for form
def _check_password(pwd):
return verify_password(current_user.id, pwd)
form = ConfirmPasswordForm(_check_password)
if request.method == 'GET':
return render_template(
'views/user-profile/authenticate.html',
thing='email address',
form=form,
back_link=url_for('.user_profile_email')
)
elif request.method == 'POST':
if NEW_EMAIL not in session:
return redirect('main.user_profile_email')
if form.validate_on_submit():
session[NEW_EMAIL_PASSWORD_CONFIRMED] = True
send_verify_code(current_user.id, 'email', to=session[NEW_EMAIL])
return redirect(url_for('.user_profile_email_confirm'))
return render_template(
'views/user-profile/authenticate.html',
thing='email address',
form=form,
back_link=url_for('.user_profile_email')
)
@main.route("/user-profile/email/confirm", methods=['GET', 'POST'])
@login_required
def user_profile_email_confirm():
form = ConfirmEmailForm()
# Validate verify code for form
def _check_code(cde):
return check_verify_code(current_user.id, cde, 'email')
form = ConfirmEmailForm(_check_code)
if request.method == 'GET':
return render_template(
'views/user-profile/confirm.html',
form_field=form.email_code,
thing='email address'
)
elif request.method == 'POST':
if NEW_EMAIL_PASSWORD_CONFIRMED not in session:
return redirect('main.user_profile_email_authenticate')
if form.validate_on_submit():
current_user.email_address = session[NEW_EMAIL]
del session[NEW_EMAIL]
del session[NEW_EMAIL_PASSWORD_CONFIRMED]
update_user(current_user)
return redirect(url_for('.user_profile'))
return render_template(
'views/user-profile/confirm.html',
form_field=form.email_code,
thing='email address'
)
@main.route("/user-profile/mobile-number", methods=['GET', 'POST'])
@login_required
def user_profile_mobile_number():
form = ChangeMobileNumberForm()
form = ChangeMobileNumberForm(mobile_number=current_user.mobile_number)
if request.method == 'GET':
if current_user.is_authenticated():
form.mobile_number.data = current_user.mobile_number
return render_template(
'views/user-profile/change.html',
thing='mobile number',
form_field=form.mobile_number
)
elif request.method == 'POST':
if form.validate_on_submit():
session[NEW_MOBILE] = form.mobile_number.data
return redirect(url_for('.user_profile_mobile_number_authenticate'))
return render_template(
'views/user-profile/change.html',
thing='mobile number',
form_field=form.mobile_number
)
@main.route("/user-profile/mobile-number/authenticate", methods=['GET', 'POST'])
@login_required
def user_profile_mobile_number_authenticate():
form = ConfirmPasswordForm()
# Validate password for form
def _check_password(pwd):
return verify_password(current_user.id, pwd)
form = ConfirmPasswordForm(_check_password)
if request.method == 'GET':
return render_template(
'views/user-profile/authenticate.html',
thing='mobile number',
form=form,
back_link=url_for('.user_profile_mobile_number_confirm')
)
elif request.method == 'POST':
if NEW_MOBILE not in session:
return redirect(url_for('.user_profile_mobile_number'))
if form.validate_on_submit():
session[NEW_MOBILE_PASSWORD_CONFIRMED] = True
send_verify_code(current_user.id, 'sms', to=session[NEW_MOBILE])
return redirect(url_for('.user_profile_mobile_number_confirm'))
return render_template(
'views/user-profile/authenticate.html',
thing='mobile number',
form=form,
back_link=url_for('.user_profile_mobile_number_confirm')
)
@main.route("/user-profile/mobile-number/confirm", methods=['GET', 'POST'])
@login_required
def user_profile_mobile_number_confirm():
form = ConfirmMobileNumberForm()
# Validate verify code for form
def _check_code(cde):
return check_verify_code(current_user.id, cde, 'sms')
if request.method == 'GET':
return render_template(
'views/user-profile/confirm.html',
form_field=form.sms_code,
thing='mobile number'
)
elif request.method == 'POST':
if NEW_MOBILE_PASSWORD_CONFIRMED not in session:
return redirect(url_for('.user_profile_mobile_number'))
form = ConfirmMobileNumberForm(_check_code)
if form.validate_on_submit():
current_user.mobile_number = session[NEW_MOBILE]
del session[NEW_MOBILE]
del session[NEW_MOBILE_PASSWORD_CONFIRMED]
update_user(current_user)
return redirect(url_for('.user_profile'))
return render_template(
'views/user-profile/confirm.html',
form_field=form.sms_code,
thing='mobile number'
)
@main.route("/user-profile/password", methods=['GET', 'POST'])
@login_required
def user_profile_password():
form = ChangePasswordForm()
# Validate password for form
def _check_password(pwd):
return verify_password(current_user.id, pwd)
form = ChangePasswordForm(_check_password)
if request.method == 'GET':
return render_template(
'views/user-profile/change-password.html',
form=form
)
elif request.method == 'POST':
if form.validate_on_submit():
current_user.set_password(form.new_password.data)
update_user(current_user)
return redirect(url_for('.user_profile'))
return render_template(
'views/user-profile/change-password.html',
form=form
)

12
app/main/views/verify.py
View File

@@ -11,7 +11,7 @@ from client.errors import HTTPError
from flask_login import login_user
from app.main import main
from app.main.dao import users_dao, verify_codes_dao
from app.main.dao import users_dao
from app.main.forms import VerifyForm
@@ -20,13 +20,11 @@ def verify():
# TODO there needs to be a way to regenerate a session id
# or handle gracefully.
user_id = session['user_details']['id']
codes = verify_codes_dao.get_codes(user_id)
form = VerifyForm(codes)
def _check_code(code, code_type):
return users_dao.check_verify_code(user_id, code, code_type)
form = VerifyForm(_check_code)
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')
try:
user = users_dao.get_user_by_id(user_id)
activated_user = users_dao.activate_user(user)
@@ -37,5 +35,7 @@ def verify():
abort(404)
else:
raise e
finally:
del session['user_details']
return render_template('views/verify.html', form=form)

126
app/models.py
View File

@@ -1,126 +0,0 @@
import datetime
from app import db
from flask import current_app
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
DATE_FORMAT = "%Y-%m-%d"
class VerifyCodes(db.Model):
__tablename__ = 'verify_codes'
code_types = ['email', 'sms']
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, index=True, unique=False, nullable=False)
code = db.Column(db.String, nullable=False)
code_type = db.Column(db.Enum(code_types, name='verify_code_types'), index=False, unique=False, nullable=False)
expiry_datetime = db.Column(db.DateTime, nullable=False)
code_used = db.Column(db.Boolean, default=False)
class Roles(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
role = db.Column(db.String, nullable=False, unique=True)
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String, nullable=False, index=True, unique=False)
email_address = db.Column(db.String(255), nullable=False, index=True, unique=True)
password = db.Column(db.String, index=False, unique=False, nullable=False)
mobile_number = db.Column(db.String, index=False, unique=False, nullable=False)
created_at = db.Column(db.DateTime,
index=False,
unique=False,
nullable=False,
default=datetime.datetime.now)
updated_at = db.Column(db.DateTime,
index=False,
unique=False,
nullable=True,
onupdate=datetime.datetime.now)
password_changed_at = db.Column(db.DateTime, index=False, unique=False, nullable=True)
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'), index=True, unique=False, nullable=False)
logged_in_at = db.Column(db.DateTime, nullable=True)
failed_login_count = db.Column(db.Integer, nullable=False, default=0)
# TODO should this be an enum?
state = db.Column(db.String, nullable=False, default='pending')
def serialize(self):
serialized = {
'id': self.id,
'name': self.name,
'emailAddress': self.email_address,
'createdAt': self.created_at.strftime(DATETIME_FORMAT),
'updatedAt': self.updated_at.strftime(DATETIME_FORMAT),
'role': self.role,
'passwordChangedAt': self.password_changed_at.strftime(DATETIME_FORMAT),
'failedLoginCount': self.failed_login_count
}
return filter_null_value_fields(serialized)
def is_authenticated(self):
return True
def is_active(self):
if self.state != 'active':
return False
else:
return True
def is_anonymous(self):
return False
def get_id(self):
return self.id
def is_locked(self):
if self.failed_login_count < current_app.config['MAX_FAILED_LOGIN_COUNT']:
return False
else:
return True
# user_to_service = db.Table(
# 'user_to_service',
# db.Model.metadata,
# db.Column('user_id', db.Integer, db.ForeignKey('users.id')),
# db.Column('service_id', db.Integer, db.ForeignKey('services.id'))
# )
# class Service(db.Model):
# __tablename__ = 'services'
# id = db.Column(db.Integer, primary_key=True)
# name = db.Column(db.String(255), nullable=False, unique=True)
# created_at = db.Column(db.DateTime, index=False, unique=False, nullable=False)
# active = db.Column(db.Boolean, index=False, unique=False, nullable=False)
# limit = db.Column(db.BigInteger, index=False, unique=False, nullable=False)
# users = db.relationship('User', secondary=user_to_service, backref=db.backref('user_to_service', lazy='dynamic'))
# restricted = db.Column(db.Boolean, index=False, unique=False, nullable=False)
# def serialize(self):
# serialized = {
# 'id': self.id,
# 'name': self.name,
# 'createdAt': self.created_at.strftime(DATETIME_FORMAT),
# 'active': self.active,
# 'restricted': self.restricted,
# 'limit': self.limit,
# 'user': self.users.serialize()
# }
# return filter_null_value_fields(serialized)
def filter_null_value_fields(obj):
return dict(
filter(lambda x: x[1] is not None, obj.items())
)

23
app/notify_client/sender.py
View File

@@ -1,32 +1,9 @@
from random import randint
from flask import url_for, current_app
from itsdangerous import URLSafeTimedSerializer, SignatureExpired
from app.main.dao import verify_codes_dao
from app import notifications_api_client
def create_verify_code():
return ''.join(["%s" % randint(0, 9) for _ in range(0, 5)])
def send_sms_code(user_id, mobile_number):
sms_code = create_verify_code()
verify_codes_dao.add_code(user_id=user_id, code=sms_code, code_type='sms')
notifications_api_client.send_sms(mobile_number=mobile_number,
message=sms_code)
return sms_code
def send_email_code(user_id, email):
email_code = create_verify_code()
verify_codes_dao.add_code(user_id=user_id, code=email_code, code_type='email')
notifications_api_client.send_email(email_address=email,
from_address='notify@digital.cabinet-office.gov.uk',
message=email_code,
subject='Verification code')
return email_code
def send_change_password_email(email):
link_to_change_password = url_for('.new_password', token=generate_token(email), _external=True)
notifications_api_client.send_email(email_address=email,

60
app/notify_client/user_api_client.py
View File

@@ -45,23 +45,15 @@ class UserApiClient(BaseAPIClient):
user_data = self.put(url, data=data)
return User(user_data['data'], max_failed_login_count=self.max_failed_login_count)
def verify_password(self, user, password):
def verify_password(self, user_id, password):
try:
data = user.serialize()
url = "/user/{}/verify/password".format(user.id)
data["password"] = password
resp = self.post(url, data=data)
if resp.status_code == 204:
return True
url = "/user/{}/verify/password".format(user_id)
data = {"password": password}
self.post(url, data=data)
return True
except HTTPError as e:
if e.status_code == 400 or e.status_code == 404:
return False
# TODO temp work around until client fixed
except InvalidResponse as e:
if e.status_code == 204:
return True
else:
raise e
def get_user_by_email(self, email_address):
users = self.get_users()
@@ -70,6 +62,27 @@ class UserApiClient(BaseAPIClient):
return user[0]
return None
def send_verify_code(self, user_id, code_type, to=None):
data = {'code_type': code_type}
if to:
data['to'] = to
endpoint = '/user/{}/code'.format(user_id)
resp = self.post(endpoint, data=data)
def check_verify_code(self, user_id, code, code_type):
data = {'code_type': code_type, 'code': code}
endpoint = '/user/{}/verify/code'.format(user_id)
try:
resp = self.post(endpoint, data=data)
return True, ''
except HTTPError as e:
if e.status_code == 400 or e.status_code == 404:
if 'Code not found' in e.message:
return False, 'Code not found'
elif 'Code has expired' in e.message:
return False, 'Code has expired'
raise e
class User(object):
def __init__(self, fields, max_failed_login_count=3):
@@ -154,11 +167,16 @@ class User(object):
return self.failed_login_count >= self.max_failed_login_count
def serialize(self):
return {"id": self.id,
"name": self.name,
"email_address": self.email_address,
"mobile_number": self.mobile_number,
"password_changed_at": self.password_changed_at,
"state": self.state,
"failed_login_count": self.failed_login_count
}
dct = {"id": self.id,
"name": self.name,
"email_address": self.email_address,
"mobile_number": self.mobile_number,
"password_changed_at": self.password_changed_at,
"state": self.state,
"failed_login_count": self.failed_login_count}
if getattr(self, '_password', None):
dct['password'] = self._password
return dct
def set_password(self, pwd):
self._password = pwd