metube/app/tests/test_url_guard.py

"""Tests for the SSRF URL guard (``url_guard.validate_url``)."""

from __future__ import annotations

import socket
import unittest
from unittest import mock

import url_guard
from url_guard import (
    validate_url,
    _address_allowed_at_connect,
    _address_is_global,
    _guarded_getaddrinfo,
    _proxy_endpoint,
    install_socket_guard,
)


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_is_rejected(self):
        # Fail closed: an unresolvable host cannot be verified as non-internal.
        with mock.patch("url_guard.socket.getaddrinfo", side_effect=socket.gaierror):
            self.assertIsNotNone(validate_url("http://does-not-resolve.example/x"))


class ConnectAddressPolicyTests(unittest.TestCase):
    """Connect-time policy: allow global, plus loopback only when the caller has
    established that this destination is the operator's configured proxy."""

    def test_global_allowed(self):
        self.assertTrue(_address_allowed_at_connect("142.250.1.1"))

    def test_loopback_blocked_by_default(self):
        # A blanket loopback allowance is what let manifest-derived media URLs
        # reach services on the server's own loopback interface.
        self.assertFalse(_address_allowed_at_connect("127.0.0.1"))
        self.assertFalse(_address_allowed_at_connect("::1"))

    def test_loopback_allowed_only_when_opted_in(self):
        self.assertTrue(_address_allowed_at_connect("127.0.0.1", allow_loopback=True))
        self.assertTrue(_address_allowed_at_connect("::1", allow_loopback=True))

    def test_opt_in_does_not_widen_beyond_loopback(self):
        self.assertFalse(_address_allowed_at_connect("169.254.169.254", allow_loopback=True))
        self.assertFalse(_address_allowed_at_connect("10.0.0.5", allow_loopback=True))

    def test_link_local_metadata_blocked(self):
        self.assertFalse(_address_allowed_at_connect("169.254.169.254"))

    def test_private_blocked(self):
        self.assertFalse(_address_allowed_at_connect("10.0.0.5"))
        self.assertFalse(_address_allowed_at_connect("192.168.1.10"))

    def test_ipv4_mapped_metadata_blocked(self):
        self.assertFalse(_address_allowed_at_connect("::ffff:169.254.169.254"))


class TunnelledIPv4Tests(unittest.TestCase):
    """IPv6 transition forms that carry an IPv4 address the outer address hides.

    ``is_global`` looks only at the outer address, so a form that tunnels an
    internal IPv4 has to be unwrapped before it is judged (GHSA-5mq5-qr7m-f4wx).
    """

    def test_nat64_well_known_prefix_blocked(self):
        # 2000::/3 global unicast on its face; carries the metadata address.
        self.assertFalse(_address_is_global("64:ff9b::a9fe:a9fe"))
        self.assertFalse(_address_is_global("64:ff9b::7f00:1"))
        self.assertFalse(_address_allowed_at_connect("64:ff9b::a9fe:a9fe"))

    def test_nat64_carrying_a_public_address_allowed(self):
        self.assertTrue(_address_is_global("64:ff9b::8.8.8.8"))

    def test_ipv4_compatible_form_blocked(self):
        # The deprecated ::/96 form, likewise global-looking to is_global.
        self.assertFalse(_address_is_global("::a9fe:a9fe"))
        self.assertFalse(_address_allowed_at_connect("::a9fe:a9fe"))

    def test_sixtofour_and_teredo_stay_blocked(self):
        # Python rejects these ranges wholesale. Unwrapping must not promote a
        # blocked address to an allowed one just because the payload is global.
        self.assertFalse(_address_is_global("2002:a9fe:a9fe::"))
        self.assertFalse(_address_is_global("2002:0808:0808::"))
        self.assertFalse(_address_is_global("2001:0:4136:e378:8000:63bf:3fff:fdd2"))

    def test_plain_addresses_unaffected(self):
        self.assertTrue(_address_is_global("142.250.1.1"))
        self.assertTrue(_address_is_global("2607:f8b0:4004:c07::64"))
        self.assertFalse(_address_is_global("not-an-ip"))

    def test_tunnelled_form_blocked_at_ingress(self):
        with mock.patch(
            "url_guard.socket.getaddrinfo",
            return_value=_addrinfo("64:ff9b::a9fe:a9fe", family=socket.AF_INET6),
        ):
            self.assertIsNotNone(validate_url("http://nat64.example/x"))


class ProxyEndpointParsingTests(unittest.TestCase):
    def test_explicit_port(self):
        self.assertEqual(_proxy_endpoint("http://127.0.0.1:9050"), ("127.0.0.1", 9050))

    def test_default_port_per_scheme(self):
        self.assertEqual(_proxy_endpoint("socks5://127.0.0.1"), ("127.0.0.1", 1080))
        self.assertEqual(_proxy_endpoint("http://127.0.0.1"), ("127.0.0.1", 80))

    def test_bare_host_port(self):
        self.assertEqual(_proxy_endpoint("127.0.0.1:8080"), ("127.0.0.1", 8080))

    def test_hostname_lowercased(self):
        self.assertEqual(_proxy_endpoint("http://LocalHost.:9050"), ("localhost", 9050))

    def test_ipv6_literal(self):
        self.assertEqual(_proxy_endpoint("http://[::1]:9050"), ("::1", 9050))

    def test_empty_and_invalid(self):
        self.assertIsNone(_proxy_endpoint(""))
        self.assertIsNone(_proxy_endpoint("   "))
        self.assertIsNone(_proxy_endpoint(None))
        self.assertIsNone(_proxy_endpoint("http://"))


class GuardedGetaddrinfoTests(unittest.TestCase):
    def setUp(self):
        # Default state: no proxy configured, so no loopback destination allowed.
        saved = set(url_guard._allowed_loopback_endpoints)
        url_guard._allowed_loopback_endpoints = set()
        self.addCleanup(lambda: setattr(url_guard, "_allowed_loopback_endpoints", saved))

    def test_internal_only_raises(self):
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("169.254.169.254")):
            with self.assertRaises(socket.gaierror):
                _guarded_getaddrinfo("metadata", 80)

    def test_filters_internal_keeps_global(self):
        # Split-horizon rebinding: keep the public address, drop the internal one.
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("142.250.1.1", "10.0.0.1")):
            results = _guarded_getaddrinfo("mixed", 80)
        self.assertEqual([r[4][0] for r in results], ["142.250.1.1"])

    def test_loopback_blocked_without_matching_proxy(self):
        # The advisory case: an m3u8 segment URL pointing at a loopback service.
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")):
            with self.assertRaises(socket.gaierror):
                _guarded_getaddrinfo("127.0.0.1", 9999)

    def test_loopback_allowed_at_configured_proxy_endpoint(self):
        url_guard._allowed_loopback_endpoints = {("127.0.0.1", 9050)}
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")):
            results = _guarded_getaddrinfo("127.0.0.1", 9050)
        self.assertEqual([r[4][0] for r in results], ["127.0.0.1"])

    def test_loopback_blocked_at_other_port_on_proxy_host(self):
        # Same host as the proxy, different port: still off limits.
        url_guard._allowed_loopback_endpoints = {("127.0.0.1", 9050)}
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")):
            with self.assertRaises(socket.gaierror):
                _guarded_getaddrinfo("127.0.0.1", 9999)

    def test_proxy_reachable_by_hostname(self):
        url_guard._allowed_loopback_endpoints = {("localhost", 9050)}
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")):
            results = _guarded_getaddrinfo("localhost", 9050)
        self.assertEqual([r[4][0] for r in results], ["127.0.0.1"])

    def test_string_port_is_normalised(self):
        url_guard._allowed_loopback_endpoints = {("127.0.0.1", 9050)}
        with mock.patch("url_guard._real_getaddrinfo", return_value=_addrinfo("127.0.0.1")):
            results = _guarded_getaddrinfo("127.0.0.1", "9050")
        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 setUp(self):
        original, saved = socket.getaddrinfo, set(url_guard._allowed_loopback_endpoints)
        self.addCleanup(lambda: setattr(socket, "getaddrinfo", original))
        self.addCleanup(lambda: setattr(url_guard, "_allowed_loopback_endpoints", saved))
        # Keep the host's own environment out of the assertions below.
        patcher = mock.patch("url_guard.urllib.request.getproxies", return_value={})
        self.getproxies = patcher.start()
        self.addCleanup(patcher.stop)

    def test_install_replaces_and_is_idempotent(self):
        install_socket_guard()
        self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo)
        # Re-installing must not wrap the wrapper (real fn captured at import).
        install_socket_guard()
        self.assertIs(socket.getaddrinfo, url_guard._guarded_getaddrinfo)

    def test_no_proxy_means_no_loopback_allowance(self):
        install_socket_guard()
        self.assertEqual(url_guard._allowed_loopback_endpoints, set())

    def test_explicit_proxy_is_registered(self):
        install_socket_guard(proxy_urls=("socks5://127.0.0.1:9050",))
        self.assertEqual(url_guard._allowed_loopback_endpoints, {("127.0.0.1", 9050)})

    def test_unset_proxy_option_is_ignored(self):
        # ytdl_opts.get('proxy') is None when the operator configured no proxy.
        install_socket_guard(proxy_urls=(None,))
        self.assertEqual(url_guard._allowed_loopback_endpoints, set())

    def test_environment_proxies_are_registered(self):
        self.getproxies.return_value = {"http": "http://127.0.0.1:8080"}
        install_socket_guard()
        self.assertEqual(url_guard._allowed_loopback_endpoints, {("127.0.0.1", 8080)})

    def test_endpoints_reset_between_installs(self):
        install_socket_guard(proxy_urls=("http://127.0.0.1:8080",))
        install_socket_guard(proxy_urls=(None,))
        self.assertEqual(url_guard._allowed_loopback_endpoints, set())


if __name__ == "__main__":
    unittest.main()