resk-mcp · Documentation

Security model

Exactly what resk-mcp enforces at the MCP boundary, in which order, what each control can and cannot catch, and how to harden a real deployment.

• Version 0.1.4
• In-process controls
• Fail closed by default

Threats and controls

Enforcement order

Knowing the order saves debugging time, because each stage fails with a different kind of response:

1. Rate limit — slowapi evaluates the key function (JWT user_id if the signature verifies, client IP otherwise) and may answer 429 before anything else runs.

2. Bearer extraction — FastAPI's HTTPBearer requires an Authorization: Bearer … header. A missing header fails with HTTP 403.

3. Body shape — RawMCPRequest requires id, method and params, and the method must start with tool/, resource/ or prompt/. Violations return HTTP 422 with FastAPI's validation detail.

4. Token verification — verify_jwt_token() checks signature, algorithm and expiry. Failures return HTTP 200 with MCP error -33000.

5. Injection scan — detect_prompt_injection(params); a hit returns -33003.

6. PII scan — detect_pii(params); a hit returns -33003.

7. Token budget — context_manager.is_within_limits(params); over budget returns -33002.

8. Tracking and dispatch — the interaction counter increments, then the tool runs. Exceptions in the tool return -32603.

Authentication

Tokens are ordinary JWTs signed with JWT_SECRET. The only claim the library consumes is user_id; exp is set automatically by create_jwt_token() and enforced on verification.

from resk_mcp import create_jwt_token, verify_jwt_token
from resk_mcp.auth import AuthError

token = create_jwt_token(user_id="[email protected]")
# override per call if needed:
short = create_jwt_token(user_id="[email protected]", expires_delta_minutes=5)

try:
    claims = verify_jwt_token(token)
    print(claims["user_id"])
except AuthError as exc:
    print("rejected:", exc)   # "Token has expired" / "Invalid token: ..."

If you mint tokens with another library, replicate this contract exactly: same secret, same algorithm, a user_id claim, and an expiry. A token without user_id still authenticates, but rate limiting falls back to the IP address and the logs show unknown_user.

Operational rules

• Generate the secret with at least 32 characters of entropy (secrets.token_urlsafe(48)).
• Never commit it. Load it from the environment or a secret manager.
• Rotate on exposure: rotation invalidates every outstanding token, which is the point.
• Keep lifetimes short (30 minutes is the default) and mint per user, not per deployment.
• JWT_ALGORITHM=none is rejected at startup; do not disable that check.

Injection screening

detect_prompt_injection() walks the params structure recursively (dicts, lists, strings, including dictionary keys) and rejects the request when any string contains one of the built-in phrases, case-insensitively:

"ignore previous instructions"
"disregard previous context"
"act as though"
"new context:"
"system prompt:"
"malicious input"

This is a tripwire for the obvious attacks, not a classifier. Text like "set aside your earlier directions" passes. Treat a hit as a signal worth logging and investigating, and keep a semantic filter on the model side — for example resk-llm, which the optional MCPSecurityManager also uses.

If your tools legitimately handle text that contains these phrases (a security scanner ingesting payloads, a red-team harness), you need a different validation path: the list is module-level in resk_mcp.validation and is not configurable through config.yaml.

PII screening

detect_pii() scans string values only — dictionary keys are deliberately skipped to avoid false positives on parameter names such as email or phone. Two patterns ship today:

The practical consequence is important: a tool that takes an email address or a phone number as an argument will always be rejected with -33003. This is enforced unconditionally — there is no allowlist and the validation.pii_detection_level key in the sample configuration is not read.

Design around it rather than weakening it: pass an opaque user or record identifier and resolve it server-side, or keep the sensitive value in a database the tool reads itself. If a tool genuinely must receive raw PII, you need to adapt the validation layer (see the validation helpers) and accept the trade-off explicitly.

Rate limiting

The limit is a slowapi decorator on /mcp_secure, configured through RATE_LIMIT. The key function tries the bearer token first: a valid signature yields user:<user_id>, anything else yields ip:<client>. Two consequences:

• An attacker with a valid token is limited per identity, not per IP — sharing an egress IP does not let them multiply their quota.
• An attacker without a token is limited per IP, and requests that fail authentication still consume that IP budget.

Counters live in process memory. Behind a load balancer or with several uvicorn workers, each process enforces its own limit; if you need a global quota, enforce it at the gateway (nginx limit_req, Cloudflare, an API gateway) as well.

Token budget

MAX_TOKEN_PER_REQUEST bounds the estimated size of params. The estimator serializes the structure to JSON and divides the character count by four. It is deterministic, cheap and transport-agnostic, but it under-estimates for languages that tokenize inefficiently (CJK, code) and over-estimates for very repetitive ASCII payloads.

The budget applies to params, not to tool results. A tool that returns a 2 MB document is not blocked by this control — cap output size in the tool itself.

Schema validation

Pydantic validates the envelope; per-tool parameter schemas are opt-in via ToolSchemaValidator. Register a JSON Schema and validate at the top of the tool, or wrap the function so validation runs automatically:

from resk_mcp.schema_validation import ToolSchemaValidator

validator = ToolSchemaValidator()

ADD_SCHEMA = {
    "type": "object",
    "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
    "required": ["a", "b"],
    "additionalProperties": False,
}

@validator.from_decorator(ADD_SCHEMA)
async def add(a: float, b: float) -> float:
    return a + b

A validation failure raises SchemaValidationError with a list of errors. If you let it propagate, the dispatcher falls back through its lookup paths and the client receives -32601 ("not found or couldn't be executed") — technically true, but unhelpful. Catch it inside the tool and return a structured error so the caller learns which parameter is wrong. Generate schemas from Pydantic models with generate_schema_from_model(MyModel) to avoid drift between the two definitions.

Optional hardening layer

resk_mcp.security.MCPSecurityManager adds RESK-LLM's detectors on top: heuristic filtering, content moderation, URL and IP detection, canary tokens, invisible-text and homoglyph analysis. It is not required by SecureMCPServer, and importing it requires the separate resk-llm package, which is not a dependency of mcp-resk 0.1.0:

pip install resk-llm

from resk_mcp.security import MCPSecurityManager

manager = MCPSecurityManager(
    enable_heuristic_filter=True,
    enable_content_moderation=False,
    enable_pii_detection=True,
    enable_url_detection=True,
    enable_ip_protection=True,
)

secured_params, info = manager.secure_mcp_request(
    "tool/calculator/add", {"a": 2, "b": 3}, user_id="[email protected]"
)
if info["is_blocked"]:
    ...  # refuse to execute

leak = manager.check_response("tool output here")
if leak["has_leaked_tokens"]:
    ...  # redact and alert

You can extend detection at runtime with add_attack_pattern(), add_suspicious_pattern() and add_suspicious_keyword(). Because this layer is wired manually, it sees exactly what you pass it — including responses, which the built-in pipeline never inspects.

Error codes and status codes

Audit trail

Every accepted request increments a counter for its tool, resource or prompt; successful tool executions increment the tool counter again. Server logs record the authenticated user_id on each call and a warning whenever a request is rejected, so shipping stdout to your log stack gives you a workable audit trail. The dashboard exposes the same counters in memory. Nothing is persisted by the library itself — export logs if you need retention.

Hardening checklist

• Generate a 48-byte secret, store it outside the repository, and rotate it on any suspicion.
• Bind to 127.0.0.1 and terminate TLS at a reverse proxy; expose only /mcp_secure publicly.
• Keep RESK_MCP_DOCS=false and DASHBOARD_AUTH_ENABLED=true with a long random password.
• Run a single worker, or move rate limiting to the gateway as well.
• Set MAX_TOKEN_PER_REQUEST from real payload sizes rather than leaving the default, and cap tool output size in code.
• Pass opaque identifiers instead of raw emails and phone numbers.
• Add ToolSchemaValidator to every tool that accepts structured input.
• Filter tool outputs on the consumer side; requests being screened does not make responses safe.
• Log rejections with their code, and alert on spikes of -33003.
• Run pip-audit and keep dependencies current — the project ships a CI job for it.