feat: added restrictions on CCIP read durin calls

SSRF Mitigation for CCIP Read:

- validate_ccip_url_scheme() — HTTPS-only by default; HTTP allowed via opt-in
- validate_ccip_url_host() / async_validate_ccip_url_host() — resolves hostname and blocks private/reserved IP ranges (127.0.0.0/8, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, etc.)
- Type aliases: CcipUrlValidator, AsyncCcipUrlValidator
- Provider config (base.py, async_base.py):
- ccip_read_allow_http: bool = False
- ccip_read_url_validator — optional user-supplied hook to reject/allow URLs
- Handler changes (exception_handling.py, async_exception_handling.py):
- Scheme + host validation before each HTTP request
- allow_redirects=False on all requests
- Validation failures continue to next URL (consistent with existing error handling)

- Wiring (eth.py, async_eth.py):
- _durin_call passes provider config to handlers

Tests:
- tests/core/utilities/test_ccip_url_validation.py — 23 unit tests for scheme/host validation
- tests/core/contracts/test_offchain_lookup.py — 6 new integration tests (HTTP rejection, allow_http, custom validator, private IP blocking, redirect prevention)
- Updated test mocks to patch socket.getaddrinfo and assert allow_redirects=False

Author: fselmo

tests/core/contracts/test_offchain_lookup.py

@@ -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)

tests/core/utilities/test_ccip_url_validation.py

@@ -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