Scope boundary

Owns the conditional, protocol-independent end-to-end envelope profile for exact message-byte integrity, sender and audience binding, operation and target binding, policy-defined freshness, atomic single-use replay state, key rotation, and conditional request-response binding across AI communication paths whose transport is insufficient.

AID-H-004.002: AID-H-004.005 authenticates and freshness-binds a message across intermediaries; AID-H-004.002 authenticates the workload, service, API caller, and protected channel. Transport or caller authentication is necessary but may end at a proxy, queue, or session boundary before the message reaches its final consumer.

AID-H-004.003: AID-H-004.005 provides the generic end-to-end envelope integrity and replay primitive; AID-H-004.003 owns A2A peer, task/context ownership, state transition, stream order, callback, and per-request authorization semantics. A2A deployments retain their protocol-native checks and add this primitive only on applicable message paths.

AID-H-018.008: AID-H-004.005 atomically rejects an altered, stale, or reused message envelope; AID-H-018.008 prevents duplicate external side effects across retries, redelivery, and ambiguous outcomes. Single-use message acceptance does not guarantee exactly-once provider behavior, and an idempotency key does not prove message authenticity.

AID-H-012.004: AID-H-004.005 provides the reusable end-to-end message-envelope primitive across untrusted intermediaries; AID-H-012.004 owns authenticity, freshness, producer identity, and consumption binding for externally computed reward payloads. When the message is a learning reward, the reward-channel result is scored only under AID-H-012.004 rather than again as a generic envelope PASS.

AID-H-034: AID-H-004.005 owns a generic envelope profile across untrusted or asynchronous intermediaries; AID-H-034 owns MCP-specific transport, OAuth, routing, handles, Tasks, subscriptions, operation budgets, idempotency boundaries, and disable-state enforcement. MCP-specific nonce, proof-of-possession, handle, task, or commit-fence mechanisms remain with their MCP controls and do not replace an applicable end-to-end envelope.

Warning

mTLS Alone Does Not Decide Applicability

mTLS can be sufficient when one authenticated channel reaches the final consumer without TLS termination, store-and-forward, asynchronous delivery, 0-RTT, cross-boundary relay, or security-critical end-to-end assurance requirements. Its presence is not itself a NOT_APPLICABLE reason. Preserve a complete message-path and topology inventory showing every termination, intermediary, retry, queue, callback, and trust boundary before declaring this control not applicable.

Use a deterministic signature profile suited to the chosen format: RFC 9421 constructs its own HTTP signature base, JWS signs exact payload bytes, and COSE should use deterministic CBOR. Use JCS only when a profile must reconstruct equivalent JSON across serializers; do not impose it on every signature format.

Implementation guidance

• AID-H-004.005-G001: Define one deterministic signed-message profile that binds exact content bytes to sender, audience, operation, target, freshness, unique message ID, and conditional request context.
• AID-H-004.005-G002: Consume each verified message ID atomically before protected processing and retain replay state for the policy-defined acceptance window.
• AID-H-004.005-G003: Inventory every AI message path and require the end-to-end profile only where topology, delivery, trust, or security-critical semantics make transport protection insufficient.

AID-H-004.005-G001

Select and pin one signature profile

Prefer a named standard supported by both endpoints. RFC 9421 signs its defined HTTP signature base; JWS signs exact payload bytes; COSE uses a deterministic CBOR profile. The example below uses JWS over one exact UTF-8 envelope. The receiver verifies the bytes before parsing and never reconstructs the signed payload, so JCS is neither needed nor claimed. Keep TLS and service authentication in place.

Sign and verify the exact JWS payload

# File: messaging/signed_envelope.py
from __future__ import annotations
import base64
import hashlib
import json
import time
import uuid
from jwcrypto import jwk, jws

REQUIRED = {
    "schema_version", "issuer", "audience", "method", "target", "content_type",
    "body_sha256", "body_b64", "issued_at", "expires_at", "message_id",
    "request_sha256"
}

def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")

def unb64url(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))

def sign_message(*, body: bytes, issuer: str, audience: str, method: str,
                 target: str, lifetime_seconds: int, key_id: str,
                 private_jwk: dict, request_bytes: bytes | None = None) -> str:
    if not body or lifetime_seconds <= 0 or not all(
        isinstance(value, str) and value for value in
        (issuer, audience, method, target, key_id)
    ):
        raise ValueError("signed-message inputs are incomplete")
    now = int(time.time())
    envelope = {
        "schema_version": "aidefend.signed-message.v1",
        "issuer": issuer,
        "audience": audience,
        "method": method,
        "target": target,
        "content_type": "application/json",
        "body_sha256": hashlib.sha256(body).hexdigest(),
        "body_b64": b64url(body),
        "issued_at": now,
        "expires_at": now + lifetime_seconds,
        "message_id": str(uuid.uuid4()),
        "request_sha256": hashlib.sha256(request_bytes).hexdigest()
            if request_bytes is not None else None,
    }
    payload = json.dumps(
        envelope, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")
    token = jws.JWS(payload)
    token.add_signature(
        jwk.JWK(**private_jwk),
        protected=json.dumps({"alg": "ES256", "kid": key_id,
                              "typ": "aidefend-message+jws"})
    )
    return token.serialize(compact=True)

def verify_message(*, compact: str, public_jwk: dict, expected_key_id: str,
                   expected_issuer: str, expected_audience: str,
                   expected_method: str, expected_target: str,
                   maximum_lifetime_seconds: int,
                   expected_request_bytes: bytes | None = None) -> tuple[dict, bytes]:
    token = jws.JWS()
    token.deserialize(compact)
    header = token.jose_header
    if header != {"alg": "ES256", "kid": expected_key_id,
                  "typ": "aidefend-message+jws"}:
        raise PermissionError("JWS protected header differs from signed policy")
    token.verify(jwk.JWK(**public_jwk))
    envelope = json.loads(token.payload.decode("utf-8", errors="strict"))
    if not isinstance(envelope, dict) or set(envelope) != REQUIRED:
        raise ValueError("signed envelope schema differs")
    if envelope["schema_version"] != "aidefend.signed-message.v1" or any(
        envelope[field] != expected for field, expected in {
            "issuer": expected_issuer, "audience": expected_audience,
            "method": expected_method, "target": expected_target
        }.items()
    ):
        raise PermissionError("signed envelope binding differs")
    now = int(time.time())
    issued, expires = envelope["issued_at"], envelope["expires_at"]
    if (isinstance(issued, bool) or isinstance(expires, bool) or
        not isinstance(issued, int) or not isinstance(expires, int) or
        issued > now or expires <= now or expires <= issued or
        expires - issued > maximum_lifetime_seconds):
        raise PermissionError("message is future-dated, stale, or overlong")
    body = unb64url(envelope["body_b64"])
    if hashlib.sha256(body).hexdigest() != envelope["body_sha256"]:
        raise PermissionError("message body digest differs")
    expected_request = hashlib.sha256(expected_request_bytes).hexdigest()         if expected_request_bytes is not None else None
    if envelope["request_sha256"] != expected_request:
        raise PermissionError("request-response binding differs")
    return envelope, body

Action: Resolve kid only through the authenticated, signed trust policy for the expected sender; do not trust issuer or key URLs from the message. Pin allowed algorithms, rotate keys with an overlap window, cap decoded bytes before parsing, and sign security-critical responses with the originating request digest when the profile requires it. Process only the verified returned body.

Evidence

Retain the exact compact JWS, verified payload digest, policy and key-set version, expected endpoint bindings, verifier build digest, outcome, and key-rotation readback. Negative tests alter each binding, body byte, header, key, lifetime, and request digest independently.

AID-H-004.005-G002

Make replay rejection one atomic decision

Signature verification and freshness checks run first. Before any protected handler executes, reserve the verified issuer and message ID in a highly available authoritative store. A read-then-write check is racy. The database clock and a unique key provide one decision under concurrency.

PostgreSQL replay ledger

CREATE TABLE signed_message_replay (
  issuer text NOT NULL,
  message_id uuid NOT NULL,
  message_sha256 char(64) NOT NULL CHECK (message_sha256 ~ '^[0-9a-f]{64}$'),
  request_sha256 char(64),
  first_seen_at timestamptz NOT NULL DEFAULT clock_timestamp(),
  expires_at timestamptz NOT NULL,
  PRIMARY KEY (issuer, message_id),
  CHECK (expires_at > first_seen_at)
);
REVOKE ALL ON signed_message_replay FROM PUBLIC;

# File: messaging/replay_gate.py
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from uuid import UUID

def consume_once(conn, *, compact_jws: str, envelope: dict) -> None:
    message_id = UUID(envelope["message_id"])
    expires_at = datetime.fromtimestamp(envelope["expires_at"], timezone.utc)
    request_sha256 = envelope["request_sha256"]
    if request_sha256 is not None and (
        not isinstance(request_sha256, str) or len(request_sha256) != 64
    ):
        raise ValueError("request digest is malformed")
    digest = hashlib.sha256(compact_jws.encode("ascii")).hexdigest()
    with conn.transaction():
        database_now = conn.execute(
            "SELECT clock_timestamp()"
        ).fetchone()[0]
        if expires_at <= database_now:
            raise PermissionError("message expired before replay reservation")
        inserted = conn.execute(
            """INSERT INTO signed_message_replay
               (issuer, message_id, message_sha256, request_sha256, expires_at)
               VALUES (%s, %s, %s, %s, %s)
               ON CONFLICT DO NOTHING
               RETURNING message_id""",
            (envelope["issuer"], message_id, digest, request_sha256, expires_at)
        ).fetchone()
        if inserted is None:
            raise PermissionError("message replay rejected")

Action: Call consume_once after signature and binding validation but before protected processing. Keep rows until the signed policy's maximum message lifetime, clock-skew allowance, retry horizon, and evidence retention permit deletion. Database timeout, replica inconsistency, failover ambiguity, or write error is ERROR and blocks the operation.

Boundary with effect idempotency

The ledger prevents a second acceptance of the same message. It does not make an external payment, email, tool call, or cloud operation exactly once. Use AID-H-018.008 for transactional outbox, provider idempotency key, commit/readback, and ambiguous-outcome recovery. Do not hold this replay transaction open around an uncontrolled remote side effect.

Verify

Run simultaneous duplicate deliveries, queue redelivery, process restart, store failover, same ID with changed bytes, expired messages, clock skew, and cleanup-boundary tests. Retain the unique-conflict result, store commit receipt, database time, message digest, policy version, and independent query proving one accepted row.

AID-H-004.005-G003

Evaluate applicability from actual topology

Build a signed inventory for agent, MCP, model-serving, webhook, event, queue, callback, administrative, and kill-switch paths. Record producer and final consumer, every TLS termination, gateway, proxy, broker, store-and-forward stage, retry source, 0-RTT setting, tenant or administrative boundary, and whether the message triggers a security-critical state change.

Fail-closed applicability evaluator

# File: messaging/evaluate_paths.py
from __future__ import annotations
import json
import sys
from pathlib import Path

FIELDS = {
    "path_id", "producer", "final_consumer", "tls_terminations",
    "store_and_forward", "asynchronous_delivery", "zero_rtt",
    "crosses_trust_boundary", "security_critical_command", "retry_sources",
    "topology_readback_sha256"
}

document = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
if (not isinstance(document, dict) or document.get("schema_version") !=
        "aidefend.message-path-inventory.v1" or
        not isinstance(document.get("paths"), list) or not document["paths"]):
    raise ValueError("complete message-path inventory is required")

results = []
seen = set()
for path in document["paths"]:
    if not isinstance(path, dict) or set(path) != FIELDS:
        raise ValueError("message-path schema differs")
    if path["path_id"] in seen or not path["topology_readback_sha256"]:
        raise ValueError("path identity or topology readback is invalid")
    seen.add(path["path_id"])
    requires = (
        path["tls_terminations"] > 1 or path["store_and_forward"] or
        path["asynchronous_delivery"] or path["zero_rtt"] or
        path["crosses_trust_boundary"] or path["security_critical_command"] or
        bool(path["retry_sources"])
    )
    results.append({"path_id": path["path_id"],
                    "outcome": "APPLICABLE" if requires else "NOT_APPLICABLE"})
print(json.dumps({"results": results}, sort_keys=True))

Action: Treat an incomplete inventory, stale topology export, unknown TLS terminator, undocumented queue, or unverified 0-RTT and retry configuration as INSUFFICIENT_DATA. A path may be NOT_APPLICABLE only when the final consumer receives one authenticated channel directly, no applicability condition exists, and independent configuration readback proves that claim. The mere presence of mTLS is not an exemption.

Release and runtime verification

For every applicable path, test a changed body, wrong audience, wrong method or target, stale and future envelope, duplicate concurrent delivery, broker redelivery, restarted consumer, rotated key, unavailable replay store, and 0-RTT replay. Bind the inventory, topology export, signed profile, negative results, and complete path population to the release. Re-evaluate after any gateway, mesh, queue, callback, trust-boundary, or retry change.

For external email, webhook, and messaging triggers, consume this integrity/replay primitive in AID-H-004.007. A provider-authenticated delivery still requires initiator/resource authorization and binding to a permitted workflow before it can create an agent run.

Open-source tools

• jwcrypto
• pyca/cryptography
• PostgreSQL
• Valkey

Threat mappings

MITRE ATLAS

• AML.T0073 Impersonation

MAESTRO

• Integration Risks (L7)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI07:2026 Insecure Inter-Agent Communication

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-3.1.2 Trusted Agent Spoofing
• AISubtech-4.3.4 Replay Exploitation
• AITech-3.1 Masquerading / Obfuscation / Impersonation
• AITech-4.3 Protocol Manipulation

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component

Databricks AI Security Framework 3.0

• Agents - Core 13.12: Agent Communication Poisoning
• Agents - Tools MCP Server 13.25: Insecure Communication