Memory and Context Poisoning

Description

Adversarial data persisted to RAG indexes, vector stores, agent memory, or conversation history corrupts future decision-making, enables cross-session attacks, and leaks data between tenants.

Mitigations

Memory Validation

Validate all data before it enters persistent memory:

• Apply the same content sanitization used for input processing (strip hidden text, normalize encoding, detect injection patterns)
• Classify content by sensitivity level before storage and enforce retrieval policies based on the current user's clearance
• Implement memory write permissions: not all agent interactions should be allowed to persist data
• Require explicit, auditable approval before storing data in long-term memory

Tenant Isolation

Enforce strict data boundaries in multi-tenant deployments:

• Use separate vector store collections or namespaces per tenant with no cross-tenant query capability
• Tag all stored data with tenant ID and enforce tenant filtering at the retrieval layer, not just the application layer
• Audit retrieval results to verify no cross-tenant leakage occurs
• Implement tenant isolation tests as part of CI/CD for memory subsystems

Context Freshness Checks

Prevent stale or poisoned context from accumulating influence:

• Assign time-to-live (TTL) to all stored memories and embeddings; expire and re-validate periodically
• Weight recent, verified information higher than older or unverified memories in retrieval ranking
• Implement provenance tracking: every piece of stored data records its source, ingestion time, and validation status
• Provide administrators with tools to search, review, and purge specific memories or entire categories

Retrieval Result Filtering

Filter retrieved context before it reaches the agent:

• Apply injection detection to retrieved documents, not just user inputs
• Limit the number of retrieved chunks to reduce the attack surface for poisoned embeddings
• Use a diversity threshold in retrieval to prevent a single source from dominating the context
• Flag and quarantine retrieved content that contains instruction-like patterns

Memory Audit Trail

Maintain complete observability over the memory lifecycle:

• Log every write, read, update, and delete operation on the memory store
• Record which agent, user, and task caused each memory write
• Implement anomaly detection on memory write patterns (sudden spikes, unusual content types, cross-tenant access attempts)
• Provide forensic tools to trace a specific agent decision back to the memory entries that influenced it

Code Examples

Memory Integrity Validator

import re
import time
import hashlib
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum


class ContentRisk(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    BLOCKED = "blocked"


INJECTION_PATTERNS = [
    re.compile(r"(?i)(system|override|ignore previous|forget|disregard).*?(instruction|prompt|rule|policy)"),
    re.compile(r"(?i)(remember|store|save|persist).*?(password|secret|key|token|credential)"),
    re.compile(r"(?i)(you are now|switch to|new role|act as|pretend)"),
    re.compile(r"(?i)(for all future|from now on|always|in every)"),
]


@dataclass
class MemoryEntry:
    content: str
    source: str
    tenant_id: str
    agent_id: str
    task_id: str
    created_at: float = field(default_factory=time.time)
    ttl_seconds: int = 86400  # 24h default
    content_hash: str = ""
    risk_level: str = ""
    validated: bool = False


class MemoryIntegrityValidator:
    """Validates and controls writes to agent memory stores."""

    def __init__(self, max_entries_per_tenant: int = 10000):
        self._max_entries = max_entries_per_tenant
        self._tenant_counts: dict[str, int] = {}
        self._write_log: list[dict] = []

    def validate_write(self, entry: MemoryEntry) -> dict:
        """Validate content before allowing storage in memory."""
        result = {
            "allowed": True,
            "risk": ContentRisk.LOW.value,
            "flags": [],
        }

        # Check tenant capacity
        count = self._tenant_counts.get(entry.tenant_id, 0)
        if count >= self._max_entries:
            result["allowed"] = False
            result["flags"].append("Tenant memory capacity exceeded")
            return result

        # Scan for injection patterns
        for pattern in INJECTION_PATTERNS:
            if pattern.search(entry.content):
                result["risk"] = ContentRisk.HIGH.value
                result["flags"].append(f"Injection pattern detected: {pattern.pattern}")

        # Check for credential-like content
        if re.search(r"(?i)(api[_-]?key|password|secret|bearer|token)\s*[:=]\s*\S+", entry.content):
            result["risk"] = ContentRisk.BLOCKED.value
            result["allowed"] = False
            result["flags"].append("Credential-like content detected — blocked from storage")

        # Block if risk is too high
        if result["risk"] == ContentRisk.HIGH.value:
            result["allowed"] = False
            result["flags"].append("High-risk content requires manual review before storage")

        # Compute content hash for integrity tracking
        entry.content_hash = hashlib.sha256(entry.content.encode()).hexdigest()
        entry.risk_level = result["risk"]
        entry.validated = result["allowed"]

        # Log the write attempt
        self._write_log.append({
            "tenant_id": entry.tenant_id,
            "agent_id": entry.agent_id,
            "content_hash": entry.content_hash,
            "risk": result["risk"],
            "allowed": result["allowed"],
            "flags": result["flags"],
            "timestamp": time.time(),
        })

        if result["allowed"]:
            self._tenant_counts[entry.tenant_id] = count + 1

        return result

    def validate_retrieval(
        self, entries: list[MemoryEntry], requesting_tenant: str
    ) -> list[MemoryEntry]:
        """Filter retrieved entries for tenant isolation and freshness."""
        now = time.time()
        filtered = []
        for entry in entries:
            # Enforce tenant isolation
            if entry.tenant_id != requesting_tenant:
                continue
            # Enforce TTL
            if now - entry.created_at > entry.ttl_seconds:
                continue
            # Only return validated entries
            if not entry.validated:
                continue
            filtered.append(entry)
        return filtered

Evidence Requirements

• Memory write audit logs with content hashes, risk classifications, and validation decisions
• Tenant isolation test results proving no cross-tenant retrieval leakage
• TTL compliance reports showing expired entries are properly purged
• Injection pattern detection statistics with examples of blocked write attempts
• Retrieval provenance traces linking agent decisions to specific memory entries