Scope boundary

Owns client-side schema and size validation, content sanitization, renderer isolation, terminal-escape handling, and denial of implicit execution for server responses. It does not make server-returned content semantically trustworthy.

AID-H-034.005: AID-H-028.004 safely parses and renders content after it reaches the client; AID-H-034.005 constrains and labels tool outputs, resources, prompts, and input requests before they leave the server. Server output controls do not replace a hostile-input-safe client renderer.

Implementation guidance

• AID-H-028.004-G001: Validate every incoming MCP message against a strict schema and enforce maximum payload size, nesting depth, and read timeout before parsing or rendering.
• AID-H-028.004-G002: Sanitize all rendered response content and harden the renderer so server-supplied markup, scripts, or terminal control sequences cannot execute or mislead the user.
• AID-H-028.004-G003: Require explicit user confirmation before executing any server-suggested action, and display the action, target, and parameters in a normalized read-only confirmation view.
• AID-H-028.004-G004: Require explicit user confirmation before opening server-suggested URLs, and display the full normalized destination without prefetching or preauthenticated links.

AID-H-028.004-G001

Concept:

Reject malformed or oversized messages as early as possible. This reduces parser edge cases, memory pressure, and unsafe fallback behavior.

Example Node.js schema validation with Ajv

// File: src/mcp/validate-incoming-message.mjs
import Ajv from "ajv";

const ajv = new Ajv({allErrors: true, removeAdditional: false});
function compileMessageValidator(maxTypeCharacters) {
  if (!Number.isSafeInteger(maxTypeCharacters) || maxTypeCharacters <= 0) {
    throw new Error("verified MCP type-length bound is invalid");
  }
  return ajv.compile({
    type: "object",
    required: ["type", "payload"],
    additionalProperties: false,
    properties: {
      type: { type: "string", maxLength: maxTypeCharacters },
      payload: { type: "object" }
    }
  });
}

export function validateIncomingMessage(raw, policy) {
  const { maxBytes, maxTypeCharacters } = policy || {};
  if (!Number.isSafeInteger(maxBytes) || maxBytes <= 0) {
    throw new Error("verified MCP message-size bound is invalid");
  }
  const validate = compileMessageValidator(maxTypeCharacters);
  if (Buffer.byteLength(raw, "utf8") > maxBytes) {
    throw new Error("MCP message too large");
  }
  const parsed = JSON.parse(raw);
  if (!validate(parsed)) {
    throw new Error(`Schema validation failed: ${ajv.errorsText(validate.errors)}`);
  }
  return parsed;
}

Operational notes: Apply separate limits for binary attachments, structured payloads, and recursive nested objects. Log validation failures with message type and size, not full message contents.

Production implementation

At the framed reader/parser, enforce policy-derived byte, depth, item and read-time bounds, validate the pinned MCP schema with unknown-field denial, reject duplicate/ambiguous encodings and release only an immutable validated object.

Independent verification

A separately credentialed verifier independently reads back or replays wire bytes, parser counters, schema result and downstream object hash with oversize, deep, slow, duplicate-key, truncation and schema-drift fixtures.

AID-H-028.004-G002

Concept:

If the client renders markdown, HTML, or rich content, sanitize before render and keep the renderer in a restricted environment. For terminal clients, strip ANSI escape sequences that can rewrite or hide output.

Example DOMPurify usage in an Electron renderer

// File: src/mcp/render-safe-html.mjs
import DOMPurify from "dompurify";

export function renderSafeHtml(untrustedHtml) {
  const clean = DOMPurify.sanitize(untrustedHtml, {
    ALLOWED_TAGS: ["b", "i", "strong", "em", "p", "ul", "li", "code", "pre", "a"],
    ALLOWED_ATTR: ["href", "title"],
    FORBID_TAGS: ["script", "iframe", "object"],
    FORBID_ATTR: ["onerror", "onclick", "onload"],
  });
  document.getElementById("content").innerHTML = clean;
}

Operational notes: In Electron, keep nodeIntegration disabled, enable contextIsolation, and apply a restrictive Content Security Policy. For terminal clients, strip or neutralize ANSI escape codes before display.

Production implementation

At the trusted renderer boundary, sanitize against the signed allowlist, neutralize active URLs and terminal controls, render in a non-privileged surface with no script or host bridge, and refuse content when sanitizer or policy readback fails.

Independent verification

A separately credentialed verifier independently reads back or replays sanitized bytes and effective DOM/terminal output with script, event-handler, CSS URL, bidi, ANSI, markdown-link and sanitizer-failure fixtures.

AID-H-028.004-G003

Concept:

The client should never silently execute a server-suggested action. Intercept suggested actions, convert them to a structured confirmation object, and show the user exactly what will happen.

Example confirmation object

{
  "server": "finance-mcp",
  "action": "create_ticket",
  "target": "jira://SEC-1234",
  "parameters": {
    "priority": "high",
    "summary": "Rotate exposed API key"
  },
  "risk_tier": "high"
}

Operational notes: Do not allow hidden parameters, free-form HTML in confirmations, or auto-approve for newly discovered servers. If you later add allowlists, keep them narrow and auditable.

Production implementation

At the client-controlled confirmation broker, canonicalize the exact action, target and parameters, display their immutable digest and policy risk, obtain fresh policy-required approval, then rehash before forwarding; cancellation, mutation or missing approval denies execution.

Independent verification

A separately credentialed verifier independently reads back or replays proposal, normalized view, approval receipt and final dispatched bytes with mutation, hidden parameter, stale approval, cancellation and dispatcher-failure fixtures.

AID-H-028.004-G004

Concept:

URL-mode elicitation is a special case of server-suggested action because navigation can leak tokens, open a phishing destination, trigger a magic link, or cause a browser-agent to perform a side effect. The client must not prefetch, auto-open, shorten, style-hide, or silently rewrite the destination. Show the full normalized URL in a read-only confirmation view, block preauthenticated URLs by default, and record the user's explicit consent before navigation. Browser/session isolation after consent belongs to AID-I-008.003.

Authoritative one-time consent ledger

CREATE TABLE url_navigation_decisions (
  decision_id TEXT PRIMARY KEY,
  consent_id TEXT UNIQUE,
  receipt_sha256 CHAR(64) UNIQUE,
  server_id TEXT NOT NULL,
  normalized_url TEXT NOT NULL,
  user_id TEXT NOT NULL,
  session_id TEXT NOT NULL,
  verdict TEXT NOT NULL CHECK (verdict IN ('allow', 'deny')),
  reason TEXT NOT NULL,
  receipt_expires_at_epoch BIGINT,
  decided_at_epoch BIGINT NOT NULL,
  CHECK (
    (verdict = 'allow' AND consent_id IS NOT NULL AND receipt_sha256 IS NOT NULL AND receipt_expires_at_epoch IS NOT NULL)
    OR verdict = 'deny'
  )
);
REVOKE ALL ON url_navigation_decisions FROM PUBLIC;
GRANT INSERT, SELECT ON url_navigation_decisions TO url_consent_broker;

URL confirmation gate

# File: mcp_client/url_elicitation_gate.py
from __future__ import annotations
import hashlib, ipaddress, json, math, os, subprocess, tempfile, time, uuid
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import parse_qsl, urlsplit, urlunsplit
import jwt
import psycopg

POLICY_KEY = Path("/opt/aidefend/trust/url-consent-policy.pub")
SENSITIVE_QUERY_KEYS = {"token", "code", "session", "auth", "signature", "sig", "key", "jwt", "sso"}

@dataclass(frozen=True)
class BrokerPolicy:
    policy_version: str; issuer: str; audience: str; key_id: str; public_key_pem: str
    allowed_origins: frozenset[str]; allowed_ports: frozenset[int]
    max_receipt_lifetime_seconds: int; clock_skew_seconds: int

@dataclass(frozen=True)
class UrlConsentDecision:
    decision_id: str; server_id: str; normalized_url: str; user_id: str; session_id: str
    verdict: str; reason: str; consent_id: str | None; receipt_sha256: str | None
    receipt_expires_at_epoch: int | None; decided_at_epoch: int

def verify_timeout() -> float:
    value = float(os.environ["VERIFIED_H028_URL_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(value) or value <= 0: raise ValueError("policy verification timeout must be finite and positive")
    return value

def load_policy(path: str, signature: str) -> BrokerPolicy:
    payload = Path(path).read_bytes()
    with tempfile.NamedTemporaryFile(prefix="aidefend-url-policy-", delete=False) as handle:
        handle.write(payload); snapshot = Path(handle.name)
    try:
        subprocess.run(["cosign", "verify-blob", "--key", str(POLICY_KEY), "--bundle", signature, str(snapshot)], check=True, capture_output=True, text=True, timeout=verify_timeout())
    finally:
        snapshot.unlink(missing_ok=True)
    raw = json.loads(payload)
    fields = {"schema_version", "policy_version", "issuer", "audience", "key_id", "public_key_path", "public_key_sha256", "allowed_origins", "allowed_ports", "max_receipt_lifetime_seconds", "clock_skew_seconds"}
    if not isinstance(raw, dict) or set(raw) != fields or raw["schema_version"] != "aidefend.url-consent-policy.v1": raise ValueError("URL consent policy schema mismatch")
    for field in ("policy_version", "issuer", "audience", "key_id", "public_key_path"):
        if not isinstance(raw[field], str) or not raw[field]: raise ValueError(field + " is required")
    origins = raw["allowed_origins"]
    if not isinstance(origins, list) or not origins or len(origins) != len(set(origins)) or any(not isinstance(origin, str) or not origin.startswith("https://") or any(char in origin[len("https://"):] for char in "/?#") for origin in origins): raise ValueError("allowed origin population is invalid")
    ports = raw["allowed_ports"]
    if not isinstance(ports, list) or not ports or len(ports) != len(set(ports)) or any(isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535 for port in ports): raise ValueError("allowed port population is invalid")
    maximum, skew = raw["max_receipt_lifetime_seconds"], raw["clock_skew_seconds"]
    if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1: raise ValueError("maximum consent lifetime must be positive")
    if isinstance(skew, bool) or not isinstance(skew, int) or skew < 0: raise ValueError("clock skew must be non-negative")
    key_path = Path(raw["public_key_path"])
    if not key_path.is_absolute() or key_path.is_symlink() or not key_path.is_file(): raise ValueError("receipt verification key must be one absolute non-link file")
    key_bytes = key_path.read_bytes(); key_digest = raw["public_key_sha256"]
    if not isinstance(key_digest, str) or len(key_digest) != 64 or any(char not in "0123456789abcdef" for char in key_digest) or hashlib.sha256(key_bytes).hexdigest() != key_digest: raise ValueError("receipt verification key digest differs")
    return BrokerPolicy(raw["policy_version"], raw["issuer"], raw["audience"], raw["key_id"], key_bytes.decode("utf-8"), frozenset(origins), frozenset(ports), maximum, skew)

def canonical_host(hostname: str) -> str:
    try:
        address = ipaddress.ip_address(hostname)
        return f"[{address.compressed}]" if address.version == 6 else address.compressed
    except ValueError:
        value = hostname.encode("idna").decode("ascii").lower().rstrip(".")
        if not value: raise ValueError("missing_hostname")
        return value

def normalize_for_display(raw_url: str, policy: BrokerPolicy) -> str:
    if not isinstance(raw_url, str) or raw_url != raw_url.strip() or any(ord(char) < 0x20 for char in raw_url) or "\\" in raw_url or "#" in raw_url: raise ValueError("ambiguous_navigation_url")
    parts = urlsplit(raw_url)
    if parts.scheme.lower() != "https" or not parts.hostname: raise ValueError("https_destination_required")
    if parts.username is not None or parts.password is not None: raise ValueError("userinfo_not_allowed")
    host = canonical_host(parts.hostname)
    try: port = parts.port or 443
    except ValueError as error: raise ValueError("invalid_port") from error
    if port not in policy.allowed_ports: raise ValueError("port_not_allowed")
    origin = "https://" + host + ("" if port == 443 else f":{port}")
    if origin not in policy.allowed_origins: raise ValueError("origin_not_allowed")
    return urlunsplit(("https", origin[len("https://"):], parts.path or "/", parts.query, ""))

def contains_preauth_material(normalized_url: str) -> bool:
    keys = {key.casefold() for key, _ in parse_qsl(urlsplit(normalized_url).query, keep_blank_values=True)}
    return bool(keys & SENSITIVE_QUERY_KEYS)

def persist_and_readback(decision: UrlConsentDecision, write_dsn: str, read_dsn: str) -> None:
    values = (decision.decision_id, decision.consent_id, decision.receipt_sha256, decision.server_id, decision.normalized_url, decision.user_id, decision.session_id, decision.verdict, decision.reason, decision.receipt_expires_at_epoch, decision.decided_at_epoch)
    with psycopg.connect(write_dsn) as connection:
        with connection.cursor() as cursor:
            cursor.execute("""INSERT INTO url_navigation_decisions (decision_id, consent_id, receipt_sha256, server_id, normalized_url, user_id, session_id, verdict, reason, receipt_expires_at_epoch, decided_at_epoch) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING RETURNING decision_id""", values)
            inserted = cursor.fetchone()
        connection.commit()
    if inserted is None: raise PermissionError("consent receipt replayed or decision identity duplicated")
    with psycopg.connect(read_dsn) as connection:
        with connection.cursor() as cursor:
            cursor.execute("""SELECT decision_id, consent_id, receipt_sha256, server_id, normalized_url, user_id, session_id, verdict, reason, receipt_expires_at_epoch, decided_at_epoch FROM url_navigation_decisions WHERE decision_id=%s""", (decision.decision_id,))
            readback = cursor.fetchone()
    if readback != values: raise RuntimeError("authoritative consent-ledger readback differs")

def deny(server_id: str, normalized: str, user_id: str, session_id: str, reason: str, write_dsn: str, read_dsn: str) -> UrlConsentDecision:
    decision = UrlConsentDecision(str(uuid.uuid4()), server_id, normalized, user_id, session_id, "deny", reason, None, None, None, int(time.time()))
    persist_and_readback(decision, write_dsn, read_dsn); return decision

def require_user_url_consent(server_id: str, raw_url: str, authenticated_user_id: str, authenticated_session_id: str, receipt_token: str | None, policy: BrokerPolicy, write_dsn: str, read_dsn: str) -> UrlConsentDecision:
    if any(not isinstance(value, str) or not value for value in (server_id, authenticated_user_id, authenticated_session_id)): raise ValueError("authenticated broker context is incomplete")
    normalized = normalize_for_display(raw_url, policy)
    if contains_preauth_material(normalized): return deny(server_id, normalized, authenticated_user_id, authenticated_session_id, "preauthenticated_url_blocked", write_dsn, read_dsn)
    if not isinstance(receipt_token, str) or not receipt_token: return deny(server_id, normalized, authenticated_user_id, authenticated_session_id, "signed_consent_missing", write_dsn, read_dsn)
    header = jwt.get_unverified_header(receipt_token)
    if not isinstance(header, dict) or set(header) != {"alg", "kid", "typ"} or header != {"alg": "ES256", "kid": policy.key_id, "typ": "JWT"}: raise PermissionError("consent receipt header denied")
    claims = jwt.decode(receipt_token, policy.public_key_pem, algorithms=["ES256"], audience=policy.audience, issuer=policy.issuer, leeway=policy.clock_skew_seconds, options={"require": ["iss", "aud", "sub", "session_id", "server_id", "normalized_url", "jti", "iat", "nbf", "exp"]})
    required = {"iss", "aud", "sub", "session_id", "server_id", "normalized_url", "jti", "iat", "nbf", "exp"}
    if not isinstance(claims, dict) or set(claims) != required: raise PermissionError("consent receipt claim population differs")
    expected = {"sub": authenticated_user_id, "session_id": authenticated_session_id, "server_id": server_id, "normalized_url": normalized}
    if any(claims[field] != value for field, value in expected.items()): raise PermissionError("consent receipt binding differs")
    if not isinstance(claims["jti"], str) or not claims["jti"]: raise PermissionError("consent ID missing")
    if any(isinstance(claims[field], bool) or not isinstance(claims[field], int) for field in ("iat", "nbf", "exp")): raise PermissionError("consent receipt time claims must be integer epochs")
    now = int(time.time())
    if not claims["iat"] <= claims["nbf"] < claims["exp"] or claims["iat"] > now + policy.clock_skew_seconds or claims["exp"] - claims["iat"] > policy.max_receipt_lifetime_seconds: raise PermissionError("consent receipt lifetime denied")
    receipt_sha256 = hashlib.sha256(receipt_token.encode()).hexdigest()
    decision = UrlConsentDecision(str(uuid.uuid4()), server_id, normalized, authenticated_user_id, authenticated_session_id, "allow", "signed_one_time_user_consent", claims["jti"], receipt_sha256, claims["exp"], now)
    persist_and_readback(decision, write_dsn, read_dsn); return decision

Verification test

# File: tests/test_url_elicitation_gate.py
import os, time, uuid
import jwt, pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from mcp_client.url_elicitation_gate import BrokerPolicy, require_user_url_consent

TEST_DSN = os.environ["TEST_URL_CONSENT_LEDGER_DSN"]

def fixture_policy(tmp_path):
    private = ec.generate_private_key(ec.SECP256R1())
    public_path = tmp_path / "consent.pub"
    public_path.write_bytes(private.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo))
    policy = BrokerPolicy("test-v1", "https://consent.test", "url-broker-test", "test-key", public_path.read_text(encoding="utf-8"), frozenset({"https://example.com"}), frozenset({443}), 300, 2)
    return private, policy

def issue(private, policy, *, server="mcp-server-a", user="user-7", session="session-9", url="https://example.com/docs?a=1&b=2", jti=None):
    now = int(time.time())
    claims = {"iss": policy.issuer, "aud": policy.audience, "sub": user, "session_id": session, "server_id": server, "normalized_url": url, "jti": jti or str(uuid.uuid4()), "iat": now, "nbf": now, "exp": now + 60}
    key = private.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())
    return jwt.encode(claims, key, algorithm="ES256", headers={"kid": policy.key_id, "typ": "JWT"})

def test_blocks_preauthenticated_url_and_reads_back(tmp_path):
    _, policy = fixture_policy(tmp_path)
    decision = require_user_url_consent("mcp-server-a", "https://example.com/login?token=abc", "user-7", "session-9", None, policy, TEST_DSN, TEST_DSN)
    assert decision.verdict == "deny" and decision.reason == "preauthenticated_url_blocked"

def test_signed_receipt_is_bound_and_single_use(tmp_path):
    private, policy = fixture_policy(tmp_path)
    token = issue(private, policy)
    first = require_user_url_consent("mcp-server-a", "HTTPS://Example.com/docs?a=1&b=2", "user-7", "session-9", token, policy, TEST_DSN, TEST_DSN)
    assert first.verdict == "allow" and first.normalized_url == "https://example.com/docs?a=1&b=2"
    with pytest.raises(PermissionError, match="replayed"):
        require_user_url_consent("mcp-server-a", "https://example.com/docs?a=1&b=2", "user-7", "session-9", token, policy, TEST_DSN, TEST_DSN)

def test_receipt_cannot_cross_session_or_destination(tmp_path):
    private, policy = fixture_policy(tmp_path); token = issue(private, policy)
    with pytest.raises(PermissionError, match="binding differs"):
        require_user_url_consent("mcp-server-a", "https://example.com/other", "user-7", "session-10", token, policy, TEST_DSN, TEST_DSN)

Action: Route every server-suggested URL, link-opening request, and navigation elicitation through the trusted broker before prefetch or browser launch. Derive user and session identities from authenticated middleware, require the signed one-time receipt for an allow, atomically consume it in the authoritative ledger, and require exact readback before launch. Retain the policy/receipt digests and database decision record; do not treat a caller boolean or local JSONL file as consent evidence.

Production implementation

At the trusted URL broker, parse and display the full canonical HTTPS destination, reject userinfo, fragments, disallowed ports/schemes and unapproved origins, prohibit prefetch and credential attachment, obtain fresh confirmation, then revalidate immediately before launch.

Independent verification

A separately credentialed verifier independently reads back or replays displayed and effective destination, DNS/redirect result and launch receipt with homograph, userinfo, redirect, DNS-change, alternate-port and cancel fixtures.

Open-source tools

• DOMPurify
• Ajv
• Zod (strict message-schema validation)
• Pydantic
• Electron
• @electron/fuses (Electron runtime hardening)
• nh3

Commercial / hosted tools

• Snyk Code (pre-release SAST verification only)
• Contrast Security (pre-release IAST/RASP verification only)

Threat mappings

MITRE ATLAS

• AML.T0011 User Execution
• AML.T0053 AI Agent Tool Invocation
• AML.T0011.002 User Execution: Poisoned AI Agent Tool
• AML.T0078 Drive-by Compromise (malicious server response used as delivery mechanism)
• AML.T0102 Generate Malicious Commands (server generates actions or payloads the client might execute)
• AML.T0105 Escape to Host (unsafe client handling allows escape to host context)
• AML.T0077 LLM Response Rendering
• AML.T0113 Steal Web Session Cookie (renderer hardening blocks malicious markup from reading session cookies)

MAESTRO

• Compromised Framework Components (L3) (parser or renderer exploitation)
• Integration Risks (L7) (unsafe response handling at the client boundary)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (exact-action confirmation and no-implicit-execution limit the client-side consequences of malicious MCP responses but do not detect instruction content later supplied to a model)
• LLM03:2026 Excessive Agency
• LLM06:2026 Unbounded Consumption (message byte, nesting-depth, and read-time bounds stop oversized or pathological MCP responses from exhausting the client, while model-inference budgets remain separate)

OWASP ML Top 10 2023

• ML09:2023 Output Integrity Attack (malicious output attacking downstream client logic)

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation (unsafe client handling of suggested actions)
• ASI05:2026 Unexpected Code Execution (RCE) (server response triggers client execution)
• ASI09:2026 Human-Agent Trust Exploitation (explicit confirmation prevents server-suggested actions from laundering consent through trusted client UI)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.3 Unsafe System / Browser / File Execution
• AISubtech-13.1.2 Memory Flooding
• AISubtech-13.1.4 Application Denial of Service
• AISubtech-2.1.1 Execution Approval Bypass
• AISubtech-4.3.1 Schema Inconsistencies
• AISubtech-4.3.4 Replay Exploitation
• AITech-12.1 Tool Exploitation
• AITech-13.1 Disruption of Availability
• AITech-2.1 Excessive Agency
• AITech-4.3 Protocol Manipulation

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (client parser and renderer are insecure integration surfaces)
• RA: Rogue Actions (server-suggested actions require explicit user confirmation to prevent rogue execution)

Databricks AI Security Framework 3.0

• Agents - Tools MCP Client 13.32: Client-Side Code Execution
• Agents - Tools MCP Client 13.28: UI/UX Deception