CVE-2026-40072

@@ -1,4 +1,5 @@
 import pytest
+import socket
 
 from eth_abi import (
     abi,
@@ -14,10 +15,14 @@
     to_hex_if_bytes,
 )
 from web3.exceptions import (
+    MultipleFailedRequests,
     OffchainLookup,
     TooManyRequests,
     Web3ValidationError,
 )
+from web3.utils import (
+    handle_offchain_lookup,
+)
 
 # "test offchain lookup" as an abi-encoded string
 OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA = "0x0000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000001474657374206f6666636861696e206c6f6f6b7570000000000000000000000000"  # noqa: E501
@@ -208,3 +213,198 @@ def test_offchain_lookup_raises_on_continuous_redirect(
     )
     with pytest.raises(TooManyRequests, match="Too many CCIP read redirects"):
         offchain_lookup_contract.caller.continuousOffchainLookup()
+
+
+# -- SSRF mitigation tests -- #
+
+
+def test_offchain_lookup_rejects_http_urls_by_default(
+    offchain_lookup_contract,
+    monkeypatch,
+):
+    """HTTP URLs should be rejected by default (only HTTPS allowed)."""
+    to_hex_if_bytes(offchain_lookup_contract.address).lower()
+
+    # Patch getaddrinfo so host validation passes
+    def _mock_getaddrinfo(host, port, *args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("1.2.3.4", 0))]
+
+    monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+
+    payload = {
+        "sender": offchain_lookup_contract.address,
+        "urls": [
+            "http://web3.py/gateway/{sender}/{data}.json",
+        ],
+        "callData": OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA,
+        "callbackFunction": b"\x00\x00\x00\x00",
+        "extraData": b"",
+    }
+    transaction = {"to": offchain_lookup_contract.address}
+
+    with pytest.raises(MultipleFailedRequests):
+        handle_offchain_lookup(payload, transaction)
+
+
+def test_offchain_lookup_allows_http_urls_when_configured(
+    w3,
+    offchain_lookup_contract,
+    monkeypatch,
+):
+    """HTTP URLs should be allowed when ccip_read_allow_http=True on provider."""
+    normalized_address = to_hex_if_bytes(offchain_lookup_contract.address)
+    mock_offchain_lookup_request_response(
+        monkeypatch,
+        mocked_request_url=f"https://web3.py/gateway/{normalized_address}/{OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA}.json",  # noqa: E501
+        mocked_json_data=WEB3PY_AS_HEXBYTES,
+    )
+
+    w3.provider.ccip_read_allow_http = True
+    try:
+        response = offchain_lookup_contract.caller.testOffchainLookup(
+            OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA
+        )
+        assert abi.decode(["string"], response)[0] == "web3py"
+    finally:
+        w3.provider.ccip_read_allow_http = False
+
+
+def test_offchain_lookup_custom_url_validator_rejects(
+    offchain_lookup_contract,
+    monkeypatch,
+):
+    """Custom url_validator on provider that rejects should skip URLs."""
+    from web3.utils.exception_handling import (
+        handle_offchain_lookup,
+    )
+
+    # Patch getaddrinfo so host validation passes
+    def _mock_getaddrinfo(host, port, *args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("1.2.3.4", 0))]
+
+    monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+
+    def reject_all(url):
+        raise Web3ValidationError(f"Rejected by policy: {url}")
+
+    payload = {
+        "sender": offchain_lookup_contract.address,
+        "urls": [
+            "https://web3.py/gateway/{sender}/{data}.json",
+        ],
+        "callData": OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA,
+        "callbackFunction": b"\x00\x00\x00\x00",
+        "extraData": b"",
+    }
+    transaction = {"to": offchain_lookup_contract.address}
+
+    with pytest.raises(MultipleFailedRequests):
+        handle_offchain_lookup(payload, transaction, url_validator=reject_all)
+
+
+def test_offchain_lookup_custom_url_validator_on_provider(
+    w3,
+    offchain_lookup_contract,
+    monkeypatch,
+):
+    """Custom url_validator set on provider is honored via _durin_call."""
+
+    # Patch getaddrinfo so host validation passes
+    def _mock_getaddrinfo(host, port, *args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("1.2.3.4", 0))]
+
+    monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+
+    validator_calls = []
+
+    def tracking_validator(url):
+        validator_calls.append(url)
+        raise Web3ValidationError(f"Rejected by policy: {url}")
+
+    w3.provider.ccip_read_url_validator = tracking_validator
+    try:
+        with pytest.raises(MultipleFailedRequests):
+            offchain_lookup_contract.caller.testOffchainLookup(
+                OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA
+            )
+        assert len(validator_calls) > 0
+    finally:
+        w3.provider.ccip_read_url_validator = None
+
+
+def test_offchain_lookup_rejects_private_ip(
+    offchain_lookup_contract,
+    monkeypatch,
+):
+    """URLs resolving to private IPs should be rejected."""
+    from web3.utils.exception_handling import (
+        handle_offchain_lookup,
+    )
+
+    def _mock_getaddrinfo(host, port, *args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 0))]
+
+    monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+
+    payload = {
+        "sender": offchain_lookup_contract.address,
+        "urls": [
+            "https://web3.py/gateway/{sender}/{data}.json",
+        ],
+        "callData": OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA,
+        "callbackFunction": b"\x00\x00\x00\x00",
+        "extraData": b"",
+    }
+    transaction = {"to": offchain_lookup_contract.address}
+
+    with pytest.raises(MultipleFailedRequests):
+        handle_offchain_lookup(payload, transaction)
+
+
+def test_offchain_lookup_redirect_not_followed(
+    offchain_lookup_contract,
+    monkeypatch,
+):
+    """302 redirects should not be followed (treated as non-2xx, try next URL)."""
+    from web3.utils.exception_handling import (
+        handle_offchain_lookup,
+    )
+
+    # Patch getaddrinfo so host validation passes
+    def _mock_getaddrinfo(host, port, *args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("1.2.3.4", 0))]
+
+    monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+
+    class Mock302Response:
+        status_code = 302
+
+        @staticmethod
+        def raise_for_status():
+            raise Exception("called raise_for_status()")
+
+    def _mock_get(*args, **kwargs):
+        assert kwargs.get("allow_redirects") is False
+        return Mock302Response()
+
+    def _mock_post(*args, **kwargs):
+        assert kwargs.get("allow_redirects") is False
+        return Mock302Response()
+
+    monkeypatch.setattr("requests.Session.get", _mock_get)
+    monkeypatch.setattr("requests.Session.post", _mock_post)
+
+    payload = {
+        "sender": offchain_lookup_contract.address,
+        "urls": [
+            "https://web3.py/gateway/{sender}/{data}.json",
+            "https://web3.py/gateway",
+        ],
+        "callData": OFFCHAIN_LOOKUP_CONTRACT_TEST_DATA,
+        "callbackFunction": b"\x00\x00\x00\x00",
+        "extraData": b"",
+    }
+    transaction = {"to": offchain_lookup_contract.address}
+
+    with pytest.raises(MultipleFailedRequests):
+        handle_offchain_lookup(payload, transaction)

@@ -0,0 +1,122 @@
+import pytest
+import socket
+
+from web3.exceptions import (
+    Web3ValidationError,
+)
+from web3.utils.ccip_url_validation import (
+    validate_ccip_url_host,
+    validate_ccip_url_scheme,
+)
+
+# -- validate_ccip_url_scheme tests -- #
+
+
+class TestValidateCcipUrlScheme:
+    def test_https_passes(self):
+        validate_ccip_url_scheme("https://example.com/api", allow_http=False)
+
+    def test_http_fails_by_default(self):
+        with pytest.raises(Web3ValidationError, match="non-HTTPS"):
+            validate_ccip_url_scheme("http://example.com/api")
+
+    def test_http_passes_with_allow_http(self):
+        validate_ccip_url_scheme("http://example.com/api", allow_http=True)
+
+    def test_ftp_always_fails(self):
+        with pytest.raises(Web3ValidationError, match="not allowed"):
+            validate_ccip_url_scheme("ftp://example.com/file")
+
+    def test_ftp_fails_even_with_allow_http(self):
+        with pytest.raises(Web3ValidationError, match="not allowed"):
+            validate_ccip_url_scheme("ftp://example.com/file", allow_http=True)
+
+    def test_file_scheme_fails(self):
+        with pytest.raises(Web3ValidationError, match="not allowed"):
+            validate_ccip_url_scheme("file:///etc/passwd")
+
+    def test_file_scheme_fails_with_allow_http(self):
+        with pytest.raises(Web3ValidationError, match="not allowed"):
+            validate_ccip_url_scheme("file:///etc/passwd", allow_http=True)
+
+
+# -- validate_ccip_url_host tests -- #
+
+
+class TestValidateCcipUrlHost:
+    def _patch_getaddrinfo(self, monkeypatch, ip):
+        def _mock_getaddrinfo(host, port, *args, **kwargs):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (ip, 0))]
+
+        monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+
+    def test_public_ip_passes(self, monkeypatch):
+        self._patch_getaddrinfo(monkeypatch, "8.8.8.8")
+        validate_ccip_url_host("https://example.com/api")
+
+    @pytest.mark.parametrize(
+        "blocked_ip",
+        [
+            "127.0.0.1",
+            "127.0.0.2",
+            "10.0.0.1",
+            "10.255.255.255",
+            "172.16.0.1",
+            "172.31.255.255",
+            "192.168.0.1",
+            "192.168.1.100",
+            "169.254.0.1",
+            "0.0.0.0",
+        ],
+    )
+    def test_blocked_ipv4(self, monkeypatch, blocked_ip):
+        self._patch_getaddrinfo(monkeypatch, blocked_ip)
+        with pytest.raises(Web3ValidationError, match="blocked private/reserved"):
+            validate_ccip_url_host("https://example.com/api")
+
+    def test_blocked_ipv6_loopback(self, monkeypatch):
+        def _mock_getaddrinfo(host, port, *args, **kwargs):
+            return [(socket.AF_INET6, socket.SOCK_STREAM, 0, "", ("::1", 0, 0, 0))]
+
+        monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+        with pytest.raises(Web3ValidationError, match="blocked private/reserved"):
+            validate_ccip_url_host("https://example.com/api")
+
+    def test_unresolvable_hostname(self, monkeypatch):
+        def _mock_getaddrinfo(host, port, *args, **kwargs):
+            raise socket.gaierror("Name or service not known")
+
+        monkeypatch.setattr("socket.getaddrinfo", _mock_getaddrinfo)
+        with pytest.raises(Web3ValidationError, match="could not be resolved"):
+            validate_ccip_url_host("https://nonexistent.invalid/api")
+
+    def test_no_hostname(self):
+        with pytest.raises(Web3ValidationError, match="no hostname"):
+            validate_ccip_url_host("https:///path")
+
+
+# -- custom validator tests -- #
+
+
+class TestCustomUrlValidator:
+    def test_custom_validator_called_and_can_reject(self):
+        calls = []
+
+        def reject_validator(url):
+            calls.append(url)
+            raise Web3ValidationError(f"Rejected: {url}")
+
+        with pytest.raises(Web3ValidationError, match="Rejected"):
+            reject_validator("https://example.com/api")
+
+        assert len(calls) == 1
+        assert calls[0] == "https://example.com/api"
+
+    def test_custom_validator_can_allow(self):
+        calls = []
+
+        def allow_validator(url):
+            calls.append(url)
+
+        allow_validator("https://example.com/api")
+        assert len(calls) == 1

@@ -1,4 +1,5 @@
 import pytest
+import socket
 
 from aiohttp import (
     ClientSession,
@@ -13,6 +14,19 @@
     Web3ValidationError,
 )
 
+
+def _mock_getaddrinfo_public(monkeypatch):
+    """Patch socket.getaddrinfo to return a public IP for CCIP test domains."""
+    _original = socket.getaddrinfo
+
+    def _patched(host, port, *args, **kwargs):
+        if host == "web3.py":
+            return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("1.2.3.4", 0))]
+        return _original(host, port, *args, **kwargs)
+
+    monkeypatch.setattr("socket.getaddrinfo", _patched)
+
+
 # the encoded calldata for the initiating ``addr(namehash(name))`` call
 ENCODED_ADDR_CALLDATA = "0x3b3b57de42041b0018edd29d7c17154b0c671acc0502ea0b3693cafbeadf58e6beaaa16c"  # noqa: E501
 
@@ -124,6 +138,8 @@ def status(self):
 
 
 def test_offchain_resolution_with_get_request(ens, monkeypatch):
+    _mock_getaddrinfo_public(monkeypatch)
+
     # mock GET response with real return data from 'offchainexample.eth' resolver
     def mock_get(*args, **kwargs):
         return MockHttpSuccessResponse("get", *args, **kwargs)
@@ -134,6 +150,8 @@ def mock_get(*args, **kwargs):
 
 
 def test_offchain_resolution_with_post_request(ens, monkeypatch):
+    _mock_getaddrinfo_public(monkeypatch)
+
     # mock POST response with real return data from 'offchainexample.eth' resolver
     def mock_post(*args, **kwargs):
         return MockHttpSuccessResponse("post", *args, **kwargs)
@@ -150,6 +168,8 @@ def test_offchain_resolution_raises_when_all_supplied_urls_fail(ens):
 
 
 def test_offchain_resolution_with_improperly_formatted_http_response(ens, monkeypatch):
+    _mock_getaddrinfo_public(monkeypatch)
+
     def mock_get(*args, **_):
         return MockHttpBadFormatResponse(*args)
 
@@ -189,6 +209,8 @@ def test_offchain_resolver_function_call_raises_with_ccip_read_disabled(
 
 @pytest.mark.asyncio
 async def test_async_offchain_resolution_with_get_request(async_ens, monkeypatch):
+    _mock_getaddrinfo_public(monkeypatch)
+
     # mock GET response with real return data from 'offchainexample.eth' resolver
     async def mock_get(*args, **kwargs):
         return AsyncMockHttpSuccessResponse("get", *args, **kwargs)
@@ -200,6 +222,8 @@ async def mock_get(*args, **kwargs):
 
 @pytest.mark.asyncio
 async def test_async_offchain_resolution_with_post_request(async_ens, monkeypatch):
+    _mock_getaddrinfo_public(monkeypatch)
+
     # mock POST response with real return data from 'offchainexample.eth' resolver
     async def mock_post(*args, **kwargs):
         return AsyncMockHttpSuccessResponse("post", *args, **kwargs)
@@ -220,6 +244,8 @@ async def test_async_offchain_resolution_raises_when_all_supplied_urls_fail(asyn
 async def test_async_offchain_resolution_with_improperly_formatted_http_response(
     async_ens, monkeypatch
 ):
+    _mock_getaddrinfo_public(monkeypatch)
+
     async def mock_get(*args, **_):
         return AsyncMockHttpBadFormatResponse(*args)
 

@@ -281,6 +281,8 @@ async def _durin_call(
                 durin_calldata = await async_handle_offchain_lookup(
                     offchain_lookup.payload,
                     transaction,
+                    allow_http=self.w3.provider.ccip_read_allow_http,
+                    url_validator=self.w3.provider.ccip_read_url_validator,
                 )
                 transaction["data"] = durin_calldata
 

@@ -260,6 +260,8 @@ def _durin_call(
                 durin_calldata = handle_offchain_lookup(
                     offchain_lookup.payload,
                     transaction,
+                    allow_http=self.w3.provider.ccip_read_allow_http,
+                    url_validator=self.w3.provider.ccip_read_url_validator,
                 )
                 transaction["data"] = durin_calldata
 

@@ -63,6 +63,9 @@
     from web3.providers.persistent import (  # noqa: F401
         RequestProcessor,
     )
+    from web3.utils.ccip_url_validation import (
+        AsyncCcipUrlValidator,
+    )
 
 
 class AsyncBaseProvider:
@@ -78,6 +81,8 @@ class AsyncBaseProvider:
     has_persistent_connection = False
     global_ccip_read_enabled: bool = True
     ccip_read_max_redirects: int = 4
+    ccip_read_allow_http: bool = False
+    ccip_read_url_validator: "AsyncCcipUrlValidator | None" = None
 
     def __init__(
         self,

@@ -50,6 +50,9 @@
     from web3._utils.batching import (
         RequestBatcher,
     )
+    from web3.utils.ccip_url_validation import (
+        CcipUrlValidator,
+    )
 
 
 class BaseProvider:
@@ -65,6 +68,8 @@ class BaseProvider:
     has_persistent_connection = False
     global_ccip_read_enabled: bool = True
     ccip_read_max_redirects: int = 4
+    ccip_read_allow_http: bool = False
+    ccip_read_url_validator: "CcipUrlValidator | None" = None
 
     def __init__(
         self,

@@ -27,11 +27,18 @@
 from web3.types import (
     TxParams,
 )
+from web3.utils.ccip_url_validation import (
+    AsyncCcipUrlValidator,
+    async_validate_ccip_url_host,
+    validate_ccip_url_scheme,
+)
 
 
 async def async_handle_offchain_lookup(
     offchain_lookup_payload: dict[str, Any],
     transaction: TxParams,
+    allow_http: bool = False,
+    url_validator: AsyncCcipUrlValidator | None = None,
 ) -> bytes:
     formatted_sender = to_hex_if_bytes(offchain_lookup_payload["sender"]).lower()
     formatted_data = to_hex_if_bytes(offchain_lookup_payload["callData"]).lower()
@@ -50,16 +57,27 @@ async def async_handle_offchain_lookup(
             .replace("{data}", str(formatted_data))
         )
 
+        try:
+            validate_ccip_url_scheme(formatted_url, allow_http=allow_http)
+            await async_validate_ccip_url_host(formatted_url)
+            if url_validator is not None:
+                await url_validator(formatted_url)
+        except Web3ValidationError:
+            continue
+
         try:
             if "{data}" in url and "{sender}" in url:
                 response = await session.get(
-                    formatted_url, timeout=ClientTimeout(DEFAULT_HTTP_TIMEOUT)
+                    formatted_url,
+                    timeout=ClientTimeout(DEFAULT_HTTP_TIMEOUT),
+                    allow_redirects=False,
                 )
             else:
                 response = await session.post(
                     formatted_url,
                     json={"data": formatted_data, "sender": formatted_sender},
                     timeout=ClientTimeout(DEFAULT_HTTP_TIMEOUT),
+                    allow_redirects=False,
                 )
         except Exception:
             continue  # try next url if timeout or issues making the request

@@ -0,0 +1,105 @@
+import asyncio
+import ipaddress
+import socket
+from typing import (
+    Awaitable,
+    Callable,
+)
+from urllib.parse import (
+    urlparse,
+)
+
+from web3.exceptions import (
+    Web3ValidationError,
+)
+
+CcipUrlValidator = Callable[[str], None]
+AsyncCcipUrlValidator = Callable[[str], Awaitable[None]]
+
+BLOCKED_IP_NETWORKS = [
+    ipaddress.ip_network("127.0.0.0/8"),
+    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("172.16.0.0/12"),
+    ipaddress.ip_network("192.168.0.0/16"),
+    ipaddress.ip_network("169.254.0.0/16"),
+    ipaddress.ip_network("0.0.0.0/8"),
+    ipaddress.ip_network("::1/128"),
+    ipaddress.ip_network("fe80::/10"),
+    ipaddress.ip_network("fc00::/7"),
+    ipaddress.ip_network("::/128"),
+]
+
+
+def validate_ccip_url_scheme(url: str, allow_http: bool = False) -> None:
+    parsed = urlparse(url)
+    scheme = parsed.scheme.lower()
+
+    if scheme == "https":
+        return
+
+    if scheme == "http" and allow_http:
+        return
+
+    if scheme == "http":
+        raise Web3ValidationError(
+            f"CCIP Read request to non-HTTPS URL '{url}' is not allowed. "
+            "Set ``ccip_read_allow_http=True`` on the provider to allow HTTP URLs."
+        )
+
+    raise Web3ValidationError(
+        f"CCIP Read request with scheme '{scheme}' is not allowed. "
+        "Only HTTPS URLs are permitted."
+    )
+
+
+def _check_ip_blocked(ip_str: str) -> bool:
+    try:
+        addr = ipaddress.ip_address(ip_str)
+    except ValueError:
+        return False
+    return any(addr in network for network in BLOCKED_IP_NETWORKS)
+
+
+def validate_ccip_url_host(url: str) -> None:
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        raise Web3ValidationError(f"CCIP Read URL '{url}' has no hostname.")
+
+    try:
+        addrinfos = socket.getaddrinfo(hostname, None)
+    except socket.gaierror:
+        raise Web3ValidationError(
+            f"CCIP Read URL hostname '{hostname}' could not be resolved."
+        )
+
+    for addrinfo in addrinfos:
+        ip_str = str(addrinfo[4][0])
+        if _check_ip_blocked(ip_str):
+            raise Web3ValidationError(
+                f"CCIP Read request to '{url}' is not allowed: "
+                f"resolved IP '{ip_str}' is in a blocked private/reserved range."
+            )
+
+
+async def async_validate_ccip_url_host(url: str) -> None:
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        raise Web3ValidationError(f"CCIP Read URL '{url}' has no hostname.")
+
+    loop = asyncio.get_running_loop()
+    try:
+        addrinfos = await loop.run_in_executor(None, socket.getaddrinfo, hostname, None)
+    except socket.gaierror:
+        raise Web3ValidationError(
+            f"CCIP Read URL hostname '{hostname}' could not be resolved."
+        )
+
+    for addrinfo in addrinfos:
+        ip_str = str(addrinfo[4][0])
+        if _check_ip_blocked(ip_str):
+            raise Web3ValidationError(
+                f"CCIP Read request to '{url}' is not allowed: "
+                f"resolved IP '{ip_str}' is in a blocked private/reserved range."
+            )

@@ -24,11 +24,18 @@
 from web3.types import (
     TxParams,
 )
+from web3.utils.ccip_url_validation import (
+    CcipUrlValidator,
+    validate_ccip_url_host,
+    validate_ccip_url_scheme,
+)
 
 
 def handle_offchain_lookup(
     offchain_lookup_payload: dict[str, Any],
     transaction: TxParams,
+    allow_http: bool = False,
+    url_validator: CcipUrlValidator | None = None,
 ) -> bytes:
     formatted_sender = to_hex_if_bytes(offchain_lookup_payload["sender"]).lower()
     formatted_data = to_hex_if_bytes(offchain_lookup_payload["callData"]).lower()
@@ -47,14 +54,27 @@ def handle_offchain_lookup(
             .replace("{data}", str(formatted_data))
         )
 
+        try:
+            validate_ccip_url_scheme(formatted_url, allow_http=allow_http)
+            validate_ccip_url_host(formatted_url)
+            if url_validator is not None:
+                url_validator(formatted_url)
+        except Web3ValidationError:
+            continue
+
         try:
             if "{data}" in url and "{sender}" in url:
-                response = session.get(formatted_url, timeout=DEFAULT_HTTP_TIMEOUT)
+                response = session.get(
+                    formatted_url,
+                    timeout=DEFAULT_HTTP_TIMEOUT,
+                    allow_redirects=False,
+                )
             else:
                 response = session.post(
                     formatted_url,
                     json={"data": formatted_data, "sender": formatted_sender},
                     timeout=DEFAULT_HTTP_TIMEOUT,
+                    allow_redirects=False,
                 )
         except Exception:
             continue  # try next url if timeout or issues making the request

@@ -39,6 +39,10 @@
     RequestCacheValidationThreshold,
     SimpleCache,
 )
+from .ccip_url_validation import (
+    AsyncCcipUrlValidator,
+    CcipUrlValidator,
+)
 from .exception_handling import (
     handle_offchain_lookup,
 )
@@ -73,6 +77,8 @@
     "async_handle_offchain_lookup",
     "RequestCacheValidationThreshold",
     "SimpleCache",
+    "AsyncCcipUrlValidator",
+    "CcipUrlValidator",
     "EthSubscription",
     "handle_offchain_lookup",
 ]

@@ -64,6 +64,24 @@ def assert_contains_log(
     assert log_entry["transactionHash"] == HexBytes(txn_hash_with_log)
 
 
+def _mock_getaddrinfo_public(
+    monkeypatch: "MonkeyPatch",
+) -> None:
+    # Patch socket.getaddrinfo to return a public IP for CCIP test domains
+    # so that CCIP URL host validation passes during tests. Pass through
+    # to the real getaddrinfo for all other hosts (e.g. 127.0.0.1 for geth).
+    import socket as _socket
+
+    _original_getaddrinfo = _socket.getaddrinfo
+
+    def _patched_getaddrinfo(host: Any, port: Any, *args: Any, **kwargs: Any) -> Any:
+        if host == "web3.py":
+            return [(_socket.AF_INET, _socket.SOCK_STREAM, 0, "", ("1.2.3.4", 0))]
+        return _original_getaddrinfo(host, port, *args, **kwargs)
+
+    monkeypatch.setattr("socket.getaddrinfo", _patched_getaddrinfo)
+
+
 def mock_offchain_lookup_request_response(
     monkeypatch: "MonkeyPatch",
     http_method: Literal["GET", "POST"] = "GET",
@@ -75,6 +93,8 @@ def mock_offchain_lookup_request_response(
     sender: str = None,
     calldata: str = None,
 ) -> None:
+    _mock_getaddrinfo_public(monkeypatch)
+
     class MockedResponse:
         status_code = mocked_status_code
 
@@ -94,6 +114,7 @@ def _mock_specific_request(
         # mock response only to specified url while validating appropriate fields
         if url_from_args == mocked_request_url:
             assert kwargs["timeout"] == DEFAULT_HTTP_TIMEOUT
+            assert kwargs.get("allow_redirects") is False
             if http_method.upper() == "POST":
                 assert kwargs["json"] == {"data": calldata, "sender": sender}
             return MockedResponse()
@@ -121,6 +142,8 @@ def async_mock_offchain_lookup_request_response(
     sender: str = None,
     calldata: str = None,
 ) -> None:
+    _mock_getaddrinfo_public(monkeypatch)
+
     class AsyncMockedResponse:
         status = mocked_status_code
 
@@ -144,6 +167,7 @@ async def _mock_specific_request(
         # mock response only to specified url while validating appropriate fields
         if url_from_args == mocked_request_url:
             assert kwargs["timeout"] == ClientTimeout(DEFAULT_HTTP_TIMEOUT)
+            assert kwargs.get("allow_redirects") is False
             if http_method.upper() == "POST":
                 assert kwargs["json"] == {"data": calldata, "sender": sender}
             return AsyncMockedResponse()