From b94af0801c77e2c14e39996a41c9bc541a44e1d8 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Fri, 11 Sep 2026 14:33:28 +0200 Subject: [PATCH 1/5] feat(mcp): add the security core for auth-claim tenant injection Phase 2 of custom MCP tools scopes every query a profile runs to the tenant carried in the caller's verified token. This is the first of two slices: the three functions that carry the security property, with no configuration model and no server wiring, so all of it is unit-testable. `resolve_injected_claim` reads one claim and accepts only a single non-empty unpadded string. It sits directly below `authorization_values`, which reads `access_token.claims` under the opposite rules -- that reader space-splits a string and coerces list members, because an absent scope only ever denies, whereas a widened value here grants. Keeping the two adjacent, each commented with why it differs from its neighbour, is what stops them being unified. The type check is the load-bearing one. `_formatted_tag_value` escapes each element before joining them with `|`, so a list claim renders a genuine cross-tenant union that no character scan of the output can tell from a legitimate one -- the `|` is structure rather than content. The scalar `|` check stays as a backstop for the character class changing again. `build_injected_filter` builds one tag equality per entry, ANDs them, and refuses the whole request when any single claim is unusable. Its match-all check runs on each clause alone, before combining: an intersection elides a `*` operand, so a check on the combined expression would pass while the tenant clause had silently vanished. `validate_inject_against_schema` fails startup when an injected field is absent, is not a tag, or is NOINDEX. A NOINDEX field is the worst of the three because it returns nothing and so looks like correct scoping. `_is_match_all_filter` moves from `redisvl/index/index.py` to `redisvl/query/filter.py` as public `is_match_all_filter`, with its one call site updated. It exists rather than a bare `str(expr) != "*"` because an un-initialized `FilterExpression` raises on render; that case now has a test. Injected fields are tag fields only. Text equality is an exact-phrase match and text tokenizes on punctuation, so `@t:("acme-corp")` also matches `acme-corp-eu` -- widening, and `no_stem` does not fix it. Every guard was mutation-checked by reverting it alone. One did not fail any test: the match-all check is unreachable behind the claim reader's own empty check, so it is now pinned by a test that substitutes that reader, which is exactly the future it insures against. --- redisvl/index/index.py | 23 +- redisvl/mcp/auth.py | 237 ++++++++++++++- redisvl/query/filter.py | 25 ++ tests/unit/test_filter.py | 21 ++ .../test_mcp/test_auth_claim_injection.py | 284 ++++++++++++++++++ 5 files changed, 568 insertions(+), 22 deletions(-) create mode 100644 tests/unit/test_mcp/test_auth_claim_injection.py diff --git a/redisvl/index/index.py b/redisvl/index/index.py index 573a94e2..e97a0ca1 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -72,7 +72,7 @@ TextQuery, ) from redisvl.query.aggregate import AggregateHybridQuery -from redisvl.query.filter import FilterExpression +from redisvl.query.filter import FilterExpression, is_match_all_filter from redisvl.redis.connection import ( RedisConnectionFactory, _split_from_existing_kwargs, @@ -233,30 +233,11 @@ class BulkResult: dry_run: bool = False -def _is_match_all_filter(filter_expression: str | FilterExpression | None) -> bool: - """Return True if the filter would match every document in the index. - - Guards the bulk ``*_by_filter`` methods against an accidental full-index - wipe/update. ``None`` is treated as match-all because it defaults to - ``FilterExpression("*")`` downstream; a default/empty ``FilterExpression`` - (whose ``str()`` raises) is likewise treated as match-all rather than - surfacing an opaque error. - """ - if filter_expression is None: - return True - try: - rendered = str(filter_expression).strip() - except ValueError: - # Improperly initialized FilterExpression() - treat as the match-all sentinel - return True - return rendered in ("", "*") - - def _require_specific_filter( filter_expression: str | FilterExpression | None, allow_all: bool ) -> None: """Raise unless the filter is specific or the caller opted into match-all.""" - if not allow_all and _is_match_all_filter(filter_expression): + if not allow_all and is_match_all_filter(filter_expression): raise ValueError( "Refusing to run a bulk operation that matches all documents. " "Pass a specific filter_expression, set allow_all=True to override, " diff --git a/redisvl/mcp/auth.py b/redisvl/mcp/auth.py index 7d6fab3d..b08b98bb 100644 --- a/redisvl/mcp/auth.py +++ b/redisvl/mcp/auth.py @@ -6,16 +6,29 @@ Auth applies only to HTTP transports; ``stdio`` is never authenticated. FastMCP imports are deferred so this module stays importable without the ``mcp`` extra. + +Reading values *out* of a verified token lives here too, next to the wiring that +decides which tokens are verified at all. Two readers coexist deliberately: +``authorization_values`` normalizes widely because an absent scope only denies, +and ``resolve_injected_claim`` refuses everything but a single non-empty string +because a widened value grants. Split across modules, someone unifies them and +reintroduces the cross-tenant union. """ +import logging +from collections.abc import Sequence from pathlib import Path -from typing import Any +from typing import Any, Protocol import yaml from redisvl.mcp.config import MCPAuthConfig, _substitute_env from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError from redisvl.mcp.settings import MCPSettings +from redisvl.query.filter import FilterExpression, Tag, is_match_all_filter +from redisvl.schema import IndexSchema + +logger = logging.getLogger(__name__) def peek_yaml_auth(config_path: str | None) -> dict[str, Any] | None: @@ -226,3 +239,225 @@ def ensure_read_scope(server: Any) -> None: 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")) + + +class InjectSpec(Protocol): + """One locked injection entry: an index field and the claim that fills it. + + A protocol rather than an import of the configuration model, so the + security core has no dependency on the shape of the YAML that reaches it. + A v1.1 code tool assembles its own specs and gets the same guarantees. + """ + + @property + def field(self) -> str: ... + + @property + def claim(self) -> str: ... + + +def _injection_refused(claim: str, tool_name: str, reason: str) -> RedisVLMCPError: + """Build the single refusal raised by every injection failure. + + The claim name is deliberately in the client-facing message. A JWT is + signed, not encrypted, so a client holding a valid token can already read + its own claim names, and an unauthenticated client never reaches this code + because the HTTP layer rejects it first. Naming the claim and the tool is + what turns a misspelled claim name from an opaque permanent failure into a + one-line diagnosis. + """ + message = ( + f"Tool '{tool_name}' requires the '{claim}' claim to scope every query, " + f"and {reason}; refusing to run an unscoped query" + ) + logger.warning("%s", message) + return RedisVLMCPError( + message, + code=MCPErrorCode.FORBIDDEN, + retryable=False, + ) + + +def resolve_injected_claim(claim: str, *, tool_name: str) -> str: + """Resolve one token claim into a value safe to inject as a tag equality. + + Deliberately the mirror image of :func:`authorization_values`, which is + directly above for that reason. That reader normalizes widely -- it + space-splits a string and coerces list members -- because an absent scope + only ever denies. Here a widened value *grants*, so every shape but a + single non-empty string is refused. + + Unlike :func:`ensure_tool_scope`, a tokenless request raises rather than + returning. The scope gate reads ``None`` as "stdio, so no gate applies"; + reached here, the same exit would attach no tenant clause and run the query + against every tenant. + """ + from fastmcp.server.dependencies import get_access_token + + access_token = get_access_token() + if access_token is None: + raise _injection_refused( + claim, tool_name, "this request carries no verified token" + ) + + claims = getattr(access_token, "claims", None) or {} + if claim not in claims: + raise _injection_refused(claim, tool_name, "the token does not carry it") + + value = claims[claim] + if not isinstance(value, str): + # The load-bearing check. `_formatted_tag_value` escapes each element + # *before* joining them with `|`, so a list claim renders as a genuine + # cross-tenant union -- `@tenant_id:{a|b}` -- that no character scan of + # the output can distinguish from a legitimate one, because the `|` is + # structure rather than content. Type is the only place to catch it. + raise _injection_refused( + claim, + tool_name, + f"it holds a {type(value).__name__} rather than a single string value", + ) + + if not value.strip(): + raise _injection_refused(claim, tool_name, "it is empty") + + if value != value.strip(): + # Not normalized to the stripped value: two tenants named `acme` and + # `acme ` would then collapse onto one. Surrounding whitespace cannot + # be meaningful in a tag equality, so refuse it instead of guessing. + raise _injection_refused(claim, tool_name, "it is padded with whitespace") + + if "|" in value: + # Defence in depth. `Tag` escapes `|` inside a single value since + # 0.27.1, so this is unreachable through the `==` path today; it is the + # backstop for that character class changing again. `test_injected_pipe + # _cannot_union_across_tenants` pins the rendering property itself. + raise _injection_refused(claim, tool_name, "it contains the union operator '|'") + + return value + + +def _injected_tag_field(schema: IndexSchema, field_name: str) -> Any: + """Return the schema field an injection entry names, or ``None``. + + ``None`` means "not usable for injection" for any reason -- absent, wrong + type, or unindexed. Callers decide whether that is a startup failure or a + request-time refusal. + """ + field = schema.fields.get(field_name) + if field is None or field.type != "tag": + return None + if getattr(field.attrs, "no_index", False): + return None + return field + + +def build_injected_filter( + inject_specs: Sequence[InjectSpec], + schema: IndexSchema, + *, + tool_name: str, +) -> FilterExpression: + """Build the tenant-scoping expression for the current request. + + Entries AND together, and any one unusable claim refuses the whole request + rather than narrowing by the entries that did resolve. + """ + if not inject_specs: + # Callers guard on truthiness before reaching here; an empty list would + # otherwise have to return a match-all, which is the one value this + # function exists to make unreachable. + raise RedisVLMCPError( + f"Tool '{tool_name}' asked for claim injection with no entries " + "configured; refusing to run an unscoped query", + code=MCPErrorCode.INTERNAL_ERROR, + retryable=False, + ) + + combined: FilterExpression | None = None + clauses: list[str] = [] + + for spec in inject_specs: + value = resolve_injected_claim(spec.claim, tool_name=tool_name) + + if _injected_tag_field(schema, spec.field) is None: + # Startup validation already rejected this, so reaching it means the + # bound schema changed under a registered tool. Re-checking costs a + # dict lookup and keeps the guarantee tied to the schema actually in + # force rather than to the one present at registration. + raise _injection_refused( + spec.claim, + tool_name, + f"its target field '{spec.field}' is no longer an indexed tag " + "field on the bound index", + ) + + clause = Tag(spec.field) == value + if is_match_all_filter(clause): + # The check has to run here, on the clause alone. An intersection + # elides a `*` operand, so `locked & match_all` renders as `locked` + # and a check on the combined expression would pass while the + # tenant clause had silently vanished. + raise _injection_refused( + spec.claim, + tool_name, + "it renders as a filter that matches every document", + ) + + clauses.append(str(clause)) + combined = clause if combined is None else combined & clause + + assert combined is not None # for mypy; the empty case raised above + rendered = str(combined) + missing = [clause for clause in clauses if clause not in rendered] + if missing: + # Operands render verbatim into `(left right)`, so containment is exact + # rather than approximate. If a future change to `format_expression` + # ever drops one, fail the request instead of serving a query that is + # scoped to fewer tenants' worth of clauses than were configured. + raise RedisVLMCPError( + f"Tool '{tool_name}' built an injected filter that lost the clauses " + f"{missing}; refusing to run a query that may not be scoped", + code=MCPErrorCode.INTERNAL_ERROR, + retryable=False, + ) + + return combined + + +def validate_inject_against_schema( + inject_specs: Sequence[InjectSpec], + schema: IndexSchema, + *, + profile_name: str, +) -> None: + """Fail startup when an injection entry names a field it cannot scope by. + + Configuration validation cannot do this: the schema is only known once the + binding has been inspected at startup. Without it, a profile would register + cleanly and then refuse every request -- or, worse for a `no_index` field, + return nothing and look like correct scoping. + """ + field_names = ", ".join(sorted(schema.field_names)) + + for spec in inject_specs: + field = schema.fields.get(spec.field) + if field is None: + raise ValueError( + f"custom_tools '{profile_name}' lock.inject references unknown " + f"field '{spec.field}' on index '{schema.index.name}'; " + f"available: {field_names}" + ) + if field.type != "tag": + raise ValueError( + f"custom_tools '{profile_name}' lock.inject field '{spec.field}' " + f"is a {field.type} field; injection requires a tag field, " + "because text equality is an exact-phrase match that tokenizes " + "on punctuation and so matches neighbouring values too" + ) + if getattr(field.attrs, "no_index", False): + raise ValueError( + f"custom_tools '{profile_name}' lock.inject field '{spec.field}' " + "is declared NOINDEX, so a filter on it matches nothing; every " + "call would return an empty result set that looks like correct " + "scoping" + ) diff --git a/redisvl/query/filter.py b/redisvl/query/filter.py index 9d62c606..9b7b5901 100644 --- a/redisvl/query/filter.py +++ b/redisvl/query/filter.py @@ -920,6 +920,31 @@ def __str__(self) -> str: return self._filter +def is_match_all_filter(filter_expression: str | FilterExpression | None) -> bool: + """Return True if the filter would match every document in the index. + + ``None`` counts as match-all because it defaults to ``FilterExpression("*")`` + downstream, and a default/empty ``FilterExpression`` -- whose ``str()`` + raises -- is treated the same way rather than surfacing an opaque error. + That raising case is the reason this exists rather than a bare + ``str(expr) != "*"`` at each call site. + + Two callers depend on it. The bulk ``*_by_filter`` methods use it to refuse + an accidental full-index wipe, and the MCP claim-injection path uses it to + reject an injected clause that renders as the wildcard -- which an + intersection would otherwise elide, deleting the clause instead of + narrowing by it. + """ + if filter_expression is None: + return True + try: + rendered = str(filter_expression).strip() + except ValueError: + # Improperly initialized FilterExpression() - treat as the match-all sentinel + return True + return rendered in ("", "*") + + def render_filter(filter_expression: str | FilterExpression | None) -> str | None: """Render a filter expression, or None when it selects every document. diff --git a/tests/unit/test_filter.py b/tests/unit/test_filter.py index 1c36f565..8377e9f4 100644 --- a/tests/unit/test_filter.py +++ b/tests/unit/test_filter.py @@ -20,6 +20,7 @@ Text, Timestamp, intersect_with_filter, + is_match_all_filter, render_filter, ) @@ -1125,3 +1126,23 @@ def test_render_filter_with_filter_expression_inputs(): ) def test_intersect_with_filter(filter_expression, expected): assert intersect_with_filter("@text:(fox)", filter_expression) == expected + + +@pytest.mark.parametrize( + "filter_expression, expected", + [ + pytest.param(None, True, id="none"), + pytest.param("*", True, id="wildcard-string"), + pytest.param(" * ", True, id="padded-wildcard"), + pytest.param("", True, id="empty-string"), + pytest.param(FilterExpression("*"), True, id="wildcard-expression"), + # The reason this helper exists rather than a bare `str(expr) != "*"` at + # each call site: an un-initialized FilterExpression raises on render. + pytest.param(FilterExpression(), True, id="uninitialized-raises"), + pytest.param(Tag("t") == "", True, id="empty-tag-renders-wildcard"), + pytest.param(Tag("t") == "acme", False, id="tag-equality"), + pytest.param("@t:{acme}", False, id="rendered-string"), + ], +) +def test_is_match_all_filter(filter_expression, expected): + assert is_match_all_filter(filter_expression) is expected diff --git a/tests/unit/test_mcp/test_auth_claim_injection.py b/tests/unit/test_mcp/test_auth_claim_injection.py new file mode 100644 index 00000000..bda3dac4 --- /dev/null +++ b/tests/unit/test_mcp/test_auth_claim_injection.py @@ -0,0 +1,284 @@ +"""Unit tests for auth-claim tenant injection. + +The property under test is the one the feature exists to provide: *a client +presenting a validly-signed token cannot make the model widen or escape the +tenant scope carried in that token.* Every case below is a route someone could +take to an unscoped or a cross-tenant query. +""" + +from dataclasses import dataclass + +import pytest + +# These tests monkeypatch fastmcp.server.dependencies.get_access_token, which +# 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 ( + build_injected_filter, + resolve_injected_claim, + validate_inject_against_schema, +) +from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError +from redisvl.schema import IndexSchema + + +@dataclass(frozen=True) +class _Spec: + """Stands in for the configuration model Stack 02 adds. + + The security core takes a spec list rather than a config object, so this + double is the whole contract: a field name and a claim name. + """ + + field: str + claim: str + + +class _AccessToken: + def __init__(self, claims=None): + self.claims = claims or {} + + +def _schema() -> IndexSchema: + return IndexSchema.from_dict( + { + "index": {"name": "kb", "prefix": "kb", "storage_type": "hash"}, + "fields": [ + {"name": "content", "type": "text"}, + {"name": "org_id", "type": "tag", "attrs": {"case_sensitive": True}}, + {"name": "region", "type": "tag", "attrs": {"case_sensitive": True}}, + {"name": "rating", "type": "numeric"}, + {"name": "shadow", "type": "tag", "attrs": {"no_index": True}}, + { + "name": "embedding", + "type": "vector", + "attrs": { + "algorithm": "flat", + "dims": 3, + "distance_metric": "cosine", + "datatype": "float32", + }, + }, + ], + } + ) + + +def _token(monkeypatch, claims): + """Install a request-scoped token, or none at all when claims is None.""" + token = None if claims is None else _AccessToken(claims) + monkeypatch.setattr( + "fastmcp.server.dependencies.get_access_token", lambda: token, raising=False + ) + return token + + +# --- resolve_injected_claim ------------------------------------------------ + + +def test_a_verified_claim_resolves_to_its_value(monkeypatch): + _token(monkeypatch, {"org_id": "acme"}) + assert resolve_injected_claim("org_id", tool_name="search_kb") == "acme" + + +def test_a_tokenless_request_is_refused_rather_than_passed_through(monkeypatch): + # The scope gate returns early here, reading a missing token as "stdio, so + # no gate applies". Injection must invert that: the same exit would attach + # no tenant clause and query every tenant. + _token(monkeypatch, None) + with pytest.raises(RedisVLMCPError) as exc: + resolve_injected_claim("org_id", tool_name="search_kb") + assert exc.value.code == MCPErrorCode.FORBIDDEN + assert exc.value.retryable is False + + +@pytest.mark.parametrize( + "claims", + [ + pytest.param({}, id="no-claims"), + pytest.param({"tenant": "acme"}, id="claim-misspelled"), + pytest.param({"org_id": None}, id="null"), + pytest.param({"org_id": ""}, id="empty"), + pytest.param({"org_id": " "}, id="whitespace-only"), + pytest.param({"org_id": " acme"}, id="left-padded"), + pytest.param({"org_id": "acme "}, id="right-padded"), + pytest.param({"org_id": ["acme", "victim"]}, id="list"), + pytest.param({"org_id": {"id": "acme"}}, id="dict"), + pytest.param({"org_id": True}, id="bool"), + pytest.param({"org_id": 42}, id="int"), + pytest.param({"org_id": "acme|victim"}, id="pipe"), + ], +) +def test_an_unusable_claim_is_refused(monkeypatch, claims): + _token(monkeypatch, claims) + with pytest.raises(RedisVLMCPError) as exc: + resolve_injected_claim("org_id", tool_name="search_kb") + assert exc.value.code == MCPErrorCode.FORBIDDEN + assert exc.value.retryable is False + + +def test_the_refusal_names_the_claim_and_the_tool(monkeypatch): + # A misspelled claim name is otherwise an opaque permanent failure. The + # disclosure is free: a JWT is signed, not encrypted, so a client holding a + # valid token can already read its own claim names. + _token(monkeypatch, {}) + with pytest.raises(RedisVLMCPError) as exc: + resolve_injected_claim("org_id", tool_name="search_kb") + assert "org_id" in str(exc.value) + assert "search_kb" in str(exc.value) + + +# --- build_injected_filter ------------------------------------------------- + + +def test_one_entry_scopes_the_query_to_the_claim(monkeypatch): + _token(monkeypatch, {"org_id": "acme"}) + expression = build_injected_filter( + [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" + ) + assert str(expression) == "@org_id:{acme}" + + +def test_two_entries_and_together(monkeypatch): + _token(monkeypatch, {"org_id": "acme", "region": "eu"}) + expression = build_injected_filter( + [_Spec("org_id", "org_id"), _Spec("region", "region")], + _schema(), + tool_name="search_kb", + ) + rendered = str(expression) + assert "@org_id:{acme}" in rendered + assert "@region:{eu}" in rendered + assert " | " not in rendered + + +def test_one_unusable_claim_refuses_the_whole_request(monkeypatch): + # Not "narrow by the entries that resolved" -- a partially applied scope is + # a wider scope than the one configured. + _token(monkeypatch, {"org_id": "acme"}) + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter( + [_Spec("org_id", "org_id"), _Spec("region", "region")], + _schema(), + tool_name="search_kb", + ) + assert exc.value.code == MCPErrorCode.FORBIDDEN + + +def test_no_entries_refuses_rather_than_matching_everything(monkeypatch): + _token(monkeypatch, {"org_id": "acme"}) + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter([], _schema(), tool_name="search_kb") + assert exc.value.code == MCPErrorCode.INTERNAL_ERROR + + +def test_a_field_that_stopped_being_an_indexed_tag_is_refused(monkeypatch): + # Startup validation rejects these, so reaching here means the bound schema + # changed under a tool that was already registered. + _token(monkeypatch, {"org_id": "acme"}) + for field in ("rating", "shadow", "absent"): + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter( + [_Spec(field, "org_id")], _schema(), tool_name="search_kb" + ) + assert exc.value.code == MCPErrorCode.FORBIDDEN + + +def test_an_injected_clause_cannot_be_elided_by_the_intersection(monkeypatch): + # `Tag(f) == ""` renders as the match-all `*`, and an intersection drops a + # `*` operand -- so `locked & injected` would render as `locked` alone with + # the tenant clause silently gone. The guard therefore runs on the injected + # clause by itself, before any combination. + from redisvl.query.filter import Tag + + injected = Tag("org_id") == "" + locked = Tag("status") == "resolved" + assert str(locked & injected) == str(locked) + + _token(monkeypatch, {"org_id": ""}) + with pytest.raises(RedisVLMCPError): + build_injected_filter( + [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" + ) + + +def test_a_clause_that_renders_match_all_is_refused_even_so(monkeypatch): + """The second, independent layer over the claim reader's own empty check. + + Today the reader rejects an empty claim first, so this guard cannot fire + through any real token -- substituting the reader is the only way to reach + it, which is precisely the future it insures against: a reader that admits + a value ``Tag`` renders as the wildcard. Without it, `locked & injected` + would render as `locked` alone and the request would run unscoped. + """ + monkeypatch.setattr( + "redisvl.mcp.auth.resolve_injected_claim", + lambda claim, *, tool_name: "", + ) + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter( + [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" + ) + assert exc.value.code == MCPErrorCode.FORBIDDEN + assert "matches every document" in str(exc.value) + + +def test_injected_pipe_cannot_union_across_tenants(monkeypatch): + """Canary for the rendering property, not for a character class. + + ``Tag`` escapes ``|`` inside a single value as of 0.27.1. That character + class has already changed once, so this asserts the property the guarantee + actually rests on -- an injected value cannot produce a clause that matches + a tenant other than the one named -- rather than asserting which characters + are in which set. + """ + from redisvl.query.filter import Tag + + # Property one: a scalar value carrying `|` is escaped, so it is content + # rather than structure. + assert str(Tag("org_id") == "acme|victim") == "@org_id:{acme\\|victim}" + + # Property two: a list value is *not* escaped into one value -- it renders + # a genuine union. This is why the claim reader rejects on type, and why a + # character scan of the rendered output would not close it. + assert str(Tag("org_id") == ["acme", "victim"]) == "@org_id:{acme|victim}" + + # So neither shape can reach a query through injection. + for value in ("acme|victim", ["acme", "victim"]): + _token(monkeypatch, {"org_id": value}) + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter( + [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" + ) + assert exc.value.code == MCPErrorCode.FORBIDDEN + + +# --- validate_inject_against_schema ---------------------------------------- + + +def test_a_valid_injected_field_passes_validation(): + validate_inject_against_schema( + [_Spec("org_id", "org_id"), _Spec("region", "region")], + _schema(), + profile_name="search_kb", + ) + + +@pytest.mark.parametrize( + "field, expected", + [ + pytest.param("absent", "unknown field", id="absent"), + pytest.param("content", "requires a tag field", id="text"), + pytest.param("rating", "requires a tag field", id="numeric"), + pytest.param("embedding", "requires a tag field", id="vector"), + pytest.param("shadow", "NOINDEX", id="no-index"), + ], +) +def test_an_unusable_injected_field_fails_startup(field, expected): + with pytest.raises(ValueError) as exc: + validate_inject_against_schema( + [_Spec(field, "org_id")], _schema(), profile_name="search_kb" + ) + assert expected in str(exc.value) + assert "search_kb" in str(exc.value) From bd9f5c53a07712e85bb62db39633058a46457ca8 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 1 Oct 2026 16:42:32 +0200 Subject: [PATCH 2/5] feat(mcp): scope custom tool queries to a tenant from a verified token claim A profile can now take its tenant from the caller's verified token instead of trusting the model to pass a filter. `lock.inject` names an index tag field and the claim that fills it; the value is AND-combined into every query the profile runs, never appears in the advertised input schema, and is left out of the description's field hints. The injected expression pre-folds into the locked side of the merge rather than travelling beside it. `merge_locked_filter` returns the caller's filter untouched when nothing is locked, so only a non-None locked side forces the caller's filter through the escape backstop. A profile with injection and no static lock would otherwise AND the tenant clause and skip that check. Startup refuses an injecting profile when authentication is not enabled, on any transport, and when authentication is configured but the server runs over stdio, which FastMCP never authenticates. `run_async` now records the transport, resolving an omitted one the way FastMCP does, since it is the only point in the process that knows it. Both checks run before any binding connects, because they depend only on the loaded config. The CLI needs no matching check: it reaches the server-side one, and so does any embedder. Config load rejects an empty `inject` list, a field injected twice, a field also constrained by `lock.filter`, `required: false`, and a missing or non-`claim` `from`. Startup rejects an injected field that is absent, not a tag, or NOINDEX, and warns when it is not CASESENSITIVE, since `Acme` and `acme` would otherwise be one tenant. A changed tool surface after an in-process restart stays a warning, except when injection is configured on either side of the change, where it is now fatal. Adding injection to a profile registered without it would otherwise leave that tool serving every tenant while the config says it does not. `_validate_custom_tools` on the server becomes `_validate_custom_tools_against_schema`, ending its name collision with `MCPConfig._validate_custom_tools`. The DSL clause walker moves from the profile module into config, where load-time validation now needs it. The docs gain the feature, its startup checks, and its threat model, and lose two statements that RedisVL never maps claims to query filters. --- docs/concepts/mcp.md | 66 +++- docs/user_guide/how_to_guides/mcp.md | 80 +++++ .../how_to_guides/mcp_authentication.md | 24 +- redisvl/mcp/config.py | 80 +++++ redisvl/mcp/server.py | 78 ++++- redisvl/mcp/tools/profiles.py | 84 +++-- redisvl/mcp/tools/search.py | 22 +- .../test_mcp/test_claim_injection.py | 297 ++++++++++++++++++ tests/unit/test_mcp/test_config.py | 126 ++++++++ tests/unit/test_mcp/test_profiles_unit.py | 117 +++++++ tests/unit/test_mcp/test_server_unit.py | 205 +++++++++++- 11 files changed, 1135 insertions(+), 44 deletions(-) create mode 100644 tests/integration/test_mcp/test_claim_injection.py diff --git a/docs/concepts/mcp.md b/docs/concepts/mcp.md index 8d22d9c3..8557f56c 100644 --- a/docs/concepts/mcp.md +++ b/docs/concepts/mcp.md @@ -99,7 +99,7 @@ Use read-only mode when Redis is serving approved content to assistants and anot ## Authentication and Authorization -The HTTP transports can require a JWT bearer token issued by an existing identity provider. The server validates the token signature, issuer, and audience, and can gate read vs write by scope or role claim. This is coarse, per-tool authorization; it does not map token claims to Redis ACL users or per-tenant filters, which remain a gateway concern. The `stdio` transport is local and is never authenticated. +The HTTP transports can require a JWT bearer token issued by an existing identity provider. The server validates the token signature, issuer, and audience, and can gate read vs write by scope or role claim. A custom tool profile can also scope every query to a tenant carried in the token; see [Tenant Scoping From Token Claims](#tenant-scoping-from-token-claims). What the server does not do is map token claims to Redis ACL users or to separate indexes, which remains a gateway concern. The `stdio` transport is local and is never authenticated. For configuration and the gateway boundary, see {doc}`/user_guide/how_to_guides/mcp_authentication`. @@ -197,6 +197,70 @@ Because adding near-duplicate tools makes tool selection harder rather than easi Misconfiguration fails at startup rather than at the first call. Among the checks: a name colliding with a built-in or using a reserved `redisvl-`/`redisvl_` prefix; a duplicate tool name; a missing or unknown `index`; a `params` key that is not a real argument; `max` on anything but `limit`, or a cap above the binding's `max_limit`; hiding `query`; locking `return_fields` while also exposing them; and a locked filter or projection naming a field the bound index does not have. Unrecognized keys are rejected too, so a typo in `lock` fails loudly instead of silently producing a tool that reads as locked but enforces nothing. +### Tenant Scoping From Token Claims + +When several tenants share one index, separated by a field such as `org_id`, exposing `search-records` makes the tenant boundary depend on the model remembering to pass a filter. One forgotten filter is a cross-tenant read. A profile can remove that knob: `lock.inject` reads the tenant from the caller's verified token and AND-combines it into every query the profile runs. + +```yaml +custom_tools: + - name: search-customer-kb + index: customer_kb + description: Search this customer's knowledge base. + lock: + inject: + - field: org_id # the tenant field in the index schema + from: claim # the value comes from the verified token + claim: "https://acme.example/org" # the claim name your identity provider emits + required: true +``` + +`field` and `claim` are independent names: `field` is what the index schema calls the tenant column, and `claim` is what the identity provider calls it. A token carrying `"https://acme.example/org": "acme"` makes every query from that caller run as `@org_id:{acme} AND `. + +The model cannot see or set the injected value. The field is absent from the tool's input schema, and it is left out of the field hints appended to the tool description, so the model is not told the field exists. If the model filters on it anyway, its clause ANDs with the injected one: it can narrow within its own tenant, and naming another tenant matches nothing. The rest of the profile works as before, so a static `lock.filter` on another field and a model-supplied filter both still apply. + +#### What Counts as a Usable Claim + +The claim must be a single, non-empty string. Anything else refuses the request with a `forbidden` error, and no query runs: + +| Claim value | Why it is refused | +|---|---| +| Absent, or the request has no token | There is no tenant to scope to. | +| `null` or `""` | An empty tag value would drop the tenant clause from the query entirely. | +| A list, such as `["acme", "victim"]` | It would render as a union, `@org_id:{acme\|victim}`, which spans both tenants. | +| An object, number or boolean | It is not a tenant identifier. | +| Padded with whitespace, or containing `\|` | Neither can be a real tenant identifier, and refusing is safer than guessing. | + +A list is refused because of its type, not because of what it renders as. The union it produces is indistinguishable from one a caller could legitimately ask for, so no inspection of the finished query could catch it. + +With several `inject` entries, every entry ANDs into the query, and one unusable claim refuses the whole request rather than narrowing by the entries that did resolve. + +#### What Fails at Startup + +Injection is checked at startup wherever the configuration alone can show it would not work: + +- authentication is not enabled, on any transport, including an unauthenticated loopback HTTP bind and any `--allow-unauthenticated` bind; +- authentication is configured but the server runs over `stdio`, which is never authenticated; +- the injected field is absent from the bound index, is not a tag field, or is declared `NOINDEX`; +- an `inject` list is empty, names one field twice, or names a field that `lock.filter` also constrains; +- `required` is anything but `true`, or `from` is anything but `claim`. + +An injected field must be a tag. Text equality is a phrase match over tokenised text, and text is tokenised on punctuation, so the phrase `acme-corp` would also match `acme-corp-eu`. + +The tool set registers once per process. If a restart reloads a configuration that differs from the registered one, the server normally logs a warning and keeps the old tools. When injection is configured on either side of the change, startup fails instead, because keeping the old tenant scoping in force is not something a log line should report. + +#### Threat Model + +The guarantee is precise: a client presenting a validly signed token cannot make the model widen or escape the tenant scope carried in that token. The trust boundary is the identity provider, not the MCP client, so the guarantee holds only while these hold: + +- The token is genuinely verified. Use a real signing key and an asymmetric algorithm. The server refuses to start an injecting profile without authentication, but it does not check which algorithm you configured. +- The identity provider assigns the claim. If a tenant can mint its own token, or set the claim itself, nothing here stops it reading another tenant's data. +- Every document carries its tenant. The tenant field must be stamped on each document and indexed as a tag. A document without the field matches no tenant, so it is invisible rather than shared. +- Tenant identifiers differ by more than case. Tag fields fold case unless declared `CASESENSITIVE`, so `Acme` and `acme` would be one tenant. The server warns at startup when an injected field is not case-sensitive. + +Where tenants share one index, the injected filter is the only isolation boundary. There is no Redis ACL or keyspace separation behind it, so a defect in filter combination or claim validation is a full cross-tenant read. If you need defence in depth, separate tenants at the Redis layer as well. + +Listing the tenant claim under `auth.required_claims` is a cheap outer layer: the verifier then rejects a token that lacks the claim before any tool runs. That check confirms only that the claim is present. Its value is still validated by the profile on every call. + ## Why Use MCP Instead of Direct RedisVL Calls Use RedisVL MCP when you want a standard tool boundary for agent frameworks or assistants that already speak MCP. diff --git a/docs/user_guide/how_to_guides/mcp.md b/docs/user_guide/how_to_guides/mcp.md index 1bfe2f53..c2e31744 100644 --- a/docs/user_guide/how_to_guides/mcp.md +++ b/docs/user_guide/how_to_guides/mcp.md @@ -340,6 +340,78 @@ Rules worth knowing: Misconfiguration fails at startup, not at the first call — a name colliding with a built-in or using a reserved `redisvl-`/`redisvl_` prefix, a duplicate name, a missing or unknown `index`, a cap above `max_limit`, hiding `query`, locking `return_fields` while also exposing them, or a locked filter or projection naming a field the index does not have. Unrecognized keys are rejected too, so a typo in `lock` fails loudly instead of quietly producing a tool that reads as locked but enforces nothing. +### Tenant Scoping With Claim Injection + +When tenants share one index, a profile can take the tenant from the caller's verified token instead of trusting the model to pass a filter. This needs authentication, so the example configures it; see {doc}`mcp_authentication` for the rest of that block. + +The index needs the tenant on every document, as a tag. Declare it `CASESENSITIVE` unless your identity provider guarantees one case, because tag fields otherwise treat `Acme` and `acme` as the same tenant: + +```yaml +# The RedisVL schema the index was created from +index: + name: customer-kb + prefix: kb +fields: + - name: content + type: text + - name: org_id + type: tag + attrs: + case_sensitive: true +``` + +Then point a profile at it: + +```yaml +server: + redis_url: redis://localhost:6379 + builtin_tools: + search-records: disabled # otherwise the unscoped built-in stays available + auth: + type: jwt + jwks_uri: ${MCP_JWKS_URI} + issuer: ${MCP_ISSUER} + audience: api://redisvl-mcp + required_claims: [exp, iat, "https://acme.example/org"] + +indexes: + customer_kb: + redis_name: customer-kb + search: + type: fulltext + runtime: + text_field_name: content + +custom_tools: + - name: search-customer-kb + index: customer_kb + description: Search this customer's knowledge base. + lock: + inject: + - field: org_id + from: claim + claim: "https://acme.example/org" + required: true +``` + +Serve it over HTTP: + +```bash +rvl mcp --config /path/to/mcp_config.yaml --transport streamable-http +``` + +What the client sees for `search-customer-kb`: + +- `query` (required), `limit`, `offset`, `filter` and `return_fields`, with no argument for `org_id`. +- A description whose field hints list `content` but not `org_id`. +- Results from its own tenant only. A `filter` naming `org_id` ANDs with the injected value, so naming another tenant returns nothing. + +Disable `search-records`, as above, unless every tenant may read the whole index. The built-in has no injected scope, so leaving it enabled hands every caller an unscoped way round the profile. + +Listing the tenant claim under `required_claims` makes the verifier reject a token without it before any tool runs. That checks presence only: the profile still validates the value on every call, and refuses a missing, empty, list-valued or otherwise unusable claim with a `forbidden` error before any query runs. + +The server refuses to start an injecting profile without authentication, over `stdio`, or on a field the index does not hold as an indexed tag. For what the guarantee covers and what it rests on, read the threat model in {doc}`/concepts/mcp`. + ## Tool Contracts RedisVL MCP exposes a small, implementation-owned contract. @@ -721,3 +793,11 @@ If the vectorizer dims do not match the configured vector field dims, startup fa ### Hybrid Config Requires Native Runtime Support Some hybrid params depend on native hybrid support in Redis and redis-py. If your environment does not support that path, remove native-only params such as `knn_ef_runtime` or upgrade Redis and redis-py. + +### Claim Injection Requires Authentication + +A profile with `lock.inject` refuses to start when authentication is not enabled, or when the server runs over `stdio`, because neither can supply a verified token. Configure `server.auth` and serve over `sse` or `streamable-http`. The check runs before the server connects to Redis, so it reports even when Redis is unreachable. + +### Claim Injection Fails Every Request With `forbidden` + +The token is verified but its claim is unusable: missing, empty, padded with whitespace, or not a single string. The error names the claim and the tool. A misspelled `claim` name in the config is the usual cause, since a JWT claim name such as `https://acme.example/org` must match exactly. diff --git a/docs/user_guide/how_to_guides/mcp_authentication.md b/docs/user_guide/how_to_guides/mcp_authentication.md index a6cd7747..643f5345 100644 --- a/docs/user_guide/how_to_guides/mcp_authentication.md +++ b/docs/user_guide/how_to_guides/mcp_authentication.md @@ -45,9 +45,7 @@ On each request it checks: `search-records` and a **write scope** to call `upsert-records`. ```{important} -This is **coarse** authorization: it decides whether a caller may connect and -whether it may read or write. It does **not** map token claims to a Redis ACL -user, a per-tenant index, or query filters. See [The Authorization Boundary](#the-authorization-boundary). +This is **coarse** authorization: it decides whether a caller may connect and whether it may read or write. It does **not** map token claims to a Redis ACL user or a per-tenant index. A custom tool profile can additionally scope its queries to a tenant claim; see [The Authorization Boundary](#the-authorization-boundary). ``` ## OAuth: Which Part RedisVL Handles @@ -184,29 +182,24 @@ A token like the following would then pass the read gate, because ## The Authorization Boundary -RedisVL MCP authenticates the caller and gates read vs write. It does **not** -translate token claims (such as a tenant id or role) into a specific Redis ACL -user, a per-tenant index, or injected query filters. The server holds one Redis -connection for one index, established at startup. +RedisVL MCP authenticates the caller, gates read vs write, and can scope every query a custom tool profile runs to a tenant carried in the token. It does **not** translate token claims into a specific Redis ACL user or a per-tenant index: every caller shares the server's Redis connection, established at startup. -Fine-grained, per-tenant data isolation belongs in a **gateway or policy layer** -in front of the MCP server, which validates the token, looks up a binding of -claim to Redis identity, and injects credentials and filters. +To scope queries by tenant inside RedisVL, add `lock.inject` to a profile; see Tenant Scoping With Claim Injection in {doc}`mcp`. That filter is then the only thing separating tenants, so read the threat model in {doc}`/concepts/mcp` before relying on it. + +Isolation enforced by Redis itself, rather than by a query filter, belongs in a **gateway or policy layer** in front of the MCP server, which validates the token, looks up a binding of claim to Redis identity, and injects the matching credentials. ```mermaid flowchart LR subgraph Gateway["Gateway / policy layer (out of scope for RedisVL)"] - T[Validate token] --> M["Map claims to
Redis user + index + filters"] + T[Validate token] --> M["Map claims to
Redis user + index"] end subgraph RedisVL["RedisVL MCP (this guide)"] - A[Validate JWT] --> S[Gate read / write by scope] + A[Validate JWT] --> S[Gate read / write by scope] --> I["Inject tenant filter
(profiles with lock.inject)"] end Client --> Gateway --> RedisVL --> Redis[(Redis)] ``` -Use RedisVL's JWT validation for authentication and coarse read/write -authorization. Layer a gateway on top when you need per-tenant Redis ACL -enforcement. +Use RedisVL's JWT validation for authentication, read/write authorization, and tenant-scoped queries. Layer a gateway on top when you need per-tenant Redis ACL enforcement. When such a gateway or reverse proxy terminates the connection and forwards a rewritten `Host` header, set `server.transport_security.enabled: false` (or @@ -216,4 +209,5 @@ proxy's rewritten `Host` is not rejected by the Host/Origin guard. ## See Also - {doc}`mcp`: run and configure the RedisVL MCP server. +- {doc}`/concepts/mcp`: custom tool profiles, tenant scoping, and its threat model. diff --git a/redisvl/mcp/config.py b/redisvl/mcp/config.py index 07a056db..244e0298 100644 --- a/redisvl/mcp/config.py +++ b/redisvl/mcp/config.py @@ -627,6 +627,53 @@ def _validate_max(self) -> "MCPProfileParamConfig": return self +def iter_filter_clauses(node: Any) -> Any: + """Yield every leaf clause of a JSON filter DSL expression.""" + if not isinstance(node, dict): + return + for operator in ("and", "or"): + if operator in node: + for child in node[operator] or []: + yield from iter_filter_clauses(child) + return + if "not" in node: + yield from iter_filter_clauses(node["not"]) + return + yield node + + +class MCPProfileInjectConfig(BaseModel): + """One tenant-scoping filter whose value comes from the caller's verified token. + + ``field`` and ``claim`` are independent author-chosen strings: ``field`` is + what the index schema calls the tenant column, ``claim`` is what the identity + provider calls it. Only ``field`` can be checked at startup, against the + bound schema; ``claim`` cannot, because no token exists until a request. + """ + + # `from` is a Python keyword, so the attribute is `source`. populate_by_name + # keeps it constructible from Python as well as from YAML. + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + field: str = Field(..., min_length=1) + # Required rather than defaulted, and a single-value Literal today: the + # source is the whole point of the entry, so it is stated, not inferred. + source: Literal["claim"] = Field(..., alias="from") + claim: str = Field(..., min_length=1) + # Optional injection is unscoped injection, so `false` is not a value this + # accepts. The key exists so the proposal's published YAML loads verbatim. + required: Literal[True] = True + + @model_validator(mode="after") + def _validate_inject(self) -> "MCPProfileInjectConfig": + """Reject a blank field or claim name, which min_length lets through.""" + if not self.field.strip(): + raise ValueError("custom_tools lock.inject field must not be blank") + if not self.claim.strip(): + raise ValueError("custom_tools lock.inject claim must not be blank") + return self + + class MCPProfileLockConfig(BaseModel): """Author-locked arguments that the model cannot override or remove.""" @@ -634,6 +681,7 @@ class MCPProfileLockConfig(BaseModel): return_fields: list[str] | None = None filter: dict[str, Any] | None = None + inject: list[MCPProfileInjectConfig] | None = None @model_validator(mode="after") def _validate_lock(self) -> "MCPProfileLockConfig": @@ -653,6 +701,38 @@ def _validate_lock(self) -> "MCPProfileLockConfig": raise ValueError( "custom_tools lock.return_fields must contain non-empty strings" ) + + if self.inject is not None: + if not self.inject: + raise ValueError( + "custom_tools lock.inject must contain at least one entry; " + "drop the key for a profile that does not inject" + ) + + injected = [entry.field for entry in self.inject] + duplicated = sorted({name for name in injected if injected.count(name) > 1}) + if duplicated: + raise ValueError( + "custom_tools lock.inject names the same field more than " + f"once: {', '.join(duplicated)}" + ) + + # A static lock and an injected value on one field are two answers + # to "which tenant", and they AND together -- so a mismatch matches + # nothing and a match makes one of them dead config. Either way the + # author meant only one of them. + locked = { + clause.get("field") + for clause in iter_filter_clauses(self.filter) + if isinstance(clause.get("field"), str) + } + colliding = sorted(locked.intersection(injected)) + if colliding: + raise ValueError( + "custom_tools lock.inject and lock.filter both constrain " + f"{', '.join(colliding)}; an injected field must not also " + "appear in lock.filter" + ) return self diff --git a/redisvl/mcp/server.py b/redisvl/mcp/server.py index 554ccc18..1ce44090 100644 --- a/redisvl/mcp/server.py +++ b/redisvl/mcp/server.py @@ -31,6 +31,14 @@ logger = logging.getLogger(__name__) + +def _config_injects(config: Any) -> bool: + """Report whether any configured profile injects a claim-derived filter.""" + if config is None: + return False + return any(profile.lock.inject for profile in config.custom_tools) + + try: from fastmcp import FastMCP except ImportError: @@ -73,6 +81,10 @@ def __init__(self, settings: MCPSettings): self._semaphore: asyncio.Semaphore | None = None self._tools_registered = False self._registered_tool_fingerprint = "" + self._registered_tools_inject = False + # Set by run_async, the only point in the process that knows it. None + # means an embedder started the server without naming a transport. + self._transport: str | None = None # Lifecycle management self._lifecycle_state = _LifecycleState.INITIAL # Server lifecycle @@ -118,6 +130,15 @@ async def run_async( outermost, rejecting DNS-rebinding requests before auth or tool handlers. ``stdio`` is untouched. """ + # Resolved the way FastMCP resolves it, so `run_async()` with no + # argument is recorded as the stdio it will actually serve. Imported + # here because this module stays importable without the `mcp` extra. + import fastmcp + + self._transport = ( + transport if transport is not None else fastmcp.settings.transport + ) + if transport in ("sse", "streamable-http"): host = transport_kwargs.get("host", "127.0.0.1") port = transport_kwargs.get("port", 8000) @@ -275,7 +296,7 @@ async def _probe_native_hybrid_search(index: AsyncSearchIndex) -> bool: return hasattr(client.ft(index.schema.index.name), "hybrid_search") - def _validate_custom_tools(self) -> None: + def _validate_custom_tools_against_schema(self) -> None: """Fail startup on profiles that do not fit their bound index schema. Config load already checked naming, parameter policy, and that a pinned @@ -293,6 +314,39 @@ def _validate_custom_tools(self) -> None: runtime = self._bindings[binding_id] validate_profile_against_schema(profile, runtime.schema) + def _verify_injection_has_a_token(self, profiles: Any) -> None: + """Refuse to start a claim-injection profile that can never succeed. + + Keyed off "auth is enabled", not off the transport name alone, so it + covers an unauthenticated loopback bind and any --allow-unauthenticated + bind as well as stdio. Without a verified token every call would be + refused at request time -- safe, but advertised as a tool that works. + + Auth configured under stdio is the second case: FastMCP never + authenticates stdio, so the verifier exists and is never consulted. Only + a transport recorded by run_async counts; an embedder that never names + one is left to the request-time refusal, which still fails closed. + """ + injecting = sorted(profile.name for profile in profiles if profile.lock.inject) + if not injecting: + return + + names = ", ".join(injecting) + if not self._auth_enabled: + raise ValueError( + f"custom_tools {names} inject a tenant filter from a token " + "claim, but authentication is not enabled, so there is no " + "verified token to read. Configure server.auth (or " + "REDISVL_MCP_AUTH_*) and serve over an HTTP transport." + ) + if self._transport == "stdio": + raise ValueError( + f"custom_tools {names} inject a tenant filter from a token " + "claim, but the server is running over stdio, which is never " + "authenticated. Serve over an HTTP transport (sse or " + "streamable-http) so the configured auth applies." + ) + @staticmethod def _tool_surface_fingerprint(config: Any) -> str: """Summarize the config that a registered tool set baked in.""" @@ -318,8 +372,22 @@ def _register_tools(self) -> None: # The dangerous direction is an operator disabling a tool or tightening # a lock and believing the restart applied it. if self._tools_registered: - current = self._tool_surface_fingerprint(getattr(self, "config", None)) + config = getattr(self, "config", None) + current = self._tool_surface_fingerprint(config) if current != self._registered_tool_fingerprint: + # For a tenant boundary "the old tools are still in effect" + # is not a log line. Either side counts: adding injection + # to a profile registered without it leaves that tool + # serving every tenant while the config says otherwise. + if self._registered_tools_inject or _config_injects(config): + raise RuntimeError( + "MCP tool configuration changed since tools were " + "registered, and claim injection is configured on " + "one side of the change. Tools register once per " + "process, so the previously registered tenant " + "scoping would stay in effect. Restart the process " + "to apply the new configuration." + ) logger.warning( "MCP tool configuration (built-in or custom) changed " "since tools were registered, but tools register once per " @@ -381,6 +449,7 @@ def _register_tools(self) -> None: self._warn_on_unusable_tool_surface(registered) self._registered_tool_fingerprint = self._tool_surface_fingerprint(config) + self._registered_tools_inject = _config_injects(config) self._tools_registered = True def _warn_on_unusable_tool_surface(self, registered: list[str]) -> None: @@ -550,6 +619,9 @@ async def _initialize_runtime_resources(self) -> None: """Load config and initialize every configured binding independently.""" self.config = load_mcp_config(self._config_path) self._verify_auth_not_stale() + # Before any binding connects: this depends only on the loaded config, + # so an unworkable one should not first need a reachable Redis. + self._verify_injection_has_a_token(self.config.custom_tools) # The semaphore is a single process-wide concurrency ceiling shared by # all bindings; take the max across bindings. This means the most # permissive binding sets the cap — e.g. five bindings each configured @@ -571,7 +643,7 @@ async def _initialize_runtime_resources(self) -> None: ) # Validate before registering so a bad profile fails startup rather than # leaving a half-registered tool set behind. - self._validate_custom_tools() + self._validate_custom_tools_against_schema() self._register_tools() async def _initialize_binding( diff --git a/redisvl/mcp/tools/profiles.py b/redisvl/mcp/tools/profiles.py index 8fae2796..0731a2f0 100644 --- a/redisvl/mcp/tools/profiles.py +++ b/redisvl/mcp/tools/profiles.py @@ -12,12 +12,17 @@ """ import inspect +import logging from typing import Annotated, Any, Optional from pydantic import Field -from redisvl.mcp.auth import ensure_read_scope -from redisvl.mcp.config import MCPCustomToolConfig +from redisvl.mcp.auth import ( + build_injected_filter, + ensure_read_scope, + validate_inject_against_schema, +) +from redisvl.mcp.config import MCPCustomToolConfig, iter_filter_clauses from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError from redisvl.mcp.filters import parse_filter from redisvl.mcp.tools.search import ( @@ -28,6 +33,8 @@ from redisvl.query.filter import FilterExpression from redisvl.schema import IndexSchema +logger = logging.getLogger(__name__) + # Ordered so the generated signature reads like the built-in's: required query # first, then the optional narrowing arguments. Real type objects rather than # string annotations, so schema generation never depends on resolving a forward @@ -56,11 +63,16 @@ def build_profile_description(profile: MCPCustomToolConfig, schema: IndexSchema) if profile.suppress_schema_hints: return description + # Narrower than suppressing the hints wholesale: the model still learns + # which fields it may legitimately narrow on, and only the tenant field it + # must not reason about is withheld. + injected = frozenset(entry.field for entry in profile.lock.inject or ()) + parts = [description] if profile.param_exposed("filter"): - parts.append(_build_filter_hint(schema)) + parts.append(_build_filter_hint(schema, exclude=injected)) if profile.param_exposed("return_fields"): - parts.append(_build_return_fields_hint(schema)) + parts.append(_build_return_fields_hint(schema, exclude=injected)) return " ".join(parts) @@ -132,20 +144,36 @@ def validate_profile_against_schema( # Raises RedisVLMCPError(INVALID_FILTER) naming the offending field. resolve_locked_filter(profile, schema) + if profile.lock.inject: + validate_inject_against_schema( + profile.lock.inject, schema, profile_name=profile.name + ) + _warn_on_case_insensitive_inject(profile, schema) + -def _iter_filter_clauses(node: Any) -> Any: - """Yield every leaf clause of a JSON filter DSL expression.""" - if not isinstance(node, dict): - return - for operator in ("and", "or"): - if operator in node: - for child in node[operator] or []: - yield from _iter_filter_clauses(child) - return - if "not" in node: - yield from _iter_filter_clauses(node["not"]) - return - yield node +def _warn_on_case_insensitive_inject( + profile: MCPCustomToolConfig, schema: IndexSchema +) -> None: + """Warn when an injected tag field folds case, which merges tenants. + + A tag field is case-insensitive unless declared ``CASESENSITIVE``, so the + tenants ``Acme`` and ``acme`` are one tenant to Redis. A warning rather than + a failure: an IdP that only ever emits lower-case identifiers is safe, and + only the operator knows whether theirs does. The value is deliberately not + normalized here -- that would hide the collision rather than surface it. + """ + for entry in profile.lock.inject or (): + field = schema.fields.get(entry.field) + if field is not None and not getattr(field.attrs, "case_sensitive", False): + logger.warning( + "custom_tools '%s' injects claim '%s' into tag field '%s', which " + "is not CASESENSITIVE, so tenant identifiers differing only in " + "case match the same documents. Declare the field CASESENSITIVE " + "unless the identity provider guarantees a single case.", + profile.name, + entry.claim, + entry.field, + ) def _validate_locked_exists_fields( @@ -165,7 +193,7 @@ def _validate_locked_exists_fields( if profile.lock.filter is None: return - for clause in _iter_filter_clauses(profile.lock.filter): + for clause in iter_filter_clauses(profile.lock.filter): if str(clause.get("op", "")).lower() != "exists": continue field_name = clause.get("field") @@ -229,8 +257,12 @@ def register_profile_tool( ``limit`` collapses its cap into a fixed result count. The wrapper then ignores anything the profile does not expose, so a lock holds even if a caller reaches it without schema validation. + + The one thing resolved per call is the injected tenant filter, because its + value belongs to the caller's token rather than to the configuration. """ locked_filter = resolve_locked_filter(profile, schema) + inject_specs = profile.lock.inject or [] locked_return_fields = profile.lock.return_fields limit_cap = profile.param_max("limit") exposes_limit = profile.param_exposed("limit") @@ -242,6 +274,20 @@ def register_profile_tool( async def profile_tool(**kwargs: Any) -> dict[str, Any]: ensure_read_scope(server) + # Per request, because the value is the caller's. Folded into the locked + # side rather than passed alongside it: `merge_locked_filter` returns the + # caller's filter untouched when nothing is locked, so only a non-None + # locked side forces the caller's filter through the escape backstop. + # A fresh local, so the registration-time expression is never rebound. + effective_locked = locked_filter + if inject_specs: + injected = build_injected_filter( + inject_specs, schema, tool_name=profile.name + ) + effective_locked = ( + injected if effective_locked is None else effective_locked & injected + ) + # A hidden argument is already absent from the advertised schema, so a # compliant client cannot send one. Ignoring it here too means the lock # does not depend on the client or the schema layer holding up. @@ -280,7 +326,7 @@ def supplied(name: str, default: Any = None) -> Any: if locked_return_fields is not None else supplied("return_fields") ), - locked_filter=locked_filter, + locked_filter=effective_locked, # Passed down rather than checked here: an omitted limit resolves to # the binding default inside search_records, so that is the only # place able to bound both paths. diff --git a/redisvl/mcp/tools/search.py b/redisvl/mcp/tools/search.py index 81382120..38ab4076 100644 --- a/redisvl/mcp/tools/search.py +++ b/redisvl/mcp/tools/search.py @@ -29,22 +29,34 @@ ) -def _build_filter_hint(schema: IndexSchema) -> str: - """Describe fields with typed operator support in the JSON filter DSL.""" +def _build_filter_hint( + schema: IndexSchema, exclude: frozenset[str] = frozenset() +) -> str: + """Describe fields with typed operator support in the JSON filter DSL. + + ``exclude`` drops fields the model must not reason about, such as a + claim-injected tenant field: filtering on it is harmless -- the caller's + clause ANDs with the injected one -- but naming it contradicts the promise + that the model never sees the knob. + """ filter_fields = [ f"{field.name}({getattr(field.type, 'value', field.type)})" for field in schema.fields.values() - if field.type in _DSL_FILTER_FIELD_TYPES + if field.type in _DSL_FILTER_FIELD_TYPES and field.name not in exclude ] if not filter_fields: return "Object filter fields: none." return "Object filter fields: " + ", ".join(filter_fields) + "." -def _build_return_fields_hint(schema: IndexSchema) -> str: +def _build_return_fields_hint( + schema: IndexSchema, exclude: frozenset[str] = frozenset() +) -> str: """Describe all fields that callers can request in `return_fields`.""" returnable_fields = [ - field.name for field in schema.fields.values() if field.type != "vector" + field.name + for field in schema.fields.values() + if field.type != "vector" and field.name not in exclude ] if not returnable_fields: return "Allowed return_fields: none." diff --git a/tests/integration/test_mcp/test_claim_injection.py b/tests/integration/test_mcp/test_claim_injection.py new file mode 100644 index 00000000..c1cd26f1 --- /dev/null +++ b/tests/integration/test_mcp/test_claim_injection.py @@ -0,0 +1,297 @@ +"""End-to-end tests for claim-injected tenant scoping against a real Redis. + +Two tenants share one index, separated by an ``org_id`` tag. Tokens are real +RS256 JWTs minted with FastMCP's ``RSAKeyPair`` and verified by the server's own +configured verifier, so the claims the profile reads are the ones a verified +request would carry -- only the request-context lookup is substituted. +""" + +from pathlib import Path + +import pytest +import yaml + +fastmcp = pytest.importorskip( + "fastmcp", reason="fastmcp not installed (install redisvl[mcp])" +) +from fastmcp.server.auth.providers.jwt import RSAKeyPair + +from redisvl.index import AsyncSearchIndex +from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError +from redisvl.mcp.server import RedisVLMCPServer +from redisvl.mcp.settings import MCPSettings +from redisvl.schema import IndexSchema + +ISSUER = "https://auth.acme.example/" +AUDIENCE = "api://redisvl-mcp" +ORG_CLAIM = "https://acme.example/org" +TOOL = "search-customer-kb" + +_PROFILE = { + "name": TOOL, + "description": "Search this customer's knowledge base.", + "lock": {"inject": [{"field": "org_id", "from": "claim", "claim": ORG_CLAIM}]}, +} + + +@pytest.fixture(scope="module") +def key() -> RSAKeyPair: + return RSAKeyPair.generate() + + +@pytest.fixture +async def tenant_index(async_client, worker_id): + schema = IndexSchema.from_dict( + { + "index": { + "name": f"mcp-tenants-{worker_id}", + "prefix": f"mcp-tenants:{worker_id}", + "storage_type": "hash", + }, + "fields": [ + {"name": "content", "type": "text"}, + {"name": "org_id", "type": "tag", "attrs": {"case_sensitive": True}}, + {"name": "category", "type": "tag"}, + {"name": "notes", "type": "text"}, + # NOINDEX is only accepted alongside SORTABLE. + { + "name": "shadow", + "type": "tag", + "attrs": {"no_index": True, "sortable": True}, + }, + ], + } + ) + index = AsyncSearchIndex(schema=schema, redis_client=async_client) + await index.create(overwrite=True, drop=True) + # Identical content across tenants, so only the tenant clause can separate + # them -- a query that leaks would match all four. + await index.load( + [ + { + "id": "a1", + "content": "refund policy", + "org_id": "acme", + "category": "billing", + }, + { + "id": "a2", + "content": "refund window", + "org_id": "acme", + "category": "billing", + }, + { + "id": "v1", + "content": "refund policy", + "org_id": "victim", + "category": "billing", + }, + { + "id": "v2", + "content": "refund window", + "org_id": "victim", + "category": "billing", + }, + ], + id_field="id", + ) + yield index + await index.delete(drop=True) + + +@pytest.fixture +def config_path(tmp_path: Path, redis_url: str, key: RSAKeyPair): + def factory(redis_name: str, custom_tools: list[dict], *, auth: bool = True) -> str: + server: dict = {"redis_url": redis_url} + if auth: + server["auth"] = { + "type": "jwt", + "public_key": key.public_key, + "issuer": ISSUER, + "audience": AUDIENCE, + } + config = { + "server": server, + "indexes": { + "kb": { + "redis_name": redis_name, + "search": {"type": "fulltext"}, + "runtime": {"text_field_name": "content"}, + } + }, + "custom_tools": custom_tools, + } + path = tmp_path / f"{redis_name}-inject.yaml" + path.write_text(yaml.safe_dump(config), encoding="utf-8") + return str(path) + + return factory + + +@pytest.fixture +async def started(tenant_index, config_path): + servers: list[RedisVLMCPServer] = [] + + async def start(custom_tools=(_PROFILE,), **kwargs) -> RedisVLMCPServer: + server = RedisVLMCPServer( + MCPSettings( + config=config_path( + tenant_index.schema.index.name, list(custom_tools), **kwargs + ) + ) + ) + await server.startup() + servers.append(server) + return server + + yield start + + for server in servers: + await server.shutdown() + + +async def _as_caller(monkeypatch, server, key, claims): + """Verify a real token through the server's verifier and make it current.""" + token = key.create_token( + subject="user-42", + issuer=ISSUER, + audience=AUDIENCE, + additional_claims=claims, + ) + access = await server.auth.verify_token(token) + assert access is not None, "the server's verifier rejected the minted token" + monkeypatch.setattr( + "fastmcp.server.dependencies.get_access_token", lambda: access, raising=False + ) + + +async def _search(server, **kwargs): + tool = await server.get_tool(TOOL) + assert tool is not None + return await tool.fn(query="refund", **kwargs) + + +def _orgs(result) -> set: + return {hit["record"]["org_id"] for hit in result["results"]} + + +async def test_a_tenants_token_returns_only_that_tenants_documents( + started, key, monkeypatch +): + server = await started() + await _as_caller(monkeypatch, server, key, {ORG_CLAIM: "acme"}) + + result = await _search(server) + + assert len(result["results"]) == 2 + assert _orgs(result) == {"acme"} + + +@pytest.mark.parametrize( + "caller_filter, expected", + [ + # Naming the other tenant ANDs with the injected one: nothing matches. + pytest.param({"field": "org_id", "op": "eq", "value": "victim"}, 0, id="swap"), + # The `or` stays nested inside the AND, so it narrows within acme rather + # than hoisting to the top level and reaching victim's documents. + pytest.param( + { + "or": [ + {"field": "org_id", "op": "eq", "value": "victim"}, + {"field": "category", "op": "eq", "value": "billing"}, + ] + }, + 2, + id="or-widening", + ), + pytest.param( + {"not": {"field": "org_id", "op": "eq", "value": "acme"}}, + 0, + id="negation", + ), + ], +) +async def test_a_caller_filter_cannot_widen_or_escape_the_tenant( + started, key, monkeypatch, caller_filter, expected +): + server = await started() + await _as_caller(monkeypatch, server, key, {ORG_CLAIM: "acme"}) + + result = await _search(server, filter=caller_filter) + + # An exact count, not just "no victim rows": a subset check alone would + # pass on an empty result for the wrong reason. + assert len(result["results"]) == expected + assert _orgs(result) <= {"acme"} + + +async def test_the_injected_field_is_absent_from_the_advertised_schema(started): + server = await started() + tool = await server.get_tool(TOOL) + + assert "org_id" not in tool.parameters["properties"] + assert tool.parameters["additionalProperties"] is False + # Nor is it named in the description's field hints. + assert "org_id" not in tool.description + assert "category(tag)" in tool.description + + +@pytest.mark.parametrize( + "claims", + [ + pytest.param({}, id="absent"), + pytest.param({ORG_CLAIM: ""}, id="empty"), + pytest.param({ORG_CLAIM: " "}, id="whitespace"), + pytest.param({ORG_CLAIM: ["acme", "victim"]}, id="array"), + pytest.param({ORG_CLAIM: {"id": "acme"}}, id="object"), + pytest.param({ORG_CLAIM: "acme|victim"}, id="pipe"), + ], +) +async def test_an_unusable_claim_is_refused_before_any_query( + started, key, monkeypatch, claims +): + server = await started() + await _as_caller(monkeypatch, server, key, claims) + + queried: list = [] + index = server.resolve_binding("kb").index + original = index.query + + async def spy(query): + queried.append(query) + return await original(query) + + monkeypatch.setattr(index, "query", spy) + + with pytest.raises(RedisVLMCPError) as exc: + await _search(server) + + assert exc.value.code == MCPErrorCode.FORBIDDEN + assert queried == [] + + +@pytest.mark.parametrize( + "inject_field, auth, expected", + [ + pytest.param("org_id", False, "authentication is not enabled", id="auth-off"), + pytest.param("missing", True, "unknown field 'missing'", id="field-absent"), + pytest.param("notes", True, "requires a tag field", id="field-text"), + pytest.param("shadow", True, "NOINDEX", id="field-noindex"), + ], +) +async def test_startup_refuses_an_injection_profile_that_cannot_scope( + tenant_index, config_path, inject_field, auth, expected +): + profile = { + **_PROFILE, + "lock": { + "inject": [{"field": inject_field, "from": "claim", "claim": ORG_CLAIM}] + }, + } + server = RedisVLMCPServer( + MCPSettings( + config=config_path(tenant_index.schema.index.name, [profile], auth=auth) + ) + ) + with pytest.raises(ValueError, match=expected): + await server.startup() diff --git a/tests/unit/test_mcp/test_config.py b/tests/unit/test_mcp/test_config.py index e44704e1..a16267f9 100644 --- a/tests/unit/test_mcp/test_config.py +++ b/tests/unit/test_mcp/test_config.py @@ -967,3 +967,129 @@ def test_load_mcp_config_parses_custom_tools_from_yaml(tmp_path: Path): # Disabling the built-in a profile supersedes is the motivating pairing, so # confirm both halves survive one load. assert config.server.builtin_tool_enabled("search-records") is False + + +# -------------------------------------------------------------------------- +# lock.inject -- claim-sourced tenant scoping +# -------------------------------------------------------------------------- + +_INJECT = {"field": "tenant_id", "from": "claim", "claim": "org", "required": True} + + +def test_the_proposals_published_injection_yaml_loads_verbatim(tmp_path: Path): + # Copied from the approved proposal, comments included. An operator pasting + # it must hit neither a rejected key nor a rejected position. + config_path = tmp_path / "mcp.yaml" + config_path.write_text( + """ +server: + redis_url: redis://localhost:6379 +indexes: + customer_kb: + redis_name: customer-kb + search: + type: fulltext + runtime: + text_field_name: content +custom_tools: + - name: search_customer_kb + kind: profile + based_on: search-records + index: customer_kb + description: Search this customer's knowledge base. + suppress_schema_hints: true # default for injection profiles; see below + lock: + inject: + - field: tenant_id # a field in THIS index's schema + from: claim # source is the auth token, not the LLM + claim: "https://acme.example/org" # a claim name THIS IdP emits + required: true # absent/empty/non-scalar => reject; never unscoped +""".strip(), + encoding="utf-8", + ) + + (entry,) = load_mcp_config(str(config_path)).custom_tools[0].lock.inject + + assert (entry.field, entry.source, entry.claim, entry.required) == ( + "tenant_id", + "claim", + "https://acme.example/org", + True, + ) + + +@pytest.mark.parametrize( + "patch, expected", + [ + # Optional injection is unscoped injection. + pytest.param({"required": False}, "required", id="required-false"), + pytest.param({"from": "header"}, "from", id="unknown-source"), + pytest.param({"claim": " "}, "claim must not be blank", id="blank-claim"), + pytest.param({"field": " "}, "field must not be blank", id="blank-field"), + pytest.param({"default": "acme"}, "default", id="misspelled-or-extra-key"), + ], +) +def test_an_unusable_inject_entry_is_rejected(patch, expected): + entry = {**_INJECT, **patch} + config = _raw_config_with_profiles(_profile_dict(lock={"inject": [entry]})) + with pytest.raises(ValueError, match=expected): + MCPConfig.model_validate(config) + + +def test_an_inject_entry_must_name_its_source(): + # Stated rather than defaulted: the source is the whole point of the entry. + entry = {key: value for key, value in _INJECT.items() if key != "from"} + config = _raw_config_with_profiles(_profile_dict(lock={"inject": [entry]})) + with pytest.raises(ValueError, match="from"): + MCPConfig.model_validate(config) + + +def test_an_empty_inject_list_is_rejected(): + config = _raw_config_with_profiles(_profile_dict(lock={"inject": []})) + with pytest.raises(ValueError, match="at least one entry"): + MCPConfig.model_validate(config) + + +def test_injecting_one_field_twice_is_rejected(): + second = {**_INJECT, "claim": "other_org"} + config = _raw_config_with_profiles( + _profile_dict(lock={"inject": [_INJECT, second]}) + ) + with pytest.raises(ValueError, match="same field more than once: tenant_id"): + MCPConfig.model_validate(config) + + +@pytest.mark.parametrize( + "locked_filter", + [ + pytest.param({"field": "tenant_id", "op": "eq", "value": "acme"}, id="leaf"), + pytest.param( + { + "and": [ + {"field": "category", "op": "eq", "value": "kb"}, + {"not": {"field": "tenant_id", "op": "eq", "value": "x"}}, + ] + }, + id="nested", + ), + ], +) +def test_an_injected_field_cannot_also_be_statically_locked(locked_filter): + config = _raw_config_with_profiles( + _profile_dict(lock={"filter": locked_filter, "inject": [_INJECT]}) + ) + with pytest.raises(ValueError, match="both constrain tenant_id"): + MCPConfig.model_validate(config) + + +def test_injection_composes_with_a_static_lock_on_another_field(): + config = _raw_config_with_profiles( + _profile_dict( + lock={ + "filter": {"field": "category", "op": "eq", "value": "kb"}, + "inject": [_INJECT], + } + ) + ) + profile = MCPConfig.model_validate(config).custom_tools[0] + assert [entry.field for entry in profile.lock.inject] == ["tenant_id"] diff --git a/tests/unit/test_mcp/test_profiles_unit.py b/tests/unit/test_mcp/test_profiles_unit.py index 906e64a4..b1c125b3 100644 --- a/tests/unit/test_mcp/test_profiles_unit.py +++ b/tests/unit/test_mcp/test_profiles_unit.py @@ -918,3 +918,120 @@ async def test_profile_tool_threads_the_locked_filter_into_every_search_mode( # branch would leave vector and hybrid profiles silently unscoped. assert mode == expected_mode assert str(kwargs["filter_expression"]) == "(@category:{resolved} @rating:[4 +inf])" + + +# -------------------------------------------------------------------------- +# Claim injection +# -------------------------------------------------------------------------- + +INJECT_CATEGORY = [{"field": "category", "from": "claim", "claim": "org"}] + + +def _as_caller(monkeypatch, claims): + """Make the current request carry these claims, or no token when None.""" + pytest.importorskip( + "fastmcp", reason="fastmcp not installed (install redisvl[mcp])" + ) + token = None if claims is None else SimpleNamespace(claims=claims) + monkeypatch.setattr( + "fastmcp.server.dependencies.get_access_token", lambda: token, raising=False + ) + + +def test_description_hints_omit_the_injected_field_and_keep_the_rest(): + description = build_profile_description( + _profile(lock={"inject": INJECT_CATEGORY}), _schema() + ) + + # Narrower than suppressing the hints: the model still learns what it may + # filter and return on, and only the tenant field is withheld. + assert "Object filter fields: content(text), rating(numeric)." in description + assert "Allowed return_fields: content, rating." in description + assert "category" not in description + + +@pytest.mark.asyncio +async def test_profile_tool_scopes_every_query_to_the_callers_claim(monkeypatch): + built = _capture_text_queries(monkeypatch) + _as_caller(monkeypatch, {"org": "acme"}) + fn = _register(FakeServer(), _profile(lock={"inject": INJECT_CATEGORY})) + + await fn(query="jam") + + assert str(built[0]["filter_expression"]) == "@category:{acme}" + + +@pytest.mark.asyncio +async def test_profile_tool_ands_the_injected_clause_with_a_static_lock(monkeypatch): + built = _capture_text_queries(monkeypatch) + _as_caller(monkeypatch, {"org": "acme"}) + locked = {"field": "rating", "op": "gte", "value": 4} + fn = _register( + FakeServer(), _profile(lock={"filter": locked, "inject": INJECT_CATEGORY}) + ) + + await fn(query="jam") + + assert str(built[0]["filter_expression"]) == "(@rating:[4 +inf] @category:{acme})" + + +@pytest.mark.asyncio +async def test_injection_alone_still_routes_the_caller_filter_through_the_backstop( + monkeypatch, +): + # The pre-fold seam. `merge_locked_filter` returns the caller's filter + # untouched when nothing is locked, so if the injected expression travelled + # beside the locked side instead of becoming it, a profile with no static + # lock would AND the tenant clause and skip the escape check entirely. + _capture_text_queries(monkeypatch) + _as_caller(monkeypatch, {"org": "acme"}) + checked: list[str] = [] + monkeypatch.setattr( + "redisvl.mcp.tools.search._reject_escapable_filter", + lambda caller: checked.append(str(caller)), + ) + fn = _register(FakeServer(), _profile(lock={"inject": INJECT_CATEGORY})) + + await fn(query="jam", filter={"field": "rating", "op": "gte", "value": 4}) + + assert checked == ["@rating:[4 +inf]"] + + +@pytest.mark.asyncio +async def test_profile_tool_runs_no_query_without_a_verified_token(monkeypatch): + built = _capture_text_queries(monkeypatch) + _as_caller(monkeypatch, None) + fn = _register(FakeServer(), _profile(lock={"inject": INJECT_CATEGORY})) + + with pytest.raises(RedisVLMCPError) as exc: + await fn(query="jam") + + assert exc.value.code == MCPErrorCode.FORBIDDEN + assert built == [] + + +def test_startup_validation_rejects_injecting_into_a_non_tag_field(): + profile = _profile( + lock={"inject": [{"field": "content", "from": "claim", "claim": "org"}]} + ) + with pytest.raises(ValueError, match="requires a tag field"): + validate_profile_against_schema(profile, _schema()) + + +def test_startup_warns_when_the_injected_tag_folds_case(caplog): + # `category` is a plain tag, so `Acme` and `acme` would be one tenant. + with caplog.at_level("WARNING", logger="redisvl.mcp.tools.profiles"): + validate_profile_against_schema( + _profile(lock={"inject": INJECT_CATEGORY}), _schema() + ) + assert any("not CASESENSITIVE" in record.message for record in caplog.records) + + +def test_startup_does_not_warn_for_a_case_sensitive_injected_tag(caplog): + schema = _schema() + schema.fields["category"].attrs.case_sensitive = True + with caplog.at_level("WARNING", logger="redisvl.mcp.tools.profiles"): + validate_profile_against_schema( + _profile(lock={"inject": INJECT_CATEGORY}), schema + ) + assert not any("CASESENSITIVE" in record.message for record in caplog.records) diff --git a/tests/unit/test_mcp/test_server_unit.py b/tests/unit/test_mcp/test_server_unit.py index a8f878da..7c82212a 100644 --- a/tests/unit/test_mcp/test_server_unit.py +++ b/tests/unit/test_mcp/test_server_unit.py @@ -181,6 +181,7 @@ def fake_register_profile_tools(server): server._bindings = bindings server._tools_registered = False server._registered_tool_fingerprint = "" + server._registered_tools_inject = False server.tool = object() server.config = config server.mcp_settings = SimpleNamespace(read_only=False) @@ -356,6 +357,7 @@ def test_register_tools_names_index_ids_when_discovery_is_disabled(monkeypatch): } server._tools_registered = False server._registered_tool_fingerprint = "" + server._registered_tools_inject = False server.tool = object() server.config = _config_with(builtin_tools={"list-indexes": "disabled"}) server.mcp_settings = SimpleNamespace(read_only=False) @@ -388,6 +390,7 @@ def test_register_tools_omits_index_ids_when_discovery_is_available(monkeypatch) } server._tools_registered = False server._registered_tool_fingerprint = "" + server._registered_tools_inject = False server.tool = object() server.config = None server.mcp_settings = SimpleNamespace(read_only=False) @@ -414,6 +417,7 @@ def test_register_tools_warns_when_builtin_config_changed_after_registration( server.tool = object() server._tools_registered = True server._registered_tool_fingerprint = "" + server._registered_tools_inject = False server.config = _config_with(builtin_tools={"upsert-records": "disabled"}) with caplog.at_level(logging.WARNING, logger="redisvl.mcp.server"): @@ -450,6 +454,7 @@ def test_register_tools_gives_upsert_the_same_index_ids_as_search(monkeypatch): } server._tools_registered = False server._registered_tool_fingerprint = "" + server._registered_tools_inject = False server.tool = object() server.config = _config_with(builtin_tools={"list-indexes": "disabled"}) server.mcp_settings = SimpleNamespace(read_only=False) @@ -578,7 +583,7 @@ def test_validate_custom_tools_checks_each_profile_against_its_bound_schema( server._bindings["knowledge"].schema.marker = "knowledge-schema" server._bindings["tickets"].schema.marker = "tickets-schema" - server._validate_custom_tools() + server._validate_custom_tools_against_schema() # Each profile is validated against the schema of the binding it is pinned to. assert validated == [("resolved-search", "tickets-schema")] @@ -604,6 +609,7 @@ def test_register_tools_warns_when_profile_config_changed_after_registration( server.tool = object() server._tools_registered = True server._registered_tool_fingerprint = "" + server._registered_tools_inject = False server.config = _config_with( custom_tools=[ { @@ -624,3 +630,200 @@ def test_register_tools_warns_when_profile_config_changed_after_registration( for record in caplog.records if "changed since tools were registered" in record.message ] + + +# -------------------------------------------------------------------------- +# Claim injection: refusing a profile that can never read a token +# -------------------------------------------------------------------------- + +_INJECTING = { + "name": "tenant-search", + "description": "Search this tenant.", + "lock": {"inject": [{"field": "category", "from": "claim", "claim": "org"}]}, +} + + +def _injection_server(*, auth_enabled, transport, custom_tools=None): + """A server shell holding a validated injecting config.""" + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server.config = _config_with(custom_tools=custom_tools or [_INJECTING]) + server._bindings = {"knowledge": _binding_runtime("knowledge")} + server._auth_enabled = auth_enabled + server._transport = transport + return server + + +@pytest.mark.parametrize("transport", ["stdio", "streamable-http", "sse", None]) +def test_injection_refuses_to_start_without_auth_on_any_transport( + monkeypatch, transport +): + # Keyed off "auth is enabled", never off the transport name alone, so an + # unauthenticated HTTP bind -- loopback or --allow-unauthenticated -- is + # refused exactly as stdio is. + server = _injection_server(auth_enabled=False, transport=transport) + with pytest.raises(ValueError, match="authentication is not enabled"): + server._verify_injection_has_a_token(server.config.custom_tools) + + +def test_injection_refuses_to_start_with_auth_configured_under_stdio(monkeypatch): + # FastMCP never authenticates stdio, so the verifier exists and is never + # consulted: every call would be refused at request time. + server = _injection_server(auth_enabled=True, transport="stdio") + with pytest.raises(ValueError, match="running over stdio"): + server._verify_injection_has_a_token(server.config.custom_tools) + + +@pytest.mark.parametrize("transport", ["streamable-http", "sse", None]) +def test_injection_starts_with_auth_over_http_or_an_unnamed_transport( + monkeypatch, transport +): + # None is an embedder that never named a transport; it is left to the + # request-time refusal, which still fails closed. + server = _injection_server(auth_enabled=True, transport=transport) + server._verify_injection_has_a_token(server.config.custom_tools) + + +def test_a_server_without_injection_needs_no_auth(monkeypatch): + server = _injection_server( + auth_enabled=False, + transport="stdio", + custom_tools=[{"name": "open-search", "description": "Search open."}], + ) + server._verify_injection_has_a_token(server.config.custom_tools) + + +@pytest.mark.asyncio +async def test_an_injection_profile_without_auth_fails_before_redis_is_touched( + monkeypatch, +): + # The refusal depends only on the loaded config, so an unworkable one must + # not first need a reachable Redis and an existing index to be reported. + connected: list[str] = [] + + async def no_connect(self, binding_id, binding): + connected.append(binding_id) + raise AssertionError("a binding was initialized before the refusal") + + monkeypatch.setattr( + "redisvl.mcp.server.load_mcp_config", + lambda path: _config_with(custom_tools=[_INJECTING]), + ) + monkeypatch.setattr(RedisVLMCPServer, "_verify_auth_not_stale", lambda self: None) + monkeypatch.setattr(RedisVLMCPServer, "_initialize_binding", no_connect) + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server._config_path = "unused.yaml" + server._auth_enabled = False + server._transport = None + + with pytest.raises(ValueError, match="authentication is not enabled"): + await server._initialize_runtime_resources() + assert connected == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "transport, recorded", + [ + ("streamable-http", "streamable-http"), + # FastMCP resolves an omitted transport to its configured default, so + # `run_async()` must be recorded as the stdio it will actually serve. + (None, "stdio"), + ], +) +async def test_run_async_records_the_transport_it_will_serve( + monkeypatch, transport, recorded +): + fastmcp = pytest.importorskip( + "fastmcp", reason="fastmcp not installed (install redisvl[mcp])" + ) + monkeypatch.setattr(fastmcp.settings, "transport", "stdio") + + async def no_serve(self, transport=None, show_banner=None, **kwargs): + return None + + monkeypatch.setattr(fastmcp.FastMCP, "run_async", no_serve) + monkeypatch.setattr( + "redisvl.mcp.server.build_host_origin_middleware", lambda *args: [] + ) + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server._transport_security = None + + await server.run_async(transport=transport) + + assert server._transport == recorded + + +def _reregister_after_change(*, registered_with_inject, new_custom_tools): + """Simulate a stop/start against an edited config on an already-registered server.""" + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server._bindings = {"knowledge": _binding_runtime("knowledge")} + server.tool = object() + server._tools_registered = True + server._registered_tool_fingerprint = "the-config-at-registration" + server._registered_tools_inject = registered_with_inject + server.config = _config_with(custom_tools=new_custom_tools) + server._register_tools() + + +@pytest.mark.parametrize( + "registered_with_inject, new_custom_tools", + [ + # Tightening or loosening an injection that is already live. + pytest.param( + True, [{"name": "tenant-search", "description": "x"}], id="removed" + ), + # The dangerous direction: injection added to a profile registered without + # it would keep serving every tenant while the config says otherwise. + pytest.param(False, [_INJECTING], id="added"), + ], +) +def test_a_changed_tool_surface_is_fatal_when_injection_is_on_either_side( + registered_with_inject, new_custom_tools +): + with pytest.raises(RuntimeError, match="claim injection is configured"): + _reregister_after_change( + registered_with_inject=registered_with_inject, + new_custom_tools=new_custom_tools, + ) + + +def test_registration_remembers_that_it_installed_injection(monkeypatch): + # The flag the fatal check reads has to be set by registration itself. A + # first registration with injection, then a reload that drops it, is the + # case where only that remembered flag knows the live tools are scoped. + for target in ( + "register_list_indexes_tool", + "register_search_tool", + "register_upsert_tool", + ): + monkeypatch.setattr(f"redisvl.mcp.server.{target}", lambda *a, **k: None) + monkeypatch.setattr( + "redisvl.mcp.server.register_profile_tools", lambda server: ["tenant-search"] + ) + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server._bindings = {"knowledge": _binding_runtime("knowledge")} + server._tools_registered = False + server._registered_tool_fingerprint = "" + server._registered_tools_inject = False + server.tool = object() + server.mcp_settings = SimpleNamespace(read_only=False) + server.config = _config_with(custom_tools=[_INJECTING]) + server._register_tools() + + server.config = _config_with( + custom_tools=[{"name": "tenant-search", "description": "x"}] + ) + with pytest.raises(RuntimeError, match="claim injection is configured"): + server._register_tools() + + +def test_a_changed_tool_surface_without_injection_still_only_warns(caplog): + with caplog.at_level(logging.WARNING, logger="redisvl.mcp.server"): + _reregister_after_change( + registered_with_inject=False, + new_custom_tools=[{"name": "open-search", "description": "Search."}], + ) + assert any( + "changed since tools were registered" in record.message + for record in caplog.records + ) From 718d79f51dba1eb8145fc57905ab5d62765f14a1 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 1 Oct 2026 17:05:16 +0200 Subject: [PATCH 3/5] fix(mcp): refuse claim characters the tag parser splits on, and allow `|` A six-perspective review of the claim-injection stack, measured on Redis 8.4, found that the claim reader let through values that match a different tenant. Inside tag braces the query parser reads every control character and the backtick as a term separator -- the controls even when escaped -- so `\x01acme` matched the tenant `acme` and `acme\tcorp` matched `acme corp`. A sweep of U+0000-U+024F, the general punctuation block, and the CJK and fullwidth ranges found exactly 33 such code points: U+0001-U+001F, U+007F, and U+0060. The claim reader now refuses any control character or backtick. Escaping the backtick in `Tag` generally is left to its own change. The `|` refusal goes. `Tag` escapes `|` inside a single value and Redis matches the escaped form literally, so the check protected nothing and refused real identifiers: an Auth0 `sub` is shaped `auth0|64f1c2`. A list claim, which is the shape that genuinely unions, is still refused on type. The request-time field re-check in `build_injected_filter` is removed, with its `schema` parameter. It checked the schema captured at registration, so it could never fire, and its comment claimed the opposite. Startup already re-validates the freshly inspected schema on every start. The clause-containment check gains the test it lacked: reverting it failed nothing, so the earlier claim that every guard had been mutation-checked was wrong. The case-folding warning moves into `validate_inject_against_schema`, beside the hard schema checks, so any caller gets it rather than only profiles. A refusal is no longer logged twice: FastMCP already logs the raised error. --- redisvl/mcp/auth.py | 84 +++++++------ .../test_mcp/test_auth_claim_injection.py | 110 +++++++++++------- 2 files changed, 116 insertions(+), 78 deletions(-) diff --git a/redisvl/mcp/auth.py b/redisvl/mcp/auth.py index b08b98bb..d2c8e462 100644 --- a/redisvl/mcp/auth.py +++ b/redisvl/mcp/auth.py @@ -16,6 +16,7 @@ """ import logging +import unicodedata from collections.abc import Sequence from pathlib import Path from typing import Any, Protocol @@ -266,11 +267,12 @@ def _injection_refused(claim: str, tool_name: str, reason: str) -> RedisVLMCPErr what turns a misspelled claim name from an opaque permanent failure into a one-line diagnosis. """ + # Not logged here: FastMCP logs every exception a tool raises, so a second + # line would only duplicate it. message = ( f"Tool '{tool_name}' requires the '{claim}' claim to scope every query, " f"and {reason}; refusing to run an unscoped query" ) - logger.warning("%s", message) return RedisVLMCPError( message, code=MCPErrorCode.FORBIDDEN, @@ -326,41 +328,39 @@ def resolve_injected_claim(claim: str, *, tool_name: str) -> str: # be meaningful in a tag equality, so refuse it instead of guessing. raise _injection_refused(claim, tool_name, "it is padded with whitespace") - if "|" in value: - # Defence in depth. `Tag` escapes `|` inside a single value since - # 0.27.1, so this is unreachable through the `==` path today; it is the - # backstop for that character class changing again. `test_injected_pipe - # _cannot_union_across_tenants` pins the rendering property itself. - raise _injection_refused(claim, tool_name, "it contains the union operator '|'") + if any(_reads_as_separator(character) for character in value): + # Measured on Redis 8.4: inside tag braces the query parser reads every + # control character and the backtick as a term separator, the controls + # even when escaped. So `\x01acme` and `acme` plus a backtick match the + # tenant `acme`, and `acme\tcorp` matches `acme corp`. Escaping cannot + # neutralise the controls, so the value is refused rather than rendered. + # + # `|` is deliberately allowed. `Tag` escapes it inside a single value, + # and Redis matches the escaped form literally, so an identifier such as + # an Auth0 `sub` (`auth0|64f1c2`) scopes to exactly its own documents. + raise _injection_refused( + claim, tool_name, "it contains a character the query parser splits on" + ) return value -def _injected_tag_field(schema: IndexSchema, field_name: str) -> Any: - """Return the schema field an injection entry names, or ``None``. - - ``None`` means "not usable for injection" for any reason -- absent, wrong - type, or unindexed. Callers decide whether that is a startup failure or a - request-time refusal. - """ - field = schema.fields.get(field_name) - if field is None or field.type != "tag": - return None - if getattr(field.attrs, "no_index", False): - return None - return field +def _reads_as_separator(character: str) -> bool: + """Report whether RediSearch splits a tag query term on this character.""" + return character == "`" or unicodedata.category(character) == "Cc" def build_injected_filter( inject_specs: Sequence[InjectSpec], - schema: IndexSchema, *, tool_name: str, ) -> FilterExpression: """Build the tenant-scoping expression for the current request. Entries AND together, and any one unusable claim refuses the whole request - rather than narrowing by the entries that did resolve. + rather than narrowing by the entries that did resolve. Which fields may be + injected is settled at startup by :func:`validate_inject_against_schema`, + which re-runs against the freshly inspected schema on every start. """ if not inject_specs: # Callers guard on truthiness before reaching here; an empty list would @@ -379,18 +379,6 @@ def build_injected_filter( for spec in inject_specs: value = resolve_injected_claim(spec.claim, tool_name=tool_name) - if _injected_tag_field(schema, spec.field) is None: - # Startup validation already rejected this, so reaching it means the - # bound schema changed under a registered tool. Re-checking costs a - # dict lookup and keeps the guarantee tied to the schema actually in - # force rather than to the one present at registration. - raise _injection_refused( - spec.claim, - tool_name, - f"its target field '{spec.field}' is no longer an indexed tag " - "field on the bound index", - ) - clause = Tag(spec.field) == value if is_match_all_filter(clause): # The check has to run here, on the clause alone. An intersection @@ -436,6 +424,9 @@ def validate_inject_against_schema( binding has been inspected at startup. Without it, a profile would register cleanly and then refuse every request -- or, worse for a `no_index` field, return nothing and look like correct scoping. + + A field that is usable but folds case only warns: see + :func:`_warn_on_case_folding`. """ field_names = ", ".join(sorted(schema.field_names)) @@ -461,3 +452,28 @@ def validate_inject_against_schema( "call would return an empty result set that looks like correct " "scoping" ) + _warn_on_case_folding(spec, field, profile_name=profile_name) + + +def _warn_on_case_folding(spec: InjectSpec, field: Any, *, profile_name: str) -> None: + """Warn when an injected tag field folds case, which can merge tenants. + + A tag field is case-insensitive unless declared ``CASESENSITIVE``, and the + folding is Unicode-wide rather than ASCII-only: measured on Redis 8.4, + ``Acme`` and ``acme`` merge, as do a Kelvin sign and ``K``, and composed and + decomposed accented forms. A warning rather than a failure, because an + identity provider that only ever emits one canonical form is safe, and only + the operator knows whether theirs does. The value is deliberately not + normalized here: that would hide a collision rather than surface it. + """ + if getattr(field.attrs, "case_sensitive", False): + return + logger.warning( + "custom_tools '%s' injects claim '%s' into tag field '%s', which is not " + "CASESENSITIVE, so tenant identifiers that differ only in case match the " + "same documents. Declare the field CASESENSITIVE unless the identity " + "provider guarantees a single case.", + profile_name, + spec.claim, + spec.field, + ) diff --git a/tests/unit/test_mcp/test_auth_claim_injection.py b/tests/unit/test_mcp/test_auth_claim_injection.py index bda3dac4..1523427c 100644 --- a/tests/unit/test_mcp/test_auth_claim_injection.py +++ b/tests/unit/test_mcp/test_auth_claim_injection.py @@ -107,7 +107,13 @@ def test_a_tokenless_request_is_refused_rather_than_passed_through(monkeypatch): pytest.param({"org_id": {"id": "acme"}}, id="dict"), pytest.param({"org_id": True}, id="bool"), pytest.param({"org_id": 42}, id="int"), - pytest.param({"org_id": "acme|victim"}, id="pipe"), + # Each of these matched a different tenant on Redis 8.4: the query + # parser splits a tag term on control characters and the backtick. + pytest.param({"org_id": "\x01acme"}, id="leading-control"), + pytest.param({"org_id": "acme\x7f"}, id="trailing-delete"), + pytest.param({"org_id": "acme\tcorp"}, id="interior-tab"), + pytest.param({"org_id": "acme\x00"}, id="nul"), + pytest.param({"org_id": "acme`"}, id="backtick"), ], ) def test_an_unusable_claim_is_refused(monkeypatch, claims): @@ -135,7 +141,7 @@ def test_the_refusal_names_the_claim_and_the_tool(monkeypatch): def test_one_entry_scopes_the_query_to_the_claim(monkeypatch): _token(monkeypatch, {"org_id": "acme"}) expression = build_injected_filter( - [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" + [_Spec("org_id", "org_id")], tool_name="search_kb" ) assert str(expression) == "@org_id:{acme}" @@ -144,7 +150,6 @@ def test_two_entries_and_together(monkeypatch): _token(monkeypatch, {"org_id": "acme", "region": "eu"}) expression = build_injected_filter( [_Spec("org_id", "org_id"), _Spec("region", "region")], - _schema(), tool_name="search_kb", ) rendered = str(expression) @@ -160,7 +165,6 @@ def test_one_unusable_claim_refuses_the_whole_request(monkeypatch): with pytest.raises(RedisVLMCPError) as exc: build_injected_filter( [_Spec("org_id", "org_id"), _Spec("region", "region")], - _schema(), tool_name="search_kb", ) assert exc.value.code == MCPErrorCode.FORBIDDEN @@ -169,22 +173,10 @@ def test_one_unusable_claim_refuses_the_whole_request(monkeypatch): def test_no_entries_refuses_rather_than_matching_everything(monkeypatch): _token(monkeypatch, {"org_id": "acme"}) with pytest.raises(RedisVLMCPError) as exc: - build_injected_filter([], _schema(), tool_name="search_kb") + build_injected_filter([], tool_name="search_kb") assert exc.value.code == MCPErrorCode.INTERNAL_ERROR -def test_a_field_that_stopped_being_an_indexed_tag_is_refused(monkeypatch): - # Startup validation rejects these, so reaching here means the bound schema - # changed under a tool that was already registered. - _token(monkeypatch, {"org_id": "acme"}) - for field in ("rating", "shadow", "absent"): - with pytest.raises(RedisVLMCPError) as exc: - build_injected_filter( - [_Spec(field, "org_id")], _schema(), tool_name="search_kb" - ) - assert exc.value.code == MCPErrorCode.FORBIDDEN - - def test_an_injected_clause_cannot_be_elided_by_the_intersection(monkeypatch): # `Tag(f) == ""` renders as the match-all `*`, and an intersection drops a # `*` operand -- so `locked & injected` would render as `locked` alone with @@ -198,9 +190,7 @@ def test_an_injected_clause_cannot_be_elided_by_the_intersection(monkeypatch): _token(monkeypatch, {"org_id": ""}) with pytest.raises(RedisVLMCPError): - build_injected_filter( - [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" - ) + build_injected_filter([_Spec("org_id", "org_id")], tool_name="search_kb") def test_a_clause_that_renders_match_all_is_refused_even_so(monkeypatch): @@ -217,41 +207,55 @@ def test_a_clause_that_renders_match_all_is_refused_even_so(monkeypatch): lambda claim, *, tool_name: "", ) with pytest.raises(RedisVLMCPError) as exc: - build_injected_filter( - [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" - ) + build_injected_filter([_Spec("org_id", "org_id")], tool_name="search_kb") assert exc.value.code == MCPErrorCode.FORBIDDEN assert "matches every document" in str(exc.value) -def test_injected_pipe_cannot_union_across_tenants(monkeypatch): - """Canary for the rendering property, not for a character class. +def test_a_pipe_in_a_claim_is_content_not_a_union(monkeypatch): + """Canary for the rendering property the guarantee rests on. - ``Tag`` escapes ``|`` inside a single value as of 0.27.1. That character - class has already changed once, so this asserts the property the guarantee - actually rests on -- an injected value cannot produce a clause that matches - a tenant other than the one named -- rather than asserting which characters - are in which set. + An identifier such as an Auth0 ``sub`` carries ``|``. ``Tag`` escapes it + inside a single value, so the claim scopes to exactly its own documents; + the integration suite proves Redis matches the escaped form literally. A + list claim is the shape that genuinely unions, and is refused on type. """ from redisvl.query.filter import Tag - # Property one: a scalar value carrying `|` is escaped, so it is content - # rather than structure. - assert str(Tag("org_id") == "acme|victim") == "@org_id:{acme\\|victim}" + _token(monkeypatch, {"org_id": "auth0|64f1c2"}) + expression = build_injected_filter( + [_Spec("org_id", "org_id")], tool_name="search_kb" + ) + assert str(expression) == "@org_id:{auth0\\|64f1c2}" - # Property two: a list value is *not* escaped into one value -- it renders - # a genuine union. This is why the claim reader rejects on type, and why a - # character scan of the rendered output would not close it. + # The structure a list produces is a real union, indistinguishable from a + # legitimate one by inspecting the output -- hence the type check. assert str(Tag("org_id") == ["acme", "victim"]) == "@org_id:{acme|victim}" + _token(monkeypatch, {"org_id": ["acme", "victim"]}) + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter([_Spec("org_id", "org_id")], tool_name="search_kb") + assert exc.value.code == MCPErrorCode.FORBIDDEN - # So neither shape can reach a query through injection. - for value in ("acme|victim", ["acme", "victim"]): - _token(monkeypatch, {"org_id": value}) - with pytest.raises(RedisVLMCPError) as exc: - build_injected_filter( - [_Spec("org_id", "org_id")], _schema(), tool_name="search_kb" - ) - assert exc.value.code == MCPErrorCode.FORBIDDEN + +def test_a_combination_that_loses_a_clause_is_refused(monkeypatch): + # The last check before a query runs. `format_expression` elides a `*` + # operand by design; if any future change made it drop a real one, the + # query would be scoped by fewer clauses than were configured. + from redisvl.query.filter import FilterExpression + + _token(monkeypatch, {"org_id": "acme", "region": "eu"}) + monkeypatch.setattr( + FilterExpression, + "format_expression", + staticmethod(lambda left, right, operator_str: str(left)), + ) + with pytest.raises(RedisVLMCPError) as exc: + build_injected_filter( + [_Spec("org_id", "org_id"), _Spec("region", "region")], + tool_name="search_kb", + ) + assert exc.value.code == MCPErrorCode.INTERNAL_ERROR + assert "lost the clauses" in str(exc.value) # --- validate_inject_against_schema ---------------------------------------- @@ -282,3 +286,21 @@ def test_an_unusable_injected_field_fails_startup(field, expected): ) assert expected in str(exc.value) assert "search_kb" in str(exc.value) + + +def test_validation_warns_when_an_injected_tag_folds_case(caplog): + schema = _schema() + schema.fields["org_id"].attrs.case_sensitive = False + with caplog.at_level("WARNING", logger="redisvl.mcp.auth"): + validate_inject_against_schema( + [_Spec("org_id", "org_id")], schema, profile_name="search_kb" + ) + assert any("not CASESENSITIVE" in record.message for record in caplog.records) + + +def test_validation_does_not_warn_for_a_case_sensitive_tag(caplog): + with caplog.at_level("WARNING", logger="redisvl.mcp.auth"): + validate_inject_against_schema( + [_Spec("org_id", "org_id")], _schema(), profile_name="search_kb" + ) + assert not any("CASESENSITIVE" in record.message for record in caplog.records) From 57bc6e8db81f1934ec6340d1ccb0d7bf973a904a Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 1 Oct 2026 17:13:48 +0200 Subject: [PATCH 4/5] fix(mcp): refuse an unscoped route to a tenant-scoped index Injection isolates a tool, but tenant data lives in an index, and every tool passes the same read-scope gate. The six-perspective review found that an injecting profile's index stayed reachable without the tenant clause: an enabled `search-records` read every tenant, an enabled `upsert-records` let one tenant retag another's document as its own (hash storage merges a partial record into the existing key), and a second profile without `lock.inject` on the same index was simply unscoped. Only the docs warned. Startup now refuses that surface, naming each route it found. Writes are no route to a read-only index, server-wide or per binding, since those refuse per call; an unscoped profile on a different index is no route either. The check depends only on config, so it runs beside the token check, before any binding connects. This is configuration that voids the guarantee it declares, which is why it fails rather than warns. Merged in from the security core: the claim reader refuses control characters and the backtick and now accepts `|`, and `build_injected_filter` no longer takes a schema. The profile wrapper follows suit, gates injection on `is not None` so an empty list would reach the builder's refusal rather than skip it, and drops its case-folding warning, which moved beside the other schema checks. Tests: a wire-level test drives a real FastMCP server through an in-memory client and shows a call naming the injected field is rejected before the wrapper runs, replacing an assertion that could never fail. The integration suite keeps what depends on real Redis -- including a live check that an Auth0-shaped claim `auth0|64f1c2` matches only its own document -- and drops the cases the unit matrices already own. Docs: the claim that the model is never told the field exists was false, since results and `list-indexes` show it, and is corrected. The threat model gains the write path, separator splitting, index-time trimming, JSON arrays and Unicode-wide case folding; the stdio refusal is qualified to the entrypoints that record a transport; the worked example disables `upsert-records`, without which it would no longer start. --- docs/concepts/mcp.md | 19 +- docs/user_guide/how_to_guides/mcp.md | 17 +- redisvl/mcp/server.py | 55 +++++- redisvl/mcp/tools/profiles.py | 40 +--- .../test_mcp/test_claim_injection.py | 173 +++++++++++------- tests/unit/test_mcp/test_profiles_unit.py | 46 +++-- tests/unit/test_mcp/test_server_unit.py | 106 +++++++++++ 7 files changed, 327 insertions(+), 129 deletions(-) diff --git a/docs/concepts/mcp.md b/docs/concepts/mcp.md index 8557f56c..a458a69c 100644 --- a/docs/concepts/mcp.md +++ b/docs/concepts/mcp.md @@ -216,7 +216,7 @@ custom_tools: `field` and `claim` are independent names: `field` is what the index schema calls the tenant column, and `claim` is what the identity provider calls it. A token carrying `"https://acme.example/org": "acme"` makes every query from that caller run as `@org_id:{acme} AND `. -The model cannot see or set the injected value. The field is absent from the tool's input schema, and it is left out of the field hints appended to the tool description, so the model is not told the field exists. If the model filters on it anyway, its clause ANDs with the injected one: it can narrow within its own tenant, and naming another tenant matches nothing. The rest of the profile works as before, so a static `lock.filter` on another field and a model-supplied filter both still apply. +The model cannot set the injected value. The field is absent from the tool's input schema, a call that names it is rejected, and it is left out of the field hints appended to the tool description. It is not hidden outright: unless `lock.return_fields` excludes it, results carry the field, and `list-indexes` describes the whole schema. What the model sees there is only ever its own tenant. If the model filters on the field, its clause ANDs with the injected one, so it can narrow within its own tenant and naming another tenant matches nothing. The rest of the profile works as before, so a static `lock.filter` on another field and a model-supplied filter both still apply. #### What Counts as a Usable Claim @@ -228,7 +228,10 @@ The claim must be a single, non-empty string. Anything else refuses the request | `null` or `""` | An empty tag value would drop the tenant clause from the query entirely. | | A list, such as `["acme", "victim"]` | It would render as a union, `@org_id:{acme\|victim}`, which spans both tenants. | | An object, number or boolean | It is not a tenant identifier. | -| Padded with whitespace, or containing `\|` | Neither can be a real tenant identifier, and refusing is safer than guessing. | +| Padded with whitespace | It cannot be a real tenant identifier, and refusing is safer than guessing which tenant was meant. | +| Containing a control character or a backtick | The query parser splits a tag term on these, so a value such as `acme` followed by a control character matches the tenant `acme`. | + +A `|` inside a single string is accepted. It is escaped, so an identifier such as the Auth0 subject `auth0|64f1c2` matches only its own documents. A list is refused because of its type, not because of what it renders as. The union it produces is indistinguishable from one a caller could legitimately ask for, so no inspection of the finished query could catch it. @@ -236,10 +239,11 @@ With several `inject` entries, every entry ANDs into the query, and one unusable #### What Fails at Startup -Injection is checked at startup wherever the configuration alone can show it would not work: +Injection is checked at startup wherever the configuration alone can show it would not hold: - authentication is not enabled, on any transport, including an unauthenticated loopback HTTP bind and any `--allow-unauthenticated` bind; -- authentication is configured but the server runs over `stdio`, which is never authenticated; +- authentication is configured but the server runs over `stdio`, which is never authenticated (checked when the server starts through `rvl mcp` or `run_async`; an embedder that calls `startup()` directly is not, and every call is then refused at request time instead); +- the index is also reachable without the tenant scope: through `search-records`, through `upsert-records` unless the index is read-only, or through another custom tool on the same index that does not inject; - the injected field is absent from the bound index, is not a tag field, or is declared `NOINDEX`; - an `inject` list is empty, names one field twice, or names a field that `lock.filter` also constrains; - `required` is anything but `true`, or `from` is anything but `claim`. @@ -250,12 +254,13 @@ The tool set registers once per process. If a restart reloads a configuration th #### Threat Model -The guarantee is precise: a client presenting a validly signed token cannot make the model widen or escape the tenant scope carried in that token. The trust boundary is the identity provider, not the MCP client, so the guarantee holds only while these hold: +The guarantee is narrow: a client presenting a validly signed token cannot make the model widen or escape the tenant scope carried in that token. The trust boundary is the identity provider, not the MCP client, so the guarantee holds only while these hold: - The token is genuinely verified. Use a real signing key and an asymmetric algorithm. The server refuses to start an injecting profile without authentication, but it does not check which algorithm you configured. - The identity provider assigns the claim. If a tenant can mint its own token, or set the claim itself, nothing here stops it reading another tenant's data. -- Every document carries its tenant. The tenant field must be stamped on each document and indexed as a tag. A document without the field matches no tenant, so it is invisible rather than shared. -- Tenant identifiers differ by more than case. Tag fields fold case unless declared `CASESENSITIVE`, so `Acme` and `acme` would be one tenant. The server warns at startup when an injected field is not case-sensitive. +- Only trusted ingestion writes the index. The server refuses `upsert-records` on a writable scoped index, because a write can retag another tenant's document as the writer's own. Whatever loads documents outside the server is inside the trust boundary. +- Every document carries exactly its tenant. Stamp the tenant field on each document, indexed as a tag, with the identifier exactly as the identity provider emits it. Redis normalises the stored value, not the claim: it splits it on the field's separator (`,` by default), so a document stamped `acme,victim` belongs to both tenants; it trims surrounding whitespace; and on JSON storage it indexes every element of an array. A document without the field matches no tenant, so it is invisible rather than shared. +- Tenant identifiers differ by more than case. Tag fields fold case unless declared `CASESENSITIVE`, and the folding is Unicode-wide: `Acme` and `acme` are one tenant, and so are a Kelvin sign and `K`, or composed and decomposed forms of an accented letter. The server warns at startup when an injected field is not case-sensitive. Where tenants share one index, the injected filter is the only isolation boundary. There is no Redis ACL or keyspace separation behind it, so a defect in filter combination or claim validation is a full cross-tenant read. If you need defence in depth, separate tenants at the Redis layer as well. diff --git a/docs/user_guide/how_to_guides/mcp.md b/docs/user_guide/how_to_guides/mcp.md index c2e31744..a142bef2 100644 --- a/docs/user_guide/how_to_guides/mcp.md +++ b/docs/user_guide/how_to_guides/mcp.md @@ -366,7 +366,8 @@ Then point a profile at it: server: redis_url: redis://localhost:6379 builtin_tools: - search-records: disabled # otherwise the unscoped built-in stays available + search-records: disabled # both built-ins reach the index unscoped, + upsert-records: disabled # so the server refuses to start with either on auth: type: jwt jwks_uri: ${MCP_JWKS_URI} @@ -402,15 +403,15 @@ rvl mcp --config /path/to/mcp_config.yaml --transport streamable-http What the client sees for `search-customer-kb`: -- `query` (required), `limit`, `offset`, `filter` and `return_fields`, with no argument for `org_id`. +- `query` (required), `limit`, `offset`, `filter` and `return_fields`, with no argument for `org_id`. A call that passes `org_id` anyway is rejected. - A description whose field hints list `content` but not `org_id`. -- Results from its own tenant only. A `filter` naming `org_id` ANDs with the injected value, so naming another tenant returns nothing. +- Results from its own tenant only. A `filter` naming `org_id` ANDs with the injected value, so naming another tenant returns nothing. Results still carry `org_id`, always with the caller's own value, unless you lock `return_fields` to leave it out. -Disable `search-records`, as above, unless every tenant may read the whole index. The built-in has no injected scope, so leaving it enabled hands every caller an unscoped way round the profile. +The server refuses to start while anything else can reach the same index without the tenant scope: `search-records`, `upsert-records` unless the index is `read_only`, or another custom tool on that index without `lock.inject`. Each would hand every caller a way round the profile, and a write could retag another tenant's document as the writer's own. Ingest documents outside the server, stamping `org_id` exactly as the identity provider emits it. Listing the tenant claim under `required_claims` makes the verifier reject a token without it before any tool runs. That checks presence only: the profile still validates the value on every call, and refuses a missing, empty, list-valued or otherwise unusable claim with a `forbidden` error before any query runs. -The server refuses to start an injecting profile without authentication, over `stdio`, or on a field the index does not hold as an indexed tag. For what the guarantee covers and what it rests on, read the threat model in {doc}`/concepts/mcp`. +The server refuses to start an injecting profile without authentication, over `stdio`, or on a field the index does not hold as an indexed tag. The `stdio` check applies when the server starts through `rvl mcp` or `run_async`. For what the guarantee covers and what it rests on, read the threat model in {doc}`/concepts/mcp`. ## Tool Contracts @@ -798,6 +799,10 @@ Some hybrid params depend on native hybrid support in Redis and redis-py. If you A profile with `lock.inject` refuses to start when authentication is not enabled, or when the server runs over `stdio`, because neither can supply a verified token. Configure `server.auth` and serve over `sse` or `streamable-http`. The check runs before the server connects to Redis, so it reports even when Redis is unreachable. +### Claim Injection Refuses an Unscoped Route + +A profile with `lock.inject` refuses to start while `search-records`, `upsert-records` on a writable index, or another custom tool without `lock.inject` can reach the same index. The error lists each route it found. Disable the built-ins under `server.builtin_tools`, mark the index `read_only`, or add the same `lock.inject` to the other custom tools. + ### Claim Injection Fails Every Request With `forbidden` -The token is verified but its claim is unusable: missing, empty, padded with whitespace, or not a single string. The error names the claim and the tool. A misspelled `claim` name in the config is the usual cause, since a JWT claim name such as `https://acme.example/org` must match exactly. +The token is verified but its claim is unusable: missing, empty, padded with whitespace, not a single string, or containing a control character or backtick. The error names the claim and the tool. A misspelled `claim` name in the config is the usual cause, since a JWT claim name such as `https://acme.example/org` must match exactly. diff --git a/redisvl/mcp/server.py b/redisvl/mcp/server.py index 1ce44090..8cc29984 100644 --- a/redisvl/mcp/server.py +++ b/redisvl/mcp/server.py @@ -347,6 +347,58 @@ def _verify_injection_has_a_token(self, profiles: Any) -> None: "streamable-http) so the configured auth applies." ) + def _verify_no_unscoped_route_to_injected_indexes(self, config: Any) -> None: + """Refuse a tool surface that reaches a tenant-scoped index unscoped. + + Injection isolates a tool, but tenant data lives in an index, and every + tool passes the same read scope gate. So any other route to that index + -- the generic search, a write, or a profile without the injection -- + hands every caller the data the profile was meant to fence off, and a + write can retag another tenant's document as the writer's own. That is + configuration which voids the guarantee it declares, not a redundant + surface, so it is refused rather than warned about. + """ + injected = sorted( + { + config.resolved_profile_index(profile) + for profile in config.custom_tools + if profile.lock.inject is not None + } + ) + if not injected: + return + + routes: list[str] = [] + if config.server.builtin_tool_enabled("search-records"): + routes.append("search-records") + if config.server.builtin_tool_enabled("upsert-records"): + writable = [ + binding_id + for binding_id in injected + if not ( + self.mcp_settings.read_only or config.indexes[binding_id].read_only + ) + ] + if writable: + routes.append("upsert-records") + routes.extend( + f"custom tool '{profile.name}'" + for profile in config.custom_tools + if profile.lock.inject is None + and config.resolved_profile_index(profile) in injected + ) + if not routes: + return + + raise ValueError( + f"Index {', '.join(repr(i) for i in injected)} is scoped by an " + "injected token claim, but is also reachable without that scope " + f"through {', '.join(routes)}. Each of those bypasses the tenant " + "filter. Disable the built-ins with server.builtin_tools (for " + "example 'search-records: disabled'), mark the index read_only to " + "stop writes, or give the other custom tools the same lock.inject." + ) + @staticmethod def _tool_surface_fingerprint(config: Any) -> str: """Summarize the config that a registered tool set baked in.""" @@ -619,9 +671,10 @@ async def _initialize_runtime_resources(self) -> None: """Load config and initialize every configured binding independently.""" self.config = load_mcp_config(self._config_path) self._verify_auth_not_stale() - # Before any binding connects: this depends only on the loaded config, + # Before any binding connects: both depend only on the loaded config, # so an unworkable one should not first need a reachable Redis. self._verify_injection_has_a_token(self.config.custom_tools) + self._verify_no_unscoped_route_to_injected_indexes(self.config) # The semaphore is a single process-wide concurrency ceiling shared by # all bindings; take the max across bindings. This means the most # permissive binding sets the cap — e.g. five bindings each configured diff --git a/redisvl/mcp/tools/profiles.py b/redisvl/mcp/tools/profiles.py index 0731a2f0..674ef770 100644 --- a/redisvl/mcp/tools/profiles.py +++ b/redisvl/mcp/tools/profiles.py @@ -12,7 +12,6 @@ """ import inspect -import logging from typing import Annotated, Any, Optional from pydantic import Field @@ -33,8 +32,6 @@ from redisvl.query.filter import FilterExpression from redisvl.schema import IndexSchema -logger = logging.getLogger(__name__) - # Ordered so the generated signature reads like the built-in's: required query # first, then the optional narrowing arguments. Real type objects rather than # string annotations, so schema generation never depends on resolving a forward @@ -148,32 +145,6 @@ def validate_profile_against_schema( validate_inject_against_schema( profile.lock.inject, schema, profile_name=profile.name ) - _warn_on_case_insensitive_inject(profile, schema) - - -def _warn_on_case_insensitive_inject( - profile: MCPCustomToolConfig, schema: IndexSchema -) -> None: - """Warn when an injected tag field folds case, which merges tenants. - - A tag field is case-insensitive unless declared ``CASESENSITIVE``, so the - tenants ``Acme`` and ``acme`` are one tenant to Redis. A warning rather than - a failure: an IdP that only ever emits lower-case identifiers is safe, and - only the operator knows whether theirs does. The value is deliberately not - normalized here -- that would hide the collision rather than surface it. - """ - for entry in profile.lock.inject or (): - field = schema.fields.get(entry.field) - if field is not None and not getattr(field.attrs, "case_sensitive", False): - logger.warning( - "custom_tools '%s' injects claim '%s' into tag field '%s', which " - "is not CASESENSITIVE, so tenant identifiers differing only in " - "case match the same documents. Declare the field CASESENSITIVE " - "unless the identity provider guarantees a single case.", - profile.name, - entry.claim, - entry.field, - ) def _validate_locked_exists_fields( @@ -262,7 +233,10 @@ def register_profile_tool( value belongs to the caller's token rather than to the configuration. """ locked_filter = resolve_locked_filter(profile, schema) - inject_specs = profile.lock.inject or [] + # `None` means "does not inject". Gated on that rather than truthiness, so + # an empty list -- which config validation rejects -- would still reach the + # builder's own refusal instead of silently skipping the tenant clause. + inject_specs = profile.lock.inject locked_return_fields = profile.lock.return_fields limit_cap = profile.param_max("limit") exposes_limit = profile.param_exposed("limit") @@ -280,10 +254,8 @@ async def profile_tool(**kwargs: Any) -> dict[str, Any]: # locked side forces the caller's filter through the escape backstop. # A fresh local, so the registration-time expression is never rebound. effective_locked = locked_filter - if inject_specs: - injected = build_injected_filter( - inject_specs, schema, tool_name=profile.name - ) + if inject_specs is not None: + injected = build_injected_filter(inject_specs, tool_name=profile.name) effective_locked = ( injected if effective_locked is None else effective_locked & injected ) diff --git a/tests/integration/test_mcp/test_claim_injection.py b/tests/integration/test_mcp/test_claim_injection.py index c1cd26f1..ded8d09f 100644 --- a/tests/integration/test_mcp/test_claim_injection.py +++ b/tests/integration/test_mcp/test_claim_injection.py @@ -1,9 +1,13 @@ """End-to-end tests for claim-injected tenant scoping against a real Redis. -Two tenants share one index, separated by an ``org_id`` tag. Tokens are real -RS256 JWTs minted with FastMCP's ``RSAKeyPair`` and verified by the server's own +Tenants share one index, separated by an ``org_id`` tag. Tokens are real RS256 +JWTs minted with FastMCP's ``RSAKeyPair`` and verified by the server's own configured verifier, so the claims the profile reads are the ones a verified request would carry -- only the request-context lookup is substituted. + +The unit suites own the exhaustive matrices. What lives here is what depends on +real Redis behaviour: how the merged query actually matches, how a value Redis +parses specially is handled, and what startup does against an inspected index. """ from pathlib import Path @@ -33,6 +37,14 @@ "lock": {"inject": [{"field": "org_id", "from": "claim", "claim": ORG_CLAIM}]}, } +# Both built-ins reach the index without the tenant scope, so a server with an +# injecting profile refuses to start unless they are off. +_SCOPED_BUILTINS = {"search-records": "disabled", "upsert-records": "disabled"} + + +def _doc(doc_id: str, org_id: str, content: str = "refund policy") -> dict: + return {"id": doc_id, "content": content, "org_id": org_id, "category": "billing"} + @pytest.fixture(scope="module") def key() -> RSAKeyPair: @@ -52,7 +64,6 @@ async def tenant_index(async_client, worker_id): {"name": "content", "type": "text"}, {"name": "org_id", "type": "tag", "attrs": {"case_sensitive": True}}, {"name": "category", "type": "tag"}, - {"name": "notes", "type": "text"}, # NOINDEX is only accepted alongside SORTABLE. { "name": "shadow", @@ -65,33 +76,18 @@ async def tenant_index(async_client, worker_id): index = AsyncSearchIndex(schema=schema, redis_client=async_client) await index.create(overwrite=True, drop=True) # Identical content across tenants, so only the tenant clause can separate - # them -- a query that leaks would match all four. + # them: a query that leaked would match every document below. await index.load( [ - { - "id": "a1", - "content": "refund policy", - "org_id": "acme", - "category": "billing", - }, - { - "id": "a2", - "content": "refund window", - "org_id": "acme", - "category": "billing", - }, - { - "id": "v1", - "content": "refund policy", - "org_id": "victim", - "category": "billing", - }, - { - "id": "v2", - "content": "refund window", - "org_id": "victim", - "category": "billing", - }, + _doc("a1", "acme"), + _doc("a2", "acme", "refund window"), + _doc("v1", "victim"), + _doc("v2", "victim", "refund window"), + # An Auth0-shaped subject beside the two halves it must not be + # confused with, so a `|` claim that unioned would match three. + _doc("p1", "auth0|64f1c2"), + _doc("p2", "auth0"), + _doc("p3", "64f1c2"), ], id_field="id", ) @@ -101,8 +97,15 @@ async def tenant_index(async_client, worker_id): @pytest.fixture def config_path(tmp_path: Path, redis_url: str, key: RSAKeyPair): - def factory(redis_name: str, custom_tools: list[dict], *, auth: bool = True) -> str: - server: dict = {"redis_url": redis_url} + def factory( + redis_name: str, + custom_tools: list[dict], + *, + auth: bool = True, + builtin_tools: dict = _SCOPED_BUILTINS, + read_only: bool = False, + ) -> str: + server: dict = {"redis_url": redis_url, "builtin_tools": builtin_tools} if auth: server["auth"] = { "type": "jwt", @@ -115,6 +118,7 @@ def factory(redis_name: str, custom_tools: list[dict], *, auth: bool = True) -> "indexes": { "kb": { "redis_name": redis_name, + "read_only": read_only, "search": {"type": "fulltext"}, "runtime": {"text_field_name": "content"}, } @@ -171,8 +175,8 @@ async def _search(server, **kwargs): return await tool.fn(query="refund", **kwargs) -def _orgs(result) -> set: - return {hit["record"]["org_id"] for hit in result["results"]} +def _orgs(result) -> list: + return sorted(hit["record"]["org_id"] for hit in result["results"]) async def test_a_tenants_token_returns_only_that_tenants_documents( @@ -181,17 +185,25 @@ async def test_a_tenants_token_returns_only_that_tenants_documents( server = await started() await _as_caller(monkeypatch, server, key, {ORG_CLAIM: "acme"}) - result = await _search(server) + assert _orgs(await _search(server)) == ["acme", "acme"] + + +async def test_a_pipe_in_the_claim_is_one_identifier_not_a_union( + started, key, monkeypatch +): + # Redis matches the escaped `\|` literally, so the claim reaches exactly its + # own document and neither of the halves a union would also have matched. + server = await started() + await _as_caller(monkeypatch, server, key, {ORG_CLAIM: "auth0|64f1c2"}) - assert len(result["results"]) == 2 - assert _orgs(result) == {"acme"} + assert _orgs(await _search(server)) == ["auth0|64f1c2"] @pytest.mark.parametrize( "caller_filter, expected", [ # Naming the other tenant ANDs with the injected one: nothing matches. - pytest.param({"field": "org_id", "op": "eq", "value": "victim"}, 0, id="swap"), + pytest.param({"field": "org_id", "op": "eq", "value": "victim"}, [], id="swap"), # The `or` stays nested inside the AND, so it narrows within acme rather # than hoisting to the top level and reaching victim's documents. pytest.param( @@ -201,12 +213,12 @@ async def test_a_tenants_token_returns_only_that_tenants_documents( {"field": "category", "op": "eq", "value": "billing"}, ] }, - 2, + ["acme", "acme"], id="or-widening", ), pytest.param( {"not": {"field": "org_id", "op": "eq", "value": "acme"}}, - 0, + [], id="negation", ), ], @@ -217,21 +229,15 @@ async def test_a_caller_filter_cannot_widen_or_escape_the_tenant( server = await started() await _as_caller(monkeypatch, server, key, {ORG_CLAIM: "acme"}) - result = await _search(server, filter=caller_filter) + # Exact, not merely "no victim rows": a subset check alone would pass on an + # empty result for the wrong reason. + assert _orgs(await _search(server, filter=caller_filter)) == expected - # An exact count, not just "no victim rows": a subset check alone would - # pass on an empty result for the wrong reason. - assert len(result["results"]) == expected - assert _orgs(result) <= {"acme"} - -async def test_the_injected_field_is_absent_from_the_advertised_schema(started): +async def test_the_description_does_not_name_the_injected_field(started): server = await started() tool = await server.get_tool(TOOL) - assert "org_id" not in tool.parameters["properties"] - assert tool.parameters["additionalProperties"] is False - # Nor is it named in the description's field hints. assert "org_id" not in tool.description assert "category(tag)" in tool.description @@ -239,12 +245,11 @@ async def test_the_injected_field_is_absent_from_the_advertised_schema(started): @pytest.mark.parametrize( "claims", [ - pytest.param({}, id="absent"), - pytest.param({ORG_CLAIM: ""}, id="empty"), - pytest.param({ORG_CLAIM: " "}, id="whitespace"), + # The shape that genuinely unions, refused on type. pytest.param({ORG_CLAIM: ["acme", "victim"]}, id="array"), - pytest.param({ORG_CLAIM: {"id": "acme"}}, id="object"), - pytest.param({ORG_CLAIM: "acme|victim"}, id="pipe"), + # Without the refusal this matched the tenant `acme` on Redis 8.4: the + # query parser splits a tag term on control characters. + pytest.param({ORG_CLAIM: "\x01acme"}, id="control-character"), ], ) async def test_an_unusable_claim_is_refused_before_any_query( @@ -271,27 +276,61 @@ async def spy(query): @pytest.mark.parametrize( - "inject_field, auth, expected", + "custom_tools, kwargs, expected", [ - pytest.param("org_id", False, "authentication is not enabled", id="auth-off"), - pytest.param("missing", True, "unknown field 'missing'", id="field-absent"), - pytest.param("notes", True, "requires a tag field", id="field-text"), - pytest.param("shadow", True, "NOINDEX", id="field-noindex"), + pytest.param( + [_PROFILE], {"auth": False}, "authentication is not enabled", id="auth-off" + ), + # Inspection has to preserve NOINDEX for this to be caught. + pytest.param( + [ + { + **_PROFILE, + "lock": { + "inject": [ + {"field": "shadow", "from": "claim", "claim": ORG_CLAIM} + ] + }, + } + ], + {}, + "NOINDEX", + id="field-noindex", + ), + pytest.param( + [_PROFILE], + {"builtin_tools": {"upsert-records": "disabled"}}, + "through search-records", + id="unscoped-search", + ), + pytest.param( + [_PROFILE], + {"builtin_tools": {"search-records": "disabled"}}, + "through upsert-records", + id="unscoped-write", + ), + pytest.param( + [_PROFILE, {"name": "open-search", "description": "Search the kb."}], + {}, + "custom tool 'open-search'", + id="unscoped-profile", + ), ], ) -async def test_startup_refuses_an_injection_profile_that_cannot_scope( - tenant_index, config_path, inject_field, auth, expected +async def test_startup_refuses_an_injection_profile_that_cannot_hold( + tenant_index, config_path, custom_tools, kwargs, expected ): - profile = { - **_PROFILE, - "lock": { - "inject": [{"field": inject_field, "from": "claim", "claim": ORG_CLAIM}] - }, - } server = RedisVLMCPServer( MCPSettings( - config=config_path(tenant_index.schema.index.name, [profile], auth=auth) + config=config_path(tenant_index.schema.index.name, custom_tools, **kwargs) ) ) with pytest.raises(ValueError, match=expected): await server.startup() + + +async def test_a_read_only_index_may_keep_upsert_enabled(started): + # Writes to a read-only binding are refused per call, so upsert-records is + # no route to it. + server = await started(builtin_tools={"search-records": "disabled"}, read_only=True) + assert await server.get_tool(TOOL) is not None diff --git a/tests/unit/test_mcp/test_profiles_unit.py b/tests/unit/test_mcp/test_profiles_unit.py index b1c125b3..bf941747 100644 --- a/tests/unit/test_mcp/test_profiles_unit.py +++ b/tests/unit/test_mcp/test_profiles_unit.py @@ -1018,20 +1018,38 @@ def test_startup_validation_rejects_injecting_into_a_non_tag_field(): validate_profile_against_schema(profile, _schema()) -def test_startup_warns_when_the_injected_tag_folds_case(caplog): - # `category` is a plain tag, so `Acme` and `acme` would be one tenant. - with caplog.at_level("WARNING", logger="redisvl.mcp.tools.profiles"): - validate_profile_against_schema( - _profile(lock={"inject": INJECT_CATEGORY}), _schema() - ) - assert any("not CASESENSITIVE" in record.message for record in caplog.records) +@pytest.mark.asyncio +async def test_a_client_cannot_pass_the_injected_field_over_the_wire(monkeypatch): + # The real `tools/call` path, on a real FastMCP server with a real client, + # rather than the wrapper called directly. FastMCP validates arguments + # against the narrowed signature, so a field the profile does not expose is + # rejected before the wrapper runs -- not dropped, and not passed through. + fastmcp = pytest.importorskip( + "fastmcp", reason="fastmcp not installed (install redisvl[mcp])" + ) + _as_caller(monkeypatch, {"org": "acme"}) + searched: list[dict[str, Any]] = [] + async def record_search(server, **kwargs): + searched.append(kwargs) + return {"results": []} -def test_startup_does_not_warn_for_a_case_sensitive_injected_tag(caplog): - schema = _schema() - schema.fields["category"].attrs.case_sensitive = True - with caplog.at_level("WARNING", logger="redisvl.mcp.tools.profiles"): - validate_profile_against_schema( - _profile(lock={"inject": INJECT_CATEGORY}), schema + monkeypatch.setattr("redisvl.mcp.tools.profiles.search_records", record_search) + mcp = fastmcp.FastMCP("tenant-test") + register_profile_tool( + mcp, _profile(lock={"inject": INJECT_CATEGORY}), "knowledge", _schema() + ) + + async with fastmcp.Client(mcp) as client: + (tool,) = await client.list_tools() + assert "category" not in tool.inputSchema["properties"] + assert tool.inputSchema["additionalProperties"] is False + + refused = await client.call_tool( + tool.name, {"query": "jam", "category": "victim"}, raise_on_error=False ) - assert not any("CASESENSITIVE" in record.message for record in caplog.records) + assert refused.is_error + assert searched == [] + + await client.call_tool(tool.name, {"query": "jam"}) + assert str(searched[0]["locked_filter"]) == "@category:{acme}" diff --git a/tests/unit/test_mcp/test_server_unit.py b/tests/unit/test_mcp/test_server_unit.py index 7c82212a..1f30c367 100644 --- a/tests/unit/test_mcp/test_server_unit.py +++ b/tests/unit/test_mcp/test_server_unit.py @@ -827,3 +827,109 @@ def test_a_changed_tool_surface_without_injection_still_only_warns(caplog): "changed since tools were registered" in record.message for record in caplog.records ) + + +# -------------------------------------------------------------------------- +# Claim injection: refusing an unscoped route to a tenant-scoped index +# -------------------------------------------------------------------------- + +_BUILTINS_OFF = {"search-records": "disabled", "upsert-records": "disabled"} + + +def _route_check(*, builtin_tools, custom_tools=None, read_only=False, indexes=None): + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server.mcp_settings = SimpleNamespace(read_only=read_only) + if indexes is None: + config = _config_with( + builtin_tools=builtin_tools, custom_tools=custom_tools or [_INJECTING] + ) + else: + config = MCPConfig.model_validate( + { + "server": { + "redis_url": "redis://localhost:6379", + "builtin_tools": builtin_tools, + }, + "indexes": { + index_id: { + "redis_name": f"{index_id}-index", + "search": {"type": "fulltext"}, + "runtime": {"text_field_name": "content"}, + } + for index_id in indexes + }, + "custom_tools": custom_tools, + } + ) + server._verify_no_unscoped_route_to_injected_indexes(config) + + +@pytest.mark.parametrize( + "builtin_tools, custom_tools, expected", + [ + pytest.param( + {"upsert-records": "disabled"}, None, "through search-records", id="search" + ), + # A write can retag another tenant's document as the writer's own. + pytest.param( + {"search-records": "disabled"}, None, "through upsert-records", id="upsert" + ), + pytest.param( + _BUILTINS_OFF, + [_INJECTING, {"name": "open-search", "description": "Search."}], + "custom tool 'open-search'", + id="unscoped-profile", + ), + ], +) +def test_an_unscoped_route_to_an_injected_index_is_refused( + builtin_tools, custom_tools, expected +): + with pytest.raises(ValueError, match=expected): + _route_check(builtin_tools=builtin_tools, custom_tools=custom_tools) + + +def test_a_fully_scoped_surface_starts(): + _route_check(builtin_tools=_BUILTINS_OFF) + + +@pytest.mark.parametrize("server_read_only", [False, True]) +def test_upsert_is_no_route_to_a_read_only_index(server_read_only): + # Server-wide read-only, or the binding's own flag, both refuse writes per + # call, so upsert-records cannot reach the index either way. + server = RedisVLMCPServer.__new__(RedisVLMCPServer) + server.mcp_settings = SimpleNamespace(read_only=server_read_only) + raw = { + "server": { + "redis_url": "redis://localhost:6379", + "builtin_tools": {"search-records": "disabled"}, + }, + "indexes": { + "knowledge": { + "redis_name": "docs-index", + "read_only": not server_read_only, + "search": {"type": "fulltext"}, + "runtime": {"text_field_name": "content"}, + } + }, + "custom_tools": [_INJECTING], + } + server._verify_no_unscoped_route_to_injected_indexes(MCPConfig.model_validate(raw)) + + +def test_an_unscoped_profile_on_another_index_is_no_route(): + _route_check( + builtin_tools=_BUILTINS_OFF, + indexes=("knowledge", "public"), + custom_tools=[ + {**_INJECTING, "index": "knowledge"}, + {"name": "public-search", "description": "Search.", "index": "public"}, + ], + ) + + +def test_a_server_without_injection_keeps_every_route(): + _route_check( + builtin_tools={}, + custom_tools=[{"name": "open-search", "description": "Search open."}], + ) From 01a6b58a02ad6bae0075f82503ce40041f453d53 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Fri, 2 Oct 2026 15:07:42 +0200 Subject: [PATCH 5/5] fix(mcp): require every tool on an injected index to inject the same scope The startup check that refuses unscoped routes to a tenant-scoped index treated any `lock.inject` as the tenant scope. A second profile on the same index injecting a different field passed, and so did one injecting the same field from a different claim. Either reads across the tenants the first profile separates: scoped by `region` alone, a caller sees every organisation's documents in its region. The scope is a property of the index, so every custom tool on it must now inject the same set of (field, claim) entries, compared as a set so entry order does not matter. A tool injecting nothing is the empty set and is caught by the same comparison, which replaces the separate "no `lock.inject`" branch. The error lists each tool on the index with the scope it injects, so the mismatch is visible in one line. --- docs/concepts/mcp.md | 2 +- docs/user_guide/how_to_guides/mcp.md | 4 +- redisvl/mcp/server.py | 74 +++++++++++++------ .../test_mcp/test_claim_injection.py | 2 +- tests/unit/test_mcp/test_server_unit.py | 60 ++++++++++++++- 5 files changed, 115 insertions(+), 27 deletions(-) diff --git a/docs/concepts/mcp.md b/docs/concepts/mcp.md index a458a69c..c227db4b 100644 --- a/docs/concepts/mcp.md +++ b/docs/concepts/mcp.md @@ -243,7 +243,7 @@ Injection is checked at startup wherever the configuration alone can show it wou - authentication is not enabled, on any transport, including an unauthenticated loopback HTTP bind and any `--allow-unauthenticated` bind; - authentication is configured but the server runs over `stdio`, which is never authenticated (checked when the server starts through `rvl mcp` or `run_async`; an embedder that calls `startup()` directly is not, and every call is then refused at request time instead); -- the index is also reachable without the tenant scope: through `search-records`, through `upsert-records` unless the index is read-only, or through another custom tool on the same index that does not inject; +- the index is also reachable without the tenant scope: through `search-records`, through `upsert-records` unless the index is read-only, or through another custom tool on the same index that does not inject exactly the same entries, since a tool scoped by another field, or by the same field from another claim, reads across the tenants this one separates; - the injected field is absent from the bound index, is not a tag field, or is declared `NOINDEX`; - an `inject` list is empty, names one field twice, or names a field that `lock.filter` also constrains; - `required` is anything but `true`, or `from` is anything but `claim`. diff --git a/docs/user_guide/how_to_guides/mcp.md b/docs/user_guide/how_to_guides/mcp.md index a142bef2..0f01f52b 100644 --- a/docs/user_guide/how_to_guides/mcp.md +++ b/docs/user_guide/how_to_guides/mcp.md @@ -407,7 +407,7 @@ What the client sees for `search-customer-kb`: - A description whose field hints list `content` but not `org_id`. - Results from its own tenant only. A `filter` naming `org_id` ANDs with the injected value, so naming another tenant returns nothing. Results still carry `org_id`, always with the caller's own value, unless you lock `return_fields` to leave it out. -The server refuses to start while anything else can reach the same index without the tenant scope: `search-records`, `upsert-records` unless the index is `read_only`, or another custom tool on that index without `lock.inject`. Each would hand every caller a way round the profile, and a write could retag another tenant's document as the writer's own. Ingest documents outside the server, stamping `org_id` exactly as the identity provider emits it. +The server refuses to start while anything else can reach the same index without the tenant scope: `search-records`, `upsert-records` unless the index is `read_only`, or another custom tool on that index whose `lock.inject` differs, including one with none. Each would hand every caller a way round the profile, and a write could retag another tenant's document as the writer's own. Ingest documents outside the server, stamping `org_id` exactly as the identity provider emits it. Listing the tenant claim under `required_claims` makes the verifier reject a token without it before any tool runs. That checks presence only: the profile still validates the value on every call, and refuses a missing, empty, list-valued or otherwise unusable claim with a `forbidden` error before any query runs. @@ -801,7 +801,7 @@ A profile with `lock.inject` refuses to start when authentication is not enabled ### Claim Injection Refuses an Unscoped Route -A profile with `lock.inject` refuses to start while `search-records`, `upsert-records` on a writable index, or another custom tool without `lock.inject` can reach the same index. The error lists each route it found. Disable the built-ins under `server.builtin_tools`, mark the index `read_only`, or add the same `lock.inject` to the other custom tools. +A profile with `lock.inject` refuses to start while `search-records`, `upsert-records` on a writable index, or another custom tool on the same index whose `lock.inject` differs, including one with none. The error lists each route it found, and each tool's injected scope. Disable the built-ins under `server.builtin_tools`, mark the index `read_only`, and give every custom tool on the index the same `lock.inject`. ### Claim Injection Fails Every Request With `forbidden` diff --git a/redisvl/mcp/server.py b/redisvl/mcp/server.py index a5d47400..7bf87fbc 100644 --- a/redisvl/mcp/server.py +++ b/redisvl/mcp/server.py @@ -32,6 +32,15 @@ logger = logging.getLogger(__name__) +def _describe_scope(scope: frozenset[tuple[str, str]]) -> str: + """Render one tool's injected scope for a startup error.""" + if not scope: + return "injects nothing" + return "injects " + ", ".join( + f"{field} from claim '{claim}'" for field, claim in sorted(scope) + ) + + def _config_injects(config: Any) -> bool: """Report whether any configured profile injects a claim-derived filter.""" if config is None: @@ -352,22 +361,32 @@ def _verify_no_unscoped_route_to_injected_indexes(self, config: Any) -> None: Injection isolates a tool, but tenant data lives in an index, and every tool passes the same read scope gate. So any other route to that index - -- the generic search, a write, or a profile without the injection -- - hands every caller the data the profile was meant to fence off, and a - write can retag another tenant's document as the writer's own. That is - configuration which voids the guarantee it declares, not a redundant - surface, so it is refused rather than warned about. + -- the generic search, a write, or a custom tool injecting a different + scope or none -- hands every caller the data the profile was meant to + fence off, and a write can retag another tenant's document as the + writer's own. That is configuration which voids the guarantee it + declares, not a redundant surface, so it is refused rather than warned + about. + + The scope is a property of the index, so every custom tool on it must + inject the same entries. Injecting *something* is not enough: a second + tool scoped by another field, or by the same field from another claim, + reads across the tenants the first one separates. """ + scopes: dict[str, dict[str, frozenset[tuple[str, str]]]] = {} + for profile in config.custom_tools: + scopes.setdefault(config.resolved_profile_index(profile), {})[ + profile.name + ] = frozenset( + (entry.field, entry.claim) for entry in profile.lock.inject or () + ) injected = sorted( - { - config.resolved_profile_index(profile) - for profile in config.custom_tools - if profile.lock.inject is not None - } + index_id for index_id, by_tool in scopes.items() if any(by_tool.values()) ) if not injected: return + problems: list[str] = [] routes: list[str] = [] if config.server.builtin_tool_enabled("search-records"): routes.append("search-records") @@ -381,22 +400,33 @@ def _verify_no_unscoped_route_to_injected_indexes(self, config: Any) -> None: ] if writable: routes.append("upsert-records") - routes.extend( - f"custom tool '{profile.name}'" - for profile in config.custom_tools - if profile.lock.inject is None - and config.resolved_profile_index(profile) in injected - ) - if not routes: + if routes: + problems.append( + "it is also reachable without that scope through " + f"{', '.join(routes)}" + ) + + for index_id in injected: + by_tool = scopes[index_id] + if len(set(by_tool.values())) > 1: + problems.append( + f"the custom tools on index '{index_id}' do not inject the " + "same scope: " + + "; ".join( + f"'{name}' {_describe_scope(scope)}" + for name, scope in sorted(by_tool.items()) + ) + ) + if not problems: return raise ValueError( f"Index {', '.join(repr(i) for i in injected)} is scoped by an " - "injected token claim, but is also reachable without that scope " - f"through {', '.join(routes)}. Each of those bypasses the tenant " - "filter. Disable the built-ins with server.builtin_tools (for " - "example 'search-records: disabled'), mark the index read_only to " - "stop writes, or give the other custom tools the same lock.inject." + f"injected token claim, but {'; and '.join(problems)}. Each of those " + "bypasses the tenant filter. Disable the built-ins with " + "server.builtin_tools (for example 'search-records: disabled'), mark " + "the index read_only to stop writes, and give every custom tool on " + "the index the same lock.inject." ) @staticmethod diff --git a/tests/integration/test_mcp/test_claim_injection.py b/tests/integration/test_mcp/test_claim_injection.py index ded8d09f..cf936cf2 100644 --- a/tests/integration/test_mcp/test_claim_injection.py +++ b/tests/integration/test_mcp/test_claim_injection.py @@ -312,7 +312,7 @@ async def spy(query): pytest.param( [_PROFILE, {"name": "open-search", "description": "Search the kb."}], {}, - "custom tool 'open-search'", + "'open-search' injects nothing", id="unscoped-profile", ), ], diff --git a/tests/unit/test_mcp/test_server_unit.py b/tests/unit/test_mcp/test_server_unit.py index 1f30c367..52528619 100644 --- a/tests/unit/test_mcp/test_server_unit.py +++ b/tests/unit/test_mcp/test_server_unit.py @@ -877,9 +877,44 @@ def _route_check(*, builtin_tools, custom_tools=None, read_only=False, indexes=N pytest.param( _BUILTINS_OFF, [_INJECTING, {"name": "open-search", "description": "Search."}], - "custom tool 'open-search'", + "'open-search' injects nothing", id="unscoped-profile", ), + # Injecting *something* is not the tenant scope. Scoped by another + # field, the second tool reads across the tenants the first separates. + pytest.param( + _BUILTINS_OFF, + [ + _INJECTING, + { + "name": "region-search", + "description": "Search.", + "lock": { + "inject": [{"field": "rating", "from": "claim", "claim": "r"}] + }, + }, + ], + "'region-search' injects rating from claim 'r'", + id="different-field", + ), + # The same field read from another claim scopes by a different value. + pytest.param( + _BUILTINS_OFF, + [ + _INJECTING, + { + "name": "other-org-search", + "description": "Search.", + "lock": { + "inject": [ + {"field": "category", "from": "claim", "claim": "alt"} + ] + }, + }, + ], + "'other-org-search' injects category from claim 'alt'", + id="different-claim", + ), ], ) def test_an_unscoped_route_to_an_injected_index_is_refused( @@ -933,3 +968,26 @@ def test_a_server_without_injection_keeps_every_route(): builtin_tools={}, custom_tools=[{"name": "open-search", "description": "Search open."}], ) + + +def test_tools_injecting_the_same_scope_share_an_index(): + # Entry order does not matter: the scope is the set of (field, claim) pairs. + two_entry = [ + {"field": "category", "from": "claim", "claim": "org"}, + {"field": "rating", "from": "claim", "claim": "tier"}, + ] + _route_check( + builtin_tools=_BUILTINS_OFF, + custom_tools=[ + { + "name": "tenant-search", + "description": "Search.", + "lock": {"inject": two_entry}, + }, + { + "name": "tenant-recent", + "description": "Recent.", + "lock": {"inject": list(reversed(two_entry))}, + }, + ], + )