guard json schema ref retrieval against internal targets

src/confluent_kafka/schema_registry/common/json_schema.py

@@ -1,7 +1,10 @@
 import decimal
+import ipaddress
 import logging
+import socket
 from io import BytesIO
 from typing import List, Optional, Set, Union
+from urllib.parse import urlsplit
 
 import httpx
 import referencing
@@ -58,8 +61,39 @@ def __exit__(self, *args):
         return False
 
 
+def _is_blocked_ip(ip) -> bool:
+    mapped = getattr(ip, 'ipv4_mapped', None)
+    if mapped is not None:
+        ip = mapped
+    return (
+        ip.is_private
+        or ip.is_loopback
+        or ip.is_link_local
+        or ip.is_reserved
+        or ip.is_multicast
+        or ip.is_unspecified
+    )
+
+
+def _guard_external_uri(uri: str):
+    parts = urlsplit(uri)
+    if parts.scheme not in ('http', 'https'):
+        raise ValueError("Refusing to retrieve schema from non-HTTP(S) URI: {}".format(uri))
+    host = parts.hostname
+    if not host:
+        raise ValueError("Refusing to retrieve schema from URI without a host: {}".format(uri))
+    try:
+        infos = socket.getaddrinfo(host, parts.port or (443 if parts.scheme == 'https' else 80))
+    except socket.gaierror as ex:
+        raise ValueError("Could not resolve schema URI host {}: {}".format(host, ex))
+    for info in infos:
+        if _is_blocked_ip(ipaddress.ip_address(info[4][0])):
+            raise ValueError("Refusing to retrieve schema from non-public address: {}".format(uri))
+
+
 def _retrieve_via_httpx(uri: str):
-    response = httpx.get(uri)
+    _guard_external_uri(uri)
+    response = httpx.get(uri, follow_redirects=False)
     return Resource.from_contents(response.json(), default_specification=DEFAULT_SPEC)
 
 

tests/schema_registry/test_json_schema_retrieve.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2024 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from unittest import mock
+
+import pytest
+
+from confluent_kafka.schema_registry.common import json_schema
+
+
+# Numeric hosts so resolution needs no network; schemes other than http(s)
+# are rejected before any lookup.
+@pytest.mark.parametrize("uri", [
+    "http://169.254.169.254/latest/meta-data/",
+    "http://127.0.0.1:8080/internal",
+    "http://10.0.0.5/schema",
+    "http://[::1]/schema",
+    "file:///etc/passwd",
+    "gopher://127.0.0.1/x",
+])
+def test_retrieve_via_httpx_refuses_non_public(uri):
+    with mock.patch.object(json_schema.httpx, "get") as get:
+        with pytest.raises(ValueError):
+            json_schema._retrieve_via_httpx(uri)
+        get.assert_not_called()
+
+
+def test_retrieve_via_httpx_allows_public_host():
+    response = mock.Mock()
+    response.json.return_value = {"type": "object"}
+
+    def gai(host, port, *args, **kwargs):
+        return [(2, 1, 6, "", ("93.184.216.34", port))]
+
+    with mock.patch.object(json_schema.socket, "getaddrinfo", side_effect=gai), \
+            mock.patch.object(json_schema.httpx, "get", return_value=response) as get:
+        json_schema._retrieve_via_httpx("http://schemas.example.com/draft-07/schema")
+        get.assert_called_once()