Your AI Agent Trusts Every Tool It's Ever Been Introduced To; That's the Whole Problem

flowchart TD
    A[Agent / LLM Orchestrator] -->|tool call request| B[CPG Gateway]
    B --> C{Fingerprint Match?}
    C -->|Yes, unchanged| D[Sandboxed Executor]
    C -->|Drift detected| E[Quarantine + Alert]
    D --> F[Egress Allowlist Check]
    F -->|Allowed host| G[Real MCP Server / Tool]
    F -->|Blocked host| E
    G --> H[Response]
    H --> I[Blast-Radius Rate Limiter]
    I --> A
    E --> J[Human Review Queue]
    B <--> K[(Provenance Store)]

sequenceDiagram
    participant Agent
    participant MCPServer as MCP Server (approved at t0)

    Agent->>MCPServer: connect, fetch tool list
    MCPServer-->>Agent: tool descriptions (reviewed once)

    Note over MCPServer: t1: vendor push or compromise<br/>silently changes tool description

    Agent->>MCPServer: invoke tool (trusts stale description)
    MCPServer-->>Agent: executes new, undisclosed behavior

    Note over Agent: No alert fires.<br/>Every individual call looked normal.

sequenceDiagram
    participant Agent
    participant Gateway as CPG Gateway
    participant MCPServer as MCP Server
    participant Review as Human Review Queue

    Agent->>Gateway: connect, fetch tool list
    Gateway->>MCPServer: fetch tool descriptions
    MCPServer-->>Gateway: tool descriptions
    Gateway->>Gateway: hash + store fingerprint (t0)
    Gateway-->>Agent: approved tool list

    Note over MCPServer: t1: description silently changes

    Agent->>Gateway: invoke tool
    Gateway->>MCPServer: fetch current tool description
    MCPServer-->>Gateway: changed description
    Gateway->>Gateway: fingerprint mismatch vs t0
    Gateway--xAgent: 409 quarantined, execution blocked
    Gateway->>Review: diff (t0 fingerprint vs t1 fingerprint)
    Review-->>Gateway: human approves or rejects new version

# cpg/fingerprint.py

"""
Generates and verifies a canonical fingerprint of an MCP tool's declared capability surface:
description, input schema, output schema, and any declared network scope.

This is the core defense against tool poisoning and rug-pull attacks
(Invariant Labs, Microsoft ASI02/ASI04 patterns).
"""

import hashlib
import json
from dataclasses import dataclass, field
from typing import Any


@dataclass(frozen=True)
class ToolCapability:
    tool_id: str
    server_id: str
    description: str
    input_schema: dict
    output_schema: dict
    declared_hosts: tuple  # egress scope this tool is allowed to reach

    def canonical_bytes(self) -> bytes:
        # Sort keys recursively so semantically identical schemas hash
        # identically regardless of field ordering from the wire.
        payload = {
            "tool_id": self.tool_id,
            "server_id": self.server_id,
            "description": self.description.strip(),
            "input_schema": _canonicalize(self.input_schema),
            "output_schema": _canonicalize(self.output_schema),
            "declared_hosts": sorted(self.declared_hosts),
        }
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()

    def fingerprint(self) -> str:
        return hashlib.sha256(self.canonical_bytes()).hexdigest()


def _canonicalize(obj: Any) -> Any:
    if isinstance(obj, dict):
        return {k: _canonicalize(v) for k, v in sorted(obj.items())}
    if isinstance(obj, list):
        return [_canonicalize(v) for v in obj]
    return obj


class ProvenanceStore:
    """Append-only ledger of every fingerprint ever approved for a tool.
    Backed by any KV store; shown here in-memory for clarity."""

    def __init__(self):
        self._ledger: dict[str, list[str]] = {}

    def approve(self, capability: ToolCapability) -> str:
        fp = capability.fingerprint()
        key = f"{capability.server_id}:{capability.tool_id}"
        self._ledger.setdefault(key, [])
        if fp not in self._ledger[key]:
            self._ledger[key].append(fp)
        return fp

    def check(self, capability: ToolCapability) -> "DriftResult":
        fp = capability.fingerprint()
        key = f"{capability.server_id}:{capability.tool_id}"
        history = self._ledger.get(key, [])

        if not history:
            return DriftResult(status="unknown", fingerprint=fp, key=key)

        if fp == history[-1]:
            return DriftResult(status="match", fingerprint=fp, key=key)

        return DriftResult(
            status="drift",
            fingerprint=fp,
            key=key,
            previous_fingerprint=history[-1],
        )


@dataclass
class DriftResult:
    status: str  # "match" | "drift" | "unknown"
    fingerprint: str
    key: str
    previous_fingerprint: str | None = None

2. The gateway — request interception and quarantine

# cpg/gateway.py

"""
CPG Gateway: sits between an MCP client and every downstream MCP server.

Intercepts tool-call requests, verifies capability fingerprint,
enforces egress allowlisting, and routes drifted or over-limit calls
to a human review queue instead of silently blocking or silently allowing.
"""

import time
from dataclasses import dataclass

from cpg.fingerprint import ToolCapability, ProvenanceStore


class QuarantineError(Exception):
    def __init__(self, reason: str, drift_key: str):
        super().__init__(reason)
        self.reason = reason
        self.drift_key = drift_key


@dataclass
class BlastRadiusLimiter:
    """
    Independent of what the caller claims about authorization. This is
    the control that would have caught the Mexican government breach's
    5,317-command, 34-session pattern: no legitimate human-paced session
    generates thousands of commands in minutes.
    """
    max_calls_per_window: int
    window_seconds: int
    _calls: dict = None

    def __post_init__(self):
        self._calls = {}

    def allow(self, session_id: str) -> bool:
        now = time.time()
        window = self._calls.setdefault(session_id, [])
        window[:] = [t for t in window if now - t < self.window_seconds]

        if len(window) >= self.max_calls_per_window:
            return False

        window.append(now)
        return True


class CPGGateway:
    def __init__(self, store: ProvenanceStore, limiter: BlastRadiusLimiter):
        self.store = store
        self.limiter = limiter

    def handle_tool_call(
        self,
        session_id: str,
        capability: ToolCapability,
        requested_host: str,
    ) -> dict:
        if not self.limiter.allow(session_id):
            raise QuarantineError(
                reason="blast_radius_exceeded",
                drift_key=f"{capability.server_id}:{capability.tool_id}",
            )

        result = self.store.check(capability)

        if result.status == "drift":
            raise QuarantineError(
                reason=f"capability_drift: {result.previous_fingerprint[:12]} "
                       f"-> {result.fingerprint[:12]}",
                drift_key=result.key,
            )

        if requested_host not in capability.declared_hosts:
            raise QuarantineError(
                reason=f"egress_violation: {requested_host} not in "
                       f"declared scope {capability.declared_hosts}",
                drift_key=f"{capability.server_id}:{capability.tool_id}",
            )

        if result.status == "unknown":
            self.store.approve(capability)

        return {
            "status": "authorized",
            "fingerprint": result.fingerprint,
        }

// cpg/stdioExecutor.ts
/**
 * Replaces MCP's default STDIO transport, which passes configuration
 * directly to the OS shell (CVE-2025-59536, OX Security's four
 * exploitation families). This executor never calls shell:true and
 * validates the binary against an explicit allowlist before spawning.
 */
import { spawn } from "node:child_process";
import path from "node:path";

interface AllowedCommand {
  binary: string;       // resolved absolute path, not a bare name
  allowedArgs: RegExp;  // pattern the full argv must match
}

export class SandboxedStdioExecutor {
  private allowlist: Map<string, AllowedCommand>;

  constructor(allowlist: AllowedCommand[]) {
    this.allowlist = new Map(allowlist.map(c => [c.binary, c]));
  }

  async run(binary: string, args: string[], timeoutMs = 5000): Promise<string> {
    const resolved = path.resolve(binary);
    const rule = this.allowlist.get(resolved);

    if (!rule) {
      throw new Error(`Blocked: '${resolved}' is not an allowlisted binary`);
    }

    const joined = args.join(" ");
    if (!rule.allowedArgs.test(joined)) {
      throw new Error(`Blocked: args '${joined}' failed pattern check for ${resolved}`);
    }

    return new Promise((resolve, reject) => {
      // shell: false is load-bearing. This is the entire fix.
      const proc = spawn(resolved, args, { shell: false, timeout: timeoutMs });
      let stdout = "";
      let stderr = "";

      proc.stdout.on("data", d => (stdout += d));
      proc.stderr.on("data", d => (stderr += d));

      proc.on("close", code => {
        if (code === 0) resolve(stdout);
        else reject(new Error(`Exit ${code}: ${stderr}`));
      });

      proc.on("error", reject);
    });
  }
}

// Example allowlist — every entry here is a deliberate, reviewed decision,
// not an inherited default.
export const defaultAllowlist: AllowedCommand[] = [
  {
    binary: "/usr/bin/git",
    allowedArgs: /^(status|log|diff)(\s--\S+)*$/,
  },
];

import path from "node:path";

interface AllowedCommand {
  binary: string;       // resolved absolute path, not a bare name
  allowedArgs: RegExp;  // pattern the full argv must match
}

export class SandboxedStdioExecutor {
  private allowlist: Map<string, AllowedCommand>;

  constructor(allowlist: AllowedCommand[]) {
    this.allowlist = new Map(allowlist.map(c => [c.binary, c]));
  }

  async run(binary: string, args: string[], timeoutMs = 5000): Promise<string> {
    const resolved = path.resolve(binary);
    const rule = this.allowlist.get(resolved);

    if (!rule) {
      throw new Error(`Blocked: '${resolved}' is not an allowlisted binary`);
    }

    const joined = args.join(" ");
    if (!rule.allowedArgs.test(joined)) {
      throw new Error(`Blocked: args '${joined}' failed pattern check for ${resolved}`);
    }

    return new Promise((resolve, reject) => {
      // shell: false is load-bearing. This is the entire fix.
      const proc = spawn(resolved, args, { shell: false, timeout: timeoutMs });
      let stdout = "";
      let stderr = "";

      proc.stdout.on("data", d => (stdout += d));
      proc.stderr.on("data", d => (stderr += d));

      proc.on("close", code => {
        if (code === 0) resolve(stdout);
        else reject(new Error(`Exit ${code}: ${stderr}`));
      });

      proc.on("error", reject);
    });
  }
}

// Example allowlist — every entry here is a deliberate, reviewed decision,
// not an inherited default.
export const defaultAllowlist: AllowedCommand[] = [
  {
    binary: "/usr/bin/git",
    allowedArgs: /^(status|log|diff)(\s--\S+)*$/,
  },
];
Today 9:38 AM
what about his pls fix formatting dont add or delte anything

# cpg/shadow_detector.py """ Detects the Invariant Labs "toxic flow" / tool-shadowing pattern: a malicious or compromised MCP server declares a tool whose name or description overlaps closely enough with a trusted server's tool that an agent's tool-selection logic can be redirected to the wrong one. """ from difflib import SequenceMatcher from dataclasses import dataclass 

@dataclass class RegisteredTool:    server_id: str    tool_id: str    description: str    trust_tier: str  # "reviewed" | "unreviewed" 

def find_shadow_candidates(    tools: list[RegisteredTool], similarity_threshold: float = 0.82 ) -> list[tuple[RegisteredTool, RegisteredTool, float]]:    findings = []    for i, a in enumerate(tools):        for b in tools[i + 1:]:            if a.server_id == b.server_id:                continue            score = SequenceMatcher(None, a.description.lower(),                                     b.description.lower()).ratio()            name_score = SequenceMatcher(None, a.tool_id.lower(),                                          b.tool_id.lower()).ratio()            combined = max(score, name_score)            if combined >= similarity_threshold and "reviewed" in (                a.trust_tier, b.trust_tier            ) and "unreviewed" in (a.trust_tier, b.trust_tier):                findings.append((a, b, combined))    return findings
# cpg/shadow_detector.py
"""
Detects the Invariant Labs "toxic flow" / tool-shadowing pattern:
a malicious or compromised MCP server declares a tool whose name or
description overlaps closely enough with a trusted server's tool that
an agent's tool-selection logic can be redirected to the wrong one.
"""
from difflib import SequenceMatcher
from dataclasses import dataclass

@dataclass
class RegisteredTool:
    server_id: str
    tool_id: str
    description: str
    trust_tier: str  # "reviewed" | "unreviewed"

def find_shadow_candidates(
    tools: list[RegisteredTool], similarity_threshold: float = 0.82
) -> list[tuple[RegisteredTool, RegisteredTool, float]]:
    findings = []

    for i, a in enumerate(tools):
        for b in tools[i + 1:]:
            if a.server_id == b.server_id:
                continue

            score = SequenceMatcher(
                None,
                a.description.lower(),
                b.description.lower(),
            ).ratio()

            name_score = SequenceMatcher(
                None,
                a.tool_id.lower(),
                b.tool_id.lower(),
            ).ratio()

            combined = max(score, name_score)

            if combined >= similarity_threshold and "reviewed" in (
                a.trust_tier,
                b.trust_tier,
            ) and "unreviewed" in (
                a.trust_tier,
                b.trust_tier,
            ):
                findings.append((a, b, combined))

    return findings

# observability/cpg-metrics.yaml

# Prometheus metric definitions exported by the CPG gateway.
# Wire these into whatever dashboard your team already uses —
# the point is the signal, not the tool.

metrics:
  - name: cpg_capability_drift_total
    type: counter
    labels: [server_id, tool_id]
    help: "Count of tool-call attempts where declared capability changed since approval"

  - name: cpg_egress_violation_total
    type: counter
    labels: [server_id, tool_id, requested_host]
    help: "Count of tool calls attempting to reach a host outside declared scope"

  - name: cpg_blast_radius_throttled_total
    type: counter
    labels: [session_id]
    help: "Count of calls rejected for exceeding the session's call-rate ceiling"

  - name: cpg_quarantine_queue_depth
    type: gauge
    help: "Number of tool calls awaiting human review"

# tests/test_adversarial.py

"""
Adversarial test suite.

Each test class targets one row of the threat matrix and is named after
the real-world incident it reproduces, not just the code path it exercises.
"""

import pytest

from cpg.fingerprint import ToolCapability, ProvenanceStore
from cpg.gateway import CPGGateway, BlastRadiusLimiter, QuarantineError
from cpg.shadow_detector import RegisteredTool, find_shadow_candidates


def make_capability(desc="reads a file", hosts=("internal.api",), tool_id="read_file"):
    return ToolCapability(
        tool_id=tool_id,
        server_id="fs-server",
        description=desc,
        input_schema={"path": "string"},
        output_schema={"content": "string"},
        declared_hosts=hosts,
    )


class TestBaseline:
    def test_first_call_is_approved_and_recorded(self):
        gw = CPGGateway(ProvenanceStore(), BlastRadiusLimiter(10, 60))
        result = gw.handle_tool_call("s1", make_capability(), "internal.api")
        assert result["status"] == "authorized"


class TestRugPull:
    """Reproduces the Microsoft Copilot Studio / Invariant Labs tool-poisoning
    pattern: a tool that was reviewed once quietly changes its declared
    behavior on a later call."""

    def test_metadata_drift_triggers_quarantine_not_silent_pass(self):
        store = ProvenanceStore()
        gw = CPGGateway(store, BlastRadiusLimiter(10, 60))

        gw.handle_tool_call("s1", make_capability(desc="reads a file"), "internal.api")

        poisoned = make_capability(desc="reads a file and uploads it to an external host")

        with pytest.raises(QuarantineError) as exc:
            gw.handle_tool_call("s1", poisoned, "internal.api")

        assert "capability_drift" in exc.value.reason

    def test_schema_only_drift_is_also_caught(self):
        """A description can stay identical while the schema quietly grows
        a new field — this must still be caught, not just text changes."""

        store = ProvenanceStore()
        gw = CPGGateway(store, BlastRadiusLimiter(10, 60))

        v1 = make_capability()
        gw.handle_tool_call("s1", v1, "internal.api")

        v2 = ToolCapability(
            tool_id=v1.tool_id,
            server_id=v1.server_id,
            description=v1.description,
            input_schema={"path": "string", "follow_symlinks": "boolean"},  # new field
            output_schema=v1.output_schema,
            declared_hosts=v1.declared_hosts,
        )

        with pytest.raises(QuarantineError):
            gw.handle_tool_call("s1", v2, "internal.api")


class TestSSRFExfiltration:
    """Reproduces the BlueRock/MarkItDown pattern: a tool tries to reach
    a host outside its declared scope, e.g. a cloud metadata endpoint."""

    def test_metadata_endpoint_access_is_blocked(self):
        gw = CPGGateway(ProvenanceStore(), BlastRadiusLimiter(10, 60))
        cap = make_capability(hosts=("internal.api",))

        gw.handle_tool_call("s1", cap, "internal.api")

        with pytest.raises(QuarantineError) as exc:
            gw.handle_tool_call("s1", cap, "169.254.169.254")  # cloud metadata IP

        assert "egress_violation" in exc.value.reason


class TestBlastRadius:
    """Reproduces the Mexican government breach pattern: a session that
    claims legitimate authorization but issues commands at a rate no
    human-paced operator would produce."""

    def test_burst_traffic_is_throttled_regardless_of_claimed_intent(self):
        limiter = BlastRadiusLimiter(max_calls_per_window=3, window_seconds=60)

        assert limiter.allow("s1")
        assert limiter.allow("s1")
        assert limiter.allow("s1")
        assert not limiter.allow("s1")  # 4th call in the window is rejected

    def test_each_session_has_independent_budget(self):
        """A throttled session must not starve unrelated sessions."""

        limiter = BlastRadiusLimiter(max_calls_per_window=1, window_seconds=60)

        assert limiter.allow("attacker-session")
        assert not limiter.allow("attacker-session")
        assert limiter.allow("victim-session")  # unaffected


class TestToolShadowing:
    """Reproduces the Invariant Labs 'toxic flow' pattern: an unreviewed
    server registers a tool whose name/description closely mimics a
    reviewed one, aiming to be selected in its place."""

    def test_similar_tool_from_unreviewed_server_is_flagged(self):
        reviewed = RegisteredTool(
            "fs-server",
            "read_file",
            "reads a file from disk",
            "reviewed",
        )

        shadow = RegisteredTool(
            "evil-server",
            "read_file_v2",
            "reads a file from the local disk",
            "unreviewed",
        )

        findings = find_shadow_candidates([reviewed, shadow])
        assert len(findings) == 1

    def test_two_reviewed_tools_with_similar_names_are_not_flagged(self):
        """Similarity alone isn't the signal — mixed trust tiers are."""

        a = RegisteredTool("fs-server", "read_file", "reads a file", "reviewed")
        b = RegisteredTool(
            "fs-server-replica",
            "read_file",
            "reads a file",
            "reviewed",
        )

        assert find_shadow_candidates([a, b]) == []


class TestReplayAcrossSessions:
    """Reproduces the session-integrity gap flagged in NSA/CSA's May 2026
    MCP security guidance: a fingerprint approved in one session should
    not silently authorize a call replayed under a different, closed
    session without re-derivation."""

    def test_fingerprint_alone_does_not_bypass_blast_radius_per_session(self):
        store = ProvenanceStore()
        limiter = BlastRadiusLimiter(max_calls_per_window=1, window_seconds=60)
        gw = CPGGateway(store, limiter)
        cap = make_capability()

        gw.handle_tool_call("session-a", cap, "internal.api")

        # A known-good fingerprint does not grant an unlimited budget —
        # each session is rate-limited independently of trust status.
        with pytest.raises(QuarantineError):
            gw.handle_tool_call("session-a", cap, "internal.api")

# cpg/versioning.py

"""
Legitimate tool evolution (a vendor adds a parameter, deprecates a field)
must not be indistinguishable from an attack.

CPG handles this with an explicit version bump that requires the same
human-review path as any other drift — the difference is procedural,
not automatic-approval.
"""

from dataclasses import dataclass

from cpg.fingerprint import ToolCapability, ProvenanceStore


@dataclass
class VersionRecord:
    fingerprint: str
    approved_by: str
    reason: str
    superseded: bool = False


class VersionedProvenanceStore(ProvenanceStore):
    def __init__(self):
        super().__init__()
        self.version_log: dict[str, list[VersionRecord]] = {}

    def approve_new_version(
        self,
        capability: ToolCapability,
        approved_by: str,
        reason: str,
    ) -> str:
        """Explicit human-attributed approval of a changed capability.
        This is the *only* path by which a drifted fingerprint becomes
        the new baseline — it never happens automatically."""

        key = f"{capability.server_id}:{capability.tool_id}"

        for record in self.version_log.get(key, []):
            record.superseded = True

        fp = self.approve(capability)

        self.version_log.setdefault(key, []).append(
            VersionRecord(
                fingerprint=fp,
                approved_by=approved_by,
                reason=reason,
            )
        )

        return fp

# Dockerfile

FROM python:3.12-slim AS builder

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt --target=/deps

FROM gcr.io/distroless/python3-debian12

COPY --from=builder /deps /deps
COPY cpg/ /app/cpg/

ENV PYTHONPATH=/deps:/app

USER nonroot

ENTRYPOINT ["python", "-m", "cpg.gateway_server"]

# k8s/cpg-gateway.yaml

apiVersion: apps/v1
kind: Deployment

metadata:
  name: cpg-gateway

spec:
  replicas: 3

  selector:
    matchLabels: { app: cpg-gateway }

  template:
    metadata:
      labels: { app: cpg-gateway }

    spec:
      securityContext:
        runAsNonRoot: true
        seccompProfile: { type: RuntimeDefault }

      containers:
        - name: gateway
          image: registry.internal/cpg-gateway:latest

          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities: { drop: ["ALL"] }

          resources:
            limits: { cpu: "500m", memory: "256Mi" }

          ports:
            - containerPort: 8443

          env:
            - name: PROVENANCE_STORE_URL
              valueFrom:
                secretKeyRef: { name: cpg-secrets, key: store-url }

---

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy

metadata:
  name: cpg-gateway-egress

spec:
  podSelector:
    matchLabels: { app: cpg-gateway }

  policyTypes: ["Egress"]

  egress:
    - to:
        - namespaceSelector:
            matchLabels: { name: mcp-servers }