diff --git a/lms/djangoapps/lti_provider/migrations/0004_require_user_account.py b/lms/djangoapps/lti_provider/migrations/0004_require_user_account.py
new file mode 100644
index 000000000000..e6238e312da0
--- /dev/null
+++ b/lms/djangoapps/lti_provider/migrations/0004_require_user_account.py
@@ -0,0 +1,26 @@
+# Generated by Django 3.2.22 on 2023-11-06 09:47
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('lti_provider', '0003_auto_20161118_1040'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='lticonsumer',
+            name='require_user_account',
+            field=models.BooleanField(blank=True, default=False, help_text='When checked, the LTI content will load only for learners who have an account in this instance. This is required only for linking learner accounts with the LTI consumer. See the Open edX LTI Provider documentation for more details.'),
+        ),
+        migrations.AlterField(
+            model_name='ltiuser',
+            name='edx_user',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL),
+        ),
+    ]
diff --git a/lms/djangoapps/lti_provider/models.py b/lms/djangoapps/lti_provider/models.py
index 0cb165eeef69..94582d4bf0b4 100644
--- a/lms/djangoapps/lti_provider/models.py
+++ b/lms/djangoapps/lti_provider/models.py
@@ -14,6 +14,7 @@
 
 from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
 from django.db import models
+from django.utils.translation import gettext as _
 from opaque_keys.edx.django.models import CourseKeyField, UsageKeyField
 
 from openedx.core.djangolib.fields import CharNullField
@@ -34,6 +35,11 @@ class LtiConsumer(models.Model):
     consumer_key = models.CharField(max_length=32, unique=True, db_index=True, default=short_token)
     consumer_secret = models.CharField(max_length=32, unique=True, default=short_token)
     instance_guid = CharNullField(max_length=255, blank=True, null=True, unique=True)
+    require_user_account = models.BooleanField(blank=True, default=False, help_text=_(
+        "When checked, the LTI content will load only for learners who have an account "
+        "in this instance. This is required only for linking learner accounts with "
+        "the LTI consumer. See the Open edX LTI Provider documentation for more details."
+    ))
 
     @staticmethod
     def get_or_supplement(instance_guid, consumer_key):
@@ -140,7 +146,7 @@ class LtiUser(models.Model):
     """
     lti_consumer = models.ForeignKey(LtiConsumer, on_delete=models.CASCADE)
     lti_user_id = models.CharField(max_length=255)
-    edx_user = models.OneToOneField(User, on_delete=models.CASCADE)
+    edx_user = models.ForeignKey(User, on_delete=models.CASCADE)
 
     class Meta:
         unique_together = ('lti_consumer', 'lti_user_id')
diff --git a/lms/djangoapps/lti_provider/tests/test_users.py b/lms/djangoapps/lti_provider/tests/test_users.py
index d945f6b9a65a..1d19d8995be9 100644
--- a/lms/djangoapps/lti_provider/tests/test_users.py
+++ b/lms/djangoapps/lti_provider/tests/test_users.py
@@ -7,8 +7,9 @@
 from unittest.mock import MagicMock, PropertyMock, patch
 
 import pytest
-from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
+from django.contrib.auth.models import AnonymousUser, User  # lint-amnesty, pylint: disable=imported-auth-user
 from django.core.exceptions import PermissionDenied
+from django.db.utils import IntegrityError
 from django.test import TestCase
 from django.test.client import RequestFactory
 
@@ -92,15 +93,22 @@ def setUp(self):
         self.old_user = UserFactory.create()
         self.request = RequestFactory().post('/')
         self.request.user = self.old_user
+        self.auto_linking_consumer = LtiConsumer(
+            consumer_name='AutoLinkingConsumer',
+            consumer_key='AutoLinkingKey',
+            consumer_secret='AutoLinkingSecret',
+            require_user_account=True
+        )
+        self.auto_linking_consumer.save()
 
-    def create_lti_user_model(self):
+    def create_lti_user_model(self, consumer=None):
         """
         Generate and save a User and an LTI user model
         """
         edx_user = User(username=self.edx_user_id)
         edx_user.save()
         lti_user = LtiUser(
-            lti_consumer=self.lti_consumer,
+            lti_consumer=consumer or self.lti_consumer,
             lti_user_id=self.lti_user_id,
             edx_user=edx_user
         )
@@ -140,6 +148,38 @@ def test_authentication_with_wrong_user(self, create_user, switch_user):
         assert not create_user.called
         switch_user.assert_called_with(self.request, lti_user, self.lti_consumer)
 
+    def test_auto_linking_of_users_using_lis_person_contact_email_primary(self, create_user, switch_user):
+        request = RequestFactory().post("/", {"lis_person_contact_email_primary": self.old_user.email})
+        request.user = self.old_user
+
+        users.authenticate_lti_user(request, self.lti_user_id, self.lti_consumer)
+        create_user.assert_called_with(self.lti_user_id, self.lti_consumer)
+
+        users.authenticate_lti_user(request, self.lti_user_id, self.auto_linking_consumer)
+        create_user.assert_called_with(self.lti_user_id, self.auto_linking_consumer, self.old_user.email)
+
+    def test_raise_exception_trying_to_auto_link_unauthenticate_user(self, create_user, switch_user):
+        request = RequestFactory().post("/")
+        request.user = AnonymousUser()
+
+        with self.assertRaises(PermissionDenied):
+            users.authenticate_lti_user(request, self.lti_user_id, self.auto_linking_consumer)
+
+    def test_raise_exception_on_mismatched_user_and_lis_email(self, create_user, switch_user):
+        request = RequestFactory().post("/", {"lis_person_contact_email_primary": "wrong_email@example.com"})
+        request.user = self.old_user
+
+        with self.assertRaises(PermissionDenied):
+            users.authenticate_lti_user(request, self.lti_user_id, self.auto_linking_consumer)
+
+    def test_authenticate_unauthenticated_user_after_auto_linking_of_user_account(self, create_user, switch_user):
+        lti_user = self.create_lti_user_model(self.auto_linking_consumer)
+        self.request.user = AnonymousUser()
+
+        users.authenticate_lti_user(self.request, self.lti_user_id, self.auto_linking_consumer)
+        assert not create_user.called
+        switch_user.assert_called_with(self.request, lti_user, self.auto_linking_consumer)
+
 
 class CreateLtiUserTest(TestCase):
     """
@@ -154,16 +194,17 @@ def setUp(self):
             consumer_secret='TestSecret'
         )
         self.lti_consumer.save()
+        self.existing_user = UserFactory.create()
 
     def test_create_lti_user_creates_auth_user_model(self):
         users.create_lti_user('lti_user_id', self.lti_consumer)
-        assert User.objects.count() == 1
+        assert User.objects.count() == 2
 
     @patch('uuid.uuid4', return_value='random_uuid')
     @patch('lms.djangoapps.lti_provider.users.generate_random_edx_username', return_value='edx_id')
     def test_create_lti_user_creates_correct_user(self, uuid_mock, _username_mock):
         users.create_lti_user('lti_user_id', self.lti_consumer)
-        assert User.objects.count() == 1
+        assert User.objects.count() == 2
         user = User.objects.get(username='edx_id')
         assert user.email == 'edx_id@lti.example.com'
         uuid_mock.assert_called_with()
@@ -173,10 +214,34 @@ def test_unique_username_created(self, username_mock):
         User(username='edx_id').save()
         users.create_lti_user('lti_user_id', self.lti_consumer)
         assert username_mock.call_count == 2
-        assert User.objects.count() == 2
+        assert User.objects.count() == 3
         user = User.objects.get(username='new_edx_id')
         assert user.email == 'new_edx_id@lti.example.com'
 
+    def test_existing_user_is_linked(self):
+        lti_user = users.create_lti_user('lti_user_id', self.lti_consumer, self.existing_user.email)
+        assert lti_user.lti_consumer == self.lti_consumer
+        assert lti_user.edx_user == self.existing_user
+
+    def test_only_one_lti_user_edx_user_for_each_lti_consumer(self):
+        users.create_lti_user('lti_user_id', self.lti_consumer, self.existing_user.email)
+
+        with pytest.raises(IntegrityError):
+            users.create_lti_user('lti_user_id', self.lti_consumer, self.existing_user.email)
+
+    def test_create_multiple_lti_users_for_edx_user_if_lti_consumer_varies(self):
+        lti_consumer_2 = LtiConsumer(
+            consumer_name="SecondConsumer",
+            consumer_key="SecondKey",
+            consumer_secret="SecondSecret",
+        )
+        lti_consumer_2.save()
+
+        lti_user_1 = users.create_lti_user('lti_user_id', self.lti_consumer, self.existing_user.email)
+        lti_user_2 = users.create_lti_user('lti_user_id', lti_consumer_2, self.existing_user.email)
+
+        assert lti_user_1.edx_user == lti_user_2.edx_user
+
 
 class LtiBackendTest(TestCase):
     """
diff --git a/lms/djangoapps/lti_provider/tests/test_views.py b/lms/djangoapps/lti_provider/tests/test_views.py
index ebb7eda154d6..8328c669cf5a 100644
--- a/lms/djangoapps/lti_provider/tests/test_views.py
+++ b/lms/djangoapps/lti_provider/tests/test_views.py
@@ -5,6 +5,7 @@
 
 from unittest.mock import MagicMock, patch
 
+from django.contrib.auth.models import AnonymousUser
 from django.test import TestCase
 from django.test.client import RequestFactory
 from django.urls import reverse
@@ -58,7 +59,7 @@ def build_launch_request(extra_post_data=None, param_to_delete=None):
         del post_data[param_to_delete]
     request = RequestFactory().post('/', data=post_data)
     request.user = UserFactory.create()
-    request.session = {}
+    request.session = MagicMock()
     return request
 
 
@@ -82,6 +83,14 @@ def setUp(self):
         )
         self.consumer.save()
 
+        self.auto_link_consumer = models.LtiConsumer(
+            consumer_name='auto-link-consumer',
+            consumer_key='consumer_key_2',
+            consumer_secret='secret_2',
+            require_user_account=True
+        )
+        self.auto_link_consumer.save()
+
 
 class LtiLaunchTest(LtiTestMixin, TestCase):
     """
@@ -189,6 +198,36 @@ def test_lti_consumer_record_supplemented_with_guid(self, _render):
         )
         assert consumer.instance_guid == 'consumer instance guid'
 
+    @patch('lms.djangoapps.lti_provider.views.render_to_response')
+    def test_unauthenticated_user_shown_error_when_require_user_account_is_enabled(self, render_error):
+        """
+        Verify that an error page is shown instead of LTI Content for an unauthenticated user,
+        when the `require_user_account` flag is enabled for the LTI Consumer.
+        """
+        request = build_launch_request({'oauth_consumer_key': 'consumer_key_2'})
+        request.user = AnonymousUser()
+
+        views.lti_launch(request, str(COURSE_KEY), str(USAGE_KEY))
+
+        render_error.assert_called()
+        assert render_error.call_args[0][0] == "lti_provider/user-auth-error.html"
+
+    @patch('lms.djangoapps.lti_provider.views.render_to_response')
+    def test_auth_error_shown_when_lis_email_is_different_from_user_email(self, render_error):
+        """
+        When the `require_user_account` flag is enabled for the LTI Consumer, verify that
+        an error page is shown instead of LTI Content if the authenticated user's email
+        doesn't match the `lis_person_contact_email_primary` value from LTI Launch.
+        """
+        # lis email different from logged in user
+        request = build_launch_request({
+            'oauth_consumer_key': 'consumer_key_2',
+            'lis_person_contact_email_primary': 'random_email@test.com'
+        })
+
+        views.lti_launch(request, str(COURSE_KEY), str(USAGE_KEY))
+        render_error.assert_called()
+
 
 class LtiLaunchTestRender(LtiTestMixin, RenderXBlockTestMixin, ModuleStoreTestCase):
     """
diff --git a/lms/djangoapps/lti_provider/users.py b/lms/djangoapps/lti_provider/users.py
index 7c5a9c88ebaa..c2373522b960 100644
--- a/lms/djangoapps/lti_provider/users.py
+++ b/lms/djangoapps/lti_provider/users.py
@@ -28,14 +28,25 @@ def authenticate_lti_user(request, lti_user_id, lti_consumer):
     If the currently logged-in user does not match the user specified by the LTI
     launch, log out the old user and log in the LTI identity.
     """
+    lis_email = request.POST.get("lis_person_contact_email_primary")
+
     try:
         lti_user = LtiUser.objects.get(
             lti_user_id=lti_user_id,
             lti_consumer=lti_consumer
         )
-    except LtiUser.DoesNotExist:
+    except LtiUser.DoesNotExist as exc:
         # This is the first time that the user has been here. Create an account.
-        lti_user = create_lti_user(lti_user_id, lti_consumer)
+        if lti_consumer.require_user_account:
+            # Verify that the email from the LTI Launch and the logged-in user are the same
+            # before linking the LtiUser with the edx_user.
+            if request.user.is_authenticated and request.user.email == lis_email:
+                lti_user = create_lti_user(lti_user_id, lti_consumer, lis_email)
+            else:
+                # Ask the user to login before linking.
+                raise PermissionDenied() from exc
+        else:
+            lti_user = create_lti_user(lti_user_id, lti_consumer)
 
     if not (request.user.is_authenticated and
             request.user == lti_user.edx_user):
@@ -44,34 +55,36 @@ def authenticate_lti_user(request, lti_user_id, lti_consumer):
         switch_user(request, lti_user, lti_consumer)
 
 
-def create_lti_user(lti_user_id, lti_consumer):
+def create_lti_user(lti_user_id, lti_consumer, email=None):
     """
     Generate a new user on the edX platform with a random username and password,
     and associates that account with the LTI identity.
     """
-    edx_password = str(uuid.uuid4())
+    edx_user = User.objects.filter(email=email).first() if email else None
 
-    created = False
-    while not created:
-        try:
-            edx_user_id = generate_random_edx_username()
-            edx_email = f"{edx_user_id}@{settings.LTI_USER_EMAIL_DOMAIN}"
-            with transaction.atomic():
-                edx_user = User.objects.create_user(
-                    username=edx_user_id,
-                    password=edx_password,
-                    email=edx_email,
-                )
-                # A profile is required if PREVENT_CONCURRENT_LOGINS flag is set.
-                # TODO: We could populate user information from the LTI launch here,
-                # but it's not necessary for our current uses.
-                edx_user_profile = UserProfile(user=edx_user)
-                edx_user_profile.save()
-            created = True
-        except IntegrityError:
-            # The random edx_user_id wasn't unique. Since 'created' is still
-            # False, we will retry with a different random ID.
-            pass
+    if not edx_user:
+        created = False
+        edx_password = str(uuid.uuid4())
+        while not created:
+            try:
+                edx_user_id = generate_random_edx_username()
+                edx_email = f"{edx_user_id}@{settings.LTI_USER_EMAIL_DOMAIN}"
+                with transaction.atomic():
+                    edx_user = User.objects.create_user(
+                        username=edx_user_id,
+                        password=edx_password,
+                        email=edx_email,
+                    )
+                    # A profile is required if PREVENT_CONCURRENT_LOGINS flag is set.
+                    # TODO: We could populate user information from the LTI launch here,
+                    # but it's not necessary for our current uses.
+                    edx_user_profile = UserProfile(user=edx_user)
+                    edx_user_profile.save()
+                    created = True
+            except IntegrityError:
+                # The random edx_user_id wasn't unique. Since 'created' is still
+                # False, we will retry with a different random ID.
+                pass
 
     lti_user = LtiUser(
         lti_consumer=lti_consumer,
diff --git a/lms/djangoapps/lti_provider/views.py b/lms/djangoapps/lti_provider/views.py
index ea2102e41893..1106908112ed 100644
--- a/lms/djangoapps/lti_provider/views.py
+++ b/lms/djangoapps/lti_provider/views.py
@@ -6,8 +6,10 @@
 import logging
 
 from django.conf import settings
+from django.core.exceptions import PermissionDenied
 from django.http import Http404, HttpResponseBadRequest, HttpResponseForbidden
 from django.views.decorators.csrf import csrf_exempt
+from common.djangoapps.edxmako.shortcuts import render_to_response
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey, UsageKey
 
@@ -88,7 +90,17 @@ def lti_launch(request, course_id, usage_id):
 
     # Create an edX account if the user identifed by the LTI launch doesn't have
     # one already, and log the edX account into the platform.
-    authenticate_lti_user(request, params['user_id'], lti_consumer)
+    try:
+        authenticate_lti_user(request, params['user_id'], lti_consumer)
+    except PermissionDenied:
+        request.session.flush()
+        context = {
+            "login_link": request.build_absolute_uri(settings.LOGIN_URL),
+            "allow_iframing": True,
+            "disable_header": True,
+            "disable_footer": True,
+        }
+        return render_to_response("lti_provider/user-auth-error.html", context)
 
     # Store any parameters required by the outcome service in order to report
     # scores back later. We know that the consumer exists, since the record was
diff --git a/lms/templates/lti_provider/user-auth-error.html b/lms/templates/lti_provider/user-auth-error.html
new file mode 100644
index 000000000000..a807e49f56b1
--- /dev/null
+++ b/lms/templates/lti_provider/user-auth-error.html
@@ -0,0 +1,32 @@
+<%page expression_filter="h" />
+<%namespace name='static' file='../static_content.html'/>
+<%!
+from django.utils.translation import gettext as _
+from openedx.core.djangolib.markup import HTML, Text
+%>
+<%inherit file="../main.html" />
+
+<section class="outside-app">
+  <h1>
+    ${Text(_("There was an error when loading this module!"))}
+  </h1>
+  <p>
+    ${Text(_("This module is available only for users who are signed into {platform}. "
+        "Kindly sign in and refresh this page to load the content."
+      )).format(
+        platform=HTML("<em>{}</em>").format(static.get_platform_name())
+      )}
+  </p>
+  <p class="my-3">
+    <strong>${_("NOTE:")}</strong>
+    ${Text(_(
+      "The email used to sign into this platform and {platform} should be the same."
+    )).format(
+      platform=HTML("<em>{}</em>").format(static.get_platform_name())
+    )}
+  </p>
+
+  <p class="mt-3">
+    <a class="btn btn-primary" href="${login_link}" target="_blank">${_("Sign in")}</a>
+  </p>
+</section>