diff --git a/README.md b/README.md index 19151c5..5ffa728 100644 --- a/README.md +++ b/README.md @@ -927,8 +927,11 @@ PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v - The web form is unauthenticated and listens on `0.0.0.0` by default. Restrict the published port with host firewall/network ACLs or place it behind an - authenticated reverse proxy. URL validation checks only the `http`/`https` - shape; yt-dlp performs the outbound request. + authenticated reverse proxy. Submitted URLs must use `http`/`https`, carry no + embedded userinfo, and resolve (by exact, normalized hostname match) to an + official YouTube host: `youtube.com`, `www.youtube.com`, `m.youtube.com`, + `music.youtube.com`, or `youtu.be`. URL, submitter, note, and total request + size are also server-side bounded. yt-dlp performs the outbound request. - MQTT username/password support is available, but application-level TLS is not configured here. Keep this connection on a trusted/protected network or add a separately managed secure tunnel/proxy, and grant the Controller only the diff --git a/app.py b/app.py index 7926c57..d98df83 100644 --- a/app.py +++ b/app.py @@ -224,12 +224,36 @@ HTML_FORM = """ """ +ALLOWED_YOUTUBE_HOSTS = frozenset( + { + "youtube.com", + "www.youtube.com", + "m.youtube.com", + "music.youtube.com", + "youtu.be", + } +) + +MAX_URL_LENGTH = 2048 +MAX_SUBMITTER_LENGTH = 200 +MAX_NOTE_LENGTH = 1000 +MAX_REQUEST_CONTENT_LENGTH = 32 * 1024 + + def _valid_media_url(value: str) -> bool: + if not value or len(value) > MAX_URL_LENGTH: + return False try: parsed = urlsplit(value) + hostname = parsed.hostname + has_userinfo = parsed.username is not None or parsed.password is not None except ValueError: return False - return parsed.scheme.lower() in {"http", "https"} and bool(parsed.netloc) + if parsed.scheme.lower() not in {"http", "https"}: + return False + if has_userinfo or not hostname: + return False + return hostname.lower() in ALLOWED_YOUTUBE_HOSTS def _submission_message(result: dict[str, object]) -> str: @@ -277,6 +301,7 @@ def create_app( web_app.config.update( TESTING=testing, SECRET_KEY=secret_key or "controller-test-only-secret", + MAX_CONTENT_LENGTH=MAX_REQUEST_CONTENT_LENGTH, ) web_app.extensions["controller_service"] = service web_app.extensions["controller_config"] = config @@ -285,6 +310,11 @@ def create_app( def index(): return render_template_string(HTML_FORM) + @web_app.errorhandler(413) + def _submission_too_large(_error): + flash("The submission was too large.") + return redirect(url_for("index")) + @web_app.post("/submit") def submit(): url = request.form.get("url", "").strip() @@ -292,7 +322,16 @@ def create_app( note = request.form.get("note", "").strip() or None if not _valid_media_url(url): - flash("Invalid URL. Use a complete http:// or https:// address.") + flash( + "Invalid URL. Only YouTube links are accepted " + "(e.g. https://www.youtube.com/... or https://youtu.be/...)." + ) + return redirect(url_for("index")) + if submitter is not None and len(submitter) > MAX_SUBMITTER_LENGTH: + flash(f"Your name must be {MAX_SUBMITTER_LENGTH} characters or fewer.") + return redirect(url_for("index")) + if note is not None and len(note) > MAX_NOTE_LENGTH: + flash(f"The note must be {MAX_NOTE_LENGTH} characters or fewer.") return redirect(url_for("index")) try: diff --git a/tests/test_app.py b/tests/test_app.py index a3f7a38..9a5b068 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -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:pass@www.youtube.com/watch?v=abc", + "https://user@www.youtube.com/watch?v=abc", + "https://www.youtube.com@evil.com/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}