feat(AUT-266): enforce fail-fast allowlist validation and prevent raw input reflection

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

@@ -8,6 +8,7 @@
 import json
 import logging
 
+from django.core.exceptions import ValidationError
 from django.conf import settings
 from django.contrib.auth import authenticate, get_user_model
 from django.contrib.auth import login as django_login
@@ -17,6 +18,7 @@
 from django.urls import reverse
 from django.utils.decorators import method_decorator
 from django.utils.translation import gettext as _
+from django.core.validators import validate_email
 from django.views.decorators.csrf import csrf_exempt, csrf_protect, ensure_csrf_cookie
 from django.views.decorators.debug import sensitive_post_parameters
 from django.views.decorators.http import require_http_methods
@@ -49,6 +51,7 @@
 from openedx.core.djangoapps.user_authn.cookies import get_response_with_refreshed_jwt_cookies, set_logged_in_cookies
 from openedx.core.djangoapps.user_authn.exceptions import AuthFailedError, VulnerablePasswordError
 from openedx.core.djangoapps.user_authn.tasks import check_pwned_password_and_send_track_event
+from openedx.core.djangoapps.user_authn.views.registration_form import validate_username
 from openedx.core.djangoapps.user_authn.toggles import (
     is_require_third_party_auth_enabled,
     should_redirect_to_authn_microfrontend,
@@ -64,6 +67,8 @@
 AUDIT_LOG = logging.getLogger("audit")
 USER_MODEL = get_user_model()
 PASSWORD_RESET_INITIATED = "edx.user.passwordreset.initiated"
+LOGIN_INFO_ERROR = _("There was an error receiving your login information. Please email us.")
+LOGIN_INVALID_INPUT_EMAIL = "[invalid input]"
 
 
 def _do_third_party_auth(request):
@@ -123,6 +128,26 @@ def _get_user_by_username(username):
         return None
 
 
+def _validate_login_identifier(identifier):
+    """
+    Validate the submitted login identifier before any auth logic runs.
+    """
+    if "@" in identifier:
+        try:
+            validate_email(identifier)
+        except ValidationError as exc:
+            raise AuthFailedError(LOGIN_INFO_ERROR) from exc
+        return
+
+    if not accounts.USERNAME_MIN_LENGTH <= len(identifier) <= accounts.USERNAME_MAX_LENGTH:
+        raise AuthFailedError(LOGIN_INFO_ERROR)
+
+    try:
+        validate_username(identifier)
+    except ValidationError as exc:
+        raise AuthFailedError(LOGIN_INFO_ERROR) from exc
+
+
 def _get_user_by_email_or_username(request, api_version):
     """
     Finds a user object in the database based on the given request, ignores all fields except for email and username.
@@ -133,7 +158,7 @@ def _get_user_by_email_or_username(request, api_version):
         login_fields = ["email_or_username", "password"]
 
     if any(f not in request.POST.keys() for f in login_fields):
-        raise AuthFailedError(_("There was an error receiving your login information. Please email us."))
+        raise AuthFailedError(LOGIN_INFO_ERROR)
 
     email_or_username = request.POST.get("email", None) or request.POST.get("email_or_username", None)
     user = _get_user_by_email(email_or_username)
@@ -567,15 +592,21 @@ def login_user(request, api_version="v1"):  # pylint: disable=too-many-statement
     third_party_auth_requested = third_party_auth.is_enabled() and pipeline.running(request)
     first_party_auth_requested = any(bool(request.POST.get(p)) for p in ["email", "email_or_username", "password"])
     is_user_third_party_authenticated = False
+    user = None
 
     set_custom_attribute("login_user_course_id", request.POST.get("course_id"))
 
     if is_require_third_party_auth_enabled() and not third_party_auth_requested:
         return HttpResponseForbidden(
             "Third party authentication is required to login. Username and password were received instead."
         )
-    possibly_authenticated_user = None
     try:
+        login_identifier = request.POST.get("email_or_username")
+        if login_identifier is None:
+            login_identifier = request.POST.get("email")
+        if login_identifier is not None:
+            _validate_login_identifier(login_identifier)
+
         if third_party_auth_requested and not first_party_auth_requested:
             # The user has already authenticated via third-party auth and has not
             # asked to do first party auth by supplying a username or password. We
@@ -693,10 +724,7 @@ def login_user(request, api_version="v1"):  # pylint: disable=too-many-statement
         error_code = response_content.get("error_code")
         if error_code:
             set_custom_attribute("login_error_code", error_code)
-        email_or_username_key = "email" if api_version == API_V1 else "email_or_username"
-        email_or_username = request.POST.get(email_or_username_key, None)
-        email_or_username = possibly_authenticated_user.email if possibly_authenticated_user else email_or_username
-        response_content["email"] = email_or_username
+        response_content["email"] = user.email if user else LOGIN_INVALID_INPUT_EMAIL
     except VulnerablePasswordError as error:
         response_content = error.get_response()
         log.exception(response_content)

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

@@ -63,7 +63,10 @@
     RegistrationFormFactory,
     get_registration_extension_form
 )
-from openedx.core.djangoapps.user_authn.views.utils import get_auto_generated_username
+from openedx.core.djangoapps.user_authn.views.utils import (
+    get_auto_generated_username,
+    get_total_registration_time_validation_error,
+)
 from openedx.core.djangoapps.user_authn.tasks import check_pwned_password_and_send_track_event
 from openedx.core.djangoapps.user_authn.toggles import (
     is_require_third_party_auth_enabled,
@@ -587,6 +590,20 @@ def post(self, request):
             )
 
         data = request.POST.copy()
+
+        validation_error = get_total_registration_time_validation_error(data)
+        if validation_error:
+            AUDIT_LOG.warning(
+                "Registration rejected due to invalid total_registration_time payload digest=%s",
+                validation_error["value_digest"],
+            )
+            return self._create_response(
+                request,
+                {validation_error["field"]: [{"user_message": validation_error["user_message"]}]},
+                status_code=400,
+                error_code=validation_error["error_code"],
+            )
+
         self._handle_terms_of_service(data)
 
         if is_auto_generated_username_enabled() and 'username' not in data:

openedx/core/djangoapps/user_authn/views/tests/test_login.py

@@ -245,6 +245,7 @@ def test_login_fail_no_user_exists(self):
         self._assert_response(
             response, success=False, value=self.LOGIN_FAILED_WARNING, status_code=400
         )
+        self._assert_response(response, success=False, email_value="[invalid input]")
         self._assert_audit_log(mock_audit_log, 'warning', ['Login failed', 'Unknown user email', email_hash])
 
     @patch.dict("django.conf.settings.FEATURES", {'SQUELCH_PII_IN_LOGS': True})
@@ -255,6 +256,7 @@ def test_login_fail_no_user_exists_no_pii(self):
             self.password,
         )
         self._assert_response(response, success=False, value=self.LOGIN_FAILED_WARNING)
+        self._assert_response(response, success=False, email_value="[invalid input]")
         self._assert_audit_log(mock_audit_log, 'warning', ['Login failed', 'Unknown user email'])
         self._assert_not_in_audit_log(mock_audit_log, 'warning', [nonexistent_email])
 
@@ -264,13 +266,15 @@ def test_login_fail_wrong_password(self):
             'wrong_password',
         )
         self._assert_response(response, success=False, value=self.LOGIN_FAILED_WARNING)
+        self._assert_response(response, success=False, email_value=self.user_email)
         self._assert_audit_log(mock_audit_log, 'warning',
                                ['Login failed', 'password for', str(self.user.id), 'invalid'])
 
     @patch.dict("django.conf.settings.FEATURES", {'SQUELCH_PII_IN_LOGS': True})
     def test_login_fail_wrong_password_no_pii(self):
         response, mock_audit_log = self._login_response(self.user_email, 'wrong_password')
         self._assert_response(response, success=False, value=self.LOGIN_FAILED_WARNING)
+        self._assert_response(response, success=False, email_value=self.user_email)
         self._assert_audit_log(
             mock_audit_log, 'warning', ['Login failed', 'password for', str(self.user.id), 'invalid']
         )
@@ -324,6 +328,7 @@ def test_login_not_activated_no_pii(self):
             self.password
         )
         self._assert_response(response, success=False, error_code="inactive-user")
+        self._assert_response(response, success=False, email_value=self.user_email)
         self._assert_audit_log(mock_audit_log, 'warning', ['Login failed', 'Account not active for user'])
         self._assert_not_in_audit_log(mock_audit_log, 'warning', ['test'])
 
@@ -711,7 +716,16 @@ def _login_response(
             result = self.client.post(self.url, post_params, **extra)
         return result, mock_audit_log
 
-    def _assert_response(self, response, success=None, value=None, status_code=None, error_code=None):
+    def _assert_response(
+        self,
+        response,
+        success=None,
+        value=None,
+        status_code=None,
+        error_code=None,
+        email_value=None,
+        absent_keys=None,
+    ):
         """
         Assert that the response has the expected status code and returned a valid
         JSON-parseable dict.
@@ -742,6 +756,13 @@ def _assert_response(self, response, success=None, value=None, status_code=None,
                    (str(response_dict['value']), str(value)))
             assert value in response_dict['value'], msg
 
+        if email_value is not None:
+            assert response_dict['email'] == email_value
+
+        if absent_keys is not None:
+            for key in absent_keys:
+                assert key not in response_dict
+
     def _assert_redirect_url(self, response, expected_redirect_url):
         """
         Assert that the redirect URL is in the response and has the expected value.
@@ -1081,13 +1102,32 @@ def test_invalid_credentials(self):
             "password": "invalid"
         })
         self.assertHttpBadRequest(response)
+        self._assert_response(response, success=False, email_value=self.EMAIL)
 
         # Invalid email address
         response = self.client.post(self.url, {
             "email": "invalid@example.com",
             "password": self.PASSWORD,
         })
         self.assertHttpBadRequest(response)
+        self._assert_response(response, success=False, email_value="[invalid input]")
+
+    @ddt.data(
+        "bad input!",
+        "ab",
+        "not-an-email@",
+        "",
+    )
+    @patch('openedx.core.djangoapps.user_authn.views.login._get_user_by_email_or_username')
+    def test_login_rejects_invalid_login_identifier(self, invalid_identifier, mock_user_lookup):
+        response = self.client.post(self.url_v2, {
+            "email_or_username": invalid_identifier,
+            "password": self.PASSWORD,
+        })
+
+        self.assertHttpBadRequest(response)
+        self._assert_response(response, success=False, email_value="[invalid input]")
+        assert not mock_user_lookup.called
 
     @ddt.data(True, False)
     def test_missing_login_params(self, is_api_v1):

openedx/core/djangoapps/user_authn/views/tests/test_register.py

@@ -1,5 +1,6 @@
 """Tests for account creation"""
 
+import hashlib
 import json
 from datetime import datetime
 from unittest import mock, skipIf, skipUnless
@@ -2092,6 +2093,47 @@ def setUp(self):  # pylint: disable=arguments-differ
         super(RegistrationViewTestV1, self).setUp()  # lint-amnesty, pylint: disable=bad-super-call
         self.url = reverse("user_api_registration_v2")
 
+    def test_register_invalid_total_registration_time(self):
+        invalid_total_registration_time = '57.664" AND "1"="1" --'
+        response = self.client.post(self.url, {
+            "email": self.EMAIL,
+            "name": self.NAME,
+            "username": self.USERNAME,
+            "password": self.PASSWORD,
+            "honor_code": "true",
+            "total_registration_time": invalid_total_registration_time,
+        })
+
+        self.assertHttpBadRequest(response)
+        response_json = json.loads(response.content.decode('utf-8'))
+        self.assertDictEqual(
+            response_json,
+            {
+                "total_registration_time": [{
+                    "user_message": "Enter a valid registration time value.",
+                }],
+                "error_code": "invalid-total-registration-time",
+            }
+        )
+
+        expected_digest = hashlib.shake_128(invalid_total_registration_time.encode("utf-8")).hexdigest(16)
+        with LogCapture() as logger:
+            self.client.post(self.url, {
+                "email": self.EMAIL,
+                "name": self.NAME,
+                "username": "another_user",
+                "password": self.PASSWORD,
+                "honor_code": "true",
+                "total_registration_time": invalid_total_registration_time,
+            })
+            logger.check_present(
+                (
+                    'audit',
+                    'WARNING',
+                    f'Registration rejected due to invalid total_registration_time payload digest={expected_digest}'
+                )
+            )
+
     @override_settings(
         REGISTRATION_EXTRA_FIELDS={
             "level_of_education": "optional",

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

@@ -2,6 +2,7 @@
 User Auth Views Utils
 """
 import logging
+import hashlib
 import re
 from typing import Dict
 
@@ -26,6 +27,35 @@
 
 log = logging.getLogger(__name__)
 API_V1 = 'v1'
+UUID4_REGEX = '[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}'
+ENTERPRISE_ENROLLMENT_URL_REGEX = fr'/enterprise/{UUID4_REGEX}/course/{settings.COURSE_KEY_REGEX}/enroll'
+TOTAL_REGISTRATION_TIME_RE = re.compile(r'^\d+(?:\.\d+)?$')
+
+
+def get_total_registration_time_validation_error(data):
+    """
+    Validate optional registration timing telemetry with an allowlist pattern.
+
+    Returns:
+        dict | None: error details when invalid, otherwise None.
+    """
+    total_registration_time = data.get('total_registration_time')
+    if total_registration_time is None:
+        total_registration_time = data.get('totalRegistrationTime')
+
+    if total_registration_time in (None, ''):
+        return None
+
+    normalized_value = str(total_registration_time).strip()
+    if TOTAL_REGISTRATION_TIME_RE.fullmatch(normalized_value):
+        return None
+
+    return {
+        "field": "total_registration_time",
+        "user_message": _("Enter a valid registration time value."),
+        "error_code": "invalid-total-registration-time",
+        "value_digest": hashlib.shake_128(normalized_value.encode("utf-8")).hexdigest(16),
+    }
 
 
 def third_party_auth_context(request, redirect_to, tpa_hint=None):