Sample code for: MCP Server Security: How to Harden a Python MCP Server

mcp-server-security/README.md

@@ -0,0 +1,37 @@
+# MCP Server Security: How to Harden a Python MCP Server
+
+This folder provides the code examples for the Real Python tutorial
+[MCP Server Security: How to Harden a Python MCP Server](https://realpython.com/mcp-server-security/).
+Everything is a harmless local simulation: the "secrets" are fake and no
+network access is needed.
+
+## Setup
+
+```console
+$ python -m venv venv
+$ source venv/bin/activate
+(venv) $ python -m pip install -r requirements.txt
+```
+
+Python 3.11 or later is required. The sandbox demos need
+[Bubblewrap](https://github.com/containers/bubblewrap) (`bwrap`) on Linux.
+
+## Run
+
+```console
+(venv) $ python scan_server.py "$(which python)" vulnerable_server.py
+(venv) $ python attack_demo.py "$(which python)" vulnerable_server.py
+(venv) $ python sandbox_demo.py
+(venv) $ python audience_demo.py
+(venv) $ python checklist.py
+```
+
+- `vulnerable_server.py`: a server with a poisoned description, an
+  over-broad shell tool, and unrestricted file reads
+- `scan_server.py`: lists a server's tools and flags risky metadata
+- `attack_demo.py`: plays an agent that obeyed the poisoned description
+- `hardened_server.py`: narrow tools plus bearer-token auth over HTTP
+- `sandbox.py` and `sandbox_demo.py`: run a server inside `bwrap`
+- `audience_demo.py`: a token for another server, with and without the
+  audience check
+- `checklist.py`: before and after results for every check

mcp-server-security/attack_demo.py

@@ -0,0 +1,31 @@
+"""Play the part of an agent that obeyed a poisoned tool description."""
+
+import asyncio
+import os
+import sys
+
+from mcp import Client, StdioServerParameters
+
+os.environ["DEMO_API_TOKEN"] = "sk-demo-not-a-real-secret"
+
+
+async def main(command: str, *args: str) -> None:
+    params = StdioServerParameters(
+        command=command, args=list(args), env=dict(os.environ)
+    )
+    async with Client(params) as client:
+        attempts = {
+            "read_file": {"path": "demo_home/.ssh/id_demo"},
+            "run_shell": {"command": "env | grep DEMO_"},
+        }
+        for name, arguments in attempts.items():
+            try:
+                result = await client.call_tool(name, arguments)
+                print(f"{name} -> is_error={result.is_error}")
+                print("  " + result.content[0].text.strip())
+            except Exception as error:
+                print(f"{name} -> {type(error).__name__}: {error}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main(*sys.argv[1:]))

mcp-server-security/audience_demo.py

@@ -0,0 +1,49 @@
+"""Show a token minted for another server, with and without audience checks."""
+
+import os
+import subprocess
+import sys
+import time
+
+import httpx2
+
+URL = "http://127.0.0.1:8765/mcp"
+HEADERS = {"Accept": "application/json, text/event-stream"}
+BODY = {
+    "jsonrpc": "2.0",
+    "id": 1,
+    "method": "initialize",
+    "params": {
+        "protocolVersion": "2025-06-18",
+        "capabilities": {},
+        "clientInfo": {"name": "demo", "version": "0"},
+    },
+}
+OTHER = "demo-other-server-token"
+
+
+def probe(validate: str) -> None:
+    env = {
+        **os.environ,
+        "NOTES_TOKEN": "demo-notes-token",
+        "OTHER_SERVER_TOKEN": OTHER,
+        "VALIDATE_AUDIENCE": validate,
+    }
+    server = subprocess.Popen(
+        [sys.executable, "hardened_server.py"],
+        env=env,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        time.sleep(3)
+        headers = {**HEADERS, "Authorization": f"Bearer {OTHER}"}
+        status = httpx2.post(URL, json=BODY, headers=headers).status_code
+        print(f"validate_token_resource={validate == '1'}: HTTP {status}")
+    finally:
+        server.terminate()
+        server.wait()
+
+
+probe("0")
+probe("1")

mcp-server-security/checklist.py

@@ -0,0 +1,164 @@
+"""Before/after hardening checklist: vulnerable vs hardened server."""
+
+import asyncio
+import os
+import shutil
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import httpx2
+from mcp import Client, StdioServerParameters
+from mcp.client.streamable_http import streamable_http_client
+
+from sandbox import sandboxed_params
+from scan_server import scan_tools
+
+HERE = Path(__file__).parent
+URL = "http://127.0.0.1:8765/mcp"
+HEADERS = {"Accept": "application/json, text/event-stream"}
+GOOD, OTHER = "demo-notes-token", "demo-other-server-token"
+BODY = {"jsonrpc": "2.0", "id": 1, "method": "ping"}
+
+
+def report(check: str, before: str, after: str) -> None:
+    print(f"{check:<34}{before:<12}{after}")
+
+
+async def call_ok(client: Client, name: str, args: dict) -> bool:
+    try:
+        result = await client.call_tool(name, args)
+    except Exception:
+        return False
+    return not result.is_error
+
+
+async def output(client: Client, name: str, args: dict) -> str:
+    try:
+        result = await client.call_tool(name, args)
+    except Exception as error:
+        return f"EXC {error}"
+    return result.content[0].text if result.content else ""
+
+
+async def status(token: str | None) -> int:
+    headers = dict(HEADERS)
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    async with httpx2.AsyncClient() as http:
+        return (await http.post(URL, json=BODY, headers=headers)).status_code
+
+
+async def sandbox_checks() -> None:
+    async with Client(sandboxed_params("vulnerable_server.py")) as boxed:
+        wrote = "Read-only" not in await output(
+            boxed, "run_shell", {"command": "touch /app/pwned"}
+        )
+        net = "unreachable" not in await output(
+            boxed,
+            "run_shell",
+            {
+                "command": 'python3 -c "import socket; '
+                "socket.create_connection(('1.1.1.1', 80), timeout=3)\""
+            },
+        )
+        seen = await call_ok(
+            boxed, "read_file", {"path": "demo_home/notes.txt"}
+        )
+    report(
+        "Sandbox: write to disk",
+        "allowed",
+        "allowed" if wrote else "blocked",
+    )
+    report(
+        "Sandbox: outbound network",
+        "allowed",
+        "allowed" if net else "blocked",
+    )
+    report(
+        "Sandbox: read demo_home",
+        "allowed",
+        "allowed" if seen else "blocked",
+    )
+
+
+async def main() -> None:
+    os.environ["DEMO_API_TOKEN"] = "sk-demo-not-a-real-secret"
+    python = sys.executable
+    vuln = StdioServerParameters(
+        command=python,
+        args=["vulnerable_server.py"],
+        env=dict(os.environ),
+    )
+    server_env = {
+        **os.environ,
+        "NOTES_TOKEN": GOOD,
+        "OTHER_SERVER_TOKEN": OTHER,
+    }
+    server = subprocess.Popen(
+        [python, "hardened_server.py"],
+        env=server_env,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        time.sleep(3)
+        report("CHECK", "BEFORE", "AFTER")
+        async with Client(vuln) as before:
+            vuln_flags = await scan_tools(before, verbose=False)
+            leak = "sk-demo" in await output(
+                before, "run_shell", {"command": "echo $DEMO_API_TOKEN"}
+            )
+            read_any = await call_ok(
+                before, "read_file", {"path": "demo_home/.ssh/id_demo"}
+            )
+            shell = await call_ok(before, "run_shell", {"command": "id"})
+        http = httpx2.AsyncClient(headers={"Authorization": f"Bearer {GOOD}"})
+        async with Client(
+            streamable_http_client(URL, http_client=http)
+        ) as after:
+            hard_flags = await scan_tools(after, verbose=False)
+            escape = await call_ok(
+                after, "read_note", {"name": "../demo_home/notes.txt"}
+            )
+            tools = {tool.name for tool in (await after.list_tools()).tools}
+        report(
+            "Tool metadata scanner findings", str(vuln_flags), str(hard_flags)
+        )
+        report(
+            "Arbitrary shell tool",
+            "exposed" if shell else "-",
+            "exposed" if "run_shell" in tools else "removed",
+        )
+        report(
+            "Read outside allowed folder",
+            "allowed" if read_any else "blocked",
+            "allowed" if escape else "blocked",
+        )
+        report(
+            "Secret visible to tools", "leaked" if leak else "hidden", "n/a"
+        )
+        report(
+            "No token -> HTTP status", "n/a (stdio)", str(await status(None))
+        )
+        report(
+            "Other server's token -> status",
+            "n/a (stdio)",
+            str(await status(OTHER)),
+        )
+        report("Own token -> MCP session", "n/a (stdio)", "connected")
+        if shutil.which("bwrap") is None:
+            for check in (
+                "write to disk",
+                "outbound network",
+                "read demo_home",
+            ):
+                report(f"Sandbox: {check}", "allowed", "skipped")
+            return
+        await sandbox_checks()
+    finally:
+        server.terminate()
+
+
+asyncio.run(main())

mcp-server-security/demo_home/.ssh/id_demo

@@ -0,0 +1 @@
+FAKE-PRIVATE-KEY-FOR-DEMO-ONLY

mcp-server-security/demo_home/notes.txt

@@ -0,0 +1 @@
+public notes

mcp-server-security/hardened_server.py

@@ -0,0 +1,83 @@
+"""The same notes server, hardened: narrow tools, bearer auth, audience."""
+
+import hmac
+import os
+from pathlib import Path
+
+from mcp.server.auth.middleware.auth_context import get_access_token
+from mcp.server.auth.provider import AccessToken
+from mcp.server.auth.settings import AuthSettings
+from mcp.server.mcpserver import MCPServer
+
+HOST, PORT = "127.0.0.1", 8765
+RESOURCE_URL = f"http://{HOST}:{PORT}/mcp"
+WORKSPACE = Path(__file__).parent.joinpath("workspace").resolve()
+MAX_NOTE_CHARS = 200
+
+# Tokens come from the environment, never from tool arguments or code.
+# In production an authorization server issues them; this maps each one
+# to the resource (audience) it was minted for.
+TOKENS = {
+    os.environ.get("NOTES_TOKEN", ""): ("notes-client", RESOURCE_URL),
+    os.environ.get("OTHER_SERVER_TOKEN", ""): (
+        "notes-client",
+        "http://127.0.0.1:9000/mcp",
+    ),
+}
+
+
+class StaticTokenVerifier:
+    async def verify_token(self, token: str) -> AccessToken | None:
+        for known, (client_id, resource) in TOKENS.items():
+            if known and hmac.compare_digest(known.encode(), token.encode()):
+                return AccessToken(
+                    token=token,
+                    client_id=client_id,
+                    scopes=["notes:read", "notes:write"],
+                    resource=resource,
+                )
+        return None
+
+
+mcp = MCPServer(
+    "notes-helper",
+    token_verifier=StaticTokenVerifier(),
+    auth=AuthSettings(
+        issuer_url="http://127.0.0.1:9999",
+        resource_server_url=RESOURCE_URL,
+        required_scopes=["notes:read"],
+        validate_token_resource=os.environ.get("VALIDATE_AUDIENCE", "1")
+        == "1",
+    ),
+)
+
+
+def safe_path(name: str) -> Path:
+    """Resolve name inside WORKSPACE or raise ValueError."""
+    path = (WORKSPACE / name).resolve()
+    if not path.is_relative_to(WORKSPACE) or path.suffix != ".txt":
+        raise ValueError("Only .txt notes inside the workspace are allowed")
+    return path
+
+
+@mcp.tool()
+def read_note(name: str) -> str:
+    """Read a .txt note from the notes workspace."""
+    return safe_path(name).read_text(encoding="utf-8")
+
+
+@mcp.tool()
+def add_note(name: str, text: str) -> str:
+    """Save a short note as a .txt file in the notes workspace."""
+    token = get_access_token()
+    if token is None or "notes:write" not in token.scopes:
+        raise ValueError("Missing scope: notes:write")
+    if len(text) > MAX_NOTE_CHARS:
+        raise ValueError(f"Note longer than {MAX_NOTE_CHARS} characters")
+    with safe_path(name).open("x", encoding="utf-8") as file:
+        file.write(text)
+    return f"Saved {name}"
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http", host=HOST, port=PORT)

mcp-server-security/requirements.txt

@@ -0,0 +1,2 @@
+mcp==2.3.0
+httpx2==2.16.0