fix: restrict $ref resolution to local refs only (SSRF/LFI) (#3502)

Co-authored-by: Claude <noreply@anthropic.com>

src/fastmcp/utilities/json_schema.py

@@ -53,6 +53,26 @@ def _defs_have_cycles(defs: dict[str, Any]) -> bool:
     return any(state[name] == UNVISITED and _has_cycle(name) for name in defs)
 
 
+def _strip_remote_refs(obj: Any) -> Any:
+    """Return a deep copy of *obj* with non-local ``$ref`` values removed.
+
+    Local refs (starting with ``#``) are kept intact.  Remote refs
+    (``http://``, ``https://``, ``file://``, or any other URI scheme) are
+    stripped so that ``jsonref.replace_refs`` never attempts to fetch an
+    external resource.  This prevents SSRF / LFI when proxying schemas
+    from untrusted servers.
+    """
+    if isinstance(obj, dict):
+        ref = obj.get("$ref")
+        if isinstance(ref, str) and not ref.startswith("#"):
+            # Drop the remote $ref key; keep all other keys.
+            return {k: _strip_remote_refs(v) for k, v in obj.items() if k != "$ref"}
+        return {k: _strip_remote_refs(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_strip_remote_refs(item) for item in obj]
+    return obj
+
+
 def dereference_refs(schema: dict[str, Any]) -> dict[str, Any]:
     """Resolve all $ref references in a JSON schema by inlining definitions.
 
@@ -67,6 +87,11 @@ def dereference_refs(schema: dict[str, Any]) -> dict[str, Any]:
     this function falls back to resolving only the root-level $ref while preserving
     $defs for nested references.
 
+    Only local ``$ref`` values (those starting with ``#``) are resolved.
+    Remote URIs (``http://``, ``file://``, etc.) are stripped before
+    resolution to prevent SSRF / local-file-inclusion attacks when proxying
+    schemas from untrusted servers.
+
     Args:
         schema: JSON schema dict that may contain $ref references
 
@@ -82,6 +107,9 @@ def dereference_refs(schema: dict[str, Any]) -> dict[str, Any]:
         >>> resolved = dereference_refs(schema)
         >>> # Result: {"properties": {"cat": {"enum": ["a", "b"], "type": "string", "default": "a"}}}
     """
+    # Strip any remote $ref values before processing to prevent SSRF / LFI.
+    schema = _strip_remote_refs(schema)
+
     # Circular $defs can't be fully inlined — jsonref.replace_refs produces
     # Python dicts with object-identity cycles that Pydantic's model_dump
     # rejects with "Circular reference detected (id repeated)".

tests/utilities/test_json_schema.py

@@ -1,5 +1,10 @@
+from unittest.mock import patch
+
+from jsonref import replace_refs
+
 from fastmcp.utilities.json_schema import (
     _prune_param,
+    _strip_remote_refs,
     compress_schema,
     dereference_refs,
     resolve_root_ref,
@@ -628,3 +633,126 @@ class TestResolveRootRef:
 
         # Should return original schema unchanged
         assert result is schema
+
+
+class TestStripRemoteRefs:
+    """Tests for _strip_remote_refs which prevents SSRF/LFI via $ref."""
+
+    def test_preserves_local_ref(self):
+        schema = {"$ref": "#/$defs/Foo"}
+        assert _strip_remote_refs(schema) == {"$ref": "#/$defs/Foo"}
+
+    def test_strips_http_ref(self):
+        schema = {"$ref": "http://evil.com/schema.json"}
+        assert _strip_remote_refs(schema) == {}
+
+    def test_strips_https_ref(self):
+        schema = {"$ref": "https://evil.com/schema.json"}
+        assert _strip_remote_refs(schema) == {}
+
+    def test_strips_file_ref(self):
+        schema = {"$ref": "file:///etc/passwd"}
+        assert _strip_remote_refs(schema) == {}
+
+    def test_preserves_siblings_when_stripping(self):
+        schema = {
+            "$ref": "http://evil.com/schema.json",
+            "description": "keep me",
+            "default": 42,
+        }
+        result = _strip_remote_refs(schema)
+        assert result == {"description": "keep me", "default": 42}
+
+    def test_strips_nested_remote_refs(self):
+        schema = {
+            "properties": {
+                "safe": {"$ref": "#/$defs/Safe"},
+                "evil": {"$ref": "http://169.254.169.254/latest/meta-data/"},
+            }
+        }
+        result = _strip_remote_refs(schema)
+        assert result["properties"]["safe"] == {"$ref": "#/$defs/Safe"}
+        assert "$ref" not in result["properties"]["evil"]
+
+    def test_strips_remote_refs_in_lists(self):
+        schema = {
+            "anyOf": [
+                {"$ref": "#/$defs/Good"},
+                {"$ref": "file:///etc/credentials.json"},
+            ]
+        }
+        result = _strip_remote_refs(schema)
+        assert result["anyOf"][0] == {"$ref": "#/$defs/Good"}
+        assert "$ref" not in result["anyOf"][1]
+
+    def test_deep_nesting(self):
+        schema = {
+            "properties": {
+                "a": {
+                    "type": "object",
+                    "properties": {"b": {"$ref": "https://internal-service/secret"}},
+                }
+            }
+        }
+        result = _strip_remote_refs(schema)
+        assert "$ref" not in result["properties"]["a"]["properties"]["b"]
+
+
+class TestDereferenceRefsRemoteRefSafety:
+    """Verify dereference_refs never fetches remote URIs."""
+
+    def test_http_ref_not_fetched(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "name": {"$ref": "http://evil.com/schema.json"},
+            },
+        }
+        with patch(
+            "fastmcp.utilities.json_schema.replace_refs", wraps=replace_refs
+        ) as mock:
+            result = dereference_refs(schema)
+            # The remote $ref should have been stripped before replace_refs
+            if mock.called:
+                call_schema = mock.call_args[0][0]
+                assert "$ref" not in call_schema.get("properties", {}).get("name", {})
+        # Result should not contain the remote $ref
+        assert "$ref" not in result.get("properties", {}).get("name", {})
+
+    def test_file_ref_not_fetched(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "secret": {"$ref": "file:///etc/passwd"},
+            },
+        }
+        result = dereference_refs(schema)
+        assert "$ref" not in result.get("properties", {}).get("secret", {})
+
+    def test_cloud_metadata_ref_not_fetched(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "creds": {
+                    "$ref": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
+                },
+            },
+        }
+        result = dereference_refs(schema)
+        assert "$ref" not in result.get("properties", {}).get("creds", {})
+
+    def test_local_refs_still_resolved(self):
+        schema = {
+            "$defs": {"Status": {"type": "string", "enum": ["a", "b"]}},
+            "type": "object",
+            "properties": {
+                "status": {"$ref": "#/$defs/Status"},
+                "evil": {"$ref": "https://evil.com/inject"},
+            },
+        }
+        result = dereference_refs(schema)
+        # Local ref should be resolved
+        assert result["properties"]["status"] == {"type": "string", "enum": ["a", "b"]}
+        # Remote ref should be stripped
+        assert "$ref" not in result["properties"]["evil"]
+        assert "$defs" not in result