AID-H-020.001: Chunk-Level Integrity Signing

Version 1.20261015 · Actionable sub-technique

Provide fine-grained RAG integrity by binding each canonical chunk, source version, embedding bytes, and embedding-model identity to a signed manifest stored in an external retention-locked trust domain, then verifying that manifest before context assembly. A hash stored beside content in the same vector database is not accepted as evidence because a vector-store writer could alter both.

Pillars: data

Lifecycle: building, operation

Scope boundary

Owns signed integrity of each canonical RAG chunk, source version, embedding bytes, and embedding-model identity using trust evidence outside the writable vector store, verified before context assembly. It proves integrity, not entitlement or truth.

AID-M-002.002: AID-H-020.001 verifies fine-grained retrieval units at runtime; AID-M-002.002 verifies whole datasets, models, images, manifests, and other macro artifacts. A valid source artifact does not prove each derived chunk and embedding remained correctly bound.

AID-H-020.003: AID-H-020.001 proves the retrieved chunk and embedding are authentic and unchanged; AID-H-020.003 proves the requesting principal is entitled to read them. Authentic content can still be unauthorized.

AID-H-002.007: AID-H-020.001 protects chunks after ingestion and transformation; AID-H-002.007 verifies external source-media provenance before OCR and chunking. Source provenance establishes the initial verdict, while chunk signing preserves integrity from that point forward.

Implementation guidance

AID-H-020.001-G001: Bind every chunk and embedding to an exact-schema, version-pinned, future-retained signed manifest before context assembly.
AID-H-020.001-G002: Resolve KMS identities, trusted signer set, object-lock retention, and retrieval freshness from one pinned-key signed RAG-integrity policy and preflight the exact object version and record schema.

AID-H-020.001-G001

Bind every chunk and embedding to an exact-schema, version-pinned, future-retained signed manifest before context assembly.

Before you begin

Retrieval pins that version, verifies a trusted signing-key ID, requires future retention, and recomputes chunk and embedding digests before context assembly.

External manifest writer

# File: rag_integrity/manifest_store.py
from __future__ import annotations

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

import boto3
import numpy as np


BUCKET = os.environ["RAG_MANIFEST_BUCKET"]
SIGNING_KEY = os.environ["RAG_MANIFEST_SIGNING_KMS_KEY_ID"]
STORAGE_KEY = os.environ["RAG_MANIFEST_STORAGE_KMS_KEY_ID"]
RETENTION_SECONDS = int(os.environ["RAG_MANIFEST_RETENTION_SECONDS"])
if RETENTION_SECONDS <= 0:
    raise RuntimeError("retention must be a positive signed-policy value")
kms = boto3.client("kms")
s3 = boto3.client("s3")


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


def embedding_sha256(values: list[float]) -> str:
    vector = np.asarray(values, dtype="<f4")
    if vector.ndim != 1 or vector.size == 0 or not np.isfinite(vector).all():
        raise ValueError("embedding must be one finite non-empty vector")
    return hashlib.sha256(vector.tobytes(order="C")).hexdigest()


def write_manifest(
    chunk_id: str,
    source_uri: str,
    source_version: str,
    chunk_text: str,
    embedding: list[float],
    embedding_model_sha256: str,
) -> dict:
    if not all([chunk_id, source_uri, source_version, embedding_model_sha256]):
        raise ValueError("manifest identity fields are required")
    now = datetime.now(timezone.utc)
    record = {
        "schema_version": "aidefend.rag-chunk-record.v1",
        "chunk_id": chunk_id,
        "source_uri": source_uri,
        "source_version": source_version,
        "chunk_sha256": hashlib.sha256(chunk_text.encode()).hexdigest(),
        "embedding_sha256": embedding_sha256(embedding),
        "embedding_dtype": "float32-little-endian",
        "embedding_model_sha256": embedding_model_sha256,
        "created_at": now.isoformat(),
    }
    payload = canonical(record)
    digest = hashlib.sha256(payload).digest()
    signed = kms.sign(
        KeyId=SIGNING_KEY,
        Message=digest,
        MessageType="DIGEST",
        SigningAlgorithm="ECDSA_SHA_256",
    )
    envelope = {
        "schema_version": "aidefend.rag-chunk-envelope.v1",
        "record": record,
        "record_sha256": digest.hex(),
        "signing_key_id": signed["KeyId"],
        "signing_algorithm": "ECDSA_SHA_256",
        "signature": base64.b64encode(signed["Signature"]).decode(),
    }
    object_key = f"chunks/{chunk_id}/{source_version}/{digest.hex()}.json"
    response = s3.put_object(
        Bucket=BUCKET,
        Key=object_key,
        Body=canonical(envelope),
        ContentType="application/json",
        IfNoneMatch="*",
        ServerSideEncryption="aws:kms",
        SSEKMSKeyId=STORAGE_KEY,
        ObjectLockMode="COMPLIANCE",
        ObjectLockRetainUntilDate=now + timedelta(seconds=RETENTION_SECONDS),
    )
    version_id = response.get("VersionId")
    if not version_id:
        raise RuntimeError("versioned Object-Lock store did not return VersionId")
    return {
        "manifest_key": object_key,
        "manifest_version_id": version_id,
        "record_sha256": digest.hex(),
    }

Retrieval-time verifier

# File: rag_integrity/retrieval_gate.py
from __future__ import annotations

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

import boto3

from rag_integrity.manifest_store import BUCKET, canonical, embedding_sha256


TRUSTED_KEYS = set(json.loads(os.environ["RAG_TRUSTED_SIGNING_KEY_IDS_JSON"]))
MIN_REMAINING_RETENTION_SECONDS = int(os.environ["RAG_MIN_REMAINING_RETENTION_SECONDS"])
kms = boto3.client("kms")
s3 = boto3.client("s3")


def verify_retrieved_chunk(doc) -> dict:
    key = doc.metadata.get("manifest_key")
    version_id = doc.metadata.get("manifest_version_id")
    if not isinstance(key, str) or not key.startswith("chunks/") or not version_id:
        raise PermissionError("version-pinned chunk manifest is missing")

    response = s3.get_object(Bucket=BUCKET, Key=key, VersionId=str(version_id))
    if response.get("VersionId") != version_id:
        raise PermissionError("manifest VersionId differs")
    if response.get("ObjectLockMode") != "COMPLIANCE":
        raise PermissionError("manifest is not in compliance retention")
    retain_until = response.get("ObjectLockRetainUntilDate")
    if not isinstance(retain_until, datetime):
        raise PermissionError("manifest retention evidence is unavailable")
    minimum = datetime.now(timezone.utc) + timedelta(seconds=MIN_REMAINING_RETENTION_SECONDS)
    if retain_until.astimezone(timezone.utc) <= minimum:
        raise PermissionError("manifest retention has expired or is below policy")

    envelope = json.loads(response["Body"].read())
    required_envelope = {
        "schema_version", "record", "record_sha256", "signing_key_id",
        "signing_algorithm", "signature",
    }
    if set(envelope) != required_envelope:
        raise PermissionError("manifest envelope schema differs")
    if envelope["schema_version"] != "aidefend.rag-chunk-envelope.v1":
        raise PermissionError("manifest envelope version is unsupported")
    record = envelope["record"]
    if not isinstance(record, dict) or record.get("schema_version") != "aidefend.rag-chunk-record.v1":
        raise PermissionError("manifest record schema differs")

    digest = hashlib.sha256(canonical(record)).digest()
    if not hmac.compare_digest(digest.hex(), envelope["record_sha256"]):
        raise PermissionError("manifest record digest differs")
    if key.rsplit("/", 1)[-1] != digest.hex() + ".json":
        raise PermissionError("content-addressed manifest key differs")
    if envelope["signing_key_id"] not in TRUSTED_KEYS:
        raise PermissionError("manifest signing key is not trusted")
    verified = kms.verify(
        KeyId=envelope["signing_key_id"],
        Message=digest,
        MessageType="DIGEST",
        Signature=base64.b64decode(envelope["signature"], validate=True),
        SigningAlgorithm="ECDSA_SHA_256",
    )
    if not verified.get("SignatureValid"):
        raise PermissionError("manifest signature is invalid")

    expected = {
        "chunk_id": doc.metadata.get("chunk_id"),
        "source_uri": doc.metadata.get("source_uri"),
        "source_version": doc.metadata.get("source_version"),
        "embedding_model_sha256": doc.metadata.get("embedding_model_sha256"),
    }
    if any(not hmac.compare_digest(str(record.get(k, "")), str(v or "")) for k, v in expected.items()):
        raise PermissionError("chunk identity binding differs")
    chunk_hash = hashlib.sha256(doc.page_content.encode()).hexdigest()
    if not hmac.compare_digest(chunk_hash, record["chunk_sha256"]):
        raise PermissionError("chunk content is tampered")
    if getattr(doc, "embedding", None) is None:
        raise PermissionError("embedding is unavailable: insufficient_data")
    if not hmac.compare_digest(embedding_sha256(doc.embedding), record["embedding_sha256"]):
        raise PermissionError("embedding is tampered")

    return {
        "schema_version": "aidefend.rag-chunk-verification.v1",
        "chunk_id": record["chunk_id"],
        "manifest_key": key,
        "manifest_version_id": version_id,
        "record_sha256": digest.hex(),
        "signing_key_id": envelope["signing_key_id"],
        "retain_until": retain_until.isoformat(),
    }

Evidence: Sign the retrieval decision outside the vector-store trust domain and bind it to the retrieval request, policy, manifest version, chunk/vector digests, and context-assembly result. Missing version, expired retention, unavailable vector, signature failure, or schema mismatch is insufficient_data and fails closed.

AID-H-020.001-G002

Resolve KMS identities, trusted signer set, object-lock retention, and retrieval freshness from one pinned-key signed RAG-integrity policy and preflight the exact object version and record schema.

Mandatory production policy envelope

The writer and verifier above are implementation components. Do not populate their environment from deployment defaults or request values. A bootstrap process first verifies one exact-schema policy under a key pinned in the workload image, injects only those verified values, and independently heads the exact S3 VersionId before allowing retrieval verification.

# File: rag_integrity/verified_policy.py
from __future__ import annotations

import json
import math
import os
import re
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

import boto3


POLICY_PATH = Path("/etc/aidefend/rag-integrity/policy.json")
POLICY_SIGNATURE = Path("/etc/aidefend/rag-integrity/policy.sig")
POLICY_TRUST_KEY = Path("/opt/aidefend/trust/rag-integrity-policy-authority.pub")
POLICY_FIELDS = {
    "schema_version", "policy_version", "manifest_bucket",
    "manifest_key_prefix", "signing_kms_key_id", "storage_kms_key_id",
    "trusted_signing_kms_key_ids", "retention_seconds",
    "minimum_remaining_retention_seconds", "retention_policy_sha256",
}
ENVELOPE_FIELDS = {
    "schema_version", "record", "record_sha256", "signing_key_id",
    "signing_algorithm", "signature",
}
RECORD_FIELDS = {
    "schema_version", "chunk_id", "source_uri", "source_version",
    "chunk_sha256", "embedding_sha256", "embedding_dtype",
    "embedding_model_sha256", "created_at",
}


def required_verify_timeout() -> float:
    value = float(os.environ["VERIFIED_H020_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(value) or value <= 0: raise RuntimeError("policy verification timeout must be finite and positive")
    return value

POLICY_VERIFY_TIMEOUT_SECONDS = required_verify_timeout()

def strict_json_object(raw: bytes) -> dict:
    def unique_pairs(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate signed-JSON key: {key}")
            value[key] = item
        return value

    def reject_constant(value):
        raise ValueError(f"non-finite signed-JSON number: {value}")

    document = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_pairs, parse_constant=reject_constant,
    )
    if not isinstance(document, dict):
        raise ValueError("signed JSON root must be an object")
    return document


def load_policy() -> dict:
    raw = POLICY_PATH.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h020-policy-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run([
            "cosign", "verify-blob", "--key", str(POLICY_TRUST_KEY),
            "--bundle", str(POLICY_SIGNATURE), str(snapshot),
        ], check=True, capture_output=True, text=True, timeout=POLICY_VERIFY_TIMEOUT_SECONDS)
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)
    policy = strict_json_object(raw)
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise RuntimeError("RAG-integrity policy schema mismatch")
    if policy["schema_version"] != "aidefend.rag-integrity-policy.v1":
        raise RuntimeError("RAG-integrity policy version unsupported")
    for field in (
        "policy_version", "manifest_bucket", "manifest_key_prefix",
        "signing_kms_key_id", "storage_kms_key_id", "retention_policy_sha256",
    ):
        if not isinstance(policy[field], str) or not policy[field]:
            raise RuntimeError(f"RAG-integrity policy field missing: {field}")
    trusted = policy["trusted_signing_kms_key_ids"]
    if (
        not isinstance(trusted, list) or not trusted
        or trusted != sorted(set(trusted))
        or policy["signing_kms_key_id"] not in trusted
    ):
        raise RuntimeError("trusted signing-key registry invalid")
    for field in ("retention_seconds", "minimum_remaining_retention_seconds"):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise RuntimeError(f"RAG-integrity policy value invalid: {field}")
    if policy["minimum_remaining_retention_seconds"] > policy["retention_seconds"]:
        raise RuntimeError("remaining-retention requirement exceeds retention policy")
    return policy


def configure_components(policy: dict) -> None:
    os.environ["RAG_MANIFEST_BUCKET"] = policy["manifest_bucket"]
    os.environ["RAG_MANIFEST_SIGNING_KMS_KEY_ID"] = policy["signing_kms_key_id"]
    os.environ["RAG_MANIFEST_STORAGE_KMS_KEY_ID"] = policy["storage_kms_key_id"]
    os.environ["RAG_MANIFEST_RETENTION_SECONDS"] = str(policy["retention_seconds"])
    os.environ["RAG_TRUSTED_SIGNING_KEY_IDS_JSON"] = json.dumps(
        policy["trusted_signing_kms_key_ids"], separators=(",", ":")
    )
    os.environ["RAG_MIN_REMAINING_RETENTION_SECONDS"] = str(
        policy["minimum_remaining_retention_seconds"]
    )


def preflight_exact_manifest(doc, policy: dict) -> dict:
    key = doc.metadata.get("manifest_key")
    version_id = doc.metadata.get("manifest_version_id")
    prefix = policy["manifest_key_prefix"].rstrip("/") + "/"
    if (
        not isinstance(key, str) or not key.startswith(prefix)
        or ".." in key.split("/") or not isinstance(version_id, str) or not version_id
    ):
        raise PermissionError("manifest locator violates signed policy")
    s3 = boto3.client("s3")
    head = s3.head_object(
        Bucket=policy["manifest_bucket"], Key=key, VersionId=version_id
    )
    if head.get("VersionId") != version_id:
        raise PermissionError("manifest head returned another VersionId")
    if head.get("ObjectLockMode") != "COMPLIANCE":
        raise PermissionError("manifest version lacks compliance retention")
    retain_until = head.get("ObjectLockRetainUntilDate")
    minimum = datetime.now(timezone.utc) + timedelta(
        seconds=policy["minimum_remaining_retention_seconds"]
    )
    if not isinstance(retain_until, datetime) or retain_until.astimezone(timezone.utc) <= minimum:
        raise PermissionError("manifest retention is below signed policy")
    response = s3.get_object(
        Bucket=policy["manifest_bucket"], Key=key, VersionId=version_id
    )
    if response.get("VersionId") != version_id:
        raise PermissionError("manifest fetch returned another VersionId")
    envelope = json.loads(response["Body"].read())
    if not isinstance(envelope, dict) or set(envelope) != ENVELOPE_FIELDS:
        raise PermissionError("manifest envelope schema mismatch")
    record = envelope.get("record")
    if not isinstance(record, dict) or set(record) != RECORD_FIELDS:
        raise PermissionError("chunk record schema mismatch")
    if envelope["signing_key_id"] not in set(policy["trusted_signing_kms_key_ids"]):
        raise PermissionError("chunk signer is not in signed key registry")
    return {
        "policy_version": policy["policy_version"],
        "retention_policy_sha256": policy["retention_policy_sha256"],
        "manifest_version_id": version_id,
        "retain_until": retain_until.isoformat(),
    }

Execution order: verify the policy, call configure_components before importing the writer/retrieval modules, then call preflight_exact_manifest and the retrieval verifier. Bind the resulting policy version, retention-policy digest, exact VersionId, KMS signer identity, chunk/vector/model digests, and context-assembly decision into an externally signed receipt. Policy/signature failure, missing head metadata, version drift, schema drift, or insufficient future retention is ERROR or INSUFFICIENT_DATA and fails closed.

Open-source tools

Cosign
pyca/cryptography
Sigstore Rekor v2 (signed chunk-manifest digest transparency anchoring; manifest storage and verification remain external)

Source-available / open-weight tools

immudb (Business Source License 1.1; source-available)

Commercial / hosted tools

AWS Key Management Service (AWS KMS)
Amazon S3 Object Lock
Azure Key Vault
Azure Blob Storage immutable storage
Google Cloud Key Management Service
Google Cloud Storage Bucket Lock

Threat mappings

MITRE ATLAS

AML.T0099 AI Agent Tool Data Poisoning
AML.T0070 RAG Poisoning
AML.T0059 Erode Dataset Integrity
AML.T0071 False RAG Entry Injection (integrity verification detects injected entries)

MAESTRO

Data Tampering (L2)
Compromised RAG Pipelines (L2)

OWASP LLM Top 10 2026

LLM05:2026 Data and Model Poisoning
LLM09:2026 Vector and Embedding Weaknesses

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

ASI06:2026 Memory & Context Poisoning (chunk-level signing detects tampered retrieval content)