mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-07-16 19:30:13 -04:00
check users' session id.
when a user enters their 2FA code, the API will store a random UUID
against them in the database - this code is then stored on the cookie
on the front end.
At the beginning of each authenticated request, we do the following
steps:
* Retrieve the user's cookie, and get the user_id from it
* Request that user's details from the database
* populate current_user with the DB model
* run the login_required decorator, which calls
current_user.is_authenticated
is_authenticated now also checks that the database model matches the
cookie for session_id. The potential states and meanings are as follows:
database | cookie | meaning
----------+--------+---------
None | None | New user, or system just been deployed.
| | Redirect to start page.
----------+--------+---------
'abc' | None | New browser (or cleared cookies). Redirect to
| | start page.
----------+--------+---------
None | 'abc' | Invalid state (cookie is set from user obj, so
| | would only happen if DB is cleared)
----------+--------+---------
'abc' | 'abc' | Same browser. Business as usual
----------+--------+---------
'abc' | 'def' | Different browser in cookie - db has been changed
| | since then. Redirect to start
This commit is contained in:
@@ -50,6 +50,7 @@ from app.notify_client.user_api_client import UserApiClient
|
||||
from app.notify_client.events_api_client import EventsApiClient
|
||||
from app.notify_client.provider_client import ProviderClient
|
||||
from app.notify_client.organisations_client import OrganisationsClient
|
||||
from app.notify_client.models import AnonymousUser
|
||||
|
||||
login_manager = LoginManager()
|
||||
csrf = CsrfProtect()
|
||||
@@ -103,6 +104,7 @@ def create_app():
|
||||
login_manager.login_view = 'main.sign_in'
|
||||
login_manager.login_message_category = 'default'
|
||||
login_manager.session_protection = None
|
||||
login_manager.anonymous_user = AnonymousUser
|
||||
|
||||
from app.main import main as main_blueprint
|
||||
application.register_blueprint(main_blueprint)
|
||||
|
||||
@@ -76,7 +76,13 @@ def sign_in():
|
||||
).format(password_reset=url_for('.forgot_password'))
|
||||
))
|
||||
|
||||
return render_template('views/signin.html', form=form, again=bool(request.args.get('next')))
|
||||
other_device = current_user.logged_in_elsewhere()
|
||||
return render_template(
|
||||
'views/signin.html',
|
||||
form=form,
|
||||
again=bool(request.args.get('next')),
|
||||
other_device=other_device
|
||||
)
|
||||
|
||||
|
||||
@login_manager.unauthorized_handler
|
||||
|
||||
@@ -26,6 +26,9 @@ def two_factor():
|
||||
if form.validate_on_submit():
|
||||
try:
|
||||
user = user_api_client.get_user(user_id)
|
||||
# the user will have a new current_session_id set by the API - store it in the cookie so we can match it in
|
||||
# future requests
|
||||
session['current_session_id'] = user.current_session_id
|
||||
services = service_api_client.get_active_services({'user_id': str(user_id)}).get('data', [])
|
||||
# Check if coming from new password page
|
||||
if 'password' in session['user_details']:
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from flask_login import (UserMixin, login_fresh)
|
||||
from flask_login import UserMixin, AnonymousUserMixin, login_fresh
|
||||
from flask import session
|
||||
|
||||
|
||||
class User(UserMixin):
|
||||
@@ -13,20 +14,25 @@ class User(UserMixin):
|
||||
self._state = fields.get('state')
|
||||
self.max_failed_login_count = max_failed_login_count
|
||||
self.platform_admin = fields.get('platform_admin')
|
||||
self.current_session_id = fields.get('current_session_id')
|
||||
|
||||
def get_id(self):
|
||||
return self.id
|
||||
|
||||
def logged_in_elsewhere(self):
|
||||
return session.get('current_session_id') != self.current_session_id
|
||||
|
||||
@property
|
||||
def is_active(self):
|
||||
return self.state == 'active'
|
||||
|
||||
@property
|
||||
def is_authenticated(self):
|
||||
# To handle remember me token renewal
|
||||
if not login_fresh():
|
||||
return False
|
||||
return super(User, self).is_authenticated
|
||||
return (
|
||||
login_fresh() and
|
||||
not self.logged_in_elsewhere() and
|
||||
super(User, self).is_authenticated
|
||||
)
|
||||
|
||||
@property
|
||||
def id(self):
|
||||
@@ -114,15 +120,18 @@ class User(UserMixin):
|
||||
return self.failed_login_count >= self.max_failed_login_count
|
||||
|
||||
def serialize(self):
|
||||
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,
|
||||
"permissions": [x for x in self._permissions]}
|
||||
if getattr(self, '_password', None):
|
||||
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,
|
||||
"permissions": [x for x in self._permissions],
|
||||
"current_session_id": self.current_session_id
|
||||
}
|
||||
if hasattr(self, '_password'):
|
||||
dct['password'] = self._password
|
||||
return dct
|
||||
|
||||
@@ -174,3 +183,9 @@ class InvitedUser(object):
|
||||
else:
|
||||
data['permissions'] = self.permissions
|
||||
return data
|
||||
|
||||
|
||||
class AnonymousUser(AnonymousUserMixin):
|
||||
# set the anonymous user so that if a new browser hits us we don't error http://stackoverflow.com/a/19275188
|
||||
def logged_in_elsewhere(self):
|
||||
return False
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import uuid
|
||||
|
||||
from flask import session
|
||||
from notifications_python_client.errors import HTTPError
|
||||
|
||||
from app.notify_client import NotifyAdminAPIClient
|
||||
|
||||
@@ -13,9 +13,15 @@
|
||||
|
||||
{% if again %}
|
||||
<h1 class="heading-large">You need to sign in again</h1>
|
||||
<p>
|
||||
We sign you out if you haven’t used Notify for a while.
|
||||
</p>
|
||||
{% if other_device %}
|
||||
<p>
|
||||
We signed you out because you logged in to Notify on another device.
|
||||
</p>
|
||||
{% else %}
|
||||
<p>
|
||||
We signed you out because you haven’t used Notify for a while.
|
||||
</p>
|
||||
{% endif %}
|
||||
{% else %}
|
||||
<h1 class="heading-large">Sign in</h1>
|
||||
<p>
|
||||
|
||||
@@ -80,7 +80,7 @@ def test_if_existing_user_accepts_twice_they_redirect_to_sign_in(
|
||||
page.select('main p')[0].text.strip(),
|
||||
) == (
|
||||
'You need to sign in again',
|
||||
'We sign you out if you haven’t used Notify for a while.',
|
||||
'We signed you out because you haven’t used Notify for a while.',
|
||||
)
|
||||
|
||||
|
||||
@@ -106,7 +106,7 @@ def test_existing_user_of_service_get_redirected_to_signin(
|
||||
page.select('main p')[0].text.strip(),
|
||||
) == (
|
||||
'You need to sign in again',
|
||||
'We sign you out if you haven’t used Notify for a while.',
|
||||
'We signed you out because you haven’t used Notify for a while.',
|
||||
)
|
||||
assert mock_accept_invite.call_count == 1
|
||||
|
||||
@@ -141,7 +141,7 @@ def test_existing_signed_out_user_accept_invite_redirects_to_sign_in(
|
||||
page.select('main p')[0].text.strip(),
|
||||
) == (
|
||||
'You need to sign in again',
|
||||
'We sign you out if you haven’t used Notify for a while.',
|
||||
'We signed you out because you haven’t used Notify for a while.',
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -1,16 +1,73 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from flask import url_for
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
|
||||
def test_render_sign_in_returns_sign_in_template(
|
||||
def test_render_sign_in_template_for_new_user(
|
||||
client
|
||||
):
|
||||
response = client.get(url_for('main.sign_in'))
|
||||
response = client.get(url_for('main.sign_in', next=None))
|
||||
assert response.status_code == 200
|
||||
assert 'Sign in' in response.get_data(as_text=True)
|
||||
assert 'Email address' in response.get_data(as_text=True)
|
||||
assert 'Password' in response.get_data(as_text=True)
|
||||
assert 'Forgot your password?' in response.get_data(as_text=True)
|
||||
resp = response.get_data(as_text=True)
|
||||
assert 'Sign in' in resp
|
||||
assert 'Email address' in resp
|
||||
assert 'Password' in resp
|
||||
assert 'Forgot your password?' in resp
|
||||
assert 'If you do not have an account, you can' in resp
|
||||
assert 'Sign in again' not in resp
|
||||
|
||||
|
||||
def test_sign_in_explains_session_timeout(client):
|
||||
response = client.get(url_for('main.sign_in', next='/foo'))
|
||||
assert response.status_code == 200
|
||||
assert 'We signed you out because you haven’t used Notify for a while.' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_sign_in_explains_other_browser(logged_in_client, api_user_active, mocker):
|
||||
api_user_active.current_session_id = str(uuid.UUID(int=1))
|
||||
mocker.patch('app.user_api_client.get_user', return_value=api_user_active)
|
||||
|
||||
with logged_in_client.session_transaction() as session:
|
||||
session['current_session_id'] = str(uuid.UUID(int=2))
|
||||
|
||||
response = logged_in_client.get(url_for('main.sign_in', next='/foo'))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert 'We signed you out because you logged in to Notify on another device' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_doesnt_redirect_to_sign_in_if_no_session_info(logged_in_client, api_user_active):
|
||||
assert api_user_active.current_session_id is None
|
||||
with logged_in_client.session_transaction() as session:
|
||||
session['current_session_id'] = None
|
||||
|
||||
response = logged_in_client.get(url_for('main.choose_service'))
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize('db_sess_id, cookie_sess_id', [
|
||||
pytest.mark.xfail((None, None)), # OK - not used notify since browser signout was implemented
|
||||
|
||||
(uuid.UUID(int=1), None), # BAD - has used other browsers before but this is a brand new browser with no cookie
|
||||
(uuid.UUID(int=1), uuid.UUID(int=2)), # BAD - this person has just signed in on a different browser
|
||||
])
|
||||
def test_redirect_to_sign_in_if_logged_in_from_other_browser(
|
||||
logged_in_client,
|
||||
api_user_active,
|
||||
mocker,
|
||||
db_sess_id,
|
||||
cookie_sess_id
|
||||
):
|
||||
api_user_active.current_session_id = db_sess_id
|
||||
mocker.patch('app.user_api_client.get_user', return_value=api_user_active)
|
||||
with logged_in_client.session_transaction() as session:
|
||||
session['current_session_id'] = str(cookie_sess_id)
|
||||
|
||||
response = logged_in_client.get(url_for('main.choose_service'))
|
||||
assert response.status_code == 302
|
||||
assert response.location == url_for('main.sign_in', next='/services', _external=True)
|
||||
|
||||
|
||||
def test_logged_in_user_redirects_to_choose_service(
|
||||
|
||||
Reference in New Issue
Block a user