fix: block SSRF via user-submitted URLs

User-submitted URLs were passed straight to yt-dlp's generic extractor,
letting the server fetch internal endpoints (cloud metadata, loopback,
RFC1918 hosts). Add a url_guard.validate_url check at every URL ingress
(add, subscribe, and nested playlist recursion) that rejects non-http(s)
schemes and hosts resolving to non-global addresses, while leaving bare
video IDs and search prefixes untouched.

app/subscriptions.py

@@ -19,6 +19,7 @@ import yt_dlp.networking.impersonate
 import bg_tasks
 from dl_formats import merge_ytdl_option_layers
 from state_store import AtomicJsonStore, read_legacy_shelf
+from url_guard import validate_url
 
 log = logging.getLogger("subscriptions")
 
@@ -113,6 +114,9 @@ def extract_flat_playlist(
                 nested_url = _entry_video_url(ent)
                 if not nested_url:
                     continue
+                # nested_url comes from remote playlist content; guard it too.
+                if validate_url(nested_url) is not None:
+                    continue
                 nested_info, nested_entries = extract_flat_playlist(
                     config,
                     nested_url,
@@ -542,6 +546,12 @@ class SubscriptionManager:
         url = self._normalize_url(url)
         if not url:
             return {"status": "error", "msg": "Missing URL"}
+        # SSRF guard: block non-http(s) schemes and internal/metadata hosts
+        # before yt-dlp fetches the feed. May do a DNS lookup, so run off-loop.
+        url_error = await asyncio.get_running_loop().run_in_executor(None, validate_url, url)
+        if url_error is not None:
+            log.warning('Rejected subscription URL "%s": %s', url, url_error)
+            return {"status": "error", "msg": url_error}
         try:
             title_regex_stored = validate_title_regex(title_regex)
         except re.error as exc:

app/tests/test_url_guard.py

@@ -0,0 +1,102 @@
+"""Tests for the SSRF URL guard (``url_guard.validate_url``)."""
+
+from __future__ import annotations
+
+import socket
+import unittest
+from unittest import mock
+
+from url_guard import validate_url
+
+
+def _addrinfo(*addrs, family=socket.AF_INET):
+    return [(family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (addr, 0)) for addr in addrs]
+
+
+class NonUrlInputTests(unittest.TestCase):
+    """Bare IDs and yt-dlp search/extractor prefixes must pass untouched."""
+
+    def test_bare_video_id_allowed(self):
+        self.assertIsNone(validate_url("dQw4w9WgXcQ"))
+
+    def test_ytsearch_prefix_allowed(self):
+        self.assertIsNone(validate_url("ytsearch:some song"))
+
+    def test_empty_string_allowed(self):
+        self.assertIsNone(validate_url(""))
+
+    def test_non_string_rejected(self):
+        self.assertIsNotNone(validate_url(None))
+
+
+class SchemeTests(unittest.TestCase):
+    def test_file_scheme_blocked(self):
+        self.assertIsNotNone(validate_url("file:///etc/passwd"))
+
+    def test_ftp_scheme_blocked(self):
+        self.assertIsNotNone(validate_url("ftp://example.com/x"))
+
+    def test_data_scheme_blocked(self):
+        self.assertIsNotNone(validate_url("data://text/plain;base64,AAAA"))
+
+
+class HostnameBlocklistTests(unittest.TestCase):
+    def test_localhost_blocked_without_lookup(self):
+        with mock.patch("url_guard.socket.getaddrinfo") as gai:
+            self.assertIsNotNone(validate_url("http://localhost:8080/x"))
+            gai.assert_not_called()
+
+    def test_localhost_subdomain_blocked(self):
+        self.assertIsNotNone(validate_url("http://foo.localhost/x"))
+
+    def test_gcp_metadata_name_blocked(self):
+        self.assertIsNotNone(validate_url("http://metadata.google.internal/x"))
+
+
+class AddressResolutionTests(unittest.TestCase):
+    def _validate_with_addrs(self, url, *addrs, family=socket.AF_INET):
+        with mock.patch("url_guard.socket.getaddrinfo", return_value=_addrinfo(*addrs, family=family)):
+            return validate_url(url)
+
+    def test_public_https_allowed(self):
+        self.assertIsNone(self._validate_with_addrs("https://youtube.com/watch?v=x", "142.250.1.1"))
+
+    def test_public_http_allowed(self):
+        self.assertIsNone(self._validate_with_addrs("http://example.com/x", "93.184.216.34"))
+
+    def test_link_local_metadata_blocked(self):
+        self.assertIsNotNone(self._validate_with_addrs("http://metadata/x", "169.254.169.254"))
+
+    def test_loopback_ipv4_blocked(self):
+        self.assertIsNotNone(self._validate_with_addrs("http://127.0.0.1/x", "127.0.0.1"))
+
+    def test_private_rfc1918_blocked(self):
+        self.assertIsNotNone(self._validate_with_addrs("http://intranet/x", "10.0.0.5"))
+
+    def test_decimal_ip_form_blocked(self):
+        # 2852039166 == 169.254.169.254; the OS resolver normalizes it.
+        self.assertIsNotNone(self._validate_with_addrs("http://2852039166/x", "169.254.169.254"))
+
+    def test_ipv6_loopback_blocked(self):
+        self.assertIsNotNone(
+            self._validate_with_addrs("http://[::1]/x", "::1", family=socket.AF_INET6)
+        )
+
+    def test_ipv4_mapped_ipv6_metadata_blocked(self):
+        self.assertIsNotNone(
+            self._validate_with_addrs(
+                "http://evil/x", "::ffff:169.254.169.254", family=socket.AF_INET6
+            )
+        )
+
+    def test_mixed_public_and_private_blocked(self):
+        # If any resolved address is internal, reject the whole URL.
+        self.assertIsNotNone(self._validate_with_addrs("http://mixed/x", "142.250.1.1", "127.0.0.1"))
+
+    def test_resolution_failure_defers_to_ytdlp(self):
+        with mock.patch("url_guard.socket.getaddrinfo", side_effect=socket.gaierror):
+            self.assertIsNone(validate_url("http://does-not-resolve.example/x"))
+
+
+if __name__ == "__main__":
+    unittest.main()

app/url_guard.py

@@ -0,0 +1,89 @@
+"""Lightweight SSRF guard for user-submitted URLs.
+
+MeTube hands user-submitted URLs to yt-dlp, whose generic extractor will fetch
+any ``http(s)`` URL. Without a guard, an attacker can make the server fetch
+internal endpoints (cloud metadata services, loopback, RFC1918 hosts, etc.) and
+have the response saved to the download directory and served back.
+
+This module provides a single cheap validator applied at every URL ingress. It
+intentionally does NOT attempt DNS-rebinding pinning, redirect-chain
+re-validation, or validation of every media URL yt-dlp derives from remote
+metadata — network isolation (e.g. Docker) remains the backstop for those.
+"""
+
+import ipaddress
+import logging
+import socket
+from urllib.parse import urlsplit
+
+log = logging.getLogger('url_guard')
+
+_ALLOWED_SCHEMES = ('http', 'https')
+
+# Hostnames that must be blocked without needing a lookup. ``localhost`` and any
+# subdomain of it are conventionally loopback, and the GCP metadata name is a
+# well-known SSRF target that may resolve via a resolver we don't control.
+_BLOCKED_HOSTNAMES = ('localhost', 'metadata.google.internal')
+
+
+def _hostname_is_blocked(hostname: str) -> bool:
+    host = hostname.rstrip('.').lower()
+    for blocked in _BLOCKED_HOSTNAMES:
+        if host == blocked or host.endswith('.' + blocked):
+            return True
+    return False
+
+
+def _address_is_global(addr: str) -> bool:
+    try:
+        ip = ipaddress.ip_address(addr)
+    except ValueError:
+        return False
+    # Unwrap IPv4-mapped/compatible IPv6 (e.g. ::ffff:169.254.169.254) so the
+    # embedded IPv4 address is judged on its own merits.
+    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
+        ip = ip.ipv4_mapped
+    return ip.is_global
+
+
+def validate_url(url: str) -> str | None:
+    """Return an error message if the URL is disallowed, else ``None``.
+
+    Inputs without a ``://`` scheme separator (bare video IDs, ``ytsearch:``
+    and other yt-dlp search/extractor prefixes) are allowed unchanged so that
+    non-URL entries keep working.
+    """
+    if not isinstance(url, str):
+        return 'Invalid URL'
+
+    candidate = url.strip()
+    if '://' not in candidate:
+        # Not an absolute URL: bare video IDs, ytsearch: prefixes, etc.
+        return None
+
+    parts = urlsplit(candidate)
+    scheme = parts.scheme.lower()
+    if scheme not in _ALLOWED_SCHEMES:
+        return f'URL scheme "{parts.scheme}" is not allowed (only http and https)'
+
+    hostname = parts.hostname
+    if not hostname:
+        return 'URL is missing a host'
+
+    if _hostname_is_blocked(hostname):
+        return f'Refusing to fetch internal host "{hostname}"'
+
+    try:
+        addrinfo = socket.getaddrinfo(hostname, parts.port, proto=socket.IPPROTO_TCP)
+    except socket.gaierror:
+        # Let yt-dlp surface a normal resolution error rather than masking it.
+        return None
+    except (UnicodeError, ValueError):
+        return f'Invalid host "{hostname}"'
+
+    for family, _type, _proto, _canonname, sockaddr in addrinfo:
+        addr = sockaddr[0]
+        if not _address_is_global(addr):
+            return f'Refusing to fetch internal address "{addr}" for host "{hostname}"'
+
+    return None

app/ytdl.py

@@ -25,6 +25,7 @@ from dl_formats import get_format, get_opts, AUDIO_FORMATS, merge_ytdl_option_la
 from datetime import datetime
 from state_store import AtomicJsonStore, from_json_compatible, read_legacy_shelf, to_json_compatible
 from subscriptions import _entry_id
+from url_guard import validate_url
 
 log = logging.getLogger('ytdl')
 
@@ -1444,6 +1445,13 @@ class DownloadQueue:
             return {'status': 'ok'}
         else:
             already.add(url)
+        # SSRF guard: reject non-http(s) schemes and hosts resolving to
+        # internal/loopback/link-local/metadata addresses before yt-dlp fetches
+        # anything. run_in_executor because validate_url may perform a DNS lookup.
+        url_error = await asyncio.get_running_loop().run_in_executor(None, validate_url, url)
+        if url_error is not None:
+            log.warning('Rejected URL "%s": %s', url, url_error)
+            return {'status': 'error', 'msg': url_error}
         try:
             entry = await asyncio.get_running_loop().run_in_executor(
                 None,