mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-07-18 21:44:11 -04:00
Merge branch 'main' into notify-api-387
This commit is contained in:
@@ -39,5 +39,4 @@ from app.main.views import ( # noqa isort:skip
|
||||
uploads,
|
||||
user_profile,
|
||||
verify,
|
||||
webauthn_credentials,
|
||||
)
|
||||
|
||||
@@ -1058,10 +1058,6 @@ class BasePermissionsForm(StripWhitespaceForm):
|
||||
login_authentication=user.auth_type
|
||||
)
|
||||
|
||||
# If a user logs in with a security key, we generally don't want a service admin to be able to change this.
|
||||
# As well as enforcing this in the backend, we need to delete the auth radios to prevent validation errors.
|
||||
if user.webauthn_auth:
|
||||
del form.login_authentication
|
||||
return form
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from flask import abort, flash, redirect, render_template, request, url_for
|
||||
from flask import flash, redirect, render_template, request, url_for
|
||||
from flask_login import current_user
|
||||
from notifications_python_client.errors import HTTPError
|
||||
|
||||
@@ -56,8 +56,6 @@ def archive_user(user_id):
|
||||
@user_is_platform_admin
|
||||
def change_user_auth(user_id):
|
||||
user = User.from_id(user_id)
|
||||
if user.webauthn_auth:
|
||||
abort(403)
|
||||
|
||||
form = AuthTypeForm(auth_type=user.auth_type)
|
||||
|
||||
|
||||
@@ -141,9 +141,8 @@ def edit_user_permissions(service_id, user_id):
|
||||
folder_permissions=form.folder_permissions.data,
|
||||
set_by_id=current_user.id,
|
||||
)
|
||||
# Only change the auth type if this is supported for a service. If a user logs in with a
|
||||
# security key, we generally don't want them to be able to use something less secure.
|
||||
if service_has_email_auth and not user.auth_type == 'webauthn_auth':
|
||||
# Only change the auth type if this is supported for a service.
|
||||
if service_has_email_auth:
|
||||
user.update(auth_type=form.login_authentication.data)
|
||||
return redirect(url_for('.manage_users', service_id=service_id))
|
||||
|
||||
|
||||
@@ -47,8 +47,6 @@ def new_password(token):
|
||||
if user.email_auth:
|
||||
# they've just clicked an email link, so have done an email auth journey anyway. Just log them in.
|
||||
return log_in_user(user.id)
|
||||
elif user.webauthn_auth:
|
||||
return redirect(url_for('main.two_factor_webauthn', next=request.args.get('next')))
|
||||
else:
|
||||
# send user a 2fa sms code
|
||||
user.send_verify_code()
|
||||
|
||||
@@ -59,8 +59,6 @@ def sign_in():
|
||||
return redirect(url_for('.two_factor_sms', next=redirect_url))
|
||||
if user.email_auth:
|
||||
return redirect(url_for('.two_factor_email_sent', next=redirect_url))
|
||||
if user.webauthn_auth:
|
||||
return redirect(url_for('.two_factor_webauthn', next=redirect_url))
|
||||
|
||||
# Vague error message for login in case of user not known, locked, inactive or password not verified
|
||||
flash(Markup(
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import json
|
||||
|
||||
from flask import (
|
||||
abort,
|
||||
current_app,
|
||||
redirect,
|
||||
render_template,
|
||||
@@ -88,18 +87,6 @@ def two_factor_sms():
|
||||
return render_template('views/two-factor-sms.html', form=form, redirect_url=redirect_url)
|
||||
|
||||
|
||||
@main.route('/two-factor-webauthn', methods=['GET'])
|
||||
@redirect_to_sign_in
|
||||
def two_factor_webauthn():
|
||||
user_id = session['user_details']['id']
|
||||
user = User.from_id(user_id)
|
||||
|
||||
if not user.webauthn_auth:
|
||||
abort(403)
|
||||
|
||||
return render_template('views/two-factor-webauthn.html')
|
||||
|
||||
|
||||
@main.route('/re-validate-email', methods=['GET'])
|
||||
def revalidate_email_sent():
|
||||
title = 'Email resent' if request.args.get('email_resent') else 'Check your email'
|
||||
|
||||
@@ -11,7 +11,6 @@ from flask import (
|
||||
url_for,
|
||||
)
|
||||
from flask_login import current_user
|
||||
from notifications_python_client.errors import HTTPError
|
||||
from notifications_utils.url_safe_token import check_token
|
||||
|
||||
from app import user_api_client
|
||||
@@ -25,7 +24,6 @@ from app.main.forms import (
|
||||
ChangeMobileNumberForm,
|
||||
ChangeNameForm,
|
||||
ChangePasswordForm,
|
||||
ChangeSecurityKeyNameForm,
|
||||
ConfirmPasswordForm,
|
||||
ServiceOnOffSettingForm,
|
||||
TwoFactorForm,
|
||||
@@ -266,77 +264,3 @@ def user_profile_disable_platform_admin_view():
|
||||
'views/user-profile/disable-platform-admin-view.html',
|
||||
form=form
|
||||
)
|
||||
|
||||
|
||||
@main.route("/user-profile/security-keys", methods=['GET'])
|
||||
@user_is_logged_in
|
||||
def user_profile_security_keys():
|
||||
if not current_user.can_use_webauthn:
|
||||
abort(403)
|
||||
|
||||
return render_template(
|
||||
'views/user-profile/security-keys.html',
|
||||
)
|
||||
|
||||
|
||||
@main.route(
|
||||
"/user-profile/security-keys/<uuid:key_id>/manage",
|
||||
methods=['GET', 'POST'],
|
||||
endpoint="user_profile_manage_security_key"
|
||||
)
|
||||
@main.route(
|
||||
"/user-profile/security-keys/<uuid:key_id>/delete",
|
||||
methods=['GET'],
|
||||
endpoint="user_profile_confirm_delete_security_key"
|
||||
)
|
||||
@user_is_logged_in
|
||||
def user_profile_manage_security_key(key_id):
|
||||
if not current_user.can_use_webauthn:
|
||||
abort(403)
|
||||
|
||||
security_key = current_user.webauthn_credentials.by_id(key_id)
|
||||
|
||||
if not security_key:
|
||||
abort(404)
|
||||
|
||||
form = ChangeSecurityKeyNameForm(security_key_name=security_key.name)
|
||||
|
||||
if form.validate_on_submit():
|
||||
if form.security_key_name.data != security_key.name:
|
||||
user_api_client.update_webauthn_credential_name_for_user(
|
||||
user_id=current_user.id,
|
||||
credential_id=key_id,
|
||||
new_name_for_credential=form.security_key_name.data
|
||||
)
|
||||
return redirect(url_for('.user_profile_security_keys'))
|
||||
|
||||
if (request.endpoint == "main.user_profile_confirm_delete_security_key"):
|
||||
flash("Are you sure you want to delete this security key?", 'delete')
|
||||
|
||||
return render_template(
|
||||
'views/user-profile/manage-security-key.html',
|
||||
security_key=security_key,
|
||||
form=form
|
||||
)
|
||||
|
||||
|
||||
@main.route("/user-profile/security-keys/<uuid:key_id>/delete", methods=['POST'])
|
||||
@user_is_logged_in
|
||||
def user_profile_delete_security_key(key_id):
|
||||
if not current_user.can_use_webauthn:
|
||||
abort(403)
|
||||
|
||||
try:
|
||||
user_api_client.delete_webauthn_credential_for_user(
|
||||
user_id=current_user.id,
|
||||
credential_id=key_id
|
||||
)
|
||||
except HTTPError as e:
|
||||
message = "Cannot delete last remaining webauthn credential for user"
|
||||
if e.message == message:
|
||||
flash("You cannot delete your last security key.")
|
||||
return redirect(url_for('.user_profile_manage_security_key', key_id=key_id))
|
||||
else:
|
||||
raise e
|
||||
|
||||
return redirect(url_for('.user_profile_security_keys'))
|
||||
|
||||
@@ -1,161 +0,0 @@
|
||||
from fido2 import cbor
|
||||
from fido2.client import ClientData
|
||||
from fido2.ctap2 import AuthenticatorData
|
||||
from flask import abort, current_app, flash, redirect, request, session, url_for
|
||||
from flask_login import current_user
|
||||
|
||||
from app.main import main
|
||||
from app.models.user import User
|
||||
from app.models.webauthn_credential import RegistrationError, WebAuthnCredential
|
||||
from app.notify_client.user_api_client import user_api_client
|
||||
from app.utils.login import (
|
||||
email_needs_revalidating,
|
||||
log_in_user,
|
||||
redirect_to_sign_in,
|
||||
)
|
||||
from app.utils.user import user_is_logged_in
|
||||
|
||||
|
||||
@main.route('/webauthn/register')
|
||||
@user_is_logged_in
|
||||
def webauthn_begin_register():
|
||||
if not current_user.can_use_webauthn:
|
||||
abort(403)
|
||||
|
||||
server = current_app.webauthn_server
|
||||
|
||||
registration_data, state = server.register_begin(
|
||||
{
|
||||
"id": bytes(current_user.id, 'utf-8'),
|
||||
"name": current_user.email_address,
|
||||
"displayName": current_user.name,
|
||||
},
|
||||
credentials=current_user.webauthn_credentials.as_cbor,
|
||||
user_verification="discouraged", # don't ask for PIN
|
||||
authenticator_attachment="cross-platform",
|
||||
)
|
||||
|
||||
session["webauthn_registration_state"] = state
|
||||
return cbor.encode(registration_data)
|
||||
|
||||
|
||||
@main.route('/webauthn/register', methods=['POST'])
|
||||
@user_is_logged_in
|
||||
def webauthn_complete_register():
|
||||
if 'webauthn_registration_state' not in session:
|
||||
return cbor.encode("No registration in progress"), 400
|
||||
|
||||
try:
|
||||
credential = WebAuthnCredential.from_registration(
|
||||
session.pop("webauthn_registration_state"),
|
||||
cbor.decode(request.get_data()),
|
||||
)
|
||||
except RegistrationError as e:
|
||||
current_app.logger.info(f'User {current_user.id} could not register a new webauthn token - {e}')
|
||||
abort(400)
|
||||
|
||||
current_user.create_webauthn_credential(credential)
|
||||
current_user.update(auth_type='webauthn_auth')
|
||||
|
||||
flash((
|
||||
'Registration complete. Next time you sign in to Notify '
|
||||
'you’ll be asked to use your security key.'
|
||||
), 'default_with_tick')
|
||||
|
||||
return cbor.encode('')
|
||||
|
||||
|
||||
@main.route('/webauthn/authenticate', methods=['GET'])
|
||||
@redirect_to_sign_in
|
||||
def webauthn_begin_authentication():
|
||||
"""
|
||||
Initiate the authentication flow. This is called after the user clicks the "Check security key" button.
|
||||
|
||||
1. Get the user's credentials out of the database to present to the browser. The browser will only let you use a
|
||||
credential in that list.
|
||||
2. Call webauthn_server.authenticate_begin. This returns the authentication data, which includes the challenge and
|
||||
the origin domain to authenticate with. This also returns the state, which we store in the cookie so we can ensure
|
||||
the challenge is correct in webauthn_complete_authentication
|
||||
"""
|
||||
# get user from session
|
||||
user_to_login = User.from_id(session['user_details']['id'])
|
||||
|
||||
if not user_to_login.webauthn_auth:
|
||||
abort(403)
|
||||
|
||||
authentication_data, state = current_app.webauthn_server.authenticate_begin(
|
||||
credentials=user_to_login.webauthn_credentials.as_cbor,
|
||||
user_verification="discouraged", # don't ask for PIN
|
||||
)
|
||||
session["webauthn_authentication_state"] = state
|
||||
return cbor.encode(authentication_data)
|
||||
|
||||
|
||||
@main.route('/webauthn/authenticate', methods=['POST'])
|
||||
@redirect_to_sign_in
|
||||
def webauthn_complete_authentication():
|
||||
"""
|
||||
Complete the authentication flow. This is called after the user taps on their security key.
|
||||
|
||||
1. Try verifying the signed challenge returned from the browser with each public key we have in the database for
|
||||
that user.
|
||||
2. If succesful, log the user in, setting up the session etc. Then return the URL they should be redirected to.
|
||||
"""
|
||||
user_id = session['user_details']['id']
|
||||
user_to_login = User.from_id(user_id)
|
||||
|
||||
_verify_webauthn_authentication(user_to_login)
|
||||
redirect = _complete_webauthn_login_attempt(user_to_login)
|
||||
|
||||
return cbor.encode({'redirect_url': redirect.location}), 200
|
||||
|
||||
|
||||
def _verify_webauthn_authentication(user):
|
||||
"""
|
||||
Check that the presented security key is valid, has signed the right challenge, and belongs to the user
|
||||
we're trying to log in.
|
||||
"""
|
||||
state = session.pop("webauthn_authentication_state")
|
||||
request_data = cbor.decode(request.get_data())
|
||||
|
||||
try:
|
||||
current_app.webauthn_server.authenticate_complete(
|
||||
state=state,
|
||||
credentials=user.webauthn_credentials.as_cbor,
|
||||
credential_id=request_data['credentialId'],
|
||||
client_data=ClientData(request_data['clientDataJSON']),
|
||||
auth_data=AuthenticatorData(request_data['authenticatorData']),
|
||||
signature=request_data['signature']
|
||||
)
|
||||
except ValueError as exc:
|
||||
# We don't expect to reach this case in normal situations - normally errors (such as using the wrong
|
||||
# security key) will be caught in the browser inside `window.navigator.credentials.get`, and the js will
|
||||
# error first meaning it doesn't send the POST request to this method. If this method is called but the key
|
||||
# couldn't be authenticated, something went wrong along the way, probably:
|
||||
# * The browser didn't implement the webauthn standard correctly, and let something through it shouldn't have
|
||||
# * The key itself is in some way corrupted, or of lower security standard
|
||||
current_app.logger.info(f'User {user.id} could not sign in using their webauthn token - {exc}')
|
||||
user.complete_webauthn_login_attempt(is_successful=False)
|
||||
abort(403)
|
||||
|
||||
|
||||
def _complete_webauthn_login_attempt(user):
|
||||
"""
|
||||
* check the user hasn't gone over their max logins
|
||||
* check that the user's email is validated
|
||||
* if succesful, update current_session_id, log in date, and then redirect
|
||||
"""
|
||||
redirect_url = request.args.get('next')
|
||||
|
||||
# normally API handles this when verifying an sms or email code but since the webauthn logic happens in the
|
||||
# admin we need a separate call that just finalises the login in the database
|
||||
logged_in, _ = user.complete_webauthn_login_attempt()
|
||||
if not logged_in:
|
||||
# user account is locked as too many failed logins
|
||||
abort(403)
|
||||
|
||||
if email_needs_revalidating(user):
|
||||
user_api_client.send_verify_code(user.id, 'email', None, redirect_url)
|
||||
return redirect(url_for('.revalidate_email_sent', next=redirect_url))
|
||||
|
||||
return log_in_user(user.id)
|
||||
Reference in New Issue
Block a user