Scope boundary

Owns minimization, redaction, cache policy, retention, and deletion for logs, traces, temporary files, crash artifacts, and persistence directly controlled by the MCP or tool client. OS-wide histories outside the client's control are not claimed by this sub-technique.

Implementation guidance

• AID-H-028.003-G001: Apply automatic PII and secret redaction to client-side logs before writing to disk, and keep structured logs field-based so sensitive fields can be consistently scrubbed.
• AID-H-028.003-G002: Apply TTL-based retention and purge policies to client-managed caches, temp files, and protocol traces, and encrypt persistent caches where caching is necessary.
• AID-H-028.003-G003: Disable protocol-level debug logging in production and scrub client-generated crash reports so in-flight payloads and secrets are not written to disk by default.

AID-H-028.003-G001

Delivery level: reusable structured-logging formatter

Redact the fully rendered message, not only record.msg. Python logging keeps format arguments separately and interpolates them later, so changing the format string while leaving record.args intact leaks secrets supplied through logger.info("token=%s", token). Prefer a closed structured schema and never serialize arbitrary record extras.

Sanitize message arguments and exception text before the sink

# File: mcp_client/security_logging.py
from __future__ import annotations

import json
import logging
import os
import re
import traceback
from datetime import datetime, timezone
from typing import Any

EMAIL = re.compile(r"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b")
API_KEY = re.compile(r"\b(?:sk|pk|rk)_[A-Za-z0-9_-]{16,}\b")
BEARER = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._~+/-]{16,}=*")
ALLOWED_EXTRA_FIELDS = ("event_name", "request_id", "server_id", "method", "outcome")


def required_positive_int(name: str) -> int:
    raw = os.environ[name]
    value = int(raw)
    if value < 1:
        raise RuntimeError(f"{name} must be positive")
    return value


MAX_FIELD_CHARACTERS = required_positive_int("MCP_SECURITY_LOG_MAX_FIELD_CHARACTERS")


def redact_text(value: Any) -> str:
    text = str(value)
    truncated = text[:MAX_FIELD_CHARACTERS]
    redacted = EMAIL.sub("[REDACTED_EMAIL]", truncated)
    redacted = API_KEY.sub("[REDACTED_API_KEY]", redacted)
    redacted = BEARER.sub("Bearer [REDACTED_TOKEN]", redacted)
    return redacted


class SecurityJsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        try:
            rendered_message = record.getMessage()
        except Exception:
            rendered_message = "[UNRENDERABLE_LOG_MESSAGE]"
        event: dict[str, Any] = {
            "timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": redact_text(rendered_message),
        }
        for field in ALLOWED_EXTRA_FIELDS:
            value = getattr(record, field, None)
            if value is not None:
                event[field] = redact_text(value)
        if record.exc_info:
            exception_text = "".join(traceback.format_exception(*record.exc_info))
            event["exception"] = redact_text(exception_text)
        # The returned JSON is the only value sent to the handler. record.args,
        # arbitrary extras, stack locals, and raw exception objects are not emitted.
        return json.dumps(event, ensure_ascii=False, separators=(",", ":"), sort_keys=True)


def build_security_logger(path: str) -> logging.Logger:
    handler = logging.FileHandler(path, encoding="utf-8")
    handler.setFormatter(SecurityJsonFormatter())
    logger = logging.getLogger("mcp-client.security")
    logger.handlers.clear()
    logger.propagate = False
    logger.setLevel(logging.INFO)
    logger.addHandler(handler)
    return logger

Regression-test the actual sink bytes

# File: tests/test_security_logging.py
from pathlib import Path

from mcp_client.security_logging import build_security_logger


def test_format_arguments_and_exception_are_redacted(tmp_path: Path) -> None:
    output = tmp_path / "security.jsonl"
    logger = build_security_logger(str(output))
    synthetic_api_key_fixture = "sk_" + ("NOT_A_REAL_CREDENTIAL_" * 2)
    logger.info("user=%s token=%s", "alice@example.com", synthetic_api_key_fixture)
    try:
        raise RuntimeError("Bearer abcdefghijklmnopqrstuvwxyz012345")
    except RuntimeError:
        logger.exception("request failed for %s", "bob@example.com")
    for handler in logger.handlers:
        handler.flush()
    sink_bytes = output.read_text(encoding="utf-8")
    assert "alice@example.com" not in sink_bytes
    assert "bob@example.com" not in sink_bytes
    assert synthetic_api_key_fixture not in sink_bytes
    assert "abcdefghijklmnopqrstuvwxyz012345" not in sink_bytes
    assert "[REDACTED_" in sink_bytes

Application code should log event metadata and stable identifiers, not prompt, response, tool-result, authorization header, or raw argument bodies. Seed sink-level tests with every organization-specific secret/PII class before adding a new pattern. Encryption, retention, access control, and emergency full-fidelity capture remain explicit logging-platform dependencies; a failure to construct the approved sanitized sink must stop security logging rather than fall back to an unredacted handler.

Action: Make this formatter the only production handler path for the covered security logger and block deployment if the regression fixtures expose formatted arguments, exception secrets, arbitrary extras, or oversized raw fields.

AID-H-028.003-G002

Concept:

Do not rely on indefinite local caches. Every client-managed cache should have a retention time and a cleanup job. If the cache must persist across restarts, prefer an encrypted store.

Example cache cleanup job

import os
from pathlib import Path
import os
import time

CACHE_DIR = Path("./cache")
TTL_SECONDS = int(os.environ["VERIFIED_MCP_CACHE_TTL_SECONDS"])
if TTL_SECONDS < 1:
    raise RuntimeError("verified cache TTL must be positive")

def purge_expired_files():
    now = time.time()
    for path in CACHE_DIR.glob("**/*"):
        if path.is_file():
            age = now - path.stat().st_mtime
            if age > TTL_SECONDS:
                path.unlink(missing_ok=True)

if __name__ == "__main__":
    purge_expired_files()

Operational notes: Keep a small inventory of all client-managed artifact paths. Review that inventory during release testing so new cache locations do not silently bypass the purge job.

Production implementation

At cache/temp creation and scheduled cleanup, derive expiry and encryption requirements from verified policy, bind objects to server/principal/tenant scope, deny persistence without an expiry, purge expired generations and independently confirm deletion/read denial.

Independent verification

A separately credentialed verifier independently reads back or replays the complete object inventory, encrypted-at-rest metadata, expiry queue, deletion result and post-purge read probe across restart and cleanup failure.

AID-H-028.003-G003

Delivery level: production startup gate

Do not infer production from an optional environment string. The launcher verifies one signed telemetry profile before importing the MCP client, rejects every unknown mode, and treats process-level debug flags only as untrusted override attempts. The platform deployment adapter is responsible for mounting the correct profile for the workload identity.

Verify the exact profile and deny raw capture

# File: mcp_client/telemetry_startup_gate.py
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any


@dataclass(frozen=True)
class TelemetryProfile:
    profile_version: str
    mode: str
    allow_protocol_debug: bool
    allow_payload_bodies: bool
    allow_local_crash_dumps: bool
    profile_sha256: str


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = value
    return result


def enabled_environment_flag(name: str) -> bool:
    value = os.environ.get(name)
    if value is None:
        return False
    # CPython enables PYTHONFAULTHANDLER for every non-empty value,
    # including "0", "false", and whitespace. Mirror that behavior.
    if name == "PYTHONFAULTHANDLER":
        return value != ""
    normalized = value.strip().lower()
    if normalized in {"1", "true", "yes", "on"}:
        return True
    if normalized in {"0", "false", "no", "off", ""}:
        return False
    raise RuntimeError(f"ambiguous debug flag: {name}")


def load_verified_profile(
    profile_path: Path, bundle_path: Path, public_key: Path,
    timeout_seconds: float,
) -> TelemetryProfile:
    if not math.isfinite(timeout_seconds) or timeout_seconds <= 0:
        raise ValueError("profile verification timeout must be finite and positive")
    payload = profile_path.read_bytes()
    with tempfile.NamedTemporaryFile(prefix="telemetry-profile-", delete=False) as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
        snapshot = Path(handle.name)
    try:
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(public_key),
                "--bundle", str(bundle_path), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
    finally:
        snapshot.unlink(missing_ok=True)
    raw = json.loads(
        payload.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"non-finite JSON number: {value}")),
    )
    required = {
        "schema_version", "profile_version", "mode",
        "allow_protocol_debug", "allow_payload_bodies",
        "allow_local_crash_dumps",
    }
    if not isinstance(raw, dict) or set(raw) != required:
        raise ValueError("telemetry profile schema mismatch")
    if raw["schema_version"] != "aidefend.mcp-telemetry-profile.v1":
        raise ValueError("unsupported telemetry profile")
    if raw["mode"] not in {"production", "staging", "development"}:
        raise ValueError("unknown deployment mode")
    if not isinstance(raw["profile_version"], str) or not raw["profile_version"]:
        raise ValueError("profile_version is required")
    for field in (
        "allow_protocol_debug", "allow_payload_bodies", "allow_local_crash_dumps"
    ):
        if type(raw[field]) is not bool:
            raise ValueError(f"{field} must be boolean")
    profile = TelemetryProfile(
        profile_version=raw["profile_version"],
        mode=raw["mode"],
        allow_protocol_debug=raw["allow_protocol_debug"],
        allow_payload_bodies=raw["allow_payload_bodies"],
        allow_local_crash_dumps=raw["allow_local_crash_dumps"],
        profile_sha256=hashlib.sha256(payload).hexdigest(),
    )
    if profile.mode == "production" and any((
        profile.allow_protocol_debug,
        profile.allow_payload_bodies,
        profile.allow_local_crash_dumps,
    )):
        raise PermissionError("production profile enables forbidden raw telemetry")
    return profile


def enforce_startup(profile: TelemetryProfile) -> None:
    flag_permissions = {
        "DEBUG_PROTOCOL_LOGGING": profile.allow_protocol_debug,
        "HTTP_DEBUG": profile.allow_protocol_debug,
        "MCP_TRACE_PAYLOADS": profile.allow_payload_bodies,
        "PYTHONFAULTHANDLER": profile.allow_local_crash_dumps,
    }
    enabled_overrides = [
        name for name in flag_permissions if enabled_environment_flag(name)
    ]
    denied_overrides = [
        name for name in enabled_overrides if not flag_permissions[name]
    ]
    if denied_overrides:
        raise PermissionError(
            "runtime attempted capture not permitted by its exact signed flag: "
            + ",".join(denied_overrides)
        )
    if profile.mode == "production" and enabled_overrides:
        raise PermissionError("production runtime contains a raw-debug override")


PROFILE_PATH = Path("/opt/aidefend/policy/mcp-telemetry-profile.json")
PROFILE_BUNDLE = Path("/opt/aidefend/policy/mcp-telemetry-profile.sigstore.json")
PROFILE_PUBLIC_KEY = Path("/opt/aidefend/trust/mcp-telemetry-profile.pub")


def admitted_profile_from_environment() -> TelemetryProfile:
    timeout_raw = os.environ.get("MCP_TELEMETRY_PROFILE_VERIFY_TIMEOUT_SECONDS")
    if not isinstance(timeout_raw, str) or not timeout_raw:
        raise RuntimeError("profile verification timeout is missing")
    profile = load_verified_profile(
        PROFILE_PATH, PROFILE_BUNDLE, PROFILE_PUBLIC_KEY, float(timeout_raw)
    )
    enforce_startup(profile)
    return profile

Regression-test native and application flag semantics

# File: tests/test_telemetry_startup_gate.py
import pytest

from mcp_client.telemetry_startup_gate import (
    TelemetryProfile, enforce_startup, enabled_environment_flag,
)


FALSE_PROFILE = TelemetryProfile(
    profile_version="test", mode="production",
    allow_protocol_debug=False, allow_payload_bodies=False,
    allow_local_crash_dumps=False, profile_sha256="0" * 64,
)
APP_FLAGS = ("DEBUG_PROTOCOL_LOGGING", "HTTP_DEBUG", "MCP_TRACE_PAYLOADS")


def clear_flags(monkeypatch: pytest.MonkeyPatch) -> None:
    for name in (*APP_FLAGS, "PYTHONFAULTHANDLER"):
        monkeypatch.delenv(name, raising=False)


@pytest.mark.parametrize("value", ["0", "false", " "])
def test_python_faulthandler_any_nonempty_value_is_enabled(
    monkeypatch: pytest.MonkeyPatch, value: str,
) -> None:
    clear_flags(monkeypatch)
    monkeypatch.setenv("PYTHONFAULTHANDLER", value)
    assert enabled_environment_flag("PYTHONFAULTHANDLER") is True
    with pytest.raises(PermissionError):
        enforce_startup(FALSE_PROFILE)


@pytest.mark.parametrize("name", APP_FLAGS)
@pytest.mark.parametrize("value", ["0", "false", "no", "off", ""])
def test_app_owned_flags_keep_explicit_false_semantics(
    monkeypatch: pytest.MonkeyPatch, name: str, value: str,
) -> None:
    clear_flags(monkeypatch)
    monkeypatch.setenv(name, value)
    assert enabled_environment_flag(name) is False
    enforce_startup(FALSE_PROFILE)

PYTHONFAULTHANDLER=0 and PYTHONFAULTHANDLER=false still enable CPython fault-handler output because both values are non-empty. In contrast, 0, false, no, off, and the empty string remain explicit disabled values for the three application-owned flags.

Call the gate before client imports

# File: run_mcp_client.py
from mcp_client.telemetry_startup_gate import admitted_profile_from_environment

TELEMETRY_PROFILE = admitted_profile_from_environment()

# Import only after the gate, so protocol libraries cannot initialize debug sinks first.
from mcp_client.application import run

raise SystemExit(run(telemetry_profile=TELEMETRY_PROFILE))

The application logging and crash-report adapters must consume this immutable profile and expose effective configuration for readback; they must not read APP_ENV or independently reinterpret optional debug variables. Release tests should exercise every flag independently and in combination, explicitly proving that PYTHONFAULTHANDLER=0 and false are enabled while the same values disable application-owned flags, then force a secret-bearing protocol error and crash and inspect every supported local and uploaded artifact. Missing profile, invalid signature, unknown mode, ambiguous flag, production raw-capture option, or effective-config mismatch blocks startup or release.

Action: Start the application only through this wrapper; do not import application code or open a listener until the pinned signed telemetry profile passes every production-mode check.

Open-source tools

• Microsoft Presidio
• structlog (structured logging with redaction hooks)
• Loguru (structured logging with redaction hooks)
• pino (structured logging with redaction hooks)
• SQLCipher (encrypted local SQLite caches where persistent cache is necessary)
• Vector (log redaction and remap transforms)

Commercial / hosted tools

• Google Cloud Sensitive Data Protection API (DLP scanning)
• Microsoft Purview Data Loss Prevention
• Forcepoint Data Loss Prevention
• Broadcom Symantec Data Loss Prevention

Threat mappings

MITRE ATLAS

• AML.T0057 LLM Data Leakage (responses retained locally in clear text)
• AML.T0085 Data from AI Services (AI service data exposed through local artifacts)

MAESTRO

• Data Exfiltration (L2) (cache or temp-file leakage)
• Data Leakage through Observability (L5) (debug logs, traces, crash reports)

OWASP LLM Top 10 2026

• LLM02:2026 Sensitive Information Disclosure

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.036 Leaking information from user interactions (interaction data persists in local logs or traces)

Cisco Integrated AI Security and Safety Framework

• AISubtech-16.1.1 Logging Sensitive Conversations
• AISubtech-7.4.1 Token Theft
• AISubtech-8.3.3 Personally Identifiable Information Exposure
• AITech-16.1 Eavesdropping
• AITech-7.4 Token Manipulation
• AITech-8.2 Data Exfiltration / Exposure
• AITech-8.3 Information Disclosure

Google Secure AI Framework 2.0 - Risks

• SDD: Sensitive Data Disclosure (client-side artifacts leak sensitive AI interaction data)
• EDH: Excessive Data Handling (client caches and logs retain data beyond session scope)

Databricks AI Security Framework 3.0

• Agents - Tools MCP Client 13.30: Client-Side Data Leakage