prevent request forgery

learning_resources/etl/ovs.py

@@ -3,7 +3,7 @@
 import logging
 from collections import defaultdict
 from collections.abc import Generator
-from urllib.parse import urlencode, urlparse, urlunparse
+from urllib.parse import quote, urlencode, urlparse, urlunparse
 
 import requests
 from django.conf import settings
@@ -25,10 +25,56 @@
 OVS_API_PATH = "/api/v0/public/videos/"
 
 
+def allowed_media_hosts() -> set[str]:
+    """
+    Return the set of hosts that OVS media urls are allowed to point at.
+
+    Returns:
+        set of lowercased host patterns, including the OVS API host
+    """
+    hosts = {host.lower() for host in settings.OVS_ALLOWED_MEDIA_HOSTS if host}
+    api_host = urlparse(settings.OVS_API_BASE_URL or "").hostname
+    if api_host:
+        hosts.add(api_host.lower())
+    return hosts
+
+
+def is_allowed_media_url(url: str | None) -> bool:
+    """
+    Check that a url from an OVS payload points at an allowlisted host over https.
+
+    OVS payloads (including webhook payloads, which are untrusted input) carry
+    absolute urls for thumbnails, streaming sources and subtitles. We fetch some
+    of those urls server-side and hand the rest to browsers, so a forged payload
+    could otherwise direct requests at an arbitrary, possibly internal, host.
+
+    Args:
+        url: the url to check
+
+    Returns:
+        True if the url is an https url on an allowlisted host
+    """
+    if not url:
+        return False
+    parsed = urlparse(url)
+    if parsed.scheme != "https":
+        return False
+    hostname = (parsed.hostname or "").lower()
+    if not hostname:
+        return False
+    return any(
+        hostname == allowed or (allowed.startswith(".") and hostname.endswith(allowed))
+        for allowed in allowed_media_hosts()
+    )
+
+
 def _get_cloudfront_domain(video_data: dict) -> str | None:
     """
     Extract the CloudFront domain from thumbnail or source URLs.
 
+    Only urls on an allowlisted host are considered, so the returned domain can
+    safely be used to build further urls.
+
     Args:
         video_data: OVS video API response dict
 
@@ -38,15 +84,13 @@ def _get_cloudfront_domain(video_data: dict) -> str | None:
     # Try thumbnails first
     for thumbnail in video_data.get("videothumbnail_set", []):
         cf_url = thumbnail.get("cloudfront_url", "")
-        if cf_url:
-            parsed = urlparse(cf_url)
-            return parsed.netloc
+        if is_allowed_media_url(cf_url):
+            return urlparse(cf_url).hostname
     # Fall back to sources
     for source in video_data.get("sources", []):
         src = source.get("src", "")
-        if src:
-            parsed = urlparse(src)
-            return parsed.netloc
+        if is_allowed_media_url(src):
+            return urlparse(src).hostname
     return None
 
 
@@ -68,9 +112,12 @@ def _build_caption_urls(video_data: dict) -> list[dict]:
 
     captions = []
     for subtitle in video_data.get("videosubtitle_set", []):
-        s3_key = subtitle.get("s3_object_key", "")
+        s3_key = quote(subtitle.get("s3_object_key", "").lstrip("/"))
         if s3_key:
             url = f"https://{cf_domain}/{s3_key}"
+            if not is_allowed_media_url(url):
+                log.warning("Skipping OVS caption url on disallowed host: %s", url)
+                continue
             captions.append(
                 {
                     "language": subtitle.get("language", "en"),
@@ -93,7 +140,10 @@ def _get_cover_image_url(video_data: dict) -> str | None:
     """
     thumbnails = video_data.get("videothumbnail_set", [])
     if thumbnails:
-        return thumbnails[0].get("cloudfront_url")
+        cover_url = thumbnails[0].get("cloudfront_url")
+        if is_allowed_media_url(cover_url):
+            return cover_url
+        log.warning("Ignoring OVS cover image url on disallowed host: %s", cover_url)
     return None
 
 
@@ -111,8 +161,11 @@ def _get_source_url(video_data: dict) -> str | None:
         src = source.get("src", "")
         if not src:
             continue
-        if urlparse(src).path.endswith(".m3u8"):
+        if not urlparse(src).path.endswith(".m3u8"):
+            continue
+        if is_allowed_media_url(src):
             return src
+        log.warning("Ignoring OVS streaming source on disallowed host: %s", src)
     return None
 
 
@@ -146,7 +199,8 @@ def _get_resource_url(video_data: dict) -> str:
     """
     Get the user-facing URL for a video resource.
 
-    Uses cta_link if available, otherwise builds a URL from OVS_API_BASE_URL.
+    Uses cta_link if it points at an allowlisted host, otherwise builds a URL
+    from OVS_API_BASE_URL.
 
     Args:
         video_data: OVS video API response dict
@@ -156,9 +210,11 @@ def _get_resource_url(video_data: dict) -> str:
     """
     cta_link = video_data.get("cta_link")
     if cta_link:
-        return cta_link
+        if is_allowed_media_url(cta_link):
+            return cta_link
+        log.warning("Ignoring OVS cta_link on disallowed host: %s", cta_link)
     base_url = settings.OVS_API_BASE_URL.rstrip("/")
-    return f"{base_url}/videos/{video_data['key']}"
+    return f"{base_url}/videos/{quote(str(video_data['key']))}"
 
 
 def extract(*, url=None) -> Generator[dict, None, None]:
@@ -263,7 +319,7 @@ def transform_collection(collection_data: dict) -> dict:
         "platform": PlatformType.ovs.name,
         "title": collection_data.get("title", ""),
         "description": collection_data.get("description", ""),
-        "url": f"{base_url}/collections/{collection_data['key']}",
+        "url": f"{base_url}/collections/{quote(str(collection_data['key']))}",
         "published": True,
     }
 
@@ -313,6 +369,12 @@ def _fetch_transcript(caption_urls: list[dict]) -> str:
     """
     for caption in caption_urls:
         if caption.get("language") == "en":
+            if not is_allowed_media_url(caption.get("url")):
+                log.warning(
+                    "Refusing to fetch transcript from disallowed host: %s",
+                    caption.get("url"),
+                )
+                break
             try:
                 result = extract_text_from_url(caption["url"], mime_type="text/vtt")
                 if result and result.get("content"):

learning_resources/etl/ovs_test.py

@@ -2,6 +2,7 @@
 
 import json
 from unittest.mock import Mock
+from urllib.parse import urlparse
 
 import pytest
 import requests
@@ -23,6 +24,7 @@
     extract,
     get_ovs_transcripts,
     get_ovs_videos_for_transcripts_job,
+    is_allowed_media_url,
     transform,
     transform_collection,
     transform_video,
@@ -40,8 +42,9 @@
 
 @pytest.fixture(autouse=True)
 def ovs_settings(settings):
-    """Ensure OVS_API_BASE_URL is set for all tests in this module"""
+    """Ensure OVS url settings are set for all tests in this module"""
     settings.OVS_API_BASE_URL = OVS_TEST_BASE_URL
+    settings.OVS_ALLOWED_MEDIA_HOSTS = [".cloudfront.net", "example.com"]
     return settings
 
 
@@ -686,6 +689,117 @@ def test_fetch_transcript_tika_returns_none(mocker):
     assert result == ""
 
 
+def test_fetch_transcript_disallowed_host(mocker):
+    """Should not request a transcript from a host outside the allowlist"""
+    mock_extract = mocker.patch("learning_resources.etl.ovs.extract_text_from_url")
+    caption_urls = [
+        {"language": "en", "url": "http://169.254.169.254/latest/meta-data/"},
+    ]
+    assert _fetch_transcript(caption_urls) == ""
+    mock_extract.assert_not_called()
+
+
+class TestMediaUrlAllowlist:
+    """Tests for is_allowed_media_url and the url filtering that depends on it"""
+
+    @pytest.mark.parametrize(
+        ("url", "expected"),
+        [
+            pytest.param(
+                "https://d1rlgptj9v7p9j.cloudfront.net/a.vtt", True, id="subdomain"
+            ),
+            pytest.param("https://example.com/a.vtt", True, id="exact_host"),
+            pytest.param(f"{OVS_TEST_BASE_URL}/videos/abc", True, id="ovs_api_host"),
+            pytest.param("http://example.com/a.vtt", False, id="not_https"),
+            pytest.param(
+                "https://evilcloudfront.net/a.vtt", False, id="suffix_lookalike"
+            ),
+            pytest.param("https://notexample.com/a.vtt", False, id="host_lookalike"),
+            pytest.param(
+                "https://example.com.evil.io/a.vtt", False, id="host_prefix_lookalike"
+            ),
+            pytest.param(
+                "https://example.com@evil.io/a.vtt", False, id="userinfo_confusion"
+            ),
+            pytest.param(
+                "http://169.254.169.254/latest/meta-data/", False, id="link_local"
+            ),
+            pytest.param("file:///etc/passwd", False, id="file_scheme"),
+            pytest.param("", False, id="empty"),
+            pytest.param(None, False, id="none"),
+        ],
+    )
+    def test_is_allowed_media_url(self, url, expected):
+        """Only https urls on allowlisted hosts are permitted"""
+        assert is_allowed_media_url(url) is expected
+
+    def test_source_url_on_disallowed_host_ignored(self):
+        """A streaming source on an unknown host is not used"""
+        assert (
+            _get_source_url({"sources": [{"src": "https://evil.io/video__index.m3u8"}]})
+            is None
+        )
+
+    def test_cover_image_on_disallowed_host_ignored(self):
+        """A thumbnail on an unknown host is not used"""
+        assert (
+            _get_cover_image_url(
+                {"videothumbnail_set": [{"cloudfront_url": "https://evil.io/t.jpg"}]}
+            )
+            is None
+        )
+
+    def test_caption_urls_require_allowed_domain(self):
+        """Captions are dropped when no allowlisted domain can be determined"""
+        assert (
+            _build_caption_urls(
+                {
+                    "videothumbnail_set": [
+                        {"cloudfront_url": "https://evil.io/t.jpg"},
+                    ],
+                    "sources": [{"src": "https://evil.io/video__index.m3u8"}],
+                    "videosubtitle_set": [
+                        {"s3_object_key": "subtitles/a.vtt", "language": "en"}
+                    ],
+                }
+            )
+            == []
+        )
+
+    def test_caption_urls_escape_object_key(self):
+        """A subtitle key cannot break out of the allowlisted host"""
+        captions = _build_caption_urls(
+            {
+                "videothumbnail_set": [
+                    {"cloudfront_url": "https://abc.cloudfront.net/t.jpg"},
+                ],
+                "videosubtitle_set": [
+                    {"s3_object_key": "/../@evil.io/a.vtt", "language": "en"}
+                ],
+            }
+        )
+        assert len(captions) == 1
+        assert urlparse(captions[0]["url"]).hostname == "abc.cloudfront.net"
+
+    def test_resource_url_ignores_disallowed_cta_link(self, settings):
+        """cta_link pointing off-allowlist falls back to the OVS url"""
+        settings.OVS_API_BASE_URL = OVS_TEST_BASE_URL
+        video = {"key": "abc123", "cta_link": "https://evil.io/watch/abc123"}
+        assert _get_resource_url(video) == f"{OVS_TEST_BASE_URL}/videos/abc123"
+
+    def test_transform_video_skips_disallowed_source(self):
+        """A video whose only source is off-allowlist is not transformed"""
+        assert (
+            transform_video(
+                {
+                    "key": "abc123",
+                    "sources": [{"src": "https://evil.io/video__index.m3u8"}],
+                }
+            )
+            is None
+        )
+
+
 def test_filters_ovs_videos_without_transcripts(ovs_platform):
     """Should return OVS videos with empty transcripts"""
     video = VideoFactory.create(

main/settings_course_etl.py

@@ -2,7 +2,7 @@
 Django settings specific to learning_resources ingestion
 """
 
-from main.envs import get_bool, get_int, get_string
+from main.envs import get_bool, get_int, get_list_of_str, get_string
 
 # EDX API Credentials
 EDX_API_URL = get_string("EDX_API_URL", None)
@@ -121,6 +121,15 @@
     "OVS_API_BASE_URL",
     None,
 )
+# Hosts that OVS media urls (thumbnails, streaming sources, subtitles) may point
+# at. OVS payloads arrive over a webhook, so any url in them is untrusted input:
+# without an allowlist a forged payload could point our requests at an arbitrary
+# (possibly internal) host. The OVS_API_BASE_URL host is always allowed. An entry
+# beginning with a "." matches any subdomain of that domain.
+OVS_ALLOWED_MEDIA_HOSTS = get_list_of_str(
+    "OVS_ALLOWED_MEDIA_HOSTS",
+    [".cloudfront.net"],
+)
 
 # course catalog podcast etl settings
 OPEN_PODCAST_DATA_BRANCH = get_string("OPEN_PODCAST_DATA_BRANCH", "master")

openapi/specs/v1.yaml

@@ -13798,9 +13798,11 @@ components:
         key:
           type: string
           minLength: 1
+          pattern: ^[A-Za-z0-9._-]{1,255}$
         video_id:
           type: string
           minLength: 1
+          pattern: ^[A-Za-z0-9._-]{1,255}$
         delete:
           type: boolean
           default: false