Merge pull request #22 from alphagov/implement_two_factor

Implement two factor

app/main/__init__.py

@@ -3,4 +3,4 @@ from flask import Blueprint
 main = Blueprint('main', __name__)
 
 
-from app.main.views import index, sign_in, register, verify
+from app.main.views import index, sign_in, register, two_factor, verify

app/main/forms.py

@@ -1,6 +1,6 @@
 from flask import session
 from flask_wtf import Form
-from wtforms import StringField, PasswordField, IntegerField
+from wtforms import StringField, PasswordField
 from wtforms.validators import DataRequired, Email, Length, Regexp
 
 from app.main.encryption import checkpw
@@ -41,6 +41,14 @@ class RegisterUserForm(Form):
                                          Blacklist(message='That password is blacklisted, too common')])
 
 
+class TwoFactorForm(Form):
+    sms_code = StringField('sms code', validators=[DataRequired(message='Please enter your code'),
+                                                   Regexp(regex=verify_code, message='Code must be 5 digits')])
+
+    def validate_sms_code(self, a):
+        validate_code(self.sms_code, session['sms_code'])
+
+
 class VerifyForm(Form):
     sms_code = StringField("Text message confirmation code",
                            validators=[DataRequired(message='SMS code can not be empty'),
@@ -50,17 +58,18 @@ class VerifyForm(Form):
                                          Regexp(regex=verify_code, message='Code must be 5 digits')])
 
     def validate_email_code(self, a):
-        if self.email_code.data is not None:
+        validate_code(self.email_code, session['email_code'])
-            if checkpw(str(self.email_code.data), session['email_code']) is False:
-                self.email_code.errors.append('Code does not match')
-                return False
-        else:
-            return True
 
     def validate_sms_code(self, a):
-        if self.sms_code.data is not None:
+        validate_code(self.sms_code, session['sms_code'])
-            if checkpw(str(self.sms_code.data), session['sms_code']) is False:
+
-                self.sms_code.errors.append('Code does not match')
+
-                return False
+def validate_code(field, code):
+    if field.data is not None:
+        if checkpw(str(field.data), code) is False:
+            field.errors.append('Code does not match')
+            return False
         else:
             return True
+    else:
+        return True

app/main/views/__init__.py

@@ -0,0 +1,30 @@
+from random import randint
+from app import admin_api_client
+from app.main.exceptions import AdminApiClientException
+
+
+def create_verify_code():
+    return ''.join(["%s" % randint(0, 9) for _ in range(0, 5)])
+
+
+def send_sms_code(mobile_number):
+    sms_code = create_verify_code()
+    try:
+        admin_api_client.send_sms(mobile_number, message=sms_code, token=admin_api_client.auth_token)
+    except:
+        raise AdminApiClientException('Exception when sending sms.')
+    return sms_code
+
+
+def send_email_code(email):
+    email_code = create_verify_code()
+    try:
+        admin_api_client.send_email(email_address=email,
+                                    from_str='notify@digital.cabinet-office.gov.uk',
+                                    message=email_code,
+                                    subject='Verification code',
+                                    token=admin_api_client.auth_token)
+    except:
+        raise AdminApiClientException('Exception when sending email.')
+
+    return email_code

app/main/views/index.py

@@ -41,11 +41,6 @@ def addservice():
     return render_template('add-service.html')
 
 
-@main.route("/two-factor")
-def twofactor():
-    return render_template('two-factor.html')
-
-
 @main.route("/send-sms")
 def sendsms():
     return render_template('send-sms.html')

app/main/views/register.py

@@ -1,15 +1,14 @@
 from datetime import datetime, timedelta
-from random import randint
 
 from flask import render_template, redirect, jsonify, session
 from sqlalchemy.exc import SQLAlchemyError
 
-from app import admin_api_client
 from app.main import main
 from app.main.dao import users_dao
 from app.main.encryption import hashpw
 from app.main.exceptions import AdminApiClientException
 from app.main.forms import RegisterUserForm
+from app.main.views import send_sms_code, send_email_code
 from app.models import User
 
 
@@ -44,30 +43,3 @@ def process_register():
     else:
         return jsonify(form.errors), 400
     return redirect('/verify')
-
-
-def send_sms_code(mobile_number):
-    sms_code = _create_code()
-    try:
-        admin_api_client.send_sms(mobile_number, message=sms_code, token=admin_api_client.auth_token)
-    except:
-        raise AdminApiClientException('Exception when sending sms.')
-    return sms_code
-
-
-def send_email_code(email):
-    email_code = _create_code()
-    try:
-        admin_api_client.send_email(email_address=email,
-                                    from_str='notify@digital.cabinet-office.gov.uk',
-                                    message=email_code,
-                                    subject='Verification code',
-                                    token=admin_api_client.auth_token)
-    except:
-        raise AdminApiClientException('Exception when sending email.')
-
-    return email_code
-
-
-def _create_code():
-    return ''.join(["%s" % randint(0, 9) for _ in range(0, 5)])

app/main/views/sign_in.py

@@ -1,10 +1,12 @@
 from flask import render_template, redirect, jsonify
-from flask_login import login_user
+from flask import session
 
 from app.main import main
 from app.main.dao import users_dao
 from app.main.encryption import checkpw
+from app.main.encryption import hashpw
 from app.main.forms import LoginForm
+from app.main.views import send_sms_code
 
 
 @main.route("/sign-in", methods=(['GET']))
@@ -24,7 +26,9 @@ def process_sign_in():
         if not user.is_active():
             return jsonify(active_user=False), 401
         if checkpw(form.password.data, user.password):
-            login_user(user)
+            sms_code = send_sms_code(user.mobile_number)
+            session['user_id'] = user.id
+            session['sms_code'] = hashpw(sms_code)
         else:
             users_dao.increment_failed_login_count(user.id)
             return jsonify(authorization=False), 401

app/main/views/two_factor.py

@@ -0,0 +1,23 @@
+from flask import render_template, redirect, jsonify, session
+from flask_login import login_user
+
+from app.main import main
+from app.main.dao import users_dao
+from app.main.forms import TwoFactorForm
+
+
+@main.route("/two-factor", methods=['GET'])
+def render_two_factor():
+    return render_template('two-factor.html', form=TwoFactorForm())
+
+
+@main.route('/two-factor', methods=['POST'])
+def process_two_factor():
+    form = TwoFactorForm()
+
+    if form.validate_on_submit():
+        user = users_dao.get_user_by_id(session['user_id'])
+        login_user(user)
+        return redirect('/dashboard')
+    else:
+        return jsonify(form.errors), 400

app/templates/two-factor.html

@@ -12,15 +12,18 @@ GOV.UK Notify | Text verification
 
 		<p>We've sent you a text message with a verification code.</p>
 
-  		<p>
-  		<label class="form-label" for="email">Enter verification code<br>
-  		<input class="form-control-1-4" id="email" type="text"><br>
-  		<span class="font-xsmall"><a href="verification-not-received">I haven't received a text</a></span>
-  		</p>
 
-  		<p>
+		<form autocomplete="off" action="" method="post">
-  		<a class="button" href="dashboard" role="button">Continue</a>
+			{{ form.hidden_tag() }}
-  		</p>
+  			<p>
+				<label class="form-label">Enter verification code</label><br>
+				{{ form.sms_code(class="form-control-1-4", autocomplete="off") }} <br>
+				<span class="font-xsmall"><a href="verification-not-received">I haven't received a text</a></span>
+  			</p>
+  			<p>
+  			<button class="button" href="dashboard" role="button">Continue</button>
+  			</p>
+		</form>
 	</div>
 </div>
 

requirements_for_test.txt

@@ -1,4 +1,4 @@
 -r requirements.txt
 pep8==1.5.7
 pytest==2.8.1
-pytest-mock==0.8.1
+pytest-mock==0.8.1

tests/app/main/views/__init__.py

@@ -0,0 +1,16 @@
+from datetime import datetime
+
+from app.main.dao import users_dao
+from app.models import User
+
+
+def create_test_user():
+    user = User(name='Test User',
+                password='somepassword',
+                email_address='test@user.gov.uk',
+                mobile_number='+441234123412',
+                created_at=datetime.now(),
+                role_id=1,
+                state='pending')
+    users_dao.insert_user(user)
+    return user

tests/app/main/views/test_sign_in.py

@@ -13,7 +13,8 @@ def test_render_sign_in_returns_sign_in_template(notifications_admin):
     assert 'Forgotten password?' in response.get_data(as_text=True)
 
 
-def test_process_sign_in_return_2fa_template(notifications_admin, notifications_admin_db):
+def test_process_sign_in_return_2fa_template(notifications_admin, notifications_admin_db, mocker):
+    _set_up_mocker(mocker)
     user = User(email_address='valid@example.gov.uk',
                 password='val1dPassw0rd!',
                 mobile_number='+441234123123',
@@ -79,3 +80,8 @@ def test_should_return_401_when_user_does_not_exist(notifications_admin, notific
                                                             'password': 'doesNotExist!'})
 
     assert response.status_code == 401
+
+
+def _set_up_mocker(mocker):
+    mocker.patch("app.admin_api_client.send_sms")
+    mocker.patch("app.admin_api_client.send_email")

tests/app/main/views/test_two_factor.py

@@ -0,0 +1,60 @@
+from flask import json
+
+from app.main.encryption import hashpw
+from tests.app.main.views import create_test_user
+
+
+def test_should_render_two_factor_page(notifications_admin, notifications_admin_db):
+    response = notifications_admin.test_client().get('/two-factor')
+    assert response.status_code == 200
+    assert '''We've sent you a text message with a verification code.''' in response.get_data(as_text=True)
+
+
+def test_should_login_user_and_redirect_to_dashboard(notifications_admin, notifications_admin_db):
+    with notifications_admin.test_client() as client:
+        with client.session_transaction() as session:
+            user = create_test_user()
+            session['user_id'] = user.id
+            session['sms_code'] = hashpw('12345')
+        response = client.post('/two-factor',
+                               data={'sms_code': '12345'})
+
+        assert response.status_code == 302
+        assert response.location == 'http://localhost/dashboard'
+
+
+def test_should_return_400_with_sms_code_error_when_sms_code_is_wrong(notifications_admin, notifications_admin_db):
+    with notifications_admin.test_client() as client:
+        with client.session_transaction() as session:
+            user = create_test_user()
+            session['user_id'] = user.id
+            session['sms_code'] = hashpw('12345')
+        response = client.post('/two-factor',
+                               data={'sms_code': '23456'})
+        assert response.status_code == 400
+        assert {'sms_code': ['Code does not match']} == json.loads(response.get_data(as_text=True))
+
+
+def test_should_return_400_when_sms_code_is_empty(notifications_admin, notifications_admin_db):
+    with notifications_admin.test_client() as client:
+        with client.session_transaction() as session:
+            user = create_test_user()
+            session['user_id'] = user.id
+            session['sms_code'] = hashpw('12345')
+        response = client.post('/two-factor')
+        assert response.status_code == 400
+        assert {'sms_code': ['Please enter your code']} == json.loads(response.get_data(as_text=True))
+
+
+def test_should_return_400_when_sms_code_is_too_short(notifications_admin, notifications_admin_db):
+    with notifications_admin.test_client() as client:
+        with client.session_transaction() as session:
+            user = create_test_user()
+            session['user_id'] = user.id
+            session['sms_code'] = hashpw('23467')
+        response = client.post('/two-factor', data={'sms_code': '2346'})
+        assert response.status_code == 400
+        data = json.loads(response.get_data(as_text=True))
+        assert len(data.keys()) == 1
+        assert 'sms_code' in data
+        assert data['sms_code'].sort() == ['Code must be 5 digits', 'Code does not match'].sort()

tests/app/main/views/test_verify.py

@@ -1,8 +1,8 @@
-from datetime import datetime
+from flask import json
 
 from app.main.dao import users_dao
 from app.main.encryption import hashpw
-from app.models import User
+from tests.app.main.views import create_test_user
 
 
 def test_should_return_verify_template(notifications_admin, notifications_admin_db):
@@ -14,7 +14,7 @@ def test_should_return_verify_template(notifications_admin, notifications_admin_
 def test_should_redirect_to_add_service_when_code_are_correct(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('12345')
             session['email_code'] = hashpw('23456')
@@ -28,7 +28,7 @@ def test_should_redirect_to_add_service_when_code_are_correct(notifications_admi
 def test_should_activate_user_after_verify(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('12345')
             session['email_code'] = hashpw('23456')
@@ -43,7 +43,7 @@ def test_should_activate_user_after_verify(notifications_admin, notifications_ad
 def test_should_return_400_when_sms_code_is_wrong(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('12345')
             session['email_code'] = hashpw('23456')
@@ -51,14 +51,13 @@ def test_should_return_400_when_sms_code_is_wrong(notifications_admin, notificat
                                data={'sms_code': '98765',
                                      'email_code': '23456'})
         assert response.status_code == 400
-        assert 'sms_code' in response.get_data(as_text=True)
+        assert {'sms_code': ['Code does not match']} == json.loads(response.get_data(as_text=True))
-        assert 'Code does not match' in response.get_data(as_text=True)
 
 
 def test_should_return_400_when_email_code_is_wrong(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('12345')
             session['email_code'] = hashpw('98456')
@@ -66,58 +65,57 @@ def test_should_return_400_when_email_code_is_wrong(notifications_admin, notific
                                data={'sms_code': '12345',
                                      'email_code': '23456'})
         assert response.status_code == 400
-        print(response.get_data(as_text=True))
+        assert {'email_code': ['Code does not match']} == json.loads(response.get_data(as_text=True))
-        assert 'email_code' in response.get_data(as_text=True)
-        assert 'Code does not match' in response.get_data(as_text=True)
 
 
 def test_should_return_400_when_sms_code_is_missing(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('12345')
             session['email_code'] = hashpw('98456')
         response = client.post('/verify',
-                               data={'email_code': '23456'})
+                               data={'email_code': '98456'})
         assert response.status_code == 400
-        assert 'SMS code can not be empty' in response.get_data(as_text=True)
+        assert {'sms_code': ['SMS code can not be empty']} == json.loads(response.get_data(as_text=True))
 
 
 def test_should_return_400_when_email_code_is_missing(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('23456')
             session['email_code'] = hashpw('23456')
         response = client.post('/verify',
                                data={'sms_code': '23456'})
         assert response.status_code == 400
-        assert 'Email code can not be empty' in response.get_data(as_text=True)
+        assert {'email_code': ['Email code can not be empty']} == json.loads(response.get_data(as_text=True))
 
 
 def test_should_return_400_when_email_code_has_letter(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('23456')
             session['email_code'] = hashpw('23456')
         response = client.post('/verify',
                                data={'sms_code': '23456',
                                      'email_code': 'abcde'})
-        data = response.get_data(as_text=True)
         assert response.status_code == 400
+        data = json.loads(response.get_data(as_text=True))
+        expected = {'email_code': ['Code does not match', 'Code must be 5 digits']}
+        assert len(data.keys()) == 1
         assert 'email_code' in data
-        assert 'Code does not match' in data
+        assert data['email_code'].sort() == expected['email_code'].sort()
-        assert 'Code must be 5 digits' in data
 
 
 def test_should_return_400_when_sms_code_is_too_short(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('23456')
             session['email_code'] = hashpw('23456')
@@ -125,16 +123,17 @@ def test_should_return_400_when_sms_code_is_too_short(notifications_admin, notif
                                data={'sms_code': '2345',
                                      'email_code': '23456'})
         assert response.status_code == 400
-        data = response.get_data(as_text=True)
+        data = json.loads(response.get_data(as_text=True))
+        expected = {'sms_code': ['Code must be 5 digits', 'Code does not match']}
+        assert len(data.keys()) == 1
         assert 'sms_code' in data
-        assert 'Code must be 5 digits' in data
+        assert data['sms_code'].sort() == expected['sms_code'].sort()
-        assert 'Code does not match' in data
 
 
 def test_should_return_302_when_email_code_starts_with_zero(notifications_admin, notifications_admin_db):
     with notifications_admin.test_client() as client:
         with client.session_transaction() as session:
-            user = _create_test_user()
+            user = create_test_user()
             session['user_id'] = user.id
             session['sms_code'] = hashpw('23456')
             session['email_code'] = hashpw('09765')
@@ -143,15 +142,3 @@ def test_should_return_302_when_email_code_starts_with_zero(notifications_admin,
                                      'email_code': '09765'})
         assert response.status_code == 302
         assert response.location == 'http://localhost/add-service'
-
-
-def _create_test_user():
-    user = User(name='Test User',
-                password='somepassword',
-                email_address='test@user.gov.uk',
-                mobile_number='+441234123412',
-                created_at=datetime.now(),
-                role_id=1,
-                state='pending')
-    users_dao.insert_user(user)
-    return user