fix: port security advisory patches to teak (BB-10886)

README.rst

@@ -195,6 +195,23 @@ separately. At a bare minimum, you will need to run the `Authentication MFE`_,
 .. _Learner Home MFE: https://github.com/openedx/frontend-app-learner-dashboard
 .. _Learning MFE: https://github.com/openedx/frontend-app-learning/
 
+Security Deployment Requirements
+********************************
+
+Some platform features require a **shared** Django cache backend (Redis or
+Memcached) to function correctly across multiple LMS nodes:
+
+* **LTI Provider** — OAuth nonce replay protection stores seen nonces in the
+  Django ``default`` cache. A per-process backend (e.g. ``LocMemCache``) will
+  not detect replays that arrive on a different node. See
+  `lms/djangoapps/lti_provider/README.rst`_ for details.
+
+Tutor-based deployments satisfy this requirement automatically. For bare-metal
+or custom deployments, verify that ``CACHES['default']`` points at a shared
+Redis or Memcached instance before enabling these features.
+
+.. _lms/djangoapps/lti_provider/README.rst: lms/djangoapps/lti_provider/README.rst
+
 License
 *******
 

cms/djangoapps/contentstore/rest_api/v1/views/tests/test_videos.py

@@ -1,20 +1,25 @@
 """
 Unit tests for course settings views.
 """
-from unittest.mock import patch
+from datetime import datetime
+from unittest.mock import MagicMock, patch
 
 import ddt
+import pytz
 from django.conf import settings
 from django.contrib.staticfiles.storage import staticfiles_storage
 from django.urls import reverse
 from edx_toggles.toggles import WaffleSwitch
 from edx_toggles.toggles.testutils import override_waffle_switch
 from edxval.api import (
+    create_profile,
+    create_video,
     get_3rd_party_transcription_plans,
     get_transcript_credentials_state_for_org,
     get_transcript_preferences,
 )
 from rest_framework import status
+from rest_framework.test import APIClient
 
 from cms.djangoapps.contentstore.video_storage_handlers import get_all_transcript_languages
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
@@ -135,3 +140,106 @@ def test_VideoTranscriptEnabledFlag_enabled(self):
             response = self.client.get(self.url)
             self.assertIn("is_ai_translations_enabled", response.data)
             self.assertTrue(response.data["is_ai_translations_enabled"])
+
+
+class VideoDownloadViewTest(CourseTestCase):
+    """
+    Tests for VideoDownloadView.
+
+    The download endpoint fetches each requested ``files[].url`` server-side and
+    returns the bytes inside a zip. Those URLs must therefore be restricted to
+    the course's own video URLs, otherwise the endpoint is an SSRF primitive
+    (see GHSA-fpf9-9rpr-jvrx).
+    """
+
+    ALLOWED_URL = "http://example.com/profile1/test.mp4"
+    # An internal address an attacker might try to reach via SSRF.
+    SSRF_URL = "http://169.254.169.254/latest/meta-data/"
+
+    def setUp(self):
+        super().setUp()
+        # reverse() with only course_id resolves to the download route (the
+        # usage route with the same name additionally requires edx_video_id).
+        self.url = reverse(
+            "cms.djangoapps.contentstore:v1:video_usage",
+            kwargs={"course_id": self.course.id},
+        )
+        self.api_client = APIClient()
+        self.api_client.force_authenticate(user=self.user)
+        create_profile("profile1")
+        create_video({
+            "edx_video_id": "test-video",
+            "client_video_id": "test.mp4",
+            "duration": 42.0,
+            "status": "file_complete",
+            "courses": [str(self.course.id)],
+            "created": datetime.now(pytz.utc),
+            "encoded_videos": [
+                {
+                    "profile": "profile1",
+                    "url": self.ALLOWED_URL,
+                    "file_size": 1600,
+                    "bitrate": 100,
+                },
+            ],
+        })
+
+    @patch("cms.djangoapps.contentstore.video_storage_handlers.requests.get")
+    def test_download_allowed_url(self, mock_get):
+        """A URL that belongs to the course's videos is fetched and zipped."""
+        mock_get.return_value = MagicMock(
+            content=b"video-bytes",
+            headers={"Content-Type": "video/mp4"},
+        )
+        response = self.api_client.put(
+            self.url,
+            data={"files": [{"url": self.ALLOWED_URL, "name": "test.mp4"}]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)  # noqa: PT009
+        mock_get.assert_called_once_with(self.ALLOWED_URL, allow_redirects=True)
+
+    @patch("cms.djangoapps.contentstore.video_storage_handlers.requests.get")
+    def test_rejects_url_not_belonging_to_course(self, mock_get):
+        """
+        A URL that is not one of the course's video URLs is rejected before any
+        server-side request is made (SSRF protection).
+        """
+        response = self.api_client.put(
+            self.url,
+            data={"files": [{"url": self.SSRF_URL, "name": "evil.txt"}]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)  # noqa: PT009
+        mock_get.assert_not_called()
+
+    @patch("cms.djangoapps.contentstore.video_storage_handlers.requests.get")
+    def test_rejects_when_any_url_is_disallowed(self, mock_get):
+        """
+        A request mixing an allowed URL with a disallowed one is rejected
+        outright, without fetching the allowed URL either.
+        """
+        response = self.api_client.put(
+            self.url,
+            data={"files": [
+                {"url": self.ALLOWED_URL, "name": "test.mp4"},
+                {"url": self.SSRF_URL, "name": "evil.txt"},
+            ]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)  # noqa: PT009
+        mock_get.assert_not_called()
+
+    @patch("cms.djangoapps.contentstore.video_storage_handlers.requests.get")
+    def test_non_staff_user_denied(self, mock_get):
+        """A user without studio read access cannot reach the fetch path."""
+        __, nonstaff_user = self.create_non_staff_authed_user_client()
+        client = APIClient()
+        client.force_authenticate(user=nonstaff_user)
+        response = client.put(
+            self.url,
+            data={"files": [{"url": self.ALLOWED_URL, "name": "test.mp4"}]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)  # noqa: PT009
+        mock_get.assert_not_called()

cms/djangoapps/contentstore/video_storage_handlers.py

@@ -47,6 +47,7 @@
 from path import Path as path
 from pytz import UTC
 from rest_framework import status as rest_status
+from rest_framework.exceptions import ValidationError
 from rest_framework.response import Response
 from tempfile import NamedTemporaryFile, mkdtemp
 from wsgiref.util import FileWrapper
@@ -242,6 +243,29 @@ def send_zip(zip_file, size=None):
     return response
 
 
+def get_course_video_download_urls(course_key_string):
+    """
+    Return the set of encoded-video URLs that legitimately belong to the given
+    course, as recorded in VAL.
+
+    The video download endpoint only ever needs to fetch URLs that were already
+    surfaced to the client by the video listing. Restricting fetches to this set
+    prevents server-side request forgery (SSRF) via attacker-supplied URLs.
+    """
+    videos, __ = get_videos_for_course(
+        course_key_string,
+        VideoSortField.created,
+        SortDirection.desc,
+        None,
+    )
+    return {
+        encoding['url']
+        for video in videos
+        for encoding in video['encoded_videos']
+        if encoding.get('url')
+    }
+
+
 def create_video_zip(course_key_string, files):
     """
     Generates the video zip, or returns None if there was an error.
@@ -254,6 +278,13 @@ def create_video_zip(course_key_string, files):
     root_dir = path(mkdtemp())
     video_dir = root_dir + '/' + name
     zip_folder = None
+    # Only allow fetching URLs that belong to this course's videos. Anything
+    # else (internal services, cloud metadata endpoints, arbitrary hosts) is a
+    # potential SSRF target and is rejected before any request is made.
+    allowed_urls = get_course_video_download_urls(course_key_string)
+    for file in files:
+        if file['url'] not in allowed_urls:
+            raise ValidationError(f"Invalid video download url: {file['url']}")
     try:
         for file in files:
             url = file['url']

lms/djangoapps/instructor/views/instructor_dashboard.py

@@ -420,6 +420,11 @@ def set_course_mode_price(request, course_id):
     """
     set the new course price and add new entry in the CourseModesArchive Table
     """
+    if not request.user.is_staff:
+        return JsonResponse(
+            {'message': _("You do not have permission to perform this action.")},
+            status=403
+        )
     try:
         course_price = int(request.POST['course_price'])
     except ValueError:

lms/djangoapps/lti_provider/README.rst

@@ -0,0 +1,31 @@
+LTI Provider
+############
+
+This Django app implements the server (provider) side of the `1EdTech LTI
+1.1`_ specification, allowing external tools and platforms to launch content
+inside the Open edX LMS using OAuth 1.0-signed requests. LTI 1.2 and 1.3
+are not supported by this app.
+
+.. _1EdTech LTI 1.1: https://www.imsglobal.org/lti/ltiv1p2/ltiIMGv1p2.html
+
+Security Requirements
+*********************
+
+Shared Cache Backend (Required for Multi-Node Deployments)
+===========================================================
+
+The LTI provider protects against OAuth replay attacks by storing each
+``oauth_nonce`` in Django's cache after it is first seen and rejecting any
+subsequent request that presents the same nonce within the validity window
+(±5 minutes around the ``oauth_timestamp``).
+
+**This protection only works correctly when all LMS nodes share the same cache
+backend.** If you run more than one LMS process or server, you must configure
+Django's ``default`` cache to use a shared backend such as Redis or Memcached.
+A per-process backend (e.g. Django's built-in ``LocMemCache``) keeps a
+separate in-memory store per process, so a replayed request arriving on a
+different node will not be detected.
+
+Tutor-based deployments use Redis by default and satisfy this requirement
+automatically. For bare-metal or custom deployments, verify that
+``CACHES['default']`` is pointed at a shared Redis or Memcached instance.

lms/djangoapps/lti_provider/signature_validator.py

@@ -2,7 +2,10 @@
 Subclass of oauthlib's RequestValidator that checks an OAuth signature.
 """
 
+import time
 
+from django.core.cache import cache
+from edx_django_utils.cache import get_cache_key
 from oauthlib.oauth1 import RequestValidator, SignatureOnlyEndpoint
 
 
@@ -63,10 +66,37 @@ def validate_timestamp_and_nonce(self, client_key, timestamp, nonce,
         in which the timestamp marks a request as valid. This method signature
         is required by the oauthlib library.
 
+        Rejects requests whose timestamp falls outside a ±5-minute window, then
+        performs an atomic check-and-store of the nonce so that replayed requests
+        are rejected.
+
+        We use Django's cache.add() directly rather than TieredCache because
+        nonce replay prevention requires an atomic check-and-store: cache.add()
+        only writes the key when it does not already exist and returns False
+        immediately if it does. TieredCache has no equivalent atomic primitive,
+        so a separate get-then-set would leave a race window that an attacker
+        could exploit.
+
+        Security requirement: this protection only holds across all LMS nodes if
+        the Django cache backend is shared (e.g. Redis or Memcached). A
+        per-process backend such as LocMemCache will not prevent replays that
+        arrive on a different node. See lms/djangoapps/lti_provider/README.rst.
+
         :return: True if the OAuth nonce and timestamp are valid, False if they
         are not.
         """
-        return True
+        try:
+            if abs(time.time() - int(timestamp)) > 300:
+                return False
+        except (ValueError, TypeError):
+            return False
+
+        # Nonces are scoped per client_key so that two different LTI consumers
+        # generating the same nonce string do not interfere with each other.
+        cache_key = get_cache_key(namespace="lti_nonce", client_key=client_key, nonce=nonce)
+        # cache.add() atomically sets the key only if absent.
+        # Returns True (new nonce, allow) or False (already seen, reject).
+        return cache.add(cache_key, 1, 600)
 
     def validate_client_key(self, client_key, request):
         """

lms/djangoapps/lti_provider/tests/test_signature_validator.py

@@ -11,6 +11,9 @@
 
 from lms.djangoapps.lti_provider.models import LtiConsumer
 from lms.djangoapps.lti_provider.signature_validator import SignatureValidator
+from openedx.core.djangolib.testing.utils import CacheIsolationMixin
+
+FIXED_TIMESTAMP = 1_000_000_000
 
 
 def get_lti_consumer():
@@ -118,3 +121,49 @@ def test_verification_parameters(self, verify_mock):
         SignatureValidator(self.lti_consumer).verify(request)
         verify_mock.assert_called_once_with(
             request.build_absolute_uri(), 'POST', body.encode('utf-8'), headers)
+
+
+@patch('lms.djangoapps.lti_provider.signature_validator.time.time', return_value=FIXED_TIMESTAMP)
+class TimestampAndNonceValidatorTest(CacheIsolationMixin, TestCase):
+    ENABLED_CACHES = ['default']
+    """
+    Tests for validate_timestamp_and_nonce. The clock is frozen via a class-level
+    patch so every test method receives a consistent `time.time` return value.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.validator = SignatureValidator(get_lti_consumer())
+
+    def _call(self, timestamp, nonce, client_key='Consumer Key'):
+        return self.validator.validate_timestamp_and_nonce(
+            client_key, str(timestamp), nonce, request=None,
+        )
+
+    def test_valid_timestamp_and_new_nonce(self, _mock_time):
+        assert self._call(FIXED_TIMESTAMP, 'nonce-a')
+
+    def test_stale_timestamp_rejected(self, _mock_time):
+        assert not self._call(FIXED_TIMESTAMP - 301, 'nonce-b')
+
+    def test_future_timestamp_rejected(self, _mock_time):
+        assert not self._call(FIXED_TIMESTAMP + 301, 'nonce-c')
+
+    def test_malformed_timestamp_rejected(self, _mock_time):
+        assert not self.validator.validate_timestamp_and_nonce(
+            'Consumer Key', 'not-a-number', 'nonce-d', request=None,
+        )
+
+    def test_replay_rejected(self, _mock_time):
+        assert self._call(FIXED_TIMESTAMP, 'nonce-e')
+        assert not self._call(FIXED_TIMESTAMP, 'nonce-e')
+
+    def test_different_nonce_same_consumer_accepted(self, _mock_time):
+        assert self._call(FIXED_TIMESTAMP, 'nonce-f1')
+        assert self._call(FIXED_TIMESTAMP, 'nonce-f2')
+
+    def test_same_nonce_different_consumer_accepted(self, _mock_time):
+        # Nonces are scoped per client_key; the same nonce string from two
+        # different consumers must not block each other.
+        assert self._call(FIXED_TIMESTAMP, 'nonce-g', client_key='Consumer Key')
+        assert self._call(FIXED_TIMESTAMP, 'nonce-g', client_key='Other Key')