diff --git a/src/backend/InvenTree/InvenTree/helpers_model.py b/src/backend/InvenTree/InvenTree/helpers_model.py index 925046aef1..7fe10777b3 100644 --- a/src/backend/InvenTree/InvenTree/helpers_model.py +++ b/src/backend/InvenTree/InvenTree/helpers_model.py @@ -1,5 +1,7 @@ """Provides helper functions used throughout the InvenTree project that access the database.""" +import contextlib +import contextvars import io import ipaddress import socket @@ -85,6 +87,66 @@ def construct_absolute_url(*arg, base_url=None, request=None): return urljoin(base_url, relative_url) +_ssrf_guard_active = contextvars.ContextVar('ssrf_guard_active', default=False) +_real_getaddrinfo = socket.getaddrinfo + + +def _is_unsafe_ip(host: str) -> bool: + """Return True if the given (already-resolved) address string is private/reserved.""" + ip = ipaddress.ip_address(host) + return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved + + +def _guarded_getaddrinfo(host, *args, **kwargs): + """Drop-in replacement for socket.getaddrinfo which enforces SSRF checks while active. + + DNS resolution is otherwise delegated to the OS resolver on every connection + attempt an HTTP client makes - including ones made well after an earlier, + now-discarded resolution was validated by `validate_url_no_ssrf()`. An attacker + controlling authoritative DNS for their own domain can answer a first ("check") + lookup with a public IP and a later ("use") lookup with a private/internal one + (DNS rebinding), bypassing a validation step that only inspects the first answer. + + Patching this at the socket layer means the *actual* resolution used to open the + real connection - whichever call that turns out to be - is the one that gets + checked, so there is no window between "validated" and "connected" for the + answer to change. The guard is only enforced while `ssrf_safe_context()` is + active in the current context, so unrelated resolutions (e.g. other threads + talking to the database/cache) are never affected. + """ + result = _real_getaddrinfo(host, *args, **kwargs) + + if _ssrf_guard_active.get(): + for _family, _type, _proto, _canonname, sockaddr in result: + if _is_unsafe_ip(sockaddr[0]): + raise socket.gaierror( + f'Host {host!r} resolved to a private or reserved address' + ) + + return result + + +socket.getaddrinfo = _guarded_getaddrinfo + + +@contextlib.contextmanager +def ssrf_safe_context(): + """Enforce SSRF IP validation on every DNS resolution performed within this block. + + Wrap any code which resolves a user/attacker-influenced hostname and then + connects to it (e.g. `requests.get()`, or a library like WeasyPrint doing its + own resolution internally) in this context manager so that the resolution + backing the *actual* connection is validated, not just an earlier, separate + lookup whose result is discarded. See `_guarded_getaddrinfo` for why this + closes the DNS-rebinding TOCTOU gap that a standalone pre-check cannot. + """ + token = _ssrf_guard_active.set(True) + try: + yield + finally: + _ssrf_guard_active.reset(token) + + def validate_url_no_ssrf(url): """Validate that a URL does not point to a private/internal network address. @@ -109,9 +171,7 @@ def validate_url_no_ssrf(url): raise ValueError(_('Invalid URL: hostname could not be resolved')) for _family, _type, _proto, _canonname, sockaddr in addrinfo: - ip = ipaddress.ip_address(sockaddr[0]) - - if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: + if _is_unsafe_ip(sockaddr[0]): raise ValueError(_('URL points to a private or reserved IP address')) @@ -158,36 +218,40 @@ def download_image_from_url( headers = {'User-Agent': user_agent} if user_agent else None try: - response = requests.get( - remote_url, - timeout=timeout, - allow_redirects=False, - stream=True, - headers=headers, - ) - - # Handle redirects manually to validate each destination - max_redirects = 5 - redirect_count = 0 - - while response.is_redirect and redirect_count < max_redirects: - redirect_url = response.headers.get('Location') - if not redirect_url: - break - - # Validate the redirect destination against SSRF - validator(redirect_url) - validate_url_no_ssrf(redirect_url) - - redirect_count += 1 + # SSRF protection: guard every DNS resolution made while actually connecting + # (not just the pre-check above) so a hostname cannot pass validation with + # one IP and then be connected to via a different, private one (DNS rebinding). + with ssrf_safe_context(): response = requests.get( - redirect_url, + remote_url, timeout=timeout, allow_redirects=False, stream=True, headers=headers, ) + # Handle redirects manually to validate each destination + max_redirects = 5 + redirect_count = 0 + + while response.is_redirect and redirect_count < max_redirects: + redirect_url = response.headers.get('Location') + if not redirect_url: + break + + # Validate the redirect destination against SSRF + validator(redirect_url) + validate_url_no_ssrf(redirect_url) + + redirect_count += 1 + response = requests.get( + redirect_url, + timeout=timeout, + allow_redirects=False, + stream=True, + headers=headers, + ) + if redirect_count >= max_redirects: raise ValueError(_('Too many redirects')) diff --git a/src/backend/InvenTree/InvenTree/tests.py b/src/backend/InvenTree/InvenTree/tests.py index 52251ea1cd..bf9bb13790 100644 --- a/src/backend/InvenTree/InvenTree/tests.py +++ b/src/backend/InvenTree/InvenTree/tests.py @@ -776,6 +776,41 @@ class TestHelpers(TestCase): large_img, timeout=10, max_size=10 * 1024 * 1024 ) + def test_download_image_dns_rebind_blocked(self): + """A hostname that resolves safely for validation but privately for the real fetch must be blocked. + + Regression test: `validate_url_no_ssrf()` used to resolve the hostname once, + validate that result, and then discard it - the actual `requests.get()` call + resolved the same hostname again independently. An attacker controlling DNS + for their own domain could answer the first ("check") lookup with a public IP + and every subsequent ("use") lookup with a private/internal one (DNS + rebinding), so the real connection went somewhere the validator never saw. + """ + public_addrinfo = [(2, 1, 6, '', ('93.184.216.34', 0))] + private_addrinfo = [(2, 1, 6, '', ('127.0.0.1', 0))] + + calls = {'n': 0} + + def rebinding_getaddrinfo(host, *args, **kwargs): + calls['n'] += 1 + return public_addrinfo if calls['n'] == 1 else private_addrinfo + + with mock.patch.object( + InvenTree.helpers_model, + '_real_getaddrinfo', + side_effect=rebinding_getaddrinfo, + ): + # The pre-check sees the safe public IP and passes; the *actual* + # connection attempt made by requests.get() must be independently + # guarded and must fail once it resolves to the private IP. + with self.assertRaisesRegex(Exception, 'Connection error'): + InvenTree.helpers_model.download_image_from_url( + 'http://rebind.example.com/image.png' + ) + + # Both the validation-time and connection-time lookups must have happened. + self.assertEqual(calls['n'], 2) + def test_model_mixin(self): """Test the getModelsWithMixin function.""" from InvenTree.models import InvenTreeBarcodeMixin diff --git a/src/backend/InvenTree/report/fetcher.py b/src/backend/InvenTree/report/fetcher.py index 114941ac34..ae4bc982a4 100644 --- a/src/backend/InvenTree/report/fetcher.py +++ b/src/backend/InvenTree/report/fetcher.py @@ -22,8 +22,16 @@ class InvenTreeURLFetcher(URLFetcher): scheme = parsed.scheme.lower() if scheme in ('data', 'http', 'https'): + from InvenTree.helpers_model import ssrf_safe_context + self._validate_http_url(url, parsed) - return super().fetch(url, headers) + + # SSRF protection: guard the DNS resolution WeasyPrint performs internally + # when it actually opens the connection, not just the pre-check above - a + # hostname could otherwise pass validation with one IP and then be + # connected to via a different, private one moments later (DNS rebinding). + with ssrf_safe_context(): + return super().fetch(url, headers) if scheme == 'file': logger.warning("InvenTreeURLFetcher: blocked file:// URL: '%s'", url) diff --git a/src/backend/InvenTree/report/tests.py b/src/backend/InvenTree/report/tests.py index 780245420c..3d4f558ea4 100644 --- a/src/backend/InvenTree/report/tests.py +++ b/src/backend/InvenTree/report/tests.py @@ -1,6 +1,7 @@ """Unit testing for the various report models.""" import os +import socket import tempfile from io import StringIO from pathlib import Path @@ -1077,6 +1078,49 @@ class URLFetcherTest(TestCase): self.fetcher.fetch('data:image/png;base64,abc123') self.fetcher.fetch('data:text/css;base64,abc123') + def test_dns_rebind_is_blocked_at_fetch_time(self): + """A hostname that resolves safely for validation but privately for the real fetch must be blocked. + + Regression test: `validate_url_no_ssrf()` used to resolve the hostname once, + validate that result, and then discard it, so `InvenTreeURLFetcher.fetch()` + would delegate straight to WeasyPrint's own fetcher, which resolves the + hostname *again* independently. An attacker controlling DNS for their own + domain could answer the first ("check") lookup with a public IP and every + subsequent ("use") lookup with a private/internal one (DNS rebinding), so the + real request WeasyPrint made went somewhere the validator never saw. + """ + set_global_setting('REPORT_FETCH_URLS', True, change_user=None) + + public_addrinfo = [(2, 1, 6, '', ('93.184.216.34', 0))] + private_addrinfo = [(2, 1, 6, '', ('127.0.0.1', 0))] + + calls = {'n': 0} + + def rebinding_getaddrinfo(host, *args, **kwargs): + calls['n'] += 1 + return public_addrinfo if calls['n'] == 1 else private_addrinfo + + def fake_weasyprint_fetch(_self, url, headers=None): + # Simulate WeasyPrint's own fetch-time DNS resolution, performed + # independently of the validation InvenTreeURLFetcher already did. + socket.getaddrinfo('rebind.example.com', None) + return {'string': b'should never be reached'} + + import InvenTree.helpers_model as helpers_model + + with patch.object( + helpers_model, '_real_getaddrinfo', side_effect=rebinding_getaddrinfo + ): + with patch( + 'weasyprint.urls.URLFetcher.fetch', side_effect=fake_weasyprint_fetch + ): + with self.assertRaises(socket.gaierror): + self.fetcher.fetch('http://rebind.example.com/image.png') + + # The validation-time lookup (safe) and the fetch-time lookup (private) + # must both have happened for this to be a meaningful regression test. + self.assertEqual(calls['n'], 2) + class DefaultTemplateFileTest(TestCase): """Unit tests for building the default report and label template files."""