AID-I-004.009: Peer-Agent Message Context Quarantine & Promotion
Keep every authenticated peer-agent message outside active and persistent model context until the exact received bytes pass transport-admission and content-validation controls and a trusted promotion service emits a single-use, work-scope-bound receipt for a bounded typed projection. Promoted content remains explicitly peer_untrusted data with no instruction, policy, credential, tool, or delegated-authority semantics. Missing, stale, unsupported, partially covered, or unverifiable evidence remains quarantined and is INSUFFICIENT_DATA or ERROR, never PASS.
Pillars: app, infra
Lifecycle: building, operation
Scope boundary
Owns transient quarantine and active-context promotion for messages received from peer agents over A2A, custom agent protocols, gateways, relays, or sub-agent runtimes. It applies when a peer-message path exists and a trusted context assembler can enforce typed trust metadata. Promoted content remains peer-untrusted data with no instruction, policy, credential, tool, delegated-authority, or durable-persistence semantics.
AID-H-004.003: AID-I-004.009 quarantines and promotes peer-message content into active context; AID-H-004.003 authenticates the peer, protects the channel and message, prevents replay, and authorizes the A2A task. A transport-admission PASS is required input but does not authorize model-context admission or change peer content into trusted instructions.
AID-H-002.002: AID-I-004.009 owns quarantine state and context admission; AID-H-002.002 validates the received bytes for schema, encoding, size, and prompt-injection risk. Content validation is a required signed input, while this control decides whether a bounded projection can enter active context and with which trust semantics.
AID-I-004.004: AID-I-004.009 promotes a bounded peer-message projection only into transient active context; AID-I-004.004 promotes an approved write into durable trusted memory. A context-promotion receipt never authorizes persistence, trusted-memory status, or reuse beyond its bounded work scope.
AID-H-017.007: AID-I-004.009 owns the peer-message quarantine and context-promotion state machine; AID-H-017.007 is an optional dual-LLM analysis mechanism for untrusted content. The analyzer can supply evidence but does not own quarantine storage, promotion atomicity, trust metadata, or context-assembler enforcement.
Implementation guidance
- AID-I-004.009-G001: Write the exact authenticated peer-message bytes to a deny-by-default quarantine and issue a signed, read-back-verified receipt bound to peer identity, task, policy, content population, and expiry.
- AID-I-004.009-G002: Promote only a signed, validated, bounded projection through a compare-and-swap state transition, and force the context assembler to preserve peer_untrusted data semantics with zero delegated authority.
AID-I-004.009-G001
Write the exact authenticated peer-message bytes to a deny-by-default quarantine and issue a signed, read-back-verified receipt bound to peer identity, task, policy, content population, and expiry.
Before you begin
Apply this path only to an actual peer-agent ingress. The authenticated transport must already have produced an AID-H-004.003 admission receipt for the exact message bytes. Use PostgreSQL 15+, psycopg[binary]>=3.2,<4, and Cosign 2.4+. The context assembler and agent runtime must have no read privilege on raw_message.
1. Create a durable quarantine with separated roles
-- File: db/004_peer_context_quarantine.sql
CREATE TYPE peer_context_state AS ENUM ('QUARANTINED', 'PROMOTED', 'REJECTED');
CREATE TABLE a2a_context_quarantine (
quarantine_id uuid PRIMARY KEY,
tenant_id text NOT NULL,
source_agent_id text NOT NULL,
target_agent_id text NOT NULL,
message_id text NOT NULL,
work_scope_id text NOT NULL,
task_id text,
context_id text,
raw_sha256 text NOT NULL CHECK (raw_sha256 ~ '^[0-9a-f]{64}$'),
raw_message bytea NOT NULL,
h004_receipt_sha256 text NOT NULL CHECK (h004_receipt_sha256 ~ '^[0-9a-f]{64}$'),
ingress_policy_sha256 text NOT NULL CHECK (ingress_policy_sha256 ~ '^[0-9a-f]{64}$'),
state peer_context_state NOT NULL DEFAULT 'QUARANTINED',
row_version bigint NOT NULL DEFAULT 0,
created_at timestamptz NOT NULL,
expires_at timestamptz NOT NULL CHECK (expires_at > created_at),
quarantine_receipt jsonb NOT NULL,
quarantine_bundle bytea NOT NULL,
context_fragment jsonb,
validation_receipt jsonb,
promotion_receipt jsonb,
promotion_bundle bytea,
promoted_at timestamptz,
UNIQUE (tenant_id, source_agent_id, message_id),
CHECK (
(state = 'QUARANTINED' AND context_fragment IS NULL AND promotion_receipt IS NULL)
OR
(state = 'PROMOTED' AND context_fragment IS NOT NULL
AND validation_receipt IS NOT NULL AND promotion_receipt IS NOT NULL
AND promotion_bundle IS NOT NULL AND promoted_at IS NOT NULL)
OR state = 'REJECTED'
)
);
CREATE FUNCTION enforce_peer_context_transition()
RETURNS trigger
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = pg_catalog, public
AS $$
BEGIN
IF OLD.state <> 'QUARANTINED'
OR NEW.state NOT IN ('PROMOTED', 'REJECTED')
OR NEW.row_version <> OLD.row_version + 1 THEN
RAISE EXCEPTION 'invalid peer-context state transition' USING ERRCODE = '42501';
END IF;
IF ROW(
NEW.quarantine_id, NEW.tenant_id, NEW.source_agent_id, NEW.target_agent_id,
NEW.message_id, NEW.work_scope_id, NEW.task_id, NEW.context_id,
NEW.raw_sha256, NEW.raw_message, NEW.h004_receipt_sha256,
NEW.ingress_policy_sha256, NEW.created_at, NEW.expires_at,
NEW.quarantine_receipt, NEW.quarantine_bundle
) IS DISTINCT FROM ROW(
OLD.quarantine_id, OLD.tenant_id, OLD.source_agent_id, OLD.target_agent_id,
OLD.message_id, OLD.work_scope_id, OLD.task_id, OLD.context_id,
OLD.raw_sha256, OLD.raw_message, OLD.h004_receipt_sha256,
OLD.ingress_policy_sha256, OLD.created_at, OLD.expires_at,
OLD.quarantine_receipt, OLD.quarantine_bundle
) THEN
RAISE EXCEPTION 'immutable peer-context binding changed' USING ERRCODE = '42501';
END IF;
RETURN NEW;
END;
$$;
CREATE TRIGGER a2a_context_quarantine_transition
BEFORE UPDATE ON a2a_context_quarantine
FOR EACH ROW EXECUTE FUNCTION enforce_peer_context_transition();
REVOKE ALL ON a2a_context_quarantine FROM PUBLIC;
GRANT INSERT (
quarantine_id, tenant_id, source_agent_id, target_agent_id, message_id,
work_scope_id, task_id, context_id, raw_sha256, raw_message,
h004_receipt_sha256, ingress_policy_sha256, created_at, expires_at,
quarantine_receipt, quarantine_bundle
) ON a2a_context_quarantine TO a2a_quarantine_writer;
GRANT SELECT (
quarantine_id, tenant_id, source_agent_id, target_agent_id, message_id,
work_scope_id, task_id, context_id, raw_sha256, h004_receipt_sha256,
ingress_policy_sha256, state, row_version, created_at, expires_at,
quarantine_receipt, quarantine_bundle
) ON a2a_context_quarantine TO a2a_quarantine_writer;
GRANT SELECT (
quarantine_id, tenant_id, source_agent_id, target_agent_id, message_id,
work_scope_id, task_id, context_id, raw_sha256, state, row_version,
expires_at, quarantine_receipt, quarantine_bundle
) ON a2a_context_quarantine TO a2a_promotion_service;
GRANT UPDATE (
state, row_version, context_fragment, validation_receipt,
promotion_receipt, promotion_bundle, promoted_at
) ON a2a_context_quarantine TO a2a_promotion_service;
GRANT SELECT (
quarantine_id, tenant_id, source_agent_id, target_agent_id, message_id,
work_scope_id, task_id, context_id, raw_sha256, state, context_fragment,
promotion_receipt, promotion_bundle, promoted_at
) ON a2a_context_quarantine TO a2a_context_assembler;2. Verify immutable signed inputs and quarantine the exact bytes
# File: a2a_context/quarantine_ingress.py
# requirements.txt: psycopg[binary]>=3.2,<4
from __future__ import annotations
import hashlib
import json
import math
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import uuid4
import psycopg
from psycopg.types.json import Jsonb
def reject_duplicate_keys(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 reject_nonfinite(value: str) -> None:
raise ValueError(f"non-finite JSON value: {value}")
def strict_json(raw: bytes) -> dict[str, Any]:
value = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
if not isinstance(value, dict):
raise ValueError("signed document must be a JSON object")
return value
def canonical(value: dict[str, Any]) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
def sha256(raw: bytes) -> str:
return hashlib.sha256(raw).hexdigest()
def regular_bytes(path: Path, maximum: int) -> bytes:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
try:
info = os.fstat(descriptor)
if not stat.S_ISREG(info.st_mode) or info.st_size < 1 or info.st_size > maximum:
raise ValueError(f"unsafe signed input: {path}")
raw = os.read(descriptor, maximum + 1)
if len(raw) != info.st_size:
raise ValueError(f"signed input changed while reading: {path}")
return raw
finally:
os.close(descriptor)
def cosign_verify_json(payload: bytes, bundle: bytes, public_key: str) -> dict[str, Any]:
timeout = float(os.environ["A2A_COSIGN_TIMEOUT_SECONDS"])
if not math.isfinite(timeout) or timeout <= 0:
raise ValueError("invalid Cosign timeout")
with tempfile.TemporaryDirectory(prefix="a2a-verify-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_path = root / "payload.json"
bundle_path = root / "payload.bundle"
payload_path.write_bytes(payload)
bundle_path.write_bytes(bundle)
os.chmod(payload_path, 0o400)
os.chmod(bundle_path, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", public_key,
"--bundle", str(bundle_path), str(payload_path)],
check=True, capture_output=True, text=True, timeout=timeout,
)
if payload_path.read_bytes() != payload or bundle_path.read_bytes() != bundle:
raise RuntimeError("verified snapshot changed")
return strict_json(payload)
def cosign_sign_json(document: dict[str, Any], private_key: str) -> tuple[bytes, bytes]:
payload = canonical(document)
timeout = float(os.environ["A2A_COSIGN_TIMEOUT_SECONDS"])
if not math.isfinite(timeout) or timeout <= 0:
raise ValueError("invalid Cosign timeout")
with tempfile.TemporaryDirectory(prefix="a2a-sign-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_path = root / "receipt.json"
bundle_path = root / "receipt.bundle"
payload_path.write_bytes(payload)
os.chmod(payload_path, 0o400)
subprocess.run(
["cosign", "sign-blob", "--yes", "--key", private_key,
"--bundle", str(bundle_path), str(payload_path)],
check=True, capture_output=True, text=True, timeout=timeout,
)
bundle = bundle_path.read_bytes()
if not bundle:
raise RuntimeError("Cosign emitted an empty bundle")
return payload, bundle
def parse_time(value: Any) -> datetime:
if not isinstance(value, str):
raise ValueError("timestamp is missing")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("timestamp must be timezone-aware")
return parsed.astimezone(timezone.utc)
def require_text(document: dict[str, Any], key: str) -> str:
value = document.get(key)
if not isinstance(value, str) or not value or len(value) > 512:
raise ValueError(f"invalid {key}")
return value
def quarantine_peer_message(
*, raw_message: bytes, source_agent_id: str, target_agent_id: str,
h004_receipt_path: Path, h004_bundle_path: Path,
) -> dict[str, Any]:
policy_raw = regular_bytes(Path(os.environ["A2A_INGRESS_POLICY_PATH"]), 1_048_576)
policy_bundle = regular_bytes(
Path(os.environ["A2A_INGRESS_POLICY_BUNDLE"]), 1_048_576
)
policy = cosign_verify_json(
policy_raw, policy_bundle, os.environ["A2A_POLICY_VERIFY_KEY"]
)
required_policy_fields = {
"schema_version", "policy_version", "maximum_message_bytes",
"quarantine_ttl_seconds", "maximum_h004_receipt_lifetime_seconds",
"required_h004_policy_sha256", "allowed_part_media_types",
"allowed_peer_pairs",
}
if set(policy) != required_policy_fields:
raise ValueError("ingress policy fields do not exactly match schema")
if policy["schema_version"] != "aidefend.peer-context-ingress-policy.v1":
raise ValueError("unsupported ingress policy schema")
policy_sha = sha256(policy_raw)
for key in (
"maximum_message_bytes", "quarantine_ttl_seconds",
"maximum_h004_receipt_lifetime_seconds",
):
if isinstance(policy[key], bool) or not isinstance(policy[key], int) or policy[key] < 1:
raise ValueError(f"invalid policy bound: {key}")
expected_h004_policy = policy["required_h004_policy_sha256"]
if (
not isinstance(expected_h004_policy, str)
or len(expected_h004_policy) != 64
or any(character not in "0123456789abcdef" for character in expected_h004_policy)
):
raise ValueError("invalid required H-004 policy digest")
if (
not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
or not isinstance(policy["allowed_part_media_types"], list)
or not policy["allowed_part_media_types"]
or any(
not isinstance(value, str) or not value
for value in policy["allowed_part_media_types"]
)
or not isinstance(policy["allowed_peer_pairs"], list)
or not policy["allowed_peer_pairs"]
or any(
not isinstance(route, dict)
or set(route) != {"source_agent_id", "target_agent_id"}
or any(not isinstance(value, str) or not value for value in route.values())
for route in policy["allowed_peer_pairs"]
)
):
raise ValueError("ingress policy version, media types, or peer routes are invalid")
if len(raw_message) < 2 or len(raw_message) > policy["maximum_message_bytes"]:
raise ValueError("message is empty or exceeds signed maximum")
message = strict_json(raw_message)
message_id = require_text(message, "messageId")
if message.get("role") not in {"ROLE_USER", "ROLE_AGENT"}:
raise ValueError("invalid A2A 1.0 role")
task_id = message.get("taskId")
if task_id is not None and (
not isinstance(task_id, str) or not task_id or len(task_id) > 512
):
raise ValueError("invalid taskId")
context_id = message.get("contextId")
if context_id is not None and (
not isinstance(context_id, str) or not context_id or len(context_id) > 512
):
raise ValueError("invalid contextId")
parts = message.get("parts")
if not isinstance(parts, list) or not parts:
raise ValueError("message part population is empty")
content_members = {"text", "raw", "url", "data"}
for part in parts:
if not isinstance(part, dict):
raise ValueError("message part is not an object")
selected = content_members.intersection(part)
if len(selected) != 1:
raise ValueError("A2A 1.0 Part must contain exactly one content member")
selected_member = next(iter(selected))
media_type = part.get("mediaType")
if media_type is None and selected_member == "text":
media_type = "text/plain"
elif media_type is None and selected_member == "data":
media_type = "application/json"
if (
not isinstance(media_type, str)
or media_type not in policy["allowed_part_media_types"]
):
raise ValueError("part media type is missing or not admitted")
peer = {"source_agent_id": source_agent_id, "target_agent_id": target_agent_id}
if peer not in policy["allowed_peer_pairs"]:
raise PermissionError("peer route is not admitted")
receipt_raw = regular_bytes(h004_receipt_path, 1_048_576)
receipt_bundle = regular_bytes(h004_bundle_path, 1_048_576)
transport = cosign_verify_json(
receipt_raw, receipt_bundle, os.environ["A2A_H004_RECEIPT_VERIFY_KEY"]
)
now = datetime.now(timezone.utc)
issued_at = parse_time(transport.get("issued_at"))
expires_at = parse_time(transport.get("expires_at"))
message_sha = sha256(raw_message)
if (
transport.get("schema_version") != "aidefend.a2a-transport-admission.v1"
or transport.get("outcome") != "PASS"
or transport.get("message_sha256") != message_sha
or transport.get("message_id") != message_id
or transport.get("task_id") != task_id
or transport.get("context_id") != context_id
or transport.get("source_agent_id") != source_agent_id
or transport.get("target_agent_id") != target_agent_id
or transport.get("policy_sha256") != expected_h004_policy
or issued_at > now
or expires_at <= now
or expires_at - issued_at > timedelta(
seconds=policy["maximum_h004_receipt_lifetime_seconds"]
)
):
raise PermissionError("H-004.003 receipt is stale, incomplete, or mis-bound")
work_scope_id = require_text(transport, "work_scope_id")
quarantine_id = uuid4()
quarantine_expires = now + timedelta(seconds=policy["quarantine_ttl_seconds"])
quarantine_receipt = {
"schema_version": "aidefend.peer-context-quarantine-receipt.v1",
"control_id": "AID-I-004.009",
"outcome": "QUARANTINED",
"quarantine_id": str(quarantine_id),
"tenant_id": require_text(transport, "tenant_id"),
"source_agent_id": source_agent_id,
"target_agent_id": target_agent_id,
"message_id": message_id,
"work_scope_id": work_scope_id,
"task_id": task_id,
"context_id": context_id,
"raw_sha256": message_sha,
"h004_receipt_sha256": sha256(receipt_raw),
"ingress_policy_sha256": policy_sha,
"created_at": now.isoformat(),
"expires_at": quarantine_expires.isoformat(),
"nonce": str(uuid4()),
}
signed_receipt, signed_bundle = cosign_sign_json(
quarantine_receipt, os.environ["A2A_QUARANTINE_SIGNING_KEY"]
)
with psycopg.connect(os.environ["A2A_QUARANTINE_DSN"]) as connection:
with connection.cursor() as cursor:
cursor.execute(
"""INSERT INTO a2a_context_quarantine (
quarantine_id, tenant_id, source_agent_id, target_agent_id,
message_id, work_scope_id, task_id, context_id,
raw_sha256, raw_message,
h004_receipt_sha256, ingress_policy_sha256, created_at, expires_at,
quarantine_receipt, quarantine_bundle
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
(
quarantine_id, quarantine_receipt["tenant_id"],
source_agent_id, target_agent_id, message_id, work_scope_id,
task_id, context_id, message_sha, raw_message, sha256(receipt_raw),
policy_sha, now, quarantine_expires,
Jsonb(quarantine_receipt), signed_bundle,
),
)
connection.commit()
with psycopg.connect(os.environ["A2A_QUARANTINE_DSN"]) as connection:
row = connection.execute(
"""SELECT raw_sha256, state, quarantine_receipt, quarantine_bundle
FROM a2a_context_quarantine WHERE quarantine_id = %s""",
(quarantine_id,),
).fetchone()
if (
row is None
or row[0] != message_sha
or row[1] != "QUARANTINED"
or canonical(row[2]) != signed_receipt
or bytes(row[3]) != signed_bundle
):
raise RuntimeError("quarantine read-back verification failed")
return quarantine_receipt
3. Verify failure behavior and complete coverage
Run tests for a modified message, wrong peer, reused messageId, expired receipt, oversized payload, duplicate JSON key, missing part, unsupported content type, bad signature, wrong task, and database read-back mismatch. Every case must leave the message outside model context. Reconcile transport ingress counts against quarantine rows per tenant and deployment; any missing or duplicate record is INSUFFICIENT_DATA or ERROR, not PASS.
Action: Put this writer immediately after authenticated peer transport admission, deny runtime access to raw quarantine bytes, and require a signed read-back-verified quarantine receipt before any downstream validation or promotion step.
AID-I-004.009-G002
Promote only a signed, validated, bounded projection through a compare-and-swap state transition, and force the context assembler to preserve peer_untrusted data semantics with zero delegated authority.
Concept
Validation does not make peer content trusted instructions. The promotion service consumes the quarantined record, a signed AID-H-002.002 validation receipt, and the signed output of an isolated deterministic projector. It constructs the context fragment itself, hard-codes all authority fields to false, signs the decision, and performs a one-way compare-and-swap transition. The model-facing assembler can read only promoted fragments and must independently verify the receipt.
1. Implement the promotion transaction
# File: a2a_context/promote.py
from __future__ import annotations
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID, uuid4
import psycopg
from psycopg.types.json import Jsonb
from a2a_context.quarantine_ingress import (
canonical, cosign_sign_json, cosign_verify_json, parse_time,
regular_bytes, sha256,
)
def read_signed(
path_env: str, bundle_env: str, key_env: str
) -> tuple[dict[str, Any], bytes]:
payload = regular_bytes(Path(os.environ[path_env]), 1_048_576)
bundle = regular_bytes(Path(os.environ[bundle_env]), 1_048_576)
return cosign_verify_json(payload, bundle, os.environ[key_env]), payload
def validate_projection(
projection: Any, maximum_items: int, maximum_item_bytes: int
) -> dict[str, list[str]]:
allowed = {"facts", "requests", "artifact_refs"}
if not isinstance(projection, dict) or set(projection) != allowed:
raise ValueError("projection fields do not exactly match schema")
total = 0
result: dict[str, list[str]] = {}
for field in sorted(allowed):
values = projection[field]
if not isinstance(values, list) or len(values) > maximum_items:
raise ValueError(f"{field} is invalid or exceeds item limit")
clean: list[str] = []
for value in values:
if not isinstance(value, str) or not value.strip():
raise ValueError(f"{field} contains an invalid value")
if len(value.encode("utf-8")) > maximum_item_bytes:
raise ValueError(f"{field} contains an oversized value")
clean.append(value)
total += 1
result[field] = clean
if total == 0:
raise ValueError("projection population is empty")
return result
def promote(
*, quarantine_id: UUID, validation_receipt_path: Path,
validation_bundle_path: Path, projection_path: Path,
projection_bundle_path: Path,
) -> dict[str, Any]:
policy, policy_raw = read_signed(
"A2A_PROMOTION_POLICY_PATH", "A2A_PROMOTION_POLICY_BUNDLE",
"A2A_POLICY_VERIFY_KEY",
)
required_fields = {
"schema_version", "policy_version", "maximum_projection_items_per_field",
"maximum_projection_item_bytes", "maximum_validation_lifetime_seconds",
"required_h002_policy_sha256", "required_projector_id",
}
if set(policy) != required_fields:
raise ValueError("promotion policy fields do not exactly match schema")
if policy["schema_version"] != "aidefend.peer-context-promotion-policy.v1":
raise ValueError("unsupported promotion policy schema")
for key in (
"maximum_projection_items_per_field", "maximum_projection_item_bytes",
"maximum_validation_lifetime_seconds",
):
if isinstance(policy[key], bool) or not isinstance(policy[key], int) or policy[key] < 1:
raise ValueError(f"invalid promotion policy bound: {key}")
digest = policy["required_h002_policy_sha256"]
if (
not isinstance(digest, str)
or len(digest) != 64
or any(character not in "0123456789abcdef" for character in digest)
):
raise ValueError("invalid required H-002 policy digest")
if (
not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
or not isinstance(policy["required_projector_id"], str)
or not policy["required_projector_id"]
):
raise ValueError("promotion policy version or projector is invalid")
validation_raw = regular_bytes(validation_receipt_path, 1_048_576)
validation_bundle = regular_bytes(validation_bundle_path, 1_048_576)
validation = cosign_verify_json(
validation_raw, validation_bundle,
os.environ["A2A_H002_RECEIPT_VERIFY_KEY"],
)
projection_raw = regular_bytes(projection_path, 1_048_576)
projection_bundle = regular_bytes(projection_bundle_path, 1_048_576)
projector = cosign_verify_json(
projection_raw, projection_bundle,
os.environ["A2A_PROJECTOR_VERIFY_KEY"],
)
now = datetime.now(timezone.utc)
issued_at = parse_time(validation.get("issued_at"))
expires_at = parse_time(validation.get("expires_at"))
if (
validation.get("schema_version") != "aidefend.peer-message-validation.v1"
or validation.get("control_id") != "AID-H-002.002"
or validation.get("outcome") != "PASS"
or validation.get("complete_population") is not True
or validation.get("policy_sha256") != digest
or issued_at > now
or expires_at <= now
or expires_at - issued_at > timedelta(
seconds=policy["maximum_validation_lifetime_seconds"]
)
):
raise PermissionError("validation receipt is stale, partial, or not PASS")
if (
projector.get("schema_version") != "aidefend.peer-context-projection.v1"
or projector.get("projector_id") != policy["required_projector_id"]
or projector.get("quarantine_id") != str(quarantine_id)
):
raise PermissionError("projector output is unsupported or mis-bound")
projection_sha = sha256(projection_raw)
if validation.get("projection_sha256") != projection_sha:
raise PermissionError("validation does not bind the signed projection")
projection = validate_projection(
projector.get("projection"),
policy["maximum_projection_items_per_field"],
policy["maximum_projection_item_bytes"],
)
with psycopg.connect(os.environ["A2A_PROMOTION_DSN"]) as connection:
with connection.cursor() as cursor:
cursor.execute(
"""SELECT tenant_id, source_agent_id, target_agent_id, message_id,
task_id, context_id, raw_sha256, state, row_version,
expires_at, quarantine_receipt, quarantine_bundle,
work_scope_id
FROM a2a_context_quarantine
WHERE quarantine_id = %s FOR UPDATE""",
(quarantine_id,),
)
row = cursor.fetchone()
if row is None:
raise LookupError("INSUFFICIENT_DATA: quarantine row is missing")
if row[7] != "QUARANTINED" or row[9] <= now:
raise PermissionError("quarantine is not live and promotable")
verified_quarantine = cosign_verify_json(
canonical(row[10]), bytes(row[11]),
os.environ["A2A_QUARANTINE_VERIFY_KEY"],
)
bindings = {
"quarantine_id": str(quarantine_id),
"tenant_id": row[0],
"source_agent_id": row[1],
"target_agent_id": row[2],
"message_id": row[3],
"work_scope_id": row[12],
"task_id": row[4],
"context_id": row[5],
"raw_sha256": row[6],
}
if any(
verified_quarantine.get(key) != value
for key, value in bindings.items()
):
raise PermissionError("quarantine receipt does not bind the locked row")
if any(validation.get(key) != value for key, value in bindings.items()):
raise PermissionError("validation receipt does not bind the locked row")
if (
projector.get("raw_sha256") != row[6]
or projector.get("work_scope_id") != row[12]
or projector.get("task_id") != row[4]
or projector.get("source_agent_id") != row[1]
):
raise PermissionError("projection does not bind the quarantined message")
context_fragment = {
"schema_version": "aidefend.peer-context-fragment.v1",
"quarantine_id": str(quarantine_id),
"work_scope_id": row[12],
"task_id": row[4],
"source_agent_id": row[1],
"role": "peer_data",
"trust": "peer_untrusted",
"authority": "none",
"may_set_system_instructions": False,
"may_set_developer_instructions": False,
"may_change_policy": False,
"may_delegate_tools": False,
"may_supply_credentials": False,
"projection": projection,
}
fragment_sha = sha256(canonical(context_fragment))
promotion_receipt = {
"schema_version": "aidefend.peer-context-promotion-receipt.v1",
"control_id": "AID-I-004.009",
"outcome": "PASS",
**bindings,
"context_fragment_sha256": fragment_sha,
"validation_receipt_sha256": sha256(validation_raw),
"validation_bundle_sha256": sha256(validation_bundle),
"projection_sha256": projection_sha,
"projection_bundle_sha256": sha256(projection_bundle),
"promotion_policy_sha256": sha256(policy_raw),
"issued_at": now.isoformat(),
"nonce": str(uuid4()),
"previous_row_version": row[8],
}
signed_receipt, signed_bundle = cosign_sign_json(
promotion_receipt, os.environ["A2A_PROMOTION_SIGNING_KEY"]
)
cursor.execute(
"""UPDATE a2a_context_quarantine
SET state = 'PROMOTED', row_version = row_version + 1,
context_fragment = %s, validation_receipt = %s,
validation_receipt_raw = %s, validation_bundle = %s,
projection_raw = %s, projection_bundle = %s,
promotion_receipt = %s, promotion_bundle = %s,
promoted_at = %s
WHERE quarantine_id = %s
AND state = 'QUARANTINED' AND row_version = %s""",
(
Jsonb(context_fragment), Jsonb(validation),
validation_raw, validation_bundle,
projection_raw, projection_bundle,
Jsonb(promotion_receipt), signed_bundle, now,
quarantine_id, row[8],
),
)
if cursor.rowcount != 1:
raise RuntimeError("promotion compare-and-swap failed")
connection.commit()
with psycopg.connect(os.environ["A2A_PROMOTION_DSN"]) as connection:
readback = connection.execute(
"""SELECT state, context_fragment, promotion_receipt, promotion_bundle,
validation_receipt_raw, validation_bundle,
projection_raw, projection_bundle
FROM a2a_context_quarantine WHERE quarantine_id = %s""",
(quarantine_id,),
).fetchone()
if (
readback is None
or readback[0] != "PROMOTED"
or sha256(canonical(readback[1])) != fragment_sha
or canonical(readback[2]) != signed_receipt
or bytes(readback[3]) != signed_bundle
or bytes(readback[4]) != validation_raw
or bytes(readback[5]) != validation_bundle
or bytes(readback[6]) != projection_raw
or bytes(readback[7]) != projection_bundle
):
raise RuntimeError("promotion read-back verification failed")
return context_fragment
def load_for_context(
quarantine_id: UUID, expected_work_scope_id: str
) -> dict[str, Any]:
with psycopg.connect(os.environ["A2A_ASSEMBLER_DSN"]) as connection:
row = connection.execute(
"""SELECT work_scope_id, state, context_fragment, promotion_receipt,
promotion_bundle
FROM a2a_context_quarantine WHERE quarantine_id = %s""",
(quarantine_id,),
).fetchone()
if row is None:
raise LookupError("INSUFFICIENT_DATA: promoted fragment is missing")
if row[0] != expected_work_scope_id or row[1] != "PROMOTED":
raise PermissionError("fragment is not promoted for this work scope")
fragment = row[2]
receipt = cosign_verify_json(
canonical(row[3]), bytes(row[4]),
os.environ["A2A_PROMOTION_VERIFY_KEY"],
)
if (
receipt.get("outcome") != "PASS"
or receipt.get("quarantine_id") != str(quarantine_id)
or receipt.get("work_scope_id") != expected_work_scope_id
or receipt.get("context_fragment_sha256") != sha256(canonical(fragment))
or fragment.get("role") != "peer_data"
or fragment.get("trust") != "peer_untrusted"
or fragment.get("authority") != "none"
or any(
fragment.get(field) is not False
for field in (
"may_set_system_instructions",
"may_set_developer_instructions",
"may_change_policy",
"may_delegate_tools",
"may_supply_credentials",
)
)
):
raise PermissionError("context fragment or promotion receipt is invalid")
return fragment
2. Enforce the model-facing boundary
Serialize the return value of load_for_context only into the platform's untrusted-data channel. Never concatenate it into system/developer prompts, tool definitions, authorization input, routing policy, or credentials. Tool calls derived from the peer data still require their own authorization and high-risk-action gates. A context-promotion receipt is single-work-scope and does not authorize persistent memory; durable storage must independently pass AID-I-004.004.
Retain replay inputs
Before the compare-and-swap, place the exact validation bytes and signature bundle plus the exact projector bytes and signature bundle in the governed quarantine-evidence store, or persist immutable content-addressed references to those four objects. The concrete path above assumes four bounded PostgreSQL BYTEA NOT NULL columns named validation_receipt_raw, validation_bundle, projection_raw, and projection_bundle; add them in the same reviewed migration as this service. It stores all four in the promotion transaction, binds their digests into the signed receipt, and byte-compares them on readback. A deployment that substitutes immutable content-addressed refs must perform the same four-object readback before commit. Hashes without retrievable signed bytes cannot support later signature replay and do not close promotion evidence.
3. Verify outcomes and population
Test altered projection bytes, altered database fragments, wrong work scope, stale validation, partial population, wrong projector, replayed promotion, invalid signature, empty projection, oversized fields, and direct runtime access to raw_message. Reconcile all live quarantine rows into exactly one of promoted, rejected, or unexpired-quarantined states, with signed evidence for every transition. Missing rows or receipts are INSUFFICIENT_DATA; verification or read-back failures are ERROR; neither state may release context.
Action: Make this promotion service and verified loader the only path from peer quarantine into model context, and inject its output solely as typed peer_untrusted data with every authority flag fixed to false.
Open-source tools
- PostgreSQL
- psycopg
- Pydantic (typed projection validation before context promotion)
- Open Policy Agent (OPA) (quarantine and promotion policy decisions)
- Cosign
Threat mappings
MITRE ATLAS
- AML.T0051 LLM Prompt Injection
- AML.T0051.001 LLM Prompt Injection: Indirect
- AML.T0080 AI Agent Context Poisoning
- AML.T0080.001 AI Agent Context Poisoning: Thread
- AML.T0118 Autonomous AI Agent Communication
- AML.T0118.001 Autonomous AI Agent Communication: Direct Agent Communication (peer messages carry no instruction, credential, or delegated-authority semantics)
- AML.T0124 Autonomous Attack Orchestration (peer tasking cannot enter active context with instruction, tool, credential, or delegated-authority semantics)
MAESTRO
- Data Tampering (L2)
- Agent Goal Manipulation (L7)
- Integration Risks (L7)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
- LLM03:2026 Excessive Agency
- LLM10:2026 Improper Output Handling
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack
- ASI06:2026 Memory & Context Poisoning
- ASI07:2026 Insecure Inter-Agent Communication
- ASI08:2026 Cascading Failures
NIST Adversarial Machine Learning 2025
- NISTAML.015 Indirect Prompt Injection
- NISTAML.018 Prompt Injection
Cisco Integrated AI Security and Safety Framework
- AISubtech-1.1.3 Multi-Agent Propagation (peer messages enter only as validated, work-scoped untrusted data)
- AISubtech-14.1.2 Insufficient Access Controls (separated quarantine roles enforce peer-context admission privileges)
- AISubtech-4.3.1 Schema Inconsistencies (closed peer-message projections and trusted metadata resist schema confusion)
- AISubtech-5.1.1 Long-term / Short-term Memory Injection (quarantine and transient-only promotion restrict peer-sourced memory poison)
- AITech-1.1 Direct Prompt Injection and Jailbreak (Prompt-based Manipulation) (peer messages enter only as validated, work-scoped untrusted data)
- AITech-1.2 Indirect Prompt Injection (peer messages enter only as validated, work-scoped untrusted data)
- AITech-4.2 Context Boundary Attacks (peer messages enter only as validated, work-scoped untrusted data)
- AITech-4.3 Protocol Manipulation (closed peer-message projections and trusted metadata resist schema confusion)
- AITech-5.1 Memory System Persistence (quarantine and transient-only promotion restrict peer-sourced memory poison)
- AITech-7.2 Memory System Corruption (quarantine and transient-only promotion restrict peer-sourced memory poison)
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection
- IIC: Insecure Integrated Component
Databricks AI Security Framework 3.0
- Agents - Core 13.1: Memory Poisoning
- Agents - Core 13.6: Intent Breaking & Goal Manipulation
- Agents - Core 13.12: Agent Communication Poisoning
- Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation