diff --git a/docs/concepts/mcp.md b/docs/concepts/mcp.md index 8d22d9c3..c227db4b 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,75 @@ 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 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 + +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 | 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. + +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 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 (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 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`. + +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 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. +- 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. + +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..0f01f52b 100644 --- a/docs/user_guide/how_to_guides/mcp.md +++ b/docs/user_guide/how_to_guides/mcp.md @@ -340,6 +340,79 @@ 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 # 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} + 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 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 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 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. + +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 RedisVL MCP exposes a small, implementation-owned contract. @@ -721,3 +794,15 @@ 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 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 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` + +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/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/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/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 14097b1f..7bf87fbc 100644 --- a/redisvl/mcp/server.py +++ b/redisvl/mcp/server.py @@ -31,6 +31,23 @@ 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: + return False + return any(profile.lock.inject for profile in config.custom_tools) + + try: from fastmcp import FastMCP except ImportError: @@ -73,6 +90,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 +139,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 +305,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 +323,112 @@ 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." + ) + + 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 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( + 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") + 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") + 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 " + 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 def _tool_surface_fingerprint(config: Any) -> str: """Summarize the config that a registered tool set baked in.""" @@ -318,8 +454,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 +531,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 +701,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: 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 @@ -571,7 +726,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..674ef770 100644 --- a/redisvl/mcp/tools/profiles.py +++ b/redisvl/mcp/tools/profiles.py @@ -16,8 +16,12 @@ 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 ( @@ -56,11 +60,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 +141,10 @@ def validate_profile_against_schema( # Raises RedisVLMCPError(INVALID_FILTER) naming the offending field. resolve_locked_filter(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 + if profile.lock.inject: + validate_inject_against_schema( + profile.lock.inject, schema, profile_name=profile.name + ) def _validate_locked_exists_fields( @@ -165,7 +164,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 +228,15 @@ 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) + # `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") @@ -242,6 +248,18 @@ 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 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 + ) + # 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 +298,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..cf936cf2 --- /dev/null +++ b/tests/integration/test_mcp/test_claim_injection.py @@ -0,0 +1,336 @@ +"""End-to-end tests for claim-injected tenant scoping against a real Redis. + +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 + +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}]}, +} + +# 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: + 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"}, + # 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 leaked would match every document below. + await index.load( + [ + _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", + ) + 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, + 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", + "public_key": key.public_key, + "issuer": ISSUER, + "audience": AUDIENCE, + } + config = { + "server": server, + "indexes": { + "kb": { + "redis_name": redis_name, + "read_only": read_only, + "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) -> list: + return sorted(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"}) + + 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 _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"}, [], 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"}, + ] + }, + ["acme", "acme"], + id="or-widening", + ), + pytest.param( + {"not": {"field": "org_id", "op": "eq", "value": "acme"}}, + [], + 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"}) + + # 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 + + +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.description + assert "category(tag)" in tool.description + + +@pytest.mark.parametrize( + "claims", + [ + # The shape that genuinely unions, refused on type. + pytest.param({ORG_CLAIM: ["acme", "victim"]}, id="array"), + # 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( + 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( + "custom_tools, kwargs, expected", + [ + 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."}], + {}, + "'open-search' injects nothing", + id="unscoped-profile", + ), + ], +) +async def test_startup_refuses_an_injection_profile_that_cannot_hold( + tenant_index, config_path, custom_tools, kwargs, expected +): + server = RedisVLMCPServer( + MCPSettings( + 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_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) 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..bf941747 100644 --- a/tests/unit/test_mcp/test_profiles_unit.py +++ b/tests/unit/test_mcp/test_profiles_unit.py @@ -918,3 +918,138 @@ 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()) + + +@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": []} + + 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 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 a8f878da..52528619 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,364 @@ 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 + ) + + +# -------------------------------------------------------------------------- +# 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."}], + "'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( + 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."}], + ) + + +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))}, + }, + ], + )