fix(mcp): fail closed when the tool scope gate cannot reach its auth config

redisvl/mcp/auth.py

@@ -164,13 +164,26 @@ def ensure_tool_scope(server: Any, required_scope: str | None) -> None:
     No-ops when auth is disabled or no scope is configured. Otherwise reads the
     current access token and checks the configured authorization claim, raising
     a ``forbidden`` MCP error when the scope is absent.
+
+    Prefer :func:`ensure_read_scope` / :func:`ensure_write_scope` at a call
+    site; they resolve the scope name from the same server this reads.
     """
+    if not getattr(server, "_auth_enabled", False):
+        return
+
     auth_config = getattr(server, "auth_config", None)
-    if (
-        not getattr(server, "_auth_enabled", False)
-        or auth_config is None
-        or required_scope is None
-    ):
+    if auth_config is None:
+        # Auth is wired, so its config has to be reachable. Returning here would
+        # silently stop gating every tool the moment the attribute is renamed --
+        # a fail-open that no test would catch -- so fail closed instead.
+        raise RedisVLMCPError(
+            "MCP auth is enabled but the server's auth configuration is "
+            "unreachable; refusing to run an ungated tool",
+            code=MCPErrorCode.INTERNAL_ERROR,
+            retryable=False,
+        )
+
+    if required_scope is None:
         return
 
     from fastmcp.server.dependencies import get_access_token
@@ -190,3 +203,26 @@ def ensure_tool_scope(server: Any, required_scope: str | None) -> None:
             code=MCPErrorCode.FORBIDDEN,
             retryable=False,
         )
+
+
+def _configured_scope(server: Any, attribute: str) -> str | None:
+    """Read one configured scope name off the server's auth config.
+
+    Deliberately unguarded on the attribute itself: a renamed field on
+    ``MCPAuthConfig`` raises here rather than resolving to ``None`` and quietly
+    turning the scope gate into a no-op.
+    """
+    auth_config = getattr(server, "auth_config", None)
+    if auth_config is None:
+        return None
+    return getattr(auth_config, attribute)
+
+
+def ensure_read_scope(server: Any) -> None:
+    """Enforce the configured read scope for the current request."""
+    ensure_tool_scope(server, _configured_scope(server, "read_scope"))
+
+
+def ensure_write_scope(server: Any) -> None:
+    """Enforce the configured write scope for the current request."""
+    ensure_tool_scope(server, _configured_scope(server, "write_scope"))

redisvl/mcp/tools/list_indexes.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING, Any
 
-from redisvl.mcp.auth import ensure_tool_scope
+from redisvl.mcp.auth import ensure_read_scope
 from redisvl.mcp.runtime import BindingRuntime
 
 if TYPE_CHECKING:
@@ -103,9 +103,7 @@ def register_list_indexes_tool(server: "RedisVLMCPServer") -> None:
 
     async def list_indexes_tool():
         """FastMCP wrapper for the `list-indexes` tool."""
-        auth_config = getattr(server, "auth_config", None)
-        read_scope = auth_config.read_scope if auth_config is not None else None
-        ensure_tool_scope(server, read_scope)
+        ensure_read_scope(server)
         return list_indexes(server)
 
     server.tool(name="list-indexes", description=DEFAULT_LIST_INDEXES_DESCRIPTION)(

redisvl/mcp/tools/profiles.py

@@ -16,7 +16,7 @@
 
 from pydantic import Field
 
-from redisvl.mcp.auth import ensure_tool_scope
+from redisvl.mcp.auth import ensure_read_scope
 from redisvl.mcp.config import MCPCustomToolConfig
 from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError
 from redisvl.mcp.filters import parse_filter
@@ -240,9 +240,7 @@ def register_profile_tool(
     signature, annotations = _build_signature(profile)
 
     async def profile_tool(**kwargs: Any) -> dict[str, Any]:
-        auth_config = getattr(server, "auth_config", None)
-        read_scope = auth_config.read_scope if auth_config is not None else None
-        ensure_tool_scope(server, read_scope)
+        ensure_read_scope(server)
 
         # A hidden argument is already absent from the advertised schema, so a
         # compliant client cannot send one. Ignoring it here too means the lock

redisvl/mcp/tools/search.py

@@ -2,7 +2,7 @@
 import inspect
 from typing import Any
 
-from redisvl.mcp.auth import ensure_tool_scope
+from redisvl.mcp.auth import ensure_read_scope
 from redisvl.mcp.config import reserved_score_metadata_field_names
 from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError, map_exception
 from redisvl.mcp.filters import parse_filter
@@ -707,9 +707,7 @@ async def search_records_tool(
         return_fields: list[str] | None = None,
     ):
         """FastMCP wrapper for the `search-records` tool."""
-        auth_config = getattr(server, "auth_config", None)
-        read_scope = auth_config.read_scope if auth_config is not None else None
-        ensure_tool_scope(server, read_scope)
+        ensure_read_scope(server)
         return await search_records(
             server,
             query=query,

redisvl/mcp/tools/upsert.py

@@ -3,7 +3,7 @@
 from copy import deepcopy
 from typing import Any
 
-from redisvl.mcp.auth import ensure_tool_scope
+from redisvl.mcp.auth import ensure_write_scope
 from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError, map_exception
 from redisvl.redis.utils import array_to_buffer
 from redisvl.schema.schema import StorageType
@@ -386,9 +386,7 @@ async def upsert_records_tool(
         skip_embedding_if_present: bool | None = None,
     ):
         """FastMCP wrapper for the `upsert-records` tool."""
-        auth_config = getattr(server, "auth_config", None)
-        write_scope = auth_config.write_scope if auth_config is not None else None
-        ensure_tool_scope(server, write_scope)
+        ensure_write_scope(server)
         return await upsert_records(
             server,
             records=records,

tests/unit/test_mcp/test_auth_scope.py

@@ -6,7 +6,13 @@
 # imports fastmcp; skip the module when the optional extra is absent.
 pytest.importorskip("fastmcp", reason="fastmcp not installed (install redisvl[mcp])")
 
-from redisvl.mcp.auth import authorization_values, ensure_tool_scope, token_has_scope
+from redisvl.mcp.auth import (
+    authorization_values,
+    ensure_read_scope,
+    ensure_tool_scope,
+    ensure_write_scope,
+    token_has_scope,
+)
 from redisvl.mcp.config import MCPAuthConfig
 from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError
 
@@ -18,14 +24,24 @@ def __init__(self, scopes=None, claims=None):
 
 
 class _Cfg:
-    def __init__(self, authorization_claim="scp"):
+    def __init__(self, authorization_claim="scp", read_scope=None, write_scope=None):
         self.authorization_claim = authorization_claim
+        self.read_scope = read_scope
+        self.write_scope = write_scope
 
 
 class _Server:
-    def __init__(self, enabled=True, authorization_claim="scp"):
+    def __init__(
+        self,
+        enabled=True,
+        authorization_claim="scp",
+        read_scope=None,
+        write_scope=None,
+    ):
         self._auth_enabled = enabled
-        self.auth_config = _Cfg(authorization_claim) if enabled else None
+        self.auth_config = (
+            _Cfg(authorization_claim, read_scope, write_scope) if enabled else None
+        )
 
 
 # --- claim selection -------------------------------------------------------
@@ -118,3 +134,59 @@ def test_ensure_tool_scope_noop_when_no_token(monkeypatch):
         "fastmcp.server.dependencies.get_access_token", lambda: None, raising=False
     )
     ensure_tool_scope(_Server(), "kb.search.read")
+
+
+# --- ensure_read_scope / ensure_write_scope --------------------------------
+
+
+def test_scope_helpers_resolve_their_configured_scope(monkeypatch):
+    # The helper reads the scope name off the server, so a wrapper never has to
+    # know which auth_config field its side of the gate uses.
+    tok = _AccessToken(claims={"roles": ["kb.search.read"]})
+    monkeypatch.setattr(
+        "fastmcp.server.dependencies.get_access_token", lambda: tok, raising=False
+    )
+    server = _Server(
+        authorization_claim="roles",
+        read_scope="kb.search.read",
+        write_scope="kb.search.write",
+    )
+
+    ensure_read_scope(server)
+
+    with pytest.raises(RedisVLMCPError) as exc:
+        ensure_write_scope(server)
+    assert exc.value.code == MCPErrorCode.FORBIDDEN
+
+
+def test_scope_helpers_noop_when_auth_disabled():
+    server = _Server(enabled=False)
+    ensure_read_scope(server)
+    ensure_write_scope(server)
+
+
+def test_scope_gate_fails_closed_when_auth_config_is_unreachable(monkeypatch):
+    # Renaming the server's auth_config attribute used to make every call site
+    # resolve a None scope and return early, silently ungating every tool.
+    tok = _AccessToken(claims={"roles": []})
+    monkeypatch.setattr(
+        "fastmcp.server.dependencies.get_access_token", lambda: tok, raising=False
+    )
+    server = _Server(authorization_claim="roles", read_scope="kb.search.read")
+    server._renamed_auth_config = server.auth_config
+    del server.auth_config
+
+    with pytest.raises(RedisVLMCPError) as exc:
+        ensure_read_scope(server)
+    assert exc.value.code == MCPErrorCode.INTERNAL_ERROR
+    assert exc.value.retryable is False
+
+
+def test_scope_helper_raises_when_the_config_field_is_renamed():
+    # An unguarded getattr, so a renamed MCPAuthConfig field is a loud failure
+    # rather than a None scope that turns the gate into a no-op.
+    server = _Server(read_scope="kb.search.read")
+    del server.auth_config.read_scope
+
+    with pytest.raises(AttributeError):
+        ensure_read_scope(server)