AID-D-004.005: Runtime Turn-Chain Integrity Monitoring

Version 1.20261015 · Actionable sub-technique

Verify the structural integrity and provenance of each conversational turn and tool result by validating a canonical signed envelope, producer identity, content digest, conversation and turn binding, monotonic order, predecessor hash, freshness, and replay state. Emit a finding for tampered, reordered, replayed, missing, cross-conversation, or unverifiable turns before prompt assembly.

Pillars: app

Lifecycle: operation

Scope boundary

Owns findings for tampered, reordered, replayed, missing, cross-conversation, or unverifiable conversational turns and tool results by validating signed envelopes, producer identity, digests, ordering, predecessor hashes, freshness, and replay state.

AID-D-001.001: AID-D-004.005 verifies cryptographic provenance and sequence integrity of the turn; AID-D-001.001 analyzes whether authentic prompt content contains injection, harmful intent, or obfuscation. Authentic content can still be malicious, and semantically benign content can still be tampered or replayed.

AID-H-021.002: AID-D-004.005 emits a turn-chain integrity finding before prompt assembly; AID-H-021.002 owns fail-closed admission of signed runtime context and configuration. Detection supplies evidence, while Harden decides whether the turn may enter the runtime.

Implementation guidance

AID-D-004.005-G001

Verify every context turn against a canonical Ed25519 envelope, producer-key registry, content digest, expected population, and predecessor chain, and emit a structured finding without performing admission.
Concept

Integrity monitoring must prove who produced each turn, which bytes it covered, where the turn belongs, and whether the complete expected chain is present. Use asymmetric signatures so verification keys can be distributed without giving every producer the ability to forge another producer's evidence. Keep large text, image, audio, and document bytes out of the envelope and bind them by SHA-256.

Canonical envelope and verifier
from __future__ import annotations

# File: detect/turn_chain_integrity.py

import base64
import hashlib
import json
from dataclasses import asdict, dataclass
from typing import Callable, Literal, Sequence

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


Status = Literal["FINDING", "NO_FINDING", "INSUFFICIENT_DATA", "ERROR"]
ROOT = "0" * 64
PAYLOAD_FIELDS = {
    "schema_version", "conversation_id", "turn_id", "turn_index",
    "prev_envelope_sha256", "producer_id", "role", "content_type",
    "content_sha256", "request_id", "issued_at", "key_id",
}
ROLES = {"system", "developer", "user", "assistant", "tool"}


@dataclass(frozen=True)
class ChainFinding:
    schema_version: str
    control_id: str
    conversation_id: str
    status: Status
    reason: str
    expected_turn_count: int
    verified_turn_count: int
    failing_turn_id: str | None
    chain_head_sha256: str | None
    evidence_sha256s: list[str]


def canonical(value: dict) -> bytes:
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")


def envelope_hash(envelope: dict) -> str:
    return hashlib.sha256(canonical(envelope)).hexdigest()


def verify_turn_chain(
    entries: Sequence[dict],
    *,
    conversation_id: str,
    expected_turn_count: int,
    content_resolver: Callable[[str], bytes],
    trusted_keys: dict[tuple[str, str], Ed25519PublicKey],
) -> ChainFinding:
    evidence_hashes: list[str] = []
    previous = ROOT

    if expected_turn_count < 0 or len(entries) != expected_turn_count:
        return ChainFinding(
            "aidefend.turn_chain_finding.v1", "AID-D-004.005",
            conversation_id, "INSUFFICIENT_DATA", "population_incomplete",
            expected_turn_count, 0, None, None, [],
        )

    try:
        for expected_index, envelope in enumerate(entries):
            if not isinstance(envelope, dict) or set(envelope) != {"payload", "signature_b64"}:
                raise ValueError("envelope_schema_mismatch")
            payload = envelope["payload"]
            turn_id = payload.get("turn_id") if isinstance(payload, dict) else None
            if not isinstance(payload, dict) or set(payload) != PAYLOAD_FIELDS:
                raise ValueError("payload_schema_mismatch")
            if payload["schema_version"] != "aidefend.turn_envelope.v1":
                raise ValueError("unsupported_envelope_version")
            if payload["conversation_id"] != conversation_id:
                raise ValueError("cross_conversation_turn")
            if payload["turn_index"] != expected_index:
                raise ValueError("turn_reordered_or_missing")
            if payload["prev_envelope_sha256"] != previous:
                raise ValueError("predecessor_hash_mismatch")
            if payload["role"] not in ROLES:
                raise ValueError("unrecognized_role")

            key = trusted_keys.get((payload["producer_id"], payload["key_id"]))
            if key is None:
                return ChainFinding(
                    "aidefend.turn_chain_finding.v1", "AID-D-004.005",
                    conversation_id, "INSUFFICIENT_DATA", "producer_key_unavailable",
                    expected_turn_count, expected_index, str(turn_id), previous, evidence_hashes,
                )
            try:
                signature = base64.b64decode(envelope["signature_b64"], validate=True)
                key.verify(signature, canonical(payload))
            except (InvalidSignature, ValueError) as exc:
                raise ValueError("signature_invalid") from exc

            content = content_resolver(payload["turn_id"])
            observed_content_hash = hashlib.sha256(content).hexdigest()
            if observed_content_hash != payload["content_sha256"]:
                raise ValueError("content_digest_mismatch")

            previous = envelope_hash(envelope)
            evidence_hashes.append(previous)

        return ChainFinding(
            "aidefend.turn_chain_finding.v1", "AID-D-004.005",
            conversation_id, "NO_FINDING", "complete_chain_verified",
            expected_turn_count, len(entries), None, previous, evidence_hashes,
        )
    except (KeyError, TypeError, ValueError) as exc:
        return ChainFinding(
            "aidefend.turn_chain_finding.v1", "AID-D-004.005",
            conversation_id, "FINDING", str(exc),
            expected_turn_count, len(evidence_hashes),
            str(turn_id) if "turn_id" in locals() and turn_id else None,
            previous if previous != ROOT else None,
            evidence_hashes,
        )
    except Exception as exc:
        return ChainFinding(
            "aidefend.turn_chain_finding.v1", "AID-D-004.005",
            conversation_id, "ERROR", type(exc).__name__,
            expected_turn_count, len(evidence_hashes), None,
            previous if previous != ROOT else None,
            evidence_hashes,
        )


def publish_chain_finding(finding: ChainFinding, sink) -> None:
    sink(asdict(finding))
Verification

Replay the exact expected chain, then independently test a modified byte, wrong producer key, unknown key, duplicate turn, skipped index, reordered turn, cross-conversation turn, changed role, broken predecessor, missing content object, extra turn, and incomplete expected population. Verify that cryptographic or order violations produce FINDING, unavailable verification material produces INSUFFICIENT_DATA, infrastructure faults produce ERROR, and only complete chain coverage produces NO_FINDING.

Action: Resolve expected_turn_count from a trusted prompt-assembly manifest, not from the submitted array. Publish only hashes and identifiers. Send the finding to AID-H-021.002; do not raise an authorization exception or label authentic-but-malicious content safe.

AID-D-004.005-G002

Write turn-chain findings and expected-population manifests to append-only evidence storage controlled outside the conversation producer identities.
Concept

The same identity that can produce a turn must not be able to rewrite the evidence that later proves a gap or signature failure. Store the trusted expected-population manifest and the detector result as separate immutable objects, and bind both to the conversation ID and detector policy version.

from __future__ import annotations

# File: detect/publish_turn_chain_evidence.py

import base64
import hashlib
import json
from datetime import datetime, timedelta, timezone


def canonical(value: dict) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")


def publish_immutable(s3, *, bucket: str, key: str, evidence: dict, retention_days: int) -> dict:
    body = canonical(evidence)
    digest = hashlib.sha256(body).digest()
    retain_until = datetime.now(timezone.utc) + timedelta(days=retention_days)
    response = s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=body,
        ContentType="application/json",
        ChecksumSHA256=base64.b64encode(digest).decode("ascii"),
        ObjectLockMode="COMPLIANCE",
        ObjectLockRetainUntilDate=retain_until,
    )
    return {
        "uri": f"s3://{bucket}/{key}",
        "version_id": response["VersionId"],
        "sha256": digest.hex(),
        "retained_until": retain_until.isoformat(),
    }

Action: Enable bucket versioning and Object Lock before deployment, use a detector-specific write identity without delete permission, derive retention from versioned organizational policy, and read back the object version and checksum as verification evidence. A write failure is ERROR, not a completed detector run.

Verify complete coverage

The population is every turn-chain result, finding, manifest, write acknowledgement, readback, retention label, and expected correlation record emitted by the measured conversation population.

Evaluator and grader tamper cases

For evaluation workloads, keep the reference solution and grader internals unreadable to the evaluated identity, and make reference data, grader logic, raw observations, and prior results unwritable by that identity. Bind the evaluator image, reference digest, result-writer identity, and complete expected turn population to the chain. Exercise attempted output or log deletion, context compaction, evaluator-container code injection, adversarial grader output, missing or reordered turns, and rewritten observations. The grader writes a new result only through its dedicated evidence writer; it never edits the raw chain it evaluates.

AID-D-004.005-G003

Hash and verify out-of-band binary, multimodal, and referenced artifacts for every context turn, and report any missing, substituted, or mutable object without admitting or blocking the turn.
Resolve the complete referenced-object population

At the conversation observation point, canonicalize every attachment, image, audio segment, video segment, tool result, retrieval object, and external object reference into a manifest entry containing stable object ID, media type, byte length, SHA-256 digest, immutable storage locator, producer identity, and the context-turn envelope digest. Enumerate from both the signed turn manifest and the transport or object-store readback; a reference present on only one side is a finding, not an omitted sample.

Verify bytes outside the producer path

A separately credentialed reader fetches each immutable object, rejects redirects and mutable aliases, streams the bytes under policy limits, and recomputes length and digest. Test swapped attachments, same-name replacement, changed MIME type, partial upload, decompression expansion, missing object, late mutation, and cross-tenant locator reuse. Publish structured findings bound to the affected turn and object; preserve expected/observed population roots, per-object digests, readback identity, policy and verifier builds, and outcome counts. Admission or quarantine belongs to the consuming Harden or Isolate control.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-004.005: Runtime Turn-Chain Integrity Monitoring | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service