fix: restrict web ingest to official YouTube hosts and bound submission size

The submit form's URL validation accepted any http/https host. Restrict it
to an explicit allowlist of official YouTube hosts using normalized,
exact hostname comparison (no substring/suffix matching), reject embedded
userinfo and malformed URLs, and add server-side length bounds for the URL,
submitter, note, and total request body.
This commit is contained in:
Media
2026-08-07 17:04:05 -05:00
parent 270ede8b77
commit ec826bb8f3
3 changed files with 125 additions and 11 deletions
+79 -7
View File
@@ -53,7 +53,13 @@ class FlaskFactoryTests(unittest.TestCase):
def _app(self, service):
return self._factory()(self.config, service, testing=True)
def _post(self, service, *, url="https://example.com/playlist", **fields):
def _post(
self,
service,
*,
url="https://www.youtube.com/playlist?list=PL1234567890",
**fields,
):
app = self._app(service)
return app.test_client().post(
"/submit",
@@ -71,15 +77,29 @@ class FlaskFactoryTests(unittest.TestCase):
self.assertIn(b'name="submitter"', response.data)
self.assertIn(b'name="note"', response.data)
def test_submit_accepts_only_http_or_https_urls_with_a_network_location(self):
def test_submit_rejects_non_youtube_and_malformed_urls(self):
invalid_urls = (
"",
"http",
"https:///missing-host",
"httpfoo://example.com/video",
"ftp://example.com/video",
"httpfoo://youtube.com/video",
"ftp://youtube.com/video",
"javascript:alert(1)",
"/relative/path",
# Arbitrary external hosts, including lookalikes and substring
# tricks, must be rejected by exact normalized hostname match.
"https://example.com/video",
"https://notyoutube.com/video",
"https://youtube.com.evil.com/video",
"https://evil.com/?u=youtube.com",
"https://xyoutube.com/watch?v=abc",
"https://youtube.com.attacker.net/watch?v=abc",
# Embedded userinfo must be rejected even when the host matches.
"https://user:[email protected]/watch?v=abc",
"https://[email protected]/watch?v=abc",
"https://[email protected]/watch?v=abc",
# Oversized URL.
"https://www.youtube.com/watch?v=" + ("a" * 3000),
)
for url in invalid_urls:
with self.subTest(url=url):
@@ -89,7 +109,18 @@ class FlaskFactoryTests(unittest.TestCase):
self.assertIn(b"Invalid URL", response.data)
self.assertEqual(service.enqueue_calls, [])
for url in ("http://example.com/video", "https://example.com/video"):
def test_submit_accepts_only_explicit_official_youtube_hosts(self):
valid_urls = (
"https://youtube.com/watch?v=abc",
"https://www.youtube.com/watch?v=abc",
"http://www.youtube.com/watch?v=abc",
"https://m.youtube.com/watch?v=abc",
"https://music.youtube.com/watch?v=abc",
"https://youtu.be/abc",
"https://YouTube.com/watch?v=abc",
"https://WWW.YOUTUBE.COM/watch?v=abc",
)
for url in valid_urls:
with self.subTest(url=url):
service = FakeControllerService()
response = self._post(service, url=url)
@@ -103,7 +134,7 @@ class FlaskFactoryTests(unittest.TestCase):
response = self._post(
service,
url="https://example.com/playlist",
url="https://www.youtube.com/playlist?list=PL1234567890",
submitter=" Alice ",
note=" Before service ",
)
@@ -113,9 +144,50 @@ class FlaskFactoryTests(unittest.TestCase):
self.assertIn(b"start downloading shortly", response.data)
self.assertEqual(
service.enqueue_calls,
[("https://example.com/playlist", "Alice", "Before service")],
[
(
"https://www.youtube.com/playlist?list=PL1234567890",
"Alice",
"Before service",
)
],
)
def test_submit_rejects_oversized_submitter_and_note(self):
module = importlib.import_module("app")
service = FakeControllerService()
response = self._post(
service,
submitter="x" * (module.MAX_SUBMITTER_LENGTH + 1),
)
self.assertEqual(response.status_code, 200)
self.assertIn(b"must be", response.data)
self.assertEqual(service.enqueue_calls, [])
service = FakeControllerService()
response = self._post(
service,
note="x" * (module.MAX_NOTE_LENGTH + 1),
)
self.assertEqual(response.status_code, 200)
self.assertIn(b"must be", response.data)
self.assertEqual(service.enqueue_calls, [])
def test_submit_rejects_oversized_request_body(self):
service = FakeControllerService()
app = self._app(service)
response = app.test_client().post(
"/submit",
data={
"url": "https://www.youtube.com/watch?v=abc",
"note": "x" * 64_000,
},
follow_redirects=True,
)
self.assertEqual(response.status_code, 200)
self.assertIn(b"too large", response.data.lower())
self.assertEqual(service.enqueue_calls, [])
def test_reused_submission_preserves_existing_slot_message(self):
service = FakeControllerService(
enqueue_result={"status": "queued", "slot": 7, "reused_slot": True}