diff --git a/src/google/adk/tools/load_web_page.py b/src/google/adk/tools/load_web_page.py index d1a679f0e4..ebb2c798c0 100644 --- a/src/google/adk/tools/load_web_page.py +++ b/src/google/adk/tools/load_web_page.py @@ -158,8 +158,40 @@ def _is_blocked_hostname(hostname: str) -> bool: ) +# NAT64 well-known prefix (RFC 6052). Addresses in this range embed an IPv4 +# address in their low 32 bits and are classified as globally routable by +# ``ipaddress`` even when the embedded IPv4 address is internal, so they must be +# unwrapped and re-checked to avoid an SSRF bypass on NAT64-enabled networks. +_NAT64_WELL_KNOWN_PREFIX = ipaddress.ip_network('64:ff9b::/96') + + +def _embedded_ipv4(address: _ResolvedAddress) -> ipaddress.IPv4Address | None: + """Returns the IPv4 address embedded in an IPv6 address, if any. + + Covers the IPv4 embeddings that resolve to an IPv4 destination but are not + themselves flagged as non-global by ``ipaddress``: the NAT64 well-known + prefix (``64:ff9b::/96``) and the deprecated IPv4-compatible form + (``::a.b.c.d``). The IPv4-mapped form (``::ffff:a.b.c.d``) is included for + completeness. + """ + if not isinstance(address, ipaddress.IPv6Address): + return None + if address.ipv4_mapped is not None: + return address.ipv4_mapped + if address in _NAT64_WELL_KNOWN_PREFIX: + return ipaddress.IPv4Address(int(address) & 0xFFFFFFFF) + # Deprecated IPv4-compatible ``::a.b.c.d`` (high 96 bits zero), excluding the + # unspecified address ``::`` and loopback ``::1``. + if int(address) > 1 and (int(address) >> 32) == 0: + return ipaddress.IPv4Address(int(address) & 0xFFFFFFFF) + return None + + def _is_blocked_address(address: _ResolvedAddress) -> bool: - return not address.is_global + if not address.is_global: + return True + embedded_ipv4 = _embedded_ipv4(address) + return embedded_ipv4 is not None and not embedded_ipv4.is_global def _resolve_host_addresses(hostname: str) -> tuple[_ResolvedAddress, ...]: diff --git a/tests/unittests/tools/test_load_web_page.py b/tests/unittests/tools/test_load_web_page.py index 1639ddb036..74b7a118f3 100644 --- a/tests/unittests/tools/test_load_web_page.py +++ b/tests/unittests/tools/test_load_web_page.py @@ -14,12 +14,14 @@ from __future__ import annotations +import ipaddress import os import socket from unittest import mock from google.adk.tools.load_web_page import load_web_page import google.adk.tools.load_web_page as load_web_page_module +import pytest import requests @@ -361,6 +363,97 @@ def test_load_web_page_passes_timeout_to_proxied_get(monkeypatch): ) +@pytest.mark.parametrize( + 'ipv6_address', + [ + '64:ff9b::7f00:1', # NAT64-wrapped 127.0.0.1 (loopback) + '64:ff9b::a9fe:a9fe', # NAT64-wrapped 169.254.169.254 (cloud metadata) + '64:ff9b::a00:1', # NAT64-wrapped 10.0.0.1 (private) + '::7f00:1', # IPv4-compatible 127.0.0.1 (loopback) + '::a9fe:a9fe', # IPv4-compatible 169.254.169.254 (cloud metadata) + '::ffff:7f00:1', # IPv4-mapped 127.0.0.1 (loopback) + ], +) +def test_is_blocked_address_blocks_embedded_internal_ipv4(ipv6_address): + """Embedded IPv4 forms that resolve to internal targets must be blocked.""" + address = ipaddress.ip_address(ipv6_address) + assert load_web_page_module._is_blocked_address(address) + + +@pytest.mark.parametrize( + 'ipv6_address', + [ + '64:ff9b::5db8:d822', # NAT64-wrapped 93.184.216.34 (public) + '2606:2800:220:1:248:1893:25c8:1946', # ordinary global IPv6 + ], +) +def test_is_blocked_address_allows_embedded_public_ipv4(ipv6_address): + """Embedded IPv4 forms that resolve to public targets stay allowed.""" + address = ipaddress.ip_address(ipv6_address) + assert not load_web_page_module._is_blocked_address(address) + + +def test_load_web_page_blocks_nat64_wrapped_metadata_literal(monkeypatch): + _clear_proxy_env(monkeypatch) + mock_get = mock.Mock() + monkeypatch.setattr(load_web_page_module.requests, 'get', mock_get) + mock_send = mock.Mock() + monkeypatch.setattr(load_web_page_module.HTTPAdapter, 'send', mock_send) + + # 64:ff9b::a9fe:a9fe embeds 169.254.169.254 (the cloud metadata endpoint). + url = 'http://[64:ff9b::a9fe:a9fe]/latest/meta-data/' + result = load_web_page(url) + + assert result == f'Failed to fetch url: {url}' + mock_get.assert_not_called() + mock_send.assert_not_called() + + +def test_load_web_page_blocks_ipv4_compatible_loopback_literal(monkeypatch): + _clear_proxy_env(monkeypatch) + mock_get = mock.Mock() + monkeypatch.setattr(load_web_page_module.requests, 'get', mock_get) + mock_send = mock.Mock() + monkeypatch.setattr(load_web_page_module.HTTPAdapter, 'send', mock_send) + + # ::7f00:1 is the deprecated IPv4-compatible form of 127.0.0.1. + url = 'http://[::7f00:1]/' + result = load_web_page(url) + + assert result == f'Failed to fetch url: {url}' + mock_get.assert_not_called() + mock_send.assert_not_called() + + +def test_load_web_page_blocks_hostname_resolving_to_nat64_internal(monkeypatch): + _clear_proxy_env(monkeypatch) + monkeypatch.setattr( + load_web_page_module.socket, + 'getaddrinfo', + mock.Mock( + return_value=[( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + '', + ('64:ff9b::a9fe:a9fe', 0, 0, 0), + )] + ), + ) + mock_get = mock.Mock() + monkeypatch.setattr(load_web_page_module.requests, 'get', mock_get) + mock_send = mock.Mock() + monkeypatch.setattr(load_web_page_module.HTTPAdapter, 'send', mock_send) + + result = load_web_page('http://nat64.example.test/latest/meta-data/') + + assert ( + result == 'Failed to fetch url: http://nat64.example.test/latest/meta-data/' + ) + mock_get.assert_not_called() + mock_send.assert_not_called() + + def test_load_web_page_returns_failure_on_timeout(monkeypatch): """Verify that a timeout exception is converted to a failed to fetch message.""" _clear_proxy_env(monkeypatch)