AID-H-028.005: Client Request, Cache & Application-State Lifecycle Security
2026-07-28, isolate every request's authenticated server origin, client identity, capabilities, tenant, trace context, token, cache scope, and explicit application handles. Honor ttlMs and cacheScope, expire artifacts under versioned policy, and prevent private cache, handles, tokens, or context from crossing authorization boundaries.Pillars: app, infra
Lifecycle: building, operation
Scope boundary
Owns client-controlled per-request identity, token reference, trace context, cache scope, expiry, and explicit application-handle isolation. It must not recreate protocol sessions, sticky affinity, or hidden global client state.
AID-H-034.006: AID-H-028.005 isolates and expires client-side handle references and cached state; AID-H-034.006 authorizes each opaque application handle against server-side principal, tenant, object, scope, and version. Client partitioning cannot authorize the object represented by a server handle.
Implementation guidance
- AID-H-028.005-G001: Isolate per-request client context and explicit application handles by server origin, principal, client instance, tenant, and authorization context.
- AID-H-028.005-G002: Enforce MCP ttlMs and cacheScope on cacheable list and resource results, with authorization-bound private keys and conservative invalidation.
- AID-H-028.005-G003: Use channel-bound or proof-of-possession token mechanisms where supported to reduce replay risk from stolen client tokens.
- AID-H-028.005-G004: Use OAuth authorization-code flow with PKCE, state validation, protected-resource metadata, and resource-parameter binding for MCP server authorization.
AID-H-028.005-G001
Concept
MCP 2026-07-28 sends protocol selection and client capabilities on every request; clients SHOULD also send optional client information unless configured not to. Build a fresh request context from authenticated facts and per-request metadata. If a tool returns an application handle such as browser_id or basket_id, store only a reference bound to the same origin and authorization partition; the handle is not authentication.
# File: mcp_client/request_context.py
from __future__ import annotations
from dataclasses import dataclass, field
from time import time
@dataclass(frozen=True)
class RequestPartition:
server_origin: str
subject: str
client_instance_id: str
tenant_id: str
authorization_context_sha256: str
@dataclass
class ApplicationHandle:
handle_type: str
value_ref: str
expires_at: float
scopes: frozenset[str]
HANDLES: dict[tuple[RequestPartition, str], ApplicationHandle] = {}
def put_handle(partition: RequestPartition, handle_id: str, handle: ApplicationHandle) -> None:
if handle.expires_at <= time() or not handle.scopes:
raise ValueError("invalid application handle")
HANDLES[(partition, handle_id)] = handle
def get_handle(partition: RequestPartition, handle_id: str, required_scope: str) -> ApplicationHandle:
handle = HANDLES.get((partition, handle_id))
if handle is None or handle.expires_at <= time():
raise PermissionError("application handle unavailable")
if required_scope not in handle.scopes:
raise PermissionError("application handle scope mismatch")
return handle
def teardown_partition(partition: RequestPartition) -> None:
for key in [key for key in HANDLES if key[0] == partition]:
del HANDLES[key]
Action: Reconstruct the partition for every request from the verified server origin, current token claims, client instance, tenant, and authorization-context digest. Deny cross-partition handle access, remove expired handles, and clear a partition after logout, token revocation, tenant switch, server identity change, or policy-directed retention expiry.
Production implementation
At request creation, lookup and teardown, derive a keyed partition from authenticated context, store immutable request/handle bindings with policy expiry, reject cross-partition or stale access and purge the complete partition on identity or authorization transition.
Independent verification
A separately credentialed verifier independently reads back or replays request/handle inventory and effective lookup decisions across cross-server, cross-user, tenant-switch, logout, revocation, expiry and restart fixtures.
AID-H-028.005-G002
Concept
MCP 2026-07-28 makes server/discover, list methods, and resources/read cacheable only through explicit ttlMs and cacheScope hints. Cache only resultType: complete; input_required is never cacheable. Treat each page independently, bind private entries to the authorization context, and never invent missing v2 hints.
Strict cache gate
// File: src/cache/mcpResultCache.js
import crypto from "node:crypto";
const VERSION = "2026-07-28";
const CACHEABLE = new Set([
"server/discover", "tools/list", "prompts/list", "resources/list",
"resources/templates/list", "resources/read"
]);
const hash = value => crypto.createHash("sha256")
.update(JSON.stringify(value)).digest("hex");
function readHints(result) {
if (result?.resultType !== "complete") throw new Error("non_complete_result_not_cacheable");
if (!Object.hasOwn(result, "ttlMs") ||
!Number.isSafeInteger(result.ttlMs) || result.ttlMs < 0) {
throw new Error("v2_cache_result_missing_valid_ttlMs");
}
if (!Object.hasOwn(result, "cacheScope") ||
!["public", "private"].includes(result.cacheScope)) {
throw new Error("v2_cache_result_missing_valid_cacheScope");
}
return { ttlMs: result.ttlMs, cacheScope: result.cacheScope };
}
export function planCacheWrite(input) {
if (input.protocolVersion !== VERSION || !CACHEABLE.has(input.method)) {
throw new Error("unsupported_protocol_or_method_for_cache");
}
const hints = readHints(input.result);
if (hints.ttlMs === 0) return Object.freeze({ store: false, reason: "ttl_zero" });
const keyMaterial = {
protocolVersion: VERSION,
serverOrigin: input.serverOrigin,
method: input.method,
paramsSha256: hash(input.params || {}),
pageCursor: input.pageCursor ?? null,
contractSha256: input.contractSha256
};
if (hints.cacheScope === "private") {
if (!input.authorizationContextSha256) {
throw new Error("private_cache_requires_auth_binding");
}
keyMaterial.authorizationContextSha256 = input.authorizationContextSha256;
} else if (!input.publicResultPolicyApproved) {
throw new Error("public_cache_result_not_locally_approved");
}
return Object.freeze({
store: true,
key: hash(keyMaterial),
scope: hints.cacheScope,
expiresAtMs: Date.now() + hints.ttlMs,
resultSha256: hash(input.result)
});
}Action: emit a conformance finding and skip storage when an v2 cacheable result omits or malforms either hint. Invalidate on expiry and on server identity, contract, policy, authorization, or page change. Never reuse a private entry across subjects, tenants, scopes, clients, or token authorization contexts; allow public reuse only after independent local classification confirms no caller-specific data.
Independent verification
Replay every cacheable method including server/discover, pagination, zero TTL, malformed or missing hints, input_required, cross-user/tenant/scope access, contract change, and public/private misclassification; compare stored keys, expiry, and readback with live result bytes.
AID-H-028.005-G003
Concept:
Bearer tokens are replayable by anyone who steals them. Proof-of-possession mechanisms such as DPoP bind the token to a client-held key so the attacker also has to steal the private key, not just the token string.
Step 1: Generate a DPoP key and proof JWT
Use a client-held asymmetric key and send a signed proof on every token and resource request.
# File: mcp_client/dpop.py
from __future__ import annotations
import base64
import hashlib
import time
import uuid
import keyring
from jwcrypto import jwk, jwt
KEYRING_SERVICE = "aidefend-mcp-client-dpop"
KEYRING_ACCOUNT = "active-p256-private-jwk"
def load_client_key() -> jwk.JWK:
encoded = keyring.get_password(KEYRING_SERVICE, KEYRING_ACCOUNT)
if encoded is None:
generated = jwk.JWK.generate(kty="EC", crv="P-256")
encoded = generated.export_private()
keyring.set_password(KEYRING_SERVICE, KEYRING_ACCOUNT, encoded)
return jwk.JWK.from_json(encoded)
client_key = load_client_key()
def b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def build_dpop_proof(
http_method: str, http_url: str, access_token: str | None = None,
nonce: str | None = None,
) -> str:
public_jwk = client_key.export_public(as_dict=True)
claims = {
"jti": str(uuid.uuid4()),
"htm": http_method.upper(),
"htu": http_url,
"iat": int(time.time()),
}
if access_token is not None:
claims["ath"] = b64url(hashlib.sha256(access_token.encode("ascii")).digest())
if nonce:
claims["nonce"] = nonce
token = jwt.JWT(
header={"alg": "ES256", "typ": "dpop+jwt", "jwk": public_jwk},
claims=claims,
)
token.make_signed_token(client_key)
return token.serialize()
Step 2: Use the proof on both token issuance and resource requests
The authorization server returns a DPoP-bound token only if the token request itself carried a valid proof. Resource calls must then send the returned DPoP-bound access token in the Authorization header and a fresh DPoP proof.
# File: mcp_client/session.py
import json
import math
from urllib.parse import urlsplit, urlunsplit
import requests
from mcp_client.dpop import build_dpop_proof
def canonical_https(value: str) -> str:
parsed = urlsplit(value)
if (
parsed.scheme != "https" or not parsed.hostname
or parsed.username is not None or parsed.password is not None
or parsed.query or parsed.fragment or not parsed.path.startswith("/")
or value != urlunsplit(("https", parsed.netloc.lower(), parsed.path, "", ""))
):
raise ValueError("DPoP endpoint must be the exact canonical HTTPS URL")
return value
def bounded_json(response: requests.Response, maximum_bytes: int) -> dict:
response.raise_for_status()
raw = bytearray()
for chunk in response.iter_content(chunk_size=min(65536, maximum_bytes + 1)):
if chunk:
raw.extend(chunk)
if len(raw) > maximum_bytes:
raise RuntimeError("DPoP response exceeds the signed decoded-byte cap")
value = json.loads(bytes(raw))
if not isinstance(value, dict):
raise RuntimeError("DPoP response root must be an object")
return value
def fetch_dpop_token(token_url: str, client_id: str, client_secret: str, timeout_seconds: float, max_response_bytes: int, ca_bundle: str, client_cert: tuple[str, str]) -> str:
token_url = canonical_https(token_url)
if not math.isfinite(timeout_seconds) or timeout_seconds <= 0 or max_response_bytes <= 0:
raise ValueError("signed DPoP transport bounds are invalid")
token_proof = build_dpop_proof("POST", token_url)
with requests.post(
token_url,
data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
},
headers={"DPoP": token_proof},
timeout=timeout_seconds,
allow_redirects=False,
verify=ca_bundle, cert=client_cert, stream=True,
) as response:
body = bounded_json(response, max_response_bytes)
if body.get("token_type", "").lower() != "dpop":
raise RuntimeError("Authorization server did not issue a DPoP-bound token")
return body["access_token"]
def call_mcp_server(resource_url: str, access_token: str, timeout_seconds: float, max_response_bytes: int, ca_bundle: str, client_cert: tuple[str, str]) -> dict:
resource_url = canonical_https(resource_url)
if not math.isfinite(timeout_seconds) or timeout_seconds <= 0 or max_response_bytes <= 0:
raise ValueError("signed MCP transport bounds are invalid")
body = {
"jsonrpc": "2.0", "id": "discover-1", "method": "server/discover",
"params": {"_meta": {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {"name": "aidefend-client", "version": "1.0.0"},
"io.modelcontextprotocol/clientCapabilities": {},
}},
}
resource_proof = build_dpop_proof("POST", resource_url, access_token=access_token)
with requests.post(
resource_url, json=body,
headers={
"Authorization": f"DPoP {access_token}", "DPoP": resource_proof,
"MCP-Protocol-Version": "2026-07-28", "Mcp-Method": "server/discover",
},
timeout=timeout_seconds,
allow_redirects=False,
verify=ca_bundle, cert=client_cert, stream=True,
) as response:
return bounded_json(response, max_response_bytes)
Step 3: Verify replay resistance
Capture a token from staging, then replay it without the private key or with a mismatched DPoP proof. The server should reject the request with 401 or invalid_dpop_proof.
Action: Prefer DPoP or another proof-of-possession mechanism for MCP client authorization whenever the provider supports it. If not, compensate with shorter TTLs, hardware-backed key storage, and replay-detection telemetry, but treat plain bearer tokens as a weaker profile.
Production implementation
At token issuance and each resource call, generate a fresh DPoP proof from a protected client key, bind method/URI/nonce, require a DPoP-bound token response, and deny bearer fallback or replay when the signed provider profile requires proof-of-possession.
Independent verification
A separately credentialed verifier independently reads back or replays key identity, proof claims, token cnf binding and server response using missing-key, wrong URI/method, reused jti, nonce, bearer-downgrade and stolen-token fixtures.
AID-H-028.005-G004
Concept:
Bind every OAuth authorization to one signature-verified local route: the exact MCP protected-resource identifier, the allowed authorization-server issuer, client, redirect URI, scopes, trust store, and transport limits. Protected-resource metadata is discovery input, not a trust anchor. Derive its URL from the configured resource per RFC 9728, require its resource value to match exactly, and select only an issuer already authorized by local policy. Never accept a caller-supplied metadata URL or choose the first issuer returned by the resource.
Reusable OAuth route gate
# File: mcp_client/oauth_route_gate.py
from __future__ import annotations
import base64
import hashlib
import json
import secrets
import time
from dataclasses import asdict, dataclass
from typing import Protocol
from urllib.parse import urlencode, urlsplit, urlunsplit
import requests
def _no_duplicate_keys(pairs: list[tuple[str, object]]) -> dict:
result: dict = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
@dataclass(frozen=True)
class OAuthRoute:
route_id: str
policy_revision: str
resource: str
selected_issuer: str
allowed_issuers: tuple[str, ...]
client_id: str
redirect_uri: str
scopes: tuple[str, ...]
timeout_seconds: float
maximum_response_bytes: int
flow_ttl_seconds: int
ca_bundle: str
client_cert: tuple[str, str]
@dataclass(frozen=True)
class OAuthStart:
route_id: str
policy_revision: str
resource: str
issuer: str
client_id: str
redirect_uri: str
token_endpoint: str
state: str
code_verifier: str
expires_at: int
response_issuer_required: bool
authorization_url: str
class VerifiedRouteProvider(Protocol):
def load_verified(self, route_id: str) -> OAuthRoute:
"""Return a route only after signature, schema, expiry, and revision checks."""
raise RuntimeError("protocol-only method")
class SingleUseFlowStore(Protocol):
def put_once(self, state: str, record: dict, ttl_seconds: int) -> None:
"""Atomically insert an encrypted record; reject an existing state."""
raise RuntimeError("protocol-only method")
def pop_once(self, state: str) -> dict | None:
"""Atomically consume one unexpired record."""
raise RuntimeError("protocol-only method")
class OAuthEvidenceSink(Protocol):
def write_and_readback(self, evidence: dict) -> str:
"""Durably store non-secret evidence and return a receipt."""
raise RuntimeError("protocol-only method")
def _https_identifier(value: str, *, allow_query: bool) -> str:
parts = urlsplit(value)
if (
parts.scheme != "https" or not parts.hostname
or parts.username is not None or parts.password is not None
or parts.fragment or (parts.query and not allow_query)
):
raise ValueError("expected an unambiguous HTTPS identifier")
return value
def protected_resource_metadata_url(resource: str) -> str:
# RFC 9728: insert the well-known name between the host and resource path.
_https_identifier(resource, allow_query=True)
parts = urlsplit(resource)
path = "/.well-known/oauth-protected-resource" + (parts.path or "")
return urlunsplit((parts.scheme, parts.netloc, path, parts.query, ""))
def authorization_server_metadata_url(issuer: str) -> str:
# RFC 8414 path-aware issuer transformation.
_https_identifier(issuer, allow_query=False)
parts = urlsplit(issuer)
path = "/.well-known/oauth-authorization-server" + (parts.path or "")
return urlunsplit((parts.scheme, parts.netloc, path, "", ""))
def _endpoint(value: object) -> str:
if not isinstance(value, str):
raise ValueError("OAuth endpoint is absent")
return _https_identifier(value, allow_query=True)
def _read_json(response: requests.Response, maximum_bytes: int) -> dict:
response.raise_for_status()
raw = bytearray()
for chunk in response.iter_content(chunk_size=min(65536, maximum_bytes + 1)):
if chunk:
raw.extend(chunk)
if len(raw) > maximum_bytes:
raise RuntimeError("OAuth metadata or token response exceeds policy cap")
value = json.loads(bytes(raw), object_pairs_hook=_no_duplicate_keys)
if not isinstance(value, dict):
raise RuntimeError("OAuth JSON root must be an object")
return value
def _get_json(url: str, route: OAuthRoute) -> dict:
with requests.get(
_https_identifier(url, allow_query=True),
timeout=route.timeout_seconds,
allow_redirects=False,
verify=route.ca_bundle,
cert=route.client_cert,
stream=True,
) as response:
return _read_json(response, route.maximum_response_bytes)
def _validate_route(route: OAuthRoute) -> None:
_https_identifier(route.resource, allow_query=True)
_https_identifier(route.selected_issuer, allow_query=False)
if (
route.selected_issuer not in route.allowed_issuers
or len(route.allowed_issuers) != len(set(route.allowed_issuers))
or not route.route_id or not route.policy_revision or not route.client_id
or not route.redirect_uri or not route.scopes
or route.timeout_seconds <= 0 or route.maximum_response_bytes <= 0
or route.flow_ttl_seconds <= 0
):
raise RuntimeError("invalid verified OAuth route")
def _b64url(value: bytes) -> str:
return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii")
def start_oauth(
route_id: str,
provider: VerifiedRouteProvider,
store: SingleUseFlowStore,
) -> OAuthStart:
route = provider.load_verified(route_id)
_validate_route(route)
resource_metadata = _get_json(
protected_resource_metadata_url(route.resource), route
)
if resource_metadata.get("resource") != route.resource:
raise RuntimeError("protected_resource_metadata_mismatch")
advertised = resource_metadata.get("authorization_servers")
if (
not isinstance(advertised, list) or not advertised
or any(not isinstance(item, str) for item in advertised)
or len(advertised) != len(set(advertised))
):
raise RuntimeError("invalid_authorization_servers")
trusted_candidates = set(advertised).intersection(route.allowed_issuers)
if trusted_candidates != {route.selected_issuer}:
raise RuntimeError("authorization_server_not_uniquely_policy_selected")
server_metadata = _get_json(
authorization_server_metadata_url(route.selected_issuer), route
)
if server_metadata.get("issuer") != route.selected_issuer:
raise RuntimeError("authorization_server_issuer_mismatch")
methods = server_metadata.get("code_challenge_methods_supported")
if not isinstance(methods, list) or "S256" not in methods:
raise RuntimeError("pkce_s256_not_advertised")
authorization_endpoint = _endpoint(
server_metadata.get("authorization_endpoint")
)
token_endpoint = _endpoint(server_metadata.get("token_endpoint"))
state = secrets.token_urlsafe(32)
verifier = secrets.token_urlsafe(64)
challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest())
parameters = {
"response_type": "code",
"client_id": route.client_id,
"redirect_uri": route.redirect_uri,
"scope": " ".join(route.scopes),
"state": state,
"code_challenge": challenge,
"code_challenge_method": "S256",
"resource": route.resource,
}
separator = "&" if urlsplit(authorization_endpoint).query else "?"
started = OAuthStart(
route_id=route.route_id,
policy_revision=route.policy_revision,
resource=route.resource,
issuer=route.selected_issuer,
client_id=route.client_id,
redirect_uri=route.redirect_uri,
token_endpoint=token_endpoint,
state=state,
code_verifier=verifier,
expires_at=int(time.time()) + route.flow_ttl_seconds,
response_issuer_required=(
server_metadata.get(
"authorization_response_iss_parameter_supported"
) is True
),
authorization_url=authorization_endpoint + separator + urlencode(parameters),
)
stored = asdict(started)
stored.pop("authorization_url")
store.put_once(state, stored, route.flow_ttl_seconds)
return started
def exchange_code(
route_id: str,
callback_state: str,
callback_issuer: str | None,
code: str,
provider: VerifiedRouteProvider,
store: SingleUseFlowStore,
evidence_sink: OAuthEvidenceSink,
) -> dict:
if not callback_state or not code:
raise RuntimeError("missing_oauth_callback_parameter")
start = store.pop_once(callback_state)
if start is None:
raise RuntimeError("unknown_or_replayed_oauth_state")
route = provider.load_verified(route_id)
_validate_route(route)
expected = {
"route_id": route.route_id,
"policy_revision": route.policy_revision,
"resource": route.resource,
"issuer": route.selected_issuer,
"client_id": route.client_id,
"redirect_uri": route.redirect_uri,
}
if any(start.get(key) != value for key, value in expected.items()):
raise RuntimeError("oauth_route_changed_or_record_mismatch")
if not isinstance(start.get("expires_at"), int) or start["expires_at"] <= int(time.time()):
raise RuntimeError("oauth_state_expired")
issuer_required = start.get("response_issuer_required") is True
if issuer_required and callback_issuer is None:
raise RuntimeError("authorization_response_issuer_missing")
if callback_issuer is not None and callback_issuer != route.selected_issuer:
raise RuntimeError("authorization_response_issuer_mismatch")
if _endpoint(start.get("token_endpoint")) != start["token_endpoint"]:
raise RuntimeError("invalid_stored_token_endpoint")
with requests.post(
start["token_endpoint"],
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": route.redirect_uri,
"client_id": route.client_id,
"code_verifier": start["code_verifier"],
"resource": route.resource,
},
timeout=route.timeout_seconds,
allow_redirects=False,
verify=route.ca_bundle,
cert=route.client_cert,
stream=True,
) as response:
token = _read_json(response, route.maximum_response_bytes)
if (
not isinstance(token.get("access_token"), str)
or not token["access_token"]
or str(token.get("token_type", "")).lower() not in {"bearer", "dpop"}
or not isinstance(token.get("expires_in"), int)
or token["expires_in"] <= 0
):
raise RuntimeError("invalid_token_response")
evidence = {
"route_id": route.route_id,
"policy_revision": route.policy_revision,
"resource": route.resource,
"issuer": route.selected_issuer,
"state_sha256": hashlib.sha256(
callback_state.encode("utf-8")
).hexdigest(),
"pkce_method": "S256",
"response_issuer_required": issuer_required,
"response_issuer_validated": callback_issuer is not None,
"token_type": str(token["token_type"]).lower(),
}
evidence_sink.write_and_readback(evidence)
return token
Production implementation
The VerifiedRouteProvider must verify a centrally signed, unexpired route profile before returning it; the flow store must provide atomic single-use consume with encryption and TTL; and the evidence sink must durably acknowledge a non-secret receipt. Keep these as adapters to the organization's policy, session, and audit systems. Apply the RFC 9207 issuer rule exactly: require iss when the authorization server advertises it, compare it whenever present, and accept absence only when support was not advertised. The protected resource must still validate token audience/resource server-side under AID-H-034.002.
Independent verification
Replay the gate with a caller-selected metadata URL, substituted or multiple issuers, mismatched resource, changed policy revision, PKCE downgrade, missing or altered iss, expired and replayed state, redirect mutation, token-endpoint redirect, duplicate JSON keys, oversized responses, and evidence-sink failure. Every such case must deny before a token is accepted; the positive case must bind the stored receipt to the exact route, resource, issuer, and policy revision without storing the access token.
Open-source tools
- PyJWT
- jwcrypto
- oauthlib
- requests-oauthlib
- OpenTelemetry
Commercial / hosted tools
- Auth0
- Okta
- Datadog
Threat mappings
MITRE ATLAS
- AML.T0012 Valid Accounts (session hijacking or replay enables unauthorized access)
- AML.T0073 Impersonation (cross-session contamination or token replay enables impersonation)
MAESTRO
- Agent Identity Attack (L7) (session hijacking or replay)
- Integration Risks (L7) (cross-session state contamination)
OWASP LLM Top 10 2026
- LLM02:2026 Sensitive Information Disclosure
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI03:2026 Identity and Privilege Abuse (session hijacking enables privilege abuse)
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- AISubtech-14.1.2 Insufficient Access Controls
- AISubtech-4.2.2 Session Boundary Violation
- AISubtech-4.3.4 Replay Exploitation
- AISubtech-7.4.1 Token Theft
- AITech-4.2 Context Boundary Attacks
- AITech-4.3 Protocol Manipulation
- 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 (cross-session state contamination leaks data across trust boundaries)
- EDH: Excessive Data Handling (stale sessions retain data beyond intended lifetime)
Databricks AI Security Framework 3.0
- Agents - Tools MCP Client 13.34: Session and State Management Failures
- Agents - Core 13.9: Identity Spoofing & Impersonation (session hijacking enables impersonation)
- Platform 12.4: Unauthorized privileged access