AID-D-001.005: Recalled Memory Pre-Rehydration Scanning
Place a recall scanner between the memory loader and prompt builder. Evaluate role or authority assertions, trigger-like markers, encoded fragments, and known quarantine fingerprints, then emit a structured finding bound to the memory-record version, content digest, detector revision, and policy version.
Pillars: app, data
Lifecycle: operation
Scope boundary
Owns detector-derived findings for persistent-memory content recalled toward prompt assembly, including latent or encoded injections, authority assertions, trigger markers, and known quarantine fingerprints. It does not suppress, rewrite, quarantine, admit, or cryptographically verify the record.
AID-I-004.003: AID-D-001.005 assesses whether authentic recalled content is semantically hazardous before rehydration; AID-I-004.003 verifies that stored memory bytes and metadata are authentic and unchanged. A valid signature does not make the content safe, and a content finding does not prove storage integrity.
AID-I-004.004: AID-D-001.005 scans recalled content and emits a finding; AID-I-004.004 owns transactional promotion from quarantine into trusted memory. Recall-time detection cannot promote or quarantine a record, and prior promotion does not eliminate the need to rescan content under the current policy.
Implementation guidance
- AID-D-001.005-G001: Rescan every versioned memory record on the recall path before prompt assembly and emit a detector-derived, content-bound finding for downstream policy consumers.
- AID-D-001.005-G002: Maintain a tenant-aware, versioned quarantine fingerprint store and emit a content-bound match finding when recalled memory matches a confirmed malicious pattern; downstream Harden or Isolate controls decide whether recall is denied or quarantined.
AID-D-001.005-G001
Detector boundary and signed inputs
Run this detector after an authenticated memory loader resolves the exact tenant, record ID, record version, and recall request, but before prompt assembly. The regular expressions below are narrow defense-in-depth signals; the primary score must come from a pinned, calibrated prompt-safety detector invoked by this module. Load thresholds from a signature-verified policy and quarantine fingerprints from the tenant-scoped store in AID-D-001.005-G002. The detector derives its own finding from scores and matched bytes and never accepts a caller-supplied safe/quarantine verdict.
Scan and bind every recalled record
from __future__ import annotations
# File: security/recall_scan.py
import hashlib
import hmac
import unicodedata
import json
import math
import re
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Callable, Literal, Sequence
ControlStatus = Literal["PASS", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]
ROLE_ASSERTION_RE = re.compile(
r"(?i)\b(system|developer|administrator|root)\b.{0,120}"
r"\b(ignore|override|must|authority)\b"
)
ACTION_TRIGGER_RE = re.compile(
r"(?i)\b(call_tool|approve|wire_transfer|delete|exfiltrate|grant_access)\b"
)
@dataclass(frozen=True)
class MemoryEntry:
entry_id: str
tenant_id: str
memory_version: str
recall_request_id: str
content: str
@dataclass(frozen=True)
class RecallPolicy:
policy_version: str
detector_revision: str
review_score: int
escalate_score: int
quarantine_score: int
maximum_content_characters: int
def sha256_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def normalize_for_fingerprint(value: str) -> str:
return " ".join(unicodedata.normalize("NFKC", value).casefold().split())
# Version 1: NFKC, casefold, whitespace collapse, UTF-8, HMAC-SHA256.
# Import this one function from both the scanner and quarantine writer.
def recall_fingerprint(value: str, key: bytes) -> str:
if not isinstance(value, str) or not value:
raise ValueError("quarantine content is empty")
if not isinstance(key, bytes) or len(key) < 32:
raise ValueError("fingerprint HMAC key must contain at least 256 bits")
return hmac.new(key, normalize_for_fingerprint(value).encode("utf-8"), hashlib.sha256).hexdigest()
def validate_policy(policy: RecallPolicy) -> None:
values = (
policy.review_score,
policy.escalate_score,
policy.quarantine_score,
)
if (
not policy.policy_version
or not policy.detector_revision
or type(policy.maximum_content_characters) is not int
or policy.maximum_content_characters < 1
or any(type(value) is not int for value in values)
or not (0 <= values[0] < values[1] < values[2] <= 100)
):
raise ValueError("invalid signed recall-scanning policy")
def scan_recalled_entry(
entry: MemoryEntry,
policy: RecallPolicy,
*,
invoke_pinned_detector: Callable[[str], float],
known_bad_fingerprints: set[str],
fingerprint_key: bytes,
) -> dict:
validate_policy(policy)
if not all(
(
entry.entry_id,
entry.tenant_id,
entry.memory_version,
entry.recall_request_id,
)
):
raise ValueError("memory identity binding is incomplete")
if not isinstance(entry.content, str) or not entry.content:
raise ValueError("recalled content is absent")
if len(entry.content) > policy.maximum_content_characters:
raise ValueError("recalled content exceeds signed detector bound")
if any(
not isinstance(value, str)
or len(value) != 64
or set(value) - set("0123456789abcdef")
for value in known_bad_fingerprints
):
raise ValueError("quarantine fingerprint set is malformed")
content_sha256 = sha256_text(entry.content)
fingerprint = recall_fingerprint(entry.content, fingerprint_key)
detector_score = float(invoke_pinned_detector(entry.content))
if not math.isfinite(detector_score) or not 0.0 <= detector_score <= 1.0:
raise ValueError("pinned detector returned an invalid score")
matched: list[str] = []
risk = round(detector_score * 100)
if fingerprint in known_bad_fingerprints:
matched.append("known_bad_fingerprint")
risk = 100
if ROLE_ASSERTION_RE.search(entry.content):
matched.append("role_assertion_override")
risk += 30
if ACTION_TRIGGER_RE.search(entry.content):
matched.append("unsafe_action_token")
risk += 30
risk = min(risk, 100)
if risk >= policy.quarantine_score:
recommendation = "QUARANTINE"
elif risk >= policy.escalate_score:
recommendation = "ESCALATE"
elif risk >= policy.review_score:
recommendation = "REVIEW"
else:
recommendation = "NONE"
return {
"schema_version": "aidefend.recalled_memory_finding.v1",
"control_id": "AID-D-001.005",
"control_status": "PASS",
"finding_status": "FINDING" if matched or risk >= policy.review_score else "NO_FINDING",
"entry_id": entry.entry_id,
"tenant_id": entry.tenant_id,
"memory_version": entry.memory_version,
"recall_request_id": entry.recall_request_id,
"content_sha256": content_sha256,
"fingerprint_hmac_sha256": fingerprint,
"fingerprint_contract": "nfkc-casefold-whitespace-hmac-sha256.v1",
"policy_version": policy.policy_version,
"detector_revision": policy.detector_revision,
"detector_score": round(detector_score, 6),
"risk_score": risk,
"matched_signals": sorted(matched),
"recommended_disposition": recommendation,
"observed_at": datetime.now(timezone.utc).isoformat(),
}
def safe_scan(entry: MemoryEntry, policy: RecallPolicy, **kwargs: object) -> dict:
try:
return scan_recalled_entry(entry, policy, **kwargs)
except Exception as error:
content = entry.content if isinstance(entry.content, str) else ""
return {
"schema_version": "aidefend.recalled_memory_finding.v1",
"control_id": "AID-D-001.005",
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"entry_id": entry.entry_id,
"tenant_id": entry.tenant_id,
"memory_version": entry.memory_version,
"recall_request_id": entry.recall_request_id,
"content_sha256": sha256_text(content),
"policy_version": policy.policy_version,
"detector_revision": policy.detector_revision,
"error_class": type(error).__name__,
}
def reconcile_population(
expected: Sequence[tuple[str, str, str, str]],
findings: Sequence[dict],
) -> dict:
expected_keys = list(expected)
if any(
not isinstance(key, tuple)
or len(key) != 4
or any(not isinstance(value, str) or not value for value in key)
for key in expected_keys
):
raise ValueError("loader manifest memory keys are malformed")
if len(expected_keys) != len(set(expected_keys)):
raise ValueError("loader manifest contains duplicate memory keys")
observed: dict[tuple[str, str, str, str], dict] = {}
for finding in findings:
key = (
str(finding.get("tenant_id", "")),
str(finding.get("recall_request_id", "")),
str(finding.get("entry_id", "")),
str(finding.get("memory_version", "")),
)
if not all(key) or key in observed:
raise ValueError("finding identity is absent or duplicated")
observed[key] = finding
missing = sorted(set(expected_keys) - set(observed))
extra = sorted(set(observed) - set(expected_keys))
if missing or extra:
return {
"control_status": "INSUFFICIENT_DATA",
"finding_status": "UNKNOWN",
"missing": missing,
"extra": extra,
}
if any(item.get("control_status") == "ERROR" for item in observed.values()):
return {
"control_status": "ERROR",
"finding_status": "UNKNOWN",
"evaluated_count": len(observed),
}
return {
"control_status": "PASS",
"finding_status": (
"FINDING"
if any(item.get("finding_status") == "FINDING" for item in observed.values())
else "NO_FINDING"
),
"evaluated_count": len(observed),
"population_sha256": hashlib.sha256(
json.dumps(
sorted(expected_keys), separators=(",", ":")
).encode("utf-8")
).hexdigest(),
}
def emit_finding(finding: dict, append_only_sink) -> None:
append_only_sink.write(
json.dumps(finding, sort_keys=True, separators=(",", ":"), allow_nan=False)
+ "\n"
)Replay the real recall path
Build a signed fixture set containing benign memories, role-assertion injection, delayed action triggers, Unicode and whitespace variants, known quarantine fingerprints, oversized content, detector timeout/exception, NaN and out-of-range detector scores, duplicate records, identical request/entry/version keys in two tenants, a missing expected record, and an unexpected record. Invoke the same pinned detector adapter and loader-to-scanner integration used in production; assert that malformed or unmeasured inputs never produce NO_FINDING. Reorder the complete expected population and confirm that the canonical population digest is unchanged.
Operational check
For each recall request, derive the expected (tenant_id, recall_request_id, entry_id, memory_version) population from the authenticated memory-loader manifest, reconcile it with the emitted findings, then read the append-only event stream back by those keys and recompute content digests and the population roll-up. Keep raw recalled content out of ordinary telemetry. Downstream Harden or Isolate controls may consume recommended_disposition, but must make their own policy decision.
Quarantine integration: Deploy recall_fingerprint from security/recall_scan.py as the shared v1 fingerprint contract. Scanner and quarantine writer receive the same secret-manager-managed HMAC key; never accept a request-selected key. Fetch the known-bad set from the authenticated tenant/global quarantine views for that key generation. Rotate key and rebuilt fingerprints atomically; a partial generation is an error, not an empty clean set. Keep the raw content SHA-256 for evidence identity, separate from fingerprint_hmac_sha256. Test Unicode compatibility characters, case and whitespace against the store before release.
AID-D-001.005-G002
Boundary and authorization contract
Redis supplies authenticated connections, ACL-enforced key access, script serialization, and configured persistence; it does not roll back commands that already ran when a later script command fails. Keep current quarantine state and its transition history in the same per-scope, per-fingerprint list. The script below performs every check before its single mutating LPUSH, so state and audit cannot diverge inside this reference operation. Inject an authorization adapter for the authenticated actor; never accept a request field such as is_admin=true, and never interpret an unknown scope as global.
Use keyed fingerprints and one append-only state ledger
from __future__ import annotations
# File: security/quarantine_store.py
import hashlib
import hmac
import json
import re
import unicodedata
from collections.abc import Callable
from dataclasses import dataclass
from typing import Literal
import redis
from security.recall_scan import recall_fingerprint
Scope = Literal["tenant", "global"]
Action = Literal["add", "remove"]
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
HEX64 = re.compile(r"^[0-9a-f]{64}$")
MUTATE = """
local expected = tonumber(ARGV[1])
if expected == nil or expected < 0 or expected % 1 ~= 0 then
return redis.error_reply('invalid expected revision')
end
local proposed = cjson.decode(ARGV[3])
if (
type(proposed) ~= 'table'
or proposed.event_id ~= ARGV[2]
or proposed.revision ~= expected + 1
) then
return redis.error_reply('malformed proposed transition')
end
local latest = redis.call('LINDEX', KEYS[1], 0)
if latest then
local current = cjson.decode(latest)
if current.event_id == proposed.event_id then
if latest == ARGV[3] then
return latest
end
return redis.error_reply('event id reused with different content')
end
if current.revision ~= expected then
return redis.error_reply('quarantine revision conflict')
end
elseif expected ~= 0 then
return redis.error_reply('quarantine revision conflict')
end
redis.call('LPUSH', KEYS[1], ARGV[3])
return ARGV[3]
"""
@dataclass(frozen=True)
class AuthenticatedActor:
actor_id: str
authentication_event_id: str
Authorize = Callable[[AuthenticatedActor, Scope, str | None], bool]
def _identifier(value: str, name: str) -> str:
if not isinstance(value, str) or SAFE_ID.fullmatch(value) is None:
raise ValueError(name + " is absent or unsafe")
return value
def _is_sha256(value: object) -> bool:
return isinstance(value, str) and HEX64.fullmatch(value) is not None
def _strict_object(pairs: list[tuple[str, object]]) -> dict:
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate quarantine-event key: " + key)
value[key] = item
return value
def _text(value: bytes | str) -> str:
if isinstance(value, bytes):
return value.decode("utf-8", errors="strict")
if isinstance(value, str):
return value
raise TypeError("Redis response is not text")
class QuarantineFingerprintStore:
def __init__(
self,
client: redis.Redis,
*,
policy_version: str,
detector_version: str,
fingerprint_key: bytes,
authorize: Authorize,
) -> None:
self.client = client
self.policy_version = _identifier(policy_version, "policy_version")
self.detector_version = _identifier(detector_version, "detector_version")
if not isinstance(fingerprint_key, bytes) or len(fingerprint_key) < 32:
raise ValueError("fingerprint HMAC key must contain at least 256 bits")
self.fingerprint_key = fingerprint_key
self.authorize = authorize
self.script = client.register_script(MUTATE)
def fingerprint(self, text: str) -> str:
return recall_fingerprint(text, self.fingerprint_key)
def _ledger_key(
self, scope: Scope, tenant_id: str | None, fingerprint: str
) -> str:
if scope == "tenant":
segment = _identifier(tenant_id, "tenant_id")
elif scope == "global" and tenant_id is None:
segment = "global"
else:
raise ValueError("scope and tenant binding are invalid")
if not _is_sha256(fingerprint):
raise ValueError("fingerprint is not a full HMAC-SHA256")
return f"aidefend:recall_quarantine:v1:{scope}:{segment}:{fingerprint}"
def latest(
self, *, scope: Scope, tenant_id: str | None, fingerprint: str
) -> dict | None:
key = self._ledger_key(scope, tenant_id, fingerprint)
raw = self.client.lindex(key, 0)
if raw is None:
return None
value = json.loads(
_text(raw),
object_pairs_hook=_strict_object,
parse_constant=lambda item: (_ for _ in ()).throw(
ValueError("non-finite quarantine value")
),
)
required = {
"schema_version", "revision", "event_id", "action", "scope",
"tenant_id", "fingerprint_hmac_sha256", "incident_id",
"actor_id", "authentication_event_id", "policy_version",
"detector_version",
}
if (
not isinstance(value, dict)
or set(value) != required
or value["schema_version"] != "aidefend.recall_quarantine.v2"
or value["scope"] != scope
or value["tenant_id"] != tenant_id
or value["fingerprint_hmac_sha256"] != fingerprint
or value["action"] not in {"add", "remove"}
or isinstance(value["revision"], bool)
or not isinstance(value["revision"], int)
or value["revision"] < 1
or not _is_sha256(value["event_id"])
):
raise RuntimeError("quarantine ledger head is malformed or misbound")
return value
def mutate(
self,
*,
action: Action,
scope: Scope,
tenant_id: str | None,
fingerprint_hmac_sha256: str,
incident_id: str,
expected_revision: int,
actor: AuthenticatedActor,
) -> dict:
if action not in ("add", "remove") or scope not in ("tenant", "global"):
raise ValueError("unsupported quarantine action or scope")
if scope == "tenant":
tenant_id = _identifier(tenant_id, "tenant_id")
elif tenant_id is not None:
raise ValueError("global scope must not carry a tenant ID")
incident_id = _identifier(incident_id, "incident_id")
actor_id = _identifier(actor.actor_id, "actor_id")
authentication_event_id = _identifier(
actor.authentication_event_id, "authentication_event_id"
)
if (
isinstance(expected_revision, bool)
or not isinstance(expected_revision, int)
or expected_revision < 0
or not _is_sha256(fingerprint_hmac_sha256)
):
raise ValueError("revision or fingerprint is malformed")
if not self.authorize(actor, scope, tenant_id):
raise PermissionError(
"authenticated actor is not authorized for this quarantine scope"
)
base = {
"schema_version": "aidefend.recall_quarantine.v2",
"revision": expected_revision + 1,
"action": action,
"scope": scope,
"tenant_id": tenant_id,
"fingerprint_hmac_sha256": fingerprint_hmac_sha256,
"incident_id": incident_id,
"actor_id": actor_id,
"authentication_event_id": authentication_event_id,
"policy_version": self.policy_version,
"detector_version": self.detector_version,
}
canonical_base = json.dumps(
base, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
record = {
**base,
"event_id": hashlib.sha256(canonical_base).hexdigest(),
}
encoded = json.dumps(
record, sort_keys=True, separators=(",", ":"), allow_nan=False
)
returned = _text(self.script(
keys=[self._ledger_key(scope, tenant_id, fingerprint_hmac_sha256)],
args=[expected_revision, record["event_id"], encoded],
))
if not hmac.compare_digest(returned, encoded):
raise RuntimeError("Redis did not return the committed transition")
return record
def contains(self, *, tenant_id: str, fingerprint_hmac_sha256: str) -> bool:
tenant_id = _identifier(tenant_id, "tenant_id")
if not _is_sha256(fingerprint_hmac_sha256):
raise ValueError("fingerprint is malformed")
tenant = self.latest(
scope="tenant", tenant_id=tenant_id,
fingerprint=fingerprint_hmac_sha256,
)
global_entry = self.latest(
scope="global", tenant_id=None,
fingerprint=fingerprint_hmac_sha256,
)
return any(
item is not None and item["action"] == "add"
for item in (tenant, global_entry)
)Operate and verify
Grant the mutation identity only the script and list commands needed for its key pattern; deny DEL, LTRIM, arbitrary EVAL, and cross-tenant keys. Reserve global mutations for a separate curator identity. The caller reads the current revision, submits compare-and-append, and retries only after a revision conflict; a replay of the exact latest transition is idempotent. Configure Redis persistence and backup in the platform profile. Under a read-only verifier identity, read the list head and transition history back, compare event ID, revision, actor, authentication event, incident, tenant/scope, versions, and fingerprint, and reconcile it with the incident ledger. Test unknown scope, cross-tenant actor, non-curator global mutation, duplicate retry, stale revision, conflicting add/remove, malformed ledger state, interruption, and restart. Missing or malformed ledger evidence is an error, never “not quarantined.”
Open-source tools
- Guardrails AI
- NVIDIA NeMo Guardrails
- OpenSSL
Source-available / open-weight tools
- Meta Llama Guard 4 (Llama 4 Community License; open-weight)
- Granite Guardian 4.1 (Apache-2.0; open-weight)
- Qwen3Guard (Apache-2.0; open-weight)
- gpt-oss-safeguard (Apache-2.0; open-weight)
- Meta Llama Prompt Guard 2 (Llama 4 Community License; open-weight)
- Google ShieldGemma (Gemma Terms of Use; open-weight)
Commercial / hosted tools
- Check Point AI Guardrails
- Palo Alto Networks Prisma AIRS
- OpenAI Moderations API (omni-moderation-latest)
- Redis Cloud
- Redis Enterprise Software
Threat mappings
MITRE ATLAS
- AML.T0051.001 LLM Prompt Injection: Indirect (memory replay is an indirect injection path)
- AML.T0051.002 LLM Prompt Injection: Triggered (recall scanning helps catch dormant trigger-based payloads before re-execution)
- AML.T0068 LLM Prompt Obfuscation (recall scanning helps detect obfuscated or encoded payloads persisted in memory)
- AML.T0080 AI Agent Context Poisoning
- AML.T0080.000 AI Agent Context Poisoning: Memory (recalled memory is the primary exploitation path)
- AML.T0080.001 AI Agent Context Poisoning: Thread (recalled thread history contaminates conversation context)
- AML.T0094 Delay Execution of LLM Instructions (recall scanning helps intercept instructions deferred to a future turn or event)
- AML.T0070 RAG Poisoning
- AML.T0071 False RAG Entry Injection
- AML.T0130 AI Agent Response Biasing (pinned prompt-safety detectors inspect recalled memory for instruction-bearing bias before reuse)
MAESTRO
- Agent Goal Manipulation (L7) (poisoned recalled memory can redirect planning or goal interpretation)
- Input Validation Attacks (L3) (recall gate extends validation to replayed memory, not just live ingress)
- Compromised RAG Pipelines (L2) (persisted retrieved content can later re-enter context as a malicious memory artifact)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- ML01:2023 Input Manipulation Attack (recalled memory is a runtime input surface)
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack (poisoned recalled memory can hijack downstream reasoning or actions)
- ASI06:2026 Memory & Context Poisoning (recalled memory is the core exploitation path for this risk)
NIST Adversarial Machine Learning 2025
- NISTAML.015 Indirect Prompt Injection (memory replay is an indirect instruction channel)
Cisco Integrated AI Security and Safety Framework
- AISubtech-4.2.2 Session Boundary Violation
- AISubtech-5.1.1 Long-term / Short-term Memory Injection
- AITech-1.2 Indirect Prompt Injection
- AITech-4.2 Context Boundary Attacks
- AITech-5.1 Memory System Persistence
- AITech-7.2 Memory System Corruption
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection (findings identify injected instructions recalled from memory)
Databricks AI Security Framework 3.0
- Agents - Core 13.1: Memory Poisoning
- Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation (recall scanning detects spoofed/manipulated context)
- Model Serving - Inference requests 9.1: Prompt inject
- Model Serving - Inference requests 9.9: Input Resource Control (memory is a runtime input resource)