ainur/metube
1
0
Fork 0
mirror of https://github.com/alexta69/metube.git synced 2026-07-23 05:12:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: ALLOW_PRIVATE_ADDRESSES to opt out of the SSRF checks (closes #1036)

Browse Source 
The SSRF guard rejects any address that isn't globally routable, which breaks
proxy/VPN setups that resolve hosts into private or special-use ranges. The
reported case is Fake-IP clients (sing-box, Clash, Mihomo) that map YouTube to
the RFC 2544 benchmarking range 198.18.0.0/15 to tunnel the traffic; MeTube
rejected it with "Refusing to fetch internal address" even though yt-dlp itself
handles it fine.

Add a boolean ALLOW_PRIVATE_ADDRESSES (default false) that, when set, skips both
layers: validate_url returns after scheme validation without the internal-host
checks, and the connect-time socket guard is not installed. Threaded to the
download subprocess via Download so the guard sees it too. Scheme validation
(http/https only) still applies. Documented in the README as a trust-your-network
opt-out that disables SSRF protection.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Alex Shnitman
2026-07-21 07:44:38 +03:00
parent 13cb65d931
commit e061a8a5ba
6 changed files with 57 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -80,6 +80,7 @@ Certain values can be set via environment variables, using the `-e` parameter on
* __YTDL_OPTIONS_PRESETS__: Named bundles of yt-dlp options, selectable per download in the UI. See [Configuring yt-dlp options](#%EF%B8%8F-configuring-yt-dlp-options) for format and examples.
* __YTDL_OPTIONS_PRESETS_FILE__: Path to a JSON file containing presets. Monitored and reloaded automatically on changes. See [Configuring yt-dlp options](#%EF%B8%8F-configuring-yt-dlp-options).
* __ALLOW_YTDL_OPTIONS_OVERRIDES__: Whether to show a free-text field in the UI for per-download yt-dlp option overrides. Defaults to `false`. See [Configuring yt-dlp options](#%EF%B8%8F-configuring-yt-dlp-options) for details and security considerations.
* __ALLOW_PRIVATE_ADDRESSES__: Whether to allow downloads from private, loopback, link-local and other non-global addresses. Defaults to `false`, which protects against SSRF by refusing URLs that resolve to internal hosts. Set to `true` only in trusted environments — for example when routing traffic through a proxy/VPN client in Fake-IP mode (sing-box, Clash, Mihomo), which resolves hosts to the `198.18.0.0/15` range. Enabling this disables the SSRF protection entirely, so only use it when you control the network.
* __YTDL_NIGHTLY_UPDATE_TIME__: If set, will cause MeTube to use [nightly yt-dlp builds](https://github.com/yt-dlp/yt-dlp-nightly-builds) instead of the stable releases. Set to the time (`HH:MM`, 24-hour) when you want the daily upgrades and MeTube restart to happen. Defaults to empty (disabled).
### 🌐 Web Server & URLs
app/main.py
+2 -1
View File
@@ -81,6 +81,7 @@ class Config:
'YTDL_OPTIONS_PRESETS': '{}',
'YTDL_OPTIONS_PRESETS_FILE': '',
'ALLOW_YTDL_OPTIONS_OVERRIDES': 'false',
'ALLOW_PRIVATE_ADDRESSES': 'false',
'CORS_ALLOWED_ORIGINS': '',
'ROBOTS_TXT': '',
'HOST': '0.0.0.0',
@@ -96,7 +97,7 @@ class Config:
'YTDL_NIGHTLY_UPDATE_TIME': '',
}
_BOOLEAN = ('DOWNLOAD_DIRS_INDEXABLE', 'CUSTOM_DIRS', 'CREATE_CUSTOM_DIRS', 'DELETE_FILE_ON_TRASHCAN', 'HTTPS', 'ENABLE_ACCESSLOG', 'ALLOW_YTDL_OPTIONS_OVERRIDES')
_BOOLEAN = ('DOWNLOAD_DIRS_INDEXABLE', 'CUSTOM_DIRS', 'CREATE_CUSTOM_DIRS', 'DELETE_FILE_ON_TRASHCAN', 'HTTPS', 'ENABLE_ACCESSLOG', 'ALLOW_YTDL_OPTIONS_OVERRIDES', 'ALLOW_PRIVATE_ADDRESSES')
def __init__(self):
for k, v in self._DEFAULTS.items():
app/subscriptions.py
+3 -2
View File
@@ -115,7 +115,7 @@ def extract_flat_playlist(
if not nested_url:
continue
# nested_url comes from remote playlist content; guard it too.
if validate_url(nested_url) is not None:
if validate_url(nested_url, allow_private=getattr(config, "ALLOW_PRIVATE_ADDRESSES", False)) is not None:
continue
nested_info, nested_entries = extract_flat_playlist(
config,
@@ -548,7 +548,8 @@ class SubscriptionManager:
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)
url_error = await asyncio.get_running_loop().run_in_executor(
None, partial(validate_url, url, allow_private=getattr(self.config, "ALLOW_PRIVATE_ADDRESSES", False)))
if url_error is not None:
log.warning('Rejected subscription URL "%s": %s', url, url_error)
return {"status": "error", "msg": url_error}
app/tests/test_url_guard.py
+26
View File
@@ -145,6 +145,32 @@ class GuardedGetaddrinfoTests(unittest.TestCase):
self.assertEqual([r[4][0] for r in results], ["127.0.0.1"])
class AllowPrivateBypassTests(unittest.TestCase):
"""ALLOW_PRIVATE_ADDRESSES: trusted proxy/VPN environments opt out of the
SSRF address checks (e.g. Fake-IP clients that resolve to 198.18.0.0/15)."""
def test_internal_address_allowed_when_bypassed(self):
# Fake-IP benchmarking range that is_global rejects by default.
self.assertIsNone(validate_url("http://www.youtube.com/x", allow_private=True))
def test_private_host_allowed_when_bypassed(self):
# No DNS lookup needed: the bypass returns before resolution.
with mock.patch("url_guard.socket.getaddrinfo") as gai:
self.assertIsNone(validate_url("http://192.168.1.1/x", allow_private=True))
gai.assert_not_called()
def test_scheme_still_enforced_when_bypassed(self):
self.assertIsNotNone(validate_url("file:///etc/passwd", allow_private=True))
def test_socket_guard_not_installed_when_bypassed(self):
original = socket.getaddrinfo
try:
install_socket_guard(allow_private=True)
self.assertIs(socket.getaddrinfo, original)
finally:
socket.getaddrinfo = original
class InstallSocketGuardTests(unittest.TestCase):
def test_install_replaces_and_is_idempotent(self):
original = socket.getaddrinfo
app/url_guard.py
+17 -2
View File
@@ -92,7 +92,7 @@ def _guarded_getaddrinfo(host, *args, **kwargs):
return allowed
def install_socket_guard() -> None:
def install_socket_guard(allow_private: bool = False) -> None:
"""Enforce the no-internal-hosts policy at actual connection time.
``validate_url`` only checks the *submitted* URL string; yt-dlp then follows
@@ -102,16 +102,27 @@ def install_socket_guard() -> None:
for any networking backend that resolves through Python's socket module
(urllib, requests). Native resolvers — notably curl_cffi/libcurl used by
``--impersonate`` — bypass this and rely on network isolation as the backstop.
When *allow_private* is set (``ALLOW_PRIVATE_ADDRESSES``), the guard is not
installed at all, so proxy/VPN setups that route through private or Fake-IP
ranges keep working.
"""
if allow_private:
return
socket.getaddrinfo = _guarded_getaddrinfo
def validate_url(url: str) -> str | None:
def validate_url(url: str, allow_private: bool = False) -> 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.
When *allow_private* is set (``ALLOW_PRIVATE_ADDRESSES``), the internal-host
and internal-address checks are skipped so that trusted proxy/VPN setups —
e.g. Fake-IP clients that resolve YouTube to ``198.18.0.0/15`` — can be used.
Scheme validation (http/https only) still applies.
"""
if not isinstance(url, str):
return 'Invalid URL'
@@ -130,6 +141,10 @@ def validate_url(url: str) -> str | None:
if not hostname:
return 'URL is missing a host'
if allow_private:
# Environment is explicitly trusted: skip the SSRF address checks.
return None
if _hostname_is_blocked(hostname):
return f'Refusing to fetch internal host "{hostname}"'
app/ytdl.py
+8 -5
View File
@@ -557,11 +557,12 @@ class Download:
cls.manager.shutdown()
cls.manager = None
def __init__(self, download_dir, temp_dir, output_template, output_template_chapter, quality, format, ytdl_opts, info):
def __init__(self, download_dir, temp_dir, output_template, output_template_chapter, quality, format, ytdl_opts, info, allow_private=False):
self.download_dir = download_dir
self.temp_dir = temp_dir
self.output_template = output_template
self.output_template_chapter = output_template_chapter
self.allow_private = allow_private
self.info = info
self.format = get_format(
getattr(info, 'download_type', 'video'),
@@ -637,8 +638,9 @@ class Download:
pass
# Re-validate every outbound connection at fetch time. validate_url only
# saw the submitted URL string; this catches redirects and DNS rebinding
# to internal hosts (cloud metadata, RFC1918) that it cannot.
install_socket_guard()
# to internal hosts (cloud metadata, RFC1918) that it cannot. Skipped when
# ALLOW_PRIVATE_ADDRESSES trusts the environment (e.g. Fake-IP proxies).
install_socket_guard(self.allow_private)
log.info(f"Starting download for: {self.info.title} ({self.info.url})")
try:
debug_logging = logging.getLogger().isEnabledFor(logging.DEBUG)
@@ -1333,7 +1335,7 @@ class DownloadQueue:
if playlist_item_limit > 0:
log.info(f'playlist limit is set. Processing only first {playlist_item_limit} entries')
ytdl_options['playlistend'] = playlist_item_limit
download = Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, dl.quality, dl.format, ytdl_options, dl)
download = Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, dl.quality, dl.format, ytdl_options, dl, allow_private=self.config.ALLOW_PRIVATE_ADDRESSES)
is_upcoming = (
getattr(dl, 'live_status', None) == 'is_upcoming'
or getattr(dl, 'status', None) == 'scheduled'
@@ -1556,7 +1558,8 @@ class DownloadQueue:
# 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)
url_error = await asyncio.get_running_loop().run_in_executor(
None, partial(validate_url, url, allow_private=self.config.ALLOW_PRIVATE_ADDRESSES))
if url_error is not None:
log.warning('Rejected URL "%s": %s', url, url_error)
return {'status': 'error', 'msg': url_error}