Scope boundary

Owns cryptographic integrity of persistent memory records after an authorized write by signing canonical content and metadata and verifying signatures and hashes before readmission to context. It does not decide who may request a protected write or whether the candidate content should be promoted.

AID-I-004.006: AID-I-004.003 proves cryptographic integrity after an authorized persistent-memory write; AID-I-004.006 enforces who may write protected file-backed identity and state resources. A valid signature does not establish write authorization, and OS-level write protection does not prove that an accepted record's content and metadata remained intact.

AID-I-004.004: AID-I-004.003 verifies signed memory integrity; AID-I-004.004 controls transactional promotion from quarantined to trusted memory. An intact record can still be untrusted or ineligible, while a promotion receipt must bind to integrity evidence rather than replace it.

Implementation guidance

• AID-I-004.003-G001: Controlled writer and integrity-first loader: sign canonical metadata at write time, then verify signature and content hash before any record is returned to the agent.

AID-I-004.003-G001

Security boundary

The controlled writer is the only identity allowed to create trusted memory records. It signs a JCS-canonical metadata envelope with an asymmetric signing service, verifies the returned signature locally, and stores content plus signature under a rollback-resistant per-record version floor. The loader joins against that floor and verifies key status, signature, tenant, namespace, policy, content digest, and stored-envelope digest before returning any content. An HMAC shared with readers is not sufficient because a compromised reader could forge writes.

PostgreSQL schema

CREATE TABLE trusted_memory_records (
  record_id text NOT NULL,
  version bigint NOT NULL CHECK (version > 0),
  tenant_id text NOT NULL,
  namespace text NOT NULL,
  content text NOT NULL,
  metadata jsonb NOT NULL,
  signature bytea NOT NULL,
  signer_receipt_sha256 text NOT NULL,
  evidence_receipt_sha256 text NOT NULL,
  envelope_sha256 text NOT NULL,
  state text NOT NULL CHECK (state = 'ACTIVE'),
  PRIMARY KEY (record_id, version)
);

CREATE TABLE trusted_memory_version_floor (
  record_id text PRIMARY KEY,
  version bigint NOT NULL CHECK (version > 0),
  envelope_sha256 text NOT NULL,
  evidence_receipt_sha256 text NOT NULL
);

REVOKE ALL ON trusted_memory_records FROM PUBLIC;
REVOKE ALL ON trusted_memory_version_floor FROM PUBLIC;
GRANT INSERT, SELECT ON trusted_memory_records TO memory_controlled_writer;
GRANT INSERT, UPDATE, SELECT ON trusted_memory_version_floor
  TO memory_controlled_writer;
GRANT SELECT ON trusted_memory_records, trusted_memory_version_floor
  TO memory_integrity_loader;

Sign, store, verify, and load

# File: memory/integrity/signed_record.py
# Dependencies: psycopg[binary]>=3.2,<4, jcs>=0.2,<1,
# PyNaCl>=1.5,<2
from __future__ import annotations

import base64
import hashlib
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Callable

import jcs
from nacl.exceptions import BadSignatureError
from nacl.signing import VerifyKey


@dataclass(frozen=True)
class Signer:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    active_key_metadata: Callable[[str], dict]
    sign_and_verify: Callable[[bytes, str, str, str], dict]


@dataclass(frozen=True)
class KeyRegistry:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    active_verify_key: Callable[[str, str, str], VerifyKey]


@dataclass(frozen=True)
class Evidence:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict], dict]
    require_committed_write: Callable[[str, int, str, str], dict]


META_FIELDS = {
    "schema_version", "record_id", "version", "tenant_id", "namespace",
    "content_sha256", "issuer", "issued_at", "key_id",
    "key_registry_version", "policy_sha256",
}


def sha256(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()


def canonical(value: dict) -> bytes:
    return bytes(jcs.canonicalize(value))


def validate_metadata(metadata: dict) -> None:
    if (
        not isinstance(metadata, dict)
        or set(metadata) != META_FIELDS
        or metadata["schema_version"] != "aidefend.signed-memory.v1"
        or not isinstance(metadata["version"], int)
        or isinstance(metadata["version"], bool)
        or metadata["version"] < 1
    ):
        raise ValueError("signed memory metadata schema invalid")


def write_record(
    conn,
    readback_conn,
    *,
    record_id: str,
    tenant_id: str,
    namespace: str,
    content: str,
    issuer: str,
    policy_sha256: str,
    signer: Signer,
    key_registry: KeyRegistry,
    evidence: Evidence,
) -> dict:
    content_bytes = content.encode("utf-8")
    with conn.transaction():
        # A transaction-scoped advisory lock serializes versions even for a
        # first write. The signer and evidence clients have bounded timeouts.
        conn.execute(
            "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
            (record_id,),
        )
        floor = conn.execute(
            "SELECT version FROM trusted_memory_version_floor "
            "WHERE record_id = %s FOR UPDATE",
            (record_id,),
        ).fetchone()
        version = 1 if floor is None else int(floor[0]) + 1
        issued_at = datetime.now(timezone.utc).isoformat()
        key_metadata = signer.active_key_metadata(policy_sha256)
        unsigned = {
            "schema_version": "aidefend.signed-memory.v1",
            "record_id": record_id,
            "version": version,
            "tenant_id": tenant_id,
            "namespace": namespace,
            "content_sha256": sha256(content_bytes),
            "issuer": issuer,
            "issued_at": issued_at,
            "key_id": key_metadata["key_id"],
            "key_registry_version": key_metadata["key_registry_version"],
            "policy_sha256": policy_sha256,
        }
        payload = canonical(unsigned)
        signing = signer.sign_and_verify(
            payload,
            unsigned["key_id"],
            unsigned["key_registry_version"],
            policy_sha256,
        )
        if (
            signing["key_id"] != unsigned["key_id"]
            or signing["key_registry_version"]
                != unsigned["key_registry_version"]
        ):
            raise ValueError("signer changed the selected key binding")
        signature = base64.b64decode(
            signing["signature_b64"], validate=True
        )
        verify_key = key_registry.active_verify_key(
            unsigned["key_id"],
            issued_at,
            unsigned["key_registry_version"],
        )
        verify_key.verify(payload, signature)
        envelope_sha256 = sha256(
            canonical({
                "metadata": unsigned,
                "signature_b64": signing["signature_b64"],
            })
        )
        prepared = evidence.append_and_verify({
            "schema_version": "aidefend.signed-memory-write.v1",
            "stage": "PREPARED_BEFORE_DATABASE_COMMIT",
            "record_id": record_id,
            "version": version,
            "tenant_id": tenant_id,
            "namespace": namespace,
            "content_sha256": unsigned["content_sha256"],
            "envelope_sha256": envelope_sha256,
            "signer_receipt_sha256": signing["receipt_sha256"],
            "policy_sha256": policy_sha256,
            "outcome": "INSUFFICIENT_DATA",
        })
        conn.execute(
            "INSERT INTO trusted_memory_records "
            "(record_id, version, tenant_id, namespace, content, metadata, "
            "signature, signer_receipt_sha256, evidence_receipt_sha256, "
            "envelope_sha256, state) "
            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'ACTIVE')",
            (
                record_id, version, tenant_id, namespace, content, unsigned,
                signature, signing["receipt_sha256"],
                prepared["receipt_sha256"], envelope_sha256,
            ),
        )
        conn.execute(
            "INSERT INTO trusted_memory_version_floor "
            "(record_id, version, envelope_sha256, evidence_receipt_sha256) "
            "VALUES (%s,%s,%s,%s) "
            "ON CONFLICT (record_id) DO UPDATE SET "
            "version = EXCLUDED.version, "
            "envelope_sha256 = EXCLUDED.envelope_sha256, "
            "evidence_receipt_sha256 = EXCLUDED.evidence_receipt_sha256",
            (
                record_id, version, envelope_sha256,
                prepared["receipt_sha256"],
            ),
        )
    committed = readback_conn.execute(
        "SELECT r.state, r.envelope_sha256, f.version, "
        "f.envelope_sha256, r.evidence_receipt_sha256 "
        "FROM trusted_memory_records r "
        "JOIN trusted_memory_version_floor f "
        "ON f.record_id = r.record_id AND f.version = r.version "
        "WHERE r.record_id = %s AND r.version = %s",
        (record_id, version),
    ).fetchone()
    if committed != (
        "ACTIVE", envelope_sha256, version, envelope_sha256,
        prepared["receipt_sha256"],
    ):
        raise RuntimeError("independent post-commit database readback mismatch")
    event = evidence.append_and_verify({
        "schema_version": "aidefend.signed-memory-write.v1",
        "stage": "COMMITTED_AND_INDEPENDENTLY_READ_BACK",
        "record_id": record_id,
        "version": version,
        "tenant_id": tenant_id,
        "namespace": namespace,
        "content_sha256": unsigned["content_sha256"],
        "envelope_sha256": envelope_sha256,
        "signer_receipt_sha256": signing["receipt_sha256"],
        "prepared_receipt_sha256": prepared["receipt_sha256"],
        "policy_sha256": policy_sha256,
        "outcome": "PASS",
    })
    return {
        "record_id": record_id,
        "version": version,
        "envelope_sha256": envelope_sha256,
        "evidence_receipt_sha256": event["receipt_sha256"],
    }


def load_verified_record(
    conn,
    *,
    record_id: str,
    expected_tenant_id: str,
    expected_namespace: str,
    expected_policy_sha256: str,
    key_registry: KeyRegistry,
    evidence: Evidence,
) -> str:
    row = conn.execute(
        "SELECT r.version, r.tenant_id, r.namespace, r.content, "
        "r.metadata, r.signature, r.signer_receipt_sha256, "
        "r.evidence_receipt_sha256, r.envelope_sha256, "
        "f.version, f.envelope_sha256, f.evidence_receipt_sha256 "
        "FROM trusted_memory_records r "
        "JOIN trusted_memory_version_floor f "
        "ON f.record_id = r.record_id AND f.version = r.version "
        "WHERE r.record_id = %s AND r.state = 'ACTIVE'",
        (record_id,),
    ).fetchone()
    if row is None:
        raise PermissionError("current trusted record unavailable")
    (
        version, tenant_id, namespace, content, metadata, signature,
        signer_receipt_sha256, write_evidence_sha256, envelope_sha256,
        floor_version, floor_envelope_sha256, floor_evidence_sha256,
    ) = row
    validate_metadata(metadata)
    if (
        tenant_id != expected_tenant_id
        or namespace != expected_namespace
        or metadata["tenant_id"] != expected_tenant_id
        or metadata["namespace"] != expected_namespace
        or metadata["record_id"] != record_id
        or metadata["version"] != version
        or version != floor_version
        or metadata["policy_sha256"] != expected_policy_sha256
        or sha256(content.encode("utf-8")) != metadata["content_sha256"]
        or write_evidence_sha256 != floor_evidence_sha256
    ):
        raise PermissionError("record scope, policy, version, or content mismatch")
    payload = canonical(metadata)
    key = key_registry.active_verify_key(
        metadata["key_id"],
        metadata["issued_at"],
        metadata["key_registry_version"],
    )
    try:
        key.verify(payload, bytes(signature))
    except BadSignatureError as exc:
        raise PermissionError("record signature invalid") from exc
    computed_envelope = sha256(canonical({
        "metadata": metadata,
        "signature_b64": base64.b64encode(bytes(signature)).decode("ascii"),
    }))
    if (
        computed_envelope != envelope_sha256
        or envelope_sha256 != floor_envelope_sha256
    ):
        raise PermissionError("record envelope rollback or substitution")
    commit_event = evidence.require_committed_write(
        record_id=record_id,
        version=version,
        envelope_sha256=envelope_sha256,
        prepared_receipt_sha256=write_evidence_sha256,
    )
    if commit_event.get("outcome") != "PASS":
        raise PermissionError("committed-write PASS evidence unavailable")
    read_event = evidence.append_and_verify({
        "schema_version": "aidefend.signed-memory-read.v1",
        "record_id": record_id,
        "version": version,
        "tenant_id": tenant_id,
        "namespace": namespace,
        "content_sha256": metadata["content_sha256"],
        "envelope_sha256": envelope_sha256,
        "signer_receipt_sha256": signer_receipt_sha256,
        "write_evidence_receipt_sha256": write_evidence_sha256,
        "write_commit_evidence_receipt_sha256":
            commit_event["receipt_sha256"],
        "policy_sha256": expected_policy_sha256,
        "outcome": "PASS",
    })
    if not read_event.get("receipt_sha256"):
        raise PermissionError("read evidence was not independently receipted")
    return content

Before you begin

After commit, a separate read-only database identity verifies the record/floor and only then may append COMMITTED PASS evidence; the loader requires that exact final receipt before content release.

Open-source tools

• OpenBao Transit
• SPIRE
• Cosign
• pyca/cryptography

Source-available / open-weight tools

• HashiCorp Vault Community (BUSL-1.1; source-available)

Commercial / hosted tools

• AWS Key Management Service
• Azure Key Vault
• Google Cloud Key Management Service

Threat mappings

MITRE ATLAS

• AML.T0080.000 AI Agent Context Poisoning: Memory
• AML.T0070 RAG Poisoning (crypto integrity prevents direct-to-DB poisoning)
• AML.T0118 Autonomous AI Agent Communication
• AML.T0118.000 Autonomous AI Agent Communication: Communication via Shared Artifacts (unsigned direct writes to shared memory never re-enter another agent's context)

MAESTRO

• Data Tampering (L2)
• Compromised RAG Pipelines (L2) (signed records prevent pipeline tampering)

OWASP LLM Top 10 2026

• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-5.1.1 Long-term / Short-term Memory Injection (signature and version checks reject direct-store memory or RAG tampering)
• AISubtech-6.1.1 Knowledge Base Poisoning (signature and version checks reject direct-store memory or RAG tampering)
• AITech-5.1 Memory System Persistence (signature and version checks reject direct-store memory or RAG tampering)
• AITech-7.2 Memory System Corruption (signature and version checks reject direct-store memory or RAG tampering)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning (cryptographic integrity detects and prevents memory poisoning)
• Raw Data 1.7: Lack of data trustworthiness (signed records ensure data trustworthiness)