diff --git a/src/confluent_kafka/schema_registry/common/json_schema.py b/src/confluent_kafka/schema_registry/common/json_schema.py
index 234f5aab1..29c480628 100644
--- a/src/confluent_kafka/schema_registry/common/json_schema.py
+++ b/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 Any, List, Optional, Set, Union
+from urllib.parse import urlsplit
 
 import httpx
 import referencing
@@ -107,8 +110,32 @@ 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 not ip.is_global
+
+
+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)) from 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)
 
 
diff --git a/tests/schema_registry/test_json_schema_retrieve.py b/tests/schema_registry/test_json_schema_retrieve.py
new file mode 100644
index 000000000..89098c3c1
--- /dev/null
+++ b/tests/schema_registry/test_json_schema_retrieve.py
@@ -0,0 +1,59 @@
+#!/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",
+        "http://[::ffff:127.0.0.1]/schema",
+        "http://100.64.0.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()