diff --git a/docs/mcp-guard/getting-started/installation.md b/docs/mcp-guard/getting-started/installation.md index 5e6ea4e..b3db2f4 100644 --- a/docs/mcp-guard/getting-started/installation.md +++ b/docs/mcp-guard/getting-started/installation.md @@ -1,42 +1,75 @@ ---- -title: Installation -description: Install CapiscIO MCP Guard for tool-level security on MCP servers. ---- + # Installation -MCP Guard provides trust badges and identity verification for [Model Context Protocol](https://modelcontextprotocol.io) tool calls. +## PyPI Installation + +```bash +# Standalone (no MCP SDK dependency) +pip install capiscio-mcp + +# With MCP SDK integration +pip install capiscio-mcp[mcp] + +# With PoP signing/verification (requires cryptography) +pip install capiscio-mcp[crypto] + +# Full installation +pip install capiscio-mcp[mcp,crypto] +``` + +## Using uv + +```bash +uv add capiscio-mcp +uv add capiscio-mcp --extra mcp +``` ## Requirements - Python 3.10+ -- A CapiscIO account or self-hosted registry +- capiscio-core (auto-downloaded on first use) -## Install via pip +## Core Connection Modes + +capiscio-mcp connects to capiscio-core for cryptographic operations: + +### Embedded Mode (Default) + +The SDK automatically downloads and manages the core binary: ```bash pip install capiscio-mcp +# Just works! Binary downloaded on first use. ``` -For MCP SDK integration (FastMCP wrapper): +The binary is cached at `~/.capiscio/bin/capiscio`. -```bash -pip install capiscio-mcp[mcp] -``` +### External Mode -## Verify Installation +Connect to a separately managed core service: ```bash -python -c "import capiscio_mcp; print(capiscio_mcp.__version__)" +# Start core in another terminal +capiscio mcp serve --listen localhost:50051 + +# SDK connects to external core +export CAPISCIO_CORE_ADDR="localhost:50051" ``` -## What's Included +## Verify Installation -| Package | Description | -|---------|-------------| -| `capiscio-mcp` | Core guard decorator and evidence logging | -| `capiscio-mcp[mcp]` | FastMCP integration for automatic tool wrapping | +```python +from capiscio_mcp import MCP_VERSION, CORE_MIN_VERSION + +print(f"capiscio-mcp version: {MCP_VERSION}") +print(f"Required core version: {CORE_MIN_VERSION}") +``` -## Next Steps +## Environment Variables -- [Quickstart](quickstart.md) — protect your first MCP tool in 5 minutes +| Variable | Description | Default | +|----------|-------------|---------| +| `CAPISCIO_CORE_ADDR` | External core address | (embedded mode) | +| `CAPISCIO_SERVER_ORIGIN` | Server origin for guard | (auto-detect) | +| `CAPISCIO_LOG_LEVEL` | Logging verbosity | `info` | diff --git a/docs/mcp-guard/getting-started/quickstart.md b/docs/mcp-guard/getting-started/quickstart.md index 5b7382c..3d072ee 100644 --- a/docs/mcp-guard/getting-started/quickstart.md +++ b/docs/mcp-guard/getting-started/quickstart.md @@ -1,13 +1,12 @@ ---- -title: Quickstart -description: Protect your first MCP tool with CapiscIO trust badges in 5 minutes. ---- + # Quickstart -Protect your MCP tools with trust-level requirements using the `@guard` decorator. +Get started with capiscio-mcp in under 5 minutes. -## Server-Side: Protect a Tool +## Server-Side: Protect Your Tools + +The `@guard` decorator protects MCP tools with trust-level requirements: ```python from capiscio_mcp import guard @@ -15,38 +14,83 @@ from capiscio_mcp import guard @guard(min_trust_level=2) async def read_database(query: str) -> list[dict]: """Only agents with Trust Level 2+ can execute this tool.""" - # ... your tool logic + # Your database query logic here + return [{"id": 1, "name": "Example"}] +``` + +### With Configuration + +```python +from capiscio_mcp import guard, GuardConfig + +config = GuardConfig( + min_trust_level=2, + trusted_issuers=["did:web:registry.capisc.io"], + allowed_tools=["read_*", "list_*"], + require_badge=True, # Deny anonymous access +) + +@guard(config=config) +async def execute_query(sql: str) -> list[dict]: + """Execute a SQL query with full policy enforcement.""" + pass +``` + +### Sync Version + +```python +from capiscio_mcp import guard_sync + +@guard_sync(min_trust_level=2) +def read_database_sync(query: str) -> list[dict]: + """Synchronous version for non-async code.""" pass ``` -## Client-Side: Verify a Server +## Client-Side: Verify Servers + +Before connecting to an MCP server, verify its identity: ```python -from capiscio_mcp import verify_server +from capiscio_mcp import verify_server, ServerState -result = await verify_server("did:web:example.com:mcp:my-server") -if result.verified: - print(f"Server trust level: {result.trust_level}") +result = await verify_server( + server_did="did:web:mcp.example.com", + server_badge="eyJhbGc...", + transport_origin="https://mcp.example.com", +) + +if result.state == ServerState.VERIFIED_PRINCIPAL: + print(f"✅ Trusted server at Level {result.trust_level}") +elif result.state == ServerState.DECLARED_PRINCIPAL: + print("⚠️ Server identity declared but not verified") +elif result.state == ServerState.UNVERIFIED_ORIGIN: + print("❌ Server did not disclose identity") ``` -## How It Works +## Understanding Server States -1. **Agent sends badge** — the calling agent presents its CapiscIO trust badge -2. **Guard verifies** — `@guard` checks the badge signature, expiry, and trust level -3. **Evidence logged** — every invocation is recorded with a cryptographic audit trail -4. **Tool executes** — if verification passes, the tool runs normally +| State | Meaning | Action | +|-------|---------|--------| +| `VERIFIED_PRINCIPAL` | Identity cryptographically verified | Safe to proceed | +| `DECLARED_PRINCIPAL` | Identity declared but verification failed | Proceed with caution | +| `UNVERIFIED_ORIGIN` | No identity disclosed | High risk | ## Trust Levels -| Level | Name | Meaning | -|-------|------|---------| -| 0 | Self-Signed | Agent generated its own badge | -| 1 | Domain Verified | Agent proved control of its DID | -| 2 | Organization Verified | Agent's organization has been verified | -| 3 | Extended Validation | Full identity verification completed | +Per RFC-002 v1.4: + +| Level | Name | Validation | Use Case | +|-------|------|------------|----------| +| 0 | Self-Signed (SS) | None, `did:key` issuer | Local dev, testing, demos | +| 1 | Registered (REG) | Account registration | Development, internal agents | +| 2 | Domain Validated (DV) | DNS/HTTP challenge | Production, B2B agents | +| 3 | Organization Validated (OV) | DUNS/legal entity | High-trust production | +| 4 | Extended Validated (EV) | Manual review + legal | Regulated industries | ## Next Steps -- [Server-Side Guide](../guides/server-side.md) — detailed guard configuration -- [Client-Side Guide](../guides/client-side.md) — verify MCP servers before connecting -- [Evidence Logging](../guides/evidence.md) — audit trail configuration +- [Server-Side Guide](../guides/server-side.md) - Full @guard configuration +- [Client-Side Guide](../guides/client-side.md) - Server verification patterns +- [Evidence Logging](../guides/evidence.md) - Audit trail configuration +- [MCP SDK Integration](../guides/mcp-integration.md) - Use with official MCP SDK diff --git a/docs/mcp-guard/guides/client-side.md b/docs/mcp-guard/guides/client-side.md index 2751d34..90e8856 100644 --- a/docs/mcp-guard/guides/client-side.md +++ b/docs/mcp-guard/guides/client-side.md @@ -1,49 +1,173 @@ ---- -title: Verify MCP Servers -description: Client-side guide for verifying MCP server identity before connecting. ---- + -# Verify MCP Servers (Client-Side) +# Client-Side: Verifying MCP Servers -Before connecting to an MCP server, verify its identity using its DID and trust badge. +This guide covers verifying MCP server identity per RFC-007. -## Verify a Server +## Basic Usage ```python -from capiscio_mcp import verify_server +from capiscio_mcp import verify_server, ServerState -result = await verify_server("did:web:example.com:mcp:my-server") +result = await verify_server( + server_did="did:web:mcp.example.com:servers:filesystem", + server_badge="eyJhbGc...", + transport_origin="https://mcp.example.com", +) -if result.verified: - print(f"Trust level: {result.trust_level}") - print(f"Issuer: {result.issuer}") -else: - print(f"Verification failed: {result.reason}") +if result.state == ServerState.VERIFIED_PRINCIPAL: + print(f"✅ Server verified at trust level {result.trust_level}") +elif result.state == ServerState.DECLARED_PRINCIPAL: + print("⚠️ Identity declared but verification failed") + print(f" Error: {result.error_detail}") +elif result.state == ServerState.UNVERIFIED_ORIGIN: + print("❌ Server did not disclose any identity") ``` -## Verification Checks +## VerifyConfig Options + +```python +from capiscio_mcp import verify_server, VerifyConfig + +config = VerifyConfig( + # List of trusted issuer DIDs + trusted_issuers=[ + "did:web:registry.capisc.io", + ], + + # Minimum trust level required (0-4) + min_trust_level=2, + + # Accept self-signed (did:key) servers? + accept_level_zero=False, + + # Skip revocation checks (offline mode)? + offline_mode=False, + + # Skip host/path binding checks (for trusted gateways) + skip_origin_binding=False, +) -The client verifier performs the following checks: +result = await verify_server( + server_did="did:web:mcp.example.com", + server_badge="eyJhbGc...", + transport_origin="https://mcp.example.com", + config=config, +) +``` -1. **DID Resolution** — resolves the server's `did:web` to its DID Document -2. **Badge Validation** — verifies the JWS signature and claims -3. **Trust Level** — confirms the badge meets your minimum trust requirement -4. **Expiry** — ensures the badge has not expired -5. **Revocation** — checks the badge has not been revoked +## Extracting Identity from Transport -## Configuration +### From HTTP Headers ```python -from capiscio_mcp import MCPClient +from capiscio_mcp import parse_http_headers, verify_server -client = MCPClient( - min_trust_level=2, - require_pop=True, - registry_url="https://registry.capisc.io", +# Extract from HTTP response +headers = { + "Capiscio-Server-DID": "did:web:mcp.example.com", + "Capiscio-Server-Badge": "eyJhbGc...", +} + +identity = parse_http_headers(headers) + +server_did, server_badge = identity +if server_did: + result = await verify_server( + server_did=server_did, + server_badge=server_badge, + transport_origin="https://mcp.example.com", + ) +``` + +### From JSON-RPC _meta + +```python +from capiscio_mcp import parse_jsonrpc_meta, verify_server + +# Extract from MCP JSON-RPC response +response = { + "jsonrpc": "2.0", + "id": 1, + "result": {...}, + "_meta": { + "capiscio_server_did": "did:web:mcp.example.com", + "capiscio_server_badge": "eyJhbGc...", + } +} + +identity = parse_jsonrpc_meta(response.get("_meta", {})) + +server_did, server_badge = identity +if server_did: + result = await verify_server( + server_did=server_did, + server_badge=server_badge, + ) +``` + +## Server States Explained + +| State | RFC-007 Definition | Recommended Action | +|-------|-------------------|-------------------| +| `VERIFIED_PRINCIPAL` | DID verified via badge chain | ✅ Safe to proceed | +| `DECLARED_PRINCIPAL` | DID provided but verification failed | ⚠️ Warn user, consider blocking | +| `UNVERIFIED_ORIGIN` | No identity material provided | ❌ Treat as untrusted | + +## Error Handling + +```python +from capiscio_mcp import verify_server, ServerVerifyError + +try: + result = await verify_server( + server_did="did:web:mcp.example.com", + server_badge="eyJhbGc...", + transport_origin="https://mcp.example.com", + ) +except ServerVerifyError as e: + print(f"Verification error: {e.error_code}") + print(f"Detail: {e.message}") +``` + +## Error Codes + +| Code | Meaning | +|------|---------| +| `DID_INVALID` | Server DID is malformed | +| `BADGE_INVALID` | Badge signature invalid | +| `BADGE_EXPIRED` | Badge has expired | +| `BADGE_REVOKED` | Badge has been revoked | +| `TRUST_INSUFFICIENT` | Trust level below minimum | +| `ORIGIN_MISMATCH` | Transport origin doesn't match DID domain | +| `PATH_MISMATCH` | Endpoint path doesn't match DID path | +| `ISSUER_UNTRUSTED` | Badge issuer not in trusted list | + +## Synchronous Version + +```python +from capiscio_mcp import verify_server_sync + +result = verify_server_sync( + server_did="did:web:mcp.example.com", + server_badge="eyJhbGc...", + transport_origin="https://mcp.example.com", ) ``` -## Next Steps +## Strict Mode + +Raises an exception if verification fails: -- [Server Registration](server-registration.md) — register your server's DID -- [Quickstart](../getting-started/quickstart.md) — end-to-end example +```python +from capiscio_mcp.server import verify_server_strict + +# Raises ServerVerifyError if not VERIFIED_PRINCIPAL +result = await verify_server_strict( + server_did="did:web:mcp.example.com", + server_badge="eyJhbGc...", + transport_origin="https://mcp.example.com", +) +# If we get here, server is verified +print(f"Server trust level: {result.trust_level}") +``` diff --git a/docs/mcp-guard/guides/evidence.md b/docs/mcp-guard/guides/evidence.md index f0388b6..4985fed 100644 --- a/docs/mcp-guard/guides/evidence.md +++ b/docs/mcp-guard/guides/evidence.md @@ -1,60 +1,130 @@ ---- -title: Evidence Logging -description: Configure cryptographic audit trails for MCP tool invocations. ---- + # Evidence Logging -MCP Guard records a cryptographic audit trail for every tool invocation, enabling post-incident review and compliance reporting. +Every tool invocation—allowed or denied—produces an evidence record for audit and forensics. -## How It Works +## What Gets Logged -When a guarded tool is invoked, MCP Guard records: +Evidence records include: | Field | Description | |-------|-------------| -| `timestamp` | ISO 8601 invocation time | -| `tool_name` | Name of the MCP tool called | -| `agent_did` | DID of the calling agent | -| `trust_level` | Agent's verified trust level | -| `badge_jti` | Badge ID used for verification | -| `result` | `allowed` or `denied` | -| `reason` | Denial reason (if denied) | - -## Configuration +| `evidence_id` | Unique ID for this record | +| `timestamp` | When the evaluation occurred | +| `tool_name` | Tool that was invoked | +| `params_hash` | SHA-256 hash of parameters (not raw params—PII safe) | +| `decision` | ALLOW or DENY | +| `deny_reason` | Why access was denied (if applicable) | +| `agent_did` | Caller's DID (if authenticated) | +| `badge_jti` | Badge ID (if badge was used) | +| `auth_level` | ANONYMOUS, API_KEY, or BADGE | +| `trust_level` | Caller's trust level (0-4) | +| `server_origin` | Server that processed the request | + +## Accessing Evidence on Denial ```python -from capiscio_mcp import configure_evidence +from capiscio_mcp import guard, GuardError + +@guard(min_trust_level=2) +async def sensitive_operation(data: dict) -> dict: + pass + +try: + result = await sensitive_operation(data={"key": "value"}) +except GuardError as e: + print(f"Denied: {e.reason}") + print(f"Evidence ID: {e.evidence_id}") + print(f"Agent DID: {e.agent_did}") + print(f"Trust Level: {e.trust_level}") +``` + +## Evidence Storage + +Evidence can be stored in multiple backends: -configure_evidence( - enabled=True, - output="file", # "file", "stdout", or "http" - path="./evidence.jsonl", # for file output -) +### Local Storage (Default) + +Evidence stored in `~/.capiscio/evidence/`: + +```bash +export CAPISCIO_EVIDENCE_DIR="$HOME/.capiscio/evidence" ``` -## Output Formats +### Registry Storage + +Forward evidence to the CapiscIO Registry: -### File (default) +```bash +export CAPISCIO_EVIDENCE_MODE="registry" +export CAPISCIO_REGISTRY_ENDPOINT="https://registry.capisc.io/events" +export CAPISCIO_REGISTRY_API_KEY="sk_live_..." +``` + +### Hybrid Storage -Evidence records are appended to a JSONL file: +Store locally AND forward to registry: -```json -{"timestamp":"2026-01-15T10:30:00Z","tool_name":"read_database","agent_did":"did:web:agent.example.com","trust_level":2,"result":"allowed"} +```bash +export CAPISCIO_EVIDENCE_MODE="hybrid" ``` -### HTTP +## Parameters Hash -Send evidence to a remote collector: +Tool parameters are **never** sent to the core or logged directly. Instead, a deterministic hash is computed: ```python -configure_evidence( - output="http", - endpoint="https://registry.capisc.io/v1/evidence", -) +from capiscio_mcp import compute_params_hash + +params = {"query": "SELECT * FROM users", "limit": 10} +hash_value = compute_params_hash(params) + +print(hash_value) +# SHA-256 of canonical JSON: "a1b2c3d4..." ``` -## Next Steps +This keeps PII out of evidence records while allowing correlation. + +## Querying Evidence + +Evidence records can be queried via: + +1. **Local files**: JSON files in the evidence directory +2. **Registry API**: Query via CapiscIO Registry (requires API key) + +### Local Evidence Example + +```python +import json +from pathlib import Path + +evidence_dir = Path.home() / ".capiscio" / "evidence" + +for evidence_file in evidence_dir.glob("*.json"): + with open(evidence_file) as f: + record = json.load(f) + if record["decision"] == "deny": + print(f"Denial: {record['tool_name']} by {record['agent_did']}") +``` + +## Evidence Retention + +Configure retention via environment: + +```bash +# Keep evidence for 90 days +export CAPISCIO_EVIDENCE_RETENTION_DAYS="90" + +# Disable auto-cleanup +export CAPISCIO_EVIDENCE_RETENTION_DAYS="0" +``` + +## Compliance Considerations + +Evidence logging helps with: -- [Server-Side Guide](server-side.md) — protect tools with trust levels -- [MCP SDK Integration](mcp-integration.md) — automatic evidence with FastMCP +- **SOC 2**: Audit trail of access decisions +- **GDPR**: Parameters are hashed, not stored raw +- **HIPAA**: Track who accessed what tools +- **PCI DSS**: Monitor privileged operations diff --git a/docs/mcp-guard/guides/mcp-integration.md b/docs/mcp-guard/guides/mcp-integration.md index 7ff7fca..b630828 100644 --- a/docs/mcp-guard/guides/mcp-integration.md +++ b/docs/mcp-guard/guides/mcp-integration.md @@ -1,11 +1,8 @@ ---- -title: MCP SDK Integration -description: Integrate CapiscIO MCP Guard with the FastMCP SDK for automatic tool protection. ---- + # MCP SDK Integration -Integrate MCP Guard with the [FastMCP](https://github.com/jlowin/fastmcp) SDK for automatic tool wrapping and evidence logging. +Use capiscio-mcp with the official MCP Python SDK. ## Installation @@ -13,46 +10,139 @@ Integrate MCP Guard with the [FastMCP](https://github.com/jlowin/fastmcp) SDK fo pip install capiscio-mcp[mcp] ``` -## Usage with FastMCP +## Server Integration + +### CapiscioMCPServer + +The `CapiscioMCPServer` wraps the MCP SDK server with automatic identity disclosure: ```python -from fastmcp import FastMCP -from capiscio_mcp.mcp import SecureMCP +from capiscio_mcp.integrations.mcp import CapiscioMCPServer + +server = CapiscioMCPServer( + name="filesystem", + did="did:web:mcp.example.com:servers:filesystem", + badge="eyJhbGc...", # Server's trust badge +) -# Wrap your FastMCP server with MCP Guard -mcp = FastMCP("My Server") -secure = SecureMCP(mcp, min_trust_level=1) +@server.tool(min_trust_level=2) +async def read_file(path: str) -> str: + """Read a file (requires Trust Level 2+).""" + with open(path) as f: + return f.read() -@secure.tool() -async def query_database(sql: str) -> list[dict]: - """All tools registered via secure.tool() are automatically guarded.""" - return await db.execute(sql) +@server.tool(min_trust_level=3) +async def write_file(path: str, content: str) -> None: + """Write a file (requires Trust Level 3+).""" + with open(path, "w") as f: + f.write(content) + +# Run the server +if __name__ == "__main__": + server.run() ``` -## How It Works +### Features + +- **Automatic identity headers**: Adds `Capiscio-Server-DID` and `Capiscio-Server-Badge` to responses +- **Trust-level enforcement**: Uses `@guard` under the hood +- **Evidence logging**: All tool calls logged automatically -`SecureMCP` wraps the FastMCP server and: +## Client Integration -1. Intercepts incoming tool calls -2. Extracts the agent's badge from the request context -3. Verifies the badge signature, trust level, and expiry -4. Logs evidence for the invocation -5. Forwards the call to the original tool if verification passes +### CapiscioMCPClient -## Configuration +The `CapiscioMCPClient` wraps the MCP SDK client with automatic server verification: ```python -secure = SecureMCP( - mcp, - min_trust_level=2, - require_pop=True, - evidence_enabled=True, - registry_url="https://registry.capisc.io", -) +from capiscio_mcp.integrations.mcp import CapiscioMCPClient + +async with CapiscioMCPClient( + server_url="https://mcp.example.com", + min_trust_level=2, # Require verified identity + badge="eyJhbGc...", # Your client badge +) as client: + # Server identity already verified on connect + print(f"Connected at trust level {client.server_trust_level}") + + result = await client.call_tool("read_file", {"path": "/data/file.txt"}) ``` -## Next Steps +### Features + +- **Automatic verification**: Verifies server identity on connection +- **Badge attachment**: Attaches your badge to tool calls +- **Error on untrusted**: Raises exception if server not verified + +## Manual Integration + +If you're using a custom MCP setup, use the core functions directly: + +### Server Side + +```python +from mcp.server import Server +from capiscio_mcp import guard + +server = Server("my-server") -- [Server-Side Guide](server-side.md) — manual guard decorator usage -- [Evidence Logging](evidence.md) — configure audit trails -- [Server Registration](server-registration.md) — register your server DID +@server.tool() +@guard(min_trust_level=2) # Apply guard to MCP tool +async def my_tool(param: str) -> str: + return f"Result: {param}" +``` + +### Client Side + +```python +from mcp.client import Client +from capiscio_mcp import verify_server, parse_http_headers + +async def connect_and_verify(url: str): + # Connect to MCP server + async with Client(url) as client: + # Get server info + info = await client.get_server_info() + + # Parse identity from response headers (returns tuple) + server_did, server_badge = parse_http_headers(client.last_response_headers) + + if server_did: + result = await verify_server( + server_did=server_did, + server_badge=server_badge, + transport_origin=url, + ) + + if not result.is_verified: + raise RuntimeError(f"Server not verified: {result.error_detail}") + + # Proceed with verified server + return client +``` + +## Stdio Transport + +For stdio-based MCP servers, identity is passed via JSON-RPC `_meta`: + +```python +from capiscio_mcp import parse_jsonrpc_meta, verify_server + +# Server adds identity to _meta +response = { + "jsonrpc": "2.0", + "id": 1, + "result": {...}, + "_meta": { + "capiscio_server_did": "did:web:example.com", + "capiscio_server_badge": "eyJhbGc...", + } +} + +# Client extracts and verifies (returns tuple) +server_did, server_badge = parse_jsonrpc_meta(response["_meta"]) +result = await verify_server( + server_did=server_did, + server_badge=server_badge, +) +``` diff --git a/docs/mcp-guard/guides/server-registration.md b/docs/mcp-guard/guides/server-registration.md index 4e29fdc..293c790 100644 --- a/docs/mcp-guard/guides/server-registration.md +++ b/docs/mcp-guard/guides/server-registration.md @@ -1,62 +1,240 @@ ---- -title: Server Registration -description: Register your MCP server's identity with the CapiscIO registry. ---- + -# Server Registration +# Server Identity Registration -Register your MCP server's DID to establish its identity in the CapiscIO trust network. +This guide covers setting up a verifiable identity for your MCP server. -## Generate a Keypair +## Why Server Identity? + +MCP servers expose powerful tools—file systems, databases, APIs. But how do clients know they're connecting to the **real** server and not an imposter? + +Server identity registration solves this by: + +- **Generating a keypair** for cryptographic signing +- **Creating a DID** (Decentralized Identifier) for the server +- **Registering with the CapiscIO Registry** for discoverability + +## Quick Start + +```python +from capiscio_mcp import setup_server_identity + +# One-step setup: generate keys + register with registry +result = await setup_server_identity( + server_id="550e8400-e29b-41d4-a716-446655440000", # From dashboard + api_key="sk_live_...", # Registry API key + output_dir="./keys", +) + +print(f"Server DID: {result['did']}") +# did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK +``` + +## Prerequisites + +1. **Create your MCP server in the CapiscIO Dashboard** + - Go to [dashboard.capisc.io](https://dashboard.capisc.io) + - Navigate to Servers → Create Server + - Copy the server UUID + +2. **Get an API key** + - In the dashboard, go to Settings → API Keys + - Create a key with `servers:write` permission + +3. **capiscio-core must be running** + - Either embedded (auto-downloaded) or external + +## Step-by-Step Registration + +### Step 1: Generate Keypair + +Generate an Ed25519 keypair for your server: ```python from capiscio_mcp import generate_server_keypair -keypair = generate_server_keypair() -print(f"DID: {keypair.did}") -print(f"Private key saved to: {keypair.key_path}") +keys = await generate_server_keypair(output_dir="./keys") + +print(f"DID: {keys['did_key']}") +print(f"Private key: {keys['private_key_path']}") ``` -## Register with the Registry +Returns: + +| Key | Description | +|-----|-------------| +| `key_id` | Unique key identifier | +| `did_key` | The derived `did:key:z6Mk...` URI | +| `public_key_pem` | PEM-encoded public key | +| `private_key_pem` | PEM-encoded private key | +| `private_key_path` | Path to saved key file (if `output_dir` provided) | + +### Step 2: Register with Registry + +Register the DID with the CapiscIO registry: ```python -from capiscio_mcp import register_server +from capiscio_mcp import register_server_identity -result = await register_server( - did="did:web:example.com:mcp:my-server", - name="My MCP Server", - description="Database access tools", - registry_url="https://registry.capisc.io", +await register_server_identity( + server_id="550e8400-e29b-41d4-a716-446655440000", + api_key="sk_live_...", + did=keys["did_key"], + public_key=keys["public_key_pem"], ) ``` -## Using the CLI +### Combined: setup_server_identity + +For convenience, use the combined function: + +```python +from capiscio_mcp import setup_server_identity -```bash -# Generate keypair -capiscio mcp keygen --did did:web:example.com:mcp:my-server +result = await setup_server_identity( + server_id="550e8400-e29b-41d4-a716-446655440000", + api_key="sk_live_...", + output_dir="./keys", +) -# Register server -capiscio mcp register \ - --did did:web:example.com:mcp:my-server \ - --name "My MCP Server" +# Returns everything you need +print(f"DID: {result['did']}") +print(f"Private key: {result['private_key_path']}") ``` -## DID Format +## Synchronous API + +All functions have sync wrappers: -MCP server DIDs follow the `did:web` method: +```python +from capiscio_mcp import ( + generate_server_keypair_sync, + register_server_identity_sync, + setup_server_identity_sync, +) +# Sync version +result = setup_server_identity_sync( + server_id="550e8400-e29b-41d4-a716-446655440000", + api_key="sk_live_...", + output_dir="./keys", +) ``` -did:web:example.com:mcp:server-name + +## Using the Identity + +After registration, use the DID and private key for server identity disclosure: + +### With CapiscioMCPServer + +```python +from capiscio_mcp.integrations.mcp import CapiscioMCPServer + +server = CapiscioMCPServer( + name="filesystem", + did=result["did"], + private_key_path=result["private_key_path"], +) + +@server.tool(min_trust_level=2) +async def read_file(path: str) -> str: + """Server identity is automatically disclosed.""" + with open(path) as f: + return f.read() ``` -The DID Document is hosted at: +### Manual Disclosure + +Add identity headers to responses: + +```python +from fastapi import FastAPI, Response + +app = FastAPI() +@app.middleware("http") +async def add_server_identity(request, call_next): + response = await call_next(request) + response.headers["Capiscio-Server-DID"] = SERVER_DID + response.headers["Capiscio-Server-Badge"] = SERVER_BADGE + return response ``` -https://example.com/.well-known/did.json + +## Error Handling + +```python +from capiscio_mcp import ( + setup_server_identity, + RegistrationError, + KeyGenerationError, +) +from capiscio_mcp.errors import CoreConnectionError + +try: + result = await setup_server_identity( + server_id="550e8400-e29b-41d4-a716-446655440000", + api_key="sk_live_...", + ) +except CoreConnectionError as e: + print("Could not connect to capiscio-core") + print("Ensure it's running: capiscio mcp serve") +except KeyGenerationError as e: + print(f"Key generation failed: {e}") +except RegistrationError as e: + print(f"Registration failed: {e}") + if e.status_code == 401: + print("Invalid API key") + elif e.status_code == 404: + print("Server not found - create it in the dashboard first") ``` +## Environment Variables + +| Variable | Description | +|----------|-------------| +| `CAPISCIO_SERVER_ID` | Server UUID from dashboard | +| `CAPISCIO_API_KEY` | Registry API key | +| `CAPISCIO_SERVER_URL` | Registry URL (default: production) | +| `CAPISCIO_SERVER_DOMAIN` | Domain for badge issuance | +| `CAPISCIO_SERVER_PRIVATE_KEY_PEM` | PEM-encoded Ed25519 private key (ephemeral environments) | +| `CAPISCIO_CORE_ADDR` | External core address (default: embedded) | +| `CAPISCIO_SERVER_DID` | Pre-configured server DID | +| `CAPISCIO_SERVER_PRIVATE_KEY` | Path to private key PEM | + +!!! tip "Deploying to containers or serverless?" + See the [Deployment Guide](deployment.md) for Docker, Lambda, Cloud Run, and + Kubernetes examples using `CAPISCIO_SERVER_PRIVATE_KEY_PEM` for ephemeral + identity persistence. + +## Security Best Practices + +1. **Never commit private keys** + ```gitignore + # .gitignore + *.pem + keys/ + capiscio_keys/ + ``` + +2. **Use restrictive permissions** + ```bash + chmod 600 ./keys/*.pem + ``` + +3. **Rotate keys periodically** + - Generate new keypair + - Update registry with new DID + - Keep old key for transition period + +4. **Store API keys securely** + ```bash + # Use environment variables + export CAPISCIO_REGISTRY_API_KEY="sk_live_..." + ``` + ## Next Steps -- [Server-Side Guide](server-side.md) — protect tools with the @guard decorator -- [Evidence Logging](evidence.md) — audit tool invocations +- [Deployment Guide](deployment.md) - Docker, Lambda, Cloud Run, Kubernetes +- [Protect MCP Tools](server-side.md) - Add trust-level requirements +- [Client-Side Verification](client-side.md) - Verify servers before connecting +- [Evidence Logging](evidence.md) - Audit trail for all tool calls diff --git a/docs/mcp-guard/guides/server-side.md b/docs/mcp-guard/guides/server-side.md index 60d7357..4182ce7 100644 --- a/docs/mcp-guard/guides/server-side.md +++ b/docs/mcp-guard/guides/server-side.md @@ -1,11 +1,8 @@ ---- -title: Protect MCP Tools -description: Server-side guide for protecting MCP tools with the @guard decorator. ---- + -# Protect MCP Tools (Server-Side) +# Server-Side: Guarding MCP Tools -Use the `@guard` decorator to enforce trust-level requirements on your MCP tool functions. +This guide covers protecting MCP tools with the `@guard` decorator per RFC-006. ## Basic Usage @@ -13,44 +10,124 @@ Use the `@guard` decorator to enforce trust-level requirements on your MCP tool from capiscio_mcp import guard @guard(min_trust_level=2) -async def sensitive_operation(data: str) -> dict: - """Requires Trust Level 2 or higher.""" - return {"status": "ok"} +async def read_file(path: str) -> str: + """Read a file (requires Trust Level 2+).""" + with open(path) as f: + return f.read() +``` + +## GuardConfig Options + +```python +from capiscio_mcp import guard, GuardConfig + +config = GuardConfig( + # Minimum trust level required (0-4) + min_trust_level=2, + + # Accept self-signed (did:key) badges? + accept_level_zero=False, + + # List of trusted issuer DIDs + trusted_issuers=[ + "did:web:registry.capisc.io", + "did:web:internal.example.com", + ], + + # Glob patterns for allowed tool names + allowed_tools=[ + "read_*", + "list_*", + ], + + # If True, deny anonymous/API key access + require_badge=True, + + # Policy version for tracking + policy_version="v1.0", +) + +@guard(config=config) +async def execute_query(sql: str) -> list[dict]: + pass ``` -## Synchronous Tools +## Handling Denials ```python -from capiscio_mcp import guard_sync +from capiscio_mcp import guard, GuardError -@guard_sync(min_trust_level=1) -def read_config(key: str) -> str: - """Synchronous tool with trust verification.""" - return config[key] +@guard(min_trust_level=2) +async def sensitive_operation(data: dict) -> dict: + pass + +try: + result = await sensitive_operation(data={"key": "value"}) +except GuardError as e: + print(f"Access denied: {e.reason}") + print(f"Evidence ID: {e.evidence_id}") # For audit trail + print(f"Caller DID: {e.agent_did}") ``` -## Configuration Options +## Deny Reasons + +| Reason | Meaning | +|--------|---------| +| `BADGE_MISSING` | No badge provided | +| `BADGE_INVALID` | Badge signature invalid | +| `BADGE_EXPIRED` | Badge has expired | +| `BADGE_REVOKED` | Badge has been revoked | +| `TRUST_INSUFFICIENT` | Trust level too low | +| `TOOL_NOT_ALLOWED` | Tool not in allowed_tools | +| `ISSUER_UNTRUSTED` | Badge issuer not trusted | +| `POLICY_DENIED` | Custom policy denied access | + +## Different Trust Levels for Different Tools -| Parameter | Type | Default | Description | -|-----------|------|---------|-------------| -| `min_trust_level` | `int` | `1` | Minimum trust level required (0-3) | -| `require_pop` | `bool` | `False` | Require Proof of Possession (RFC-003) | -| `evidence` | `bool` | `True` | Log evidence for this invocation | +```python +from capiscio_mcp import guard + +@guard(min_trust_level=1) +async def list_files(directory: str) -> list[str]: + """Low-risk: List files (DV sufficient).""" + pass + +@guard(min_trust_level=2) +async def read_file(path: str) -> str: + """Medium-risk: Read file contents (OV required).""" + pass + +@guard(min_trust_level=3) +async def write_file(path: str, content: str) -> None: + """High-risk: Write files (OV required).""" + pass + +@guard(min_trust_level=4) +async def execute_command(cmd: str) -> str: + """Critical: Execute shell commands (EV required).""" + pass +``` -## Error Handling +## Context Access -When a badge fails verification, the guard raises `GuardError` with a descriptive message: +Access caller information within guarded functions: ```python -from capiscio_mcp.errors import GuardError +from capiscio_mcp import guard +from capiscio_mcp.guard import get_credential -try: - result = await guarded_tool("input") -except GuardError as e: - print(f"Access denied: {e}") +@guard(min_trust_level=2) +async def audit_operation(data: dict) -> dict: + credential = get_credential() + + if credential: + print(f"Caller DID: {credential.agent_did}") + print(f"Trust Level: {credential.trust_level}") + print(f"Badge JTI: {credential.badge_jti}") + + return {"status": "ok"} ``` -## Next Steps +## MCP SDK Integration -- [Evidence Logging](evidence.md) — configure audit trails -- [MCP SDK Integration](mcp-integration.md) — automatic FastMCP wrapping +See [MCP SDK Integration](mcp-integration.md) for using `@guard` with the official MCP SDK.