chore(ulmo): backport security patch

lms/djangoapps/course_home_api/outline/tests/test_view.py

@@ -23,6 +23,7 @@
 from lms.djangoapps.course_home_api.toggles import COURSE_HOME_SEND_COURSE_PROGRESS_ANALYTICS_FOR_STUDENT
 from lms.djangoapps.course_home_api.tests.utils import BaseCourseHomeTests
 from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
+from openedx.core.djangoapps.content.block_structure.api import update_course_in_cache
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 from openedx.core.djangoapps.content.learning_sequences.api import replace_course_outline
 from openedx.core.djangoapps.content.learning_sequences.data import CourseOutlineData, CourseVisibility
@@ -900,3 +901,55 @@ def test_vertical_icon_determined_by_icon_class(self):
         response = self.client.get(reverse('course-home:course-navigation', args=[self.course.id]))
         vertical_data = response.data['blocks'][str(self.vertical.location)]
         assert vertical_data['icon'] == 'video'
+
+    def test_navigation_does_not_cache_stale_data_after_publish(self):
+        """
+        Regression test: after the block structure rebuild task completes,
+        the navigation sidebar should serve fresh data.
+
+        This simulates a production scenario where:
+        1. A unit is deleted and the course is auto-published
+        2. The block structure rebuild Celery task is queued with a delay (30s by default)
+        3. A learner hits the navigation endpoint during that 30s window
+        4. The rebuild task completes (bumping block_structure_version)
+        5. Another request arrives
+
+        Without the fix, step 3 caches stale data under a key that step 5
+        also hits (because course_version changed eagerly). With the fix,
+        the cache key uses block_structure_version which only changes when
+        the rebuild completes, so step 5 gets a cache miss and fresh data.
+        """
+        self.add_blocks_to_course()
+        CourseEnrollment.enroll(self.user, self.course.id, CourseMode.VERIFIED)
+
+        # First request — populates both block structure and navigation cache
+        response = self.client.get(self.url)
+        assert response.status_code == 200
+        sequential_data = response.data['blocks'][str(self.sequential.location)]
+        assert str(self.vertical.location) in sequential_data['children']
+
+        # Delete the vertical directly in the modulestore. Signals are disabled
+        # in ModuleStoreTestCase, so the block structure cache is now stale —
+        # mirroring the 30s window in production before the rebuild task runs.
+        self.store.delete_item(self.vertical.location, self.user.id)
+        update_outline_from_modulestore(self.course.id)
+
+        # Request during the stale window — served from the pre-delete cache
+        # (block_structure_version hasn't changed yet, so same cache key).
+        response = self.client.get(self.url)
+        assert response.status_code == 200
+
+        # The vertical is still in the cache, even though it has been deleted
+        sequential_data = response.data['blocks'][str(self.sequential.location)]
+        assert str(self.vertical.location) in sequential_data['children']
+
+        # Now simulate the block structure rebuild task completing.
+        # This bumps block_structure_version → new cache key on next request.
+        update_course_in_cache(self.course.id)
+
+        # Next request has a new cache key (version bumped) → cache miss →
+        # fresh data built from updated block structure.
+        response = self.client.get(self.url)
+        assert response.status_code == 200
+        sequential_data = response.data['blocks'][str(self.sequential.location)]
+        assert str(self.vertical.location) not in sequential_data['children']

lms/djangoapps/course_home_api/outline/views.py

@@ -46,6 +46,7 @@
 from lms.djangoapps.courseware.views.views import get_cert_data
 from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
 from lms.djangoapps.utils import OptimizelyClient
+from openedx.core.djangoapps.content.block_structure.api import get_block_structure_version
 from openedx.core.djangoapps.content.learning_sequences.api import get_user_course_outline
 from openedx.core.djangoapps.content.course_overviews.api import get_course_overview_or_404
 from openedx.core.djangoapps.course_groups.cohorts import get_cohort
@@ -423,7 +424,7 @@ class CourseNavigationBlocksView(RetrieveAPIView):
 
     serializer_class = CourseBlockSerializer
     COURSE_BLOCKS_CACHE_KEY_TEMPLATE = (
-        'course_sidebar_blocks_{course_key_string}_{course_version}_{user_id}_{user_cohort_id}'
+        'course_sidebar_blocks_{course_key_string}_{block_structure_version}_{user_id}_{user_cohort_id}'
         '_{enrollment_mode}_{allow_public}_{allow_public_outline}_{is_masquerading}'
     )
     COURSE_BLOCKS_CACHE_TIMEOUT = 60 * 60  # 1 hour
@@ -458,7 +459,7 @@ def get(self, request, *args, **kwargs):
 
         cache_key = self.COURSE_BLOCKS_CACHE_KEY_TEMPLATE.format(
             course_key_string=course_key_string,
-            course_version=str(course.course_version),
+            block_structure_version=get_block_structure_version(course_key),
             user_id=request.user.id,
             enrollment_mode=getattr(enrollment, 'mode', ''),
             user_cohort_id=getattr(user_cohort, 'id', ''),

openedx/core/djangoapps/content/block_structure/api.py

@@ -8,6 +8,47 @@
 from xmodule.modulestore.django import modulestore
 
 from .manager import BlockStructureManager
+from .models import BlockStructureModel
+
+BLOCK_STRUCTURE_VERSION_KEY = 'block_structure_version:{}'
+
+
+def get_block_structure_version(course_key):
+    """
+    Returns the current block structure version for the given course.
+    This version corresponds to the data_version stored in BlockStructureModel
+    and changes each time the block structure cache is rebuilt.
+
+    Reads from cache first; on a miss, falls back to the database
+    without populating the cache. The cache is populated exclusively by
+    _update_block_structure_version after a successful rebuild, which
+    prevents readers from accidentally caching a stale version during
+    a concurrent rebuild.
+    """
+    cache_key = BLOCK_STRUCTURE_VERSION_KEY.format(course_key)
+    version = cache.get(cache_key)
+    if version is None:
+        try:
+            course_usage_key = modulestore().make_course_usage_key(course_key)
+            block_structure_model = BlockStructureModel.objects.get(data_usage_key=course_usage_key)
+            version = str(block_structure_model.data_version or '')
+        except BlockStructureModel.DoesNotExist:
+            version = ''
+    return version
+
+
+def _update_block_structure_version(course_key):
+    """
+    Reads the current data_version from BlockStructureModel and updates
+    the cached block structure version key.
+    """
+    try:
+        course_usage_key = modulestore().make_course_usage_key(course_key)
+        block_structure_model = BlockStructureModel.objects.get(data_usage_key=course_usage_key)
+        version = str(block_structure_model.data_version or '')
+    except BlockStructureModel.DoesNotExist:
+        version = ''
+    cache.set(BLOCK_STRUCTURE_VERSION_KEY.format(course_key), version, timeout=None)
 
 
 def get_course_in_cache(course_key):
@@ -29,7 +70,8 @@ def update_course_in_cache(course_key):
     block_structure.updated_collected function that updates the block
     structure in the cache for the given course_key.
     """
-    return get_block_structure_manager(course_key).update_collected_if_needed()
+    get_block_structure_manager(course_key).update_collected_if_needed()
+    _update_block_structure_version(course_key)
 
 
 def clear_course_from_cache(course_key):

openedx/core/djangoapps/notifications/base_notification.py

@@ -1,6 +1,7 @@
 """
 Base setup for Notification Apps and Types.
 """
+from django.utils.html import escape
 from django.utils.translation import gettext_lazy as _
 
 from .email_notifications import EmailCadence
@@ -11,6 +12,13 @@
 
 FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE = 'filter_audit_expired_users_with_no_role'
 
+# Context keys whose values are used as HTML tag names by content_templates
+# (e.g. `<{p}>...<{strong}>{post_title}</{strong}></{p}>`). These must NOT be
+# HTML-escaped before `template.format(**context)`; every other context value
+# must be, since it typically comes from user input (thread title, username,
+# etc.). See get_notification_content below.
+_STRUCTURAL_CONTEXT_KEYS = frozenset({'p', 'strong'})
+
 COURSE_NOTIFICATION_TYPES = {
     'new_comment_on_response': {
         'notification_app': 'discussion',
@@ -538,8 +546,16 @@ def get_notification_content(notification_type, context):
         context = context_function(context)
 
         if template:
-            # Handle grouped templates differently by modifying the context using a different function.
-            return template.format(**context)
+            # HTML-escape every context value except the structural tag-name
+            # keys, so that user-controlled input (post_title, replier_name,
+            # etc.) cannot inject `<style>` / `<script>` / other HTML into
+            # notification.content — which is rendered with `|safe` in the
+            # digest and batched email templates.
+            safe_context = {
+                key: value if key in _STRUCTURAL_CONTEXT_KEYS else escape(value)
+                for key, value in context.items()
+            }
+            return template.format(**safe_context)
 
     return ''
 

openedx/core/djangoapps/notifications/tests/test_base_notification.py

@@ -1,6 +1,8 @@
 """
 Tests for base_notification
 """
+import pytest
+
 from common.djangoapps.student.tests.factories import UserFactory
 from openedx.core.djangoapps.notifications import base_notification, models
 from openedx.core.djangoapps.notifications.models import (
@@ -262,3 +264,40 @@ def test_validate_non_core_notification_types(self):
                 assert isinstance(notification_type[key], str)
             for key in bool_keys:
                 assert isinstance(notification_type[key], bool)
+
+
+@pytest.mark.parametrize(
+    ('user_input', 'escaped'),
+    [
+        ('<style>body{background:red}</style>evil', '&lt;style&gt;body{background:red}&lt;/style&gt;evil'),
+        ('<script>alert(1)</script>', '&lt;script&gt;alert(1)&lt;/script&gt;'),
+        ('AT&T "quoted"', 'AT&amp;T &quot;quoted&quot;'),
+    ],
+)
+def test_get_notification_content_escapes_user_input(user_input, escaped):
+    """
+    Regression test for GHSA-rv5w-f4r5-h77g: user-controlled context values
+    must be HTML-escaped before being interpolated into a content_template
+    via `str.format`. Structural context keys (`p`, `strong`) are exempt so
+    the template can still emit real <p>/<strong> tags.
+    """
+    context = {'replier_name': 'alice', 'post_title': user_input}
+    content = base_notification.get_notification_content('new_response', context)
+    assert '<style>' not in content
+    assert '<script>' not in content
+    assert escaped in content
+
+
+def test_get_notification_content_preserves_structural_tags():
+    """
+    Companion to test_get_notification_content_escapes_user_input: verify
+    that the structural `p` and `strong` keys still produce real HTML tags
+    after the escape pass, and that innocuous user input renders as plain
+    text alongside them.
+    """
+    context = {'replier_name': 'alice', 'post_title': 'Hello world'}
+    content = base_notification.get_notification_content('new_response', context)
+    assert '<p>' in content
+    assert '</p>' in content
+    assert '<strong>alice</strong>' in content
+    assert '<strong>Hello world</strong>' in content

openedx/core/lib/extract_archive.py

@@ -7,7 +7,7 @@
 """
 
 import logging
-from os.path import abspath, dirname
+from os.path import abspath, commonpath, dirname
 from os.path import join as joinpath
 from os.path import realpath
 from typing import List, Union
@@ -30,8 +30,18 @@ def resolved(rpath):
 def _is_bad_path(path, base):
     """
     Is (the canonical absolute path of) `path` outside `base`?
+
+    Uses ``os.path.commonpath`` for a segment-aware containment check so that
+    sibling directories whose names happen to extend ``base`` as a string
+    prefix (e.g. ``<base>evil/...``) are correctly classified as outside.
     """
-    return not resolved(joinpath(base, path)).startswith(base)
+    target = resolved(joinpath(base, path))
+    try:
+        return commonpath([target, base]) != base
+    except ValueError:
+        # commonpath raises when paths are incomparable (e.g. mixed absolute
+        # and relative, or different drives on Windows). Treat as bad.
+        return True
 
 
 def _is_bad_link(info, base):

openedx/core/lib/tests/test_extract_archive.py

@@ -0,0 +1,94 @@
+"""
+Tests for openedx.core.lib.extract_archive.
+
+The path-containment helpers must treat ``base`` as a directory boundary, not
+a raw string prefix: a sibling whose name extends ``base`` as a string prefix
+(e.g. ``<parent>/foo_evil/x`` next to ``<parent>/foo``) is *outside* ``base``.
+These tests pin that boundary down at the helper level and end-to-end through
+``safe_extractall``.
+"""
+
+import io
+import os
+import tarfile
+import tempfile
+
+import pytest
+from django.core.exceptions import SuspiciousOperation
+from django.test import override_settings
+
+from openedx.core.lib.extract_archive import _is_bad_path, safe_extractall
+
+
+# Direct tests of the path-containment helper. No Django settings needed.
+
+def test_is_bad_path_prefix_bypass_is_rejected():
+    """
+    A sibling path whose name extends the base's name as a raw string prefix
+    (e.g. ``<parent>/Y29evil/file`` vs base ``<parent>/Y29``) is outside
+    ``base`` and must be flagged as bad.
+    """
+    with tempfile.TemporaryDirectory() as parent:
+        base = os.path.join(parent, "Y29")
+        os.mkdir(base)
+        assert _is_bad_path("../Y29evil/file", base) is True
+
+
+def test_is_bad_path_inside_base_is_accepted():
+    with tempfile.TemporaryDirectory() as parent:
+        base = os.path.join(parent, "Y29")
+        os.mkdir(base)
+        assert _is_bad_path("nested/file.txt", base) is False
+
+
+def test_is_bad_path_traversal_outside_base_is_rejected():
+    with tempfile.TemporaryDirectory() as parent:
+        base = os.path.join(parent, "Y29")
+        os.mkdir(base)
+        assert _is_bad_path("../../etc/passwd", base) is True
+
+
+# End-to-end tests of safe_extractall against crafted .tar.gz archives.
+
+def _add_file(tar, name, content=b""):
+    info = tarfile.TarInfo(name=name)
+    info.size = len(content)
+    tar.addfile(info, io.BytesIO(content))
+
+
+def _add_symlink(tar, name, linkname):
+    info = tarfile.TarInfo(name=name)
+    info.type = tarfile.SYMTYPE
+    info.linkname = linkname
+    tar.addfile(info)
+
+
+def test_safe_extractall_blocks_file_entry_with_prefix_bypass(tmp_path):
+    root = str(tmp_path)
+    extract_dir = os.path.join(root, "Y29")
+    os.mkdir(extract_dir)
+    archive = os.path.join(root, "malicious.tar.gz")
+    with tarfile.open(archive, "w:gz") as tar:
+        _add_file(tar, "../Y29evil/sentinel.txt", b"owned")
+
+    with override_settings(GITHUB_REPO_ROOT=root):
+        with pytest.raises(SuspiciousOperation):
+            safe_extractall(archive, extract_dir)
+
+    escape_target = os.path.join(root, "Y29evil", "sentinel.txt")
+    assert not os.path.exists(escape_target)
+
+
+def test_safe_extractall_blocks_symlink_target_with_prefix_bypass(tmp_path):
+    root = str(tmp_path)
+    extract_dir = os.path.join(root, "Y29")
+    os.mkdir(extract_dir)
+    archive = os.path.join(root, "malicious.tar.gz")
+    # symlink inside extract_dir pointing at a sibling whose name extends
+    # extract_dir's basename as a string prefix.
+    with tarfile.open(archive, "w:gz") as tar:
+        _add_symlink(tar, name="link", linkname="../Y29evil/secret")
+
+    with override_settings(GITHUB_REPO_ROOT=root):
+        with pytest.raises(SuspiciousOperation):
+            safe_extractall(archive, extract_dir)