Merge pull request #29502 from edx/shafqat/VAN-666

feat: VAN-666 - Reject new password that is detected as vulnerable on password reset

cms/envs/common.py

@@ -496,6 +496,15 @@ IDA_LOGOUT_URI_LIST = []
 COURSE_AUTHORING_MICROFRONTEND_URL = None
 DISCUSSIONS_MICROFRONTEND_URL = None
 LIBRARY_AUTHORING_MICROFRONTEND_URL = None
+# .. toggle_name: ENABLE_AUTHN_RESET_PASSWORD_HIBP_POLICY
+# .. toggle_implementation: DjangoSetting
+# .. toggle_default: False
+# .. toggle_description: When enabled, this toggle activates the use of the password validation
+#   HIBP Policy.
+# .. toggle_use_cases: open_edx
+# .. toggle_creation_date: 2021-12-03
+# .. toggle_tickets: https://openedx.atlassian.net/browse/VAN-666
+ENABLE_AUTHN_RESET_PASSWORD_HIBP_POLICY = False
 
 ############################# SOCIAL MEDIA SHARING #############################
 SOCIAL_SHARING_SETTINGS = {

lms/envs/common.py

@@ -4701,6 +4701,15 @@ LEARNING_MICROFRONTEND_URL = None
 # .. setting_description: Base URL of the micro-frontend-based dicussions page.
 # .. setting_warning: Also set site's courseware.discussions_mfe waffle flag.
 DISCUSSIONS_MICROFRONTEND_URL = None
+# .. toggle_name: ENABLE_AUTHN_RESET_PASSWORD_HIBP_POLICY
+# .. toggle_implementation: DjangoSetting
+# .. toggle_default: False
+# .. toggle_description: When enabled, this toggle activates the use of the password validation
+#   HIBP Policy.
+# .. toggle_use_cases: open_edx
+# .. toggle_creation_date: 2021-12-03
+# .. toggle_tickets: https://openedx.atlassian.net/browse/VAN-666
+ENABLE_AUTHN_RESET_PASSWORD_HIBP_POLICY = False
 
 ############### Settings for the ace_common plugin #################
 ACE_ENABLED_CHANNELS = ['django_email']

openedx/core/djangoapps/password_policy/hibp.py

@@ -33,7 +33,10 @@ class PwnedPasswordsAPI:
     @staticmethod
     def range(password):
         """
-        Returns a dict containing hashed password signatures along with their count
+        Returns a dict containing hashed password signatures along with their count.
+        API URL takes first 5 characters of a SHA-1 password hash (not case-sensitive).
+        API response contains suffix of every hash beginning with the specified prefix,
+        followed by a count of how many times it appears in their data set.
 
         **Argument(s):
             password: a sha-1-hashed string against which pwnedservice is invoked

openedx/core/djangoapps/user_api/accounts/__init__.py

@@ -62,6 +62,9 @@ EMAIL_CONFLICT_MSG = _(
     "Try again with a different email address."
 )
 AUTHN_EMAIL_CONFLICT_MSG = _("This email is already associated with an existing or previous edX account")
+AUTHN_PASSWORD_COMPROMISED_MSG = _(
+    "The password you entered is on a list of known compromised passwords. Please choose a different one."
+)
 USERNAME_CONFLICT_MSG = _(
     "It looks like {username} belongs to an existing account. "
     "Try again with a different username."

openedx/core/djangoapps/user_api/accounts/api.py

@@ -32,6 +32,7 @@ from openedx.core.djangoapps.user_api.errors import (
     PreferenceValidationError
 )
 from openedx.core.djangoapps.user_api.preferences.api import update_user_preferences
+from openedx.core.djangoapps.user_authn.utils import check_pwned_password
 from openedx.core.djangoapps.user_authn.views.registration_form import validate_name, validate_username
 from openedx.core.lib.api.view_utils import add_serializer_errors
 from openedx.features.enterprise_support.utils import get_enterprise_readonly_account_fields
@@ -433,17 +434,18 @@ def get_confirm_email_validation_error(confirm_email, email):
     return _validate(_validate_confirm_email, errors.AccountEmailInvalid, confirm_email, email)
 
 
-def get_password_validation_error(password, username=None, email=None):
+def get_password_validation_error(password, username=None, email=None, reset_password_page=False):
     """Get the built-in validation error message for when
     the password is invalid in some way.
 
     :param password: The proposed password (unicode).
     :param username: The username associated with the user's account (unicode).
     :param email: The email associated with the user's account (unicode).
+    :param reset_password_page: The flag that determines the validation page (bool).
     :return: Validation error message.
 
     """
-    return _validate(_validate_password, errors.AccountPasswordInvalid, password, username, email)
+    return _validate(_validate_password, errors.AccountPasswordInvalid, password, username, email, reset_password_page)
 
 
 def get_country_validation_error(country):
@@ -587,7 +589,7 @@ def _validate_confirm_email(confirm_email, email):
         raise errors.AccountEmailInvalid(accounts.REQUIRED_FIELD_CONFIRM_EMAIL_MSG)
 
 
-def _validate_password(password, username=None, email=None):
+def _validate_password(password, username=None, email=None, reset_password_page=False):
     """Validate the format of the user's password.
 
     Passwords cannot be the same as the username of the account,
@@ -598,6 +600,7 @@ def _validate_password(password, username=None, email=None):
         password (unicode): The proposed password.
         username (unicode): The username associated with the user's account.
         email (unicode): The email associated with the user's account.
+        reset_password_page (bool): The flag that determines the validation page.
 
     Returns:
         None
@@ -615,6 +618,13 @@ def _validate_password(password, username=None, email=None):
     except ValidationError as validation_err:
         raise errors.AccountPasswordInvalid(' '.join(validation_err.messages))
 
+    # TODO: VAN-666 - Restrict this feature to reset password page for now until it is
+    #  enabled on account sign in and register.
+    if settings.ENABLE_AUTHN_RESET_PASSWORD_HIBP_POLICY and reset_password_page:
+        pwned_response = check_pwned_password(password)
+        if pwned_response.get('vulnerability', 'no') == 'yes':
+            raise errors.AccountPasswordInvalid(accounts.AUTHN_PASSWORD_COMPROMISED_MSG)
+
 
 def _validate_country(country):
     """Validate the country selection.

openedx/core/djangoapps/user_authn/tasks.py

@@ -2,9 +2,7 @@
 This file contains celery tasks for sending email
 """
 
-import hashlib
 import logging
-import math
 
 from celery import shared_task
 from celery.exceptions import MaxRetriesExceededError
@@ -15,47 +13,26 @@ from edx_ace import ace
 from edx_ace.errors import RecoverableChannelDeliveryError
 from edx_ace.message import Message
 from edx_django_utils.monitoring import set_code_owner_attribute
-from rest_framework.status import HTTP_408_REQUEST_TIMEOUT
 
 from common.djangoapps.track import segment
-from openedx.core.djangoapps.password_policy.hibp import PwnedPasswordsAPI
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
+from openedx.core.djangoapps.user_authn.utils import check_pwned_password
 from openedx.core.lib.celery.task_utils import emulate_http_request
 
 log = logging.getLogger('edx.celery.task')
 
 
-def get_pwned_properties(pwned_response, password):
-    """
-    Derive different pwned parameters for analytics
-    """
-    properties = {}
-    if pwned_response == HTTP_408_REQUEST_TIMEOUT:
-        properties['vulnerability'] = 'unknown'
-        pwned_count = 0
-    else:
-        pwned_count = pwned_response.get(password, 0)
-        properties['vulnerability'] = 'yes' if pwned_count > 0 else 'no'
-
-    if pwned_count > 0:
-        properties['frequency'] = math.ceil(math.log10(pwned_count))
-
-    return properties
-
-
 @shared_task
 @set_code_owner_attribute
 def check_pwned_password_and_send_track_event(user_id, password, internal_user=False):
     """
-    Check the Pwned Databases and send its event to Segment
+    Check the Pwned Databases and send its event to Segment.
     """
     try:
-        password = hashlib.sha1(password.encode('utf-8')).hexdigest()
-        pwned_response = PwnedPasswordsAPI.range(password)
-        if pwned_response is not None:
-            properties = get_pwned_properties(pwned_response, password)
-            properties['internal_user'] = internal_user
-            segment.track(user_id, 'edx.bi.user.pwned.password.status', properties)
+        pwned_properties = check_pwned_password(password)
+        if pwned_properties:
+            pwned_properties['internal_user'] = internal_user
+            segment.track(user_id, 'edx.bi.user.pwned.password.status', pwned_properties)
     except Exception:  # pylint: disable=W0703
         log.exception(
             'Unable to get response from pwned password api for user_id: "%s"',

openedx/core/djangoapps/user_authn/utils.py

@@ -2,6 +2,8 @@
 Utility functions used during user authentication.
 """
 
+import hashlib
+import math
 import random
 import re
 from urllib.parse import urlparse  # pylint: disable=import-error
@@ -10,8 +12,10 @@ from uuid import uuid4  # lint-amnesty, pylint: disable=unused-import
 from django.conf import settings
 from django.utils import http
 from oauth2_provider.models import Application
+from rest_framework.status import HTTP_408_REQUEST_TIMEOUT
 
 from common.djangoapps.student.models import username_exists_or_retired
+from openedx.core.djangoapps.password_policy.hibp import PwnedPasswordsAPI
 from openedx.core.djangoapps.user_api.accounts import USERNAME_MAX_LENGTH
 
 
@@ -105,3 +109,28 @@ def generate_username_suggestions(name):
                     break
 
     return username_suggestions
+
+
+def check_pwned_password(password):
+    """
+        Check the Pwned Databases for vulnerable passwords.
+        check_pwned_password returns password hash suffix and a dictionary containing
+        suffix of every SHA-1 password hash beginning with the specified prefix,
+        followed by a count of how many times it appears in their data set.
+    """
+    properties = {}
+    password = hashlib.sha1(password.encode('utf-8')).hexdigest().upper()
+
+    pwned_response = PwnedPasswordsAPI.range(password)
+    if pwned_response is not None:
+        if pwned_response == HTTP_408_REQUEST_TIMEOUT:
+            properties['vulnerability'] = 'unknown'
+            pwned_count = 0
+        else:
+            pwned_count = pwned_response.get(password[5:], 0)
+            properties['vulnerability'] = 'yes' if pwned_count > 0 else 'no'
+
+        if pwned_count > 0:
+            properties['frequency'] = math.ceil(math.log10(pwned_count))
+
+    return properties

openedx/core/djangoapps/user_authn/views/password_reset.py

@@ -42,6 +42,7 @@ from openedx.core.djangoapps.user_api.helpers import FormDescription
 from openedx.core.djangoapps.user_api.models import UserRetirementRequest
 from openedx.core.djangoapps.user_api.preferences.api import get_user_preference
 from openedx.core.djangoapps.user_authn.message_types import PasswordReset, PasswordResetSuccess
+from openedx.core.djangoapps.user_authn.utils import check_pwned_password
 from openedx.core.djangolib.markup import HTML
 from common.djangoapps.student.forms import send_account_recovery_email_for_user
 from common.djangoapps.student.models import AccountRecovery, LoginFailures
@@ -745,6 +746,17 @@ class LogistrationPasswordResetView(APIView):  # lint-amnesty, pylint: disable=m
                 return Response({'reset_status': reset_status})
 
             validate_password(password, user=user)
+
+            if settings.ENABLE_AUTHN_RESET_PASSWORD_HIBP_POLICY:
+                # Checks the Pwned Databases for password vulnerability.
+                pwned_response = check_pwned_password(password)
+                if pwned_response.get('vulnerability', 'no') == 'yes':
+                    error_status = {
+                        'reset_status': reset_status,
+                        'err_msg': accounts.AUTHN_PASSWORD_COMPROMISED_MSG
+                    }
+                    return Response(error_status)
+
             form = SetPasswordForm(user, request.data)
             if form.is_valid():
                 form.save()

openedx/core/djangoapps/user_authn/views/register.py

@@ -285,14 +285,14 @@ def create_account_with_params(request, params):  # pylint: disable=too-many-sta
 
     # TODO: there is no error checking here to see that the user actually logged in successfully,
     # and is not yet an active user.
-    is_new_user(request, new_user)
+    is_new_user(form.cleaned_data['password'], new_user)
     return new_user
 
 
-def is_new_user(request, user):
+def is_new_user(password, user):
     if user is not None:
         AUDIT_LOG.info(f"Login success on new account creation - {user.username}")
-        check_pwned_password_and_send_track_event.delay(user.id, request.POST.get('password'), user.is_staff)
+        check_pwned_password_and_send_track_event.delay(user.id, password, user.is_staff)
 
 
 def _link_user_to_third_party_provider(
@@ -820,7 +820,8 @@ class RegistrationValidationView(APIView):
         username = request.data.get('username')
         email = request.data.get('email')
         password = request.data.get('password')
-        return get_password_validation_error(password, username, email)
+        reset_password_page = request.data.get('reset_password_page', 'false') == 'true'
+        return get_password_validation_error(password, username, email, reset_password_page)
 
     def country_handler(self, request):
         """ Country validator """