"""Test the official MCP server against a disposable loopback Redis process. Requires redis-server on PATH. Never connects to an existing Redis instance. Writes validation.json beside this file; no credentials appear in the report. """ import asyncio from datetime import datetime, timezone from importlib.metadata import version import json import os from pathlib import Path import secrets import shutil import socket import subprocess import sys import tempfile import time from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import redis RULES = ["reset", "on", "~mcpdemo:*", "-@all", "+ping", "+get", "+hget", "+hgetall", "+type", "+ttl"] CHECKS = [] def record(name, condition, details=None): if not condition: raise AssertionError(f"{name}: {details}") CHECKS.append({"name": name, "passed": True, "details": details}) def result_text(result): return "\n".join(part.text for part in result.content if hasattr(part, "text")) async def exercise(port, password, admin): env = {k: v for k, v in os.environ.items() if not k.startswith("REDIS_")} env.update(REDIS_USERNAME="mcp_reader", REDIS_PWD=password) parameters = StdioServerParameters( command=str(Path(sys.executable).parent / "redis-mcp-server"), args=["--host", "127.0.0.1", "--port", str(port), "--db", "0"], env=env, ) async with stdio_client(parameters) as (read, write): async with ClientSession(read, write) as session: init = await session.initialize() record("MCP initialization", bool(init.serverInfo.name), init.serverInfo.name) available = {t.name for t in (await session.list_tools()).tools} record("Core tools advertised", {"get", "hgetall", "type", "set", "scan_keys"} <= available, sorted(available)) async def call(name, args): return result_text(await session.call_tool(name, args)) text = await call("get", {"key": "mcpdemo:status"}) record("Read string through MCP", "ready" in text, text) text = await call("hgetall", {"name": "mcpdemo:session:42"}) record("Read hash through MCP", "demo-user" in text and "english" in text, text) text = await call("type", {"key": "mcpdemo:session:42"}) data = json.loads(text) record("Inspect type and TTL through MCP", data["type"] == "hash" and 0 < data["ttl"] <= 3600, data) before = admin.get("mcpdemo:status") text = await call("set", {"key": "mcpdemo:status", "value": "changed"}) record("MCP write denied", "permission" in text.lower() or "noperm" in text.lower(), text) record("Denied write leaves data unchanged", admin.get("mcpdemo:status") == before) text = await call("get", {"key": "private:sentinel"}) record("Other prefix denied", "permission" in text.lower() or "noperm" in text.lower(), text) record("Private value absent from MCP response", "PRIVATE_SENTINEL" not in text) text = await call("scan_keys", {"pattern": "*", "count": 100}) record("Key enumeration denied", "permission" in text.lower() or "noperm" in text.lower(), text) text = await call("type", {"key": "mcpdemo:missing"}) data = json.loads(text) record("Missing key has TTL minus two", data["ttl"] == -2 and data["type"] == "none", data) # Demonstrate why granting SCAN defeats key-name isolation. admin.execute_command("ACL", "SETUSER", "mcp_reader", "+scan") text = await call("scan_keys", {"pattern": "*", "count": 1000}) record("SCAN reveals names outside ACL prefix", "private:sentinel" in text) admin.execute_command("ACL", "SETUSER", "mcp_reader", "-scan") text = await call("scan_keys", {"pattern": "*", "count": 1000}) record("SCAN revoked again", "permission" in text.lower() or "noperm" in text.lower()) def main(): executable = shutil.which("redis-server") if not executable: raise SystemExit("Install redis-server and put it on PATH first.") root = Path(__file__).resolve().parent runs = root / "test-runs" runs.mkdir(exist_ok=True) run = Path(tempfile.mkdtemp(prefix="mcp-", dir=runs)) with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] log = (run / "redis.log").open("w") process = subprocess.Popen([executable, "--bind", "127.0.0.1", "--port", str(port), "--save", "", "--appendonly", "no", "--dir", str(run)], stdout=log, stderr=subprocess.STDOUT) admin = redis.Redis(host="127.0.0.1", port=port, decode_responses=True) try: for _ in range(60): if process.poll() is not None: raise RuntimeError("Disposable Redis failed to start; inspect test-runs log.") try: if admin.ping(): break except redis.ConnectionError: time.sleep(0.05) else: raise RuntimeError("Redis startup timed out") password = secrets.token_urlsafe(32) admin.set("mcpdemo:status", "ready", ex=3600) admin.hset("mcpdemo:session:42", mapping={"user": "demo-user", "language": "english"}) admin.expire("mcpdemo:session:42", 3600) admin.set("private:sentinel", "PRIVATE_SENTINEL") admin.execute_command("ACL", "SETUSER", "mcp_reader", *RULES, ">" + password) restricted = redis.Redis(host="127.0.0.1", port=port, username="mcp_reader", password=password, decode_responses=True) record("Reader authenticates", restricted.ping()) asyncio.run(asyncio.wait_for(exercise(port, password, admin), timeout=60)) result = { "testedAt": datetime.now(timezone.utc).isoformat(), "scope": "Official package over MCP stdio against isolated local Redis; no hosted database or editor UI test", "versions": {"python": sys.version.split()[0], "redis-server": admin.info()["redis_version"], **{p: version(p) for p in ["redis-mcp-server", "mcp", "redis"]}}, "checks": CHECKS, "passed": len(CHECKS), } (root / "validation.json").write_text(json.dumps(result, indent=2) + "\n") print(json.dumps(result, indent=2)) finally: admin.close() process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait() log.close() if __name__ == "__main__": main()