Agentic Ventures Wiki

F2-replicate-admin-tools-block

4 Min. Lesezeit

F2 — Account-Admin- und Raw-Tools via tools/call ohne Whitelist erreichbar

Datei: ~/source/mcps/mcp-replicate-hosted/src/mcp_replicate_hosted/main.py

Patch-Skizze

Defense-in-Depth analog mcp-vf-hosted-_BLOCKED_TOOLS-Pattern. Hier wir machen es als Allow-Liste statt Block-Liste (sicherer — neue mcp-replicate-Tools sind automatisch geblockt bis explizit freigegeben).

# main.py — direkt unter _MODEL_GATED_TOOLS
 
# Defense-in-Depth: `tools/list` ist via ToolWhitelist gefiltert (Discovery-
# Pattern), `tools/call` bleibt open fuer search_tools-Workflow. Dadurch sind
# ALLE replicate_*-Tools im Sub-MCP erreichbar — auch Admin (create_model,
# delete_model, create_training, create_deployment, raw_get, raw_post) und
# Info-Leak-Tools (get_account, get_webhook_secret).
#
# Statt jeden Admin-Tool einzeln auf eine Block-Liste zu setzen (Pattern aus
# mcp-vf-hosted: _BLOCKED_TOOLS): hier Allow-Liste, weil sub-MCP-Tool-Set
# unter unserer Kontrolle ist und neue Tools automatisch geblockt sein
# sollten.
_ALLOWED_REPLICATE_TOOLS: frozenset[str] = frozenset({
    # Prediction-Lifecycle (Read + Cancel)
    "replicate_get_prediction",
    "replicate_wait_for_prediction",
    "replicate_cancel_prediction",
    "replicate_list_predictions",
    # Predictions selbst nur via Layer-Tools (create_image etc.) oder
    # replicate_create_prediction MIT Model-Whitelist-Check (F1 + bestehender Code).
    "replicate_create_prediction",
})
 
 
_BLOCKED_REPLICATE_TOOL_MESSAGE = (
    "Dieses Tool ist serverseitig deaktiviert. Erlaubte replicate_*-Tools: "
    "create_prediction (mit Modell-Whitelist), get_prediction, wait_for_prediction, "
    "cancel_prediction, list_predictions. Plus die Layer-Tools create_image, "
    "create_text_image, create_svg_logo, create_image_from_reference, create_video. "
    "Account-Admin- und Raw-Tools sind geblockt, damit JWT-Auth nicht zum Replicate-"
    "Account-Admin-Zugriff eskaliert (Defense-in-Depth, siehe Security-Audit F2)."
)

Dann in GuardMiddleware.on_call_tool (direkt nach Emergency-Disable, vor Model-Check):

# NEU: nicht-Whitelist replicate_* hard-rejecten
if tool.startswith("replicate_") and tool not in _ALLOWED_REPLICATE_TOOLS:
    subject = _extract_subject(context)
    audit_log(
        "blocked_replicate_admin_tool_attempt",
        tool=tool,
        subject=subject,
        reason="not_in_allowlist",
    )
    raise RuntimeError(_BLOCKED_REPLICATE_TOOL_MESSAGE)

Wichtig: Layer-Tools (create_image etc.) rufen create_prediction intern via _build_sub_mcp_client — diese Sub-MCP-Calls gehen NICHT durch unsere GuardMiddleware (sind Calls innerhalb des Containers, kein neuer JWT-Request). Bleibt funktional.

Test

tests/test_blocked_replicate_admin_tools.py neu anlegen:

"""F2 regression — Admin/Raw replicate_* tools must be rejected before reaching the sub-MCP."""
 
from __future__ import annotations
 
import pytest
from unittest.mock import MagicMock, AsyncMock
 
from mcp_replicate_hosted.main import GuardMiddleware, _ALLOWED_REPLICATE_TOOLS
from mcp_replicate_hosted.config import DEFAULT_MODEL_WHITELIST
from mcp_replicate_hosted.ratelimit import RateLimiter
 
 
@pytest.fixture
def guard() -> GuardMiddleware:
    return GuardMiddleware(RateLimiter(60, 1000), DEFAULT_MODEL_WHITELIST)
 
 
def _ctx(tool: str, args: dict | None = None) -> MagicMock:
    ctx = MagicMock()
    ctx.message = MagicMock(name=tool, arguments=args or {})
    ctx.message.name = tool
    ctx.message.arguments = args or {}
    ctx.fastmcp_context = None
    return ctx
 
 
@pytest.mark.parametrize("blocked_tool", [
    "replicate_raw_get",
    "replicate_raw_post",
    "replicate_create_model",
    "replicate_delete_model",
    "replicate_delete_model_version",
    "replicate_create_training",
    "replicate_cancel_training",
    "replicate_create_deployment",
    "replicate_update_deployment",
    "replicate_delete_deployment",
    "replicate_get_account",
    "replicate_get_webhook_secret",
    "replicate_upload_file",
    "replicate_delete_file",
    "replicate_search_models",   # info-leak im hosted Kontext nicht noetig
    "replicate_list_deployments",
    "replicate_get_deployment",
])
@pytest.mark.asyncio
async def test_admin_or_raw_tool_blocked(guard: GuardMiddleware, blocked_tool: str) -> None:
    call_next = AsyncMock(return_value="should_not_be_called")
    with pytest.raises(RuntimeError, match="serverseitig deaktiviert"):
        await guard.on_call_tool(_ctx(blocked_tool), call_next)
    call_next.assert_not_called()
 
 
@pytest.mark.asyncio
async def test_allowed_replicate_tools_pass_through(guard: GuardMiddleware) -> None:
    """Allow-Liste muss alle Lifecycle-Tools + create_prediction durchlassen."""
    for tool in _ALLOWED_REPLICATE_TOOLS:
        call_next = AsyncMock(return_value="OK")
        ctx = _ctx(tool, {"prediction_id": "test"})
        # darf nicht raise mit "serverseitig deaktiviert" — andere Errors
        # (rate-limit, model-whitelist) sind hier nicht im Scope
        try:
            await guard.on_call_tool(ctx, call_next)
        except RuntimeError as e:
            assert "serverseitig deaktiviert" not in str(e), (
                f"allowed tool {tool} wurde geblockt: {e}"
            )
 
 
@pytest.mark.asyncio
async def test_layer_tools_unaffected(guard: GuardMiddleware) -> None:
    """Layer-Tools (create_image etc.) starten nicht mit replicate_, sind unaffected."""
    for tool in ("create_image", "create_text_image", "create_svg_logo",
                 "create_image_from_reference", "create_video"):
        call_next = AsyncMock(return_value="OK")
        ctx = _ctx(tool, {"prompt": "test", "model": "black-forest-labs/flux-2-pro"})
        try:
            await guard.on_call_tool(ctx, call_next)
        except RuntimeError as e:
            assert "serverseitig deaktiviert" not in str(e)

Aufwand

~30 Zeilen Code + 1 neues Test-File (~80 Zeilen). <20 Min inkl. pytest-Lauf.

Side-Effects

  • search_tools zeigt weiterhin die geblockten Tools im Discovery-Output (anhand _get_tool_catalog). Sollten wir die nicht-allowed-Tools auch aus dem search_tools-Output filtern? Ja — sonst widerspricht das Discovery (Tool ist da) der Realitaet (Call ist geblockt). Optionale zweite Aenderung:

    # In search_tools-Implementation, nach catalog-Fetch:
if t["name"].startswith("replicate_") and t["name"] not in _ALLOWED_REPLICATE_TOOLS:
    continue  # skip blocked tools from discovery

  • Marvin’s eigene Admin-Workflows (z.B. replicate_get_account aus Debug-Bedarf) gehen direkt via mcp-replicate lokal (nicht via hosted MCP) → unaffected.

  • Falls eine neue Layer-Tool jemals ein bisher geblocktes Tool intern braucht: Layer-Tools nutzen _build_sub_mcp_client → bypassed GuardMiddleware. Bleibt funktional.

Reihenfolge zu F1

F2 ist DER Hauptfix. F1 ist Subset davon (replicate_create_prediction ist allowed, also greift F1 wieder als zusaetzliche Schicht). In einem PR umsetzen, Tests grun, dann F3 dranziehen.

Graphansicht

Backlinks