diff --git a/.annotation_safe_list.yml b/.annotation_safe_list.yml
index 2a09ce7cb..67bd8f75e 100644
--- a/.annotation_safe_list.yml
+++ b/.annotation_safe_list.yml
@@ -21,6 +21,8 @@ catalog.HistoricalContentMetadata:
     ".. no_pii": "This model has no PII"
 catalog.HistoricalEnterpriseCatalog:
   ".. no_pii": "This model has no PII"
+catalog.HistoricalRestrictedCourseMetadata:
+    ".. no_pii": "This model has no PII"
 contenttypes.ContentType:
   ".. no_pii:": "This model has no PII"
 curation.HistoricalEnterpriseCurationConfig:
diff --git a/enterprise_catalog/apps/api/tasks.py b/enterprise_catalog/apps/api/tasks.py
index aa466bb85..2fef38b24 100644
--- a/enterprise_catalog/apps/api/tasks.py
+++ b/enterprise_catalog/apps/api/tasks.py
@@ -287,7 +287,7 @@ def _update_full_content_metadata_course(content_keys, dry_run=False):
         else:
             ContentMetadata.objects.bulk_update(
                 modified_content_metadata_records,
-                ['json_metadata'],
+                ['_json_metadata'],
                 batch_size=10,
             )
 
@@ -326,13 +326,14 @@ def _update_single_full_course_record(course_metadata_dict, metadata_record, cou
     for that course.
     """
     # Merge the full metadata from discovery's /api/v1/courses into the local metadata object.
-    metadata_record.json_metadata.update(course_metadata_dict)
+    metadata_record._json_metadata.update(course_metadata_dict)  # pylint: disable=protected-access
 
     _normalize_metadata_record(metadata_record)
 
     if course_review:
-        metadata_record.json_metadata['reviews_count'] = course_review.get('reviews_count')
-        metadata_record.json_metadata['avg_course_rating'] = course_review.get('avg_course_rating')
+        # pylint: disable=protected-access
+        metadata_record._json_metadata['reviews_count'] = course_review.get('reviews_count')
+        metadata_record._json_metadata['avg_course_rating'] = course_review.get('avg_course_rating')
 
     if metadata_record.json_metadata.get(FORCE_INCLUSION_METADATA_TAG_KEY):
         metadata_record.json_metadata = transform_course_metadata_to_visible(metadata_record.json_metadata)
@@ -358,11 +359,12 @@ def _normalize_metadata_record(course_metadata_record):
     normalized_metadata_input = {
         'course_metadata': course_metadata_record.json_metadata,
     }
-    course_metadata_record.json_metadata['normalized_metadata'] =\
+    # pylint: disable=protected-access
+    course_metadata_record._json_metadata['normalized_metadata'] =\
         NormalizedContentMetadataSerializer(normalized_metadata_input).data
-    course_metadata_record.json_metadata['normalized_metadata_by_run'] = {}
+    course_metadata_record._json_metadata['normalized_metadata_by_run'] = {}
     for run in course_metadata_record.json_metadata.get('course_runs', []):
-        course_metadata_record.json_metadata['normalized_metadata_by_run'].update({
+        course_metadata_record._json_metadata['normalized_metadata_by_run'].update({
             run['key']: NormalizedContentMetadataSerializer({
                 'course_run_metadata': run,
                 'course_metadata': course_metadata_record.json_metadata,
@@ -413,7 +415,7 @@ def _update_full_content_metadata_program(content_keys, dry_run=False):
                 logger.error('Could not find ContentMetadata record for content_key %s.', content_key)
                 continue
 
-            metadata_record.json_metadata.update(program_metadata_dict)
+            metadata_record._json_metadata.update(program_metadata_dict)  # pylint: disable=protected-access
             modified_content_metadata_records.append(metadata_record)
 
         if dry_run:
@@ -421,7 +423,7 @@ def _update_full_content_metadata_program(content_keys, dry_run=False):
         else:
             ContentMetadata.objects.bulk_update(
                 modified_content_metadata_records,
-                ['json_metadata'],
+                ['_json_metadata'],
                 batch_size=10,
             )
 
@@ -1200,7 +1202,12 @@ def fetch_missing_course_metadata_task(self, force=False, dry_run=False):  # pyl
     that are embedded inside a program.
     """
     logger.info('[FETCH_MISSING_METADATA] fetch_missing_course_metadata_task task started.')
-    program_metadata_list = ContentMetadata.objects.filter(content_type=PROGRAM).values_list('json_metadata', flat=True)
+    program_metadata_list = ContentMetadata.objects.filter(
+        content_type=PROGRAM
+    ).values_list(
+        '_json_metadata',
+        flat=True,
+    )
     course_keys = set()
     for program_metadata in program_metadata_list:
         if program_metadata is not None:
@@ -1266,7 +1273,7 @@ def fetch_missing_pathway_metadata_task(self, force=False, dry_run=False):  # py
     )
 
     learner_pathway_metadata_list = ContentMetadata.objects.filter(content_type=LEARNER_PATHWAY).values_list(
-        'json_metadata', flat=True,
+        '_json_metadata', flat=True,
     )
     program_uuids = set()
     course_keys = set()
diff --git a/enterprise_catalog/apps/api/tests/test_tasks.py b/enterprise_catalog/apps/api/tests/test_tasks.py
index 0dfac25bd..1505bfa37 100644
--- a/enterprise_catalog/apps/api/tests/test_tasks.py
+++ b/enterprise_catalog/apps/api/tests/test_tasks.py
@@ -223,7 +223,7 @@ def test_fetch_missing_course_metadata_task(self, mock_update_data_from_discover
         """
         test_course = 'course:edX+testX'
         course_content_metadata = ContentMetadataFactory.create(content_type=COURSE)
-        ContentMetadataFactory.create(content_type=PROGRAM, json_metadata={
+        ContentMetadataFactory.create(content_type=PROGRAM, _json_metadata={
             'courses': [
                 course_content_metadata.json_metadata,
                 {
@@ -489,7 +489,7 @@ def test_update_full_metadata(self, mock_oauth_client, mock_partition_course_key
         # the normalized metadata serializer.
         metadata_4 = ContentMetadataFactory(content_type=COURSE, content_key=course_key_4)
         metadata_4.catalog_queries.set([self.catalog_query])
-        metadata_4.json_metadata['advertised_course_run_uuid'] = None
+        metadata_4._json_metadata['advertised_course_run_uuid'] = None  # pylint: disable=protected-access
         metadata_4.save()
         non_course_metadata = ContentMetadataFactory(content_type=COURSE_RUN, content_key=non_course_key)
         non_course_metadata.catalog_queries.set([self.catalog_query])
@@ -714,7 +714,7 @@ def test_update_full_metadata_exec_ed(self, mock_oauth_client, mock_partition_co
 
         # Simulate a pre-existing ContentMetadata object freshly seeded using the response from /api/v1/search/all/
         course_metadata = ContentMetadataFactory.create(
-            content_type=COURSE, content_key=course_key, json_metadata={
+            content_type=COURSE, content_key=course_key, _json_metadata={
                 'aggregation_key': 'course:edX+testX',
                 'key': 'edX+testX',
                 'course_type': EXEC_ED_2U_COURSE_TYPE,
@@ -801,7 +801,8 @@ def setUp(self):
         self.course_metadata_published.catalog_queries.set([self.enterprise_catalog_query])
         self.course_metadata_published.tags.set([self.tag1])
         self.course_metadata_unpublished = ContentMetadataFactory(content_type=COURSE, content_key='course-2')
-        self.course_metadata_unpublished.json_metadata.get('course_runs')[0].update({
+        # pylint: disable=protected-access
+        self.course_metadata_unpublished._json_metadata.get('course_runs')[0].update({
             'status': 'unpublished',
         })
         self.course_metadata_unpublished.catalog_queries.set([self.enterprise_catalog_query])
@@ -821,7 +822,7 @@ def setUp(self):
             content_type=COURSE_RUN,
             parent_content_key='course-2',
         )
-        self.course_run_metadata_unpublished.json_metadata.update({
+        self.course_run_metadata_unpublished._json_metadata.update({
             'status': 'unpublished',
         })
         self.course_run_metadata_unpublished.catalog_queries.set([course_run_catalog_query])
@@ -874,7 +875,7 @@ def test_add_metadata_to_algolia_objects_skips_metadata_records_over_max_size(se
         short_description_string = "ayylmao".join(["" for x in range(50000)])
         full_description_string = "foobar".join(["" for x in range(50000)])
         too_big_sized_course = ContentMetadataFactory(content_type=COURSE, content_key='test-course-2')
-        too_big_sized_course.json_metadata.update(
+        too_big_sized_course._json_metadata.update(  # pylint: disable=protected-access
             {"short_description": short_description_string, "full_description": full_description_string}
         )
         too_big_sized_course.save()
@@ -1216,7 +1217,7 @@ def test_index_algolia_published_course_to_program(self, mock_search_client):
         program_2 = ContentMetadataFactory(content_type=PROGRAM, content_key='program-2')
 
         # Make program-2 hidden to make it "non-indexable". Later we will assert that it will not get indexed.
-        program_2.json_metadata.update({
+        program_2._json_metadata.update({  # pylint: disable=protected-access
             'hidden': True,
         })
         program_2.save()
@@ -1324,7 +1325,7 @@ def test_index_algolia_unpublished_course_to_program(self, mock_search_client):
         program_2.catalog_queries.set([self.enterprise_catalog_query])
 
         # Make program-2 hidden to make it "non-indexable". Later we will assert that it will not get indexed.
-        program_2.json_metadata.update({
+        program_2._json_metadata.update({  # pylint: disable=protected-access
             'hidden': True,
         })
         program_2.save()
@@ -1635,7 +1636,7 @@ def test_index_algolia_program_to_pathway(self, mock_search_client):
         program_1.catalog_queries.set([self.enterprise_catalog_query])
 
         # Make program-2 hidden to make it "non-indexable". Later we will assert that it will not get indexed.
-        program_2.json_metadata.update({
+        program_2._json_metadata.update({  # pylint: disable=protected-access
             'hidden': True,
         })
         program_2.save()
@@ -1772,7 +1773,7 @@ def test_index_algolia_all_uuids(self, mock_search_client):
         program_for_main_course = ContentMetadataFactory(content_type=PROGRAM, content_key='program-1')
         # Make the program hidden to make it "non-indexable", but ensure that it still gets indexed due to being related
         # to an indexable course.
-        program_for_main_course.json_metadata.update({
+        program_for_main_course._json_metadata.update({  # pylint: disable=protected-access
             'hidden': True,
         })
         program_for_main_course.save()
@@ -1780,7 +1781,7 @@ def test_index_algolia_all_uuids(self, mock_search_client):
         program_for_pathway.catalog_queries.set([self.enterprise_catalog_query])
         # Make the program hidden to make it "non-indexable", but ensure that it still gets indexed due to being related
         # to an indexable pathway.
-        program_for_pathway.json_metadata.update({
+        program_for_pathway._json_metadata.update({  # pylint: disable=protected-access
             'hidden': True,
         })
         program_for_pathway.save()
diff --git a/enterprise_catalog/apps/api/v1/tests/test_serializers.py b/enterprise_catalog/apps/api/v1/tests/test_serializers.py
index 5c00c771f..bdcbcb192 100644
--- a/enterprise_catalog/apps/api/v1/tests/test_serializers.py
+++ b/enterprise_catalog/apps/api/v1/tests/test_serializers.py
@@ -29,9 +29,7 @@ def test_product_source_formatting(self):
         Test that the content metadata serializer will transform product source data within the json metadata field
         from a string to a dict.
         """
-        json_metadata = self.content_metadata_item.json_metadata
-        json_metadata['product_source'] = '2u'
-        self.content_metadata_item.json_metadata = json_metadata
+        self.content_metadata_item._json_metadata.update({'product_source': '2u'})  # pylint: disable=protected-access
         self.content_metadata_item.save()
         serialized_data = ContentMetadataSerializer(self.content_metadata_item)
         assert serialized_data.data.get('product_source') == {
diff --git a/enterprise_catalog/apps/api/v1/tests/test_utils.py b/enterprise_catalog/apps/api/v1/tests/test_utils.py
index 2b0d99320..51185d1b2 100644
--- a/enterprise_catalog/apps/api/v1/tests/test_utils.py
+++ b/enterprise_catalog/apps/api/v1/tests/test_utils.py
@@ -69,10 +69,10 @@ def test_get_archived_content_count(self):
         """
         Test that archived content will increment the count.
         """
-        content_1 = ContentMetadataFactory.create(json_metadata={'course_run_statuses': ['archived']})
-        content_2 = ContentMetadataFactory.create(json_metadata={'course_run_statuses': ['unpublished', 'archived']})
+        content_1 = ContentMetadataFactory.create(_json_metadata={'course_run_statuses': ['archived']})
+        content_2 = ContentMetadataFactory.create(_json_metadata={'course_run_statuses': ['unpublished', 'archived']})
         # if there's at least one published course run, the content should not be considered archived
-        content_3 = ContentMetadataFactory.create(json_metadata={'course_run_statuses': ['published', 'archived']})
+        content_3 = ContentMetadataFactory.create(_json_metadata={'course_run_statuses': ['published', 'archived']})
 
         highlighted_content_1 = HighlightedContentFactory(content_metadata=content_1)
         highlighted_content_2 = HighlightedContentFactory(content_metadata=content_2)
diff --git a/enterprise_catalog/apps/catalog/management/commands/tests/test_set_global_average_course_rating_value.py b/enterprise_catalog/apps/catalog/management/commands/tests/test_set_global_average_course_rating_value.py
index bdc0b1b08..275c05bca 100644
--- a/enterprise_catalog/apps/catalog/management/commands/tests/test_set_global_average_course_rating_value.py
+++ b/enterprise_catalog/apps/catalog/management/commands/tests/test_set_global_average_course_rating_value.py
@@ -25,13 +25,13 @@ def test_command_averages_course_reviews(
         """
         ContentMetadataFactory(
             content_type='course',
-            json_metadata={'avg_course_rating': 5, 'reviews_count': 20}
+            _json_metadata={'avg_course_rating': 5, 'reviews_count': 20}
         )
         ContentMetadataFactory(
             content_type='course',
-            json_metadata={'avg_course_rating': 4, 'reviews_count': 10}
+            _json_metadata={'avg_course_rating': 4, 'reviews_count': 10}
         )
-        ContentMetadataFactory(json_metadata={})
+        ContentMetadataFactory(_json_metadata={})
         call_command(self.command_name)
         expected_total_average = ((5 * 20) + (4 * 10)) / 30
 
diff --git a/enterprise_catalog/apps/catalog/migrations/0040_restrictedcoursemetadata_and_more.py b/enterprise_catalog/apps/catalog/migrations/0040_restrictedcoursemetadata_and_more.py
new file mode 100644
index 000000000..f4b21bc31
--- /dev/null
+++ b/enterprise_catalog/apps/catalog/migrations/0040_restrictedcoursemetadata_and_more.py
@@ -0,0 +1,118 @@
+# Generated by Django 4.2.16 on 2024-10-09 03:45
+
+import collections
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+import jsonfield.encoder
+import jsonfield.fields
+import model_utils.fields
+import simple_history.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('catalog', '0039_alter_catalogquery_unique_together_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='RestrictedCourseMetadata',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')),
+                ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')),
+                ('content_uuid', models.UUIDField(blank=True, help_text='The UUID that represents a piece of content. This value is usually a secondary identifier to content_key in the enterprise environment.', null=True, verbose_name='Content UUID')),
+                ('content_type', models.CharField(choices=[('course', 'Course'), ('courserun', 'Course Run'), ('program', 'Program'), ('learnerpathway', 'Learner Pathway')], max_length=255)),
+                ('parent_content_key', models.CharField(blank=True, db_index=True, help_text="The key represents this content's parent. For example for course_runs content their parent course key.", max_length=255, null=True)),
+                ('_json_metadata', jsonfield.fields.JSONField(blank=True, db_column='json_metadata', default={}, dump_kwargs={'cls': jsonfield.encoder.JSONEncoder, 'indent': 4, 'separators': (',', ':')}, help_text="The metadata about a particular piece content as retrieved from the discovery service's search/all endpoint results, specified as a JSON object.", load_kwargs={'object_pairs_hook': collections.OrderedDict}, null=True)),
+                ('content_key', models.CharField(help_text='The key that represents a piece of content, such as a course, course run, or program.', max_length=255)),
+            ],
+            options={
+                'verbose_name': 'Restricted Content Metadata',
+                'verbose_name_plural': 'Restricted Content Metadata',
+            },
+        ),
+        migrations.AlterField(
+            model_name='contentmetadata',
+            name='json_metadata',
+            field=jsonfield.fields.JSONField(blank=True, db_column='json_metadata', default={}, dump_kwargs={'cls': jsonfield.encoder.JSONEncoder, 'indent': 4, 'separators': (',', ':')}, help_text="The metadata about a particular piece content as retrieved from the discovery service's search/all endpoint results, specified as a JSON object.", load_kwargs={'object_pairs_hook': collections.OrderedDict}, null=True),
+        ),
+        migrations.RenameField(
+            model_name='contentmetadata',
+            old_name='json_metadata',
+            new_name='_json_metadata',
+        ),
+        migrations.AlterField(
+            model_name='historicalcontentmetadata',
+            name='json_metadata',
+            field=jsonfield.fields.JSONField(blank=True, db_column='json_metadata', default={}, dump_kwargs={'cls': jsonfield.encoder.JSONEncoder, 'indent': 4, 'separators': (',', ':')}, help_text="The metadata about a particular piece content as retrieved from the discovery service's search/all endpoint results, specified as a JSON object.", load_kwargs={'object_pairs_hook': collections.OrderedDict}, null=True),
+        ),
+        migrations.RenameField(
+            model_name='historicalcontentmetadata',
+            old_name='json_metadata',
+            new_name='_json_metadata',
+        ),
+        migrations.AlterField(
+            model_name='catalogquery',
+            name='content_filter',
+            field=jsonfield.fields.JSONField(default=dict, dump_kwargs={'cls': jsonfield.encoder.JSONEncoder, 'ensure_ascii': False, 'indent': 4, 'separators': (',', ':')}, help_text="Query parameters which will be used to filter the discovery service's search/all endpoint results, specified as a JSON object.", load_kwargs={'object_pairs_hook': collections.OrderedDict}),
+        ),
+        migrations.CreateModel(
+            name='RestrictedRunAllowedForRestrictedCourse',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')),
+                ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')),
+                ('course', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='restricted_run_allowed_for_restricted_course', to='catalog.restrictedcoursemetadata')),
+                ('run', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='restricted_run_allowed_for_restricted_course', to='catalog.contentmetadata')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.AddField(
+            model_name='restrictedcoursemetadata',
+            name='catalog_query',
+            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='restricted_content_metadata', to='catalog.catalogquery'),
+        ),
+        migrations.AddField(
+            model_name='restrictedcoursemetadata',
+            name='unrestricted_parent',
+            field=models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='restricted_courses', to='catalog.contentmetadata'),
+        ),
+        migrations.CreateModel(
+            name='HistoricalRestrictedCourseMetadata',
+            fields=[
+                ('id', models.IntegerField(auto_created=True, blank=True, db_index=True, verbose_name='ID')),
+                ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')),
+                ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')),
+                ('content_uuid', models.UUIDField(blank=True, help_text='The UUID that represents a piece of content. This value is usually a secondary identifier to content_key in the enterprise environment.', null=True, verbose_name='Content UUID')),
+                ('content_type', models.CharField(choices=[('course', 'Course'), ('courserun', 'Course Run'), ('program', 'Program'), ('learnerpathway', 'Learner Pathway')], max_length=255)),
+                ('parent_content_key', models.CharField(blank=True, db_index=True, help_text="The key represents this content's parent. For example for course_runs content their parent course key.", max_length=255, null=True)),
+                ('_json_metadata', jsonfield.fields.JSONField(blank=True, db_column='json_metadata', default={}, dump_kwargs={'cls': jsonfield.encoder.JSONEncoder, 'indent': 4, 'separators': (',', ':')}, help_text="The metadata about a particular piece content as retrieved from the discovery service's search/all endpoint results, specified as a JSON object.", load_kwargs={'object_pairs_hook': collections.OrderedDict}, null=True)),
+                ('content_key', models.CharField(help_text='The key that represents a piece of content, such as a course, course run, or program.', max_length=255)),
+                ('history_id', models.AutoField(primary_key=True, serialize=False)),
+                ('history_date', models.DateTimeField(db_index=True)),
+                ('history_change_reason', models.CharField(max_length=100, null=True)),
+                ('history_type', models.CharField(choices=[('+', 'Created'), ('~', 'Changed'), ('-', 'Deleted')], max_length=1)),
+                ('catalog_query', models.ForeignKey(blank=True, db_constraint=False, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='+', to='catalog.catalogquery')),
+                ('history_user', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
+                ('unrestricted_parent', models.ForeignKey(blank=True, db_constraint=False, null=True, on_delete=django.db.models.deletion.DO_NOTHING, related_name='+', to='catalog.contentmetadata')),
+            ],
+            options={
+                'verbose_name': 'historical Restricted Content Metadata',
+                'verbose_name_plural': 'historical Restricted Content Metadata',
+                'ordering': ('-history_date', '-history_id'),
+                'get_latest_by': ('history_date', 'history_id'),
+            },
+            bases=(simple_history.models.HistoricalChanges, models.Model),
+        ),
+        migrations.AlterUniqueTogether(
+            name='restrictedcoursemetadata',
+            unique_together={('content_key', 'catalog_query')},
+        ),
+    ]
diff --git a/enterprise_catalog/apps/catalog/models.py b/enterprise_catalog/apps/catalog/models.py
index ce4d163d9..584a1f9e4 100644
--- a/enterprise_catalog/apps/catalog/models.py
+++ b/enterprise_catalog/apps/catalog/models.py
@@ -1,3 +1,158 @@
+"""
+SPIKE TEST:
+
+1. First, reset the sqlite database:
+
+rm enterprise_catalog/default.db && python manage.py migrate && python manage.py shell
+
+2. Then, paste everything below into the python shell:
+
+# Set up catalog and catalog query:
+# =================================
+from enterprise_catalog.apps.catalog.models import *
+from enterprise_catalog.apps.catalog.tests.factories import *
+catalog = (
+    EnterpriseCatalog.objects.filter(uuid='0fa8c2e9-ba23-4749-86d9-0bfa84991490').first()
+    or
+    EnterpriseCatalogFactory(uuid='0fa8c2e9-ba23-4749-86d9-0bfa84991490')
+)
+catalog_query = catalog.catalog_query
+catalog_2 = (
+    EnterpriseCatalog.objects.filter(uuid='2fa8c2e9-ba23-4749-86d9-0bfa84991490').first()
+    or
+    EnterpriseCatalogFactory(uuid='2fa8c2e9-ba23-4749-86d9-0bfa84991490')
+)
+catalog_query_2 = catalog_2.catalog_query
+
+# Set up pure course and runs:
+# ===============================
+course_pure = (
+    ContentMetadata.objects.filter(content_key='course_pure').first()
+    or
+    ContentMetadataFactory(content_key='course_pure', content_type='course')
+)
+course_pure.catalog_queries.set([catalog_query])
+course_pure_run1 = (
+    ContentMetadata.objects.filter(content_key='course_pure_run1').first()
+    or
+    ContentMetadataFactory(
+        content_key='course_pure_run1',
+        content_type='courserun',
+        parent_content_key=course_pure.content_key,
+    )
+)
+
+# Set up mixed course and runs:
+# ===============================
+course_mixed = (
+    ContentMetadata.objects.filter(content_key='course_mixed').first()
+    or
+    ContentMetadataFactory(content_key='course_mixed', content_type='course')
+)
+course_mixed.catalog_queries.set([catalog_query, catalog_query_2])
+course_mixed_run1 = (
+    ContentMetadata.objects.filter(content_key='course_mixed_run1').first()
+    or
+    ContentMetadataFactory(
+        content_key='course_mixed_run1',
+        content_type='courserun',
+        parent_content_key=course_mixed.content_key,
+    )
+)
+course_mixed_run2 = (
+    ContentMetadata.objects.filter(content_key='course_mixed_run2').first()
+    or
+    ContentMetadataFactory(
+        content_key='course_mixed_run2',
+        content_type='courserun',
+        parent_content_key=course_mixed.content_key,
+    )
+)
+course_mixed_run3 = (
+    ContentMetadata.objects.filter(content_key='course_mixed_run3').first()
+    or
+    ContentMetadataFactory(
+        content_key='course_mixed_run3',
+        content_type='courserun',
+        parent_content_key=course_mixed.content_key,
+    )
+)
+restricted_course_mixed, _ = RestrictedCourseMetadata.objects.get_or_create(
+    content_key=course_mixed.content_key,
+    content_type=course_mixed.content_type,
+    catalog_query=catalog_query,
+    unrestricted_parent=course_mixed,
+    _json_metadata={'RESTRICTED': 'RESTRICTED_FOR_QUERY_1'},
+)
+restricted_runs_allowed = []
+restricted_runs_allowed.append(RestrictedRunAllowedForRestrictedCourse.objects.get_or_create(
+    course=restricted_course_mixed,
+    run=course_mixed_run2,
+)[0])
+restricted_course_mixed_2, _ = RestrictedCourseMetadata.objects.get_or_create(
+    content_key=course_mixed.content_key,
+    content_type=course_mixed.content_type,
+    catalog_query=catalog_query_2,
+    unrestricted_parent=course_mixed,
+    _json_metadata={'RESTRICTED': 'RESTRICTED_FOR_QUERY_2'},
+)
+restricted_runs_allowed.append(RestrictedRunAllowedForRestrictedCourse.objects.get_or_create(
+    course=restricted_course_mixed_2,
+    run=course_mixed_run3,
+)[0])
+
+# Set up unicorn course and runs:
+# ===============================
+course_unicorn = (
+    ContentMetadata.objects.filter(content_key='course_unicorn').first()
+    or
+    ContentMetadataFactory(content_key='course_unicorn', content_type='course')
+)
+course_unicorn.catalog_queries.set([catalog_query])
+course_unicorn_run1 = (
+    ContentMetadata.objects.filter(content_key='course_unicorn_run1').first()
+    or
+    ContentMetadataFactory(
+        content_key='course_unicorn_run1',
+        content_type='courserun',
+        parent_content_key=course_unicorn.content_key,
+    )
+)
+restricted_course_unicorn, _ = RestrictedCourseMetadata.objects.get_or_create(
+    content_key=course_unicorn.content_key,
+    content_type=course_unicorn.content_type,
+    catalog_query=catalog_query,
+    unrestricted_parent=course_unicorn,
+    _json_metadata={'RESTRICTED': 'RESTRICTED_FOR_QUERY_1'},
+)
+restricted_runs_allowed.append(RestrictedRunAllowedForRestrictedCourse.objects.get_or_create(
+    course=restricted_course_unicorn,
+    run=course_unicorn_run1,
+)[0])
+
+
+3. run this to delete all objects:
+
+all_objs = [
+    catalog,
+    catalog_query,
+    catalog_2,
+    catalog_query_2,
+    course_pure,
+    course_pure_run1,
+    course_mixed,
+    course_mixed_run1,
+    course_mixed_run2,
+    course_mixed_run3,
+    restricted_course_mixed,
+    restricted_course_mixed_2,
+    course_unicorn,
+    course_unicorn_run1,
+    restricted_course_unicorn,
+] + restricted_runs_allowed
+for obj in all_objs:
+    obj.delete()
+"""
 import collections
 import json
 from logging import getLogger
@@ -252,11 +407,50 @@ def content_metadata(self):
         Helper to retrieve the content metadata associated with the catalog.
 
         Returns:
-            Queryset: The queryset of associated content metadata
+            Queryset of ContentMetadata: The queryset of associated content metadata
+        """
+        if not self.catalog_query:
+            return ContentMetadata.objects.none()
+        return self.catalog_query.contentmetadata_set.prefetch_related(
+            'restricted_run_allowed_for_restricted_course'
+        ).filter(
+            # Exclude all restricted runs (heuristic is that a run is assumed
+            # restricted if it is mapped to a restricted course via
+            # RestrictedRunAllowedForRestrictedCourse).
+            restricted_run_allowed_for_restricted_course__isnull=True,
+        )
+
+    @property
+    def content_metadata_with_restricted(self):
+        """
+        Same as self.content_metadata, but dynamically replace course json_metadata
+        with the correct version containing restricted runs allowed by the current
+        catalog query.
+
+        The technique to dynamically override ContentMetadata.json_metadata is a
+        combination of two things:
+        1. This method setting the `restricted_course_metadata_for_catalog_query`
+           attribute in the queryset to get the correct RestrictedCourseMetadata, and
+        2. The ContentMetdata.json_metadata attribute being a property that
+           dynamically uses (1) or falls back to the stored value in _json_metadata.
+
+        Returns:
+            Queryset of ContentMetadata: Same as self.content_metadata, but courses may have augmented
+            json_metadata.
         """
         if not self.catalog_query:
             return ContentMetadata.objects.none()
-        return self.catalog_query.contentmetadata_set.all()
+        related_contentmetadata = self.catalog_query.contentmetadata_set
+        # Provide json_metadata overrides via dynamic attribute if any restricted runs are allowed.
+        if self.catalog_query.restricted_runs_allowed:
+            # FYI: prefetch causes a performance penalty by introducing a 2nd database query.
+            prefetch_qs = models.Prefetch(
+                'restricted_courses',
+                queryset=RestrictedCourseMetadata.objects.filter(catalog_query=self.catalog_query),
+                to_attr='restricted_course_metadata_for_catalog_query',
+            )
+            related_contentmetadata = related_contentmetadata.prefetch_related(prefetch_qs)
+        return related_contentmetadata.all()
 
     @cached_property
     def restricted_runs_allowed(self):
@@ -324,7 +518,7 @@ def get_catalog_content_diff(self, content_keys):
         items_not_found = distinct_content_keys - found_content_keys
         return [{'content_key': item} for item in items_not_found], items_not_included, items_found
 
-    def get_matching_content(self, content_keys):
+    def get_matching_content(self, content_keys, include_restricted=False):
         """
         Returns the set of content contained within this catalog that matches
         any of the course keys, course run keys, or programs keys specified by
@@ -361,15 +555,35 @@ def get_matching_content(self, content_keys):
         # (if any) to handle the following case:
         #   - catalog contains courses and the specified content_keys are course run ids.
         searched_metadata = ContentMetadata.objects.filter(content_key__in=content_keys)
+        if include_restricted and self.catalog_query.restricted_runs_allowed:
+            # Only hide restricted runs that are not allowed by the current catalog.
+            searched_metadata = searched_metadata.prefetch_related(
+                'restricted_run_allowed_for_restricted_course'
+            ).exclude(
+                # Find all restricted runs allowed by a RestrictedCourseMetadata related to the
+                # current CatalogQuery. Do NOT exclude those.
+                ~Q(restricted_run_allowed_for_restricted_course__course__catalog_query=self.catalog_query)
+                # Exclude all other restricted runs. A run is assumed of type restricted if it
+                # is related to at least one RestrictedRunAllowedForRestrictedCourse.
+                & Q(restricted_run_allowed_for_restricted_course__isnull=False)
+            )
+        else:
+            # Hide ALL restricted runs.
+            searched_metadata = searched_metadata.exclude(
+                restricted_run_allowed_for_restricted_course__isnull=False
+            )
         parent_content_keys = {
             metadata.parent_content_key
             for metadata in searched_metadata
             if metadata.parent_content_key
         }
         query |= Q(content_key__in=parent_content_keys)
-        return self.content_metadata.filter(query)
+        if include_restricted:
+            return self.content_metadata_with_restricted.filter(query)
+        else:
+            return self.content_metadata.filter(query)
 
-    def contains_content_keys(self, content_keys):
+    def contains_content_keys(self, content_keys, include_restricted=False):
         """
         Determines whether the given ``content_keys`` are part of the catalog.
 
@@ -386,10 +600,10 @@ def contains_content_keys(self, content_keys):
             in the ``content_keys`` list (to handle cases when a catalog contains only courses,
             but course run keys are provided in the ``content_keys`` argument).
         """
-        included_content = self.get_matching_content(content_keys)
+        included_content = self.get_matching_content(content_keys, include_restricted=include_restricted)
         return included_content.exists()
 
-    def filter_content_keys(self, content_keys):
+    def filter_content_keys(self, content_keys, include_restricted=False):
         """
         Determines whether content_keys are part of the catalog.
 
@@ -419,7 +633,11 @@ def filter_content_keys(self, content_keys):
         query = Q(content_key__in=content_keys) | Q(parent_content_key__in=content_keys)
 
         items_included = set()
-        for content in self.content_metadata.filter(query).all():
+        if include_restricted:
+            accessible_metadata_qs = self.content_metadata_with_restricted
+        else:
+            accessible_metadata_qs = self.content_metadata
+        for content in accessible_metadata_qs.filter(query).all():
             if content.content_key in content_keys:
                 items_included.add(content.content_key)
             elif content.parent_content_key in content_keys:
@@ -568,13 +786,12 @@ def bulk_update(self, objs, fields, batch_size=None):
         super().bulk_update(objs, fields, batch_size=batch_size)
 
 
-class ContentMetadata(TimeStampedModel):
+class BaseContentMetadata(TimeStampedModel):
     """
-    Stores the JSON metadata for a piece of content, such as a course, course run, or program.
-    The metadata is retrieved from the Discovery service /search/all endpoint.
-
-    .. no_pii:
+    Common ContentMetadata fields.
     """
+    class Meta:
+        abstract = True
 
     content_uuid = models.UUIDField(
         null=True,
@@ -587,7 +804,6 @@ class ContentMetadata(TimeStampedModel):
             "in the enterprise environment."
         )
     )
-
     content_key = models.CharField(
         max_length=255,
         blank=False,
@@ -612,32 +828,22 @@ class ContentMetadata(TimeStampedModel):
             "The key represents this content's parent. For example for course_runs content their parent course key."
         )
     )
-
-    # one course can be associated with many programs and one program can contain many courses.
-    associated_content_metadata = models.ManyToManyField('self', blank=True)
-
-    json_metadata = JSONField(
+    _json_metadata = JSONField(
         default={},
         blank=True,
         null=True,
         load_kwargs={'object_pairs_hook': collections.OrderedDict},
         dump_kwargs={'indent': 4, 'cls': JSONEncoder, 'separators': (',', ':')},
+        db_column='json_metadata',
         help_text=_(
             "The metadata about a particular piece content as retrieved from the discovery service's search/all "
             "endpoint results, specified as a JSON object."
         )
     )
-    catalog_queries = models.ManyToManyField(CatalogQuery)
 
     history = HistoricalRecords()
-
     objects = ContentMetadataManager()
 
-    class Meta:
-        verbose_name = _("Content Metadata")
-        verbose_name_plural = _("Content Metadata")
-        app_label = 'catalog'
-
     @property
     def is_exec_ed_2u_course(self):
         # pylint: disable=no-member
@@ -666,6 +872,14 @@ def get_child_records(cls, content_metadata):
         """
         return cls.objects.filter(parent_content_key=content_metadata.content_key)
 
+    @property
+    def json_metadata(self):
+        return self._json_metadata
+
+    @json_metadata.setter
+    def json_metadata(self, new_json_metadata):
+        self._json_metadata = new_json_metadata
+
     def __str__(self):
         """
         Return human-readable string representation.
@@ -677,6 +891,119 @@ def __str__(self):
         )
 
 
+class ContentMetadata(BaseContentMetadata):
+    """
+    Stores the JSON metadata for a piece of content, such as a course, course run, or program.
+    The metadata is retrieved from the Discovery service /search/all endpoint.
+
+    .. no_pii:
+    """
+    class Meta:
+        verbose_name = _("Content Metadata")
+        verbose_name_plural = _("Content Metadata")
+        app_label = 'catalog'
+
+    # one course can be associated with many programs and one program can contain many courses.
+    associated_content_metadata = models.ManyToManyField('self', blank=True)
+
+    # one course can be part of many CatalogQueries and one CatalogQuery can contain many courses.
+    catalog_queries = models.ManyToManyField(CatalogQuery)
+
+    history = HistoricalRecords()
+
+    @property
+    def json_metadata(self):
+        """
+        Use the CatalogQuery-specific version of a course json_metadata if one exists
+        (potentially containing restricted runs allowed by that CatatlogQuery),
+        otherwise fall back to the standard unrestricted-only version.
+        """
+        restricted_course_metadata_for_catalog_query = getattr(
+            self,
+            'restricted_course_metadata_for_catalog_query',
+            None,
+        )
+        # Truthy means that the requester wants to see restricted runs AND restricted
+        # runs were actually found for this specific course and the requester's
+        # specific Catalog.
+        if restricted_course_metadata_for_catalog_query:
+            # pylint: disable=protected-access, unsubscriptable-object
+            return restricted_course_metadata_for_catalog_query[0]._json_metadata
+        return self._json_metadata
+
+    @json_metadata.setter
+    def json_metadata(self, new_json_metadata):
+        self._json_metadata = new_json_metadata
+
+
+class RestrictedCourseMetadata(BaseContentMetadata):
+    """
+    Copies of courses, but one copy for each CatalogQuery which explicitly
+    allows any restricted runs of the course.
+
+    .. no_pii:
+    """
+    class Meta:
+        verbose_name = _("Restricted Content Metadata")
+        verbose_name_plural = _("Restricted Content Metadata")
+        app_label = 'catalog'
+        unique_together = ('content_key', 'catalog_query')
+
+    # Overwrite content_key from BaseContentMetadata in order to change unique
+    # to False. Use unique_together to allow multiple copies of the same course
+    # (one for each catalog query.
+    content_key = models.CharField(
+        max_length=255,
+        blank=False,
+        null=False,
+        unique=False,
+        help_text=_(
+            "The key that represents a piece of content, such as a course, course run, or program."
+        )
+    )
+    unrestricted_parent = models.ForeignKey(
+        ContentMetadata,
+        blank=False,
+        null=True,
+        related_name='restricted_courses',
+        on_delete=models.deletion.SET_NULL,
+    )
+    catalog_query = models.ForeignKey(
+        CatalogQuery,
+        blank=False,
+        null=True,
+        related_name='restricted_content_metadata',
+        on_delete=models.deletion.SET_NULL,
+    )
+    history = HistoricalRecords()
+
+
+class RestrictedRunAllowedForRestrictedCourse(TimeStampedModel):
+    """
+    Mapping table to relate RestrictedCourseMetadata objects to restricted runs in ContentMetadata.
+
+    A run should be mapped to a restricted course IFF the RestrictedCourseMetadata's
+    catalog query explicitly allows the run. This mapping table should be generated by the
+    update-content-metadata task.
+
+    .. no_pii:
+    """
+    course = models.ForeignKey(
+        RestrictedCourseMetadata,
+        blank=False,
+        null=True,
+        related_name='restricted_run_allowed_for_restricted_course',
+        on_delete=models.deletion.SET_NULL,
+    )
+    run = models.ForeignKey(
+        ContentMetadata,
+        blank=False,
+        null=True,
+        related_name='restricted_run_allowed_for_restricted_course',
+        on_delete=models.deletion.SET_NULL,
+    )
+
+
 def content_metadata_with_type_course():
     """
     Find all ContentMetadata records with a content type of "course".
@@ -732,12 +1059,12 @@ def _get_defaults_from_metadata(entry, exists=False):
             if value := entry.get(field):
                 entry_minimal[field] = value
         if entry_minimal:
-            defaults.update({'json_metadata': entry_minimal})
+            defaults.update({'_json_metadata': entry_minimal})
     elif not exists or (content_type != 'course'):
         # Update json_metadata for non-courses when ContentMetadata object already exists. Also,
         # always include json_metadata (regardless of content type) if ContentMetadata object
         # does not yet exist in the database.
-        defaults.update({'json_metadata': entry})
+        defaults.update({'_json_metadata': entry})
     return defaults
 
 
@@ -790,9 +1117,9 @@ def _update_existing_content_metadata(existing_metadata_defaults, existing_metad
         content_metadata = existing_metadata_by_key.get(defaults['content_key'])
         if content_metadata:
             for key, value in defaults.items():
-                if key == 'json_metadata':
+                if key == '_json_metadata':
                     # merge new json_metadata with old json_metadata (i.e., don't replace it fully)
-                    content_metadata.json_metadata.update(value)
+                    content_metadata._json_metadata.update(value)  # pylint: disable=protected-access
                 else:
                     # replace attributes with new values
                     setattr(content_metadata, key, value)
@@ -803,7 +1130,7 @@ def _update_existing_content_metadata(existing_metadata_defaults, existing_metad
         for metadata in metadata_list:
             LOGGER.info(f"[Dry Run] Skipping Content Metadata update: {metadata}")
     else:
-        metadata_fields_to_update = ['content_key', 'parent_content_key', 'content_type', 'json_metadata']
+        metadata_fields_to_update = ['content_key', 'parent_content_key', 'content_type', '_json_metadata']
         batch_size = settings.UPDATE_EXISTING_CONTENT_METADATA_BATCH_SIZE
         for batched_metadata in batch(metadata_list, batch_size=batch_size):
             try:
diff --git a/enterprise_catalog/apps/catalog/tasks.py b/enterprise_catalog/apps/catalog/tasks.py
index 82f0e4df8..e3785f4fa 100644
--- a/enterprise_catalog/apps/catalog/tasks.py
+++ b/enterprise_catalog/apps/catalog/tasks.py
@@ -17,6 +17,7 @@
 @shared_task(base=LoggedTask)
 def compare_catalog_queries_to_filters_task():
     logger.info('compare_catalog_queries_to_filters starting...')
+    # NOTE: No need to exclude restricted runs because they are already filtered out via content_type.
     for content_metadata in ContentMetadata.objects.filter(content_type=COURSE):
         for enterprise_catalog in EnterpriseCatalog.objects.all():
             try:
diff --git a/enterprise_catalog/apps/catalog/tests/factories.py b/enterprise_catalog/apps/catalog/tests/factories.py
index d972bdd94..ca4445952 100644
--- a/enterprise_catalog/apps/catalog/tests/factories.py
+++ b/enterprise_catalog/apps/catalog/tests/factories.py
@@ -18,6 +18,8 @@
     EnterpriseCatalog,
     EnterpriseCatalogFeatureRole,
     EnterpriseCatalogRoleAssignment,
+    RestrictedCourseMetadata,
+    RestrictedRunAllowedForRestrictedCourse,
 )
 from enterprise_catalog.apps.core.models import User
 
@@ -80,7 +82,7 @@ class Meta:
     parent_content_key = None
 
     @factory.lazy_attribute
-    def json_metadata(self):
+    def _json_metadata(self):
         json_metadata = {
             'key': self.content_key,
             'aggregation_key': f'{self.content_type}:{self.content_key}',
@@ -187,6 +189,31 @@ def json_metadata(self):
         return json_metadata
 
 
+class RestrictedCourseMetadataFactory(factory.django.DjangoModelFactory):
+    """
+    Test factory for the `RestrictedCourseMetadata` model.
+    """
+    class Meta:
+        model = RestrictedCourseMetadata
+
+    content_key = factory.Faker('bothify', text='??????????+####')
+    content_uuid = factory.LazyFunction(uuid4)
+    content_type = COURSE
+    parent_content_key = None
+    _json_metadata = {}  # Callers are encouraged to set this.
+
+
+class RestrictedRunAllowedForRestrictedCourseFactory(factory.django.DjangoModelFactory):
+    """
+    Test factory for the `RestrictedRunAllowedForRestrictedCourse` model.
+    """
+    class Meta:
+        model = RestrictedRunAllowedForRestrictedCourse
+
+    course = factory.SubFactory(RestrictedCourseMetadataFactory)
+    run = factory.SubFactory(ContentMetadataFactory)
+
+
 class UserFactory(factory.django.DjangoModelFactory):
     username = factory.Faker('user_name')
     password = factory.PostGenerationMethodCall('set_password', USER_PASSWORD)
diff --git a/enterprise_catalog/apps/catalog/tests/test_algolia_utils.py b/enterprise_catalog/apps/catalog/tests/test_algolia_utils.py
index ced7a1da8..ad4c07332 100644
--- a/enterprise_catalog/apps/catalog/tests/test_algolia_utils.py
+++ b/enterprise_catalog/apps/catalog/tests/test_algolia_utils.py
@@ -126,7 +126,7 @@ def test_should_index_course(
         }
         course_metadata = ContentMetadataFactory.create(
             content_type=COURSE,
-            json_metadata=json_metadata,
+            _json_metadata=json_metadata,
         )
         # pylint: disable=protected-access
         assert utils._should_index_course(course_metadata) is expected_result
@@ -996,7 +996,7 @@ def test_get_program_skill_names(self, program_metadata, course_metadata, expect
         ContentMetadataFactory.create(
             content_key=program_metadata['courses'][0]['key'],
             content_type=COURSE,
-            json_metadata=course_metadata,
+            _json_metadata=course_metadata,
         )
         skill_names = utils.get_program_skill_names(program_metadata)
         self.assertEqual(sorted(skill_names), sorted(expected_skill_names))
@@ -1269,7 +1269,7 @@ def test_get_program_subjects(self, program_metadata, course_metadata, expected_
         ContentMetadataFactory.create(
             content_key=program_metadata['courses'][0]['key'],
             content_type=COURSE,
-            json_metadata=course_metadata,
+            _json_metadata=course_metadata,
         )
         program_subjects = utils.get_program_subjects(program_metadata)
         self.assertEqual(sorted(expected_subjects), sorted(program_subjects))
@@ -1311,7 +1311,7 @@ def test_get_program_level_type(self, program_metadata, course_metadata, expecte
             ContentMetadataFactory.create(
                 content_key=program_metadata['courses'][i]['key'],
                 content_type=COURSE,
-                json_metadata=course_metadata[i],
+                _json_metadata=course_metadata[i],
             )
         program_level_type = utils.get_program_level_type(program_metadata)
         self.assertEqual(expected_level_type, program_level_type)
@@ -1420,12 +1420,12 @@ def test_get_pathway_availability(self, pathway_metadata, json_metadata, expecte
             ContentMetadataFactory.create(
                 content_key=pathway_metadata['steps'][i]['courses'][i]['key'],
                 content_type=COURSE,
-                json_metadata=json_metadata[i],
+                _json_metadata=json_metadata[i],
             )
             ContentMetadataFactory.create(
                 content_key=pathway_metadata['steps'][i]['programs'][i]['uuid'],
                 content_type=PROGRAM,
-                json_metadata=json_metadata[i],
+                _json_metadata=json_metadata[i],
             )
 
         pathway_availability = utils.get_pathway_availability(pathway_metadata)
@@ -1526,12 +1526,12 @@ def test_get_pathway_partners(self, pathway_metadata, json_metadata, expected_pa
             ContentMetadataFactory.create(
                 content_key=pathway_metadata['steps'][i]['courses'][i]['key'],
                 content_type=COURSE,
-                json_metadata=json_metadata[0],
+                _json_metadata=json_metadata[0],
             )
             ContentMetadataFactory.create(
                 content_key=pathway_metadata['steps'][i]['programs'][i]['uuid'],
                 content_type=PROGRAM,
-                json_metadata=json_metadata[1],
+                _json_metadata=json_metadata[1],
             )
         pathway_partners = utils.get_pathway_partners(pathway_metadata)
         self.assertEqual(expected_partners, pathway_partners)
@@ -1586,12 +1586,12 @@ def test_get_pathway_subjects(self, pathway_metadata, json_metadata, expected_su
             ContentMetadataFactory.create(
                 content_key=pathway_metadata['steps'][i]['courses'][i]['key'],
                 content_type=COURSE,
-                json_metadata=json_metadata[0],
+                _json_metadata=json_metadata[0],
             )
             ContentMetadataFactory.create(
                 content_key=pathway_metadata['steps'][i]['programs'][i]['uuid'],
                 content_type=PROGRAM,
-                json_metadata=json_metadata[1],
+                _json_metadata=json_metadata[1],
             )
         pathway_subjects = utils.get_pathway_subjects(pathway_metadata)
         self.assertEqual(sorted(expected_subjects), sorted(pathway_subjects))
diff --git a/enterprise_catalog/apps/catalog/tests/test_models.py b/enterprise_catalog/apps/catalog/tests/test_models.py
index d9d69311c..056cd1302 100644
--- a/enterprise_catalog/apps/catalog/tests/test_models.py
+++ b/enterprise_catalog/apps/catalog/tests/test_models.py
@@ -44,7 +44,7 @@ def test_is_exec_ed_2u_course(self, content_type, course_type, expected_value):
             content_key='edX+testX',
             content_type=content_type,
         )
-        content_metadata.json_metadata['course_type'] = course_type
+        content_metadata._json_metadata['course_type'] = course_type  # pylint: disable=protected-access
         self.assertEqual(content_metadata.is_exec_ed_2u_course, expected_value)
 
     @ddt.data(
@@ -60,7 +60,7 @@ def test__should_allow_metadata(self, content_type, course_type, expected_value)
             content_key='edX+testX',
             content_type=content_type,
         )
-        content_metadata.json_metadata['course_type'] = course_type
+        content_metadata._json_metadata['course_type'] = course_type  # pylint: disable=protected-access
         self.assertEqual(_should_allow_metadata(content_metadata.json_metadata), expected_value)
 
     @override_settings(DISCOVERY_CATALOG_QUERY_CACHE_TIMEOUT=0)
@@ -479,7 +479,7 @@ def test_enrollment_url_exec_ed(
             content_key='the-content-key',
             content_type=COURSE,
         )
-        content_metadata.json_metadata.update({
+        content_metadata._json_metadata.update({  # pylint: disable=protected-access
             'course_type': EXEC_ED_2U_COURSE_TYPE,
             'entitlements': [{
                 'mode': EXEC_ED_2U_ENTITLEMENT_MODE,
@@ -584,9 +584,9 @@ def test_bulk_update_changes_modified_time(self):
         records = factories.ContentMetadataFactory.create_batch(10, modified=original_modified_time)
 
         for record in records:
-            record.json_metadata['extra_stuff'] = 'foo'
+            record._json_metadata['extra_stuff'] = 'foo'  # pylint: disable=protected-access
 
-        ContentMetadata.objects.bulk_update(records, ['json_metadata'], batch_size=10)
+        ContentMetadata.objects.bulk_update(records, ['_json_metadata'], batch_size=10)
 
         for record in records:
             record.refresh_from_db()
@@ -656,3 +656,510 @@ def test_restricted_runs_are_none(self, restricted_runs_dict):
 
         self.assertIsNone(catalog_query.restricted_runs_allowed)
         self.assertIsNone(catalog.restricted_runs_allowed)
+
+
+@ddt.ddt
+class TestRestrictedRunsModels(TestCase):
+    """
+    Tests for the following models pertaining to the "restricted runs" feature:
+    * RestrictedCourseMetadata
+    * RestrictedRunAllowedForRestrictedCourse
+    """
+
+    def _create_objects_and_relationships(
+        self,
+        create_catalog_query,
+        create_content_metadata=None,
+        create_restricted_courses=None,
+        create_restricted_run_allowed_for_restricted_course=None,
+    ):
+        """
+        """
+        catalog_queries = {
+            cq_uuid: factories.CatalogQueryFactory(
+                uuid=cq_uuid,
+                content_filter=cq_info['content_filter'] | {'force_unique': cq_uuid},
+            ) for cq_uuid, cq_info in create_catalog_query.items()
+        }
+        content_metadata = {}
+        create_content_metadata = create_content_metadata or {}
+        for course_key, course_info in create_content_metadata.items():
+            course = factories.ContentMetadataFactory(
+                content_key=course_key,
+                content_type=COURSE,
+                _json_metadata=course_info['json_metadata'],
+            )
+            content_metadata.update({course_key: course})
+            if cq_uuid := course_info['associate_with_catalog_query']:
+                course.catalog_queries.set([catalog_queries[cq_uuid]])
+            for run_key, run_info in course_info['create_runs'].items():
+                run = factories.ContentMetadataFactory(
+                    content_key=run_key,
+                    parent_content_key=course_key,
+                    content_type=COURSE_RUN,
+                )
+                if run_info['is_restricted']:
+                    # pylint: disable=protected-access
+                    run._json_metadata.update({'restriction_type': 'custom-b2b-enterprise'})
+                    run.save()
+                content_metadata.update({run_key: run})
+        restricted_courses = {
+            id: factories.RestrictedCourseMetadataFactory(
+                id=id,
+                content_key=restricted_course_info['content_key'],
+                unrestricted_parent=content_metadata[restricted_course_info['content_key']],
+                catalog_query=catalog_queries[restricted_course_info['catalog_query']],
+                _json_metadata=restricted_course_info['json_metadata'],
+            ) for id, restricted_course_info in create_restricted_courses.items()
+        } if create_restricted_courses else {}
+        for mapping_info in create_restricted_run_allowed_for_restricted_course or []:
+            factories.RestrictedRunAllowedForRestrictedCourseFactory(
+                course=restricted_courses[mapping_info['course']],
+                run=content_metadata[mapping_info['run']],
+            )
+        main_catalog = factories.EnterpriseCatalogFactory(
+            catalog_query=catalog_queries['11111111-1111-1111-1111-111111111111'],
+        )
+        return main_catalog, catalog_queries, content_metadata, restricted_courses
+
+    @ddt.data(
+        # Skip creating any content metadata at all. The result should be empty.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {},
+                },
+            },
+            'expected_json_metadata': [],
+            'expected_json_metadata_with_restricted': [],
+        },
+        # Create a simple course and run, but it is not part of the catalog. The result should be empty.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {},
+                },
+                '22222222-2222-2222-2222-222222222222': {
+                    'content_filter': {},
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '22222222-2222-2222-2222-222222222222',  # different!
+                },
+            },
+            'expected_json_metadata': [],
+            'expected_json_metadata_with_restricted': [],
+        },
+        # Create a simple course and run, associated with the catalog.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {},
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'expected_json_metadata': [
+                {'foobar': 'base metadata'},
+            ],
+            'expected_json_metadata_with_restricted': [
+                {'foobar': 'base metadata'},
+            ],
+        },
+        # Create a course with a restricted run, but the run is not allowed by the main CatalogQuery.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {
+                        'restricted_runs_allowed': {
+                            'course:edX+course': [
+                                'course-v1:edX+course+run2',
+                            ],
+                        },
+                    },
+                },
+                '22222222-2222-2222-2222-222222222222': {
+                    'content_filter': {},
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                        'course-v1:edX+course+run2': {'is_restricted': True},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'create_restricted_courses': {
+                0: {
+                    'content_key': 'edX+course',
+                    'catalog_query': '22222222-2222-2222-2222-222222222222',
+                    'json_metadata': {'foobar': 'override metadata'},
+                },
+            },
+            'create_restricted_run_allowed_for_restricted_course': [
+                {'course': 0, 'run': 'course-v1:edX+course+run2'},
+            ],
+            'expected_json_metadata': [
+                {'foobar': 'base metadata'},
+            ],
+            'expected_json_metadata_with_restricted': [
+                {'foobar': 'base metadata'},
+            ],
+        },
+        # Create a course with both an unrestricted (run1) and restricted run (run2), and the restricted run is allowed
+        # by the CatalogQuery.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {
+                        'restricted_runs_allowed': {
+                            'course:edX+course': [
+                                'course-v1:edX+course+run2',
+                            ],
+                        },
+                    },
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                        'course-v1:edX+course+run2': {'is_restricted': True},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'create_restricted_courses': {
+                0: {
+                    'content_key': 'edX+course',
+                    'catalog_query': '11111111-1111-1111-1111-111111111111',
+                    'json_metadata': {'foobar': 'override metadata'},
+                },
+            },
+            'create_restricted_run_allowed_for_restricted_course': [
+                {'course': 0, 'run': 'course-v1:edX+course+run2'},
+            ],
+            'expected_json_metadata': [
+                {'foobar': 'base metadata'},
+            ],
+            'expected_json_metadata_with_restricted': [
+                {'foobar': 'override metadata'},
+            ],
+        },
+        # Create a course with ONLY an unrestricted run (run1), and the restricted run is allowed by the CatalogQuery.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {
+                        'restricted_runs_allowed': {
+                            'course:edX+course': [
+                                'course-v1:edX+course+run2',
+                            ],
+                        },
+                    },
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': True},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'create_restricted_courses': {
+                0: {
+                    'content_key': 'edX+course',
+                    'catalog_query': '11111111-1111-1111-1111-111111111111',
+                    'json_metadata': {'foobar': 'override metadata'},
+                },
+            },
+            'create_restricted_run_allowed_for_restricted_course': [
+                {'course': 0, 'run': 'course-v1:edX+course+run1'},
+            ],
+            'expected_json_metadata': [
+                {'foobar': 'base metadata'},
+            ],
+            'expected_json_metadata_with_restricted': [
+                {'foobar': 'override metadata'},
+            ],
+        },
+    )
+    @ddt.unpack
+    def test_catalog_content_metadata_with_restricted_runs(
+        self,
+        create_catalog_query,
+        create_content_metadata=None,
+        create_restricted_courses=None,
+        create_restricted_run_allowed_for_restricted_course=None,
+        expected_json_metadata=None,
+        expected_json_metadata_with_restricted=None,
+    ):
+        """
+        Test the content_metadata() method of EnterpriseCatalog instances, as well as the newer
+        content_metadata_with_restricted() method.
+
+        The second method should cause courses to be serialized with json_metadata conditionally
+        overriden from a related RestrictedCourseMetadata instance, if one exists for the requested
+        catalog.
+        """
+        main_catalog, _, _, _ = self._create_objects_and_relationships(
+            create_catalog_query,
+            create_content_metadata,
+            create_restricted_courses,
+            create_restricted_run_allowed_for_restricted_course,
+        )
+        expected_json_metadata = expected_json_metadata or []
+        expected_json_metadata_with_restricted = expected_json_metadata_with_restricted or []
+        actual_json_metadata = [m.json_metadata for m in main_catalog.content_metadata]
+        actual_json_metadata_with_restricted = [m.json_metadata for m in main_catalog.content_metadata_with_restricted]
+        assert actual_json_metadata == expected_json_metadata
+        assert actual_json_metadata_with_restricted == expected_json_metadata_with_restricted
+
+    @ddt.data(
+        # Skip creating any content metadata at all. The result should be empty.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {},
+                },
+            },
+            'requested_content_keys': ['course-v1:edX+course+run1'],
+            'expected_json_metadata': [],
+            'expected_json_metadata_with_restricted': [],
+        },
+        # Create a simple course and run, but it is not part of the catalog. The result should be empty.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {},
+                },
+                '22222222-2222-2222-2222-222222222222': {
+                    'content_filter': {},
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '22222222-2222-2222-2222-222222222222',  # different!
+                },
+            },
+            'requested_content_keys': ['course-v1:edX+course+run1'],
+            'expected_json_metadata': [],
+            'expected_json_metadata_with_restricted': [],
+        },
+        # Create a simple course and run, associated with the catalog.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {},
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'requested_content_keys': ['course-v1:edX+course+run1'],
+            'expected_json_metadata': [
+                {'foobar': 'base metadata'},
+            ],
+            'expected_json_metadata_with_restricted': [
+                {'foobar': 'base metadata'},
+            ],
+        },
+        # Create a course with a restricted run, but the run is not allowed by the main CatalogQuery.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {
+                        'restricted_runs_allowed': {
+                            'course:edX+course': [
+                                'course-v1:edX+course+run2',
+                            ],
+                        },
+                    },
+                },
+                '22222222-2222-2222-2222-222222222222': {
+                    'content_filter': {},
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                        'course-v1:edX+course+run2': {'is_restricted': True},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'create_restricted_courses': {
+                0: {
+                    'content_key': 'edX+course',
+                    # Not the caller's catalog!
+                    'catalog_query': '22222222-2222-2222-2222-222222222222',
+                    'json_metadata': {'foobar': 'override metadata'},
+                },
+            },
+            'create_restricted_run_allowed_for_restricted_course': [
+                {'course': 0, 'run': 'course-v1:edX+course+run2'},
+            ],
+            'requested_content_keys': ['course-v1:edX+course+run2'],
+            'expected_json_metadata': [
+                # run2 was not found because it is restricted.
+            ],
+            'expected_json_metadata_with_restricted': [
+                # run2 was not found because it is not allowed by the caller's catalog.
+            ],
+        },
+        # Create a course with both an unrestricted (run1) and restricted run (run2), and the
+        # restricted run is allowed by the CatalogQuery. Request the UNRESTRICTED run (run1) anyway
+        # and assert that the override course metadata is returned.
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {
+                        'restricted_runs_allowed': {
+                            'course:edX+course': [
+                                'course-v1:edX+course+run2',
+                            ],
+                        },
+                    },
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        'course-v1:edX+course+run1': {'is_restricted': False},
+                        'course-v1:edX+course+run2': {'is_restricted': True},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'create_restricted_courses': {
+                0: {
+                    'content_key': 'edX+course',
+                    'catalog_query': '11111111-1111-1111-1111-111111111111',
+                    'json_metadata': {'foobar': 'override metadata'},
+                },
+            },
+            'create_restricted_run_allowed_for_restricted_course': [
+                {'course': 0, 'run': 'course-v1:edX+course+run2'},
+            ],
+            'requested_content_keys': ['course-v1:edX+course+run1'],
+            'expected_json_metadata': [
+                {'foobar': 'base metadata'},
+            ],
+            'expected_json_metadata_with_restricted': [
+                # We supply the override course metadata *even though* the
+                # requested content key represents an unrestricted run.
+                {'foobar': 'override metadata'},
+            ],
+        },
+        # Create a course with ONLY an unrestricted run (run1), and the restricted run is allowed by the CatalogQuery.
+        # This type of course has colloquially been referred to as "Unicorn".
+        {
+            'create_catalog_query': {
+                '11111111-1111-1111-1111-111111111111': {
+                    'content_filter': {
+                        'restricted_runs_allowed': {
+                            'course:edX+course': [
+                                'course-v1:edX+course+run2',
+                            ],
+                        },
+                    },
+                },
+            },
+            'create_content_metadata': {
+                'edX+course': {
+                    'create_runs': {
+                        # The only run is a restricted run.
+                        'course-v1:edX+course+run1': {'is_restricted': True},
+                    },
+                    'json_metadata': {'foobar': 'base metadata'},
+                    'associate_with_catalog_query': '11111111-1111-1111-1111-111111111111',
+                },
+            },
+            'create_restricted_courses': {
+                0: {
+                    'content_key': 'edX+course',
+                    'catalog_query': '11111111-1111-1111-1111-111111111111',
+                    'json_metadata': {'foobar': 'override metadata'},
+                },
+            },
+            'create_restricted_run_allowed_for_restricted_course': [
+                {'course': 0, 'run': 'course-v1:edX+course+run1'},
+            ],
+            'requested_content_keys': ['course-v1:edX+course+run1'],
+            'expected_json_metadata': [
+                # The RUN is invisible to the requester because it is restricted, so the COURSE
+                # should not be found either.
+            ],
+            'expected_json_metadata_with_restricted': [
+                {'foobar': 'override metadata'},
+            ],
+        },
+    )
+    @ddt.unpack
+    def test_get_matching_content_with_restricted_runs(
+        self,
+        create_catalog_query,
+        create_content_metadata=None,
+        create_restricted_courses=None,
+        create_restricted_run_allowed_for_restricted_course=None,
+        requested_content_keys=None,
+        expected_json_metadata=None,
+        expected_json_metadata_with_restricted=None,
+    ):
+        """
+        Test the get_matching_content() method of EnterpriseCatalog instances, both with and without
+        passing include_restricted=True.
+
+        An example of how we expect behavior to change after passing include_restricted:
+        If the requester's catalog allows a restricted run, then passing that run into
+        get_matching_content(include_restricted=False) will yield an empty list, whereas
+        get_matching_content(include_restricted=True) will return the parent course.
+
+        Restricted runs should always seem non-existent by default.
+        """
+        main_catalog, _, _, _ = self._create_objects_and_relationships(
+            create_catalog_query,
+            create_content_metadata,
+            create_restricted_courses,
+            create_restricted_run_allowed_for_restricted_course,
+        )
+        requested_content_keys = requested_content_keys or []
+        expected_json_metadata = expected_json_metadata or []
+        expected_json_metadata_with_restricted = expected_json_metadata_with_restricted or []
+        actual_json_metadata = [m.json_metadata for m in main_catalog.get_matching_content(requested_content_keys)]
+        actual_json_metadata_with_restricted = [
+            m.json_metadata for m in main_catalog.get_matching_content(requested_content_keys, include_restricted=True)
+        ]
+        assert actual_json_metadata == expected_json_metadata
+        assert actual_json_metadata_with_restricted == expected_json_metadata_with_restricted