fix: prevent SSRF in the Studio video download endpoint

The PUT /api/contentstore/v1/videos/{course_id}/download endpoint fetched
every client-supplied files[].url server-side with
requests.get(url, allow_redirects=True) and returned the bytes inside the
ZIP response. Because the URLs were never validated, an authenticated user
with studio read access could point them at internal services or cloud
metadata endpoints and exfiltrate the responses (GHSA-fpf9-9rpr-jvrx).

By design these URLs are always a subset of the course's own VAL
encoded_videos[].url values (the same data the video listing hands the
frontend). Restrict fetches to that allowlist: build the set of legitimate
URLs for the course and reject any request containing a URL outside it
before any HTTP request is made. This eliminates the SSRF rather than
merely narrowing it.

Adds VideoDownloadViewTest (the endpoint previously had no test coverage)
covering the allowed-URL success path, rejection of disallowed URLs without
any outbound request, mixed allowed/disallowed batches, and the non-staff
permission gate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: feanil

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,107 @@ 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']