RAG Security
in Production

Python · Output monitoring pipeline (simplified)

import re
from sentence_transformers import SentenceTransformer, util

INSTRUCTION_PATTERNS = [
    r"ignore (previous|all) instructions",
    r"you are now",
    r"new system prompt",
    r"disregard (your|the) (system|previous)",
    r"act as if",
]

scorer = SentenceTransformer("all-MiniLM-L6-v2")

def check_output(retrieved_chunks, generated_response, query_text):
    flags = []

    # Check each retrieved chunk for instruction patterns
    for chunk in retrieved_chunks:
        for pattern in INSTRUCTION_PATTERNS:
            if re.search(pattern, chunk["content"], re.IGNORECASE):
                flags.append({
                    "type": "instruction_pattern",
                    "chunk_id": chunk["id"],
                    "pattern": pattern,
                    "severity": "high",
                })

    # Score semantic consistency: response vs retrieved context
    context = " ".join(c["content"] for c in retrieved_chunks)
    emb_ctx, emb_resp = scorer.encode([context, generated_response])
    consistency = float(util.cos_sim(emb_ctx, emb_resp))

    if consistency < 0.45:  # tune this threshold per system
        flags.append({
            "type": "consistency_low",
            "score": consistency,
            "severity": "medium",
        })

    return flags

Python · Structured retrieval audit log entry with chain hash

import hashlib, json, time, uuid

_prev_hash = "genesis"  # initialise at startup; load from last entry in prod

def write_audit_entry(
    user_id, collection, query_vector, results, policy, flags
):
    global _prev_hash
    entry = {
        "request_id":    str(uuid.uuid4()),
        "timestamp":     time.time(),
        "user_id":       user_id,
        "collection":    collection,
        # Never log the raw vector; log a hash of it for correlation only
        "query_hash":    hashlib.sha256(
                             str(query_vector).encode()
                         ).hexdigest()[:16],
        "result_ids":    [r["id"] for r in results],
        "result_scores": [round(r["score"], 4) for r in results],
        "policy_applied": policy,
        "flags":         flags,  # output monitor findings for this request
        "prev_hash":     _prev_hash,
    }
    entry_bytes = json.dumps(entry, sort_keys=True).encode()
    entry["entry_hash"] = hashlib.sha256(entry_bytes).hexdigest()
    _prev_hash = entry["entry_hash"]
    # Write to append-only log store (separate from the vector DB)
    append_to_log(entry)
    return entry["request_id"]

Python · Drift metrics against a known-good baseline

from collections import Counter
import numpy as np

# Capture this baseline when the system ships to production
BASELINE = {
    "avg_similarity": 0.78,    # mean similarity score across 1000 queries
    "topic_distribution": {     # rough topic breakdown of retrieved content
        "product": 0.42,
        "policy": 0.31,
        "support": 0.27,
    },
    "instruction_rate": 0.002,  # fraction of chunks flagged as instruction-like
}

def compute_drift_score(window_metrics):
    scores = []

    # 1. Similarity score drift
    sim_drift = abs(
        window_metrics["avg_similarity"] - BASELINE["avg_similarity"]
    ) / BASELINE["avg_similarity"]
    scores.append(sim_drift)

    # 2. Topic distribution drift (Jensen-Shannon divergence)
    base_dist = np.array(list(BASELINE["topic_distribution"].values()))
    curr_dist = np.array(list(window_metrics["topic_distribution"].values()))
    m = (base_dist + curr_dist) / 2
    js = (np.sum(base_dist * np.log(base_dist / m + 1e-9) +
              curr_dist * np.log(curr_dist / m + 1e-9)) / 2)
    scores.append(float(js))

    # 3. Instruction-pattern rate drift
    instr_drift = (
        window_metrics["instruction_rate"] - BASELINE["instruction_rate"]
    ) / max(BASELINE["instruction_rate"], 1e-6)
    scores.append(max(instr_drift, 0))

    return np.mean(scores)  # alert if > 0.25 over a rolling window

Shell · Run a DiscoveR RAG security assessment (pre-deployment)

# Install the DiscoveR CLI
pip install mirror-discover

# Run a full RAG assessment against your staging endpoint
discover assess \
  --target https://your-rag-endpoint/query \
  --collection rag_documents \
  --auth-header "Authorization: Bearer $STAGING_TOKEN" \
  --mode rag \
  --output report.json

# The assessment covers:
#   rag_poisoning        -- adversarial document injection tests
#   prompt_injection     -- indirect injection via retrieved content
#   namespace_boundary   -- cross-tenant isolation tests
#   embedding_inversion  -- vector plaintext reconstruction tests
#   query_poisoning      -- adversarial query vector tests
#   resource_exhaustion  -- rate limit and queue depth tests

# View the risk score summary
discover report report.json --format summary

Shell · Generate and CVE-check a RAG system SBOM

# Generate a CycloneDX SBOM for your RAG Python environment
pip install cyclonedx-bom
cyclonedx-py environment --output rag-sbom.json --format json

# Add AI-specific components not captured by pip (edit rag-sbom.json):
#   embedding_model: text-embedding-3-small v2024-11
#   vector_db: qdrant 1.12.4
#   llm_api: gpt-4o 2025-02
#   encryption_lib: mirror-sdk 2.1.0

# Check all components against the NVD CVE database
pip install pip-audit
pip-audit --requirement requirements.txt --format json --output audit.json

# Reconcile: compare SBOM against what is installed
pip list --format json > installed.json
python3 reconcile_sbom.py rag-sbom.json installed.json